"""background_chain_render 단위 테스트.

OpenAI client + PromptSanitizer를 mock하여 호출 시퀀스/parent ref 흐름/
sanitize retry/text-only fallback/우주 외 잔재 방지 동작을 검증한다.
"""
from __future__ import annotations

import base64
from pathlib import Path
from typing import List
from unittest.mock import MagicMock, patch

import pytest

from app.modules.pipeline.background_chain_render import (
    _looks_like_moderation_block,
    render_node_image,
    render_one_location,
    run_background_chain_render,
)


@pytest.fixture(autouse=True)
def _no_sleep(monkeypatch):
    """moderation retry sleep을 우회 — 단위 테스트 시간 단축."""
    import app.modules.pipeline.background_chain_render as mod
    monkeypatch.setattr(mod.time, "sleep", lambda *a, **kw: None)


# ───────────────────────── 유틸 ─────────────────────────


def _png_response(byte_marker: bytes = b"PNG_BYTES"):
    """OpenAI images.edit/generate 응답 mock — b64_json 필드만."""
    item = MagicMock()
    item.b64_json = base64.b64encode(byte_marker).decode("ascii")
    resp = MagicMock()
    resp.data = [item]
    return resp


def _moderation_error():
    return RuntimeError(
        "Image generation blocked by content_policy: SAFETY violation"
    )


# ───────────────────── _looks_like_moderation_block ─────────────────────


def test_looks_like_moderation_block_detects_safety():
    assert _looks_like_moderation_block(_moderation_error())
    assert _looks_like_moderation_block(RuntimeError("moderation flagged"))
    assert _looks_like_moderation_block(RuntimeError("Content was BLOCKED"))


def test_looks_like_moderation_block_skips_transient():
    assert not _looks_like_moderation_block(RuntimeError("connection reset"))
    assert not _looks_like_moderation_block(RuntimeError("rate limit"))


# ───────────────────────── render_node_image ─────────────────────────


def test_render_node_image_happy_path_with_parent_ref(tmp_path):
    parent_png = tmp_path / "parent.png"
    parent_png.write_bytes(b"parent")
    out = tmp_path / "node.png"

    client = MagicMock()
    client.images.edit.return_value = _png_response(b"OK")

    info = render_node_image(
        openai_client=client, image_model="gpt-image-2.5-sunburst",
        prompt="photoreal still", out_path=out,
        ref_path=parent_png, sanitizer=None,
    )

    assert info["status"] == "ok"
    assert info["attempts"] == 1
    assert info["strategies"] == []
    assert out.exists()
    assert out.read_bytes() == b"OK"
    client.images.edit.assert_called_once()
    client.images.generate.assert_not_called()


def test_render_node_image_text_only_when_no_ref(tmp_path):
    out = tmp_path / "node.png"
    client = MagicMock()
    client.images.generate.return_value = _png_response(b"GEN")

    info = render_node_image(
        openai_client=client, image_model="gpt-image-2.5-sunburst",
        prompt="photoreal still", out_path=out,
        ref_path=None, sanitizer=None,
    )

    assert info["status"] == "ok"
    assert info["ref_used"] == "text_only"
    client.images.generate.assert_called_once()
    client.images.edit.assert_not_called()


def test_render_node_image_moderation_retry_with_sanitize(tmp_path):
    parent_png = tmp_path / "parent.png"; parent_png.write_bytes(b"P")
    out = tmp_path / "node.png"

    client = MagicMock()
    client.images.edit.side_effect = [
        _moderation_error(),       # attempt 1: moderation block
        _png_response(b"AFTER_SANITIZE"),  # attempt 2: success after sanitize
    ]

    sanitizer = MagicMock()
    sanitizer.sanitize.return_value = {
        "sanitized_prompt": "FILMVIS prefix + sanitized",
        "strategy": "film_previs",
        "changes": "removed gore",
    }

    info = render_node_image(
        openai_client=client, image_model="gpt-image-2.5-sunburst",
        prompt="raw bloody scene", out_path=out,
        ref_path=parent_png, sanitizer=sanitizer,
    )

    assert info["status"] == "ok"
    assert info["attempts"] == 2
    assert info["strategies"] == ["film_previs"]
    assert out.read_bytes() == b"AFTER_SANITIZE"
    sanitizer.sanitize.assert_called_once()
    args, kwargs = sanitizer.sanitize.call_args
    assert kwargs["original_prompt"] == "raw bloody scene"
    assert kwargs["attempt"] == 1


def test_render_node_image_moderation_exhausts_attempts(tmp_path):
    parent_png = tmp_path / "parent.png"; parent_png.write_bytes(b"P")
    out = tmp_path / "node.png"

    client = MagicMock()
    client.images.edit.side_effect = [_moderation_error()] * 5  # 모두 차단

    sanitizer = MagicMock()
    sanitizer.sanitize.side_effect = [
        {"sanitized_prompt": "FILMVIS s1", "strategy": "film_previs", "changes": ""},
        {"sanitized_prompt": "POSTER s2", "strategy": "movie_poster", "changes": ""},
        {"sanitized_prompt": "AFTER s3", "strategy": "aftermath", "changes": ""},
    ]

    info = render_node_image(
        openai_client=client, image_model="gpt-image-2.5-sunburst",
        prompt="bad", out_path=out,
        ref_path=parent_png, sanitizer=sanitizer, max_attempts=4,
    )

    assert info["status"] == "failed"
    assert info["attempts"] == 4
    assert info["strategies"] == ["film_previs", "movie_poster", "aftermath"]
    assert info["final_block_reason"]
    assert not out.exists()


def test_render_node_image_ref_path_missing_falls_back_to_text_only(tmp_path):
    """ref_path가 None이 아닌데 file이 없으면 text_only로 fallback하고
    ref_used도 text_only로 라벨링되어야 한다 (Claude PR #4 Issue 1 회귀 가드)."""
    missing = tmp_path / "vanished.png"  # 파일 만들지 않음
    out = tmp_path / "node.png"

    client = MagicMock()
    client.images.generate.return_value = _png_response(b"GEN")

    info = render_node_image(
        openai_client=client, image_model="gpt-image-2.5-sunburst",
        prompt="x", out_path=out,
        ref_path=missing, sanitizer=None,
    )

    assert info["status"] == "ok"
    assert info["ref_used"] == "text_only"  # ref_path was set but file missing
    client.images.generate.assert_called_once()
    client.images.edit.assert_not_called()


def test_render_node_image_sanitize_prepends_background_only_reinforcer(tmp_path):
    """sanitize 결과 prompt에 BACKGROUND-ONLY prefix가 prepend되어 generic
    sanitizer의 인물/자세 prefix를 무력화 (Codex PR #4 HIGH 3 회귀 가드)."""
    parent_png = tmp_path / "parent.png"; parent_png.write_bytes(b"P")
    out = tmp_path / "node.png"

    captured: dict = {}
    edit_call_count = {"n": 0}

    def _edit_side_effect(**kwargs):
        edit_call_count["n"] += 1
        if edit_call_count["n"] == 1:
            raise _moderation_error()
        captured["second_prompt"] = kwargs.get("prompt", "")
        return _png_response(b"OK")

    client = MagicMock()
    client.images.edit.side_effect = _edit_side_effect

    sanitizer = MagicMock()
    sanitizer.sanitize.return_value = {
        "sanitized_prompt": (
            "Movie poster key visual — characters faces dramatic poses..."
        ),
        "strategy": "movie_poster",
        "changes": "",
    }

    info = render_node_image(
        openai_client=client, image_model="gpt-image-2.5-sunburst",
        prompt="raw scene", out_path=out,
        ref_path=parent_png, sanitizer=sanitizer,
    )

    assert info["status"] == "ok"
    assert "BACKGROUND-ONLY" in captured["second_prompt"]
    # generic sanitizer prefix는 BACKGROUND-ONLY 뒤에 와야 함
    bg_idx = captured["second_prompt"].index("BACKGROUND-ONLY")
    poster_idx = captured["second_prompt"].index("Movie poster")
    assert bg_idx < poster_idx


def test_render_one_location_rejects_path_traversal_node_id(monkeypatch, tmp_path):
    """node_id가 ../ 등을 포함하면 렌더 안 하고 rejected_node_id로 마킹
    (Codex PR #4 HIGH 1 회귀 가드)."""
    _patch_prompt_gen(monkeypatch)

    plan = {
        "location_id": "L01", "location_name": "x",
        "rationale_summary": "x",
        "nodes": [{
            "id": "../escape_attempt",  # path traversal 시도
            "kind": "anchor_root",
            "label": "x", "description": "x",
            "shot_ids": ["S01_Shot1"], "parent_id": "",
            "depth": 0, "rationale": "x",
            "shared_visual_anchors_with_parent": [],
        }],
        "execution_order": ["../escape_attempt"],
        "unassigned_shots": [],
    }

    client = MagicMock()  # 호출되어선 안 됨

    out = render_one_location(
        location_id="L01",
        location_data=plan,
        image_dir=tmp_path,
        location_ref_paths={},
        openai_client=client,
        image_model="gpt-image-2.5-sunburst",
        sanitizer=None,
    )

    assert out["nodes"][0]["render_status"] == "rejected_node_id"
    assert out["failed_node_count"] == 1
    client.images.edit.assert_not_called()
    client.images.generate.assert_not_called()


def test_render_one_location_rejects_uppercase_or_special_node_id(monkeypatch, tmp_path):
    """대문자/하이픈/공백 등 snake_case ASCII 외 문자는 reject."""
    _patch_prompt_gen(monkeypatch)

    plan = {
        "location_id": "L01", "location_name": "x", "rationale_summary": "x",
        "nodes": [{
            "id": "Interior-Main Room",  # 대문자 + 하이픈 + 공백
            "kind": "anchor_root", "label": "x", "description": "x",
            "shot_ids": ["S01_Shot1"], "parent_id": "",
            "depth": 0, "rationale": "x",
            "shared_visual_anchors_with_parent": [],
        }],
        "execution_order": ["Interior-Main Room"],
        "unassigned_shots": [],
    }

    out = render_one_location(
        location_id="L01", location_data=plan,
        image_dir=tmp_path, location_ref_paths={},
        openai_client=MagicMock(), image_model="gpt-image-2.5-sunburst", sanitizer=None,
    )

    assert out["nodes"][0]["render_status"] == "rejected_node_id"


def test_render_node_image_no_sanitizer_returns_failed_on_block(tmp_path):
    parent_png = tmp_path / "parent.png"; parent_png.write_bytes(b"P")
    out = tmp_path / "node.png"

    client = MagicMock()
    client.images.edit.side_effect = [_moderation_error()]

    info = render_node_image(
        openai_client=client, image_model="gpt-image-2.5-sunburst",
        prompt="bad", out_path=out,
        ref_path=parent_png, sanitizer=None, max_attempts=4,
    )

    assert info["status"] == "failed"
    assert info["attempts"] == 1
    assert info["strategies"] == []
    assert info["final_block_reason"]


# ───────────────────────── render_one_location ─────────────────────────


def _planning_data_two_node_chain():
    return {
        "location_id": "L01",
        "location_name": "anonymous_room",
        "rationale_summary": "single location two-state chain.",
        "nodes": [
            {
                "id": "interior_main_room_day",
                "kind": "anchor_root",
                "label": "main room day wide",
                "description": "wide of the main room with side window light.",
                "shot_ids": ["S01_Shot1"],
                "parent_id": "",
                "depth": 0,
                "rationale": "anchor.",
                "shared_visual_anchors_with_parent": [],
            },
            {
                "id": "interior_main_room_night",
                "kind": "anchor_state",
                "label": "main room night wide",
                "description": "same room at night, single ceiling lamp.",
                "shot_ids": ["S02_Shot3"],
                "parent_id": "interior_main_room_day",
                "depth": 1,
                "rationale": "child of day anchor at night.",
                "shared_visual_anchors_with_parent": ["wall finish", "wood floor"],
            },
        ],
        "execution_order": [
            "interior_main_room_day",
            "interior_main_room_night",
        ],
        "unassigned_shots": [],
    }


def _patch_prompt_gen(monkeypatch, t2i_text="A photoreal still."):
    """generate_node_prompt를 stub해서 LLM 호출 우회. Tuple[str, List] 반환."""
    def _fake(*args, **kwargs):
        return t2i_text, []
    import app.modules.pipeline.background_chain_render as mod
    monkeypatch.setattr(mod, "generate_node_prompt", _fake)


def test_render_one_location_uses_location_ref_for_root_then_parent_for_child(
    monkeypatch, tmp_path,
):
    _patch_prompt_gen(monkeypatch)

    loc_ref = tmp_path / "loc_ref.png"
    loc_ref.write_bytes(b"LOC_REF")

    client = MagicMock()
    client.images.edit.side_effect = [
        _png_response(b"ROOT_PNG"),   # root anchor (uses loc ref)
        _png_response(b"CHILD_PNG"),  # child (uses parent png as ref)
    ]

    out = render_one_location(
        location_id="L01",
        location_data=_planning_data_two_node_chain(),
        image_dir=tmp_path,
        location_ref_paths={"L01": loc_ref},
        openai_client=client,
        image_model="gpt-image-2.5-sunburst",
        sanitizer=None,
    )

    assert out["failed_node_count"] == 0
    assert out["total_node_count"] == 2

    # root 파일 작성됨
    root_node = out["nodes"][0]
    assert root_node["render_status"] == "ok"
    assert root_node["ref_used"] == "location"
    root_path = Path(root_node["image_path"])
    assert root_path.exists() and root_path.read_bytes() == b"ROOT_PNG"

    # child는 root_path를 ref로 사용
    child_node = out["nodes"][1]
    assert child_node["render_status"] == "ok"
    assert child_node["ref_used"] == "parent"
    # 두 번째 edit 호출에서 image=open(parent_png) 형태로 전달됐는지
    # MagicMock이라 정확한 파일 객체 매칭은 어렵지만, edit 두 번 호출 확인
    assert client.images.edit.call_count == 2

    # shot_backgrounds 매핑
    assert {sb["shot_label"] for sb in out["shot_backgrounds"]} == {
        "S01_Shot1", "S02_Shot3",
    }


def test_render_one_location_root_fallback_to_text_only_when_no_loc_ref(
    monkeypatch, tmp_path,
):
    """location ref가 없으면 root는 images.generate (text-only)로 fallback."""
    _patch_prompt_gen(monkeypatch)

    client = MagicMock()
    client.images.generate.return_value = _png_response(b"TEXT_ROOT")
    client.images.edit.return_value = _png_response(b"CHILD")

    out = render_one_location(
        location_id="L01",
        location_data=_planning_data_two_node_chain(),
        image_dir=tmp_path,
        location_ref_paths={},  # 비어있음
        openai_client=client,
        image_model="gpt-image-2.5-sunburst",
        sanitizer=None,
    )

    assert out["nodes"][0]["ref_used"] == "text_only"
    client.images.generate.assert_called_once()
    # child는 부모 png ref → edit
    assert out["nodes"][1]["ref_used"] == "parent"


def test_render_one_location_root_failure_blocks_child_using_location_fallback(
    monkeypatch, tmp_path,
):
    """root가 실패하면 child의 ref_used는 location/text-only로 fallback (parent PNG 없음)."""
    _patch_prompt_gen(monkeypatch)

    loc_ref = tmp_path / "loc.png"; loc_ref.write_bytes(b"LR")

    client = MagicMock()
    # root: 모두 차단, child: 성공
    client.images.edit.side_effect = [
        _moderation_error(),  # root attempt 1
        _png_response(b"CHILD_OK"),  # child attempt 1 (uses location ref)
    ]

    out = render_one_location(
        location_id="L01",
        location_data=_planning_data_two_node_chain(),
        image_dir=tmp_path,
        location_ref_paths={"L01": loc_ref},
        openai_client=client,
        image_model="gpt-image-2.5-sunburst",
        sanitizer=None,  # 차단 → 즉시 실패
    )

    root, child = out["nodes"]
    assert root["render_status"] == "failed"
    assert child["render_status"] == "ok"
    assert child["ref_used"] == "location"  # 부모 PNG 없으므로 location fallback


def test_render_one_location_prompt_failure_marked_and_skips_image_call(
    monkeypatch, tmp_path,
):
    """generate_node_prompt가 raise하면 image 호출 없이 prompt_failed로 기록."""
    import app.modules.pipeline.background_chain_render as mod

    call_count = {"n": 0}

    def _fake_prompt(*a, **kw):
        call_count["n"] += 1
        if call_count["n"] == 1:
            raise ValueError("non-ASCII detected in t2i_prompt")
        return "ok prompt", []

    monkeypatch.setattr(mod, "generate_node_prompt", _fake_prompt)

    loc_ref = tmp_path / "lr.png"; loc_ref.write_bytes(b"L")
    client = MagicMock()
    client.images.edit.return_value = _png_response(b"CHILD")

    out = render_one_location(
        location_id="L01",
        location_data=_planning_data_two_node_chain(),
        image_dir=tmp_path,
        location_ref_paths={"L01": loc_ref},
        openai_client=client,
        image_model="gpt-image-2.5-sunburst",
        sanitizer=None,
    )

    root, child = out["nodes"]
    assert root["render_status"] == "prompt_failed"
    assert root["render_error"]
    # child는 정상 (parent_id가 root지만 root rendered_paths에 없으므로 location fallback)
    assert child["render_status"] == "ok"
    assert child["ref_used"] == "location"


# ───────────────────────── run_background_chain_render ─────────────────────────


def test_run_background_chain_render_empty_returns_empty(monkeypatch, tmp_path):
    out = run_background_chain_render(
        planning_data={"locations": {}},
        image_dir=tmp_path,
        location_ref_paths={},
        openai_client=MagicMock(),
        sanitizer=MagicMock(),
    )
    assert out == {"locations": {}, "_failed_count": 0}


def test_run_background_chain_render_propagates_failed_count(monkeypatch, tmp_path):
    _patch_prompt_gen(monkeypatch)

    client = MagicMock()
    # 두 location 둘 다 root 차단 (sanitizer X → 즉시 실패)
    client.images.edit.side_effect = _moderation_error()

    planning = {"locations": {
        "L01": _planning_data_two_node_chain(),
        "L02": _planning_data_two_node_chain(),
    }}

    out = run_background_chain_render(
        planning_data=planning,
        image_dir=tmp_path,
        location_ref_paths={},
        openai_client=client,
        sanitizer=None,
        max_workers=1,  # 결정적 순서
    )

    # 각 location 2 노드 모두 실패 → _failed_count = 4
    assert out["_failed_count"] == 4
    assert set(out["locations"].keys()) == {"L01", "L02"}


# ───────────────────── Phase 5.3 — _compute_chain_bg_levels DAG 분할 ─────────────────────


def test_compute_levels_all_roots_single_level():
    """parent_id 없는 root 그룹들만 → 1 level에 모두."""
    from app.modules.pipeline.background_chain_render import _compute_chain_bg_levels

    order = ["g1", "g2", "g3"]
    groups = {gid: {"parent_id": ""} for gid in order}
    levels = _compute_chain_bg_levels(order, groups, set(order))
    assert levels == [["g1", "g2", "g3"]]


def test_compute_levels_linear_chain():
    """g1 → g2 → g3 → g4 (parent 체인) → 4 sequential levels."""
    from app.modules.pipeline.background_chain_render import _compute_chain_bg_levels

    order = ["g1", "g2", "g3", "g4"]
    groups = {
        "g1": {"parent_id": ""},
        "g2": {"parent_id": "g1"},
        "g3": {"parent_id": "g2"},
        "g4": {"parent_id": "g3"},
    }
    levels = _compute_chain_bg_levels(order, groups, set(order))
    assert levels == [["g1"], ["g2"], ["g3"], ["g4"]]


def test_compute_levels_fanout_then_merge():
    """g1 → {g2, g3, g4} (모두 g1의 자식) → 2 levels (root + fanout)."""
    from app.modules.pipeline.background_chain_render import _compute_chain_bg_levels

    order = ["g1", "g2", "g3", "g4"]
    groups = {
        "g1": {"parent_id": ""},
        "g2": {"parent_id": "g1"},
        "g3": {"parent_id": "g1"},
        "g4": {"parent_id": "g1"},
    }
    levels = _compute_chain_bg_levels(order, groups, set(order))
    assert levels == [["g1"], ["g2", "g3", "g4"]]


def test_compute_levels_real_world_shape():
    """실측 PID c00bbe19와 유사한 shape: 10 root + 4 L1 + 2 L2 + 1 L3 + 1 L4."""
    from app.modules.pipeline.background_chain_render import _compute_chain_bg_levels

    order = (
        ["r1", "r2", "r3", "r4", "r5", "r6", "r7", "r8", "r9", "r10"]
        + ["c1a", "c1b", "c5a", "c5b"]
        + ["c5aa", "c1bb"]
        + ["c5aaa"]
        + ["c5aaaa"]
    )
    groups = {gid: {"parent_id": ""} for gid in ["r1", "r2", "r3", "r4", "r5", "r6", "r7", "r8", "r9", "r10"]}
    groups["c1a"] = {"parent_id": "r1"}
    groups["c1b"] = {"parent_id": "r1"}
    groups["c5a"] = {"parent_id": "r5"}
    groups["c5b"] = {"parent_id": "r5"}
    groups["c5aa"] = {"parent_id": "c5a"}
    groups["c1bb"] = {"parent_id": "c1b"}
    groups["c5aaa"] = {"parent_id": "c5aa"}
    groups["c5aaaa"] = {"parent_id": "c5aaa"}

    levels = _compute_chain_bg_levels(order, groups, set(order))
    # 5 levels: 10 roots → 4 L1 → 2 L2 → 1 L3 → 1 L4
    assert len(levels) == 5
    assert sorted(levels[0]) == sorted(["r1", "r2", "r3", "r4", "r5", "r6", "r7", "r8", "r9", "r10"])
    assert sorted(levels[1]) == sorted(["c1a", "c1b", "c5a", "c5b"])
    assert sorted(levels[2]) == sorted(["c5aa", "c1bb"])
    assert levels[3] == ["c5aaa"]
    assert levels[4] == ["c5aaaa"]


def test_compute_levels_parent_outside_renderable_treated_as_root():
    """parent_id가 renderable_set에 없으면 (skipped 등) root처럼 처리."""
    from app.modules.pipeline.background_chain_render import _compute_chain_bg_levels

    order = ["g1", "g2", "g3"]
    groups = {
        "g1": {"parent_id": ""},
        "g2": {"parent_id": "skipped_planning_group"},  # parent not renderable
        "g3": {"parent_id": "g2"},
    }
    renderable = {"g1", "g2", "g3"}  # parent "skipped_planning_group" 부재
    levels = _compute_chain_bg_levels(order, groups, renderable)
    # g1 + g2 모두 root처럼 처리 (g2의 parent가 renderable에 없음) → L0에 같이
    assert sorted(levels[0]) == ["g1", "g2"]
    assert levels[1] == ["g3"]


def test_compute_levels_empty_input():
    """빈 입력 → 빈 levels."""
    from app.modules.pipeline.background_chain_render import _compute_chain_bg_levels

    assert _compute_chain_bg_levels([], {}, set()) == []


def test_compute_levels_cycle_flushed_as_final_batch():
    """cycle (planner invariant 위반) → warning + 잔여 그룹을 마지막 level로 flush.

    이 시나리오는 planner가 막아야 하지만 안전망 검증.
    """
    from app.modules.pipeline.background_chain_render import _compute_chain_bg_levels

    order = ["g1", "g2"]
    groups = {
        "g1": {"parent_id": "g2"},  # cycle
        "g2": {"parent_id": "g1"},  # cycle
    }
    levels = _compute_chain_bg_levels(order, groups, {"g1", "g2"})
    # cycle → 한꺼번에 flush
    assert len(levels) == 1
    assert sorted(levels[0]) == ["g1", "g2"]


# ───────────────────── Phase 5.3 — _resolve_workers env handling ─────────────────────


def test_resolve_workers_default_when_unset(monkeypatch):
    from app.modules.pipeline.background_chain_render import _resolve_workers
    monkeypatch.delenv("BACKGROUND_CHAIN_RENDER_WORKERS", raising=False)
    assert _resolve_workers(default=4) == 4


def test_resolve_workers_clamps_to_cap(monkeypatch):
    from app.modules.pipeline.background_chain_render import _resolve_workers
    monkeypatch.setenv("BACKGROUND_CHAIN_RENDER_WORKERS", "100")
    assert _resolve_workers(default=4, cap=8) == 8


def test_resolve_workers_clamps_to_min(monkeypatch):
    from app.modules.pipeline.background_chain_render import _resolve_workers
    monkeypatch.setenv("BACKGROUND_CHAIN_RENDER_WORKERS", "0")
    assert _resolve_workers() == 1


def test_resolve_workers_invalid_falls_back_to_default(monkeypatch, caplog):
    import logging
    from app.modules.pipeline.background_chain_render import _resolve_workers
    monkeypatch.setenv("BACKGROUND_CHAIN_RENDER_WORKERS", "abc")
    with caplog.at_level(logging.WARNING):
        assert _resolve_workers(default=4) == 4
    assert any("not int" in rec.getMessage() for rec in caplog.records)


# ───────────────────── Phase 5.3 — 병렬 렌더 통합 (mock LLM/image) ─────────────────────


def test_planner_render_parallel_preserves_variant_order(monkeypatch, tmp_path):
    """deterministic variant_index — 병렬 실행에도 chain_bg_order 순서대로 v01/v02/v03."""
    from app.modules.pipeline.background_chain_render import _run_planner_driven_render

    # 같은 location L05의 group 3개. 모두 root → 1 level에 병렬 실행됨.
    order = ["g_first", "g_second", "g_third"]
    groups = {
        "g_first": {
            "status": "ok",
            "location_id": "L05",
            "location_name": "Living",
            "parent_id": "",
            "rationale_summary": "first",
            "nodes": [{"id": "n1", "kind": "anchor_root", "shot_ids": ["S1_Shot1"], "parent_id": "", "depth": 0, "shared_visual_anchors_with_parent": []}],
        },
        "g_second": {
            "status": "ok",
            "location_id": "L05",
            "location_name": "Living",
            "parent_id": "",
            "rationale_summary": "second",
            "nodes": [{"id": "n2", "kind": "anchor_root", "shot_ids": ["S2_Shot1"], "parent_id": "", "depth": 0, "shared_visual_anchors_with_parent": []}],
        },
        "g_third": {
            "status": "ok",
            "location_id": "L05",
            "location_name": "Living",
            "parent_id": "",
            "rationale_summary": "third",
            "nodes": [{"id": "n3", "kind": "anchor_root", "shot_ids": ["S3_Shot1"], "parent_id": "", "depth": 0, "shared_visual_anchors_with_parent": []}],
        },
    }

    # 이미지 client mock — 항상 ok
    fake_b64 = base64.b64encode(b"PNG").decode("ascii")
    client = MagicMock()
    item = MagicMock()
    item.b64_json = fake_b64
    resp = MagicMock()
    resp.data = [item]
    client.images.generate.return_value = resp
    client.images.edit.return_value = resp

    with patch(
        "app.modules.pipeline.background_chain_render.generate_node_prompt",
        return_value=("photo prompt", [{"shot_id": "S1_Shot1", "guide": "g"}]),
    ):
        out = _run_planner_driven_render(
            planner_chain_order=order,
            planning_groups=groups,
            floor_plan_specs={},
            floor_plan_paths={},
            image_dir=tmp_path,
            location_ref_paths={},
            openai_client=client,
            image_model="gpt-image-2.5-sunburst",
            sanitizer=None,
            size="1024x1024",
            quality="high",
            max_attempts=2,
            opik_metadata=None,
            shot_meta_by_id=None,
        )

    res = out["data"]["groups"]
    # variant_index: chain_bg_order대로 v01, v02, v03 (병렬에도 deterministic)
    assert res["g_first"]["variant_label"] == "v01"
    assert res["g_second"]["variant_label"] == "v02"
    assert res["g_third"]["variant_label"] == "v03"
    assert all(r["status"] == "ok" for r in res.values())
    assert out["_failed_count"] == 0


def test_planner_render_parallel_respects_dependency_order(monkeypatch, tmp_path):
    """parent group이 child보다 먼저 렌더 — child가 parent PNG를 ref로 받는지 검증."""
    from app.modules.pipeline.background_chain_render import _run_planner_driven_render

    # parent → child (linear chain)
    order = ["parent_g", "child_g"]
    groups = {
        "parent_g": {
            "status": "ok",
            "location_id": "L01",
            "location_name": "Hall",
            "parent_id": "",
            "rationale_summary": "parent",
            "nodes": [{"id": "p", "kind": "anchor_root", "shot_ids": ["S1_Shot1"], "parent_id": "", "depth": 0, "shared_visual_anchors_with_parent": []}],
        },
        "child_g": {
            "status": "ok",
            "location_id": "L01",
            "location_name": "Hall",
            "parent_id": "parent_g",
            "rationale_summary": "child",
            "nodes": [{"id": "c", "kind": "anchor_root", "shot_ids": ["S2_Shot1"], "parent_id": "", "depth": 0, "shared_visual_anchors_with_parent": []}],
        },
    }

    # ref_paths를 캡처 — parent 렌더 시 0개, child 렌더 시 parent PNG 포함
    captured_refs: List[List[Path]] = []

    def fake_render(*, ref_paths, out_path, **kw):
        captured_refs.append(list(ref_paths))
        out_path.write_bytes(b"PNG")  # parent 렌더 후 child가 ref로 사용 가능하게
        return {"status": "ok", "attempts": 1, "strategies": []}

    monkeypatch.setattr(
        "app.modules.pipeline.background_chain_render.render_node_image",
        fake_render,
    )

    with patch(
        "app.modules.pipeline.background_chain_render.generate_node_prompt",
        return_value=("photo prompt", []),
    ):
        out = _run_planner_driven_render(
            planner_chain_order=order,
            planning_groups=groups,
            floor_plan_specs={},
            floor_plan_paths={},
            image_dir=tmp_path,
            location_ref_paths={},
            openai_client=MagicMock(),
            image_model="gpt-image-2.5-sunburst",
            sanitizer=None,
            size="1024x1024",
            quality="high",
            max_attempts=2,
            opik_metadata=None,
            shot_meta_by_id=None,
        )

    res = out["data"]["groups"]
    # 순서: parent_g (level 0) → child_g (level 1)
    # captured_refs[0] = parent 렌더 시 ref_paths (없음)
    # captured_refs[1] = child 렌더 시 ref_paths (parent PNG 포함)
    assert len(captured_refs) == 2
    assert len(captured_refs[0]) == 0  # parent — no floor_plan/parent/loc_ref
    assert len(captured_refs[1]) == 1  # child — parent PNG ref 포함
    assert "L01_v01.png" in str(captured_refs[1][0])  # parent의 PNG
    assert res["child_g"]["ref_used"] == "parent_group"


def test_planner_render_empty_loc_id_records_skipped_no_loc(monkeypatch, tmp_path):
    """Claude review I2: loc_id 비어있는 group도 results에 'skipped_no_loc'으로 기록."""
    from app.modules.pipeline.background_chain_render import _run_planner_driven_render

    order = ["g_no_loc", "g_ok"]
    groups = {
        "g_no_loc": {
            "status": "ok",
            "location_id": "",  # empty
            "location_name": "Phantom",
            "parent_id": "",
            "rationale_summary": "missing loc",
        },
        "g_ok": {
            "status": "ok",
            "location_id": "L01",
            "location_name": "Hall",
            "parent_id": "",
            "rationale_summary": "ok",
            "nodes": [{"id": "n1", "kind": "anchor_root", "shot_ids": ["S1_Shot1"], "parent_id": "", "depth": 0, "shared_visual_anchors_with_parent": []}],
        },
    }

    monkeypatch.setattr(
        "app.modules.pipeline.background_chain_render.render_node_image",
        lambda **kw: (kw["out_path"].write_bytes(b"PNG"),
                      {"status": "ok", "attempts": 1, "strategies": []})[1],
    )

    with patch(
        "app.modules.pipeline.background_chain_render.generate_node_prompt",
        return_value=("photo prompt", []),
    ):
        out = _run_planner_driven_render(
            planner_chain_order=order,
            planning_groups=groups,
            floor_plan_specs={},
            floor_plan_paths={},
            image_dir=tmp_path,
            location_ref_paths={},
            openai_client=MagicMock(),
            image_model="gpt-image-2.5-sunburst",
            sanitizer=None,
            size="1024x1024",
            quality="high",
            max_attempts=2,
            opik_metadata=None,
            shot_meta_by_id=None,
        )

    res = out["data"]["groups"]
    # g_no_loc도 results에 명시적 기록 (I2 fix)
    assert "g_no_loc" in res
    assert res["g_no_loc"]["status"] == "skipped_no_loc"
    assert res["g_no_loc"]["skipped_reason"] == "empty location_id"
    # g_ok는 정상 처리
    assert res["g_ok"]["status"] == "ok"


def test_planner_render_worker_exception_captured(monkeypatch, tmp_path):
    """Claude review I1: worker thread exception이 future.result()에서 전파되어도
    같은 level의 나머지 group이 누락되지 않고 'failed' 상태로 기록된다."""
    from app.modules.pipeline.background_chain_render import _run_planner_driven_render

    order = ["g_boom", "g_ok"]
    groups = {
        "g_boom": {
            "status": "ok",
            "location_id": "L01",
            "location_name": "BoomRoom",
            "parent_id": "",
            "rationale_summary": "boom",
            "nodes": [{"id": "n1", "kind": "anchor_root", "shot_ids": ["S1_Shot1"], "parent_id": "", "depth": 0, "shared_visual_anchors_with_parent": []}],
        },
        "g_ok": {
            "status": "ok",
            "location_id": "L02",
            "location_name": "OkRoom",
            "parent_id": "",
            "rationale_summary": "ok",
            "nodes": [{"id": "n2", "kind": "anchor_root", "shot_ids": ["S2_Shot1"], "parent_id": "", "depth": 0, "shared_visual_anchors_with_parent": []}],
        },
    }

    def flaky_render(**kw):
        # g_boom용 PNG path가 들어오면 raise — 다른 group은 정상.
        if "L01_v01.png" in str(kw["out_path"]):
            raise RuntimeError("disk write OOPS")
        kw["out_path"].write_bytes(b"PNG")
        return {"status": "ok", "attempts": 1, "strategies": []}

    monkeypatch.setattr(
        "app.modules.pipeline.background_chain_render.render_node_image",
        flaky_render,
    )

    with patch(
        "app.modules.pipeline.background_chain_render.generate_node_prompt",
        return_value=("photo prompt", []),
    ):
        out = _run_planner_driven_render(
            planner_chain_order=order,
            planning_groups=groups,
            floor_plan_specs={},
            floor_plan_paths={},
            image_dir=tmp_path,
            location_ref_paths={},
            openai_client=MagicMock(),
            image_model="gpt-image-2.5-sunburst",
            sanitizer=None,
            size="1024x1024",
            quality="high",
            max_attempts=2,
            opik_metadata=None,
            shot_meta_by_id=None,
        )

    res = out["data"]["groups"]
    # 두 group 모두 results에 기록 (worker exception이 전파돼도 다른 그룹 누락 X)
    assert set(res.keys()) == {"g_boom", "g_ok"}
    assert res["g_boom"]["status"] == "failed"
    assert "OOPS" in res["g_boom"]["render_error"]
    assert res["g_ok"]["status"] == "ok"
    assert out["_failed_count"] == 1


def test_planner_render_groups_dict_preserves_planner_chain_order(monkeypatch, tmp_path):
    """Codex review L1: 같은 level의 병렬 렌더 후 data.groups dict 삽입 순서가
    planner_chain_order대로 재구성되어 deterministic."""
    import time as _time
    from app.modules.pipeline.background_chain_render import _run_planner_driven_render

    # 5개 root group — 모두 같은 level. 일부러 첫 group이 가장 느리게 끝나도록.
    order = ["g_a", "g_b", "g_c", "g_d", "g_e"]
    groups = {
        gid: {
            "status": "ok",
            "location_id": f"L0{i+1}",
            "location_name": f"Loc{i}",
            "parent_id": "",
            "rationale_summary": gid,
            "nodes": [{"id": f"n{i}", "kind": "anchor_root", "shot_ids": [f"S{i}_Shot1"], "parent_id": "", "depth": 0, "shared_visual_anchors_with_parent": []}],
        }
        for i, gid in enumerate(order)
    }

    # g_a는 0.1s 지연, 나머지는 즉시 → completion order는 b,c,d,e,a
    def staggered_render(**kw):
        out_path = kw["out_path"]
        if "L01_v01.png" in str(out_path):
            _time.sleep(0.1)
        out_path.write_bytes(b"PNG")
        return {"status": "ok", "attempts": 1, "strategies": []}

    monkeypatch.setattr(
        "app.modules.pipeline.background_chain_render.render_node_image",
        staggered_render,
    )

    with patch(
        "app.modules.pipeline.background_chain_render.generate_node_prompt",
        return_value=("photo prompt", []),
    ):
        out = _run_planner_driven_render(
            planner_chain_order=order,
            planning_groups=groups,
            floor_plan_specs={},
            floor_plan_paths={},
            image_dir=tmp_path,
            location_ref_paths={},
            openai_client=MagicMock(),
            image_model="gpt-image-2.5-sunburst",
            sanitizer=None,
            size="1024x1024",
            quality="high",
            max_attempts=2,
            opik_metadata=None,
            shot_meta_by_id=None,
        )

    # data.groups dict insertion order = planner_chain_order (completion order 아님)
    assert list(out["data"]["groups"].keys()) == order
    assert list(out["groups"].keys()) == order