"""Phase 3 — location_floor_plan step 단위 테스트."""
from __future__ import annotations

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

import pytest


def test_settings_background_mode_default_off():
    """code default 'off' 회귀 0건 보장 — pydantic Field default 직접 검증으로 .env override 영향 격리."""
    from app.core.config import Settings
    assert Settings.model_fields["background_mode"].default == "off"


def test_applicability_off_returns_false(monkeypatch):
    """background_mode='off' 시 step 비활성."""
    from app.core.config import settings
    monkeypatch.setattr(settings, "background_mode", "off")
    from app.core.applicability import _if_floor_plan_mode
    runner = MagicMock()
    assert _if_floor_plan_mode(runner) is False


def test_applicability_chain_only_returns_false(monkeypatch):
    """background_mode='chain_only' 시도 floor plan은 비활성."""
    from app.core.config import settings
    monkeypatch.setattr(settings, "background_mode", "chain_only")
    from app.core.applicability import _if_floor_plan_mode
    runner = MagicMock()
    assert _if_floor_plan_mode(runner) is False


def test_applicability_floor_plan_anchored_returns_true(monkeypatch):
    """background_mode='floor_plan_anchored' 시만 활성."""
    from app.core.config import settings
    monkeypatch.setattr(settings, "background_mode", "floor_plan_anchored")
    from app.core.applicability import _if_floor_plan_mode
    runner = MagicMock()
    assert _if_floor_plan_mode(runner) is True


def test_prompt_v1_loads():
    """location_floor_plan/1.<ts>/system.md 와 user_template.md 가 로드된다."""
    from app.modules.prompt_loader import load_prompt
    sys_text = load_prompt("location_floor_plan", "system", db=None)
    user_text = load_prompt("location_floor_plan", "user_template", db=None)
    # 핵심 키워드만 검증 — 정확한 본문 변경에 결합되지 않도록.
    assert "Top-down architectural floor plan" in sys_text
    assert "{location_short_id}" in user_text or "{{location_short_id}}" in user_text
    assert "{scenes_text}" in user_text or "{{scenes_text}}" in user_text


def test_build_user_prompt_includes_zone_markers():
    """user_prompt에 zone marker가 포함된다."""
    from app.modules.pipeline.location_floor_plan import build_user_prompt
    template = (
        "LOCATION ID: {location_short_id}\n"
        "LOCATION LABEL: {location_label}\n\n"
        "── ALL SCENES OCCURRING IN THIS LOCATION ──\n\n"
        "{scenes_text}\n\n"
        "── SELECTED SHOTS IN THIS LOCATION (visible elements) ──\n\n"
        "{selected_shots_text}\n\n"
        "── VISUAL WORLD RULES ──\n\n"
        "{visual_world_rules}\n"
    )
    out = build_user_prompt(
        template=template,
        location_short_id="L05",
        location_label="rooftop apartment",
        scenes=[
            {"scene_index": 5, "heading": "S5/거실 - DAY", "text": "민숙은 /거실에 앉아"},
            {"scene_index": 12, "heading": "S12/안방 - NIGHT", "text": "수리영이 /안방에서"},
        ],
        selected_shots=[
            {"scene_index": 5, "shot_index": 1, "description": "민숙이 TV 앞 sofa에 앉음"},
        ],
        visual_world_rules="옥탑방 layout. 투룸 구조.",
    )
    assert "L05" in out
    assert "rooftop apartment" in out
    assert "/거실" in out
    assert "/안방" in out
    assert "Shot 1: 민숙이 TV 앞" in out
    assert "옥탑방 layout" in out


def test_build_user_prompt_excludes_unselected_shots():
    """selected_shots에 없는 shot_index는 user_prompt에 포함 안 됨."""
    from app.modules.pipeline.location_floor_plan import build_user_prompt
    template = "{selected_shots_text}"
    out = build_user_prompt(
        template=template,
        location_short_id="L05",
        location_label="x",
        scenes=[],
        selected_shots=[
            {"scene_index": 5, "shot_index": 1, "description": "shot one"},
            {"scene_index": 5, "shot_index": 3, "description": "shot three"},
        ],
        visual_world_rules="",
    )
    assert "shot one" in out
    assert "shot three" in out
    assert "shot two" not in out


def test_validate_prompt_length_too_short():
    """500자 미만 prompt는 ValueError."""
    from app.modules.pipeline.location_floor_plan import validate_prompt_length
    with pytest.raises(ValueError, match="too short"):
        validate_prompt_length("short")


def test_validate_prompt_length_too_long():
    """6000자 초과 prompt는 ValueError."""
    from app.modules.pipeline.location_floor_plan import validate_prompt_length
    with pytest.raises(ValueError, match="too long"):
        validate_prompt_length("x" * 6001)


def test_validate_prompt_length_ok():
    """500-6000자는 통과."""
    from app.modules.pipeline.location_floor_plan import validate_prompt_length
    validate_prompt_length("x" * 500)   # boundary OK
    validate_prompt_length("x" * 6000)  # boundary OK


# === I2 fix: placeholder safety in user_template ===

def test_user_template_format_with_all_keys_no_keyerror():
    """user_template의 placeholder를 모두 채우면 KeyError 없이 substitution 성공.

    v1: 5 placeholder. v2: 7 placeholder (previous_floor_plans_block, building_group_context 추가).
    build_user_prompt가 새 placeholder를 빈 문자열로 채우므로 호환.
    """
    from app.modules.pipeline.location_floor_plan import build_user_prompt
    from app.modules.prompt_loader import load_prompt
    template = load_prompt("location_floor_plan", "user_template", db=None)
    out = build_user_prompt(
        template=template,
        location_short_id="L05",
        location_label="rooftop apartment",
        scenes=[{"scene_index": 5, "heading": "S5", "text": "sample text"}],
        selected_shots=[{"scene_index": 5, "shot_index": 1, "description": "sample"}],
        visual_world_rules="옥탑방 layout",
    )
    assert "L05" in out
    assert "rooftop apartment" in out
    assert "sample text" in out


def test_generate_prompt_retry_on_failure(monkeypatch):
    """gpt-5.5 호출 실패 시 retry 3회 + 최종 실패는 RuntimeError."""
    from app.modules.pipeline.location_floor_plan import generate_floor_plan_prompt

    calls = {"count": 0}
    def failing(**kwargs):
        calls["count"] += 1
        raise RuntimeError("LLM down")

    monkeypatch.setattr("app.modules.pipeline.location_floor_plan.time.sleep", lambda s: None)
    with pytest.raises(RuntimeError, match="failed after 3 retries"):
        generate_floor_plan_prompt(
            system_prompt="x",
            user_prompt="y",
            project_config=MagicMock(),
            call_text_fn=failing,
        )
    assert calls["count"] == 4  # 1 initial + 3 retries


def test_generate_prompt_succeeds_on_retry(monkeypatch):
    """첫 호출 실패 → 두 번째 성공."""
    from app.modules.pipeline.location_floor_plan import generate_floor_plan_prompt

    calls = {"count": 0}
    def flaky(**kwargs):
        calls["count"] += 1
        if calls["count"] == 1:
            raise RuntimeError("transient")
        return "x" * 1500

    monkeypatch.setattr("app.modules.pipeline.location_floor_plan.time.sleep", lambda s: None)
    out = generate_floor_plan_prompt(
        system_prompt="x",
        user_prompt="y",
        project_config=MagicMock(),
        call_text_fn=flaky,
    )
    assert len(out) == 1500
    assert calls["count"] == 2


def test_generate_prompt_short_response_triggers_retry(monkeypatch):
    """Length validation 실패 시 retry 작동 확인 — 첫 번째 short → 두 번째 OK."""
    from app.modules.pipeline.location_floor_plan import generate_floor_plan_prompt

    calls = {"count": 0}
    def short_then_ok(**kwargs):
        calls["count"] += 1
        return "x" * 100 if calls["count"] == 1 else "x" * 1500

    monkeypatch.setattr("app.modules.pipeline.location_floor_plan.time.sleep", lambda s: None)
    out = generate_floor_plan_prompt(
        system_prompt="x",
        user_prompt="y",
        project_config=MagicMock(),
        call_text_fn=short_then_ok,
    )
    assert len(out) == 1500
    assert calls["count"] == 2


def test_generate_prompt_persistently_short_consumes_retry_budget(monkeypatch):
    """Length validation이 계속 실패하면 retry 3회 다 소비 후 RuntimeError."""
    from app.modules.pipeline.location_floor_plan import generate_floor_plan_prompt

    calls = {"count": 0}
    def always_short(**kwargs):
        calls["count"] += 1
        return "x" * 100  # always under 500

    monkeypatch.setattr("app.modules.pipeline.location_floor_plan.time.sleep", lambda s: None)
    with pytest.raises(RuntimeError, match="failed after 3 retries"):
        generate_floor_plan_prompt(
            system_prompt="x",
            user_prompt="y",
            project_config=MagicMock(),
            call_text_fn=always_short,
        )
    assert calls["count"] == 4  # 1 initial + 3 retries


def test_generate_image_retry_on_failure(monkeypatch):
    """gpt-image-2 호출 실패 시 retry 3회."""
    from app.modules.pipeline.location_floor_plan import generate_floor_plan_image

    client = MagicMock()
    client.images.generate.side_effect = RuntimeError("API down")
    monkeypatch.setattr("app.modules.pipeline.location_floor_plan.time.sleep", lambda s: None)
    with pytest.raises(RuntimeError, match="failed after 3 retries"):
        generate_floor_plan_image(prompt="x" * 1500, openai_client=client)
    assert client.images.generate.call_count == 4


def test_generate_image_rejects_empty_png(monkeypatch):
    """gpt-image-2가 0-byte PNG 반환 시 retry → 최종 RuntimeError."""
    from app.modules.pipeline.location_floor_plan import generate_floor_plan_image

    client = MagicMock()
    # b64decode("") = b""
    client.images.generate.return_value.data = [MagicMock(b64_json="")]

    monkeypatch.setattr("app.modules.pipeline.location_floor_plan.time.sleep", lambda s: None)
    with pytest.raises(RuntimeError, match="failed after 3 retries"):
        generate_floor_plan_image(prompt="x" * 1500, openai_client=client)
    assert client.images.generate.call_count == 4


def test_generate_image_rejects_too_small_png(monkeypatch):
    """1024 bytes 미만 PNG 반환 시 retry → 최종 RuntimeError."""
    import base64
    from app.modules.pipeline.location_floor_plan import generate_floor_plan_image

    tiny_png = base64.b64encode(b"x" * 100).decode()
    client = MagicMock()
    client.images.generate.return_value.data = [MagicMock(b64_json=tiny_png)]

    monkeypatch.setattr("app.modules.pipeline.location_floor_plan.time.sleep", lambda s: None)
    with pytest.raises(RuntimeError, match="failed after 3 retries"):
        generate_floor_plan_image(prompt="x" * 1500, openai_client=client)
    assert client.images.generate.call_count == 4


# === Phase 5 T4: ref_paths multi-image edit ===

def test_generate_floor_plan_image_text_only_when_no_refs():
    """ref_paths=None 회귀 — Phase 3 text-only 동작 유지 (images.generate 호출)."""
    import base64
    from app.modules.pipeline.location_floor_plan import generate_floor_plan_image

    client = MagicMock()
    client.images.generate.return_value.data = [
        MagicMock(b64_json=base64.b64encode(b"x" * 2000).decode())
    ]
    out = generate_floor_plan_image(prompt="p" * 1500, openai_client=client, ref_paths=None)
    assert client.images.generate.called
    assert not client.images.edit.called
    assert len(out) == 2000


def test_generate_floor_plan_image_single_ref_uses_edit(tmp_path):
    """ref_paths=[file] 1개 → images.edit (single image edit)."""
    import base64
    from app.modules.pipeline.location_floor_plan import generate_floor_plan_image

    ref1 = tmp_path / "ref1.png"
    ref1.write_bytes(b"PNG1" * 300)  # 1200 bytes >= 1024
    client = MagicMock()
    client.images.edit.return_value.data = [
        MagicMock(b64_json=base64.b64encode(b"x" * 2000).decode())
    ]
    out = generate_floor_plan_image(
        prompt="p" * 1500, openai_client=client, ref_paths=[ref1]
    )
    assert client.images.edit.called
    assert not client.images.generate.called
    call = client.images.edit.call_args
    assert isinstance(call.kwargs["image"], list)
    assert len(call.kwargs["image"]) == 1
    assert len(out) == 2000


def test_generate_floor_plan_image_multi_ref_uses_edit(tmp_path):
    """ref_paths=[f1, f2] 2개 → images.edit with image=list of 2."""
    import base64
    from app.modules.pipeline.location_floor_plan import generate_floor_plan_image

    ref1 = tmp_path / "ref1.png"
    ref1.write_bytes(b"PNG1" * 300)
    ref2 = tmp_path / "ref2.png"
    ref2.write_bytes(b"PNG2" * 300)
    client = MagicMock()
    client.images.edit.return_value.data = [
        MagicMock(b64_json=base64.b64encode(b"x" * 2000).decode())
    ]
    out = generate_floor_plan_image(
        prompt="p" * 1500, openai_client=client, ref_paths=[ref1, ref2]
    )
    assert client.images.edit.called
    assert not client.images.generate.called
    call = client.images.edit.call_args
    assert isinstance(call.kwargs["image"], list)
    assert len(call.kwargs["image"]) == 2
    assert len(out) == 2000


def test_generate_floor_plan_image_invalid_path_falls_back_to_text_only(tmp_path):
    """ref_paths=[non-existent] → filtered out → text-only fallback."""
    import base64
    from app.modules.pipeline.location_floor_plan import generate_floor_plan_image

    bogus = tmp_path / "does_not_exist.png"
    client = MagicMock()
    client.images.generate.return_value.data = [
        MagicMock(b64_json=base64.b64encode(b"x" * 2000).decode())
    ]
    out = generate_floor_plan_image(
        prompt="p" * 1500, openai_client=client, ref_paths=[bogus]
    )
    assert client.images.generate.called
    assert not client.images.edit.called
    assert len(out) == 2000


def test_generate_floor_plan_image_too_small_ref_filtered(tmp_path):
    """ref_paths=[file_too_small (<1024 bytes)] → filtered out → text-only fallback."""
    import base64
    from app.modules.pipeline.location_floor_plan import generate_floor_plan_image

    small = tmp_path / "small.png"
    small.write_bytes(b"x" * 100)  # 100 bytes < 1024
    client = MagicMock()
    client.images.generate.return_value.data = [
        MagicMock(b64_json=base64.b64encode(b"x" * 2000).decode())
    ]
    out = generate_floor_plan_image(
        prompt="p" * 1500, openai_client=client, ref_paths=[small]
    )
    assert client.images.generate.called
    assert not client.images.edit.called
    assert len(out) == 2000


# === Task 5: LocationFloorPlanStep ===

def test_step_no_planner_cp_returns_zero(tmp_path, monkeypatch):
    """planner cp 없으면 _execute는 0 반환 (legacy: selected shot 0건 시나리오 대체)."""
    from app.core.steps.location_floor_plan_step import LocationFloorPlanStep

    monkeypatch.setattr("app.core.config.settings.projects_dir", str(tmp_path))
    monkeypatch.setattr("app.core.config.settings.openai_api_key", "test-key")
    monkeypatch.setattr("app.core.config.settings.background_mode", "floor_plan_anchored")

    runner = LocationFloorPlanStep.__new__(LocationFloorPlanStep)
    runner.project_id = ""
    runner.episode_id = "eid"
    runner.db = MagicMock()
    runner.db.query.return_value.filter.return_value.all.return_value = []
    runner.project_config = MagicMock()
    runner.build_opik_metadata = MagicMock(return_value={})

    result = runner._execute(mode="resume")
    assert result["applicable_count"] == 0
    assert result["completed_count"] == 0
    assert result["failed_count"] == 0


def test_config_hash_changes_on_mode_flip(monkeypatch):
    """background_mode 변경 시 config_hash가 달라진다."""
    from app.core.steps.location_floor_plan_step import LocationFloorPlanStep
    from app.core.config import settings

    runner = LocationFloorPlanStep.__new__(LocationFloorPlanStep)

    monkeypatch.setattr(settings, "background_mode", "off")
    h1 = runner._config_hash()
    monkeypatch.setattr(settings, "background_mode", "floor_plan_anchored")
    h2 = runner._config_hash()
    assert h1 != h2


def test_process_floor_plan_failed_records_status(tmp_path, monkeypatch):
    """LLM 실패 시 _process_floor_plan은 status='failed' + failure_reason 기록."""
    from app.core.steps.location_floor_plan_step import LocationFloorPlanStep

    runner = LocationFloorPlanStep.__new__(LocationFloorPlanStep)
    runner.project_config = MagicMock()
    runner.build_opik_metadata = MagicMock(return_value={})

    monkeypatch.setattr(
        "app.modules.pipeline.location_floor_plan.generate_floor_plan_prompt",
        lambda **kw: (_ for _ in ()).throw(RuntimeError("LLM down")),
    )
    image_dir = tmp_path / "fp"
    image_dir.mkdir()
    spec = {
        "id": "FP_L05",
        "primary_location_id": "L05",
        "building_group": "g1",
        "location_ids": ["L05"],
        "shot_count": 5,
    }
    result = runner._process_floor_plan(
        fp_id="FP_L05",
        spec=spec,
        location_canon_by_short={"L05": "canon-l05"},
        location_label_by_short={"L05": "rooftop"},
        prev_summaries=[],
        same_group_ref_paths=[],
        system_prompt="x",
        user_template="{location_short_id}",
        rules_text="",
        image_dir=image_dir,
        scenes=[],
        selected_shots=[],
    )
    assert result["status"] == "failed"
    assert "RuntimeError" in result["failure_reason"]
    assert result.get("png_path", "") == ""


# === Task 6: manifest + STEP_CLASSES + PIPELINE_STEPS 등록 ===


@pytest.mark.skip(reason="Phase 7 deprecation (T16): applicability='disabled', see test_phase5_deprecation.py")
def test_manifest_entry_exists():
    """step_manifest에 location_floor_plan 등록됨."""
    from app.core.step_manifest import STEP_MANIFEST
    assert "location_floor_plan" in STEP_MANIFEST
    e = STEP_MANIFEST["location_floor_plan"]
    assert e["category"] == "image"
    assert e["order"] == 21.5  # was 19.85 (image>analysis 불변식 위반 fix)
    assert e["applicability"] == "if_floor_plan_mode"
    assert e["step_type"] == "asset"
    assert e["lifecycle"] == "active"
    assert "shot_validator" in e["depends_on"]
    assert "shot_selection" in e["depends_on"]
    assert "scene_director" in e["depends_on"]  # C1 fix dependency


def test_step_classes_includes_location_floor_plan():
    """STEP_CLASSES에 LocationFloorPlanStep 등록됨."""
    from app.core.steps import STEP_CLASSES
    from app.core.steps.location_floor_plan_step import LocationFloorPlanStep
    assert STEP_CLASSES.get("location_floor_plan") is LocationFloorPlanStep


def test_pipeline_steps_includes_location_floor_plan():
    """llm_client.PIPELINE_STEPS 등록 — 모델 매핑 parity."""
    from app.modules.llm.llm_client import PIPELINE_STEPS
    assert "location_floor_plan" in PIPELINE_STEPS
    assert PIPELINE_STEPS["location_floor_plan"]["default"] in ("gpt", "gpt-5.5")


# === Phase 5 Task 5: planner-driven sequential redesign ===


def _planner_runner_factory(tmp_path, monkeypatch, planner_cp_data, *, project_id="p", episode_id="eid"):
    """Phase 5 T5 sequential _execute용 공통 fixture.

    project_id 하위에 background_planner manifest를 작성한다 (StepRunner._load_prev_checkpoint
    가 projects_dir/project_id/checkpoints/episodes/episode_id/<step>/manifest.json을 본다).
    """
    from app.core.steps.location_floor_plan_step import LocationFloorPlanStep

    ckpt_root = tmp_path / project_id / "checkpoints" / "episodes" / episode_id
    (ckpt_root / "background_planner").mkdir(parents=True, exist_ok=True)
    (ckpt_root / "background_planner" / "manifest.json").write_text(
        json.dumps({"data": planner_cp_data}), encoding="utf-8"
    )

    monkeypatch.setattr("app.core.config.settings.projects_dir", str(tmp_path))
    monkeypatch.setattr("app.core.config.settings.openai_api_key", "test-key")
    monkeypatch.setattr("app.core.config.settings.background_mode", "floor_plan_anchored")

    runner = LocationFloorPlanStep.__new__(LocationFloorPlanStep)
    runner.project_id = project_id
    runner.episode_id = episode_id
    runner.db = MagicMock()
    runner.db.query.return_value.filter.return_value.all.return_value = []
    runner.db.query.return_value.filter_by.return_value.first.return_value = None
    runner.project_config = MagicMock()
    runner.build_opik_metadata = MagicMock(return_value={})
    return runner


def test_t5_no_planner_checkpoint_returns_zero(tmp_path, monkeypatch):
    """planner cp 없음 → applicable_count=0."""
    from app.core.steps.location_floor_plan_step import LocationFloorPlanStep

    monkeypatch.setattr("app.core.config.settings.projects_dir", str(tmp_path))
    monkeypatch.setattr("app.core.config.settings.openai_api_key", "test-key")
    monkeypatch.setattr("app.core.config.settings.background_mode", "floor_plan_anchored")

    runner = LocationFloorPlanStep.__new__(LocationFloorPlanStep)
    runner.project_id = "p"
    runner.episode_id = "eid"
    runner.db = MagicMock()
    runner.project_config = MagicMock()
    runner.build_opik_metadata = MagicMock(return_value={})

    result = runner._execute(mode="resume")
    assert result["applicable_count"] == 0
    assert result["completed_count"] == 0
    assert result["failed_count"] == 0
    assert result["data"]["floor_plans"] == {}


def test_t5_empty_floor_plans_graceful_noop(tmp_path, monkeypatch):
    """planner.floor_plans 빈 list → graceful no-op."""
    runner = _planner_runner_factory(
        tmp_path, monkeypatch,
        {"floor_plans": [], "floor_plan_order": []},
    )
    result = runner._execute(mode="resume")
    assert result["applicable_count"] == 0
    assert result["completed_count"] == 0
    assert result["data"]["floor_plans"] == {}


def test_t5_processes_in_planner_order(tmp_path, monkeypatch):
    """floor_plan_order 순서대로 처리. FP3 → FP1 → FP2 순서.

    FP1, FP3은 같은 building_group="g1", FP2는 "g2".
    """
    planner_data = {
        "floor_plans": [
            {"id": "FP_L05", "primary_location_id": "L05", "building_group": "g1",
             "location_ids": ["L05"], "shot_count": 5},
            {"id": "FP_L11", "primary_location_id": "L11", "building_group": "g1",
             "location_ids": ["L11"], "shot_count": 4},
            {"id": "FP_L02", "primary_location_id": "L02", "building_group": "g2",
             "location_ids": ["L02"], "shot_count": 3},
        ],
        # 사용자 요구사항: FP3 → FP1 → FP2 순서로 처리. T5 spec text: 대응 ID 매핑.
        "floor_plan_order": ["FP_L11", "FP_L05", "FP_L02"],
    }
    runner = _planner_runner_factory(tmp_path, monkeypatch, planner_data)

    # entity_canon mock (location 3개)
    canon_l05 = MagicMock(id="canon-l05", short_id="L05", entity_type="location", name="rooftop")
    canon_l11 = MagicMock(id="canon-l11", short_id="L11", entity_type="location", name="hallway")
    canon_l02 = MagicMock(id="canon-l02", short_id="L02", entity_type="location", name="store")
    runner.db.query.return_value.filter.return_value.all.return_value = [canon_l05, canon_l11, canon_l02]

    # 처리 호출 순서 capture
    call_log = []
    def fake_process_floor_plan(self, *, fp_id, spec, prev_summaries, same_group_ref_paths, **kw):
        call_log.append({
            "fp_id": fp_id,
            "prev_ids": [oid for oid, _ in prev_summaries],
            "ref_count": len(same_group_ref_paths),
        })
        png_path = kw["image_dir"] / f"{fp_id}.png"
        png_path.write_bytes(b"x" * 2048)
        return {
            "status": "ok",
            "prompt_text": f"prompt for {fp_id}",
            "png_path": str(png_path),
            "fp_id": fp_id,
            "primary_location_id": spec["primary_location_id"],
            "building_group": spec["building_group"],
            "location_ids": spec["location_ids"],
            "ref_used": "text_only" if not same_group_ref_paths else "single_ref",
        }
    monkeypatch.setattr(
        "app.core.steps.location_floor_plan_step.LocationFloorPlanStep._process_floor_plan",
        fake_process_floor_plan,
    )

    result = runner._execute(mode="resume")
    assert result["applicable_count"] == 3
    assert result["completed_count"] == 3
    assert [c["fp_id"] for c in call_log] == ["FP_L11", "FP_L05", "FP_L02"]


def test_t5_prev_summaries_injected(tmp_path, monkeypatch):
    """prev 도면 요약이 다음 fp 처리 시 prev_summaries에 포함됨."""
    planner_data = {
        "floor_plans": [
            {"id": "FP_A", "primary_location_id": "L05", "building_group": "g1",
             "location_ids": ["L05"], "shot_count": 5},
            {"id": "FP_B", "primary_location_id": "L02", "building_group": "g2",
             "location_ids": ["L02"], "shot_count": 3},
        ],
        "floor_plan_order": ["FP_A", "FP_B"],
    }
    runner = _planner_runner_factory(tmp_path, monkeypatch, planner_data)
    canon_l05 = MagicMock(id="canon-l05", short_id="L05", entity_type="location", name="rooftop")
    canon_l02 = MagicMock(id="canon-l02", short_id="L02", entity_type="location", name="store")
    runner.db.query.return_value.filter.return_value.all.return_value = [canon_l05, canon_l02]

    captured = []
    def fake_process(self, *, fp_id, spec, prev_summaries, same_group_ref_paths, **kw):
        captured.append({"fp_id": fp_id, "prev_summaries": list(prev_summaries)})
        png_path = kw["image_dir"] / f"{fp_id}.png"
        png_path.write_bytes(b"x" * 2048)
        return {
            "status": "ok",
            "prompt_text": f"prompt for {fp_id} " * 50,  # not truncated
            "png_path": str(png_path),
            "fp_id": fp_id,
            "primary_location_id": spec["primary_location_id"],
            "building_group": spec["building_group"],
            "location_ids": spec["location_ids"],
            "ref_used": "text_only",
        }
    monkeypatch.setattr(
        "app.core.steps.location_floor_plan_step.LocationFloorPlanStep._process_floor_plan",
        fake_process,
    )

    runner._execute(mode="resume")
    assert captured[0]["prev_summaries"] == []
    assert len(captured[1]["prev_summaries"]) == 1
    prev_id, prev_summary = captured[1]["prev_summaries"][0]
    assert prev_id == "FP_A"
    assert isinstance(prev_summary, str) and len(prev_summary) > 0


def test_t5_same_building_group_ref_attached(tmp_path, monkeypatch):
    """같은 building_group의 prev PNG가 ref_paths에 포함됨.

    FP_A, FP_C는 g1. FP_B는 g2. floor_plan_order=[FP_A, FP_B, FP_C].
      - FP_A: refs=[]
      - FP_B: refs=[] (다른 group)
      - FP_C: refs=[FP_A.png] (같은 g1)
    """
    planner_data = {
        "floor_plans": [
            {"id": "FP_A", "primary_location_id": "L05", "building_group": "g1",
             "location_ids": ["L05"], "shot_count": 5},
            {"id": "FP_B", "primary_location_id": "L02", "building_group": "g2",
             "location_ids": ["L02"], "shot_count": 3},
            {"id": "FP_C", "primary_location_id": "L11", "building_group": "g1",
             "location_ids": ["L11"], "shot_count": 4},
        ],
        "floor_plan_order": ["FP_A", "FP_B", "FP_C"],
    }
    runner = _planner_runner_factory(tmp_path, monkeypatch, planner_data)
    canons = [
        MagicMock(id=f"canon-{s.lower()}", short_id=s, entity_type="location", name=s)
        for s in ("L05", "L11", "L02")
    ]
    runner.db.query.return_value.filter.return_value.all.return_value = canons

    captured = []
    def fake_process(self, *, fp_id, spec, prev_summaries, same_group_ref_paths, **kw):
        captured.append({"fp_id": fp_id, "ref_count": len(same_group_ref_paths),
                         "ref_basenames": [Path(p).name for p in same_group_ref_paths]})
        png_path = kw["image_dir"] / f"{fp_id}.png"
        png_path.write_bytes(b"x" * 2048)
        return {
            "status": "ok",
            "prompt_text": f"prompt {fp_id}",
            "png_path": str(png_path),
            "fp_id": fp_id,
            "primary_location_id": spec["primary_location_id"],
            "building_group": spec["building_group"],
            "location_ids": spec["location_ids"],
            "ref_used": "single_ref" if same_group_ref_paths else "text_only",
        }
    monkeypatch.setattr(
        "app.core.steps.location_floor_plan_step.LocationFloorPlanStep._process_floor_plan",
        fake_process,
    )

    runner._execute(mode="resume")
    assert captured[0] == {"fp_id": "FP_A", "ref_count": 0, "ref_basenames": []}
    assert captured[1] == {"fp_id": "FP_B", "ref_count": 0, "ref_basenames": []}
    assert captured[2]["fp_id"] == "FP_C"
    assert captured[2]["ref_count"] == 1
    assert "FP_A.png" in captured[2]["ref_basenames"]


def test_t5_image_asset_upsert_variant_v00(tmp_path, monkeypatch):
    """ImageAsset(asset_type='floor_plan', variant_index=0, variant_label='v00') UPSERT."""
    planner_data = {
        "floor_plans": [
            {"id": "FP_A", "primary_location_id": "L05", "building_group": "g1",
             "location_ids": ["L05"], "shot_count": 5},
        ],
        "floor_plan_order": ["FP_A"],
    }
    runner = _planner_runner_factory(tmp_path, monkeypatch, planner_data)
    canon_l05 = MagicMock(id="canon-l05", short_id="L05", entity_type="location", name="rooftop")
    runner.db.query.return_value.filter.return_value.all.return_value = [canon_l05]
    runner.db.query.return_value.filter_by.return_value.first.return_value = None  # no existing

    added: List[Any] = []
    runner.db.add = lambda obj: added.append(obj)

    def fake_process(self, *, fp_id, spec, **kw):
        png_path = kw["image_dir"] / f"{fp_id}.png"
        png_path.write_bytes(b"x" * 2048)
        return {
            "status": "ok",
            "prompt_text": f"prompt {fp_id}",
            "png_path": str(png_path),
            "fp_id": fp_id,
            "primary_location_id": spec["primary_location_id"],
            "building_group": spec["building_group"],
            "location_ids": spec["location_ids"],
            "ref_used": "text_only",
        }
    monkeypatch.setattr(
        "app.core.steps.location_floor_plan_step.LocationFloorPlanStep._process_floor_plan",
        fake_process,
    )

    runner._execute(mode="resume")
    assert len(added) == 1
    asset = added[0]
    assert asset.asset_type == "floor_plan"
    assert asset.entity_id == "canon-l05"
    assert asset.variant_index == 0
    assert asset.variant_label == "v00"
    assert asset.t2i_guide is None
    assert asset.is_primary == 1


def test_t5_partial_failure_continues_processing(tmp_path, monkeypatch):
    """1 fp 실패해도 나머지 fp는 계속 진행."""
    planner_data = {
        "floor_plans": [
            {"id": "FP_A", "primary_location_id": "L05", "building_group": "g1",
             "location_ids": ["L05"], "shot_count": 5},
            {"id": "FP_B", "primary_location_id": "L02", "building_group": "g2",
             "location_ids": ["L02"], "shot_count": 3},
        ],
        "floor_plan_order": ["FP_A", "FP_B"],
    }
    runner = _planner_runner_factory(tmp_path, monkeypatch, planner_data)
    canons = [
        MagicMock(id="canon-l05", short_id="L05", entity_type="location", name="rooftop"),
        MagicMock(id="canon-l02", short_id="L02", entity_type="location", name="store"),
    ]
    runner.db.query.return_value.filter.return_value.all.return_value = canons

    def fake_process(self, *, fp_id, spec, **kw):
        if fp_id == "FP_A":
            return {
                "status": "failed",
                "prompt_text": "",
                "png_path": "",
                "fp_id": fp_id,
                "primary_location_id": spec["primary_location_id"],
                "building_group": spec["building_group"],
                "location_ids": spec["location_ids"],
                "ref_used": "text_only",
                "failure_reason": "RuntimeError: LLM down",
            }
        png_path = kw["image_dir"] / f"{fp_id}.png"
        png_path.write_bytes(b"x" * 2048)
        return {
            "status": "ok",
            "prompt_text": f"prompt {fp_id}",
            "png_path": str(png_path),
            "fp_id": fp_id,
            "primary_location_id": spec["primary_location_id"],
            "building_group": spec["building_group"],
            "location_ids": spec["location_ids"],
            "ref_used": "text_only",
        }
    monkeypatch.setattr(
        "app.core.steps.location_floor_plan_step.LocationFloorPlanStep._process_floor_plan",
        fake_process,
    )

    result = runner._execute(mode="resume")
    assert result["applicable_count"] == 2
    assert result["completed_count"] == 1
    assert result["failed_count"] == 1
    fps = result["data"]["floor_plans"]
    assert fps["FP_A"]["status"] == "failed"
    assert fps["FP_B"]["status"] == "ok"


def test_t5_backward_compat_locations_alias(tmp_path, monkeypatch):
    """checkpoint에 floor_plans (new) + locations (compat alias) 둘 다 포함.

    Phase 4의 background_chain_planning_step._load_floor_plan_prompts와
    background_chain_render_step._load_floor_plan_paths가 data.locations[]
    리스트 형태를 읽어서 정상 동작해야 한다.
    """
    planner_data = {
        "floor_plans": [
            {"id": "FP_A", "primary_location_id": "L05", "building_group": "g1",
             "location_ids": ["L05"], "shot_count": 5},
        ],
        "floor_plan_order": ["FP_A"],
    }
    runner = _planner_runner_factory(tmp_path, monkeypatch, planner_data)
    canon_l05 = MagicMock(id="canon-l05", short_id="L05", entity_type="location", name="rooftop")
    runner.db.query.return_value.filter.return_value.all.return_value = [canon_l05]

    def fake_process(self, *, fp_id, spec, **kw):
        png_path = kw["image_dir"] / f"{fp_id}.png"
        png_path.write_bytes(b"x" * 2048)
        return {
            "status": "ok",
            "prompt_text": f"Top-down floor plan of {spec['primary_location_id']}",
            "png_path": str(png_path),
            "fp_id": fp_id,
            "primary_location_id": spec["primary_location_id"],
            "building_group": spec["building_group"],
            "location_ids": spec["location_ids"],
            "ref_used": "text_only",
        }
    monkeypatch.setattr(
        "app.core.steps.location_floor_plan_step.LocationFloorPlanStep._process_floor_plan",
        fake_process,
    )

    result = runner._execute(mode="resume")
    # new shape
    assert "FP_A" in result["data"]["floor_plans"]
    # compat alias for Phase 4
    assert "locations" in result["data"]
    locs = result["data"]["locations"]
    assert isinstance(locs, list)
    assert len(locs) == 1
    loc = locs[0]
    assert loc["id"] == "L05"  # primary_location short id
    assert loc["status"] == "ok"
    assert loc["prompt_text"].startswith("Top-down floor plan")
    assert loc["image_path"]  # non-empty