# Phase 4 — chain_bg + floor plan integration Implementation Plan

> **For agentic workers:** Use `superpowers:subagent-driven-development` to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking.

**Goal:** `background_mode=floor_plan_anchored` 시 `chain_bg_render`가 도면 PNG를 1순위 ref로 사용하는 multi-image edit. `chain_bg_planning`은 도면 prompt를 best-effort로 user_prompt에 prepend. mode=off / chain_only는 회귀 0건.

**Architecture:** 단일 토글 `background_mode` 분기. `location_floor_plan` checkpoint에서 `image_path` + `prompt_text`를 로드. `render_node_image`가 multi-image edit 지원. chain_bg_render manifest depends_on에 `location_floor_plan` 추가 (`not_applicable`도 satisfied). chain_bg_planning은 manifest 변경 없이 코드 내 best-effort 로드.

**Tech Stack:** Python 3.11, pytest, OpenAI SDK (gpt-image-2 multi-image edit), 기존 background_chain pipeline.

**Spec:** `docs/2026-04-29-phase4-chain-bg-floor-plan-design.md`

---

## File Structure

| 파일 | 책임 | 변경 종류 |
|---|---|---|
| `backend/app/core/step_manifest.py` | chain_bg_render depends_on `+ location_floor_plan` | 1 line edit |
| `backend/app/core/steps/background_chain_render_step.py` | `_load_floor_plan_paths` + run_background_chain_render 인자 | 함수 + 호출 |
| `backend/app/core/steps/background_chain_planning_step.py` | `_load_floor_plan_prompts` (best-effort) + 호출 | 함수 + 호출 |
| `backend/app/modules/pipeline/background_chain_render.py` | `render_one_location`에 floor_plan_path 인자 + `render_node_image` multi-image edit | 다단계 수정 |
| `backend/app/modules/pipeline/background_chain_planning.py` | user_prompt 빌드 시 floor_plan_prompts prepend | 함수 인자 + prepend |
| `backend/tests/core/test_phase4_floor_plan_chain.py` | **신규** — 단위 테스트 ~12개 | 신규 |
| `backend/tests/pipeline/test_background_chain_render.py` | 기존 회귀 + multi-image edit 신규 케이스 | append |

**프롬프트 파일 변경 없음** (chain_bg_render / chain_bg_planning 모두 system.md 변경 없음 — input은 user_prompt 빌드에서 처리).

---

## Task 1: floor plan checkpoint loader (BackgroundChainRenderStep)

**Files:**
- Modify: `backend/app/core/steps/background_chain_render_step.py`
- Test: `backend/tests/core/test_phase4_floor_plan_chain.py` (신규)

- [ ] **Step 1.1: 신규 테스트 파일 + 첫 3 테스트**

`backend/tests/core/test_phase4_floor_plan_chain.py`:

```python
"""Phase 4 — chain_bg + floor plan integration 단위 테스트."""
from __future__ import annotations

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

import pytest


def test_load_floor_plan_paths_when_off_returns_empty(tmp_path, monkeypatch):
    """background_mode='off' (location_floor_plan checkpoint 없음) → 빈 dict."""
    from app.core.steps.background_chain_render_step import BackgroundChainRenderStep

    monkeypatch.setattr("app.core.config.settings.projects_dir", str(tmp_path))
    runner = BackgroundChainRenderStep.__new__(BackgroundChainRenderStep)
    runner.project_id = "p"
    runner.episode_id = "e"

    paths = runner._load_floor_plan_paths()
    assert paths == {}


def test_load_floor_plan_paths_skips_failed_status(tmp_path, monkeypatch):
    """status='failed' location은 path 매핑에 미포함."""
    from app.core.steps.background_chain_render_step import BackgroundChainRenderStep

    ckpt = tmp_path / "p" / "checkpoints" / "episodes" / "e" / "location_floor_plan"
    ckpt.mkdir(parents=True)
    (ckpt / "manifest.json").write_text(json.dumps({
        "data": {"locations": [
            {"id": "L05", "image_path": "p/images/e/floor_plan/L05.png", "status": "ok"},
            {"id": "L02", "image_path": "p/images/e/floor_plan/L02.png", "status": "failed"},
        ]},
    }))
    # L05 PNG 파일 만들기 (exists check 통과)
    png_dir = tmp_path / "p" / "images" / "e" / "floor_plan"
    png_dir.mkdir(parents=True)
    (png_dir / "L05.png").write_bytes(b"x" * 2048)

    monkeypatch.setattr("app.core.config.settings.projects_dir", str(tmp_path / "p"))
    runner = BackgroundChainRenderStep.__new__(BackgroundChainRenderStep)
    runner.project_id = "p"
    runner.episode_id = "e"

    paths = runner._load_floor_plan_paths()
    assert "L05" in paths
    assert "L02" not in paths


def test_load_floor_plan_paths_skips_missing_file(tmp_path, monkeypatch):
    """체크포인트에 status=ok지만 PNG 파일 없으면 제외."""
    from app.core.steps.background_chain_render_step import BackgroundChainRenderStep

    ckpt = tmp_path / "p" / "checkpoints" / "episodes" / "e" / "location_floor_plan"
    ckpt.mkdir(parents=True)
    (ckpt / "manifest.json").write_text(json.dumps({
        "data": {"locations": [
            {"id": "L05", "image_path": "p/images/e/floor_plan/missing.png", "status": "ok"},
        ]},
    }))

    monkeypatch.setattr("app.core.config.settings.projects_dir", str(tmp_path / "p"))
    runner = BackgroundChainRenderStep.__new__(BackgroundChainRenderStep)
    runner.project_id = "p"
    runner.episode_id = "e"

    paths = runner._load_floor_plan_paths()
    assert paths == {}
```

- [ ] **Step 1.2: 테스트 실행 — 3 FAIL (`AttributeError: ... has no attribute '_load_floor_plan_paths'`)**

```bash
cd /Users/manta/Documents/Projects/TheRoad-I1/backend
.venv/bin/pytest tests/core/test_phase4_floor_plan_chain.py -v -k "load_floor_plan_paths"
```

- [ ] **Step 1.3: `_load_floor_plan_paths` 추가**

In `backend/app/core/steps/background_chain_render_step.py`:

```python
def _load_floor_plan_paths(self) -> Dict[str, Path]:
    """location_floor_plan checkpoint에서 status='ok' + PNG 파일 존재하는 매핑.

    Phase 4: background_mode=floor_plan_anchored 시 chain_bg_render의 1순위 ref.
    체크포인트 없거나 모든 location 실패면 빈 dict (mode=off / chain_only 회귀 보장).
    """
    cp = self._load_prev_checkpoint("location_floor_plan")
    if not cp:
        return {}
    locations = cp.get("data", {}).get("locations", []) or []
    out: Dict[str, Path] = {}
    from app.core.config import settings
    projects_root = Path(settings.projects_dir).parent
    for loc in locations:
        if loc.get("status") != "ok":
            continue
        rel = loc.get("image_path", "")
        if not rel:
            continue
        p = projects_root / rel
        if p.exists():
            out[loc["id"]] = p
    return out
```

- [ ] **Step 1.4: 테스트 통과 확인 (3 PASS)**

- [ ] **Step 1.5: Commit**

```bash
git add backend/app/core/steps/background_chain_render_step.py \
        backend/tests/core/test_phase4_floor_plan_chain.py
git commit -m "$(cat <<'EOF'
feat(phase4): _load_floor_plan_paths in BackgroundChainRenderStep

location_floor_plan checkpoint에서 status='ok' + PNG 파일 존재하는 location만 매핑.
mode=off 시 빈 dict 반환 (회귀 보장). Phase 4의 chain_bg_render multi-image edit
입력으로 사용 예정.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
EOF
)"
```

---

## Task 2: floor plan prompts loader (BackgroundChainPlanningStep)

**Files:**
- Modify: `backend/app/core/steps/background_chain_planning_step.py`
- Test: `backend/tests/core/test_phase4_floor_plan_chain.py`

- [ ] **Step 2.1: 테스트 추가 (3 cases)**

```python
def test_load_floor_plan_prompts_when_off_returns_empty(tmp_path, monkeypatch):
    """background_mode='off' (체크포인트 없음) → 빈 dict."""
    from app.core.steps.background_chain_planning_step import BackgroundChainPlanningStep

    monkeypatch.setattr("app.core.config.settings.projects_dir", str(tmp_path))
    runner = BackgroundChainPlanningStep.__new__(BackgroundChainPlanningStep)
    runner.project_id = "p"
    runner.episode_id = "e"

    prompts = runner._load_floor_plan_prompts()
    assert prompts == {}


def test_load_floor_plan_prompts_extracts_ok_only(tmp_path, monkeypatch):
    """status='ok' location의 prompt_text만 추출."""
    from app.core.steps.background_chain_planning_step import BackgroundChainPlanningStep

    ckpt = tmp_path / "p" / "checkpoints" / "episodes" / "e" / "location_floor_plan"
    ckpt.mkdir(parents=True)
    (ckpt / "manifest.json").write_text(json.dumps({
        "data": {"locations": [
            {"id": "L05", "prompt_text": "Top-down floor plan of rooftop", "status": "ok"},
            {"id": "L02", "prompt_text": "ignored", "status": "failed"},
        ]},
    }))

    monkeypatch.setattr("app.core.config.settings.projects_dir", str(tmp_path / "p"))
    runner = BackgroundChainPlanningStep.__new__(BackgroundChainPlanningStep)
    runner.project_id = "p"
    runner.episode_id = "e"

    prompts = runner._load_floor_plan_prompts()
    assert "L05" in prompts
    assert "L02" not in prompts
    assert "rooftop" in prompts["L05"]


def test_load_floor_plan_prompts_skips_empty_text(tmp_path, monkeypatch):
    """prompt_text가 빈 문자열인 location은 제외."""
    from app.core.steps.background_chain_planning_step import BackgroundChainPlanningStep

    ckpt = tmp_path / "p" / "checkpoints" / "episodes" / "e" / "location_floor_plan"
    ckpt.mkdir(parents=True)
    (ckpt / "manifest.json").write_text(json.dumps({
        "data": {"locations": [
            {"id": "L05", "prompt_text": "", "status": "ok"},
        ]},
    }))

    monkeypatch.setattr("app.core.config.settings.projects_dir", str(tmp_path / "p"))
    runner = BackgroundChainPlanningStep.__new__(BackgroundChainPlanningStep)
    runner.project_id = "p"
    runner.episode_id = "e"

    prompts = runner._load_floor_plan_prompts()
    assert prompts == {}
```

- [ ] **Step 2.2: 테스트 실행 — 3 FAIL.

- [ ] **Step 2.3: `_load_floor_plan_prompts` 추가**

In `backend/app/core/steps/background_chain_planning_step.py`:

```python
def _load_floor_plan_prompts(self) -> Dict[str, str]:
    """location_floor_plan checkpoint에서 status='ok' + non-empty prompt_text 매핑.

    Phase 4: chain_bg_planning user_prompt에 도면 prompt를 prepend하여 LLM이
    floor plan 인식한 chain bg prompt 생성. mode=off / chain_only 시 빈 dict.
    """
    cp = self._load_prev_checkpoint("location_floor_plan")
    if not cp:
        return {}
    locations = cp.get("data", {}).get("locations", []) or []
    out: Dict[str, str] = {}
    for loc in locations:
        if loc.get("status") != "ok":
            continue
        text = (loc.get("prompt_text") or "").strip()
        if not text:
            continue
        out[loc["id"]] = text
    return out
```

- [ ] **Step 2.4: 3 PASS 확인.**

- [ ] **Step 2.5: Commit**

```bash
git add backend/app/core/steps/background_chain_planning_step.py \
        backend/tests/core/test_phase4_floor_plan_chain.py
git commit -m "$(cat <<'EOF'
feat(phase4): _load_floor_plan_prompts in BackgroundChainPlanningStep

location_floor_plan checkpoint에서 status='ok' + non-empty prompt_text 추출.
chain_bg_planning user_prompt prepend용. mode=off 시 빈 dict (회귀 보장).

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
EOF
)"
```

---

## Task 3: render_node_image multi-image edit 지원

**Files:**
- Modify: `backend/app/modules/pipeline/background_chain_render.py`
- Test: `backend/tests/core/test_phase4_floor_plan_chain.py`

- [ ] **Step 3.1: 테스트 추가 (3 cases)**

```python
def test_render_node_image_multi_image_edit(tmp_path, monkeypatch):
    """ref_paths 2개 → multi-image edit API 호출."""
    from app.modules.pipeline.background_chain_render import render_node_image

    ref1 = tmp_path / "floor_plan.png"
    ref2 = tmp_path / "parent.png"
    ref1.write_bytes(b"x" * 2048)
    ref2.write_bytes(b"y" * 2048)
    out = tmp_path / "out.png"

    client = MagicMock()
    fake_b64 = __import__("base64").b64encode(b"PNG" * 1024).decode()
    client.images.edit.return_value.data = [MagicMock(b64_json=fake_b64)]

    info = render_node_image(
        openai_client=client,
        image_model="gpt-image-2",
        prompt="x" * 50,
        out_path=out,
        ref_paths=[ref1, ref2],  # 신규 인자 (List)
        sanitizer=None,
        max_attempts=1,
    )
    assert info["status"] == "ok"
    assert client.images.edit.called
    # multi-image: image= 인자가 list여야 함
    call_kwargs = client.images.edit.call_args.kwargs
    assert isinstance(call_kwargs.get("image"), list)
    assert len(call_kwargs["image"]) == 2


def test_render_node_image_single_image_fallback(tmp_path, monkeypatch):
    """ref_paths 1개 → single-image edit (기존 동작 호환)."""
    from app.modules.pipeline.background_chain_render import render_node_image

    ref = tmp_path / "ref.png"
    ref.write_bytes(b"x" * 2048)
    out = tmp_path / "out.png"

    client = MagicMock()
    fake_b64 = __import__("base64").b64encode(b"PNG" * 1024).decode()
    client.images.edit.return_value.data = [MagicMock(b64_json=fake_b64)]

    info = render_node_image(
        openai_client=client,
        image_model="gpt-image-2",
        prompt="x" * 50,
        out_path=out,
        ref_paths=[ref],
        sanitizer=None,
        max_attempts=1,
    )
    assert info["status"] == "ok"
    # single image: image= 가 file-like (list 아님)
    call_kwargs = client.images.edit.call_args.kwargs
    assert not isinstance(call_kwargs.get("image"), list)


def test_render_node_image_text_only_when_no_refs(tmp_path):
    """ref_paths 빈 list → images.generate (text-only)."""
    from app.modules.pipeline.background_chain_render import render_node_image

    out = tmp_path / "out.png"
    client = MagicMock()
    fake_b64 = __import__("base64").b64encode(b"PNG" * 1024).decode()
    client.images.generate.return_value.data = [MagicMock(b64_json=fake_b64)]

    info = render_node_image(
        openai_client=client,
        image_model="gpt-image-2",
        prompt="x" * 50,
        out_path=out,
        ref_paths=[],
        sanitizer=None,
        max_attempts=1,
    )
    assert info["status"] == "ok"
    assert client.images.generate.called
    assert not client.images.edit.called
```

- [ ] **Step 3.2: 테스트 실행 — 3 FAIL (`ref_paths` 파라미터 미존재).**

- [ ] **Step 3.3: `render_node_image` 시그니처 변경 + multi-image 로직**

In `backend/app/modules/pipeline/background_chain_render.py`, `render_node_image`:

```python
def render_node_image(
    openai_client: Any,
    image_model: str,
    prompt: str,
    out_path: Path,
    ref_paths: Optional[List[Path]] = None,  # 변경: ref_path → List
    sanitizer: Optional[PromptSanitizer] = None,
    size: str = "1024x1024",
    quality: str = "high",
    max_attempts: int = 4,
    # 하위 호환: ref_path single 인자도 받음 (deprecated)
    ref_path: Optional[Path] = None,
) -> Dict[str, Any]:
    """단일 노드 이미지 생성. ref_paths 1개 → single edit, 2+ → multi-image edit, 0 → generate."""
    # 하위 호환: ref_path → ref_paths
    if ref_paths is None and ref_path is not None:
        ref_paths = [ref_path]
    if ref_paths is None:
        ref_paths = []

    # 존재하는 파일만 사용
    valid_refs = [p for p in ref_paths if p and p.exists()]

    info: Dict[str, Any] = {
        "status": "failed",
        "attempts": 0,
        "strategies": [],
        "final_block_reason": None,
        "ref_used": "text_only" if not valid_refs else f"refs_{len(valid_refs)}",
    }
    current_prompt = prompt

    for attempt in range(1, max_attempts + 1):
        info["attempts"] = attempt
        try:
            if len(valid_refs) >= 2:
                # multi-image edit
                files = [p.open("rb") for p in valid_refs]
                try:
                    resp = openai_client.images.edit(
                        model=image_model,
                        image=files,  # list
                        prompt=current_prompt,
                        size=size,
                        quality=quality,
                        n=1,
                    )
                finally:
                    for f in files:
                        try:
                            f.close()
                        except Exception:
                            pass
            elif len(valid_refs) == 1:
                with valid_refs[0].open("rb") as f:
                    resp = openai_client.images.edit(
                        model=image_model,
                        image=f,
                        prompt=current_prompt,
                        size=size,
                        quality=quality,
                        n=1,
                    )
            else:
                resp = openai_client.images.generate(
                    model=image_model,
                    prompt=current_prompt,
                    size=size,
                    quality=quality,
                    n=1,
                )
            b64 = resp.data[0].b64_json
            if not b64:
                raise RuntimeError("empty b64 response")
            out_path.write_bytes(base64.b64decode(b64))
            info["status"] = "ok"
            return info
        except Exception as exc:
            # 기존 moderation block + retry 로직 그대로
            ... # (변경 없음)

    return info
```

⚠️ **하위 호환성**: 기존 호출이 `ref_path=path` (single) 사용 중이면 `ref_path` 파라미터로 받아 `ref_paths=[ref_path]`로 변환. 점진적 migration. 모든 caller가 `ref_paths`로 옮겨가면 `ref_path` 파라미터 deprecation 후 제거.

- [ ] **Step 3.4: 3 PASS 확인.**

- [ ] **Step 3.5: 회귀 테스트 — 기존 `tests/pipeline/test_background_chain_render.py` 17 PASS 확인.**

```bash
.venv/bin/pytest tests/pipeline/test_background_chain_render.py -v
```

- [ ] **Step 3.6: Commit**

```bash
git add backend/app/modules/pipeline/background_chain_render.py \
        backend/tests/core/test_phase4_floor_plan_chain.py
git commit -m "$(cat <<'EOF'
feat(phase4): render_node_image multi-image edit 지원

ref_paths: List[Path] 인자 추가 (backward-compat: ref_path single 인자 유지).
- 0 refs → images.generate (text-only)
- 1 ref → images.edit (single, 기존)
- 2+ refs → images.edit (multi-image, gpt-image-2 native)
ref_used 라벨에 ref count 반영.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
EOF
)"
```

---

## Task 4: render_one_location에 floor_plan_path 인자 + 우선순위 로직

**Files:**
- Modify: `backend/app/modules/pipeline/background_chain_render.py`
- Test: `backend/tests/core/test_phase4_floor_plan_chain.py`

- [ ] **Step 4.1: 테스트 추가 (2 cases)**

```python
def test_render_one_location_floor_plan_priority(tmp_path, monkeypatch):
    """floor_plan + parent 모두 있을 때 floor_plan이 1순위, parent 2순위."""
    from app.modules.pipeline.background_chain_render import render_one_location

    floor_plan = tmp_path / "fp.png"
    floor_plan.write_bytes(b"x" * 2048)
    image_dir = tmp_path / "chain"
    image_dir.mkdir()

    fake_b64 = __import__("base64").b64encode(b"PNG" * 1024).decode()
    client = MagicMock()
    client.images.edit.return_value.data = [MagicMock(b64_json=fake_b64)]

    location_data = {
        "nodes": [
            {"id": "n1", "shot_ids": ["S5_Shot1"]},
            {"id": "n2", "parent_id": "n1", "shot_ids": ["S5_Shot2"]},
        ],
        "execution_order": ["n1", "n2"],
        "location_description": "rooftop",
    }

    with patch("app.modules.pipeline.background_chain_render.generate_node_prompt",
              return_value=("photo prompt", [])):
        result = render_one_location(
            location_id="L05",
            location_data=location_data,
            image_dir=image_dir,
            location_ref_paths={},  # location ref 없음
            openai_client=client,
            image_model="gpt-image-2",
            sanitizer=None,
            max_attempts=1,
            floor_plan_path=floor_plan,  # 신규
        )

    # n1: refs = [floor_plan] (parent 없음, location ref 없음)
    # n2: refs = [floor_plan, parent(n1)]
    # 첫 호출(n1): single ref (floor_plan)
    # 둘째 호출(n2): multi (floor_plan + parent)
    calls = client.images.edit.call_args_list
    assert len(calls) == 2
    # 둘째 call: image=list, len 2
    assert isinstance(calls[1].kwargs["image"], list)
    assert len(calls[1].kwargs["image"]) == 2


def test_render_one_location_no_floor_plan_uses_legacy_flow(tmp_path):
    """floor_plan_path=None 시 기존 single-ref flow 동작."""
    from app.modules.pipeline.background_chain_render import render_one_location

    image_dir = tmp_path / "chain"
    image_dir.mkdir()
    location_ref = tmp_path / "loc.png"
    location_ref.write_bytes(b"x" * 2048)

    fake_b64 = __import__("base64").b64encode(b"PNG" * 1024).decode()
    client = MagicMock()
    client.images.edit.return_value.data = [MagicMock(b64_json=fake_b64)]

    location_data = {
        "nodes": [{"id": "n1", "shot_ids": ["S5_Shot1"]}],
        "execution_order": ["n1"],
        "location_description": "rooftop",
    }

    with patch("app.modules.pipeline.background_chain_render.generate_node_prompt",
              return_value=("p", [])):
        result = render_one_location(
            location_id="L05",
            location_data=location_data,
            image_dir=image_dir,
            location_ref_paths={"L05": location_ref},
            openai_client=client,
            image_model="gpt-image-2",
            sanitizer=None,
            max_attempts=1,
            floor_plan_path=None,  # 명시적 None
        )
    # location_ref 단일 사용 — image= 가 list 아님
    call_kwargs = client.images.edit.call_args.kwargs
    assert not isinstance(call_kwargs.get("image"), list)
```

- [ ] **Step 4.2: 2 FAIL 확인.

- [ ] **Step 4.3: render_one_location 시그니처 + 우선순위 로직**

```python
def render_one_location(
    location_id: str,
    location_data: Dict[str, Any],
    image_dir: Path,
    location_ref_paths: Dict[str, Path],
    openai_client: Any,
    image_model: str,
    sanitizer: Optional[PromptSanitizer],
    size: str = "1024x1024",
    quality: str = "high",
    max_attempts: int = 4,
    opik_metadata: Optional[Dict[str, Any]] = None,
    shot_meta_by_id: Optional[Dict[str, Dict[str, Any]]] = None,
    floor_plan_path: Optional[Path] = None,  # 신규
) -> Dict[str, Any]:
    ...
    for node_id in execution_order:
        ...
        # 기존 ref_path 단일 결정 로직 → ref_paths List 빌드로 대체
        ref_paths: List[Path] = []
        # 1순위: floor plan (있으면)
        if floor_plan_path is not None and floor_plan_path.exists():
            ref_paths.append(floor_plan_path)
        # 2순위: parent PNG (자식 node)
        if parent_id and parent_id in rendered_paths:
            candidate = rendered_paths[parent_id]
            if candidate.exists():
                ref_paths.append(candidate)
                ref_used_label = "parent"
        # 3순위: location ref (root anchor)
        if not ref_paths or (parent_id is None and len(ref_paths) < 2):
            candidate = _resolve_location_ref_path(location_id, location_ref_paths)
            if candidate is not None:
                ref_paths.append(candidate)
                ref_used_label = "location"

        ref_used_label = "floor_plan_only" if (floor_plan_path and len(ref_paths) == 1) else (
            "floor_plan+parent" if (floor_plan_path and parent_id and parent_id in rendered_paths) else (
                "floor_plan+location" if (floor_plan_path and not parent_id) else "legacy"
            )
        )

        # render_node_image 호출 — 기존 ref_path → ref_paths List
        info = render_node_image(
            openai_client=openai_client,
            image_model=image_model,
            prompt=t2i_prompt,
            out_path=out_path,
            ref_paths=ref_paths,
            sanitizer=sanitizer,
            ...
        )
        ...
        enriched_nodes[node_id] = {
            ...,
            "ref_used": ref_used_label,
        }
```

⚠️ ref_used 라벨 결정 로직은 위 분기 그대로 사용. 코드 작성 시 명확하게.

- [ ] **Step 4.4: 2 PASS + 회귀 17 PASS 확인.**

- [ ] **Step 4.5: Commit**

```bash
git add backend/app/modules/pipeline/background_chain_render.py \
        backend/tests/core/test_phase4_floor_plan_chain.py
git commit -m "$(cat <<'EOF'
feat(phase4): render_one_location에 floor_plan_path 인자 + 우선순위 ref

ref_paths 우선순위: floor_plan(1순위) → parent(자식)/location(root)(2순위).
mode=off (floor_plan_path=None) 시 기존 단일 ref 동작.
ref_used 라벨에 floor_plan_only/floor_plan+parent/floor_plan+location 추가.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
EOF
)"
```

---

## Task 5: BackgroundChainRenderStep에서 floor_plan_paths 전달

**Files:**
- Modify: `backend/app/core/steps/background_chain_render_step.py`
- Modify: `backend/app/modules/pipeline/background_chain_render.py` (`run_background_chain_render`)
- Test: `backend/tests/core/test_phase4_floor_plan_chain.py`

- [ ] **Step 5.1: 테스트 추가 (1 case — Step 통합 smoke)**

```python
def test_step_passes_floor_plan_paths_to_pipeline(tmp_path, monkeypatch):
    """BackgroundChainRenderStep._execute가 floor_plan_paths를 pipeline에 전달."""
    from app.core.steps.background_chain_render_step import BackgroundChainRenderStep

    # 가짜 체크포인트 + 도면 PNG
    pid_root = tmp_path / "p"
    ckpt_dir = pid_root / "checkpoints" / "episodes" / "e"
    (ckpt_dir / "background_chain_planning").mkdir(parents=True)
    (ckpt_dir / "background_chain_planning" / "manifest.json").write_text(json.dumps({
        "data": {"locations": {"L05": {"nodes": [], "execution_order": []}}},
    }))
    (ckpt_dir / "location_floor_plan").mkdir(parents=True)
    (ckpt_dir / "location_floor_plan" / "manifest.json").write_text(json.dumps({
        "data": {"locations": [
            {"id": "L05", "image_path": "p/images/e/floor_plan/L05.png", "status": "ok"},
        ]},
    }))
    fp_dir = pid_root / "images" / "e" / "floor_plan"
    fp_dir.mkdir(parents=True)
    (fp_dir / "L05.png").write_bytes(b"x" * 2048)

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

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

    captured = {}
    def fake_run(**kwargs):
        captured.update(kwargs)
        return {"locations": {}, "_failed_count": 0}

    monkeypatch.setattr(
        "app.modules.pipeline.background_chain_render.run_background_chain_render",
        fake_run,
    )

    runner._execute(mode="resume")
    assert "floor_plan_paths" in captured
    assert "L05" in captured["floor_plan_paths"]
```

- [ ] **Step 5.2: FAIL 확인.

- [ ] **Step 5.3: BackgroundChainRenderStep._execute 수정**

In `backend/app/core/steps/background_chain_render_step.py`:

```python
def _execute(self, mode: str = "resume") -> Dict[str, Any]:
    ...
    # 기존 location_ref_paths 로드 다음에 추가:
    floor_plan_paths = self._load_floor_plan_paths()  # Phase 4 신규
    ...
    result = run_background_chain_render(
        ...
        floor_plan_paths=floor_plan_paths,  # 신규
    )
    ...
```

In `backend/app/modules/pipeline/background_chain_render.py`, `run_background_chain_render`:

```python
def run_background_chain_render(
    *,
    planning_data: Dict[str, Any],
    image_dir: Path,
    location_ref_paths: Dict[str, Path],
    openai_client: Any,
    image_model: str = "gpt-image-2",
    sanitizer: Optional[PromptSanitizer] = None,
    size: str = "1024x1024",
    quality: str = "high",
    max_attempts: int = 4,
    opik_metadata: Optional[Dict[str, Any]] = None,
    shot_meta_by_id: Optional[Dict[str, Dict[str, Any]]] = None,
    floor_plan_paths: Optional[Dict[str, Path]] = None,  # 신규
) -> Dict[str, Any]:
    ...
    floor_plan_paths = floor_plan_paths or {}
    ...
    def _process(loc_id: str, data: Dict[str, Any]) -> Tuple[str, Dict[str, Any]]:
        ...
        return loc_id, render_one_location(
            ...,
            floor_plan_path=floor_plan_paths.get(loc_id),  # 신규
        )
```

- [ ] **Step 5.4: 1 PASS 확인 + 전체 회귀 PASS.**

- [ ] **Step 5.5: Commit**

```bash
git add backend/app/core/steps/background_chain_render_step.py \
        backend/app/modules/pipeline/background_chain_render.py \
        backend/tests/core/test_phase4_floor_plan_chain.py
git commit -m "$(cat <<'EOF'
feat(phase4): BackgroundChainRenderStep + run_background_chain_render에
floor_plan_paths 전달

Step._execute가 _load_floor_plan_paths 결과를 pipeline에 전달.
run_background_chain_render는 location별로 floor_plan_path를 dispatch.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
EOF
)"
```

---

## Task 6: chain_bg_planning에 floor plan prompt prepend

**Files:**
- Modify: `backend/app/core/steps/background_chain_planning_step.py`
- Modify: `backend/app/modules/pipeline/background_chain_planning.py`
- Test: `backend/tests/core/test_phase4_floor_plan_chain.py`

- [ ] **Step 6.1: 테스트 추가 (2 cases)**

```python
def test_chain_bg_planning_prepends_floor_plan_prompt(tmp_path):
    """floor_plan_prompts 제공 시 user_prompt 앞에 [FLOOR PLAN] 블록 prepend."""
    from app.modules.pipeline.background_chain_planning import build_planning_user_prompt

    user_prompt = build_planning_user_prompt(
        location_id="L05",
        base_prompt="LOCATION L05 — chain bg planning task",
        floor_plan_prompts={"L05": "Top-down floor plan of rooftop apartment..."},
    )
    assert user_prompt.startswith("[FLOOR PLAN")
    assert "rooftop apartment" in user_prompt
    assert "LOCATION L05" in user_prompt


def test_chain_bg_planning_no_prepend_when_empty(tmp_path):
    """floor_plan_prompts={} 시 base_prompt 그대로."""
    from app.modules.pipeline.background_chain_planning import build_planning_user_prompt

    user_prompt = build_planning_user_prompt(
        location_id="L05",
        base_prompt="LOCATION L05 — chain bg planning task",
        floor_plan_prompts={},
    )
    assert not user_prompt.startswith("[FLOOR PLAN")
    assert user_prompt == "LOCATION L05 — chain bg planning task"
```

- [ ] **Step 6.2: 2 FAIL 확인.

- [ ] **Step 6.3: `build_planning_user_prompt` 함수 추가**

In `backend/app/modules/pipeline/background_chain_planning.py`:

```python
def build_planning_user_prompt(
    location_id: str,
    base_prompt: str,
    floor_plan_prompts: Dict[str, str],
) -> str:
    """chain_bg_planning user_prompt 빌드. floor plan prompt가 있으면 prepend."""
    fp = floor_plan_prompts.get(location_id, "").strip()
    if not fp:
        return base_prompt
    return (
        f"[FLOOR PLAN — spatial layout authority for location {location_id}]\n"
        f"{fp}\n\n"
        f"[CHAIN BG PLANNING TASK]\n"
        f"{base_prompt}"
    )
```

기존 user_prompt 빌드하는 함수 안에서 마지막에 위 헬퍼로 wrapping:

```python
# 기존 user_prompt 빌드 후
user_prompt = build_planning_user_prompt(
    location_id=loc_id,
    base_prompt=user_prompt,
    floor_plan_prompts=floor_plan_prompts,
)
```

`background_chain_planning.py` 의 메인 진입 함수가 `floor_plan_prompts: Dict[str, str]` 인자를 받도록 시그니처 추가.

In `backend/app/core/steps/background_chain_planning_step.py`:

```python
def _execute(self, mode: str = "resume") -> Dict[str, Any]:
    ...
    floor_plan_prompts = self._load_floor_plan_prompts()  # 신규
    ...
    result = run_background_chain_planning(
        ...,
        floor_plan_prompts=floor_plan_prompts,  # 신규
    )
```

- [ ] **Step 6.4: 2 PASS 확인 + chain_bg_planning 기존 테스트 회귀 PASS.**

- [ ] **Step 6.5: Commit**

```bash
git add backend/app/core/steps/background_chain_planning_step.py \
        backend/app/modules/pipeline/background_chain_planning.py \
        backend/tests/core/test_phase4_floor_plan_chain.py
git commit -m "$(cat <<'EOF'
feat(phase4): chain_bg_planning에 floor_plan_prompts prepend

build_planning_user_prompt 헬퍼 신설 — floor_plan_prompts.get(loc_id)이 비어있지
않으면 [FLOOR PLAN] + [CHAIN BG PLANNING TASK] 블록으로 prepend. 빈 dict면
base_prompt 그대로 (mode=off / chain_only 회귀 보장).

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
EOF
)"
```

---

## Task 7: manifest depends_on + 회귀 + 듀얼 리뷰

**Files:**
- Modify: `backend/app/core/step_manifest.py`
- Test: `backend/tests/core/test_phase4_floor_plan_chain.py`

- [ ] **Step 7.1: 테스트 추가 (1 case)**

```python
def test_chain_bg_render_depends_on_location_floor_plan():
    """chain_bg_render manifest의 depends_on에 location_floor_plan 포함."""
    from app.core.step_manifest import STEP_MANIFEST
    deps = STEP_MANIFEST["background_chain_render"]["depends_on"]
    assert "location_floor_plan" in deps
    # 기존 의존성도 그대로 유지
    assert "background_chain_planning" in deps
```

- [ ] **Step 7.2: FAIL 확인.

- [ ] **Step 7.3: manifest 수정**

```python
"background_chain_render": {
    ...,
    "depends_on": ["background_chain_planning", "location_floor_plan"],  # 추가
    ...
},
```

- [ ] **Step 7.4: PASS 확인 + 전체 회귀 PASS.**

```bash
.venv/bin/pytest tests/core tests/pipeline tests/test_step_manifest_v3.py tests/test_manifest_fields.py tests/test_step_catalog.py tests/test_pipeline_v3_e2e.py::TestStepOrdering -v --tb=short 2>&1 | tail -10
```

Expected: 모두 PASS. order 변경 없으므로 image>analysis 불변식 영향 없음.

- [ ] **Step 7.5: 듀얼 리뷰 (Codex `--effort minimal` + Claude)**

Phase 2/3 패턴. Codex timeout 시 Claude 단독 + 사용자 명시 승인.

- [ ] **Step 7.6: 리뷰 피드백 반영 commit (있으면).

- [ ] **Step 7.7: Commit (manifest)**

```bash
git add backend/app/core/step_manifest.py backend/tests/core/test_phase4_floor_plan_chain.py
git commit -m "$(cat <<'EOF'
feat(phase4): chain_bg_render manifest depends_on에 location_floor_plan 추가

mode=off 시 location_floor_plan이 not_applicable이지만 step_runner가 satisfied로
인정 (step_runner.py:99). cascade invalidation 자동.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
EOF
)"
```

---

## Task 8: 메모리 + 푸시

- [ ] **Step 8.1: 메모리 작성**

`session_20260429_phase4.md` — Phase 4 결과 기록.

- [ ] **Step 8.2: MEMORY.md 인덱스 추가**

- [ ] **Step 8.3: 사용자 명시 승인 후 push**

---

## 진행 위치

| Day | Phase | 상태 |
|---|---|---|
| 1 | P0 | ✅ aacccd9 |
| 3 | 1b | ✅ 553fea2 |
| 4 | 2 | ✅ 04840cc |
| 5~6 | 3 | ✅ 4eac12c |
| **7** | **4** | **🔵 본 plan** |
| 8 | 5 | 대기 |
| 9~10 | 6 | 대기 |

본 plan 완료 시 6/8 완료.

---

## Self-Review (controller before dispatch)

- [ ] Spec coverage: design doc 17 섹션이 모두 task로 커버됨?
- [ ] No placeholders: TBD/TODO 등 없음?
- [ ] Type consistency: 함수/필드 이름이 task 간 일관됨?
- [ ] 듀얼 리뷰 명시: Task 7.5에 명시됨?
- [ ] 회귀 가드 3중 명시: mode default off + 기존 호출 호환 + 빈 dict fallback?