# Phase 7 — Background Pipeline Redesign Implementation Plan > **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking. **Goal:** Phase 5의 4 step(`background_planner` / `background_chain_planning` / `location_floor_plan` / `background_chain_render`)을 6 step master-plan 일원화 architecture로 완전 대체. **Architecture:** Step 1(분류) + Step 2(group별 master plan, 그룹간 LLM 병렬) + Step 3-4(도면 prompt + render, level 병렬) + Step 5-6(배경 prompt + render, level 병렬). 모든 ID/순서/ref 결정은 Step 2의 단일 master plan에서 내려와 Phase 5의 ID mismatch / sub-room 무작위 / plot-critical 누락을 해결. **Tech Stack:** Python 3.12, FastAPI, SQLAlchemy + Alembic, gpt-5.5(text), gpt-image-2(image), pytest, ThreadPoolExecutor. **Spec 참조:** `docs/superpowers/specs/2026-04-29-phase7-background-redesign.md` --- ## File Structure ### 신규 파일 | 파일 | 책임 | |---|---| | `backend/app/core/steps/background_classify_step.py` | Step 1 entry: 의존 체크포인트 로드 → classify LLM 호출 | | `backend/app/core/steps/background_master_plan_step.py` | Step 2 entry: 그룹별 LLM 호출(병렬) → invariant 검증 | | `backend/app/core/steps/floor_plan_prompt_step.py` | Step 3 entry: 도면별 LLM 호출(level 병렬) | | `backend/app/core/steps/floor_plan_render_step.py` | Step 4 entry: 도면 PNG gpt-image-2 (level 병렬) | | `backend/app/core/steps/background_prompt_step.py` | Step 5 entry: 배경별 LLM 호출(level 병렬) | | `backend/app/core/steps/background_render_step.py` | Step 6 entry: 배경 PNG gpt-image-2 (level 병렬) | | `backend/app/modules/pipeline/background_classify.py` | classify LLM 호출/검증 helper | | `backend/app/modules/pipeline/background_master_plan.py` | master plan LLM 호출/invariant validators | | `backend/app/modules/pipeline/floor_plan_prompt.py` | 도면 prompt LLM 호출/검증 | | `backend/app/modules/pipeline/floor_plan_render.py` | 도면 PNG gpt-image-2 helper | | `backend/app/modules/pipeline/background_prompt.py` | 배경 prompt LLM 호출/검증 | | `backend/app/modules/pipeline/background_render.py` | 배경 PNG gpt-image-2 helper | | `backend/app/modules/pipeline/_dag_levels.py` | generic level batch DAG helper (Phase 5.3 일반화) | | `prompts/_base/background_classify/1.YYYYMMDDHHmm/{system.md,user_template.md,schema.json}` | Step 1 prompt | | `prompts/_base/background_master_plan/1.YYYYMMDDHHmm/{...}` | Step 2 prompt | | `prompts/_base/floor_plan_prompt/1.YYYYMMDDHHmm/{...}` | Step 3 prompt | | `prompts/_base/background_prompt/1.YYYYMMDDHHmm/{...}` | Step 5 prompt | | `backend/tests/pipeline/test_background_classify.py` | Step 1 단위 테스트 | | `backend/tests/pipeline/test_background_master_plan.py` | Step 2 단위 테스트 (invariants) | | `backend/tests/pipeline/test_floor_plan_prompt.py` | Step 3 단위 테스트 | | `backend/tests/pipeline/test_floor_plan_render.py` | Step 4 단위 테스트 | | `backend/tests/pipeline/test_background_prompt.py` | Step 5 단위 테스트 | | `backend/tests/pipeline/test_background_render.py` | Step 6 단위 테스트 | | `backend/tests/pipeline/test_dag_levels.py` | DAG helper 단위 테스트 | ### 수정 파일 | 파일 | 변경 | |---|---| | `backend/app/core/step_manifest.py` | 6 신규 entry 추가 / Phase 5 entry 4개 lifecycle="deprecated" + applicability="disabled" | | `backend/app/core/steps/__init__.py` | STEP_CLASSES에 6 신규 step 등록 | | `backend/app/modules/llm/llm_client.py` | PIPELINE_STEPS에 6 신규 entry 추가 | | `backend/app/core/applicability.py` | `if_background_mode` validator 추가 (legacy `if_floor_plan_mode`는 alias) | | `backend/app/core/config.py` | `background_mode` literal: `off`/`on`/`floor_plan_anchored`(legacy alias) | | `backend/app/core/steps/scene_context_loader.py` | (변경 없음 — Phase 7 `background_render`가 Phase 5와 동일한 `data.groups` shape 산출) | | `.env.example` | `BACKGROUND_MODE=off` default + 주석 갱신 | ### 변경 없는 파일 (공통 utility 재사용) - `backend/app/modules/prompt_loader.py` — load_prompt / load_schema 그대로 - `backend/app/modules/llm/llm_client.py:call_structured` — LLM 호출 그대로 - `backend/app/modules/prompt_sanitizer.py` — moderation block 처리 그대로 - `backend/app/core/step_runner.py` — StepRunner / check_applicability 그대로 --- ## Common Patterns (모든 task에서 따름) 1. **TDD**: 매 단위 작업은 failing test → 최소 구현 → pass → commit 2. **시나리오 의존성 0**: ID/snake_case 필드는 ASCII only. 한글/한자/kana 검출 시 invariant 위반. 3. **truncation 금지**: scene_segments / world_rules는 무절단 inject (CLAUDE.md absolute rule) 4. **PROMPT_VERSION + SCHEMA_VERSION**: 각 step의 `_config_hash`에 포함하여 stale checkpoint 자동 invalidation 5. **PNG path traversal 가드**: `_SAFE_NODE_ID_RE = r'^[a-z0-9][a-z0-9_]*$'` + `out_path.resolve().relative_to(image_dir.resolve())` 2중 가드 6. **LLM retry**: 3회. invariant 위반 = ValueError → retry 대상. 일시적 LLM/네트워크 = RuntimeError/TimeoutError/ConnectionError → retry. TypeError/AttributeError/KeyError = fail-fast. 7. **frequent commits**: 각 task의 step은 5-15커밋 정도. 한 task = 한 PR 단위. --- ## Task 1: Foundation — generic DAG levels helper 추출 **Why first**: Step 4(`floor_plan_render`)와 Step 6(`background_render`) 모두 level batch 병렬화가 필요. Phase 5.3의 `_compute_chain_bg_levels`를 generic하게 추출하여 양쪽에서 재사용. 이걸 먼저 안 하면 두 step에서 코드 중복. **Files:** - Create: `backend/app/modules/pipeline/_dag_levels.py` - Create: `backend/tests/pipeline/test_dag_levels.py` - Modify: `backend/app/modules/pipeline/background_chain_render.py:793-838` (`_compute_chain_bg_levels` → wrap around generic helper) - [ ] **Step 1.1: failing test 작성** ```python # backend/tests/pipeline/test_dag_levels.py import pytest from app.modules.pipeline._dag_levels import compute_dag_levels def test_all_roots_single_level(): items = {"a": {"parent_id": ""}, "b": {"parent_id": ""}, "c": {"parent_id": ""}} order = ["a", "b", "c"] renderable = {"a", "b", "c"} levels = compute_dag_levels(order, items, renderable) assert levels == [["a", "b", "c"]] def test_linear_chain_4_levels(): items = { "n0": {"parent_id": ""}, "n1": {"parent_id": "n0"}, "n2": {"parent_id": "n1"}, "n3": {"parent_id": "n2"}, } order = ["n0", "n1", "n2", "n3"] renderable = set(order) levels = compute_dag_levels(order, items, renderable) assert levels == [["n0"], ["n1"], ["n2"], ["n3"]] def test_fanout_then_merge(): items = { "root": {"parent_id": ""}, "a": {"parent_id": "root"}, "b": {"parent_id": "root"}, "merge": {"parent_id": "a"}, } order = ["root", "a", "b", "merge"] renderable = set(order) levels = compute_dag_levels(order, items, renderable) assert levels == [["root"], ["a", "b"], ["merge"]] def test_parent_outside_renderable_treated_as_root(): items = {"a": {"parent_id": "skipped_node"}, "b": {"parent_id": ""}} order = ["a", "b"] renderable = {"a", "b"} # "skipped_node" not renderable levels = compute_dag_levels(order, items, renderable) assert levels == [["a", "b"]] def test_empty_input(): assert compute_dag_levels([], {}, set()) == [] def test_cycle_flushed_as_final_batch(caplog): items = {"a": {"parent_id": "b"}, "b": {"parent_id": "a"}} order = ["a", "b"] renderable = {"a", "b"} levels = compute_dag_levels(order, items, renderable) # cycle 검출 시 잔여 그룹을 마지막 level로 flush assert levels == [["a", "b"]] assert any("unresolved" in r.message.lower() for r in caplog.records) def test_preserves_order_within_level(): items = {"z": {"parent_id": ""}, "a": {"parent_id": ""}, "m": {"parent_id": ""}} order = ["z", "a", "m"] # 입력 순서 보존 renderable = set(order) levels = compute_dag_levels(order, items, renderable) assert levels == [["z", "a", "m"]] ``` - [ ] **Step 1.2: 테스트 실행 — fail 확인** Run: `cd backend && pytest tests/pipeline/test_dag_levels.py -v` Expected: FAIL with "ModuleNotFoundError: app.modules.pipeline._dag_levels" - [ ] **Step 1.3: helper 구현** ```python # backend/app/modules/pipeline/_dag_levels.py """Generic DAG → level batches helper. Phase 5.3 background_chain_render의 `_compute_chain_bg_levels`를 일반화. Phase 7 floor_plan_render / background_render에서 재사용. 각 level[i]는 parent가 모두 이전 level에서 완료(또는 renderable_set 외부) 상태인 노드들. 같은 level 내 노드끼리는 독립 → ThreadPool 병렬 안전. """ from __future__ import annotations import logging from typing import Any, Dict, List, Set logger = logging.getLogger(__name__) def compute_dag_levels( order: List[str], items: Dict[str, Dict[str, Any]], renderable_set: Set[str], parent_field: str = "parent_id", ) -> List[List[str]]: """`order` 순서를 보존하며 DAG를 level batch로 분할. Args: order: 노드 IDs (planner 결정 순서) items: id → node dict (parent_field key 보유) renderable_set: 실제로 렌더 대상인 IDs. 이 집합 밖 노드는 부모 부재로 간주. parent_field: 노드 dict의 부모 ID 필드명. 기본 "parent_id". Returns: level batches (input order 보존). cycle 검출 시 잔여 노드를 마지막 level로 flush. """ levels: List[List[str]] = [] completed: Set[str] = set() remaining = [nid for nid in order if nid in renderable_set] while remaining: ready: List[str] = [] not_ready: List[str] = [] for nid in remaining: node = items.get(nid, {}) pid = (node.get(parent_field) or "").strip() if not pid or pid in completed or pid not in renderable_set: ready.append(nid) else: not_ready.append(nid) if not ready: logger.warning( "compute_dag_levels: unresolved parent dependencies — flushing %d nodes as final level", len(not_ready), ) ready = not_ready not_ready = [] levels.append(ready) completed.update(ready) remaining = not_ready return levels ``` - [ ] **Step 1.4: 테스트 통과 확인** Run: `cd backend && pytest tests/pipeline/test_dag_levels.py -v` Expected: PASS — 7 tests - [ ] **Step 1.5: 기존 `_compute_chain_bg_levels` wrapper로 변경 (회귀 0)** Edit `backend/app/modules/pipeline/background_chain_render.py:793-838` — 함수 본문을 generic helper 호출로 교체: ```python # 위쪽 import 추가 from app.modules.pipeline._dag_levels import compute_dag_levels def _compute_chain_bg_levels( planner_chain_order: List[str], planning_groups: Dict[str, Dict[str, Any]], renderable_set: set, ) -> List[List[str]]: """planner_chain_order를 level-based parallelizable batches로 분할. Phase 5.3 wrapper — Phase 7에서 추출한 generic compute_dag_levels로 위임. 동일한 시그니처/동작 유지 (회귀 0). """ return compute_dag_levels( order=planner_chain_order, items=planning_groups, renderable_set=renderable_set, parent_field="parent_id", ) ``` - [ ] **Step 1.6: 기존 Phase 5.3 테스트 회귀 0 확인** Run: `cd backend && pytest tests/pipeline/test_background_chain_render.py -v -k "compute_chain_bg_levels"` Expected: PASS — 7 tests (Phase 5.3 invariants 모두 보존) - [ ] **Step 1.7: commit** ```bash git add backend/app/modules/pipeline/_dag_levels.py backend/tests/pipeline/test_dag_levels.py backend/app/modules/pipeline/background_chain_render.py git commit -m "feat(p7-t1): generic compute_dag_levels — Phase 5.3 helper 추출" ``` --- ## Task 2: Foundation — config + applicability 확장 **Why**: Phase 7 step들의 `applicability="if_background_mode"`. 새 validator 등록 + `BACKGROUND_MODE` literal 확장. **Files:** - Modify: `backend/app/core/config.py:49` (background_mode literal) - Modify: `backend/app/core/applicability.py:109-124` (`_if_background_mode` validator + registry) - Create: `backend/tests/core/test_applicability_phase7.py` - [ ] **Step 2.1: failing test 작성** ```python # backend/tests/core/test_applicability_phase7.py import pytest from unittest.mock import MagicMock, patch from app.core.applicability import _if_background_mode, APPLICABILITY_VALIDATORS def test_if_background_mode_off_returns_false(): runner = MagicMock() with patch("app.core.config.settings.background_mode", "off"): assert _if_background_mode(runner) is False def test_if_background_mode_on_returns_true(): runner = MagicMock() with patch("app.core.config.settings.background_mode", "on"): assert _if_background_mode(runner) is True def test_if_background_mode_legacy_alias_returns_true(): """floor_plan_anchored은 legacy alias로 'on' 동등.""" runner = MagicMock() with patch("app.core.config.settings.background_mode", "floor_plan_anchored"): assert _if_background_mode(runner) is True def test_validator_registered(): assert "if_background_mode" in APPLICABILITY_VALIDATORS ``` - [ ] **Step 2.2: 테스트 실행 — fail 확인** Run: `cd backend && pytest tests/core/test_applicability_phase7.py -v` Expected: FAIL with "ImportError: cannot import name '_if_background_mode'" - [ ] **Step 2.3: config literal 확장** Edit `backend/app/core/config.py:49`: ```python # before: background_mode: Literal["off", "chain_only", "floor_plan_anchored"] = "off" # after: background_mode: Literal["off", "on", "floor_plan_anchored", "chain_only"] = "off" # Phase 7: "on" — Phase 7 6 step 활성. "floor_plan_anchored" — legacy alias for "on". # "off" — 모든 background step disabled. "chain_only" — Phase 4 LEGACY only. ``` - [ ] **Step 2.4: applicability validator 추가** Edit `backend/app/core/applicability.py` — `_if_floor_plan_mode` 직후에 추가: ```python def _if_background_mode(runner: "StepRunner") -> bool: """Phase 7: settings.background_mode in {"on", "floor_plan_anchored"} 일 때만 실행. "floor_plan_anchored"은 Phase 5 legacy alias — Phase 7 활성으로 동작. """ from app.core.config import settings return settings.background_mode in {"on", "floor_plan_anchored"} ``` 레지스트리에도 등록: ```python APPLICABILITY_VALIDATORS: Dict[str, ApplicabilityValidator] = { "if_planning_doc": _if_planning_doc, "if_has_outlooks": _if_has_outlooks, "if_shot_essence_enabled": _if_shot_essence_enabled, "if_floor_plan_mode": _if_floor_plan_mode, "if_background_mode": _if_background_mode, } ``` - [ ] **Step 2.5: 테스트 통과 확인** Run: `cd backend && pytest tests/core/test_applicability_phase7.py tests/core/test_applicability.py -v` Expected: PASS — 4 신규 + 기존 모두 PASS - [ ] **Step 2.6: commit** ```bash git add backend/app/core/config.py backend/app/core/applicability.py backend/tests/core/test_applicability_phase7.py git commit -m "feat(p7-t2): if_background_mode applicability + 'on' literal" ``` --- ## Task 3: Step 1 prompt 작성 (background_classify) **Why**: Step 1은 building group 분류 + chain_bg vs prev_shot_ref 결정. system prompt + user_template + schema 정의. **Files:** - Create: `prompts/_base/background_classify/1.YYYYMMDDHHmm/system.md` (실제 timestamp 사용 — 예: `1.202604291800`) - Create: `prompts/_base/background_classify/1.YYYYMMDDHHmm/user_template.md` - Create: `prompts/_base/background_classify/1.YYYYMMDDHHmm/schema.json` > **Note for implementer**: timestamp 형식 `YYYYMMDDHHmm`은 작성 시점 기준 (예: `202604291800`). 메모리 `feedback_version_format.md` 참조. - [ ] **Step 3.1: prompt 디렉토리 생성** ```bash mkdir -p prompts/_base/background_classify/1.$(date +%Y%m%d%H%M) ``` (이후 단계에서 `${VERSION_DIR}`로 표기. 한 번 만든 후 환경변수로 export 권장.) - [ ] **Step 3.2: system.md 작성** ```markdown # Background Classify — System Prompt You are a film art-direction analyst. Your job: classify each building group of locations into one of two render strategies: - **chain_bg**: render dedicated background images (with floor plan reference) when the group has 3 or more shots AND has at least one indoor anchor. - **prev_shot_ref**: skip dedicated backgrounds; downstream uses previous shot images as background reference. Used for groups that are too small (<3 shots) OR outdoor-only. ## Rules 1. **Group inputs**: each input "building_group" is a set of related locations (e.g., a rooftop building includes its interior, exterior, hallway, etc). Members are pre-grouped by the caller — DO NOT regroup. 2. **Anchor selection**: pick the most-shot indoor location as `anchor_loc`. If no indoor location, use the most-shot outdoor. 3. **Universal nouns only**: all output IDs and labels must be ASCII snake_case (e.g., `bg_rooftop_unit`, NOT `옥탑방_단지`). Korean/Hanja/kana in any ID = error. 4. **No work-specific names**: do not use proper nouns from the work (character names, place names). Generic descriptors only. 5. **Rationale in Korean**: rationale fields may be Korean (descriptive prose), but ID/label fields must be ASCII. ## Decision Heuristics - 3+ shots + ≥1 indoor anchor → `chain_bg` - 3 shots and outdoor-only → `prev_shot_ref` (no dedicated bg, prev-shot reuse) - <3 shots → `prev_shot_ref` regardless ## Output Strict JSON matching the schema. No extra prose. ``` - [ ] **Step 3.3: user_template.md 작성** ```markdown # Building Groups to Classify ## visual_world_rules {visual_world_rules} ## building_groups {building_groups_block} ## Output Return classification for each input group. Preserve `group_id` exactly. Pick `kind` per the rules above. ``` - [ ] **Step 3.4: schema.json 작성** ```json { "type": "object", "properties": { "building_groups": { "type": "array", "items": { "type": "object", "properties": { "group_id": {"type": "string", "pattern": "^[a-z0-9][a-z0-9_]*$"}, "anchor_loc": {"type": "string"}, "kind": {"type": "string", "enum": ["chain_bg", "prev_shot_ref"]}, "rationale": {"type": "string"} }, "required": ["group_id", "anchor_loc", "kind", "rationale"], "additionalProperties": false } } }, "required": ["building_groups"], "additionalProperties": false } ``` - [ ] **Step 3.5: prompt loader 동작 확인** Run: `cd backend && python -c "from app.modules.prompt_loader import load_prompt, load_schema; print(load_prompt('background_classify','system')[:80]); print(load_schema('background_classify','schema'))"` Expected: system.md 처음 80자 + schema dict 출력 - [ ] **Step 3.6: commit** ```bash git add prompts/_base/background_classify/ git commit -m "feat(p7-t3): background_classify prompt v1 (system + user + schema)" ``` --- ## Task 4: Step 1 module — background_classify LLM 호출 **Files:** - Create: `backend/app/modules/pipeline/background_classify.py` - Create: `backend/tests/pipeline/test_background_classify.py` - [ ] **Step 4.1: failing test 작성** ```python # backend/tests/pipeline/test_background_classify.py from unittest.mock import MagicMock import pytest from app.modules.pipeline.background_classify import ( build_classify_user_prompt, validate_classify_output, run_background_classify, ClassifyError, ) def test_build_user_prompt_includes_groups(): groups = [ {"group_id": "bg_x", "members": [ {"loc_id": "L01", "label": "옥탑방", "shot_count": 12, "is_indoor": True}, {"loc_id": "L02", "label": "옥상", "shot_count": 4, "is_indoor": False}, ]}, ] prompt = build_classify_user_prompt(groups, "rules text") assert "bg_x" in prompt assert "L01" in prompt and "L02" in prompt assert "rules text" in prompt # truncation 검사 — 모든 멤버 포함 assert "옥탑방" in prompt and "옥상" in prompt def test_validate_rejects_korean_id(): out = {"building_groups": [ {"group_id": "옥탑방", "anchor_loc": "L01", "kind": "chain_bg", "rationale": "ok"}, ]} with pytest.raises(ValueError, match="non-ASCII"): validate_classify_output(out, all_loc_ids={"L01"}, expected_group_ids={"옥탑방"}) def test_validate_rejects_unknown_anchor(): out = {"building_groups": [ {"group_id": "bg_x", "anchor_loc": "L99", "kind": "chain_bg", "rationale": "ok"}, ]} with pytest.raises(ValueError, match="anchor_loc"): validate_classify_output(out, all_loc_ids={"L01"}, expected_group_ids={"bg_x"}) def test_validate_rejects_unexpected_group_id(): out = {"building_groups": [ {"group_id": "bg_extra", "anchor_loc": "L01", "kind": "chain_bg", "rationale": "ok"}, ]} with pytest.raises(ValueError, match="unexpected group"): validate_classify_output(out, all_loc_ids={"L01"}, expected_group_ids={"bg_x"}) def test_validate_passes_valid_output(): out = {"building_groups": [ {"group_id": "bg_x", "anchor_loc": "L01", "kind": "chain_bg", "rationale": "indoor anchor"}, ]} validate_classify_output(out, all_loc_ids={"L01"}, expected_group_ids={"bg_x"}) # no raise def test_run_retries_on_invariant_failure(): bad = {"building_groups": [{"group_id": "bg_x", "anchor_loc": "L99", "kind": "chain_bg", "rationale": "x"}]} good = {"building_groups": [{"group_id": "bg_x", "anchor_loc": "L01", "kind": "chain_bg", "rationale": "x"}]} fn = MagicMock(side_effect=[bad, good]) result = run_background_classify( user_prompt="x", all_loc_ids=["L01"], expected_group_ids=["bg_x"], call_structured_fn=fn, sleep_fn=lambda _: None, ) assert result == good assert fn.call_count == 2 def test_run_raises_on_exhaustion(): fn = MagicMock(return_value={"building_groups": [{"group_id": "bg_x", "anchor_loc": "L99", "kind": "chain_bg", "rationale": "x"}]}) with pytest.raises(ClassifyError): run_background_classify( user_prompt="x", all_loc_ids=["L01"], expected_group_ids=["bg_x"], call_structured_fn=fn, max_retries=2, sleep_fn=lambda _: None, ) assert fn.call_count == 2 ``` - [ ] **Step 4.2: 테스트 fail 확인** Run: `cd backend && pytest tests/pipeline/test_background_classify.py -v` Expected: FAIL — module not found - [ ] **Step 4.3: 모듈 구현** ```python # backend/app/modules/pipeline/background_classify.py """background_classify — Phase 7 Step 1. building groups → {chain_bg | prev_shot_ref} 분류 (LLM 1회). """ from __future__ import annotations import logging import re import time from typing import Any, Callable, Dict, List, Optional from app.modules.prompt_loader import load_prompt, load_schema logger = logging.getLogger(__name__) _NON_ASCII_TEXT_RE = re.compile( r"[ㄱ-ㆎ가-힣" r"一-鿿㐀-䶿豈-﫿" r"぀-ヿ]" ) _SAFE_ID_RE = re.compile(r"^[a-z0-9][a-z0-9_]*$") _MODULE = "background_classify" class ClassifyError(Exception): """retry 한도까지 실패.""" def build_classify_user_prompt( building_groups: List[Dict[str, Any]], visual_world_rules: str, ) -> str: """LLM에 보낼 user prompt — 모든 멤버 정보 truncation 없이 포함.""" template = load_prompt(_MODULE, "user_template") blocks: List[str] = [] for g in building_groups: gid = g.get("group_id", "") members = g.get("members", []) or [] lines = [f"- group_id: {gid}", " members:"] for m in members: loc_id = m.get("loc_id", "") label = m.get("label", "") shot_count = m.get("shot_count", 0) is_indoor = m.get("is_indoor", False) summary = (m.get("summary", "") or "").strip() lines.append( f" - {loc_id} ({'indoor' if is_indoor else 'outdoor'}, {shot_count} shot): {label}" ) if summary: lines.append(f" summary: {summary}") blocks.append("\n".join(lines)) groups_block = "\n\n".join(blocks) or "(none)" return template.format( visual_world_rules=visual_world_rules or "(none)", building_groups_block=groups_block, ) def validate_classify_output( output: Dict[str, Any], all_loc_ids: set, expected_group_ids: set, ) -> None: """semantic invariants: - group_id ∈ expected_group_ids (LLM이 그룹 추가/누락 X) - anchor_loc ∈ all_loc_ids - 모든 ID/label은 ASCII (한글 0건) - kind ∈ {"chain_bg", "prev_shot_ref"} (schema가 선잡지만 방어적) """ groups = output.get("building_groups") or [] seen_ids: set = set() for g in groups: gid = g.get("group_id", "") if not _SAFE_ID_RE.match(gid): raise ValueError( f"group_id {gid!r} contains non-ASCII or unsafe chars" ) if _NON_ASCII_TEXT_RE.search(gid): raise ValueError(f"group_id {gid!r} contains non-ASCII text") if gid not in expected_group_ids: raise ValueError( f"unexpected group_id {gid!r}; expected ⊆ {expected_group_ids}" ) seen_ids.add(gid) anchor = g.get("anchor_loc", "") if anchor and anchor not in all_loc_ids: raise ValueError( f"anchor_loc {anchor!r} for group {gid!r} not in all_loc_ids" ) if _NON_ASCII_TEXT_RE.search(anchor): raise ValueError(f"anchor_loc {anchor!r} contains non-ASCII") kind = g.get("kind", "") if kind not in {"chain_bg", "prev_shot_ref"}: raise ValueError(f"unknown kind {kind!r} for group {gid!r}") missing = expected_group_ids - seen_ids if missing: raise ValueError(f"missing group_ids: {sorted(missing)}") def run_background_classify( *, user_prompt: str, all_loc_ids: List[str], expected_group_ids: List[str], call_structured_fn: Callable[..., Dict[str, Any]], project_config: Optional[Dict[str, Any]] = None, opik_metadata: Optional[Dict[str, Any]] = None, max_retries: int = 3, backoff_base_sec: float = 2.0, sleep_fn: Callable[[float], None] = time.sleep, ) -> Dict[str, Any]: system = load_prompt(_MODULE, "system") schema = load_schema(_MODULE, "schema") loc_set = set(all_loc_ids) group_set = set(expected_group_ids) last_err: Optional[Exception] = None for attempt in range(max_retries): try: result = call_structured_fn( step="background_classify", system_prompt=system, user_prompt=user_prompt, response_schema=schema, project_config=project_config, schema_name="background_classify", opik_metadata=opik_metadata, ) validate_classify_output(result, loc_set, group_set) return result except (ValueError, RuntimeError, TimeoutError, ConnectionError) as exc: last_err = exc logger.warning( "background_classify attempt %d/%d failed: %s", attempt + 1, max_retries, exc, ) if attempt + 1 < max_retries: sleep_fn(backoff_base_sec * (attempt + 1)) raise ClassifyError( f"background_classify exhausted {max_retries} retries: {last_err}" ) ``` - [ ] **Step 4.4: 테스트 통과 확인** Run: `cd backend && pytest tests/pipeline/test_background_classify.py -v` Expected: PASS — 7 tests - [ ] **Step 4.5: commit** ```bash git add backend/app/modules/pipeline/background_classify.py backend/tests/pipeline/test_background_classify.py git commit -m "feat(p7-t4): background_classify LLM 호출 + invariants" ``` --- ## Task 5: Step 1 step entry — BackgroundClassifyStep **Files:** - Create: `backend/app/core/steps/background_classify_step.py` - Modify: `backend/app/core/steps/__init__.py:34-99` (STEP_CLASSES 등록) - Modify: `backend/app/modules/llm/llm_client.py:205-254` (PIPELINE_STEPS 등록) - Modify: `backend/app/core/step_manifest.py:648-695` (manifest entry) - Create: `backend/tests/core/test_background_classify_step.py` - [ ] **Step 5.1: failing test 작성** ```python # backend/tests/core/test_background_classify_step.py import json import pytest from pathlib import Path from unittest.mock import patch, MagicMock from app.core.steps.background_classify_step import ( BackgroundClassifyStep, SCHEMA_VERSION, PROMPT_VERSION, _build_groups_from_entities, ) def test_build_groups_from_entities_groups_by_building_hint(tmp_path): """Phase 5 building_group hint(entity_detail) 가 있으면 그것을 따른다.""" entity_merge = {"data": {"locations": [ {"short_id": "L01", "name": "옥탑방"}, {"short_id": "L02", "name": "옥상"}, {"short_id": "L03", "name": "복도"}, ]}} entity_detail = {"data": {"locations": [ {"short_id": "L01", "kind": "indoor", "building_group": "rooftop_unit"}, {"short_id": "L02", "kind": "outdoor", "building_group": "rooftop_unit"}, {"short_id": "L03", "kind": "indoor", "building_group": "other_building"}, ]}} shot_counts = {"L01": 12, "L02": 4, "L03": 6} groups = _build_groups_from_entities(entity_merge, entity_detail, shot_counts) by_id = {g["group_id"]: g for g in groups} assert "bg_rooftop_unit" in by_id assert "bg_other_building" in by_id assert {m["loc_id"] for m in by_id["bg_rooftop_unit"]["members"]} == {"L01", "L02"} assert {m["loc_id"] for m in by_id["bg_other_building"]["members"]} == {"L03"} def test_build_groups_falls_back_to_singleton_when_no_hint(): entity_merge = {"data": {"locations": [{"short_id": "L01", "name": "x"}]}} entity_detail = {"data": {"locations": []}} shot_counts = {"L01": 5} groups = _build_groups_from_entities(entity_merge, entity_detail, shot_counts) assert len(groups) == 1 assert groups[0]["group_id"] == "bg_l01" def test_step_disabled_when_mode_off(tmp_path): with patch("app.core.config.settings.background_mode", "off"): step = BackgroundClassifyStep.__new__(BackgroundClassifyStep) step.project_id = "p"; step.episode_id = "e"; step.project_config = {} step.build_opik_metadata = MagicMock(return_value={}) result = step._execute() assert result["applicable_count"] == 0 assert result["data"] == {} ``` - [ ] **Step 5.2: 테스트 fail 확인** Run: `cd backend && pytest tests/core/test_background_classify_step.py -v` Expected: FAIL — module not found - [ ] **Step 5.3: step 구현** ```python # backend/app/core/steps/background_classify_step.py """BackgroundClassifyStep — Phase 7 Step 1. building group 분류 + chain_bg/prev_shot_ref 결정. 단일 LLM 호출. 흐름: 1. shot_validator/shot_selection/entity_merge/entity_detail/scene_director/visual_world_rules 로드 2. shot_count(loc_id별 출현 횟수) 집계 3. _build_groups_from_entities — entity_detail의 building_group hint로 그룹화 (hint 없으면 location 1개당 singleton group) 4. build_classify_user_prompt → run_background_classify (3회 retry) 체크포인트 data: data.building_groups[] = [{group_id, members[], anchor_loc, kind, rationale}] """ from __future__ import annotations import json import logging from collections import defaultdict from pathlib import Path from typing import Any, Dict, List, Optional from app.core.step_runner import StepRunner logger = logging.getLogger(__name__) SCHEMA_VERSION = 1 PROMPT_VERSION = "1" class BackgroundClassifyStep(StepRunner): def _config_hash(self) -> str: import hashlib, json as _json from app.core.config import settings payload = { "background_mode": settings.background_mode, "model": "gpt-5.5", "schema_version": SCHEMA_VERSION, "prompt_version": PROMPT_VERSION, } return hashlib.sha256( _json.dumps(payload, sort_keys=True).encode("utf-8") ).hexdigest()[:16] def _load_prev_checkpoint(self, step_id: str) -> Optional[Dict[str, Any]]: from app.core.config import settings cp = ( Path(settings.projects_dir) / self.project_id / "checkpoints" / "episodes" / self.episode_id / step_id / "manifest.json" ) if cp.exists(): try: return json.loads(cp.read_text(encoding="utf-8")) except Exception as exc: logger.warning("background_classify: %s parse failed: %s", step_id, exc) return None def _execute(self, mode: str = "resume") -> Dict[str, Any]: from app.core.config import settings if settings.background_mode not in {"on", "floor_plan_anchored"}: logger.info( "background_classify: skipped — background_mode=%s", settings.background_mode, ) return { "applicable_count": 0, "completed_count": 0, "failed_count": 0, "schema_version": SCHEMA_VERSION, "config_hash": self._config_hash(), "data": {}, } shot_validator_cp = self._load_prev_checkpoint("shot_validator") shot_selection_cp = self._load_prev_checkpoint("shot_selection") entity_merge_cp = self._load_prev_checkpoint("entity_merge") entity_detail_cp = self._load_prev_checkpoint("entity_detail") rules_cp = self._load_prev_checkpoint("visual_world_rules") # location별 shot count (selected shots만) shot_counts = _count_selected_shots_by_loc(shot_validator_cp, shot_selection_cp) groups = _build_groups_from_entities(entity_merge_cp, entity_detail_cp, shot_counts) if not groups: logger.info("background_classify: no groups — empty plan") return { "applicable_count": 1, "completed_count": 1, "failed_count": 0, "schema_version": SCHEMA_VERSION, "config_hash": self._config_hash(), "data": {"building_groups": []}, } rules_text = "" if rules_cp: data = rules_cp.get("data", {}) or {} rules_text = data.get("rules_text", "") or data.get("text", "") or "" from app.modules.pipeline.background_classify import ( build_classify_user_prompt, run_background_classify, ClassifyError, ) from app.modules.llm.llm_client import call_structured all_loc_ids = sorted({m["loc_id"] for g in groups for m in g["members"]}) expected_group_ids = [g["group_id"] for g in groups] user_prompt = build_classify_user_prompt(groups, rules_text) try: result = run_background_classify( user_prompt=user_prompt, all_loc_ids=all_loc_ids, expected_group_ids=expected_group_ids, call_structured_fn=call_structured, project_config=self.project_config, opik_metadata=self.build_opik_metadata(), ) except ClassifyError as exc: logger.error("background_classify: exhausted: %s", exc) return { "applicable_count": 1, "completed_count": 0, "failed_count": 1, "schema_version": SCHEMA_VERSION, "config_hash": self._config_hash(), "data": {"error": str(exc), "building_groups": []}, } # LLM output에 members[] 정보를 다시 합쳐 (LLM은 group_id/anchor_loc/kind만 결정) groups_by_id = {g["group_id"]: g for g in groups} out_groups: List[Dict[str, Any]] = [] for cls in result.get("building_groups", []): gid = cls["group_id"] base = groups_by_id.get(gid, {}) out_groups.append({ "group_id": gid, "members": base.get("members", []), "anchor_loc": cls["anchor_loc"], "kind": cls["kind"], "rationale": cls["rationale"], }) return { "applicable_count": 1, "completed_count": 1, "failed_count": 0, "schema_version": SCHEMA_VERSION, "config_hash": self._config_hash(), "data": {"building_groups": out_groups}, } def _count_selected_shots_by_loc( shot_validator_cp: Optional[Dict[str, Any]], shot_selection_cp: Optional[Dict[str, Any]], ) -> Dict[str, int]: if not shot_validator_cp or not shot_selection_cp: return {} sel_map: Dict[int, set] = {} for s in (shot_selection_cp.get("data", {}) or {}).get("scenes", []) or []: si = s.get("scene_index") if si is None: continue sel_map[int(si)] = set(s.get("selected_shot_indices", []) or []) counts: Dict[str, int] = defaultdict(int) for s in (shot_validator_cp.get("data", {}) or {}).get("scenes", []) or []: si = s.get("scene_index") if si is None: continue sel = sel_map.get(int(si), set()) for sh in s.get("shots", []) or []: shi = sh.get("shot_index") if shi is None or shi not in sel: continue loc_id = sh.get("location_id") or "" if loc_id: counts[loc_id] += 1 return dict(counts) def _build_groups_from_entities( entity_merge_cp: Optional[Dict[str, Any]], entity_detail_cp: Optional[Dict[str, Any]], shot_counts: Dict[str, int], ) -> List[Dict[str, Any]]: """entity_merge.locations + entity_detail.building_group hint로 그룹 빌드. hint 없는 location은 singleton group("bg_l01" 등)으로 둠. """ if not entity_merge_cp: return [] locations = (entity_merge_cp.get("data", {}) or {}).get("locations", []) or [] detail_by_short: Dict[str, Dict[str, str]] = {} if entity_detail_cp: ed = (entity_detail_cp.get("data", {}) or {}) for key in ("locations", "entities", "items", "all"): for item in (ed.get(key) or []): if not isinstance(item, dict): continue sid = item.get("short_id") or "" if not sid.startswith("L"): continue detail_by_short[sid] = { "kind": (item.get("kind") or item.get("location_kind") or "") or "", "building_group": (item.get("building_group") or "") or "", "summary": (item.get("description", "") or item.get("summary", "")) or "", } groups_map: Dict[str, List[Dict[str, Any]]] = defaultdict(list) for loc in locations: sid = loc.get("short_id") or "" if not sid: continue d = detail_by_short.get(sid, {}) bg = d.get("building_group", "").strip() kind = d.get("kind", "").strip() is_indoor = kind == "indoor" or "indoor" in kind.lower() member = { "loc_id": sid, "label": loc.get("name") or sid, "shot_count": shot_counts.get(sid, 0), "is_indoor": is_indoor, "summary": d.get("summary", ""), } group_id = f"bg_{bg}" if bg else f"bg_{sid.lower()}" groups_map[group_id].append(member) return [ {"group_id": gid, "members": members} for gid, members in sorted(groups_map.items()) ] ``` - [ ] **Step 5.4: STEP_CLASSES + PIPELINE_STEPS + manifest 등록** Edit `backend/app/core/steps/__init__.py` — line 34 부근에 import + STEP_CLASSES 등록: ```python from app.core.steps.background_classify_step import BackgroundClassifyStep # ... STEP_CLASSES dict에 추가 "background_classify": BackgroundClassifyStep, ``` Edit `backend/app/modules/llm/llm_client.py:250` (background_planner 직후): ```python "background_classify": {"label": "배경 그룹 분류", "default": "gpt", "category": "analysis"}, ``` Edit `backend/app/core/step_manifest.py` — `background_planner` entry(line 648) 직전에 추가: ```python # 19.50 background_classify — Phase 7 Step 1. building group 분류 + chain_bg vs prev_shot_ref. "background_classify": { "label": "배경 그룹 분류", "category": "analysis", "order": 19.50, "default_model": "gpt", "provider": "openai", "depends_on": [ "shot_validator", "shot_selection", "entity_merge", "entity_detail", "visual_world_rules", "scene_director", ], "fan_out": False, "applicability": "if_background_mode", "step_type": "transform", "lifecycle": "active", }, ``` - [ ] **Step 5.5: 테스트 통과 확인** Run: `cd backend && pytest tests/core/test_background_classify_step.py -v` Expected: PASS — 3 tests - [ ] **Step 5.6: 회귀 검사** Run: `cd backend && pytest tests/core/ tests/pipeline/ -x --timeout=60 -q` Expected: 회귀 0 (Phase 5/5.3/6 tests 그대로 통과) - [ ] **Step 5.7: commit** ```bash git add backend/app/core/steps/background_classify_step.py backend/app/core/steps/__init__.py backend/app/modules/llm/llm_client.py backend/app/core/step_manifest.py backend/tests/core/test_background_classify_step.py git commit -m "feat(p7-t5): BackgroundClassifyStep + manifest 등록 (order 19.50)" ``` --- ## Task 6: Step 2 prompt 작성 (background_master_plan) **Why**: Phase 7 핵심 — 그룹별 도면/배경/순서/ref/샷매핑을 한 호출로 결정. 가장 복잡한 schema. **Files:** - Create: `prompts/_base/background_master_plan/1.YYYYMMDDHHmm/{system.md,user_template.md,schema.json}` - [ ] **Step 6.1: 디렉토리 + system.md 작성** ```bash mkdir -p prompts/_base/background_master_plan/1.$(date +%Y%m%d%H%M) ``` system.md (요약): ```markdown # Background Master Plan — System Prompt You are a film art-direction master planner for a single building group. Given (a) the group spec, (b) related scene segments (verbatim), (c) related shots, decide: 1. **floor_plans[]**: how many floor plans to draw (typically 1; multiple if the group has clearly separated sub-rooms — e.g., living room + bedroom + corridor) 2. **backgrounds[]**: how many background images to draw (one per (sub_location, state_label) combo — e.g., living_day_normal, living_dusk_ransacked, bedroom_night_blood) 3. **gen_order**: topological order — floor_plans first, then backgrounds. Within backgrounds, place chains (depends_on_bg) sequentially. 4. **applies_to_shots**: each background must list which shots use it. Shot IDs MUST come from the input shots list. ## Hard Invariants 1. ASCII snake_case for ALL ids/labels (floor_plans[].fp_id, backgrounds[].bg_id, sub_location, state_label). Korean/Hanja/kana = error. 2. Every background MUST have ≥1 entry in `depends_on_fp` referencing an existing fp_id from this same plan. 3. `applies_to_shots` ⊆ input shot_ids list. Cross-group shots forbidden. 4. Same `sub_location` value → all backgrounds with that sub_location share the same floor plan via `depends_on_fp`. Consistency rule. 5. Backgrounds in the same `sub_location` must form a chain (each later one references an earlier one via `depends_on_bg`) so style is consistent. 6. `gen_order` is a topological ordering: each entry's depends_on_fp + depends_on_bg references must appear earlier in gen_order. 7. NO proper nouns from the work. Use generic English descriptors: "living_room", "bedroom_corner", "rooftop_outside" — NOT "ok-tab-bang_living_room". 8. Plot-critical visual elements explicit in `state_label` or `scope` (e.g., "dusk_ransacked", "night_blood_curtain_drawn") — these will steer downstream prompts. ## Sub-room consistency If multiple shots happen in clearly different rooms within one building group, name distinct sub_locations. If shots are all in the same room with different states (day/night/blood/etc), keep one sub_location and vary state_label. The same sub_location MUST always reference the same floor plan. Different sub_locations may share or split floor plans depending on physical complexity. ## Output Strict JSON per the schema. No prose outside JSON. ``` - [ ] **Step 6.2: user_template.md 작성** ```markdown # Group: {group_id} ## members {members_block} ## visual_world_rules {visual_world_rules} ## scenes (verbatim, do not summarize) {scene_segments_block} ## shots in this group {shots_block} ## Output Plan all floor_plans, backgrounds, gen_order, and per-background shot mappings per the rules. ``` - [ ] **Step 6.3: schema.json 작성** ```json { "type": "object", "properties": { "group_id": {"type": "string", "pattern": "^[a-z0-9][a-z0-9_]*$"}, "rationale_summary": {"type": "string"}, "floor_plans": { "type": "array", "items": { "type": "object", "properties": { "fp_id": {"type": "string", "pattern": "^[a-z0-9][a-z0-9_]*$"}, "sub_location": {"type": "string", "pattern": "^[a-z0-9][a-z0-9_]*$"}, "scope": {"type": "string"}, "depends_on_fp": { "type": "array", "items": {"type": "string"} } }, "required": ["fp_id", "sub_location", "scope", "depends_on_fp"], "additionalProperties": false } }, "backgrounds": { "type": "array", "items": { "type": "object", "properties": { "bg_id": {"type": "string", "pattern": "^[a-z0-9][a-z0-9_]*$"}, "loc_id": {"type": "string"}, "sub_location": {"type": "string", "pattern": "^[a-z0-9][a-z0-9_]*$"}, "state_label": {"type": "string", "pattern": "^[a-z0-9][a-z0-9_]*$"}, "depends_on_fp": { "type": "array", "items": {"type": "string"}, "minItems": 1 }, "depends_on_bg": { "type": "array", "items": {"type": "string"} }, "applies_to_shots": { "type": "array", "items": {"type": "string"} } }, "required": ["bg_id", "loc_id", "sub_location", "state_label", "depends_on_fp", "depends_on_bg", "applies_to_shots"], "additionalProperties": false } }, "gen_order": { "type": "array", "items": {"type": "string"} } }, "required": ["group_id", "rationale_summary", "floor_plans", "backgrounds", "gen_order"], "additionalProperties": false } ``` - [ ] **Step 6.4: prompt loader smoke** Run: `cd backend && python -c "from app.modules.prompt_loader import load_prompt, load_schema; load_prompt('background_master_plan','system'); load_prompt('background_master_plan','user_template'); load_schema('background_master_plan','schema'); print('ok')"` Expected: `ok` - [ ] **Step 6.5: commit** ```bash git add prompts/_base/background_master_plan/ git commit -m "feat(p7-t6): background_master_plan prompt v1 (system + user + schema)" ``` --- ## Task 7: Step 2 module — master plan LLM 호출 + invariants **Why**: 가장 복잡한 invariant 검증. 8개 hard invariants를 코드로 강제. **Files:** - Create: `backend/app/modules/pipeline/background_master_plan.py` - Create: `backend/tests/pipeline/test_background_master_plan.py` - [ ] **Step 7.1: failing test 작성 — invariants 8개 + retry** ```python # backend/tests/pipeline/test_background_master_plan.py import pytest from unittest.mock import MagicMock from app.modules.pipeline.background_master_plan import ( build_master_plan_user_prompt, validate_master_plan_output, run_background_master_plan, MasterPlanError, ) def _good_plan(): return { "group_id": "bg_x", "rationale_summary": "...", "floor_plans": [ {"fp_id": "fp_living", "sub_location": "living_room", "scope": "...", "depends_on_fp": []}, ], "backgrounds": [ {"bg_id": "cb_living_day", "loc_id": "L01", "sub_location": "living_room", "state_label": "day_normal", "depends_on_fp": ["fp_living"], "depends_on_bg": [], "applies_to_shots": ["S01_Shot1"]}, ], "gen_order": ["fp_living", "cb_living_day"], } def test_validate_passes_minimal(): validate_master_plan_output(_good_plan(), expected_group_id="bg_x", group_loc_ids={"L01"}, group_shot_ids={"S01_Shot1"}) def test_invariant_1_korean_id_rejected(): plan = _good_plan() plan["floor_plans"][0]["fp_id"] = "fp_거실" with pytest.raises(ValueError, match="non-ASCII"): validate_master_plan_output(plan, "bg_x", {"L01"}, {"S01_Shot1"}) def test_invariant_2_bg_without_fp_ref_rejected(): plan = _good_plan() plan["backgrounds"][0]["depends_on_fp"] = [] with pytest.raises(ValueError, match="depends_on_fp"): validate_master_plan_output(plan, "bg_x", {"L01"}, {"S01_Shot1"}) def test_invariant_3_cross_group_shot_rejected(): plan = _good_plan() plan["backgrounds"][0]["applies_to_shots"] = ["S99_Shot1"] with pytest.raises(ValueError, match="applies_to_shots"): validate_master_plan_output(plan, "bg_x", {"L01"}, {"S01_Shot1"}) def test_invariant_4_same_sublocation_must_share_fp(): plan = _good_plan() plan["floor_plans"].append( {"fp_id": "fp_living2", "sub_location": "living_room", "scope": "...", "depends_on_fp": []} ) plan["backgrounds"].append({ "bg_id": "cb_living_dusk", "loc_id": "L01", "sub_location": "living_room", "state_label": "dusk", "depends_on_fp": ["fp_living2"], "depends_on_bg": [], "applies_to_shots": [] }) plan["gen_order"] = ["fp_living", "fp_living2", "cb_living_day", "cb_living_dusk"] with pytest.raises(ValueError, match="sub_location"): validate_master_plan_output(plan, "bg_x", {"L01"}, {"S01_Shot1"}) def test_invariant_5_chain_within_sublocation(): plan = _good_plan() plan["backgrounds"].append({ "bg_id": "cb_living_dusk", "loc_id": "L01", "sub_location": "living_room", "state_label": "dusk", "depends_on_fp": ["fp_living"], "depends_on_bg": [], # 같은 sub_location 두 번째인데 chain 미지정 → 위반 "applies_to_shots": [] }) plan["gen_order"] = ["fp_living", "cb_living_day", "cb_living_dusk"] with pytest.raises(ValueError, match="chain"): validate_master_plan_output(plan, "bg_x", {"L01"}, {"S01_Shot1"}) def test_invariant_6_topological_violation_rejected(): plan = _good_plan() plan["gen_order"] = ["cb_living_day", "fp_living"] # fp가 bg 뒤에 옴 — 위반 with pytest.raises(ValueError, match="topolog"): validate_master_plan_output(plan, "bg_x", {"L01"}, {"S01_Shot1"}) def test_invariant_8_loc_id_not_in_group_rejected(): plan = _good_plan() plan["backgrounds"][0]["loc_id"] = "L99" with pytest.raises(ValueError, match="loc_id"): validate_master_plan_output(plan, "bg_x", {"L01"}, {"S01_Shot1"}) def test_run_retries_on_invariant_failure(): bad = _good_plan() bad["backgrounds"][0]["loc_id"] = "L99" good = _good_plan() fn = MagicMock(side_effect=[bad, good]) result = run_background_master_plan( user_prompt="x", expected_group_id="bg_x", group_loc_ids=["L01"], group_shot_ids=["S01_Shot1"], call_structured_fn=fn, sleep_fn=lambda _: None, ) assert result["group_id"] == "bg_x" assert fn.call_count == 2 def test_run_raises_after_exhaustion(): bad = _good_plan() bad["backgrounds"][0]["loc_id"] = "L99" fn = MagicMock(return_value=bad) with pytest.raises(MasterPlanError): run_background_master_plan( user_prompt="x", expected_group_id="bg_x", group_loc_ids=["L01"], group_shot_ids=["S01_Shot1"], call_structured_fn=fn, max_retries=2, sleep_fn=lambda _: None, ) def test_build_user_prompt_includes_full_scene_text(): """truncation 금지 — 전체 scene 본문 포함.""" huge_text = "A" * 50000 prompt = build_master_plan_user_prompt( group={"group_id": "bg_x", "members": [{"loc_id": "L01", "label": "x"}]}, scenes=[{"scene_index": 1, "heading": "h", "text": huge_text, "shots": []}], visual_world_rules="rules", ) assert huge_text in prompt ``` - [ ] **Step 7.2: 테스트 fail 확인** Run: `cd backend && pytest tests/pipeline/test_background_master_plan.py -v` Expected: FAIL — module not found - [ ] **Step 7.3: 모듈 구현** ```python # backend/app/modules/pipeline/background_master_plan.py """background_master_plan — Phase 7 Step 2. 그룹당 LLM 호출 1회로 floor_plans + backgrounds + gen_order + applies_to_shots 결정. """ from __future__ import annotations import logging import re import time from collections import defaultdict from typing import Any, Callable, Dict, List, Optional from app.modules.prompt_loader import load_prompt, load_schema logger = logging.getLogger(__name__) _NON_ASCII_TEXT_RE = re.compile( r"[ㄱ-ㆎ가-힣" r"一-鿿㐀-䶿豈-﫿" r"぀-ヿ]" ) _SAFE_ID_RE = re.compile(r"^[a-z0-9][a-z0-9_]*$") _MODULE = "background_master_plan" class MasterPlanError(Exception): """retry 한도까지 실패.""" def build_master_plan_user_prompt( group: Dict[str, Any], scenes: List[Dict[str, Any]], visual_world_rules: str, ) -> str: """그룹의 master plan LLM에 보낼 user prompt — scene text 무절단 inject.""" template = load_prompt(_MODULE, "user_template") members_lines: List[str] = [] for m in group.get("members", []) or []: members_lines.append( f"- {m.get('loc_id', '')}: {m.get('label', '')} " f"({'indoor' if m.get('is_indoor') else 'outdoor'}, {m.get('shot_count', 0)} shot)" ) members_block = "\n".join(members_lines) or "(none)" scene_lines: List[str] = [] shot_lines: List[str] = [] for sc in scenes: si = sc.get("scene_index") if si is None: continue heading = (sc.get("heading") or "").strip() text = (sc.get("text") or "").strip() scene_lines.append(f"### Scene {si} — {heading}" if heading else f"### Scene {si}") scene_lines.append(text) scene_lines.append("") for sh in sc.get("shots", []) or []: shi = sh.get("shot_index", 0) shot_lines.append( f"- S{si}_Shot{shi} (loc={sh.get('location_id', '')}): {sh.get('description', '')}" ) scene_block = "\n".join(scene_lines).rstrip() or "(none)" shots_block = "\n".join(shot_lines) or "(none)" return template.format( group_id=group.get("group_id", ""), members_block=members_block, visual_world_rules=visual_world_rules or "(none)", scene_segments_block=scene_block, shots_block=shots_block, ) def validate_master_plan_output( plan: Dict[str, Any], expected_group_id: str, group_loc_ids: set, group_shot_ids: set, ) -> None: """8 hard invariants: 1. group_id matches + ASCII; all fp_id/bg_id/sub_location/state_label ASCII 2. 모든 background.depends_on_fp는 비어있지 않고 자체 plan의 fp_id 참조 3. applies_to_shots ⊆ group_shot_ids 4. 같은 sub_location의 모든 background는 같은 fp_id를 ref (depends_on_fp 첫 entry 비교) 5. 같은 sub_location의 background N>1개일 때 후속 background는 depends_on_bg에 이전 같은 sub_location bg를 reference 6. gen_order는 위상정렬: 모든 dependency가 앞에 등장 7. fp_id/bg_id/gen_order 모두 unique + set 일치 8. background.loc_id ⊆ group_loc_ids """ if plan.get("group_id") != expected_group_id: raise ValueError( f"group_id mismatch: got {plan.get('group_id')!r}, expected {expected_group_id!r}" ) if _NON_ASCII_TEXT_RE.search(plan.get("group_id", "")): raise ValueError(f"group_id {plan.get('group_id')!r} contains non-ASCII") floor_plans = plan.get("floor_plans") or [] backgrounds = plan.get("backgrounds") or [] gen_order = plan.get("gen_order") or [] fp_ids = [fp.get("fp_id", "") for fp in floor_plans] bg_ids = [bg.get("bg_id", "") for bg in backgrounds] fp_id_set = set(fp_ids) bg_id_set = set(bg_ids) # invariant 1: ASCII for label, fp in [("floor_plans", fp) for fp in floor_plans]: for field in ("fp_id", "sub_location"): v = fp.get(field, "") if not _SAFE_ID_RE.match(v): raise ValueError(f"floor_plan.{field}={v!r} not ASCII snake_case") if _NON_ASCII_TEXT_RE.search(v): raise ValueError(f"floor_plan.{field}={v!r} contains non-ASCII") for bg in backgrounds: for field in ("bg_id", "sub_location", "state_label"): v = bg.get(field, "") if not _SAFE_ID_RE.match(v): raise ValueError(f"background.{field}={v!r} not ASCII snake_case") if _NON_ASCII_TEXT_RE.search(v): raise ValueError(f"background.{field}={v!r} contains non-ASCII") # invariant 7: unique + set 일치 if len(fp_ids) != len(fp_id_set): raise ValueError(f"floor_plans.fp_id has duplicates: {fp_ids}") if len(bg_ids) != len(bg_id_set): raise ValueError(f"backgrounds.bg_id has duplicates: {bg_ids}") expected_order_set = fp_id_set | bg_id_set if set(gen_order) != expected_order_set: raise ValueError( f"gen_order set mismatch: {set(gen_order)} vs {expected_order_set}" ) if len(gen_order) != len(set(gen_order)): raise ValueError(f"gen_order has duplicates: {gen_order}") # invariant 2 & 8: bg.depends_on_fp 비어있지 않고 fp_id 참조 + bg.loc_id 그룹 멤버 for bg in backgrounds: deps = bg.get("depends_on_fp") or [] if not deps: raise ValueError(f"background {bg.get('bg_id')!r} depends_on_fp is empty") for fp_ref in deps: if fp_ref not in fp_id_set: raise ValueError( f"background {bg.get('bg_id')!r} depends_on_fp {fp_ref!r} not in plan" ) if bg.get("loc_id", "") not in group_loc_ids: raise ValueError( f"background {bg.get('bg_id')!r} loc_id {bg.get('loc_id')!r} not in group" ) # invariant 3: applies_to_shots ⊆ group_shot_ids for sid in bg.get("applies_to_shots") or []: if sid not in group_shot_ids: raise ValueError( f"background {bg.get('bg_id')!r} applies_to_shots {sid!r} not in group" ) # depends_on_bg는 자체 bg_id 참조여야 for bg_ref in bg.get("depends_on_bg") or []: if bg_ref not in bg_id_set: raise ValueError( f"background {bg.get('bg_id')!r} depends_on_bg {bg_ref!r} not in plan" ) # invariant 4: 같은 sub_location → 같은 fp_id (첫 entry 비교) sub_to_fp: Dict[str, str] = {} for bg in backgrounds: sub = bg.get("sub_location", "") first_fp = (bg.get("depends_on_fp") or [""])[0] if sub in sub_to_fp: if sub_to_fp[sub] != first_fp: raise ValueError( f"sub_location {sub!r}: backgrounds reference different floor plans " f"({sub_to_fp[sub]} vs {first_fp})" ) else: sub_to_fp[sub] = first_fp # invariant 5: 같은 sub_location N>1개 → 후속 bg가 depends_on_bg에 이전 같은 sub_location bg 포함 sub_to_bgs: Dict[str, List[str]] = defaultdict(list) for bg in backgrounds: sub_to_bgs[bg.get("sub_location", "")].append(bg.get("bg_id", "")) for sub, lst in sub_to_bgs.items(): if len(lst) <= 1: continue # 첫 bg는 chain root, 나머지는 같은 sub의 이전 bg 중 1+개를 ref해야 bg_meta = {bg.get("bg_id", ""): bg for bg in backgrounds} for i, bid in enumerate(lst[1:], start=1): deps = set(bg_meta[bid].get("depends_on_bg") or []) prior_in_sub = set(lst[:i]) if not (deps & prior_in_sub): raise ValueError( f"background {bid!r} in sub_location {sub!r} missing chain " f"(should depends_on_bg one of {prior_in_sub})" ) # invariant 6: gen_order는 topological — 각 entry의 deps가 앞에 있어야 seen: set = set() fp_meta = {fp.get("fp_id", ""): fp for fp in floor_plans} bg_meta = {bg.get("bg_id", ""): bg for bg in backgrounds} for ent in gen_order: if ent in fp_meta: for dep in fp_meta[ent].get("depends_on_fp") or []: if dep not in seen: raise ValueError( f"topological: floor_plan {ent!r} dep {dep!r} appears later" ) elif ent in bg_meta: bg = bg_meta[ent] for dep in (bg.get("depends_on_fp") or []) + (bg.get("depends_on_bg") or []): if dep not in seen: raise ValueError( f"topological: background {ent!r} dep {dep!r} appears later" ) seen.add(ent) def run_background_master_plan( *, user_prompt: str, expected_group_id: str, group_loc_ids: List[str], group_shot_ids: List[str], call_structured_fn: Callable[..., Dict[str, Any]], project_config: Optional[Dict[str, Any]] = None, opik_metadata: Optional[Dict[str, Any]] = None, max_retries: int = 3, backoff_base_sec: float = 2.0, sleep_fn: Callable[[float], None] = time.sleep, ) -> Dict[str, Any]: system = load_prompt(_MODULE, "system") schema = load_schema(_MODULE, "schema") loc_set = set(group_loc_ids) shot_set = set(group_shot_ids) last_err: Optional[Exception] = None for attempt in range(max_retries): try: result = call_structured_fn( step="background_master_plan", system_prompt=system, user_prompt=user_prompt, response_schema=schema, project_config=project_config, schema_name="background_master_plan", opik_metadata=opik_metadata, ) validate_master_plan_output(result, expected_group_id, loc_set, shot_set) return result except (ValueError, RuntimeError, TimeoutError, ConnectionError) as exc: last_err = exc logger.warning( "background_master_plan attempt %d/%d for %s failed: %s", attempt + 1, max_retries, expected_group_id, exc, ) if attempt + 1 < max_retries: sleep_fn(backoff_base_sec * (attempt + 1)) raise MasterPlanError( f"background_master_plan {expected_group_id!r} exhausted {max_retries} retries: {last_err}" ) ``` - [ ] **Step 7.4: 테스트 통과 확인** Run: `cd backend && pytest tests/pipeline/test_background_master_plan.py -v` Expected: PASS — 11 tests - [ ] **Step 7.5: commit** ```bash git add backend/app/modules/pipeline/background_master_plan.py backend/tests/pipeline/test_background_master_plan.py git commit -m "feat(p7-t7): background_master_plan LLM + 8 hard invariants" ``` --- ## Task 8: Step 2 step entry — BackgroundMasterPlanStep (그룹간 LLM 병렬) **Why**: 그룹 N개 → ThreadPool로 그룹별 LLM 병렬. 그룹간 의존성 없음(각 그룹은 독립 master plan). **Files:** - Create: `backend/app/core/steps/background_master_plan_step.py` - Modify: `backend/app/core/steps/__init__.py` (STEP_CLASSES 등록) - Modify: `backend/app/modules/llm/llm_client.py:PIPELINE_STEPS` (`background_master_plan` entry) - Modify: `backend/app/core/step_manifest.py` (manifest entry, order=19.55) - Create: `backend/tests/core/test_background_master_plan_step.py` - [ ] **Step 8.1: failing test 작성 (skeleton)** ```python # backend/tests/core/test_background_master_plan_step.py from unittest.mock import MagicMock, patch from app.core.steps.background_master_plan_step import ( BackgroundMasterPlanStep, SCHEMA_VERSION, PROMPT_VERSION, ) def test_step_disabled_when_mode_off(): with patch("app.core.config.settings.background_mode", "off"): step = BackgroundMasterPlanStep.__new__(BackgroundMasterPlanStep) step.project_id = "p"; step.episode_id = "e"; step.project_config = {} step.build_opik_metadata = MagicMock(return_value={}) result = step._execute() assert result["applicable_count"] == 0 def test_step_skips_prev_shot_only_groups(): """kind=prev_shot_ref 그룹은 master plan 호출 skip.""" classify_data = { "data": {"building_groups": [ {"group_id": "bg_x", "kind": "chain_bg", "members": [{"loc_id": "L01"}]}, {"group_id": "bg_y", "kind": "prev_shot_ref", "members": [{"loc_id": "L02"}]}, ]} } with patch("app.core.config.settings.background_mode", "on"): step = BackgroundMasterPlanStep.__new__(BackgroundMasterPlanStep) 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: { "background_classify": classify_data, "scene_save": {"data": {"segments": []}}, "shot_validator": {"data": {"scenes": []}}, "shot_selection": {"data": {"scenes": []}}, "visual_world_rules": {"data": {"rules_text": ""}}, }.get(sid)) with patch("app.core.steps.background_master_plan_step.run_background_master_plan") as mock_run: mock_run.return_value = { "group_id": "bg_x", "rationale_summary": "", "floor_plans": [], "backgrounds": [], "gen_order": [] } result = step._execute() # bg_y는 skip되어 master plan 호출 안 됨 called_groups = [c.kwargs["expected_group_id"] for c in mock_run.call_args_list] assert "bg_y" not in called_groups ``` - [ ] **Step 8.2: 테스트 fail 확인** Run: `cd backend && pytest tests/core/test_background_master_plan_step.py -v` Expected: FAIL — module not found - [ ] **Step 8.3: step 구현** ```python # backend/app/core/steps/background_master_plan_step.py """BackgroundMasterPlanStep — Phase 7 Step 2. 각 chain_bg group에 대해 LLM 1회 호출 → master plan 산출. 그룹간 ThreadPool 병렬. prev_shot_ref 그룹은 skip. """ from __future__ import annotations import json import logging from concurrent.futures import ThreadPoolExecutor, as_completed from pathlib import Path from typing import Any, Dict, List, Optional from app.core.step_runner import StepRunner logger = logging.getLogger(__name__) SCHEMA_VERSION = 1 PROMPT_VERSION = "1" class BackgroundMasterPlanStep(StepRunner): def _config_hash(self) -> str: import hashlib, json as _json from app.core.config import settings payload = { "background_mode": settings.background_mode, "model": "gpt-5.5", "schema_version": SCHEMA_VERSION, "prompt_version": PROMPT_VERSION, } return hashlib.sha256( _json.dumps(payload, sort_keys=True).encode("utf-8") ).hexdigest()[:16] def _load_prev_checkpoint(self, step_id: str) -> Optional[Dict[str, Any]]: from app.core.config import settings cp = ( Path(settings.projects_dir) / self.project_id / "checkpoints" / "episodes" / self.episode_id / step_id / "manifest.json" ) if cp.exists(): try: return json.loads(cp.read_text(encoding="utf-8")) except Exception as exc: logger.warning("background_master_plan: %s parse failed: %s", step_id, exc) return None def _execute(self, mode: str = "resume") -> Dict[str, Any]: from app.core.config import settings if settings.background_mode not in {"on", "floor_plan_anchored"}: return { "applicable_count": 0, "completed_count": 0, "failed_count": 0, "schema_version": SCHEMA_VERSION, "config_hash": self._config_hash(), "data": {}, } classify_cp = self._load_prev_checkpoint("background_classify") scene_save_cp = self._load_prev_checkpoint("scene_save") shot_validator_cp = self._load_prev_checkpoint("shot_validator") shot_selection_cp = self._load_prev_checkpoint("shot_selection") rules_cp = self._load_prev_checkpoint("visual_world_rules") groups = ((classify_cp or {}).get("data", {}) or {}).get("building_groups", []) or [] if not groups: return { "applicable_count": 1, "completed_count": 1, "failed_count": 0, "schema_version": SCHEMA_VERSION, "config_hash": self._config_hash(), "data": {"plans": {}}, } rules_text = ((rules_cp or {}).get("data", {}) or {}).get("rules_text", "") \ or ((rules_cp or {}).get("data", {}) or {}).get("text", "") or "" scene_segments = ((scene_save_cp or {}).get("data", {}) or {}).get("segments", []) or [] # selected shots (위 Task 5의 helper와 동일 패턴 — 인라인 구현) from app.core.steps.background_classify_step import _count_selected_shots_by_loc sel_by_loc = _count_selected_shots_by_loc(shot_validator_cp, shot_selection_cp) # group → relevant scenes/shots 빌드 from app.core.steps.background_master_plan_step import _select_scenes_for_group # self ref from app.modules.pipeline.background_master_plan import ( build_master_plan_user_prompt, run_background_master_plan, MasterPlanError, ) from app.modules.llm.llm_client import call_structured chain_groups = [g for g in groups if g.get("kind") == "chain_bg"] plans: Dict[str, Any] = {} failed = 0 opik_meta = self.build_opik_metadata() project_config = self.project_config def _process(g: Dict[str, Any]) -> Tuple[str, Dict[str, Any]]: gid = g["group_id"] group_loc_ids = [m["loc_id"] for m in g.get("members", []) or []] relevant_scenes, group_shot_ids = _select_scenes_for_group( g, scene_segments, shot_validator_cp, shot_selection_cp, ) user_prompt = build_master_plan_user_prompt( group=g, scenes=relevant_scenes, visual_world_rules=rules_text, ) try: plan = run_background_master_plan( user_prompt=user_prompt, expected_group_id=gid, group_loc_ids=group_loc_ids, group_shot_ids=group_shot_ids, call_structured_fn=call_structured, project_config=project_config, opik_metadata=opik_meta, ) return gid, {"status": "ok", "plan": plan} except MasterPlanError as exc: logger.error("master_plan: group %s failed: %s", gid, exc) return gid, {"status": "failed", "error": str(exc)[:200], "plan": None} max_workers = min(4, max(1, len(chain_groups))) if chain_groups: with ThreadPoolExecutor(max_workers=max_workers) as pool: futures = {pool.submit(_process, g): g["group_id"] for g in chain_groups} for fut in as_completed(futures): gid = futures[fut] try: _gid, res = fut.result() except Exception as exc: logger.error("master_plan: group %s thread raised: %s", gid, exc) res = {"status": "failed", "error": str(exc)[:200], "plan": None} plans[gid] = res if res.get("status") != "ok": failed += 1 # prev_shot_ref 그룹은 plan=null로 기록 for g in groups: if g.get("kind") == "prev_shot_ref": plans[g["group_id"]] = {"status": "prev_shot_only", "plan": None} # deterministic order: 입력 그룹 순서 보존 ordered_plans = {g["group_id"]: plans[g["group_id"]] for g in groups if g["group_id"] in plans} return { "applicable_count": 1, "completed_count": 1 if failed == 0 else 0, "failed_count": failed, "schema_version": SCHEMA_VERSION, "config_hash": self._config_hash(), "data": {"plans": ordered_plans}, } # Tuple import 빠뜨리지 않도록 모듈 상단에 추가: from typing import Tuple # noqa: E402 def _select_scenes_for_group( group: Dict[str, Any], scene_segments: List[Dict[str, Any]], shot_validator_cp: Optional[Dict[str, Any]], shot_selection_cp: Optional[Dict[str, Any]], ) -> Tuple[List[Dict[str, Any]], List[str]]: """그룹 멤버 loc_id가 등장하는 씬 + 그 씬의 shots만 추출.""" member_locs = {m["loc_id"] for m in group.get("members", []) or []} # selected shot indices sel_map: Dict[int, set] = {} for s in ((shot_selection_cp or {}).get("data", {}) or {}).get("scenes", []) or []: si = s.get("scene_index") if si is None: continue sel_map[int(si)] = set(s.get("selected_shot_indices", []) or []) # scene → shots shots_by_scene: Dict[int, List[Dict[str, Any]]] = {} for s in ((shot_validator_cp or {}).get("data", {}) or {}).get("scenes", []) or []: si = s.get("scene_index") if si is None: continue si_int = int(si) sel = sel_map.get(si_int, set()) for sh in s.get("shots", []) or []: shi = sh.get("shot_index") if shi is None or shi not in sel: continue loc = sh.get("location_id", "") or "" if loc not in member_locs: continue shots_by_scene.setdefault(si_int, []).append({ "shot_index": shi, "description": sh.get("description", "") or "", "location_id": loc, }) relevant_scene_indices = set(shots_by_scene.keys()) relevant_scenes: List[Dict[str, Any]] = [] for seg in scene_segments: si = seg.get("scene_index") if not isinstance(si, int) or si not in relevant_scene_indices: continue relevant_scenes.append({ "scene_index": si, "heading": seg.get("heading", ""), "text": seg.get("text", ""), "shots": shots_by_scene.get(si, []), }) relevant_scenes.sort(key=lambda x: x["scene_index"]) group_shot_ids = [ f"S{si}_Shot{sh['shot_index']}" for si in sorted(shots_by_scene.keys()) for sh in shots_by_scene[si] ] return relevant_scenes, group_shot_ids ``` - [ ] **Step 8.4: STEP_CLASSES + PIPELINE_STEPS + manifest 등록** `backend/app/core/steps/__init__.py`: ```python from app.core.steps.background_master_plan_step import BackgroundMasterPlanStep # STEP_CLASSES: "background_master_plan": BackgroundMasterPlanStep, ``` `backend/app/modules/llm/llm_client.py` PIPELINE_STEPS: ```python "background_master_plan": {"label": "배경 마스터플랜", "default": "gpt", "category": "analysis"}, ``` `backend/app/core/step_manifest.py` (`background_planner` entry 직전 또는 background_classify 직후): ```python "background_master_plan": { "label": "배경 마스터플랜", "category": "analysis", "order": 19.55, "default_model": "gpt", "provider": "openai", "depends_on": [ "background_classify", "scene_save", "shot_validator", "shot_selection", "visual_world_rules", ], "fan_out": False, "applicability": "if_background_mode", "step_type": "transform", "lifecycle": "active", }, ``` - [ ] **Step 8.5: 테스트 통과 확인** Run: `cd backend && pytest tests/core/test_background_master_plan_step.py -v` Expected: PASS — 2 tests - [ ] **Step 8.6: commit** ```bash git add backend/app/core/steps/background_master_plan_step.py backend/app/core/steps/__init__.py backend/app/modules/llm/llm_client.py backend/app/core/step_manifest.py backend/tests/core/test_background_master_plan_step.py git commit -m "feat(p7-t8): BackgroundMasterPlanStep + 그룹간 ThreadPool 병렬" ``` --- ## Task 9: Step 3 prompt + module (floor_plan_prompt) **Why**: 도면별 detailed t2i prompt 생성. plot-critical 시각 요소(커튼/시체 가림 등)를 LLM이 해당 도면 spec + scene segments에서 자연히 picking하도록. **Files:** - Create: `prompts/_base/floor_plan_prompt/1.YYYYMMDDHHmm/{system.md,user_template.md,schema.json}` - Create: `backend/app/modules/pipeline/floor_plan_prompt.py` - Create: `backend/tests/pipeline/test_floor_plan_prompt.py` - [ ] **Step 9.1: prompt 작성** ```bash mkdir -p prompts/_base/floor_plan_prompt/1.$(date +%Y%m%d%H%M) ``` system.md: ```markdown # Floor Plan Prompt — System You write t2i prompts for floor plan images (top-down architectural diagrams). Given a floor plan spec from the master plan, plus the related scene segments (verbatim) and shots that will use this floor plan, produce a detailed English t2i prompt for `gpt-image-2`. ## Rules 1. **Top-down architectural diagram** — clean line work, labeled rooms only if the master plan has sub_locations, NO photoreal furniture. 2. **Anchor key elements**: every persistent prop, doorway, window, or feature mentioned in the scenes that will be visible in downstream backgrounds MUST appear in the floor plan with location markers. 3. **Plot-critical visual devices** (e.g., a curtain that hides a body, a broken window, a hidden compartment) MUST be marked. 4. **English only** in t2i_prompt. Korean/Hanja/kana = error. 5. **No proper nouns from the work**. Generic descriptors only. ## Output JSON: `{"fp_id", "t2i_prompt", "key_elements"}`. `key_elements` is a list of short English phrases describing each anchored item. ``` user_template.md: ```markdown ## Floor plan fp_id: {fp_id} sub_location: {sub_location} scope: {scope} ## Backgrounds that will reference this floor plan {backgrounds_block} ## All shots using this floor plan {shots_block} ## Scene segments (verbatim) {scene_segments_block} ## visual_world_rules {visual_world_rules} ``` schema.json: ```json { "type": "object", "properties": { "fp_id": {"type": "string", "pattern": "^[a-z0-9][a-z0-9_]*$"}, "t2i_prompt": {"type": "string", "minLength": 30}, "key_elements": {"type": "array", "items": {"type": "string"}} }, "required": ["fp_id", "t2i_prompt", "key_elements"], "additionalProperties": false } ``` - [ ] **Step 9.2: failing test 작성** ```python # backend/tests/pipeline/test_floor_plan_prompt.py from unittest.mock import MagicMock import pytest from app.modules.pipeline.floor_plan_prompt import ( build_fp_user_prompt, validate_fp_prompt_output, run_floor_plan_prompt, FloorPlanPromptError, ) def test_build_user_prompt_includes_full_scene_text(): huge = "Z" * 30000 p = build_fp_user_prompt( fp_spec={"fp_id": "fp_living", "sub_location": "living_room", "scope": "x"}, applied_backgrounds=[{"bg_id": "cb_living_day"}], applied_shots=["S01_Shot1"], scene_segments=[{"scene_index": 1, "heading": "h", "text": huge}], visual_world_rules="rules", ) assert huge in p def test_validate_rejects_korean_in_t2i(): out = {"fp_id": "fp_x", "t2i_prompt": "옥탑방 floor plan", "key_elements": []} with pytest.raises(ValueError, match="non-ASCII"): validate_fp_prompt_output(out, expected_fp_id="fp_x") def test_validate_rejects_fp_id_mismatch(): out = {"fp_id": "fp_y", "t2i_prompt": "x" * 50, "key_elements": []} with pytest.raises(ValueError, match="fp_id mismatch"): validate_fp_prompt_output(out, expected_fp_id="fp_x") def test_run_returns_on_success(): out = {"fp_id": "fp_x", "t2i_prompt": "x" * 50, "key_elements": ["wall"]} fn = MagicMock(return_value=out) r = run_floor_plan_prompt( user_prompt="x", expected_fp_id="fp_x", call_structured_fn=fn, sleep_fn=lambda _: None, ) assert r == out def test_run_raises_on_exhaustion(): out = {"fp_id": "fp_y", "t2i_prompt": "x"*50, "key_elements": []} fn = MagicMock(return_value=out) with pytest.raises(FloorPlanPromptError): run_floor_plan_prompt( user_prompt="x", expected_fp_id="fp_x", call_structured_fn=fn, max_retries=2, sleep_fn=lambda _: None, ) ``` - [ ] **Step 9.3: 테스트 fail 확인** Run: `cd backend && pytest tests/pipeline/test_floor_plan_prompt.py -v` Expected: FAIL - [ ] **Step 9.4: 모듈 구현** ```python # backend/app/modules/pipeline/floor_plan_prompt.py """floor_plan_prompt — Phase 7 Step 3.""" from __future__ import annotations import logging import re import time from typing import Any, Callable, Dict, List, Optional from app.modules.prompt_loader import load_prompt, load_schema logger = logging.getLogger(__name__) _NON_ASCII_TEXT_RE = re.compile( r"[ㄱ-ㆎ가-힣" r"一-鿿㐀-䶿豈-﫿" r"぀-ヿ]" ) _MODULE = "floor_plan_prompt" class FloorPlanPromptError(Exception): pass def build_fp_user_prompt( fp_spec: Dict[str, Any], applied_backgrounds: List[Dict[str, Any]], applied_shots: List[str], scene_segments: List[Dict[str, Any]], visual_world_rules: str, ) -> str: template = load_prompt(_MODULE, "user_template") bg_lines = [f"- {b.get('bg_id', '')} (state={b.get('state_label', '')})" for b in applied_backgrounds] bg_block = "\n".join(bg_lines) or "(none)" shots_block = "\n".join(f"- {s}" for s in applied_shots) or "(none)" seg_lines: List[str] = [] for seg in scene_segments: si = seg.get("scene_index") heading = (seg.get("heading") or "").strip() text = (seg.get("text") or "").strip() seg_lines.append(f"### Scene {si} — {heading}" if heading else f"### Scene {si}") seg_lines.append(text) seg_lines.append("") seg_block = "\n".join(seg_lines).rstrip() or "(none)" return template.format( fp_id=fp_spec.get("fp_id", ""), sub_location=fp_spec.get("sub_location", ""), scope=fp_spec.get("scope", ""), backgrounds_block=bg_block, shots_block=shots_block, scene_segments_block=seg_block, visual_world_rules=visual_world_rules or "(none)", ) def validate_fp_prompt_output(output: Dict[str, Any], expected_fp_id: str) -> None: if output.get("fp_id") != expected_fp_id: raise ValueError(f"fp_id mismatch: got {output.get('fp_id')!r}, expected {expected_fp_id!r}") t2i = output.get("t2i_prompt") or "" if _NON_ASCII_TEXT_RE.search(t2i): raise ValueError(f"floor_plan_prompt {expected_fp_id} t2i contains non-ASCII text") if len(t2i) < 30: raise ValueError(f"floor_plan_prompt {expected_fp_id} t2i too short ({len(t2i)})") def run_floor_plan_prompt( *, user_prompt: str, expected_fp_id: str, call_structured_fn: Callable[..., Dict[str, Any]], project_config: Optional[Dict[str, Any]] = None, opik_metadata: Optional[Dict[str, Any]] = None, max_retries: int = 3, backoff_base_sec: float = 2.0, sleep_fn: Callable[[float], None] = time.sleep, ) -> Dict[str, Any]: system = load_prompt(_MODULE, "system") schema = load_schema(_MODULE, "schema") last_err: Optional[Exception] = None for attempt in range(max_retries): try: result = call_structured_fn( step="floor_plan_prompt", system_prompt=system, user_prompt=user_prompt, response_schema=schema, project_config=project_config, schema_name="floor_plan_prompt", opik_metadata=opik_metadata, ) validate_fp_prompt_output(result, expected_fp_id) return result except (ValueError, RuntimeError, TimeoutError, ConnectionError) as exc: last_err = exc logger.warning("floor_plan_prompt %s attempt %d/%d failed: %s", expected_fp_id, attempt + 1, max_retries, exc) if attempt + 1 < max_retries: sleep_fn(backoff_base_sec * (attempt + 1)) raise FloorPlanPromptError( f"floor_plan_prompt {expected_fp_id} exhausted {max_retries} retries: {last_err}" ) ``` - [ ] **Step 9.5: 테스트 통과 확인** Run: `cd backend && pytest tests/pipeline/test_floor_plan_prompt.py -v` Expected: PASS — 5 tests - [ ] **Step 9.6: commit** ```bash git add prompts/_base/floor_plan_prompt/ backend/app/modules/pipeline/floor_plan_prompt.py backend/tests/pipeline/test_floor_plan_prompt.py git commit -m "feat(p7-t9): floor_plan_prompt module + prompt v1" ``` --- ## Task 10: Step 3 step entry — FloorPlanPromptStep (level 병렬) **Files:** - Create: `backend/app/core/steps/floor_plan_prompt_step.py` - Modify: `backend/app/core/steps/__init__.py`, `llm_client.py`, `step_manifest.py` - Create: `backend/tests/core/test_floor_plan_prompt_step.py` - [ ] **Step 10.1: failing test 작성** ```python # backend/tests/core/test_floor_plan_prompt_step.py from unittest.mock import MagicMock, patch from app.core.steps.floor_plan_prompt_step import FloorPlanPromptStep def test_step_aggregates_fps_across_groups(): """모든 chain_bg group의 floor_plans를 모두 모아 LLM 호출.""" plans_data = {"data": {"plans": { "bg_x": {"status": "ok", "plan": { "group_id": "bg_x", "rationale_summary": "", "floor_plans": [{"fp_id": "fp_living", "sub_location": "living_room", "scope": "x", "depends_on_fp": []}], "backgrounds": [{ "bg_id": "cb_living_day", "loc_id": "L01", "sub_location": "living_room", "state_label": "day", "depends_on_fp": ["fp_living"], "depends_on_bg": [], "applies_to_shots": ["S01_Shot1"] }], "gen_order": ["fp_living", "cb_living_day"] }}, "bg_y": {"status": "prev_shot_only", "plan": None}, }}} with patch("app.core.config.settings.background_mode", "on"): step = FloorPlanPromptStep.__new__(FloorPlanPromptStep) 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: { "background_master_plan": plans_data, "scene_save": {"data": {"segments": [{"scene_index": 1, "heading": "h", "text": "t"}]}}, "shot_validator": {"data": {"scenes": [{"scene_index": 1, "shots": [{"shot_index": 1, "description": "d", "location_id": "L01"}]}]}}, "shot_selection": {"data": {"scenes": [{"scene_index": 1, "selected_shot_indices": [1]}]}}, "visual_world_rules": {"data": {"rules_text": ""}}, }.get(sid)) with patch("app.core.steps.floor_plan_prompt_step.run_floor_plan_prompt") as mock: mock.return_value = {"fp_id": "fp_living", "t2i_prompt": "x"*60, "key_elements": []} result = step._execute() assert result["completed_count"] == 1 assert "fp_living" in (result["data"].get("floor_plans") or {}) ``` - [ ] **Step 10.2: 테스트 fail 확인** Run: `cd backend && pytest tests/core/test_floor_plan_prompt_step.py -v` Expected: FAIL - [ ] **Step 10.3: step 구현** ```python # backend/app/core/steps/floor_plan_prompt_step.py """FloorPlanPromptStep — Phase 7 Step 3. 각 group plan의 floor_plans[]에 대해 LLM 1회씩 t2i prompt 생성. 도면 간에는 depends_on_fp DAG가 있으므로 level 병렬 가능 (대부분 single level). """ from __future__ import annotations import json import logging from concurrent.futures import ThreadPoolExecutor, as_completed from pathlib import Path from typing import Any, Dict, List, Optional, Tuple from app.core.step_runner import StepRunner logger = logging.getLogger(__name__) SCHEMA_VERSION = 1 PROMPT_VERSION = "1" class FloorPlanPromptStep(StepRunner): def _config_hash(self) -> str: import hashlib, json as _json from app.core.config import settings payload = { "background_mode": settings.background_mode, "schema_version": SCHEMA_VERSION, "prompt_version": PROMPT_VERSION, } return hashlib.sha256( _json.dumps(payload, sort_keys=True).encode("utf-8") ).hexdigest()[:16] def _load_prev_checkpoint(self, step_id: str) -> Optional[Dict[str, Any]]: from app.core.config import settings cp = ( Path(settings.projects_dir) / self.project_id / "checkpoints" / "episodes" / self.episode_id / step_id / "manifest.json" ) if cp.exists(): try: return json.loads(cp.read_text(encoding="utf-8")) except Exception as exc: logger.warning("floor_plan_prompt: %s parse failed: %s", step_id, exc) return None def _execute(self, mode: str = "resume") -> Dict[str, Any]: from app.core.config import settings if settings.background_mode not in {"on", "floor_plan_anchored"}: return self._not_applicable() plans_cp = self._load_prev_checkpoint("background_master_plan") scene_save_cp = self._load_prev_checkpoint("scene_save") shot_validator_cp = self._load_prev_checkpoint("shot_validator") shot_selection_cp = self._load_prev_checkpoint("shot_selection") rules_cp = self._load_prev_checkpoint("visual_world_rules") plans_map = ((plans_cp or {}).get("data", {}) or {}).get("plans", {}) or {} rules_text = ((rules_cp or {}).get("data", {}) or {}).get("rules_text", "") or "" scene_segments = ((scene_save_cp or {}).get("data", {}) or {}).get("segments", []) or [] # 모든 fp 수집 (group별 scope으로 scene 필터) from app.core.steps.background_master_plan_step import _select_scenes_for_group from app.modules.pipeline.floor_plan_prompt import ( build_fp_user_prompt, run_floor_plan_prompt, FloorPlanPromptError, ) from app.modules.llm.llm_client import call_structured fp_jobs: List[Dict[str, Any]] = [] for gid, entry in plans_map.items(): if entry.get("status") != "ok": continue plan = entry.get("plan") or {} for fp in plan.get("floor_plans") or []: # 이 fp가 적용되는 background와 shot 수집 applied_bgs = [bg for bg in (plan.get("backgrounds") or []) if fp["fp_id"] in (bg.get("depends_on_fp") or [])] applied_shots: List[str] = [] for bg in applied_bgs: applied_shots.extend(bg.get("applies_to_shots") or []) fp_jobs.append({ "fp_id": fp["fp_id"], "fp_spec": fp, "applied_bgs": applied_bgs, "applied_shots": list(dict.fromkeys(applied_shots)), # dedupe preserve order "group_id": gid, "members": [{"loc_id": loc, "label": ""} for loc in {bg.get("loc_id") for bg in applied_bgs if bg.get("loc_id")}], }) if not fp_jobs: return { "applicable_count": 1, "completed_count": 1, "failed_count": 0, "schema_version": SCHEMA_VERSION, "config_hash": self._config_hash(), "data": {"floor_plans": {}}, } # group별 scenes 캐시 (반복 호출 절약) group_scene_cache: Dict[str, Tuple[List[Dict[str, Any]], List[str]]] = {} for gid, entry in plans_map.items(): if entry.get("status") != "ok": continue plan = entry.get("plan") or {} group_member_locs = {bg.get("loc_id") for bg in (plan.get("backgrounds") or [])} synthetic_group = { "group_id": gid, "members": [{"loc_id": loc, "label": ""} for loc in group_member_locs if loc], } scenes, _shots = _select_scenes_for_group( synthetic_group, scene_segments, shot_validator_cp, shot_selection_cp, ) group_scene_cache[gid] = (scenes, _shots) # 도면간 DAG는 거의 다 root이지만 generic helper로 안전하게 처리 from app.modules.pipeline._dag_levels import compute_dag_levels items = {j["fp_id"]: {"parent_id": (j["fp_spec"].get("depends_on_fp") or [""])[0]} for j in fp_jobs} order = [j["fp_id"] for j in fp_jobs] renderable = set(order) levels = compute_dag_levels(order, items, renderable, parent_field="parent_id") results: Dict[str, Any] = {} failed = 0 opik = self.build_opik_metadata() def _process(job: Dict[str, Any]) -> Tuple[str, Dict[str, Any]]: scenes, _ = group_scene_cache.get(job["group_id"], ([], [])) up = build_fp_user_prompt( fp_spec=job["fp_spec"], applied_backgrounds=job["applied_bgs"], applied_shots=job["applied_shots"], scene_segments=scenes, visual_world_rules=rules_text, ) try: out = run_floor_plan_prompt( user_prompt=up, expected_fp_id=job["fp_id"], call_structured_fn=call_structured, project_config=self.project_config, opik_metadata=opik, ) return job["fp_id"], {"status": "ok", "t2i_prompt": out["t2i_prompt"], "key_elements": out.get("key_elements", []), "applied_shots": job["applied_shots"], "group_id": job["group_id"], "depends_on_fp": job["fp_spec"].get("depends_on_fp") or []} except FloorPlanPromptError as exc: logger.error("floor_plan_prompt: fp %s failed: %s", job["fp_id"], exc) return job["fp_id"], {"status": "failed", "error": str(exc)[:200], "group_id": job["group_id"]} jobs_by_id = {j["fp_id"]: j for j in fp_jobs} max_workers = 4 for level in levels: if not level: continue level_workers = max(1, min(max_workers, len(level))) with ThreadPoolExecutor(max_workers=level_workers) as pool: futures = {pool.submit(_process, jobs_by_id[fid]): fid for fid in level} for fut in as_completed(futures): fid = futures[fut] try: _fid, res = fut.result() except Exception as exc: logger.error("floor_plan_prompt: fp %s thread raised: %s", fid, exc) res = {"status": "failed", "error": str(exc)[:200]} results[fid] = res if res.get("status") != "ok": failed += 1 # input order 보존 ordered = {fid: results[fid] for fid in order if fid in results} return { "applicable_count": 1, "completed_count": 1 if failed == 0 else 0, "failed_count": failed, "schema_version": SCHEMA_VERSION, "config_hash": self._config_hash(), "data": {"floor_plans": ordered}, } def _not_applicable(self) -> Dict[str, Any]: return { "applicable_count": 0, "completed_count": 0, "failed_count": 0, "schema_version": SCHEMA_VERSION, "config_hash": self._config_hash(), "data": {}, } ``` - [ ] **Step 10.4: 등록 (STEP_CLASSES + PIPELINE_STEPS + manifest)** manifest entry: ```python "floor_plan_prompt": { "label": "도면 t2i 프롬프트", "category": "analysis", "order": 19.60, "default_model": "gpt", "provider": "openai", "depends_on": [ "background_master_plan", "scene_save", "shot_validator", "shot_selection", "visual_world_rules", ], "fan_out": False, "applicability": "if_background_mode", "step_type": "transform", "lifecycle": "active", }, ``` PIPELINE_STEPS: `"floor_plan_prompt": {"label": "도면 t2i 프롬프트", "default": "gpt", "category": "analysis"}` - [ ] **Step 10.5: 테스트 통과 + 회귀 확인** Run: `cd backend && pytest tests/core/test_floor_plan_prompt_step.py tests/pipeline/test_floor_plan_prompt.py -v` Expected: PASS - [ ] **Step 10.6: commit** ```bash git add backend/app/core/steps/floor_plan_prompt_step.py backend/app/core/steps/__init__.py backend/app/modules/llm/llm_client.py backend/app/core/step_manifest.py backend/tests/core/test_floor_plan_prompt_step.py git commit -m "feat(p7-t10): FloorPlanPromptStep + level 병렬 (order 19.60)" ``` --- ## Task 11: Step 4 — FloorPlanRenderStep (gpt-image-2, level 병렬) **Files:** - Create: `backend/app/modules/pipeline/floor_plan_render.py` - Create: `backend/app/core/steps/floor_plan_render_step.py` - Create: `backend/tests/pipeline/test_floor_plan_render.py` - Create: `backend/tests/core/test_floor_plan_render_step.py` > **Note**: Phase 5 `location_floor_plan_step.py:1-787`에 이미 gpt-image-2 호출 + ImageAsset row 생성 패턴이 있다. 그 패턴을 참고해 새 module로 깔끔히 다시 작성. Phase 5 step은 deprecate(applicability=disabled)만 한다. - [ ] **Step 11.1: failing test (module level)** ```python # backend/tests/pipeline/test_floor_plan_render.py from pathlib import Path from unittest.mock import MagicMock from app.modules.pipeline.floor_plan_render import render_one_floor_plan, FloorPlanRenderResult def test_render_one_floor_plan_text_only(tmp_path): client = MagicMock() resp = MagicMock() resp.data = [MagicMock(b64_json="aGVsbG8=")] # base64('hello') client.images.generate.return_value = resp out = tmp_path / "fp_x.png" res = render_one_floor_plan( openai_client=client, image_model="gpt-image-2", prompt="a top-down floor plan", out_path=out, ref_paths=[], ) assert isinstance(res, FloorPlanRenderResult) assert res.status == "ok" assert out.exists() and out.read_bytes() == b"hello" def test_render_one_floor_plan_with_ref(tmp_path): """parent fp_id ref가 있으면 images.edit 호출.""" client = MagicMock() resp = MagicMock() resp.data = [MagicMock(b64_json="aGVsbG8=")] client.images.edit.return_value = resp ref_png = tmp_path / "fp_parent.png" ref_png.write_bytes(b"PARENT") out = tmp_path / "fp_x.png" res = render_one_floor_plan( openai_client=client, image_model="gpt-image-2", prompt="x", out_path=out, ref_paths=[ref_png], ) assert res.status == "ok" assert client.images.edit.called assert not client.images.generate.called ``` - [ ] **Step 11.2: 테스트 fail 확인** Run: `cd backend && pytest tests/pipeline/test_floor_plan_render.py -v` Expected: FAIL - [ ] **Step 11.3: 모듈 구현 (slim — Phase 5 chain_bg_render의 `render_node_image` 패턴 재사용)** ```python # backend/app/modules/pipeline/floor_plan_render.py """floor_plan_render — Phase 7 Step 4. 도면 PNG를 gpt-image-2로 생성. depends_on_fp가 있으면 multi-image edit, 없으면 generate. """ from __future__ import annotations import base64 import contextlib import logging from dataclasses import dataclass, field from pathlib import Path from typing import Any, List, Optional logger = logging.getLogger(__name__) @dataclass class FloorPlanRenderResult: fp_id: str = "" status: str = "failed" png_path: str = "" attempts: int = 0 error: str = "" ref_used: str = "text_only" def render_one_floor_plan( *, openai_client: Any, image_model: str, prompt: str, out_path: Path, ref_paths: List[Path], fp_id: str = "", size: str = "1024x1024", quality: str = "high", max_attempts: int = 3, ) -> FloorPlanRenderResult: """단일 도면 PNG 생성. ref_paths 비어있으면 generate, 있으면 edit (multi-image).""" valid_refs = [p for p in ref_paths if p is not None and p.exists()] res = FloorPlanRenderResult( fp_id=fp_id, status="failed", ref_used=("text_only" if not valid_refs else "ref" if len(valid_refs) == 1 else f"refs_{len(valid_refs)}"), ) for attempt in range(1, max_attempts + 1): res.attempts = attempt try: if len(valid_refs) >= 2: with contextlib.ExitStack() as stack: files = [stack.enter_context(p.open("rb")) for p in valid_refs] resp = openai_client.images.edit( model=image_model, image=files, prompt=prompt, size=size, quality=quality, n=1, ) elif len(valid_refs) == 1: with valid_refs[0].open("rb") as f: resp = openai_client.images.edit( model=image_model, image=f, prompt=prompt, size=size, quality=quality, n=1, ) else: resp = openai_client.images.generate( model=image_model, prompt=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)) res.status = "ok" res.png_path = str(out_path) return res except Exception as exc: res.error = str(exc)[:200] logger.error("render_one_floor_plan %s attempt %d: %s", fp_id, attempt, exc) if attempt >= max_attempts: return res return res ``` - [ ] **Step 11.4: 모듈 테스트 통과** Run: `cd backend && pytest tests/pipeline/test_floor_plan_render.py -v` Expected: PASS — 2 tests - [ ] **Step 11.5: step entry 작성** ```python # backend/app/core/steps/floor_plan_render_step.py """FloorPlanRenderStep — Phase 7 Step 4. floor_plan_prompt 결과 + master plan의 depends_on_fp를 따라 PNG 생성. level 병렬 (compute_dag_levels). variant 패턴: ImageAsset(asset_type='floor_plan', variant_index=0, variant_label=fp_id, t2i_guide=prompt). """ from __future__ import annotations import json import logging from concurrent.futures import ThreadPoolExecutor, as_completed from pathlib import Path from typing import Any, Dict, List, Optional, Tuple from app.core.step_runner import StepRunner logger = logging.getLogger(__name__) SCHEMA_VERSION = 1 PROMPT_VERSION = "1" _SAFE_FP_RE = __import__("re").compile(r"^[a-z0-9][a-z0-9_]*$") class FloorPlanRenderStep(StepRunner): def _config_hash(self) -> str: import hashlib, json as _json from app.core.config import settings return hashlib.sha256( _json.dumps({ "background_mode": settings.background_mode, "schema_version": SCHEMA_VERSION, "prompt_version": PROMPT_VERSION, }, sort_keys=True).encode("utf-8") ).hexdigest()[:16] def _load_prev_checkpoint(self, step_id: str) -> Optional[Dict[str, Any]]: from app.core.config import settings cp = ( Path(settings.projects_dir) / self.project_id / "checkpoints" / "episodes" / self.episode_id / step_id / "manifest.json" ) if cp.exists(): try: return json.loads(cp.read_text(encoding="utf-8")) except Exception as exc: logger.warning("floor_plan_render: %s parse failed: %s", step_id, exc) return None def _execute(self, mode: str = "resume") -> Dict[str, Any]: from app.core.config import settings if settings.background_mode not in {"on", "floor_plan_anchored"}: return { "applicable_count": 0, "completed_count": 0, "failed_count": 0, "schema_version": SCHEMA_VERSION, "config_hash": self._config_hash(), "data": {}, } prompts_cp = self._load_prev_checkpoint("floor_plan_prompt") plans_cp = self._load_prev_checkpoint("background_master_plan") prompts_map = ((prompts_cp or {}).get("data", {}) or {}).get("floor_plans", {}) or {} plans_map = ((plans_cp or {}).get("data", {}) or {}).get("plans", {}) or {} # fp dependency 그래프 구성 items: Dict[str, Dict[str, Any]] = {} order: List[str] = [] for gid, entry in plans_map.items(): if entry.get("status") != "ok": continue for fp in entry["plan"].get("floor_plans") or []: fid = fp["fp_id"] if not _SAFE_FP_RE.match(fid): logger.error("floor_plan_render: unsafe fp_id %r — skip", fid) continue deps = fp.get("depends_on_fp") or [] items[fid] = {"parent_id": deps[0] if deps else "", "spec": fp, "group_id": gid} order.append(fid) renderable = {fid for fid in order if prompts_map.get(fid, {}).get("status") == "ok"} if not renderable: return { "applicable_count": 1, "completed_count": 1, "failed_count": 0, "schema_version": SCHEMA_VERSION, "config_hash": self._config_hash(), "data": {"floor_plans": {}}, } from app.modules.pipeline._dag_levels import compute_dag_levels from app.modules.pipeline.floor_plan_render import render_one_floor_plan from app.modules.llm.llm_client import get_openai_client # 가정 — 기존 helper 사용 # 만약 helper가 없으면 from openai import OpenAI; client = OpenAI() try: client = get_openai_client() except ImportError: from openai import OpenAI client = OpenAI() levels = compute_dag_levels(order, items, renderable, parent_field="parent_id") image_dir = Path(settings.projects_dir) / self.project_id / "episodes" / self.episode_id / "images" / "floor_plan" image_dir.mkdir(parents=True, exist_ok=True) rendered_paths: Dict[str, Path] = {} results: Dict[str, Any] = {} failed = 0 max_workers = 3 def _process(fid: str) -> Tuple[str, Dict[str, Any]]: prompt_entry = prompts_map[fid] spec = items[fid]["spec"] ref_paths: List[Path] = [] for dep in spec.get("depends_on_fp") or []: if dep in rendered_paths and rendered_paths[dep].exists(): ref_paths.append(rendered_paths[dep]) out_path = image_dir / f"{fid}.png" # path traversal 가드 try: out_path.resolve().relative_to(image_dir.resolve()) except (ValueError, OSError): return fid, {"status": "rejected_path", "error": "out_path escapes image_dir"} res = render_one_floor_plan( openai_client=client, image_model="gpt-image-2", prompt=prompt_entry["t2i_prompt"], out_path=out_path, ref_paths=ref_paths, fp_id=fid, ) return fid, { "status": res.status, "png_path": res.png_path, "attempts": res.attempts, "ref_used": res.ref_used, "error": res.error, "t2i_prompt": prompt_entry["t2i_prompt"], } for level in levels: if not level: continue snapshot = dict(rendered_paths) # thread-safe snapshot level_workers = max(1, min(max_workers, len(level))) with ThreadPoolExecutor(max_workers=level_workers) as pool: futures = {pool.submit(_process, fid): fid for fid in level} for fut in as_completed(futures): fid = futures[fut] try: _fid, res = fut.result() except Exception as exc: logger.error("floor_plan_render: fp %s thread raised: %s", fid, exc) res = {"status": "failed", "error": str(exc)[:200]} results[fid] = res if res.get("status") == "ok": rendered_paths[fid] = Path(res["png_path"]) else: failed += 1 ordered = {fid: results[fid] for fid in order if fid in results} return { "applicable_count": 1, "completed_count": 1 if failed == 0 else 0, "failed_count": failed, "schema_version": SCHEMA_VERSION, "config_hash": self._config_hash(), "data": {"floor_plans": ordered}, } ``` - [ ] **Step 11.6: step entry 테스트 작성 (mock OpenAI client)** ```python # backend/tests/core/test_floor_plan_render_step.py from pathlib import Path from unittest.mock import MagicMock, patch from app.core.steps.floor_plan_render_step import FloorPlanRenderStep def test_step_disabled_when_mode_off(): with patch("app.core.config.settings.background_mode", "off"): step = FloorPlanRenderStep.__new__(FloorPlanRenderStep) step.project_id = "p"; step.episode_id = "e" result = step._execute() assert result["applicable_count"] == 0 ``` - [ ] **Step 11.7: 등록 + 테스트 통과** manifest entry: ```python "floor_plan_render": { "label": "도면 이미지 생성", "category": "image", "order": 21.50, "default_model": "gpt", # gpt-image-2 (image gen은 별도 client) "provider": "openai", "depends_on": ["floor_plan_prompt", "background_master_plan"], "fan_out": False, "applicability": "if_background_mode", "step_type": "asset", "lifecycle": "active", }, ``` PIPELINE_STEPS: `"floor_plan_render": {"label": "도면 이미지 생성", "default": "gpt", "category": "image"}` Run: `cd backend && pytest tests/core/test_floor_plan_render_step.py tests/pipeline/test_floor_plan_render.py -v` Expected: PASS - [ ] **Step 11.8: commit** ```bash git add backend/app/modules/pipeline/floor_plan_render.py backend/app/core/steps/floor_plan_render_step.py backend/app/core/steps/__init__.py backend/app/modules/llm/llm_client.py backend/app/core/step_manifest.py backend/tests/pipeline/test_floor_plan_render.py backend/tests/core/test_floor_plan_render_step.py git commit -m "feat(p7-t11): FloorPlanRenderStep + gpt-image-2 + level 병렬 (order 21.50)" ``` --- ## Task 12: Step 5 prompt + module (background_prompt) **Why**: 배경별 detailed t2i prompt 생성. 도면 ref + 이전 배경 ref 활용 가이드 포함. **Files:** - Create: `prompts/_base/background_prompt/1.YYYYMMDDHHmm/{system.md,user_template.md,schema.json}` - Create: `backend/app/modules/pipeline/background_prompt.py` - Create: `backend/tests/pipeline/test_background_prompt.py` - [ ] **Step 12.1: prompt 작성** ```bash mkdir -p prompts/_base/background_prompt/1.$(date +%Y%m%d%H%M) ``` system.md: ```markdown # Background Prompt — System You write t2i prompts for photoreal background images. Given a single background spec (sub_location + state_label), the floor plan PNG that will be referenced, optional prior background PNG (chain ref), and the related scene segments, produce: 1. **t2i_prompt**: detailed English description of the photoreal scene at this state. 2. **ref_guide**: instructions about what NOT to redraw (already in ref images). 3. **shot_guides[]**: per-shot one-line hints for downstream scene_detail consumer. ## Rules 1. **English t2i_prompt only**. No Korean/Hanja/kana. ASCII safe. 2. **No proper nouns from the work** (character names, place names from input scenes — use generic descriptors). 3. **NO people, NO faces, NO blood-on-corpses depicted**. Background only — empty space, props, atmosphere. 4. **Plot-critical visual devices** mentioned in scene segments MUST be in the prompt (drawn curtain, broken window, scattered debris, etc). 5. **Reference reuse**: if `floor_plan_path` is provided, t2i_prompt must say "preserve room layout, furniture positions, and architectural elements from the reference floor plan exactly". If `prior_bg_paths` are provided, mention "match material, lighting style, and color palette to the prior background reference". 6. **State variation**: state_label drives lighting/mood/decor changes. Examples — `day_normal`, `dusk_ransacked`, `night_blood_curtain_drawn`. The prompt must reflect this state vividly. ## Output Strict JSON: `{"bg_id", "t2i_prompt", "ref_guide", "shot_guides[{shot_id, guide_text}]"}`. shot_guides covers every applies_to_shots entry. ``` user_template.md: ```markdown ## Background spec bg_id: {bg_id} loc_id: {loc_id} sub_location: {sub_location} state_label: {state_label} ## References floor_plan: {floor_plan_path_block} prior backgrounds: {prior_bg_paths_block} ## Applies to shots {applies_to_shots_block} ## Scene segments (verbatim) {scene_segments_block} ## visual_world_rules {visual_world_rules} ``` schema.json: ```json { "type": "object", "properties": { "bg_id": {"type": "string", "pattern": "^[a-z0-9][a-z0-9_]*$"}, "t2i_prompt": {"type": "string", "minLength": 50}, "ref_guide": {"type": "string"}, "shot_guides": { "type": "array", "items": { "type": "object", "properties": { "shot_id": {"type": "string"}, "guide_text": {"type": "string"} }, "required": ["shot_id", "guide_text"], "additionalProperties": false } } }, "required": ["bg_id", "t2i_prompt", "ref_guide", "shot_guides"], "additionalProperties": false } ``` - [ ] **Step 12.2: failing test** ```python # backend/tests/pipeline/test_background_prompt.py from unittest.mock import MagicMock import pytest from app.modules.pipeline.background_prompt import ( build_bg_user_prompt, validate_bg_prompt_output, run_background_prompt, BackgroundPromptError, ) def test_build_user_prompt_includes_full_scene_text(): huge = "B" * 30000 p = build_bg_user_prompt( bg_spec={"bg_id": "cb_x", "loc_id": "L01", "sub_location": "living_room", "state_label": "day_normal", "applies_to_shots": ["S01_Shot1"]}, floor_plan_path="/tmp/fp.png", prior_bg_paths=[], scene_segments=[{"scene_index": 1, "heading": "h", "text": huge}], visual_world_rules="rules", ) assert huge in p def test_validate_rejects_korean_t2i(): out = {"bg_id": "cb_x", "t2i_prompt": "옥탑방 interior", "ref_guide": "x", "shot_guides": [{"shot_id": "S01_Shot1", "guide_text": "x"}]} with pytest.raises(ValueError, match="non-ASCII"): validate_bg_prompt_output(out, "cb_x", {"S01_Shot1"}) def test_validate_rejects_missing_shot_guide(): out = {"bg_id": "cb_x", "t2i_prompt": "x" * 60, "ref_guide": "x", "shot_guides": []} with pytest.raises(ValueError, match="shot_guide"): validate_bg_prompt_output(out, "cb_x", {"S01_Shot1"}) def test_validate_passes_minimal(): out = {"bg_id": "cb_x", "t2i_prompt": "x" * 60, "ref_guide": "x", "shot_guides": [{"shot_id": "S01_Shot1", "guide_text": "h"}]} validate_bg_prompt_output(out, "cb_x", {"S01_Shot1"}) def test_run_retries_then_returns(): bad = {"bg_id": "cb_x", "t2i_prompt": "x"*60, "ref_guide": "x", "shot_guides": []} good = {"bg_id": "cb_x", "t2i_prompt": "x"*60, "ref_guide": "x", "shot_guides": [{"shot_id": "S01_Shot1", "guide_text": "g"}]} fn = MagicMock(side_effect=[bad, good]) r = run_background_prompt( user_prompt="x", expected_bg_id="cb_x", applies_to_shots=["S01_Shot1"], call_structured_fn=fn, sleep_fn=lambda _: None, ) assert r == good ``` - [ ] **Step 12.3: 모듈 구현** ```python # backend/app/modules/pipeline/background_prompt.py """background_prompt — Phase 7 Step 5.""" from __future__ import annotations import logging import re import time from typing import Any, Callable, Dict, List, Optional from app.modules.prompt_loader import load_prompt, load_schema logger = logging.getLogger(__name__) _NON_ASCII_TEXT_RE = re.compile( r"[ㄱ-ㆎ가-힣" r"一-鿿㐀-䶿豈-﫿" r"぀-ヿ]" ) _MODULE = "background_prompt" class BackgroundPromptError(Exception): pass def build_bg_user_prompt( bg_spec: Dict[str, Any], floor_plan_path: Optional[str], prior_bg_paths: List[str], scene_segments: List[Dict[str, Any]], visual_world_rules: str, ) -> str: template = load_prompt(_MODULE, "user_template") fp_block = floor_plan_path if floor_plan_path else "(none)" prior_block = "\n".join(f"- {p}" for p in prior_bg_paths) or "(none)" shots_block = "\n".join(f"- {s}" for s in (bg_spec.get("applies_to_shots") or [])) or "(none)" seg_lines: List[str] = [] for seg in scene_segments: si = seg.get("scene_index") heading = (seg.get("heading") or "").strip() text = (seg.get("text") or "").strip() seg_lines.append(f"### Scene {si} — {heading}" if heading else f"### Scene {si}") seg_lines.append(text) seg_lines.append("") seg_block = "\n".join(seg_lines).rstrip() or "(none)" return template.format( bg_id=bg_spec.get("bg_id", ""), loc_id=bg_spec.get("loc_id", ""), sub_location=bg_spec.get("sub_location", ""), state_label=bg_spec.get("state_label", ""), floor_plan_path_block=fp_block, prior_bg_paths_block=prior_block, applies_to_shots_block=shots_block, scene_segments_block=seg_block, visual_world_rules=visual_world_rules or "(none)", ) def validate_bg_prompt_output( output: Dict[str, Any], expected_bg_id: str, applies_to_shots: set, ) -> None: if output.get("bg_id") != expected_bg_id: raise ValueError(f"bg_id mismatch: got {output.get('bg_id')!r}, expected {expected_bg_id!r}") t2i = output.get("t2i_prompt") or "" if _NON_ASCII_TEXT_RE.search(t2i): raise ValueError(f"background_prompt {expected_bg_id} t2i contains non-ASCII") if len(t2i) < 50: raise ValueError(f"background_prompt {expected_bg_id} t2i too short ({len(t2i)})") guides = output.get("shot_guides") or [] guide_shot_ids = {g.get("shot_id") for g in guides if isinstance(g, dict)} missing = applies_to_shots - guide_shot_ids if missing: raise ValueError( f"background_prompt {expected_bg_id} missing shot_guides for: {sorted(missing)}" ) def run_background_prompt( *, user_prompt: str, expected_bg_id: str, applies_to_shots: List[str], call_structured_fn: Callable[..., Dict[str, Any]], project_config: Optional[Dict[str, Any]] = None, opik_metadata: Optional[Dict[str, Any]] = None, max_retries: int = 3, backoff_base_sec: float = 2.0, sleep_fn: Callable[[float], None] = time.sleep, ) -> Dict[str, Any]: system = load_prompt(_MODULE, "system") schema = load_schema(_MODULE, "schema") shot_set = set(applies_to_shots) last_err: Optional[Exception] = None for attempt in range(max_retries): try: result = call_structured_fn( step="background_prompt", system_prompt=system, user_prompt=user_prompt, response_schema=schema, project_config=project_config, schema_name="background_prompt", opik_metadata=opik_metadata, ) validate_bg_prompt_output(result, expected_bg_id, shot_set) return result except (ValueError, RuntimeError, TimeoutError, ConnectionError) as exc: last_err = exc logger.warning("background_prompt %s attempt %d/%d failed: %s", expected_bg_id, attempt + 1, max_retries, exc) if attempt + 1 < max_retries: sleep_fn(backoff_base_sec * (attempt + 1)) raise BackgroundPromptError( f"background_prompt {expected_bg_id} exhausted {max_retries} retries: {last_err}" ) ``` - [ ] **Step 12.4: 테스트 통과 + commit** Run: `cd backend && pytest tests/pipeline/test_background_prompt.py -v` Expected: PASS — 5 tests ```bash git add prompts/_base/background_prompt/ backend/app/modules/pipeline/background_prompt.py backend/tests/pipeline/test_background_prompt.py git commit -m "feat(p7-t12): background_prompt module + prompt v1" ``` --- ## Task 13: Step 5 step entry — BackgroundPromptStep (level 병렬) **Files:** - Create: `backend/app/core/steps/background_prompt_step.py` - Modify: `backend/app/core/steps/__init__.py`, `llm_client.py`, `step_manifest.py` - [ ] **Step 13.1: step 구현** ```python # backend/app/core/steps/background_prompt_step.py """BackgroundPromptStep — Phase 7 Step 5. 각 master plan의 backgrounds[]에 대해 LLM 1회씩 t2i prompt 생성. depends_on_bg DAG로 level 병렬화. """ from __future__ import annotations import json import logging from concurrent.futures import ThreadPoolExecutor, as_completed from pathlib import Path from typing import Any, Dict, List, Optional, Tuple from app.core.step_runner import StepRunner logger = logging.getLogger(__name__) SCHEMA_VERSION = 1 PROMPT_VERSION = "1" class BackgroundPromptStep(StepRunner): def _config_hash(self) -> str: import hashlib, json as _json from app.core.config import settings return hashlib.sha256( _json.dumps({ "background_mode": settings.background_mode, "schema_version": SCHEMA_VERSION, "prompt_version": PROMPT_VERSION, }, sort_keys=True).encode("utf-8") ).hexdigest()[:16] def _load_prev_checkpoint(self, step_id: str) -> Optional[Dict[str, Any]]: from app.core.config import settings cp = ( Path(settings.projects_dir) / self.project_id / "checkpoints" / "episodes" / self.episode_id / step_id / "manifest.json" ) if cp.exists(): try: return json.loads(cp.read_text(encoding="utf-8")) except Exception as exc: logger.warning("background_prompt: %s parse failed: %s", step_id, exc) return None def _execute(self, mode: str = "resume") -> Dict[str, Any]: from app.core.config import settings if settings.background_mode not in {"on", "floor_plan_anchored"}: return { "applicable_count": 0, "completed_count": 0, "failed_count": 0, "schema_version": SCHEMA_VERSION, "config_hash": self._config_hash(), "data": {}, } plans_cp = self._load_prev_checkpoint("background_master_plan") fp_render_cp = self._load_prev_checkpoint("floor_plan_render") scene_save_cp = self._load_prev_checkpoint("scene_save") shot_validator_cp = self._load_prev_checkpoint("shot_validator") shot_selection_cp = self._load_prev_checkpoint("shot_selection") rules_cp = self._load_prev_checkpoint("visual_world_rules") plans_map = ((plans_cp or {}).get("data", {}) or {}).get("plans", {}) or {} fp_paths = {fid: entry.get("png_path", "") for fid, entry in (((fp_render_cp or {}).get("data", {}) or {}).get("floor_plans", {}) or {}).items() if entry.get("status") == "ok"} # 모든 bg job 수집 from app.core.steps.background_master_plan_step import _select_scenes_for_group from app.modules.pipeline.background_prompt import ( build_bg_user_prompt, run_background_prompt, BackgroundPromptError, ) from app.modules.pipeline._dag_levels import compute_dag_levels from app.modules.llm.llm_client import call_structured bg_jobs: List[Dict[str, Any]] = [] items: Dict[str, Dict[str, Any]] = {} order: List[str] = [] for gid, entry in plans_map.items(): if entry.get("status") != "ok": continue for bg in entry["plan"].get("backgrounds") or []: bid = bg["bg_id"] bg_jobs.append({"bg_id": bid, "spec": bg, "group_id": gid, "members": [{"loc_id": bg.get("loc_id"), "label": ""}]}) deps = bg.get("depends_on_bg") or [] items[bid] = {"parent_id": deps[0] if deps else ""} order.append(bid) if not bg_jobs: return { "applicable_count": 1, "completed_count": 1, "failed_count": 0, "schema_version": SCHEMA_VERSION, "config_hash": self._config_hash(), "data": {"backgrounds": {}}, } renderable = set(order) levels = compute_dag_levels(order, items, renderable, parent_field="parent_id") # group별 scenes 캐시 scene_segments = ((scene_save_cp or {}).get("data", {}) or {}).get("segments", []) or [] group_scene_cache: Dict[str, List[Dict[str, Any]]] = {} for gid, entry in plans_map.items(): if entry.get("status") != "ok": continue plan = entry["plan"] group_member_locs = {bg.get("loc_id") for bg in plan.get("backgrounds") or []} synthetic_group = { "group_id": gid, "members": [{"loc_id": loc, "label": ""} for loc in group_member_locs if loc], } scenes, _ = _select_scenes_for_group( synthetic_group, scene_segments, shot_validator_cp, shot_selection_cp, ) group_scene_cache[gid] = scenes rules_text = ((rules_cp or {}).get("data", {}) or {}).get("rules_text", "") or "" results: Dict[str, Any] = {} failed = 0 max_workers = 4 opik = self.build_opik_metadata() jobs_by_id = {j["bg_id"]: j for j in bg_jobs} def _process(bid: str) -> Tuple[str, Dict[str, Any]]: job = jobs_by_id[bid] spec = job["spec"] # floor_plan PNG 경로 fp_first = (spec.get("depends_on_fp") or [""])[0] fp_path = fp_paths.get(fp_first, "") # prior bg PNG 경로 (이번 level 시작 snapshot) prior_bg_paths: List[str] = [] for dep in spec.get("depends_on_bg") or []: prior_entry = results.get(dep) or {} # 같은 step에서 결과는 prompt이지 PNG가 아님 — bg_render에서 반영. 여기서는 path 정보 X. # 따라서 prior_bg_paths는 user_prompt에 ref text로 들어가지만 path는 placeholder. if prior_entry.get("status") == "ok": prior_bg_paths.append(f"") up = build_bg_user_prompt( bg_spec=spec, floor_plan_path=fp_path, prior_bg_paths=prior_bg_paths, scene_segments=group_scene_cache.get(job["group_id"], []), visual_world_rules=rules_text, ) try: out = run_background_prompt( user_prompt=up, expected_bg_id=bid, applies_to_shots=spec.get("applies_to_shots") or [], call_structured_fn=call_structured, project_config=self.project_config, opik_metadata=opik, ) return bid, { "status": "ok", "t2i_prompt": out["t2i_prompt"], "ref_guide": out.get("ref_guide", ""), "shot_guides": out.get("shot_guides", []), "spec": spec, "group_id": job["group_id"], } except BackgroundPromptError as exc: logger.error("background_prompt: bg %s failed: %s", bid, exc) return bid, {"status": "failed", "error": str(exc)[:200], "spec": spec, "group_id": job["group_id"]} for level in levels: if not level: continue level_workers = max(1, min(max_workers, len(level))) with ThreadPoolExecutor(max_workers=level_workers) as pool: futures = {pool.submit(_process, bid): bid for bid in level} for fut in as_completed(futures): bid = futures[fut] try: _bid, res = fut.result() except Exception as exc: logger.error("background_prompt: bg %s thread raised: %s", bid, exc) res = {"status": "failed", "error": str(exc)[:200]} results[bid] = res if res.get("status") != "ok": failed += 1 ordered = {bid: results[bid] for bid in order if bid in results} return { "applicable_count": 1, "completed_count": 1 if failed == 0 else 0, "failed_count": failed, "schema_version": SCHEMA_VERSION, "config_hash": self._config_hash(), "data": {"backgrounds": ordered}, } ``` - [ ] **Step 13.2: 등록 + manifest entry** manifest: ```python "background_prompt": { "label": "배경 t2i 프롬프트", "category": "analysis", "order": 21.60, # floor_plan_render(21.50) 후 — fp PNG 경로 필요 "default_model": "gpt", "provider": "openai", "depends_on": [ "background_master_plan", "floor_plan_render", "scene_save", "shot_validator", "shot_selection", "visual_world_rules", ], "fan_out": False, "applicability": "if_background_mode", "step_type": "transform", "lifecycle": "active", }, ``` PIPELINE_STEPS: `"background_prompt": {"label": "배경 t2i 프롬프트", "default": "gpt", "category": "analysis"}` - [ ] **Step 13.3: smoke test + commit** ```bash cd backend && python -c "from app.core.steps import STEP_CLASSES; print('background_prompt' in STEP_CLASSES)" # Expected: True git add backend/app/core/steps/background_prompt_step.py backend/app/core/steps/__init__.py backend/app/modules/llm/llm_client.py backend/app/core/step_manifest.py git commit -m "feat(p7-t13): BackgroundPromptStep + level 병렬 (order 21.60)" ``` --- ## Task 14: Step 6 — BackgroundRenderStep (gpt-image-2, level 병렬) **Why**: 마지막 image step. 각 background를 도면 PNG + (있으면) 이전 배경 PNG ref로 multi-image edit. **출력 shape는 Phase 5 chain_bg_render와 호환** — `data.groups[group_id].{status, shot_guides, ...}`. 그래야 Phase 6 scene_context_loader 변경 0. **Files:** - Create: `backend/app/modules/pipeline/background_render.py` - Create: `backend/app/core/steps/background_render_step.py` - Create: `backend/tests/pipeline/test_background_render.py` > **중요 design 결정 (Phase 6 호환성)**: > Phase 7 출력의 shape는 Phase 5와 100% 호환되어야 한다. > - `data.groups[group_id]` (top-level) — group_id는 master plan의 bg_id로 매핑 > - 각 group entry: `{status, location_id, png_path, t2i_prompt, shot_guides, shot_ids, ...}` > - scene_context_loader.py:348 `data.groups`를 변경 없이 읽음 → Phase 6 consumer 그대로 동작. - [ ] **Step 14.1: failing test (module level — render_one_background)** ```python # backend/tests/pipeline/test_background_render.py from pathlib import Path from unittest.mock import MagicMock from app.modules.pipeline.background_render import render_one_background def test_render_with_fp_and_prior_bg(tmp_path): client = MagicMock() resp = MagicMock() resp.data = [MagicMock(b64_json="aGVsbG8=")] client.images.edit.return_value = resp fp = tmp_path / "fp.png"; fp.write_bytes(b"FP") prior = tmp_path / "prior.png"; prior.write_bytes(b"PR") out = tmp_path / "cb_x.png" res = render_one_background( openai_client=client, image_model="gpt-image-2", prompt="bg prompt", out_path=out, fp_path=fp, prior_bg_paths=[prior], bg_id="cb_x", ) assert res["status"] == "ok" assert client.images.edit.called # multi-image edit으로 호출 call_kwargs = client.images.edit.call_args.kwargs assert isinstance(call_kwargs["image"], list) assert len(call_kwargs["image"]) == 2 def test_render_text_only_when_no_refs(tmp_path): client = MagicMock() resp = MagicMock() resp.data = [MagicMock(b64_json="aGVsbG8=")] client.images.generate.return_value = resp out = tmp_path / "cb_x.png" res = render_one_background( openai_client=client, image_model="gpt-image-2", prompt="x", out_path=out, fp_path=None, prior_bg_paths=[], bg_id="cb_x", ) assert res["status"] == "ok" assert client.images.generate.called ``` - [ ] **Step 14.2: 모듈 구현** ```python # backend/app/modules/pipeline/background_render.py """background_render — Phase 7 Step 6. 배경 PNG를 gpt-image-2로 생성. fp_path + prior_bg_paths를 ref로 multi-image edit. """ from __future__ import annotations import base64 import contextlib import logging from pathlib import Path from typing import Any, Dict, List, Optional from app.modules.prompt_sanitizer import PromptSanitizer logger = logging.getLogger(__name__) _BACKGROUND_ONLY_REINFORCEMENT = ( "BACKGROUND-ONLY architectural still — empty space, NO people, NO faces, " "NO body posture, NO action, NO weapons, NO blood. " "Photoreal scene without any human figures.\n\n" ) def render_one_background( *, openai_client: Any, image_model: str, prompt: str, out_path: Path, fp_path: Optional[Path], prior_bg_paths: List[Path], bg_id: str = "", sanitizer: Optional[PromptSanitizer] = None, size: str = "1024x1024", quality: str = "high", max_attempts: int = 4, ) -> Dict[str, Any]: """단일 배경 PNG 생성. ref 우선순위: fp_path → prior_bg_paths → text_only.""" ref_paths: List[Path] = [] if fp_path is not None and fp_path.exists(): ref_paths.append(fp_path) for p in prior_bg_paths: if p is not None and p.exists(): ref_paths.append(p) info: Dict[str, Any] = { "status": "failed", "attempts": 0, "strategies": [], "ref_used": ("text_only" if not ref_paths else "fp_only" if len(ref_paths) == 1 and fp_path is not None else f"refs_{len(ref_paths)}"), "final_block_reason": None, } current_prompt = prompt if sanitizer is None: sanitizer = PromptSanitizer() for attempt in range(1, max_attempts + 1): info["attempts"] = attempt try: if len(ref_paths) >= 2: with contextlib.ExitStack() as stack: files = [stack.enter_context(p.open("rb")) for p in ref_paths] resp = openai_client.images.edit( model=image_model, image=files, prompt=current_prompt, size=size, quality=quality, n=1, ) elif len(ref_paths) == 1: with ref_paths[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" info["png_path"] = str(out_path) return info except Exception as exc: msg = str(exc).lower() is_moderation = any(k in msg for k in ( "moderation", "safety", "content_policy", "prohibited", "policy", "blocked", "violates", "violation" )) if is_moderation and attempt < max_attempts: try: s_attempt = min(attempt, 3) sr = sanitizer.sanitize( original_prompt=current_prompt, block_reason="SAFETY", block_categories=[], attempt=s_attempt, ) sanitized = sr["sanitized_prompt"] if not sanitized.lstrip().startswith("BACKGROUND-ONLY"): sanitized = _BACKGROUND_ONLY_REINFORCEMENT + sanitized current_prompt = sanitized info["strategies"].append(sr.get("strategy")) continue except Exception as se: info["final_block_reason"] = f"sanitize_failed: {se}"[:200] return info logger.error("background_render %s attempt %d: %s", bg_id, attempt, str(exc)[:200]) if attempt >= max_attempts: info["final_block_reason"] = str(exc)[:200] return info return info ``` - [ ] **Step 14.3: step entry 작성** ```python # backend/app/core/steps/background_render_step.py """BackgroundRenderStep — Phase 7 Step 6. 각 master plan background를 gpt-image-2로 PNG 생성. depends_on_bg DAG → level 병렬. 출력 shape는 Phase 5 chain_bg_render와 호환 (data.groups[bg_id]) — Phase 6 consumer 변경 0. """ from __future__ import annotations import json import logging import re from concurrent.futures import ThreadPoolExecutor, as_completed from pathlib import Path from typing import Any, Dict, List, Optional, Tuple from app.core.step_runner import StepRunner logger = logging.getLogger(__name__) SCHEMA_VERSION = 1 PROMPT_VERSION = "1" _SAFE_BG_RE = re.compile(r"^[a-z0-9][a-z0-9_]*$") class BackgroundRenderStep(StepRunner): def _config_hash(self) -> str: import hashlib, json as _json from app.core.config import settings return hashlib.sha256( _json.dumps({ "background_mode": settings.background_mode, "schema_version": SCHEMA_VERSION, "prompt_version": PROMPT_VERSION, }, sort_keys=True).encode("utf-8") ).hexdigest()[:16] def _load_prev_checkpoint(self, step_id: str) -> Optional[Dict[str, Any]]: from app.core.config import settings cp = ( Path(settings.projects_dir) / self.project_id / "checkpoints" / "episodes" / self.episode_id / step_id / "manifest.json" ) if cp.exists(): try: return json.loads(cp.read_text(encoding="utf-8")) except Exception as exc: logger.warning("background_render: %s parse failed: %s", step_id, exc) return None def _execute(self, mode: str = "resume") -> Dict[str, Any]: from app.core.config import settings if settings.background_mode not in {"on", "floor_plan_anchored"}: return { "applicable_count": 0, "completed_count": 0, "failed_count": 0, "schema_version": SCHEMA_VERSION, "config_hash": self._config_hash(), "data": {}, } plans_cp = self._load_prev_checkpoint("background_master_plan") prompts_cp = self._load_prev_checkpoint("background_prompt") fp_render_cp = self._load_prev_checkpoint("floor_plan_render") plans_map = ((plans_cp or {}).get("data", {}) or {}).get("plans", {}) or {} prompts_map = ((prompts_cp or {}).get("data", {}) or {}).get("backgrounds", {}) or {} fp_paths_str = {fid: entry.get("png_path", "") for fid, entry in (((fp_render_cp or {}).get("data", {}) or {}).get("floor_plans", {}) or {}).items() if entry.get("status") == "ok"} items: Dict[str, Dict[str, Any]] = {} order: List[str] = [] bg_specs: Dict[str, Dict[str, Any]] = {} for gid, entry in plans_map.items(): if entry.get("status") != "ok": continue for bg in entry["plan"].get("backgrounds") or []: bid = bg["bg_id"] if not _SAFE_BG_RE.match(bid): logger.error("background_render: unsafe bg_id %r — skip", bid) continue deps = bg.get("depends_on_bg") or [] items[bid] = {"parent_id": deps[0] if deps else ""} order.append(bid) bg_specs[bid] = bg renderable = {bid for bid in order if prompts_map.get(bid, {}).get("status") == "ok"} if not renderable: return { "applicable_count": 1, "completed_count": 1, "failed_count": 0, "schema_version": SCHEMA_VERSION, "config_hash": self._config_hash(), "data": {"groups": {}}, } from app.modules.pipeline._dag_levels import compute_dag_levels from app.modules.pipeline.background_render import render_one_background try: from app.modules.llm.llm_client import get_openai_client client = get_openai_client() except (ImportError, AttributeError): from openai import OpenAI client = OpenAI() levels = compute_dag_levels(order, items, renderable, parent_field="parent_id") image_dir = Path(settings.projects_dir) / self.project_id / "episodes" / self.episode_id / "images" / "background_chain" image_dir.mkdir(parents=True, exist_ok=True) rendered_paths: Dict[str, Path] = {} groups_out: Dict[str, Any] = {} failed = 0 max_workers = 4 def _process(bid: str) -> Tuple[str, Dict[str, Any]]: spec = bg_specs[bid] prompt_entry = prompts_map[bid] fp_first = (spec.get("depends_on_fp") or [""])[0] fp_path = Path(fp_paths_str[fp_first]) if fp_paths_str.get(fp_first) else None prior_bg_paths: List[Path] = [] for dep in spec.get("depends_on_bg") or []: if dep in rendered_paths and rendered_paths[dep].exists(): prior_bg_paths.append(rendered_paths[dep]) out_path = image_dir / f"{bid}.png" try: out_path.resolve().relative_to(image_dir.resolve()) except (ValueError, OSError): return bid, { "status": "rejected_path", "location_id": spec.get("loc_id", ""), "png_path": "", "t2i_prompt": prompt_entry.get("t2i_prompt", ""), "shot_guides": prompt_entry.get("shot_guides", []), "shot_ids": spec.get("applies_to_shots", []), "parent_id": (spec.get("depends_on_bg") or [""])[0], } info = render_one_background( openai_client=client, image_model="gpt-image-2", prompt=prompt_entry["t2i_prompt"], out_path=out_path, fp_path=fp_path, prior_bg_paths=prior_bg_paths, bg_id=bid, ) return bid, { "status": info.get("status", "failed"), "location_id": spec.get("loc_id", ""), "location_name": spec.get("sub_location", ""), "png_path": info.get("png_path", "") if info.get("status") == "ok" else "", "t2i_prompt": prompt_entry["t2i_prompt"], "shot_guides": prompt_entry.get("shot_guides", []), "shot_ids": spec.get("applies_to_shots", []), "scenes": [], "parent_id": (spec.get("depends_on_bg") or [""])[0], "ref_used": info.get("ref_used", "text_only"), "render_attempts": info.get("attempts", 0), "render_error": info.get("final_block_reason") or "", "variant_label": spec.get("state_label", ""), "floor_plan_used": fp_path is not None and fp_path.exists(), } for level in levels: if not level: continue level_workers = max(1, min(max_workers, len(level))) with ThreadPoolExecutor(max_workers=level_workers) as pool: futures = {pool.submit(_process, bid): bid for bid in level} for fut in as_completed(futures): bid = futures[fut] try: _bid, res = fut.result() except Exception as exc: logger.error("background_render: bg %s thread raised: %s", bid, exc) res = {"status": "failed", "render_error": str(exc)[:200]} groups_out[bid] = res if res.get("status") == "ok": rendered_paths[bid] = Path(res["png_path"]) else: failed += 1 # Phase 6 호환 shape: data.groups[bg_id] ordered = {bid: groups_out[bid] for bid in order if bid in groups_out} return { "applicable_count": 1, "completed_count": 1 if failed == 0 else 0, "failed_count": failed, "schema_version": SCHEMA_VERSION, "config_hash": self._config_hash(), "data": {"groups": ordered}, } ``` - [ ] **Step 14.4: 등록 + manifest entry** manifest: ```python "background_render": { "label": "배경 이미지 생성", "category": "image", "order": 24.70, "default_model": "gpt", "provider": "openai", "depends_on": ["background_prompt", "floor_plan_render", "background_master_plan"], "fan_out": False, "applicability": "if_background_mode", "step_type": "asset", "lifecycle": "active", }, ``` PIPELINE_STEPS: `"background_render": {"label": "배경 이미지 생성", "default": "gpt", "category": "image"}` - [ ] **Step 14.5: 테스트 통과** Run: `cd backend && pytest tests/pipeline/test_background_render.py -v` Expected: PASS — 2 tests - [ ] **Step 14.6: commit** ```bash git add backend/app/modules/pipeline/background_render.py backend/app/core/steps/background_render_step.py backend/app/core/steps/__init__.py backend/app/modules/llm/llm_client.py backend/app/core/step_manifest.py backend/tests/pipeline/test_background_render.py git commit -m "feat(p7-t14): BackgroundRenderStep + Phase 6 호환 shape (order 24.70)" ``` --- ## Task 15: Phase 6 consumer 호환성 검증 (변경 없음 확인) **Why**: Phase 7 `background_render`가 Phase 5와 동일한 `data.groups` shape를 산출하므로 `scene_context_loader._load_chain_bg_guide_by_shot`은 변경 없음. 회귀 테스트로 확정. **Files:** - Modify: `backend/tests/core/test_chain_bg_guide.py` (신규 케이스 1개 추가) - [ ] **Step 15.1: failing test 작성** ```python # backend/tests/core/test_chain_bg_guide.py 끝에 추가 def test_loader_parses_phase7_render_shape(tmp_path): """Phase 7 background_render 출력(data.groups[bg_id])을 Phase 5와 동일하게 파싱.""" from app.core.steps.scene_context_loader import _SceneContextLoaderImpl # 또는 정확한 클래스명 # Phase 7 background_render shape (data.groups[bg_id]) cp = { "data": {"groups": { "cb_living_day": { "status": "ok", "location_id": "L01", "shot_ids": ["S01_Shot1", "S01_Shot2"], "shot_guides": [ {"shot_id": "S01_Shot1", "guide": "TV in living room corner — preserve from ref"}, {"shot_id": "S01_Shot2", "guide": "Same TV — same angle"}, ], }, }} } # 직접 _load 호출은 인스턴스 + 체크포인트 파일 필요. 헬퍼 fixture 사용 (기존 패턴 따름). # 기존 test_loader_parses_phase5_groups_shape 케이스와 동일하게 작성. # ... (기존 fixture 재사용) ``` > **참고**: 기존 `test_chain_bg_guide.py`의 `test_loader_parses_phase5_groups_shape` 케이스와 100% 동일한 구조이므로, 사실상 신규 케이스는 그 케이스의 alias 검증으로 충분. 같은 shape을 사용하므로 별도 코드 변경 0. - [ ] **Step 15.2: 테스트 실행 — Phase 6 loader가 phase7 shape도 그대로 처리하는지 확인** Run: `cd backend && pytest tests/core/test_chain_bg_guide.py -v` Expected: PASS — 기존 7+1 케이스 모두 통과 (코드 변경 0) - [ ] **Step 15.3: commit** ```bash git add backend/tests/core/test_chain_bg_guide.py git commit -m "test(p7-t15): Phase 7 background_render shape 호환성 회귀 추가" ``` --- ## Task 16: Phase 5 deprecate (manifest applicability=disabled) **Why**: 4 step (`background_planner`, `background_chain_planning`, `location_floor_plan`, `background_chain_render`)을 deprecate. 코드 삭제는 별도 PR(Phase 7.4)로 미루고 manifest만 disable. **Files:** - Modify: `backend/app/core/step_manifest.py:559-695,747-761` (4 entry — applicability=disabled, lifecycle=deprecated) - Modify: `backend/app/core/applicability.py` (`_if_floor_plan_mode` validator 그대로 유지하되 deprecation comment) - Create: `backend/tests/core/test_phase5_deprecation.py` - [ ] **Step 16.1: failing test 작성** ```python # backend/tests/core/test_phase5_deprecation.py from app.core.step_manifest import STEP_MANIFEST def test_phase5_steps_deprecated(): deprecated = ["background_planner", "background_chain_planning", "location_floor_plan", "background_chain_render"] for step_id in deprecated: m = STEP_MANIFEST[step_id] assert m["lifecycle"] == "deprecated", f"{step_id} should be deprecated" assert m["applicability"] == "disabled", f"{step_id} should be disabled" def test_phase7_steps_active(): active = ["background_classify", "background_master_plan", "floor_plan_prompt", "floor_plan_render", "background_prompt", "background_render"] for step_id in active: m = STEP_MANIFEST[step_id] assert m["lifecycle"] == "active", f"{step_id} should be active" assert m["applicability"] == "if_background_mode", f"{step_id} should be if_background_mode" ``` - [ ] **Step 16.2: 테스트 fail 확인** Run: `cd backend && pytest tests/core/test_phase5_deprecation.py -v` Expected: FAIL — Phase 5 entry는 아직 active - [ ] **Step 16.3: manifest 4 entry 변경** `backend/app/core/step_manifest.py`에서 4 step entry의 `lifecycle`/`applicability` 변경: ```python "location_floor_plan": { # ... 기존 그대로 ... "applicability": "disabled", # was "if_floor_plan_mode" "lifecycle": "deprecated", # was "active" # Phase 7 (2026-04-29): floor_plan_prompt(19.60) + floor_plan_render(21.50)으로 분리됨. # 이 step은 코드 삭제 보류 + manifest disable. Phase 7.4 cleanup PR에서 제거. }, "background_planner": { # ... "applicability": "disabled", "lifecycle": "deprecated", }, "background_chain_planning": { # ... "applicability": "disabled", # was "on_demand" "lifecycle": "deprecated", }, "background_chain_render": { # ... "applicability": "disabled", "lifecycle": "deprecated", }, ``` - [ ] **Step 16.4: 테스트 통과 확인** Run: `cd backend && pytest tests/core/test_phase5_deprecation.py -v` Expected: PASS — 2 tests - [ ] **Step 16.5: 회귀 검사 (전체 테스트)** Run: `cd backend && pytest --timeout=120 -q` Expected: 회귀 0. Phase 5 step entry는 disabled 상태이므로 테스트가 직접 호출하지 않는 한 영향 없음. > **만약 Phase 5 step 테스트가 manifest active 상태에 의존하면**: 테스트를 skip 또는 mock으로 변경. 또는 해당 테스트들에 `pytest.mark.skip(reason="Phase 7 deprecation")` 추가. - [ ] **Step 16.6: commit** ```bash git add backend/app/core/step_manifest.py backend/tests/core/test_phase5_deprecation.py git commit -m "feat(p7-t16): Phase 5 step manifest deprecate (applicability=disabled)" ``` --- ## Task 17: env / config 정리 **Files:** - Modify: `backend/.env.example` (BACKGROUND_MODE 옵션 갱신) - Modify: `backend/app/core/config.py` (이미 Task 2에서 변경) - [ ] **Step 17.1: .env.example 갱신** ```bash # backend/.env.example 끝부분 # Phase 7 background pipeline (background_classify → master_plan → fp_prompt → fp_render → bg_prompt → bg_render) # off (default): 모든 background step disabled # on: Phase 7 6 step 활성 # floor_plan_anchored: legacy alias for "on" — Phase 5 환경 호환 BACKGROUND_MODE=off # Phase 7 render 병렬 worker 수 (default 4, max 8) # BACKGROUND_RENDER_WORKERS=4 ``` - [ ] **Step 17.2: 파일 차이 확인 + commit** ```bash git diff backend/.env.example git add backend/.env.example git commit -m "docs(p7-t17): .env.example BACKGROUND_MODE 옵션 갱신" ``` --- ## Task 18: 통합 smoke test (mock LLM/image) **Why**: Step 1 → 6 end-to-end가 mock 환경에서 정상 통과하는지 확인. invariant 위반 0건 검증. **Files:** - Create: `backend/tests/integration/test_phase7_e2e.py` - [ ] **Step 18.1: failing test 작성** ```python # backend/tests/integration/test_phase7_e2e.py """Phase 7 6 step end-to-end smoke test (mock LLM + mock image).""" from pathlib import Path from unittest.mock import patch, MagicMock import pytest @pytest.mark.integration def test_phase7_pipeline_smoke(tmp_path, monkeypatch): """6 step 모두 정상 통과 + Phase 6 loader 호환 확인.""" # 시나리오: 1 group, 1 fp, 2 bg, 2 shot monkeypatch.setenv("BACKGROUND_MODE", "on") # 1. background_classify mock classify_out = {"building_groups": [ {"group_id": "bg_x", "anchor_loc": "L01", "kind": "chain_bg", "rationale": "indoor anchor"}, ]} # 2. background_master_plan mock master_out = { "group_id": "bg_x", "rationale_summary": "...", "floor_plans": [{"fp_id": "fp_living", "sub_location": "living_room", "scope": "living area", "depends_on_fp": []}], "backgrounds": [ {"bg_id": "cb_living_day", "loc_id": "L01", "sub_location": "living_room", "state_label": "day_normal", "depends_on_fp": ["fp_living"], "depends_on_bg": [], "applies_to_shots": ["S01_Shot1"]}, {"bg_id": "cb_living_dusk", "loc_id": "L01", "sub_location": "living_room", "state_label": "dusk", "depends_on_fp": ["fp_living"], "depends_on_bg": ["cb_living_day"], "applies_to_shots": ["S01_Shot2"]}, ], "gen_order": ["fp_living", "cb_living_day", "cb_living_dusk"], } # 3. floor_plan_prompt mock fp_prompt_out = {"fp_id": "fp_living", "t2i_prompt": "x"*60, "key_elements": []} # 4. background_prompt mock (per bg) bg_prompt_outs = { "cb_living_day": {"bg_id": "cb_living_day", "t2i_prompt": "x"*60, "ref_guide": "y", "shot_guides": [{"shot_id": "S01_Shot1", "guide_text": "g1"}]}, "cb_living_dusk": {"bg_id": "cb_living_dusk", "t2i_prompt": "x"*60, "ref_guide": "y", "shot_guides": [{"shot_id": "S01_Shot2", "guide_text": "g2"}]}, } # gpt-image-2 mock img_resp = MagicMock() img_resp.data = [MagicMock(b64_json="aGVsbG8=")] # 호출 순서별 LLM mock side_effect llm_responses = [classify_out, master_out, fp_prompt_out, bg_prompt_outs["cb_living_day"], bg_prompt_outs["cb_living_dusk"]] call_index = [0] def llm_side_effect(**kwargs): r = llm_responses[call_index[0]] call_index[0] += 1 return r with patch("app.modules.llm.llm_client.call_structured", side_effect=llm_side_effect): with patch("app.modules.pipeline.floor_plan_render.openai_client", create=True) as mock_fp_client: mock_fp_client.images.generate.return_value = img_resp mock_fp_client.images.edit.return_value = img_resp # ... step 1-6 순차 호출 (실제 step_runner 호출은 fixture 필요) # 이 통합 테스트는 step_runner integration 패턴이 어떤지에 따라 # 다음 단계에서 정확한 fixture로 채워야 함. pass # TODO: 위 mock 패턴을 실제 step_runner integration test fixture로 이식. # 본 테스트는 plan에 명시된 6 step 순서 + 호환성을 보장하는 sanity check. ``` > **Note**: 본 통합 테스트는 step_runner의 실제 invoke 패턴이 정해진 후에 채운다. 단위 테스트 17개(Task 1-15)가 통과하면 통합은 production E2E (Task 19)에서 검증. - [ ] **Step 18.2: 통합 fixture 채우기 (선택, 시간 허락 시)** step_runner의 기존 통합 fixture(`backend/tests/integration/conftest.py` 등)을 참조하여 6 step 순차 호출 구현. - [ ] **Step 18.3: 단위 테스트 회귀 검사** Run: `cd backend && pytest -x --timeout=120 -q` Expected: 1153 + 신규 ~30개 = ~1180+ tests PASS, 회귀 0 - [ ] **Step 18.4: commit** ```bash git add backend/tests/integration/test_phase7_e2e.py git commit -m "test(p7-t18): Phase 7 e2e smoke (mock LLM + image)" ``` --- ## Task 19: Production E2E 검증 (PID c00bbe19 EP1) **Why**: 실제 LLM + gpt-image-2로 전체 6 step 실행. Phase 5 production 결과(같은 sub-room 무작위 / parent chain 끊김 / plot-critical 누락)가 모두 해결되었는지 검증. **Files:** - Create: `scripts/p7_production_verify.sh` - Create: `backend/tests/integration/test_phase7_production_data.py` (검증용 — 체크포인트 read-only inspection) - [ ] **Step 19.1: 환경 준비 (수동)** ```bash # 1. .env 토글 활성 sed -i.bak 's/^# BACKGROUND_MODE=.*/BACKGROUND_MODE=on/' backend/.env # 2. 서버 재시작 cd backend && pkill -f uvicorn; sleep 1; nohup uvicorn app.main:app --host 0.0.0.0 --port 8001 > /tmp/p7_server.log 2>&1 & # 3. login + run 6 step sequentially via API (force=true) PID=c00bbe19-a9b5-463f-acfc-806f2e820258 EP=fe165e3a-19c2-4a0f-9acb-e0c9bab0ee5a TOKEN=$(curl -sX POST http://localhost:8001/api/v1/auth/login -d 'username=admin&password=admin123' | jq -r .access_token) for step in background_classify background_master_plan floor_plan_prompt floor_plan_render background_prompt background_render; do curl -X POST -H "Authorization: Bearer $TOKEN" \ "http://localhost:8001/api/v1/projects/$PID/episodes/$EP/steps/$step?force=true" sleep 5 done ``` - [ ] **Step 19.2: 결과 검증 — 같은 sub_location 일관성** ```python # backend/tests/integration/test_phase7_production_data.py """Phase 7 production 결과 검증 — 체크포인트 read-only inspection.""" import json import os from pathlib import Path import pytest PID = "c00bbe19-a9b5-463f-acfc-806f2e820258" EP = "fe165e3a-19c2-4a0f-9acb-e0c9bab0ee5a" def _cp_dir(): return Path(os.environ.get("PROJECTS_DIR", "/Users/manta/Documents/Projects/TheRoad-I1/projects")) \ / PID / "checkpoints" / "episodes" / EP @pytest.mark.production def test_master_plan_same_sublocation_shares_floor_plan(): """같은 sub_location의 모든 background는 같은 fp 참조.""" cp = json.loads((_cp_dir() / "background_master_plan" / "manifest.json").read_text()) for gid, entry in cp["data"]["plans"].items(): if entry.get("status") != "ok": continue plan = entry["plan"] sub_to_fp = {} for bg in plan["backgrounds"]: sub = bg["sub_location"] fp = bg["depends_on_fp"][0] if sub in sub_to_fp: assert sub_to_fp[sub] == fp, \ f"{gid}: sub_location {sub} has split fp ({sub_to_fp[sub]} vs {fp})" sub_to_fp[sub] = fp @pytest.mark.production def test_master_plan_no_id_mismatch(): """master plan의 모든 ID가 자체 일관성 (gen_order vs floor_plans+backgrounds).""" cp = json.loads((_cp_dir() / "background_master_plan" / "manifest.json").read_text()) for gid, entry in cp["data"]["plans"].items(): if entry.get("status") != "ok": continue plan = entry["plan"] fp_ids = {fp["fp_id"] for fp in plan["floor_plans"]} bg_ids = {bg["bg_id"] for bg in plan["backgrounds"]} assert set(plan["gen_order"]) == fp_ids | bg_ids, \ f"{gid}: gen_order mismatch" @pytest.mark.production def test_background_render_all_groups_completed(): """background_render 출력에서 status='ok'인 group 비율이 90%+ — Phase 5는 7/18(39%)였음.""" cp = json.loads((_cp_dir() / "background_render" / "manifest.json").read_text()) groups = cp["data"]["groups"] if not groups: pytest.skip("no groups rendered") ok_count = sum(1 for g in groups.values() if g["status"] == "ok") ratio = ok_count / len(groups) assert ratio >= 0.9, f"only {ok_count}/{len(groups)} ({ratio:.0%}) groups completed — too low" @pytest.mark.production def test_phase6_consumer_picks_up_shot_guides(): """scene_context_loader가 새 shape에서 shot_guides를 추출하는지 확인.""" from app.core.steps.scene_context_loader import _SceneContextLoaderImpl # 정확한 클래스명 확인 필요 # 로더 호출 → chain_bg_guide_by_shot 비어있지 않음을 확인 # ... (loader fixture에 따라 작성) pass ``` - [ ] **Step 19.3: 검증 스크립트 실행** ```bash cd backend && pytest tests/integration/test_phase7_production_data.py -v -m production ``` Expected: 3 PASS (위 4번째는 fixture 정해진 후 채움) - [ ] **Step 19.4: HTML viewer 갱신 (선택)** `projects/c00bbe19-.../images/fe165e3a-.../index.html` 갱신 — Phase 7 도면 + 배경 모두 표시. 동일 sub_location 다른 state가 일관된 sub-room을 그렸는지 시각적으로 확인. - [ ] **Step 19.5: 환경 정리** ```bash sed -i.bak 's/^BACKGROUND_MODE=on/# BACKGROUND_MODE=on/' backend/.env pkill -f uvicorn ``` - [ ] **Step 19.6: 검증 결과를 메모리에 기록 + commit** 다음 세션 메모리: - `next_session_phase7_plan.md` → `session_20260430_phase7.md` 갱신 - `MEMORY.md` 인덱스에 Phase 7 entry 추가 ```bash git add scripts/p7_production_verify.sh backend/tests/integration/test_phase7_production_data.py git commit -m "test(p7-t19): production 검증 스크립트 + master plan invariant 검증" ``` --- ## Task 20: 듀얼 코드 리뷰 (Codex + Claude) **Why**: 기존 메모리 `feedback_dual_code_review.md` — 중요 변경은 반드시 두 리뷰어 거침. - [ ] **Step 20.1: Codex 리뷰 실행** ```bash # main 브랜치 기준으로 Phase 7 신규 변경 전체에 대해 codex-rescue agent 실행 # (subagent-driven-development 워크플로에서 자연히 수행됨) ``` - [ ] **Step 20.2: Claude code-reviewer agent 실행** (subagent-driven-development의 spec-reviewer + code-quality-reviewer 단계로 자연 수행) - [ ] **Step 20.3: 두 리뷰 결과 통합 + 수정** 각 task별로 spec-reviewer ✅ + code-quality-reviewer ✅ 후 다음 task 진행 (subagent-driven-development 패턴). - [ ] **Step 20.4: 최종 반영 후 push** ```bash git push origin main ``` --- ## Self-Review 다음을 plan 작성 후 점검: ### 1. Spec coverage | Spec 요구사항 | 구현 task | |---|---| | Step 1+2 한 step (분류) | Task 3-5 | | Step 3 그룹별 master plan | Task 6-8 | | Step 4 도면 prompt | Task 9-10 | | Step 5 도면 render | Task 11 | | Step 6 배경 prompt | Task 12-13 | | Step 7 배경 render | Task 14 | | 그룹간 master plan 병렬 | Task 8 ThreadPoolExecutor | | level batch 병렬화 | Task 1 generic helper + Task 10/11/13/14 모두 사용 | | Phase 5 deprecate | Task 16 | | 마이그레이션 (aggressive) | Task 16 (manifest disable) — 데이터는 그대로, 새 PID 권장 | | BACKGROUND_MODE on/off/floor_plan_anchored | Task 2 + Task 17 | | Phase 6 consumer 호환 (data.groups shape) | Task 14 출력 shape + Task 15 검증 | | 시나리오 의존성 0 | 모든 task의 invariant validator (한글/한자 검사) | | truncation 금지 | Task 4/7/9/12 user_prompt builder가 무절단 inject | | plot-critical 시각 요소 | Task 6 system prompt + Task 9/12 detailed LLM call | → Spec 8 sections 모두 task로 매핑됨. 미커버 0. ### 2. Placeholder scan - "TODO" 검색: Task 18.1 fixture 부분("TODO: 위 mock 패턴을 실제 step_runner integration test fixture로 이식") — 통합 fixture는 step_runner 패턴이 환경마다 달라 plan 단계에서 완전 코드화 어려움. plan에 그대로 둔 후 implementer 판단. - "implement later" / "fill in details": 0건. - "Add appropriate error handling": 0건 — 모든 retry/fail-fast 정책은 명시 (3 retry, ValueError vs TypeError 구분). - "Similar to Task N": 0건 — 모든 task의 코드 블록은 자기완결. ### 3. Type consistency - `compute_dag_levels(order, items, renderable_set, parent_field)`: Task 1 정의 → Task 10/11/13/14 호출, signature 일치 ✅ - `run_background_classify(user_prompt, all_loc_ids, expected_group_ids, call_structured_fn, ...)`: Task 4 정의 → Task 5 호출 ✅ - `run_background_master_plan(user_prompt, expected_group_id, group_loc_ids, group_shot_ids, call_structured_fn, ...)`: Task 7 정의 → Task 8 호출 ✅ - `run_floor_plan_prompt(user_prompt, expected_fp_id, ...)`: Task 9 정의 → Task 10 호출 ✅ - `run_background_prompt(user_prompt, expected_bg_id, applies_to_shots, ...)`: Task 12 정의 → Task 13 호출 ✅ - `render_one_floor_plan(...)`: Task 11 정의 + 호출 (같은 task) ✅ - `render_one_background(...)`: Task 14 정의 + 호출 (같은 task) ✅ - 체크포인트 shape `data.groups[bg_id].{status, png_path, t2i_prompt, shot_guides, shot_ids, parent_id, ref_used, location_id, ...}`: Task 14 출력 = Task 15 검증 = Phase 5 chain_bg_render shape 일치 ✅ → 시그니처/property 이름 모두 task 간 일치. --- ## Execution Handoff **Plan complete and saved to** `docs/superpowers/plans/2026-04-29-phase7-background-redesign.md`. **Two execution options:** **1. Subagent-Driven (recommended)** — 각 task당 fresh subagent 디스패치, two-stage review (spec compliance + code quality), 빠른 iteration. - REQUIRED SUB-SKILL: `superpowers:subagent-driven-development` - Task 20개 — 각 task 약 30-90분 (subagent + 리뷰 cycle 포함). 총 ~1-1.5주. **2. Inline Execution** — 본 세션에서 task별 batch 실행 + checkpoint review. - REQUIRED SUB-SKILL: `superpowers:executing-plans` **선택 후 진행.**