"""W20C / W20E7-A: BackgroundRenderStep ``shot_aware_plan`` selector
branch + ``w18j_overlap`` deprecation tests.

production-side, focused. LLM / image / VLM API call 0 (render_one_background
mocked, _resolve_openai_client / _register_image_assets / db patched). DB /
ImageAsset write 0.

Covers:
- selector branch wires to ``_run_shot_aware_plan_queue`` (legacy and
  w18j_overlap paths NOT invoked).
- ``max_attempts=1`` is forwarded to ``render_one_background``.
- node_index ordering is honoured.
- catalog grows so the next node's catalog ref resolves.
- missing plan / not production_clear / ref unresolved → image API
  not called, entry marked failed.
- legacy path remains unaffected by the new branch.
- ``w18j_overlap`` selector value is W20E7-A deprecated — branch returns
  every renderable bg as ``failed`` with a stable deprecation error and
  never imports / calls ``background_image_planner`` or
  ``_run_sequential_overlap_queue`` (the helper itself is deleted).
- W19B-3 mutex unchanged (only triggers on w18j_overlap+W20B both on).
- ``_config_hash`` stamps the shot_aware_plan selector value (opt-in
  only — legacy payload stays byte-identical).
"""
from __future__ import annotations

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

import pytest


def _touch_png(p: Path) -> Path:
    p.parent.mkdir(parents=True, exist_ok=True)
    p.write_bytes(b"\x89PNG\r\n\x1a\n")
    return p


def _bg_spec(
    *, bg_id: str, fp_id: str = "fp_a", loc_id: str = "L01"
) -> Dict[str, Any]:
    return {
        "bg_id": bg_id,
        "loc_id": loc_id,
        "sub_location": "primary_unit",
        "state_label": "day",
        "applies_to_shots": [f"{bg_id}_Shot1"],
        "depends_on_fp": [fp_id],
        "depends_on_bg": [],
    }


_RENDER_GUIDANCE_FIELDS = (
    "visible_space_directive",
    "camera_framing_directive",
    "subject_position_directive",
    "state_cue_directive",
    "negative_continuity_directive",
)


def _render_guidance_for(bg_id: str) -> Dict[str, str]:
    return {
        f: f"directive {f} for {bg_id}" for f in _RENDER_GUIDANCE_FIELDS
    }


def _plan_node(
    *,
    bg_id: str,
    node_index: int,
    mode: str,
    selected_refs=None,
    is_anchor: bool = False,
    camera_lens: str = "normal",
    render_action: str = "render_new_plate",
    reuse_target_bg_id: str = "",
) -> Dict[str, Any]:
    if selected_refs is None:
        selected_refs = []
    psi_list = [
        r.get("physical_space_id", "") for r in selected_refs
    ]
    if mode == "two_refs_distinct_spaces":
        same_space = "distinct_visible_spaces"
        why = "co-visible distinct spaces"
    else:
        same_space = "single_ref"
        why = "single ref reason"
    return {
        "bg_id": bg_id,
        "node_index": node_index,
        "mode": mode,
        "is_dwelling_identity_anchor": is_anchor,
        "rationale": f"rationale for {bg_id}",
        # W21B-w3 Commit 2a router output (stamped into the plan cp).
        "render_action": render_action,
        "reuse_target_bg_id": reuse_target_bg_id,
        "reference_decision": {
            "selected_refs": list(selected_refs),
            "rejected_refs": [],
            "same_physical_space_dedup_decision": same_space,
            "why_single_ref_or_two_refs": why,
            "physical_space_id_per_ref": psi_list,
        },
        "camera_decision": {
            "camera_unit": 1,
            "camera_cell": [0, 0],
            "look_at_unit": 2,
            "look_at_cell": [2, 0],
            "lens_enum": camera_lens,
            "fov_deg": 50 if camera_lens == "normal" else 90,
            "framing_notes": f"framing for {bg_id}",
        },
        "render_guidance": _render_guidance_for(bg_id),
    }


def _plan_for_fp(
    *,
    fp_id: str,
    nodes: List[Dict[str, Any]],
    status: str = "ok",
    production_clear: bool = True,
) -> Dict[str, Any]:
    return {
        "fp_id": fp_id,
        "shot_aware_bg_render_plan_status": status,
        "graph": {"nodes": list(nodes)},
        "production_clear": production_clear,
        "real_api_call_counts": {"image": 0, "llm": 1, "vlm": 0},
        "readback_status": "ok",
        "validators": {"all_validators_passed": True, "diagnostics": []},
        "diagnostics": [],
    }


def _build_cp_map(
    *,
    tmp_path: Path,
    fp_id: str,
    bg_specs: List[Dict[str, Any]],
    plan: Dict[str, Any] = None,
) -> Dict[str, Any]:
    fp_png = tmp_path / "fp_render" / f"{fp_id}.png"
    _touch_png(fp_png)
    plans_data = {
        "data": {
            "plans": {
                "g_a": {
                    "status": "ok",
                    "plan": {
                        "group_id": "g_a",
                        "rationale_summary": "",
                        "floor_plans": [],
                        "backgrounds": bg_specs,
                        "gen_order": [b["bg_id"] for b in bg_specs],
                    },
                }
            },
            "bg_catalog_hash": "",
            "shot_binding_hash": "",
        }
    }
    fp_render_cp = {
        "data": {
            "floor_plans": {
                fp_id: {"status": "ok", "png_path": str(fp_png)}
            }
        }
    }
    prompts_cp = {
        "data": {
            "backgrounds": {
                b["bg_id"]: {
                    "status": "ok",
                    "t2i_prompt": f"prompt body for {b['bg_id']}",
                    "shot_guides": [],
                    "objects_owned_by_background": [],
                    "spec": b,
                    "group_id": "g_a",
                }
                for b in bg_specs
            }
        }
    }
    fp_prompt_cp = {
        "data": {
            "floor_plans": {
                fp_id: {
                    "status": "ok",
                    "numbered_elements": [],
                    "camera_recommendations": [],
                }
            }
        }
    }
    shot_aware_cp = (
        {"data": {"per_fp": {fp_id: plan}}}
        if plan is not None
        else None
    )
    return {
        "background_master_plan": plans_data,
        "background_prompt": prompts_cp,
        "floor_plan_render": fp_render_cp,
        "floor_plan_prompt": fp_prompt_cp,
        "shot_aware_bg_render_plan": shot_aware_cp,
        "floor_plan_overlay_payload": None,
    }


def _new_step(tmp_path: Path, cp_map: Dict[str, Any]):
    from app.core.steps.background_render_step import BackgroundRenderStep

    step = BackgroundRenderStep.__new__(BackgroundRenderStep)
    step.project_id = "p"
    step.episode_id = "e"
    step.project_config = {}
    step.build_opik_metadata = MagicMock(return_value={})
    step._load_prev_checkpoint = MagicMock(
        side_effect=lambda sid: cp_map.get(sid)
    )
    step.db = MagicMock()
    step._register_image_assets = MagicMock()
    return step


def _apply_settings(monkeypatch, tmp_path, *, mode="shot_aware_plan",
                    w20b_enabled=False):
    monkeypatch.setattr(
        "app.core.config.settings.projects_dir", str(tmp_path)
    )
    monkeypatch.setattr(
        "app.core.config.settings.openai_api_key", "sk-test"
    )
    monkeypatch.setattr(
        "app.core.config.settings.background_mode", "on"
    )
    monkeypatch.setattr(
        "app.core.config.settings.background_render_reference_mode", mode
    )
    monkeypatch.setattr(
        "app.core.config.settings.shot_aware_bg_render_plan_enabled",
        w20b_enabled,
    )


def _success_renderer():
    def _side(*, openai_client, image_model, prompt, out_path, fp_path,
              prior_bg_paths, bg_id, **kwargs):
        _touch_png(Path(out_path))
        return {
            "status": "ok",
            "attempts": 1,
            "strategies": [],
            "ref_used": (
                "fp_only" if fp_path is not None and not prior_bg_paths
                else f"refs_{len(prior_bg_paths)}"
                if prior_bg_paths else "text_only"
            ),
            "final_block_reason": None,
            "png_path": str(out_path),
        }
    return _side


# ─── test #1: selector routes to shot_aware_plan helper, NOT legacy/w18j ───


def test_selector_routes_to_shot_aware_plan_helper(tmp_path, monkeypatch):
    _apply_settings(monkeypatch, tmp_path)
    bg_specs = [_bg_spec(bg_id="L01B01")]
    plan = _plan_for_fp(
        fp_id="fp_a",
        nodes=[
            _plan_node(
                bg_id="L01B01", node_index=0,
                mode="fp_seeded_anchor", is_anchor=True,
            )
        ],
    )
    cp_map = _build_cp_map(
        tmp_path=tmp_path, fp_id="fp_a", bg_specs=bg_specs, plan=plan
    )
    step = _new_step(tmp_path, cp_map)

    with patch(
        "app.core.steps.background_render_step._resolve_openai_client",
        return_value=MagicMock(),
    ), patch(
        "app.modules.pipeline.background_render.render_one_background",
        side_effect=_success_renderer(),
    ), patch(
        "app.core.steps.background_render_step.ThreadPoolExecutor"
    ) as mock_pool:
        result = step._execute()

    # legacy ThreadPool not used. The W19B-3 ``_run_sequential_overlap_queue``
    # helper was deleted in W20E7-A — there is no longer a way to mock or
    # invoke it; this branch never reaches the deprecated w18j_overlap
    # fail-closed loop because ``shot_aware_plan`` is the configured mode.
    assert mock_pool.call_count == 0
    # Successful single-bg run.
    assert result["completed_count"] == 1
    assert result["data"]["groups"]["L01B01"]["status"] == "ok"
    # opt-in catalog key present.
    assert "bg_reference_catalog" in result["data"]
    assert (
        result["data"]["bg_reference_catalog"][0]["source_kind"]
        == "shot_aware_plan"
    )


# ─── test #2: max_attempts=1 forwarded to render_one_background ────────


def test_shot_aware_plan_forwards_max_attempts_two(tmp_path, monkeypatch):
    _apply_settings(monkeypatch, tmp_path)
    bg_specs = [_bg_spec(bg_id="L01B01")]
    plan = _plan_for_fp(
        fp_id="fp_a",
        nodes=[
            _plan_node(
                bg_id="L01B01", node_index=0,
                mode="fp_seeded_anchor", is_anchor=True,
            )
        ],
    )
    cp_map = _build_cp_map(
        tmp_path=tmp_path, fp_id="fp_a", bg_specs=bg_specs, plan=plan
    )
    step = _new_step(tmp_path, cp_map)

    captured_kwargs: Dict[str, Any] = {}

    def _capturing(*, openai_client, image_model, prompt, out_path,
                   fp_path, prior_bg_paths, bg_id, **kwargs):
        captured_kwargs.update(kwargs)
        _touch_png(Path(out_path))
        return {
            "status": "ok", "attempts": 1, "strategies": [],
            "ref_used": "fp_only", "final_block_reason": None,
            "png_path": str(out_path),
        }

    with patch(
        "app.core.steps.background_render_step._resolve_openai_client",
        return_value=MagicMock(),
    ), patch(
        "app.modules.pipeline.background_render.render_one_background",
        side_effect=_capturing,
    ):
        step._execute()

    # 2026-07-11: moderation-전용 sanitize 재시도 허용 (정상 경로 1콜 불변)
    assert captured_kwargs.get("max_attempts") == 2


# ─── P0-1: partial completion (some bgs ok, some failed) ──────────────


def test_bg_render_partial_reports_actual_ok_count(tmp_path, monkeypatch):
    """P0-1 (W21B-w6): bg 일부만 렌더 성공하면 completed_count 는 실제 ok
    rendered 수, applicable_count 는 renderable 전체 수 → step_runner 가
    'partial' 도출.

    이전엔 completed_count=1 if failed==0 else 0 이진화 → bg 1개 실패에
    status='failed' → analysis_dispatch 즉시 break → 정상 bg + 모든
    downstream 통째 stop. 완료 기준은 status=="ok" AND png_path
    (rendered_paths 등록 기준 = downstream scene_image_pipeline 이 참조
    가능한 PNG). DB sync raise 정책은 불변.
    """
    _apply_settings(monkeypatch, tmp_path)
    bg_specs = [_bg_spec(bg_id="L01B01"), _bg_spec(bg_id="L01B02")]
    plan = _plan_for_fp(
        fp_id="fp_a",
        nodes=[
            _plan_node(
                bg_id="L01B01", node_index=0,
                mode="fp_seeded_anchor", is_anchor=True,
            ),
            _plan_node(
                bg_id="L01B02", node_index=1,
                mode="fp_seeded_anchor", is_anchor=True,
            ),
        ],
    )
    cp_map = _build_cp_map(
        tmp_path=tmp_path, fp_id="fp_a", bg_specs=bg_specs, plan=plan
    )
    step = _new_step(tmp_path, cp_map)

    def _partial_renderer(*, openai_client, image_model, prompt, out_path,
                          fp_path, prior_bg_paths, bg_id, **kwargs):
        if bg_id == "L01B02":
            return {
                "status": "failed", "attempts": 1, "strategies": [],
                "ref_used": "text_only", "final_block_reason": "boom",
                "png_path": "",
            }
        _touch_png(Path(out_path))
        return {
            "status": "ok", "attempts": 1, "strategies": [],
            "ref_used": "fp_only", "final_block_reason": None,
            "png_path": str(out_path),
        }

    with patch(
        "app.core.steps.background_render_step._resolve_openai_client",
        return_value=MagicMock(),
    ), patch(
        "app.modules.pipeline.background_render.render_one_background",
        side_effect=_partial_renderer,
    ):
        result = step._execute()

    assert result["applicable_count"] == 2
    assert result["completed_count"] == 1  # 이전 이진화면 0
    assert result["failed_count"] == 1
    completed, total, failed = (
        result["completed_count"],
        result["applicable_count"],
        result["failed_count"],
    )
    final_status = (
        "completed" if failed == 0 else ("partial" if completed > 0 else "failed")
    )
    assert final_status == "partial"
    groups = result["data"]["groups"]
    assert groups["L01B01"]["status"] == "ok"
    assert groups["L01B02"]["status"] == "failed"


# ─── test #3: node_index ordering + catalog growth between bgs ─────────


def test_shot_aware_plan_orders_by_node_index_and_grows_catalog(
    tmp_path, monkeypatch,
):
    """Plan declares two nodes for the same fp.  node_index=0 is the
    anchor (no refs); node_index=1 is reference_derived against the
    anchor.  The anchor must render first so the derived's catalog
    lookup resolves.
    """
    _apply_settings(monkeypatch, tmp_path)
    bg_specs = [
        _bg_spec(bg_id="L01B01"),
        _bg_spec(bg_id="L01B02"),
    ]
    plan = _plan_for_fp(
        fp_id="fp_a",
        nodes=[
            _plan_node(
                bg_id="L01B02", node_index=1,
                mode="reference_derived",
                selected_refs=[{
                    "ref_bg_id": "L01B01",
                    "physical_space_id": "primary",
                    "space_description": None,
                }],
                camera_lens="wide",
            ),
            _plan_node(
                bg_id="L01B01", node_index=0,
                mode="fp_seeded_anchor", is_anchor=True,
            ),
        ],
    )
    cp_map = _build_cp_map(
        tmp_path=tmp_path, fp_id="fp_a", bg_specs=bg_specs, plan=plan
    )
    step = _new_step(tmp_path, cp_map)

    captured: List[Dict[str, Any]] = []

    def _capturing(*, openai_client, image_model, prompt, out_path,
                   fp_path, prior_bg_paths, bg_id, **kwargs):
        _touch_png(Path(out_path))
        captured.append({
            "bg_id": bg_id,
            "fp_path": str(fp_path) if fp_path is not None else None,
            "prior_bg_paths": [str(p) for p in prior_bg_paths],
            "prompt": prompt,
        })
        return {
            "status": "ok", "attempts": 1, "strategies": [],
            "ref_used": (
                "fp_only" if fp_path is not None and not prior_bg_paths
                else f"refs_{len(prior_bg_paths)}"
            ),
            "final_block_reason": None,
            "png_path": str(out_path),
        }

    with patch(
        "app.core.steps.background_render_step._resolve_openai_client",
        return_value=MagicMock(),
    ), patch(
        "app.modules.pipeline.background_render.render_one_background",
        side_effect=_capturing,
    ):
        result = step._execute()

    # Order honoured: anchor first.
    assert [c["bg_id"] for c in captured] == ["L01B01", "L01B02"]
    # Anchor uses base FP only.
    assert captured[0]["fp_path"] is not None
    assert captured[0]["fp_path"].endswith("fp_a.png")
    assert captured[0]["prior_bg_paths"] == []
    # Derived uses prior BG path (catalog) only — NO FP.
    assert captured[1]["fp_path"] is None
    assert len(captured[1]["prior_bg_paths"]) == 1
    assert captured[1]["prior_bg_paths"][0].endswith("L01B01.png")

    # Audit fields land on groups_out.
    a = result["data"]["groups"]["L01B01"]
    b = result["data"]["groups"]["L01B02"]
    assert a["reference_decision"]["mode"] == "fp_seeded_anchor"
    assert b["reference_decision"]["mode"] == "reference_derived"
    assert b["reference_decision"]["source_bg_ids"] == ["L01B01"]
    assert a["camera_decision"]["lens_enum"] == "normal"
    assert b["camera_decision"]["lens_enum"] == "wide"
    assert a["shot_aware_plan_node_index"] == 0
    assert b["shot_aware_plan_node_index"] == 1
    # effective_render_prompt == render call prompt.
    assert a["effective_render_prompt"] == captured[0]["prompt"]
    assert b["effective_render_prompt"] == captured[1]["prompt"]
    # W20D: render_guidance surfaces top-level and inside reference_decision.
    assert a["render_guidance"] == _render_guidance_for("L01B01")
    assert b["render_guidance"] == _render_guidance_for("L01B02")
    assert (
        a["reference_decision"]["render_guidance"]
        == _render_guidance_for("L01B01")
    )
    # W20D: captured prompt contains camera exact fields + render_guidance.
    a_prompt = captured[0]["prompt"]
    assert "camera_unit: 1" in a_prompt
    assert "camera_cell: [0, 0]" in a_prompt
    assert "look_at_unit: 2" in a_prompt
    assert "look_at_cell: [2, 0]" in a_prompt
    assert "lens_enum: 'normal'" in a_prompt
    assert "fov_deg: 50" in a_prompt
    for v in _render_guidance_for("L01B01").values():
        assert v in a_prompt
    b_prompt = captured[1]["prompt"]
    assert "lens_enum: 'wide'" in b_prompt
    assert "fov_deg: 90" in b_prompt
    for v in _render_guidance_for("L01B02").values():
        assert v in b_prompt
    # Catalog dump carries both bgs in render order.
    catalog = result["data"]["bg_reference_catalog"]
    assert [e["bg_id"] for e in catalog] == ["L01B01", "L01B02"]
    assert catalog[0]["mode"] == "fp_seeded_anchor"
    assert catalog[1]["mode"] == "reference_derived"
    assert catalog[1]["source_bg_ids"] == ["L01B01"]


# ─── test #4: missing plan cp → all bgs failed, zero image calls ───────


def test_shot_aware_plan_missing_plan_emits_failed_no_image_calls(
    tmp_path, monkeypatch,
):
    """TASK3-B: with the missing-plan direct-plate fallback DISABLED, a
    missing fp plan stays fail-closed (legacy behaviour)."""
    _apply_settings(monkeypatch, tmp_path)
    monkeypatch.setattr(
        "app.core.config.settings."
        "background_render_missing_shot_aware_plan_fallback_enabled",
        False,
    )
    bg_specs = [_bg_spec(bg_id="L01B01")]
    cp_map = _build_cp_map(
        tmp_path=tmp_path, fp_id="fp_a", bg_specs=bg_specs, plan=None
    )
    step = _new_step(tmp_path, cp_map)

    with patch(
        "app.core.steps.background_render_step._resolve_openai_client",
        return_value=MagicMock(),
    ), patch(
        "app.modules.pipeline.background_render.render_one_background",
    ) as mock_render:
        result = step._execute()

    mock_render.assert_not_called()
    assert result["failed_count"] == 1
    entry = result["data"]["groups"]["L01B01"]
    assert entry["status"] == "failed"
    assert "shot_aware_plan" in entry["render_error"]
    # Catalog key still present for opt-in path (empty).
    assert result["data"]["bg_reference_catalog"] == []


# ─── TASK3-B: missing-plan direct-plate fallback (default ON) ──────────


def test_missing_plan_degrades_to_direct_plate_when_enabled(
    tmp_path, monkeypatch,
):
    """Default fallback ON: a missing fp plan degrades the fp's bgs to the
    direct-plate (text_only + prior-bg) render path instead of failing —
    rendered, explicitly marked, and folded into the reference catalog so
    background_render completes (no STALE_UPSTREAM cascade)."""
    _apply_settings(monkeypatch, tmp_path)
    monkeypatch.setattr(
        "app.core.config.settings."
        "background_render_missing_shot_aware_plan_fallback_enabled",
        True,
    )
    bg_specs = [_bg_spec(bg_id="L01B01")]
    cp_map = _build_cp_map(
        tmp_path=tmp_path, fp_id="fp_a", bg_specs=bg_specs, plan=None
    )
    step = _new_step(tmp_path, cp_map)

    with patch(
        "app.core.steps.background_render_step._resolve_openai_client",
        return_value=MagicMock(),
    ), patch(
        "app.modules.pipeline.background_render.render_one_background",
        side_effect=_success_renderer(),
    ) as mock_render:
        result = step._execute()

    mock_render.assert_called_once()
    entry = result["data"]["groups"]["L01B01"]
    assert entry["status"] == "ok"
    assert entry["png_path"]
    assert entry["shot_aware_plan_mode"] == "missing_plan_direct_plate_fallback"
    assert entry["render_degraded"] is True
    assert entry["fallback_reason"] == "shot_aware_plan_missing"
    assert entry["missing_fp_id"] == "fp_a"
    assert result["completed_count"] == 1
    assert result["failed_count"] == 0
    # The degraded bg is in the reference catalog (downstream sees a png).
    catalog = result["data"]["bg_reference_catalog"]
    assert any(c["bg_id"] == "L01B01" for c in catalog)
    assert any(
        c.get("source_kind") == "missing_plan_direct_plate_fallback"
        for c in catalog
    )


def test_missing_plan_no_direct_inputs_stays_failed(tmp_path, monkeypatch):
    """Codex condition: the fallback degrades ONLY when the direct-plate
    inputs are present. A bg whose t2i_prompt is empty cannot render
    text_only → it stays fail-closed even with the fallback enabled."""
    _apply_settings(monkeypatch, tmp_path)
    monkeypatch.setattr(
        "app.core.config.settings."
        "background_render_missing_shot_aware_plan_fallback_enabled",
        True,
    )
    bg_specs = [_bg_spec(bg_id="L01B01")]
    cp_map = _build_cp_map(
        tmp_path=tmp_path, fp_id="fp_a", bg_specs=bg_specs, plan=None
    )
    # Strip the direct-plate input.
    cp_map["background_prompt"]["data"]["backgrounds"]["L01B01"][
        "t2i_prompt"
    ] = ""
    step = _new_step(tmp_path, cp_map)

    with patch(
        "app.core.steps.background_render_step._resolve_openai_client",
        return_value=MagicMock(),
    ), patch(
        "app.modules.pipeline.background_render.render_one_background",
    ) as mock_render:
        result = step._execute()

    mock_render.assert_not_called()
    entry = result["data"]["groups"]["L01B01"]
    assert entry["status"] == "failed"
    assert "shot_aware_plan" in entry["render_error"]
    assert result["failed_count"] == 1


# ─── test #5: plan not production_clear → all bgs failed ───────────────


def test_shot_aware_plan_not_production_clear_emits_failed(
    tmp_path, monkeypatch,
):
    _apply_settings(monkeypatch, tmp_path)
    bg_specs = [_bg_spec(bg_id="L01B01")]
    plan = _plan_for_fp(
        fp_id="fp_a",
        nodes=[
            _plan_node(
                bg_id="L01B01", node_index=0,
                mode="fp_seeded_anchor", is_anchor=True,
            )
        ],
        production_clear=False,  # ← gate fails.
    )
    cp_map = _build_cp_map(
        tmp_path=tmp_path, fp_id="fp_a", bg_specs=bg_specs, plan=plan
    )
    step = _new_step(tmp_path, cp_map)

    with patch(
        "app.core.steps.background_render_step._resolve_openai_client",
        return_value=MagicMock(),
    ), patch(
        "app.modules.pipeline.background_render.render_one_background",
    ) as mock_render:
        result = step._execute()

    mock_render.assert_not_called()
    assert result["failed_count"] == 1
    entry = result["data"]["groups"]["L01B01"]
    assert entry["status"] == "failed"
    assert "production_clear" in entry["render_error"]


# ─── test #5b: plan production_clear is a truthy string → image 0 ──


def test_shot_aware_plan_production_clear_truthy_non_bool_emits_failed(
    tmp_path, monkeypatch,
):
    """Codex review #1 narrow gate hardening: a JSON-string round trip
    that demotes ``production_clear`` from ``True`` to ``"true"`` must
    NOT pass the gate.  The bg is failed without an image call."""
    _apply_settings(monkeypatch, tmp_path)
    bg_specs = [_bg_spec(bg_id="L01B01")]
    plan = _plan_for_fp(
        fp_id="fp_a",
        nodes=[
            _plan_node(
                bg_id="L01B01", node_index=0,
                mode="fp_seeded_anchor", is_anchor=True,
            )
        ],
    )
    plan["production_clear"] = "true"   # ← truthy but NOT True.
    cp_map = _build_cp_map(
        tmp_path=tmp_path, fp_id="fp_a", bg_specs=bg_specs, plan=plan
    )
    step = _new_step(tmp_path, cp_map)

    with patch(
        "app.core.steps.background_render_step._resolve_openai_client",
        return_value=MagicMock(),
    ), patch(
        "app.modules.pipeline.background_render.render_one_background",
    ) as mock_render:
        result = step._execute()

    mock_render.assert_not_called()
    assert result["failed_count"] == 1
    entry = result["data"]["groups"]["L01B01"]
    assert entry["status"] == "failed"
    assert "production_clear" in entry["render_error"]


# ─── test #6: plan ref_bg_id not in catalog → image call not issued ───


def test_shot_aware_plan_ref_unresolved_skips_image_call(
    tmp_path, monkeypatch,
):
    """Plan node #1 references a bg that was never rendered first
    (anchor missing from the plan). Adapter raises; bg surfaces as
    failed without an image call."""
    _apply_settings(monkeypatch, tmp_path)
    bg_specs = [_bg_spec(bg_id="L01B02")]
    plan = _plan_for_fp(
        fp_id="fp_a",
        nodes=[
            _plan_node(
                bg_id="L01B02", node_index=0,
                mode="reference_derived",
                selected_refs=[{
                    "ref_bg_id": "L01B_GHOST",
                    "physical_space_id": "primary",
                    "space_description": None,
                }],
            ),
        ],
    )
    cp_map = _build_cp_map(
        tmp_path=tmp_path, fp_id="fp_a", bg_specs=bg_specs, plan=plan
    )
    step = _new_step(tmp_path, cp_map)

    with patch(
        "app.core.steps.background_render_step._resolve_openai_client",
        return_value=MagicMock(),
    ), patch(
        "app.modules.pipeline.background_render.render_one_background",
    ) as mock_render:
        result = step._execute()

    mock_render.assert_not_called()
    entry = result["data"]["groups"]["L01B02"]
    assert entry["status"] == "failed"
    assert "L01B_GHOST" in entry["render_error"]
    assert "catalog" in entry["render_error"]
    assert result["failed_count"] == 1


# ─── test #7: legacy path unaffected by new branch ─────────────────────


def test_legacy_path_unaffected_by_shot_aware_plan_branch(
    tmp_path, monkeypatch,
):
    _apply_settings(monkeypatch, tmp_path, mode="legacy")
    bg_specs = [_bg_spec(bg_id="L01B01")]
    cp_map = _build_cp_map(
        tmp_path=tmp_path, fp_id="fp_a", bg_specs=bg_specs, plan=None
    )
    step = _new_step(tmp_path, cp_map)

    with patch(
        "app.core.steps.background_render_step._resolve_openai_client",
        return_value=MagicMock(),
    ), patch(
        "app.modules.pipeline.background_render.render_one_background",
        side_effect=_success_renderer(),
    ) as mock_render, patch(
        "app.core.steps.background_render_step.BackgroundRenderStep."
        "_run_shot_aware_plan_queue"
    ) as mock_w20c, patch(
        "app.core.steps.background_render_step.ThreadPoolExecutor"
    ) as mock_pool:
        pool_ctx = MagicMock()
        pool_ctx.__enter__ = MagicMock(return_value=pool_ctx)
        pool_ctx.__exit__ = MagicMock(return_value=False)

        def _submit(fn, *args, **kwargs):
            fut = MagicMock()
            fut.result = lambda: fn(*args, **kwargs)
            return fut
        pool_ctx.submit = _submit
        mock_pool.return_value = pool_ctx

        with patch(
            "app.core.steps.background_render_step.as_completed",
            side_effect=lambda futures: list(futures),
        ):
            result = step._execute()

    # ``shot_aware_plan`` helper not invoked. The deprecated W19B-3
    # ``_run_sequential_overlap_queue`` helper was deleted in W20E7-A,
    # so the legacy path neither calls it nor needs a patch.
    assert mock_w20c.call_count == 0
    # Legacy ThreadPool used.
    assert mock_pool.call_count >= 1
    assert mock_render.call_count == 1
    # bg_reference_catalog key NOT present in legacy path data.
    assert "bg_reference_catalog" not in result["data"]


# ─── W21B-w4 #4(C): substrate consumer combines FP + parent refs ─────────


def _stamp_dag(node, *, needs_new=True, parents=None, roles=None, anchor=None):
    """Add the schema-9 3a/3b fields a real plan cp would carry (stamped by
    mirror_plate_partition / build_reference_dag in build_render_plan_for_fp)."""
    node.update({
        "needs_new_plate": needs_new,
        "ref_tree_parents": parents or [],
        "ref_role_per_parent": roles or {},
        "plate_anchor_bg_id": anchor if anchor is not None else node["bg_id"],
    })
    return node


def test_substrate_consumer_combines_fp_plus_parent_when_flag_on(
    tmp_path, monkeypatch,
):
    """bg_render_substrate_enabled ON → a fresh reference_derived node's render
    input is FP + parent plate COMBINED (fp_plus_refs), resolved from the 3b
    ref_tree_parents. The legacy adapter path renders the same node parent-ONLY
    (no FP); this is the #4(C) behavioral change. anchor stays FP-only."""
    _apply_settings(monkeypatch, tmp_path, w20b_enabled=True)
    monkeypatch.setattr(
        "app.core.config.settings.bg_render_substrate_enabled", True
    )
    bg_specs = [_bg_spec(bg_id="L01B01"), _bg_spec(bg_id="L01B02")]
    anchor = _stamp_dag(
        _plan_node(bg_id="L01B01", node_index=0, mode="fp_seeded_anchor",
                   is_anchor=True),
        parents=[], roles={}, anchor="L01B01",
    )
    derived = _stamp_dag(
        _plan_node(bg_id="L01B02", node_index=1, mode="reference_derived",
                   selected_refs=[{"ref_bg_id": "L01B01",
                                   "physical_space_id": "primary",
                                   "space_description": None}],
                   camera_lens="wide"),
        parents=["L01B01"], roles={"L01B01": "space_continuity"},
        anchor="L01B02",
    )
    plan = _plan_for_fp(fp_id="fp_a", nodes=[derived, anchor])
    cp_map = _build_cp_map(
        tmp_path=tmp_path, fp_id="fp_a", bg_specs=bg_specs, plan=plan
    )
    step = _new_step(tmp_path, cp_map)

    captured: List[Dict[str, Any]] = []

    def _capturing(*, openai_client, image_model, prompt, out_path, fp_path,
                   prior_bg_paths, bg_id, **kwargs):
        _touch_png(Path(out_path))
        captured.append({
            "bg_id": bg_id,
            "fp_path": str(fp_path) if fp_path is not None else None,
            "prior_bg_paths": [str(p) for p in prior_bg_paths],
        })
        return {
            "status": "ok", "attempts": 1, "strategies": [],
            "ref_used": "x", "final_block_reason": None,
            "png_path": str(out_path),
        }

    with patch(
        "app.core.steps.background_render_step._resolve_openai_client",
        return_value=MagicMock(),
    ), patch(
        "app.modules.pipeline.background_render.render_one_background",
        side_effect=_capturing,
    ):
        result = step._execute()

    by_bg = {c["bg_id"]: c for c in captured}
    # anchor: FP only (no parents).
    assert by_bg["L01B01"]["fp_path"].endswith("fp_a.png")
    assert by_bg["L01B01"]["prior_bg_paths"] == []
    # derived: FP + parent COMBINED — the #4(C) substrate change (legacy = parent
    # only, no FP).
    assert by_bg["L01B02"]["fp_path"].endswith("fp_a.png")
    assert len(by_bg["L01B02"]["prior_bg_paths"]) == 1
    assert by_bg["L01B02"]["prior_bg_paths"][0].endswith("L01B01.png")
    # substrate audit surfaced on groups_out (JSON-safe subset).
    sd = result["data"]["groups"]["L01B02"]["substrate_decision"]
    assert sd["ref_used"] == "fp_plus_refs"
    assert sd["ordered_parent_bg_ids"] == ["L01B01"]
    assert sd["node_class"] == "fresh"
    assert result["data"]["groups"]["L01B01"]["substrate_decision"][
        "ref_used"] == "fp_only"


def test_substrate_flag_off_is_byte_identical_parent_only(tmp_path, monkeypatch):
    """flag OFF (default) → derived node renders parent-only via the adapter
    (no FP, no substrate_decision key). Guards the OFF byte-identical contract."""
    _apply_settings(monkeypatch, tmp_path, w20b_enabled=True)
    # bg_render_substrate_enabled left default False.
    bg_specs = [_bg_spec(bg_id="L01B01"), _bg_spec(bg_id="L01B02")]
    anchor = _stamp_dag(
        _plan_node(bg_id="L01B01", node_index=0, mode="fp_seeded_anchor",
                   is_anchor=True), anchor="L01B01")
    derived = _stamp_dag(
        _plan_node(bg_id="L01B02", node_index=1, mode="reference_derived",
                   selected_refs=[{"ref_bg_id": "L01B01",
                                   "physical_space_id": "primary",
                                   "space_description": None}]),
        parents=["L01B01"], roles={"L01B01": "space_continuity"},
        anchor="L01B02")
    plan = _plan_for_fp(fp_id="fp_a", nodes=[derived, anchor])
    cp_map = _build_cp_map(
        tmp_path=tmp_path, fp_id="fp_a", bg_specs=bg_specs, plan=plan)
    step = _new_step(tmp_path, cp_map)

    captured: List[Dict[str, Any]] = []

    def _cap(*, openai_client, image_model, prompt, out_path, fp_path,
             prior_bg_paths, bg_id, **kwargs):
        _touch_png(Path(out_path))
        captured.append({
            "bg_id": bg_id,
            "fp_path": str(fp_path) if fp_path is not None else None,
            "prior_bg_paths": [str(p) for p in prior_bg_paths],
        })
        return {"status": "ok", "attempts": 1, "strategies": [],
                "ref_used": "x", "final_block_reason": None,
                "png_path": str(out_path)}

    with patch(
        "app.core.steps.background_render_step._resolve_openai_client",
        return_value=MagicMock(),
    ), patch(
        "app.modules.pipeline.background_render.render_one_background",
        side_effect=_cap,
    ):
        result = step._execute()

    by_bg = {c["bg_id"]: c for c in captured}
    # legacy adapter: derived is parent-ONLY, no FP.
    assert by_bg["L01B02"]["fp_path"] is None
    assert by_bg["L01B02"]["prior_bg_paths"][0].endswith("L01B01.png")
    assert "substrate_decision" not in result["data"]["groups"]["L01B02"]


def test_config_hash_substrate_flag_invalidates_only_shot_aware_path(
    tmp_path, monkeypatch,
):
    """Codex Required 3 — bg_render_substrate_enabled must invalidate the
    background_render config_hash, but only on the shot_aware_plan opt-in path
    with the flag True (so legacy/default + shot_aware+flag-False keep their
    existing hash byte-identical)."""
    step = _new_step(tmp_path, {})
    monkeypatch.setattr("app.core.config.settings.background_mode", "on")

    def _h(mode, flag):
        monkeypatch.setattr(
            "app.core.config.settings.background_render_reference_mode", mode
        )
        monkeypatch.setattr(
            "app.core.config.settings.bg_render_substrate_enabled", flag
        )
        return step._config_hash()

    # legacy path: the flag never enters the payload → hash unchanged.
    assert _h("legacy", False) == _h("legacy", True)
    # shot_aware_plan path: flag True invalidates; False keeps the pre-#4(C) hash.
    sa_false = _h("shot_aware_plan", False)
    sa_true = _h("shot_aware_plan", True)
    assert sa_false != sa_true
    assert _h("shot_aware_plan", False) == sa_false  # stable / byte-identical
    # shot_aware+False is distinct from legacy (mode is stamped) — sanity.
    assert sa_false != _h("legacy", False)


def test_substrate_on_missing_parent_gracefully_falls_back_fp_only(
    tmp_path, monkeypatch,
):
    """Codex Required 1 — substrate ON: a fresh node whose ref_tree_parents plate
    is NOT in the catalog must NOT be blocked by the legacy adapter catalog gate.
    It renders FP-only with ref_used=fallback_fp_only + missing_parent_bg_ids,
    no failed group."""
    _apply_settings(monkeypatch, tmp_path, w20b_enabled=True)
    monkeypatch.setattr(
        "app.core.config.settings.bg_render_substrate_enabled", True
    )
    bg_specs = [_bg_spec(bg_id="L01B02")]
    # single derived node referencing a parent that never renders (not in plan).
    derived = _stamp_dag(
        _plan_node(bg_id="L01B02", node_index=0, mode="reference_derived",
                   selected_refs=[{"ref_bg_id": "MISSING",
                                   "physical_space_id": "primary",
                                   "space_description": None}]),
        parents=["MISSING"], roles={"MISSING": "space_continuity"},
        anchor="L01B02")
    plan = _plan_for_fp(fp_id="fp_a", nodes=[derived])
    cp_map = _build_cp_map(
        tmp_path=tmp_path, fp_id="fp_a", bg_specs=bg_specs, plan=plan)
    step = _new_step(tmp_path, cp_map)

    rendered: List[str] = []

    def _cap(*, openai_client, image_model, prompt, out_path, fp_path,
             prior_bg_paths, bg_id, **kwargs):
        _touch_png(Path(out_path))
        rendered.append(bg_id)
        return {"status": "ok", "attempts": 1, "strategies": [],
                "ref_used": "x", "final_block_reason": None,
                "png_path": str(out_path)}

    with patch(
        "app.core.steps.background_render_step._resolve_openai_client",
        return_value=MagicMock(),
    ), patch(
        "app.modules.pipeline.background_render.render_one_background",
        side_effect=_cap,
    ):
        result = step._execute()

    grp = result["data"]["groups"]["L01B02"]
    assert grp["status"] == "ok"  # NOT failed — graceful degrade
    assert "L01B02" in rendered  # image call happened (FP-only)
    sd = grp["substrate_decision"]
    assert sd["ref_used"] == "fallback_fp_only"
    assert sd["missing_parent_bg_ids"] == ["MISSING"]
    assert sd["ordered_parent_bg_ids"] == []


def test_substrate_on_non_schema9_node_fails_closed_no_image(
    tmp_path, monkeypatch,
):
    """Codex Required 2 — substrate ON + a production_clear plan node lacking the
    3a/3b fields (stale schema-8) must fail closed before any image call, not be
    silently mis-rendered as no-parent fp_only."""
    _apply_settings(monkeypatch, tmp_path, w20b_enabled=True)
    monkeypatch.setattr(
        "app.core.config.settings.bg_render_substrate_enabled", True
    )
    bg_specs = [_bg_spec(bg_id="L01B01")]
    # schema-8 node: NO needs_new_plate / ref_tree_parents / ref_role_per_parent.
    node = _plan_node(bg_id="L01B01", node_index=0, mode="fp_seeded_anchor",
                      is_anchor=True)
    plan = _plan_for_fp(fp_id="fp_a", nodes=[node])
    cp_map = _build_cp_map(
        tmp_path=tmp_path, fp_id="fp_a", bg_specs=bg_specs, plan=plan)
    step = _new_step(tmp_path, cp_map)

    rendered: List[str] = []

    def _cap(**kwargs):
        rendered.append(kwargs.get("bg_id"))
        return {"status": "ok", "attempts": 1, "strategies": [],
                "ref_used": "x", "final_block_reason": None,
                "png_path": str(kwargs["out_path"])}

    with patch(
        "app.core.steps.background_render_step._resolve_openai_client",
        return_value=MagicMock(),
    ), patch(
        "app.modules.pipeline.background_render.render_one_background",
        side_effect=_cap,
    ):
        result = step._execute()

    grp = result["data"]["groups"]["L01B01"]
    assert grp["status"] != "ok"
    assert "substrate_requires_schema9_plan_fields" in grp.get("render_error", "")
    assert rendered == []  # no image call


# ─── test #8 (W20E7-A): w18j_overlap deprecated — fail-closed, no planner ─


def test_w18j_overlap_path_deprecated_fail_closed_no_image_call(
    tmp_path, monkeypatch,
):
    """W20E7-A: ``background_render_reference_mode='w18j_overlap'`` is
    deprecated.  The branch must:
      - emit every renderable bg as ``failed`` with a stable error code
        containing ``w18j_overlap_deprecated`` and
        ``background_render_reference_mode_deprecated``;
      - never call ``render_one_background`` (zero image API attempts);
      - never invoke ``_run_shot_aware_plan_queue``;
      - never import ``app.modules.pipeline.background_image_planner``
        (asserted via the static-source guard in the next test).
    """
    _apply_settings(monkeypatch, tmp_path, mode="w18j_overlap")
    bg_specs = [_bg_spec(bg_id="L01B01"), _bg_spec(bg_id="L01B02")]
    cp_map = _build_cp_map(
        tmp_path=tmp_path, fp_id="fp_a", bg_specs=bg_specs, plan=None
    )
    step = _new_step(tmp_path, cp_map)

    with patch(
        "app.core.steps.background_render_step._resolve_openai_client",
        return_value=MagicMock(),
    ), patch(
        "app.modules.pipeline.background_render.render_one_background",
    ) as mock_render, patch(
        "app.core.steps.background_render_step.BackgroundRenderStep."
        "_run_shot_aware_plan_queue"
    ) as mock_w20c:
        result = step._execute()

    # No image API attempts on the deprecated branch.
    mock_render.assert_not_called()
    # shot_aware_plan helper is not invoked from the deprecated branch.
    assert mock_w20c.call_count == 0
    # Both renderable bgs marked failed with the deprecation error code.
    assert result["failed_count"] == 2
    assert result["completed_count"] == 0
    for bg_id in ("L01B01", "L01B02"):
        entry = result["data"]["groups"][bg_id]
        assert entry["status"] == "failed"
        assert entry["png_path"] == ""
        assert "w18j_overlap_deprecated" in entry["render_error"]
        assert (
            "background_render_reference_mode_deprecated"
            in entry["render_error"]
        )


def test_active_production_does_not_import_background_image_planner():
    """W20E7-A guard: scan active production source files (under
    ``app.core.steps`` / ``app.modules.pipeline`` / ``app.services``) for
    any ``background_image_planner`` import statement.  None must exist;
    the W19B-3 deterministic reference planner module itself is deleted
    (W20E7-A follow-up).  This test now also guards against any future
    re-introduction of the import path.
    """
    import pathlib

    repo_root = pathlib.Path(__file__).resolve().parents[3]
    trees = [
        repo_root / "backend" / "app" / "core" / "steps",
        repo_root / "backend" / "app" / "modules" / "pipeline",
        repo_root / "backend" / "app" / "services",
    ]
    offenders: List[str] = []
    for tree in trees:
        for py in tree.rglob("*.py"):
            for lineno, raw in enumerate(
                py.read_text(encoding="utf-8").splitlines(), start=1
            ):
                stripped = raw.strip()
                if not stripped or stripped.startswith("#"):
                    continue
                if "background_image_planner" not in stripped:
                    continue
                if stripped.startswith(("from ", "import ")):
                    offenders.append(
                        f"{py.relative_to(repo_root)}:{lineno}: {stripped}"
                    )
    assert offenders == [], (
        "Active production module(s) still import background_image_planner: "
        f"{offenders}"
    )


# ─── test #9: mutex (W19B-3 vs W20B) unchanged — only w18j+W20B blocks ─


def test_mutex_w18j_plus_w20b_still_blocks(tmp_path, monkeypatch):
    """The existing W19B-3 / W20B mutex guard must keep firing when
    BOTH the old opt-in (``w18j_overlap``) AND the W20B planner are
    active.  The new ``shot_aware_plan`` selector value does NOT
    change this guard's contract."""
    _apply_settings(
        monkeypatch, tmp_path,
        mode="w18j_overlap", w20b_enabled=True,
    )
    step = _new_step(tmp_path, {})

    with patch(
        "app.modules.pipeline.background_render.render_one_background",
    ) as mock_render, patch(
        "app.core.steps.background_render_step._resolve_openai_client",
    ) as mock_client:
        result = step._execute()

    mock_render.assert_not_called()
    mock_client.assert_not_called()
    assert result["failed_count"] == 1
    assert "selector_conflict" in result["data"]
    assert "w18j_overlap" in result["data"]["selector_conflict"]


def test_mutex_shot_aware_plus_w20b_does_not_block(tmp_path, monkeypatch):
    """``shot_aware_plan`` mode WITH ``shot_aware_bg_render_plan_enabled``
    is the normal happy path: the planner step is what *produces* the
    plan cp this branch consumes.  The mutex must NOT fire for this
    combination — it only fires for w18j_overlap + W20B."""
    _apply_settings(
        monkeypatch, tmp_path,
        mode="shot_aware_plan", w20b_enabled=True,
    )
    bg_specs = [_bg_spec(bg_id="L01B01")]
    plan = _plan_for_fp(
        fp_id="fp_a",
        nodes=[
            _plan_node(
                bg_id="L01B01", node_index=0,
                mode="fp_seeded_anchor", is_anchor=True,
            )
        ],
    )
    cp_map = _build_cp_map(
        tmp_path=tmp_path, fp_id="fp_a", bg_specs=bg_specs, plan=plan
    )
    step = _new_step(tmp_path, cp_map)

    with patch(
        "app.core.steps.background_render_step._resolve_openai_client",
        return_value=MagicMock(),
    ), patch(
        "app.modules.pipeline.background_render.render_one_background",
        side_effect=_success_renderer(),
    ):
        result = step._execute()

    assert "selector_conflict" not in result.get("data", {})
    assert result["completed_count"] == 1
    assert result["data"]["groups"]["L01B01"]["status"] == "ok"


# ─── test #10: _config_hash stamps shot_aware_plan selector ────────────


def test_config_hash_stamps_shot_aware_plan_selector(monkeypatch):
    """legacy path is byte-identical to a payload without the new key.
    shot_aware_plan stamps the selector value; w18j_overlap unchanged."""
    import hashlib
    import json as _json

    from app.core.steps.background_render_step import (
        PROMPT_VERSION,
        SCHEMA_VERSION,
        BackgroundRenderStep,
    )

    step = BackgroundRenderStep.__new__(BackgroundRenderStep)

    legacy_payload = {
        "background_mode": "on",
        "schema_version": SCHEMA_VERSION,
        "prompt_version": PROMPT_VERSION,
    }
    legacy_expected = hashlib.sha256(
        _json.dumps(legacy_payload, sort_keys=True).encode("utf-8")
    ).hexdigest()[:16]

    shot_aware_payload = dict(legacy_payload)
    shot_aware_payload["background_render_reference_mode"] = "shot_aware_plan"
    # TASK3-B: the missing-plan fallback policy is stamped in the
    # shot_aware_plan path (default ON) — legacy / w18j stay byte-identical.
    shot_aware_payload[
        "background_render_missing_shot_aware_plan_fallback_enabled"
    ] = True
    shot_aware_expected = hashlib.sha256(
        _json.dumps(shot_aware_payload, sort_keys=True).encode("utf-8")
    ).hexdigest()[:16]

    w18j_payload = dict(legacy_payload)
    w18j_payload["background_render_reference_mode"] = "w18j_overlap"
    w18j_expected = hashlib.sha256(
        _json.dumps(w18j_payload, sort_keys=True).encode("utf-8")
    ).hexdigest()[:16]

    monkeypatch.setattr("app.core.config.settings.background_mode", "on")
    # W-G: 이 테스트의 기대 payload 는 building-fp 스탬프가 없는 형태 — 실행
    # 환경(.env)의 flag 값과 무관하게 결정론이 되도록 명시 OFF 고정.
    monkeypatch.setattr(
        "app.core.config.settings.outdoor_building_fp_ref_enabled", False
    )
    # W-I: building anchor 스탬프도 동일 정책 — 명시 OFF 고정.
    monkeypatch.setattr(
        "app.core.config.settings.outdoor_building_anchor_ref_enabled", False
    )
    # W-K: same-place 렌더 체이닝 스탬프도 동일 정책 — 명시 OFF 고정.
    monkeypatch.setattr(
        "app.core.config.settings.same_place_render_chain_enabled", False
    )
    # W-L: outdoor aerial 스탬프도 동일 정책 — 명시 OFF 고정.
    monkeypatch.setattr(
        "app.core.config.settings.outdoor_aerial_reference_enabled", False
    )
    # W-M: plate 단계화 체인 스탬프도 동일 정책 — 명시 OFF 고정.
    monkeypatch.setattr(
        "app.core.config.settings.outdoor_plate_stage_chain_enabled", False
    )

    monkeypatch.setattr(
        "app.core.config.settings.background_render_reference_mode", "legacy"
    )
    assert step._config_hash() == legacy_expected

    monkeypatch.setattr(
        "app.core.config.settings.background_render_reference_mode",
        "shot_aware_plan",
    )
    assert step._config_hash() == shot_aware_expected

    monkeypatch.setattr(
        "app.core.config.settings.background_render_reference_mode",
        "w18j_overlap",
    )
    assert step._config_hash() == w18j_expected

    assert legacy_expected != shot_aware_expected
    assert shot_aware_expected != w18j_expected


# ─── test #11: bg not in plan graph → failed, no image call ────────────


def test_shot_aware_plan_renderable_bg_not_in_plan_emits_failed(
    tmp_path, monkeypatch,
):
    """master_plan + prompt declare bg L01B01 + L01B02.  The plan only
    covers L01B01 — L01B02 must surface as failed without an image
    call.  Plan-graph membership is exact-ID and required.
    """
    _apply_settings(monkeypatch, tmp_path)
    bg_specs = [
        _bg_spec(bg_id="L01B01"),
        _bg_spec(bg_id="L01B02"),
    ]
    plan = _plan_for_fp(
        fp_id="fp_a",
        nodes=[
            _plan_node(
                bg_id="L01B01", node_index=0,
                mode="fp_seeded_anchor", is_anchor=True,
            )
        ],
    )
    cp_map = _build_cp_map(
        tmp_path=tmp_path, fp_id="fp_a", bg_specs=bg_specs, plan=plan
    )
    step = _new_step(tmp_path, cp_map)

    render_calls: List[str] = []

    def _capturing(*, openai_client, image_model, prompt, out_path,
                   fp_path, prior_bg_paths, bg_id, **kwargs):
        render_calls.append(bg_id)
        _touch_png(Path(out_path))
        return {
            "status": "ok", "attempts": 1, "strategies": [],
            "ref_used": "fp_only", "final_block_reason": None,
            "png_path": str(out_path),
        }

    with patch(
        "app.core.steps.background_render_step._resolve_openai_client",
        return_value=MagicMock(),
    ), patch(
        "app.modules.pipeline.background_render.render_one_background",
        side_effect=_capturing,
    ):
        result = step._execute()

    # Only the bg covered by the plan was rendered.
    assert render_calls == ["L01B01"]
    a = result["data"]["groups"]["L01B01"]
    b = result["data"]["groups"]["L01B02"]
    assert a["status"] == "ok"
    assert b["status"] == "failed"
    assert "missing from plan graph" in b["render_error"]


# ─── Commit 2b — reuse_existing_plate materialize / verify ────────────
#
# render_action='reuse_existing_plate' nodes must NOT call the image API;
# they alias the target plate's path, are registered as a chain_bg
# ImageAsset row sharing the target file_path, and stay status='ok' +
# is_reuse=True so downstream scene loaders (which gate on status=='ok')
# still see them. Tests are deterministic — render_one_background mocked,
# zero real API.


def test_reuse_node_skips_image_call_and_aliases_target(
    tmp_path, monkeypatch,
):
    """A reuse_existing_plate node renders nothing; its entry aliases the
    target plate path and carries is_reuse provenance."""
    _apply_settings(monkeypatch, tmp_path)
    bg_specs = [_bg_spec(bg_id="L01B01"), _bg_spec(bg_id="L01B02")]
    plan = _plan_for_fp(
        fp_id="fp_a",
        nodes=[
            _plan_node(
                bg_id="L01B01", node_index=0,
                mode="fp_seeded_anchor", is_anchor=True,
            ),
            _plan_node(
                bg_id="L01B02", node_index=1,
                mode="reference_derived",
                selected_refs=[{
                    "ref_bg_id": "L01B01",
                    "physical_space_id": "primary",
                    "space_description": None,
                }],
                render_action="reuse_existing_plate",
                reuse_target_bg_id="L01B01",
            ),
        ],
    )
    cp_map = _build_cp_map(
        tmp_path=tmp_path, fp_id="fp_a", bg_specs=bg_specs, plan=plan
    )
    step = _new_step(tmp_path, cp_map)

    with patch(
        "app.core.steps.background_render_step._resolve_openai_client",
        return_value=MagicMock(),
    ), patch(
        "app.modules.pipeline.background_render.render_one_background",
        side_effect=_success_renderer(),
    ) as mock_render:
        result = step._execute()

    # Only the anchor (render_new_plate) calls the image API. The reuse
    # node spends zero image calls / budget.
    assert mock_render.call_count == 1
    called_bgs = [c.kwargs["bg_id"] for c in mock_render.call_args_list]
    assert called_bgs == ["L01B01"]

    a = result["data"]["groups"]["L01B01"]
    b = result["data"]["groups"]["L01B02"]
    assert a["status"] == "ok"
    assert a.get("is_reuse") is False
    assert b["status"] == "ok"
    assert b["is_reuse"] is True
    assert b["render_action"] == "reuse_existing_plate"
    assert b["reuse_target_bg_id"] == "L01B01"
    assert b["reused_from_bg_id"] == "L01B01"
    # copy-less alias: reuse png_path == target png_path, no new file.
    assert b["png_path"] == a["png_path"]
    assert b["png_path"].endswith("L01B01.png")
    assert not (tmp_path / "p" / "episodes" / "e" / "images"
                / "background_chain" / "L01B02.png").exists()
    # reuse entry keeps its OWN shot ids (not the target's).
    assert b["shot_ids"] == ["L01B02_Shot1"]
    # P0-1 (W21B-w6): per-bg count. 두 bg 모두 status=ok + png_path
    # (anchor=rendered, reuse=copy-less alias to target png) → completed=2.
    # (이전 이진화 `1 if failed==0 else 0` 면 1.)
    assert result["completed_count"] == 2
    assert result["applicable_count"] == 2
    assert result["failed_count"] == 0


def test_reuse_target_missing_fails_closed(tmp_path, monkeypatch):
    """If the reuse target never produced a plate, the reuse node is
    fail-closed (status='reuse_target_missing'), no image call, and the
    step is not clean."""
    _apply_settings(monkeypatch, tmp_path)
    bg_specs = [_bg_spec(bg_id="L01B01"), _bg_spec(bg_id="L01B02")]
    plan = _plan_for_fp(
        fp_id="fp_a",
        nodes=[
            _plan_node(
                bg_id="L01B01", node_index=0,
                mode="fp_seeded_anchor", is_anchor=True,
            ),
            _plan_node(
                bg_id="L01B02", node_index=1,
                mode="reference_derived",
                selected_refs=[{
                    "ref_bg_id": "L01B01",
                    "physical_space_id": "primary",
                    "space_description": None,
                }],
                render_action="reuse_existing_plate",
                reuse_target_bg_id="L01B01",
            ),
        ],
    )
    cp_map = _build_cp_map(
        tmp_path=tmp_path, fp_id="fp_a", bg_specs=bg_specs, plan=plan
    )
    step = _new_step(tmp_path, cp_map)

    def _anchor_fails(*, openai_client, image_model, prompt, out_path,
                      fp_path, prior_bg_paths, bg_id, **kwargs):
        # Anchor render fails → no target plate for the reuse node.
        return {
            "status": "failed", "attempts": 1, "strategies": [],
            "ref_used": "text_only",
            "final_block_reason": "boom", "png_path": "",
        }

    with patch(
        "app.core.steps.background_render_step._resolve_openai_client",
        return_value=MagicMock(),
    ), patch(
        "app.modules.pipeline.background_render.render_one_background",
        side_effect=_anchor_fails,
    ) as mock_render:
        result = step._execute()

    # Only the anchor attempted an image call; reuse spent none.
    assert mock_render.call_count == 1
    b = result["data"]["groups"]["L01B02"]
    assert b["status"] == "reuse_target_missing"
    assert b["is_reuse"] is True
    assert b["render_action"] == "reuse_existing_plate"
    assert result["failed_count"] >= 1
    assert result["completed_count"] == 0


def test_reuse_alias_registers_chain_bg_imageasset_row(
    tmp_path, monkeypatch,
):
    """Integration-ish: render 1 + reuse 1 → two chain_bg ImageAsset rows
    sharing the target file_path (copy-less alias), exactly one image call.
    _register_image_assets is exercised for real against a mock db."""
    from app.core.steps.background_render_step import BackgroundRenderStep

    _apply_settings(monkeypatch, tmp_path)
    bg_specs = [_bg_spec(bg_id="L01B01"), _bg_spec(bg_id="L01B02")]
    plan = _plan_for_fp(
        fp_id="fp_a",
        nodes=[
            _plan_node(
                bg_id="L01B01", node_index=0,
                mode="fp_seeded_anchor", is_anchor=True,
            ),
            _plan_node(
                bg_id="L01B02", node_index=1,
                mode="reference_derived",
                selected_refs=[{
                    "ref_bg_id": "L01B01",
                    "physical_space_id": "primary",
                    "space_description": None,
                }],
                render_action="reuse_existing_plate",
                reuse_target_bg_id="L01B01",
            ),
        ],
    )
    cp_map = _build_cp_map(
        tmp_path=tmp_path, fp_id="fp_a", bg_specs=bg_specs, plan=plan
    )
    # Build a step WITHOUT mocking _register_image_assets so the alias row
    # creation is exercised. EntityCanon query → one location canon.
    step = BackgroundRenderStep.__new__(BackgroundRenderStep)
    step.project_id = "p"
    step.episode_id = "e"
    step.project_config = {}
    step.build_opik_metadata = MagicMock(return_value={})
    step._load_prev_checkpoint = MagicMock(
        side_effect=lambda sid: cp_map.get(sid)
    )

    added_rows: List[Any] = []
    canon = MagicMock()
    canon.short_id = "L01"
    canon.id = "canon-L01"

    db = MagicMock()
    db.query.return_value.filter.return_value.all.return_value = [canon]
    db.query.return_value.filter_by.return_value.first.return_value = None
    db.add.side_effect = lambda row: added_rows.append(row)
    step.db = db

    with patch(
        "app.core.steps.background_render_step._resolve_openai_client",
        return_value=MagicMock(),
    ), patch(
        "app.modules.pipeline.background_render.render_one_background",
        side_effect=_success_renderer(),
    ) as mock_render:
        result = step._execute()

    assert mock_render.call_count == 1  # one real render, reuse aliased
    chain_rows = [r for r in added_rows if r.asset_type == "chain_bg"]
    by_bg = {r.variant_type: r for r in chain_rows}
    assert set(by_bg) == {"L01B01", "L01B02"}
    # alias: both rows point at the same (target) relative file_path.
    assert by_bg["L01B02"].file_path == by_bg["L01B01"].file_path
    assert by_bg["L01B01"].file_path.endswith("L01B01.png")
    # P0-1 (W21B-w6): per-bg count — anchor(rendered) + reuse(alias) 둘 다
    # status=ok + png_path → completed=2 (이전 이진화면 1).
    assert result["completed_count"] == 2


def test_verify_completion_separates_new_and_reuse_counts(
    tmp_path, monkeypatch,
):
    """verify_completion surfaces new-render vs reuse mapping counts,
    classifying via render_action / is_reuse (NOT status)."""
    from app.core.file_paths import resolve_image_path  # noqa: F401

    _apply_settings(monkeypatch, tmp_path)
    bg_specs = [_bg_spec(bg_id="L01B01"), _bg_spec(bg_id="L01B02")]
    plan = _plan_for_fp(
        fp_id="fp_a",
        nodes=[
            _plan_node(
                bg_id="L01B01", node_index=0,
                mode="fp_seeded_anchor", is_anchor=True,
            ),
            _plan_node(
                bg_id="L01B02", node_index=1,
                mode="reference_derived",
                selected_refs=[{
                    "ref_bg_id": "L01B01",
                    "physical_space_id": "primary",
                    "space_description": None,
                }],
                render_action="reuse_existing_plate",
                reuse_target_bg_id="L01B01",
            ),
        ],
    )
    cp_map = _build_cp_map(
        tmp_path=tmp_path, fp_id="fp_a", bg_specs=bg_specs, plan=plan
    )
    step = _new_step(tmp_path, cp_map)

    with patch(
        "app.core.steps.background_render_step._resolve_openai_client",
        return_value=MagicMock(),
    ), patch(
        "app.modules.pipeline.background_render.render_one_background",
        side_effect=_success_renderer(),
    ):
        result = step._execute()

    # The render produced one real PNG (the anchor target).
    target_png = (tmp_path / "p" / "episodes" / "e" / "images"
                  / "background_chain" / "L01B01.png")
    assert target_png.exists()

    # Build chain_bg ImageAsset rows mirroring what _register would write:
    # both bgs point at the target relative path (alias).
    rel = str(target_png.relative_to(Path(str(tmp_path)).parent))

    def _row(bg_id):
        r = MagicMock()
        r.variant_type = bg_id
        r.file_path = rel
        return r

    # Re-run verify with a db returning the alias rows + self_cp groups.
    cp_map_verify = dict(cp_map)
    cp_map_verify["background_render"] = {"data": result["data"]}
    step.v_step = step  # noqa
    step._load_prev_checkpoint = MagicMock(
        side_effect=lambda sid: cp_map_verify.get(sid)
    )
    step.db.query.return_value.filter.return_value.all.return_value = [
        _row("L01B01"), _row("L01B02")
    ]

    report = step.verify_completion()
    md = report.metadata
    assert md["expected_new_render_count"] == 1
    assert md["expected_reuse_count"] == 1
    assert md["valid_reuse_count"] == 1
    assert md["missing_reuse_targets"] == []
    assert report.is_complete is True


# ─── E2E11 ②: STRUCTURE FACTS + VIEW AUTHORITY 배선 ────────────────────


def _cp_map_with_facts(tmp_path):
    bg_specs = [_bg_spec(bg_id="L01B01")]
    plan = _plan_for_fp(
        fp_id="fp_a",
        nodes=[
            _plan_node(
                bg_id="L01B01", node_index=0,
                mode="fp_seeded_anchor", is_anchor=True,
            )
        ],
    )
    cp_map = _build_cp_map(
        tmp_path=tmp_path, fp_id="fp_a", bg_specs=bg_specs, plan=plan
    )
    cp_map["floor_plan_prompt"]["data"]["floor_plans"]["fp_a"][
        "numbered_elements"] = [
        {"number": 9, "category": "opening",
         "base_layer_decision": "base_opening",
         "label": "stair access landing",
         "position_hint": "southwest edge of the roof deck"},
        # base 지만 이 bg 의 use 목록 밖 — FACTS 제외 (exact join)
        {"number": 5, "category": "opening",
         "base_layer_decision": "base_opening",
         "label": "side window",
         "position_hint": "south facade"},
        # use 로 소비되는 state_overlay — FACTS 절대 제외 (BLOCKING-1)
        {"number": 14, "category": "state",
         "base_layer_decision": "state_overlay_plot_cue",
         "label": "attacker silhouette in window",
         "position_hint": "front window"},
    ]
    cp_map["floor_plan_prompt"]["data"]["floor_plans"]["fp_a"][
        "camera_recommendations"] = [
        {"bg_id": "L01B01",
         "use_numbered_elements": [9, 14],
         "ignore_numbered_elements": [5]},
    ]
    return cp_map


def test_structure_facts_flag_on_injects_prefix(tmp_path, monkeypatch):
    """flag ON: fp numbered_elements 가 STRUCTURE FACTS 로 prefix 주입 +
    VIEW AUTHORITY 절 — effective 프롬프트(생성·판정 공유)에 도달."""
    _apply_settings(monkeypatch, tmp_path)
    monkeypatch.setattr(
        "app.core.config.settings.plate_structure_facts_enabled", True,
        raising=False,
    )
    step = _new_step(tmp_path, _cp_map_with_facts(tmp_path))
    captured = {}

    def _capture(*, prompt, out_path, **kwargs):
        captured["prompt"] = prompt
        _touch_png(Path(out_path))
        return {"status": "ok", "attempts": 1, "strategies": [],
                "ref_used": "fp_only"}

    with patch(
        "app.core.steps.background_render_step._resolve_openai_client",
        return_value=MagicMock(),
    ), patch(
        "app.modules.pipeline.background_render.render_one_background",
        side_effect=_capture,
    ):
        result = step._execute()

    assert result["data"]["groups"]["L01B01"]["status"] == "ok"
    p = captured["prompt"]
    assert "STRUCTURE FACTS" in p
    assert "- 9. [opening] stair access landing — southwest edge" in p
    assert "VIEW AUTHORITY" in p
    # Codex BLOCKING-1: state_overlay 는 use 목록에 있어도 FACTS 제외,
    # base 라도 use 목록 밖(exact join)이면 제외
    assert "attacker silhouette" not in p
    assert "side window" not in p
    entry = result["data"]["groups"]["L01B01"]
    assert "STRUCTURE FACTS" in entry["reference_guidance_prefix"]


def test_structure_facts_flag_off_byte_identical(tmp_path, monkeypatch):
    """flag OFF(default): prefix 에 신규 절 0 — 기존 byte-identical."""
    _apply_settings(monkeypatch, tmp_path)
    step = _new_step(tmp_path, _cp_map_with_facts(tmp_path))
    captured = {}

    def _capture(*, prompt, out_path, **kwargs):
        captured["prompt"] = prompt
        _touch_png(Path(out_path))
        return {"status": "ok", "attempts": 1, "strategies": [],
                "ref_used": "fp_only"}

    with patch(
        "app.core.steps.background_render_step._resolve_openai_client",
        return_value=MagicMock(),
    ), patch(
        "app.modules.pipeline.background_render.render_one_background",
        side_effect=_capture,
    ):
        result = step._execute()

    assert result["data"]["groups"]["L01B01"]["status"] == "ok"
    assert "STRUCTURE FACTS" not in captured["prompt"]
    assert "VIEW AUTHORITY" not in captured["prompt"]


def test_structure_facts_flag_stamps_config_hash(monkeypatch):
    from app.core.steps.background_render_step import BackgroundRenderStep

    step = BackgroundRenderStep.__new__(BackgroundRenderStep)
    base = step._config_hash()
    monkeypatch.setattr(
        "app.core.config.settings.plate_structure_facts_enabled", True,
        raising=False,
    )
    assert step._config_hash() != base


# ─── W20F11: camera-geometry typed not_applicable → direct plate ───────


def _camera_geometry_na_plan(fp_id: str) -> Dict[str, Any]:
    return {
        "fp_id": fp_id,
        "shot_aware_bg_render_plan_status": "not_applicable",
        "not_applicable_reason_code": "no_viable_camera_look_at_pair",
        "camera_lookat_viability": {
            "camera_candidate_counts": {"1": 1},
            "look_at_candidate_counts": {"1": 0},
        },
        "graph": {"nodes": []},
        "validators": {"all_validators_passed": False, "diagnostics": []},
        "real_api_call_counts": {"image": 0, "llm": 0, "vlm": 0},
        "readback_status": "ok",
        "production_clear": False,
        "diagnostics": [],
    }


def test_camera_geometry_na_routes_to_direct_plate(tmp_path, monkeypatch):
    """W20F11 (Codex 합의): typed reason 만 direct-plate — 렌더 성공·감사
    (mode/fallback_reason/source_fp_id)·catalog 편입·failed 미계수."""
    _apply_settings(monkeypatch, tmp_path)
    bg_specs = [_bg_spec(bg_id="L01B01")]
    cp_map = _build_cp_map(
        tmp_path=tmp_path, fp_id="fp_a", bg_specs=bg_specs,
        plan=_camera_geometry_na_plan("fp_a"),
    )
    step = _new_step(tmp_path, cp_map)

    with patch(
        "app.core.steps.background_render_step._resolve_openai_client",
        return_value=MagicMock(),
    ), patch(
        "app.modules.pipeline.background_render.render_one_background",
        side_effect=_success_renderer(),
    ) as mock_render:
        result = step._execute()

    mock_render.assert_called_once()
    entry = result["data"]["groups"]["L01B01"]
    assert entry["status"] == "ok"
    assert entry["shot_aware_plan_mode"] == "camera_geometry_direct_plate"
    assert entry["render_degraded"] is True
    assert entry["fallback_reason"] == "no_viable_camera_look_at_pair"
    assert entry["source_fp_id"] == "fp_a"
    assert result["completed_count"] == 1
    assert result["failed_count"] == 0
    catalog = result["data"]["bg_reference_catalog"]
    assert any(
        c.get("source_kind") == "camera_geometry_direct_plate"
        and c.get("source_fp_id") == "fp_a"
        for c in catalog
    )


def test_camera_geometry_na_without_t2i_prompt_stays_failed(
    tmp_path, monkeypatch,
):
    """t2i_prompt 결손=direct plate 불가 — typed reason 이어도 fail-closed."""
    _apply_settings(monkeypatch, tmp_path)
    bg_specs = [_bg_spec(bg_id="L01B01")]
    cp_map = _build_cp_map(
        tmp_path=tmp_path, fp_id="fp_a", bg_specs=bg_specs,
        plan=_camera_geometry_na_plan("fp_a"),
    )
    cp_map["background_prompt"]["data"]["backgrounds"]["L01B01"][
        "t2i_prompt"
    ] = ""
    step = _new_step(tmp_path, cp_map)

    with patch(
        "app.core.steps.background_render_step._resolve_openai_client",
        return_value=MagicMock(),
    ), patch(
        "app.modules.pipeline.background_render.render_one_background",
    ) as mock_render:
        result = step._execute()

    mock_render.assert_not_called()
    entry = result["data"]["groups"]["L01B01"]
    assert entry["status"] == "failed"
    assert result["failed_count"] == 1


def test_generic_not_applicable_plan_stays_failed(tmp_path, monkeypatch):
    """reason code 없는 generic not_applicable(provider 미배선 등)=기존
    fail-closed 유지 — typed 경로가 일반 NA 를 세탁하지 않는다."""
    _apply_settings(monkeypatch, tmp_path)
    bg_specs = [_bg_spec(bg_id="L01B01")]
    plan = _camera_geometry_na_plan("fp_a")
    del plan["not_applicable_reason_code"]
    cp_map = _build_cp_map(
        tmp_path=tmp_path, fp_id="fp_a", bg_specs=bg_specs, plan=plan,
    )
    step = _new_step(tmp_path, cp_map)

    with patch(
        "app.core.steps.background_render_step._resolve_openai_client",
        return_value=MagicMock(),
    ), patch(
        "app.modules.pipeline.background_render.render_one_background",
    ) as mock_render:
        result = step._execute()

    mock_render.assert_not_called()
    entry = result["data"]["groups"]["L01B01"]
    assert entry["status"] == "failed"
    assert "not production_clear" in entry["render_error"]
    assert result["failed_count"] == 1