# Phase 5 Background Planner 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:** `background_mode=floor_plan_anchored` 활성 시 LLM-driven planner가 frequency rule + building group을 분석한 후 4-step sequential pipeline (`background_planner` → `location_floor_plan` → `background_chain_planning` → `background_chain_render`)으로 도면/배경을 생성한다. 모든 location 무차별 생성을 중단하고 3+ 샷 + 건물 내부만 도면+chain_bg, 그 외는 prev_shot_ref hybrid로 fallback. **Architecture:** 1 신규 step (`background_planner`, gpt-5.5 1 LLM call) + 3 기존 step redesign (sequential 멀티턴 + planner output 소비). HARD 제약: 한 번에 모든 배경 작성 금지 — 각 floor_plan/chain_bg = 별도 LLM call. legacy `chain_only` mode fallback 보존으로 회귀 0건. **Tech Stack:** Python 3.11, FastAPI, SQLAlchemy ORM, gpt-5.5 (text), gpt-image-2 (image), Pydantic v2 schemas with runtime enum injection, pytest, Phase 4 ExitStack multi-image edit 패턴. --- ## Pre-Flight (Implementer 시작 전) - 작업 브랜치: `main` (직접 commit, Phase 3+4 패턴) - 시작 commit: `8f6d9f1` (Phase 4 끝) - 현재 timestamp 사용 (예: `1.202604292000`) — 버전 형식 `버전.YYYYMMDDHHmm` - 모든 commit에 `Co-Authored-By: Claude Opus 4.7 (1M context) ` 포함 ## ImageAsset 데이터 모델 결정 (사용자 확정) **C안 채택**: outlook 패턴(C01 + O01/O02) → backgrounds에 동일 적용. ``` ImageAsset 컬럼 신규 3개: variant_index: int default 0 # 0=floor_plan, 1+=chain_bg states variant_label: str default "v00" # "v00", "v01"... 표시용 t2i_guide: text nullable # shot t2i 주입용 가이드 Saving 정책: L05_v00 = floor_plan (primary location of building_group "okt_room") L05_v01, L05_v02 = chain_bg variants (낮/저녁/밤 등) L11_v01, L11_v02 = chain_bg variants (자체 floor_plan 없음, planner.floor_plans[].location_ids로 L05_v00 PNG ref) ``` **Phase 5 scope**: ImageAsset 컬럼 추가 + variant_label/t2i_guide 영속화까지. **Phase 6 (이월)**: scene_detail이 shot의 location+state로 variant lookup → t2i_guide inject. --- ## Task 분해 개요 (10 tasks) > **모델 정책 (사용자 확정)**: 모든 implementer + reviewer agent는 **opus** 사용. | # | Task | Scope | Tests | Risk | |:---:|---|---|:---:|:---:| | T1 | background_planner schema + prompts | 신규 prompt v1 | 0 (data only) | 낮음 | | T1.5 | **ImageAsset migration (variant_index/label/t2i_guide)** | alembic + model | ~3 | 낮음 | | T2 | background_planner pipeline 모듈 | LLM call + validate | ~10 | 중간 | | T3 | BackgroundPlannerStep + manifest 등록 | Step + applicability | ~8 | 중간 | | T4 | floor_plan v2 prompt + image ref_paths 파라미터 | prompt v2 + pipeline 시그니처 | ~5 | 낮음 | | T5 | LocationFloorPlanStep sequential redesign + variant=v00 영속 | step _execute rewrite | ~8 | **높음** | | T6 | chain_bg_planning group-based redesign | step + pipeline 분기 | ~8 | 중간 | | T7 | chain_bg_render chain_bg_order driver + variant counter + t2i_guide UPSERT | step + pipeline + DB | ~7 | 중간 | | T8 | manifest depends_on + integration | 1 manifest 수정 + E2E | ~5 | 낮음 | | T9 | 회귀 + 듀얼 리뷰 + push | 전체 회귀 + 메모리 | — | — | --- ## Task 1: background_planner Schema + Prompts **Files:** - Create: `prompts/_base/background_planner/1.{ts}/system.md` - Create: `prompts/_base/background_planner/1.{ts}/user_template.md` - Create: `prompts/_base/background_planner/1.{ts}/schema.json` 여기서 `{ts}` = 작업 시작 시간 (예: `202604292000`). - [ ] **Step 1.1: schema.json 작성** design.md §4의 schema spec 그대로. 6 top-level keys: `rationale_summary`, `floor_plans`, `floor_plan_order`, `chain_bg_groups`, `chain_bg_order`, `prev_shot_only`. `additionalProperties: false`. 모든 nested object도 `required` 명시 + `additionalProperties: false`. - [ ] **Step 1.2: system.md 작성** 핵심 섹션: 1. Role (background generation planner) 2. Frequency Rules (HARD: 1/2-shot never, 3+outdoor never, 3+indoor required) 3. Building Group Rule (실내 + 관련 외부 같은 floor_plan에 묶기) 4. Order Constraints (floor_plan_order: building_group 인접; chain_bg_order: parent before child) 5. Output Discipline (English snake_case ID, 한국어 rationale 1줄, 작품 고유명사 0건) - [ ] **Step 1.3: user_template.md 작성** placeholders: `{visual_world_rules}`, `{location_lines}`, `{scene_blocks}`. `{scene_blocks}`는 selected shot description을 scene별로 그룹핑. shot_id format = `S{scene_index}_Shot{shot_index}`. - [ ] **Step 1.4: 프롬프트 로드 검증 (smoke)** ```bash python3 -c " from app.modules.prompt_loader import load_prompt, load_schema sys = load_prompt('background_planner', 'system') tpl = load_prompt('background_planner', 'user_template') sch = load_schema('background_planner', 'schema') print('system len:', len(sys)) print('template placeholders:', tpl.count('{')) print('schema keys:', list(sch['properties'].keys())) " ``` 기대: system len > 800, template placeholders ≥ 3, schema keys 6개. - [ ] **Step 1.5: Commit** ```bash git add prompts/_base/background_planner/ git commit -m "$(cat <<'EOF' feat(p5): background_planner prompt v1 + schema Phase 5 background_planner step의 system prompt + user template + JSON schema 신설. frequency rule(1/2/3+ shot), building group, order constraints 명시. Co-Authored-By: Claude Opus 4.7 (1M context) EOF )" ``` --- ## Task 1.5: ImageAsset Migration (variant_index, variant_label, t2i_guide) **Files:** - Create: `backend/alembic/versions/{rev}_phase5_image_asset_variants.py` - Modify: `backend/app/models/image_asset.py` (또는 ORM 정의 위치) - Modify: `backend/tests/db/test_image_asset_migration.py` (있으면, 신규 가능) > **목적**: outlook 패턴(C01 캐릭터 + O01/O02 의상)을 backgrounds로 확장. floor_plan = v00, chain_bg = v01/v02.../ each location. 기존 Phase 3+4 row는 default `variant_index=0, variant_label="v00", t2i_guide=NULL`로 backfill (default 값으로 자동). - [ ] **Step 1.5.1: alembic revision 생성** ```bash cd backend && alembic revision --autogenerate -m "phase5_image_asset_variants" ``` 생성된 파일 확인 후 수동 편집 (autogenerate 신뢰 X — 명시적으로 작성): ```python def upgrade() -> None: op.add_column("image_asset", sa.Column("variant_index", sa.Integer(), nullable=False, server_default="0")) op.add_column("image_asset", sa.Column("variant_label", sa.String(length=8), nullable=False, server_default="v00")) op.add_column("image_asset", sa.Column("t2i_guide", sa.Text(), nullable=True)) # 기존 row는 server_default로 자동 backfill됨 def downgrade() -> None: op.drop_column("image_asset", "t2i_guide") op.drop_column("image_asset", "variant_label") op.drop_column("image_asset", "variant_index") ``` - [ ] **Step 1.5.2: ImageAsset ORM 모델 업데이트** ```python # app/models/image_asset.py class ImageAsset(Base): ... variant_index: Mapped[int] = mapped_column(Integer, nullable=False, default=0, server_default="0") variant_label: Mapped[str] = mapped_column(String(8), nullable=False, default="v00", server_default="v00") t2i_guide: Mapped[Optional[str]] = mapped_column(Text, nullable=True) ``` `__table_args__`에 unique constraint가 (project_id, asset_type, short_id)였다면 → (project_id, asset_type, short_id, variant_index)로 확장. **기존 unique 위반 회귀 검증 필수.** - [ ] **Step 1.5.3: migration 적용 + 회귀 테스트** ```bash cd backend && alembic upgrade head pytest tests/db/ -v pytest tests/core/test_location_floor_plan.py -v # Phase 3 회귀 pytest tests/core/test_phase4_floor_plan_chain.py -v # Phase 4 회귀 ``` 기대: 모두 PASS. 기존 row가 default 값으로 자동 backfill됨. - [ ] **Step 1.5.4: rollback 검증 (사용자 안전망)** ```bash alembic downgrade -1 alembic upgrade head ``` 기대: down/up 양방향 동작. - [ ] **Step 1.5.5: Commit** ```bash git add backend/alembic/versions/ backend/app/models/image_asset.py git commit -m "$(cat <<'EOF' feat(p5): ImageAsset variant 컬럼 추가 (variant_index/label/t2i_guide) outlook 패턴 확장: floor_plan=v00, chain_bg=v01+. shot t2i 주입용 t2i_guide 영속화. 기존 Phase 3+4 row는 server_default로 자동 backfill. Unique constraint에 variant_index 포함. Co-Authored-By: Claude Opus 4.7 (1M context) EOF )" ``` --- ## Task 2: background_planner Pipeline 모듈 **Files:** - Create: `backend/app/modules/pipeline/background_planner.py` - Create: `backend/tests/core/test_background_planner_pipeline.py` (이 task의 unit test) - [ ] **Step 2.1: 실패 테스트 — build_planner_user_prompt** ```python # tests/core/test_background_planner_pipeline.py def test_build_planner_user_prompt_includes_all_selected_shots(): selected_shots_by_scene = { 1: [{"shot_index": 1, "description": "S1 desc1"}], 5: [{"shot_index": 1, "description": "S5 desc1"}, {"shot_index": 2, "description": "S5 desc2"}], } locations = [ {"short_id": "L01", "name": "옥탑방", "kind": "indoor", "description": "..."}, {"short_id": "L02", "name": "골목", "kind": "outdoor", "description": "..."}, ] prompt = build_planner_user_prompt( selected_shots_by_scene=selected_shots_by_scene, location_lines=[f"{l['short_id']} ({l['kind']}): {l['name']}" for l in locations], visual_world_rules="rule1", scene_primary_locations={1: "L01", 5: "L02"}, ) assert "S1_Shot1" in prompt assert "S5_Shot1" in prompt and "S5_Shot2" in prompt assert "L01" in prompt and "L02" in prompt assert "rule1" in prompt ``` - [ ] **Step 2.2: 테스트 실행 → fail 확인** ```bash cd backend && pytest tests/core/test_background_planner_pipeline.py -v ``` 기대: ImportError (모듈 미존재). - [ ] **Step 2.3: 최소 구현 — build_planner_user_prompt** `pipeline/background_planner.py`: ```python from typing import Dict, List from app.modules.prompts.loader import load_prompt def build_planner_user_prompt( selected_shots_by_scene: Dict[int, List[Dict]], location_lines: List[str], visual_world_rules: str, scene_primary_locations: Dict[int, str], ) -> str: template = load_prompt("background_planner", "user_template") scene_blocks = [] for scene_idx in sorted(selected_shots_by_scene.keys()): shots = selected_shots_by_scene[scene_idx] primary = scene_primary_locations.get(scene_idx, "") block = f"Scene {scene_idx} (primary_location={primary}):\n" for sh in shots: block += f" S{scene_idx}_Shot{sh['shot_index']}: {sh.get('description', '')}\n" scene_blocks.append(block) return template.format( visual_world_rules=visual_world_rules or "(none)", location_lines="\n".join(location_lines) or "(none)", scene_blocks="\n".join(scene_blocks) or "(none)", ) ``` - [ ] **Step 2.4: pass 확인** ```bash pytest tests/core/test_background_planner_pipeline.py::test_build_planner_user_prompt_includes_all_selected_shots -v ``` 기대: PASS. - [ ] **Step 2.5: 실패 테스트 — validate_planner_output (frequency violation)** ```python def test_validate_planner_output_rejects_floor_plan_for_2_shot_location(): plan = { "rationale_summary": "x", "floor_plans": [{ "id": "FP_L02", "building_group": "g1", "location_ids": ["L02"], "primary_location_id": "L02", "rationale": "위반", "shot_count": 2, }], "floor_plan_order": ["FP_L02"], "chain_bg_groups": [], "chain_bg_order": [], "prev_shot_only": [], } with pytest.raises(ValueError, match="2 shot"): validate_planner_output(plan, all_location_ids=["L02"], all_shot_ids=["S1_Shot1", "S1_Shot2"]) ``` - [ ] **Step 2.6: 구현 — validate_planner_output** `validate_planner_output(plan, all_location_ids, all_shot_ids)` semantic invariants: 1. 모든 floor_plans[].location_ids ⊆ all_location_ids 2. 모든 chain_bg_groups[].shot_ids ⊆ all_shot_ids 3. floor_plan_order = floor_plans IDs (set equality) 4. chain_bg_order = chain_bg_groups IDs 5. parent_id이 비지 않으면 chain_bg_order에서 자신보다 앞에 있어야 6. **frequency rule**: floor_plans[].shot_count ≥ 3 (2-shot에 도면 금지) 7. building_group이 같은 floor_plans는 floor_plan_order에서 인접 8. 한국어 0건 in 모든 ID/snake_case 필드 (rationale_summary는 한국어 OK) - [ ] **Step 2.7: pass 확인 + 추가 invariant 테스트** `validate_planner_output` 7개 invariant마다 fail/pass 테스트 추가 (최소 6 tests). - [ ] **Step 2.8: run_background_planner 구현** ```python def run_background_planner( *, project_config: Dict[str, Any], user_prompt: str, location_short_ids: List[str], shot_ids: List[str], opik_metadata: Dict[str, Any], call_structured_fn: Callable, ) -> Dict[str, Any]: system = load_prompt("background_planner", "system") schema = load_schema("background_planner", "schema") schema = _inject_runtime_enums(schema, location_short_ids, shot_ids) last_err = None for attempt in range(3): try: result = call_structured_fn( step="background_planner", system_prompt=system, user_prompt=user_prompt, response_schema=schema, project_config=project_config, schema_name="background_planner", opik_metadata=opik_metadata, ) validate_planner_output(result, location_short_ids, shot_ids) return result except (ValueError, LLMError) as e: last_err = e time.sleep(2 * (attempt + 1)) raise PlannerError(f"background_planner failed after 3 retries: {last_err}") ``` - [ ] **Step 2.9: run_background_planner 테스트** mock `call_structured_fn`으로 retry 동작 검증 (1회 실패 → 2회 성공 / 3회 실패 → raise). - [ ] **Step 2.10: pytest 전체 실행** ```bash pytest tests/core/test_background_planner_pipeline.py -v ``` 기대: 10 passed. - [ ] **Step 2.11: Commit** ```bash git add backend/app/modules/pipeline/background_planner.py backend/tests/core/test_background_planner_pipeline.py git commit -m "$(cat <<'EOF' feat(p5): background_planner pipeline 모듈 build_planner_user_prompt + validate_planner_output (7 semantic invariants) + run_background_planner (gpt-5.5 retry 3회). 10 unit tests. Co-Authored-By: Claude Opus 4.7 (1M context) EOF )" ``` --- ## Task 3: BackgroundPlannerStep + Manifest **Files:** - Create: `backend/app/core/steps/background_planner_step.py` - Modify: `backend/app/core/steps/__init__.py` - Modify: `backend/app/core/step_manifest.py` - Modify: `backend/app/modules/llm/llm_client.py` - Create: `backend/tests/core/test_background_planner_step.py` - [ ] **Step 3.1: 실패 테스트 — applicability** ```python def test_background_planner_skipped_when_mode_off(tmp_path, monkeypatch): monkeypatch.setattr("app.core.config.settings.background_mode", "off") step = BackgroundPlannerStep(...) result = step._execute(mode="resume") assert result["applicable_count"] == 0 assert result["completed_count"] == 0 ``` - [ ] **Step 3.2: BackgroundPlannerStep 구현** ```python class BackgroundPlannerStep(StepRunner): step_id = "background_planner" def _execute(self, mode="resume") -> Dict[str, Any]: if settings.background_mode != "floor_plan_anchored": return {"applicable_count": 0, "completed_count": 0, "failed_count": 0, "data": {}} # 1. 의존 체크포인트 로드 shot_validator_cp = self._load_prev_checkpoint("shot_validator") shot_selection_cp = self._load_prev_checkpoint("shot_selection") scene_director_cp = self._load_prev_checkpoint("scene_director") 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") # 2. selected shot 추출 selected_shots_by_scene = _extract_selected_shots(shot_validator_cp, shot_selection_cp) # 3. location lines (kind 보강 from entity_detail) location_lines, all_location_ids = _build_location_lines(entity_merge_cp, entity_detail_cp) # 4. scene → primary_location scene_primary = {sc["scene_index"]: sc.get("primary_location", "") for sc in scene_director_cp["data"]["scenes"]} # 5. shot_id 모음 all_shot_ids = [f"S{sc}_Shot{sh['shot_index']}" for sc, shots in selected_shots_by_scene.items() for sh in shots] # 6. user_prompt user_prompt = build_planner_user_prompt( selected_shots_by_scene=selected_shots_by_scene, location_lines=location_lines, visual_world_rules=rules_cp["data"].get("rules_text", "") if rules_cp else "", scene_primary_locations=scene_primary, ) # 7. LLM 호출 plan = run_background_planner( project_config=self.project_config, user_prompt=user_prompt, location_short_ids=all_location_ids, shot_ids=all_shot_ids, opik_metadata=self.build_opik_metadata(), call_structured_fn=call_structured, ) return { "applicable_count": 1, "completed_count": 1, "failed_count": 0, "data": plan, } ``` - [ ] **Step 3.3: __init__.py 등록** ```python from app.core.steps.background_planner_step import BackgroundPlannerStep STEP_CLASSES["background_planner"] = BackgroundPlannerStep ``` - [ ] **Step 3.4: step_manifest.py entry 추가** design.md §9의 spec 그대로. order=19.55, applicability="if_floor_plan_mode", fan_out=False, depends_on=6 step. 위치는 `background_chain_planning` (19.6) 직전. - [ ] **Step 3.5: llm_client.py PIPELINE_STEPS 추가** ```python "background_planner": {"label": "배경 생성 Planner", "default": "gpt-5.5", "category": "analysis"}, ``` - [ ] **Step 3.6: 추가 테스트 (8 케이스)** - mode=off → applicable_count=0 - mode=chain_only → applicable_count=0 - mode=floor_plan_anchored + 의존 모두 OK → completed_count=1 - shot_selection 없음 → 빈 selected_shots → 빈 plan - 6 location 중 3+ shot indoor 1개 → floor_plans 1개 - 모두 outdoor → floor_plans 0개 - LLM retry 모두 실패 → failed_count=1, raise X (graceful) - validate 위반 → retry 후 raise - [ ] **Step 3.7: pytest 실행** ```bash pytest tests/core/test_background_planner_step.py -v ``` 기대: 8 passed. - [ ] **Step 3.8: 회귀 — manifest 불변식 + 의존 그래프** ```bash pytest tests/core/test_step_manifest.py -v pytest tests/core/test_image_steps_after_analysis.py -v ``` 기대: 모두 PASS (background_planner=19.55 < image=21+, 불변식 만족). - [ ] **Step 3.9: Commit** ```bash git add backend/app/core/steps/background_planner_step.py backend/app/core/steps/__init__.py backend/app/core/step_manifest.py backend/app/modules/llm/llm_client.py backend/tests/core/test_background_planner_step.py git commit -m "$(cat <<'EOF' feat(p5): BackgroundPlannerStep 신설 (order 19.55) mode=floor_plan_anchored 시 1 LLM call로 frequency 분석 + building group + floor_plan_order/chain_bg_order 결정. mode=off/chain_only는 not_applicable. manifest + STEP_CLASSES + PIPELINE_STEPS 등록. 8 unit tests. Co-Authored-By: Claude Opus 4.7 (1M context) EOF )" ``` --- ## Task 4: floor_plan v2 Prompt + Image ref_paths **Files:** - Create: `prompts/_base/location_floor_plan/2.{ts}/system.md` - Create: `prompts/_base/location_floor_plan/2.{ts}/user_template.md` - Modify: `backend/app/modules/pipeline/location_floor_plan.py` (`generate_floor_plan_image` 시그니처) - Modify: `backend/tests/core/test_location_floor_plan.py` (ref_paths 케이스) - [ ] **Step 4.1: prompt v2 system.md 작성** v1 system.md 내용 보존 + 마지막에 추가: ``` ## Multi-turn Context (if provided) [PREVIOUS FLOOR PLANS] block lists previously generated floor plans in this episode (one short summary per line). When this block is present: - For floor plans of related buildings (same building_group), maintain consistent architectural style, era, and world-building context. - For unrelated buildings, ensure visual distinctness (different style/era). [BUILDING GROUP CONTEXT] block lists which previous floor plans belong to the SAME building group as the current one. The PNG outputs of those plans are attached as image references. Use them as spatial layout authority for shared interior/exterior structures. ``` - [ ] **Step 4.2: prompt v2 user_template.md 작성** v1 5 placeholders + 2 추가: ``` {previous_floor_plans_block} {building_group_context} ``` 빈 문자열일 때 자동 생략되도록 (헤더+내용 통째로 비우기). - [ ] **Step 4.3: 실패 테스트 — generate_floor_plan_image with ref_paths** ```python def test_generate_floor_plan_image_uses_multi_image_edit_when_refs_provided(tmp_path): ref1 = tmp_path / "ref1.png"; ref1.write_bytes(b"PNG1" * 300) ref2 = tmp_path / "ref2.png"; ref2.write_bytes(b"PNG2" * 300) client = MagicMock() client.images.edit.return_value.data = [MagicMock(b64_json=base64.b64encode(b"x"*2000).decode())] out = generate_floor_plan_image("test prompt", openai_client=client, ref_paths=[ref1, ref2]) assert client.images.edit.called call = client.images.edit.call_args assert isinstance(call.kwargs["image"], list) assert len(call.kwargs["image"]) == 2 ``` - [ ] **Step 4.4: 구현 — generate_floor_plan_image 시그니처 변경** ```python import contextlib def generate_floor_plan_image( prompt: str, openai_client, *, ref_paths: Optional[List[Path]] = None, model: str = "gpt-image-2", size: str = "1024x1024", quality: str = "high", max_retries: int = 3, ) -> bytes: valid_refs = [p for p in (ref_paths or []) if p.exists() and p.stat().st_size >= 1024] last_err = None for attempt in range(max_retries): try: if not valid_refs: resp = openai_client.images.generate(model=model, prompt=prompt, size=size, quality=quality) else: with contextlib.ExitStack() as stack: files = [stack.enter_context(p.open("rb")) for p in valid_refs] resp = openai_client.images.edit( model=model, image=files, prompt=prompt, size=size, quality=quality, ) data = resp.data[0] png_bytes = base64.b64decode(data.b64_json) if hasattr(data, "b64_json") and data.b64_json else _fetch_url(data.url) if len(png_bytes) < 1024: raise ImageGenerationError("empty/truncated PNG") return png_bytes except Exception as e: last_err = e time.sleep(2 * (attempt + 1)) raise ImageGenerationError(f"floor plan image gen failed: {last_err}") ``` - [ ] **Step 4.5: 회귀 — ref_paths=None text-only 동작** ```python def test_generate_floor_plan_image_text_only_when_no_refs(tmp_path): client = MagicMock() client.images.generate.return_value.data = [MagicMock(b64_json=base64.b64encode(b"x"*2000).decode())] out = generate_floor_plan_image("p", openai_client=client, ref_paths=None) assert client.images.generate.called assert not client.images.edit.called ``` - [ ] **Step 4.6: 회귀 + 신규 5 케이스 통과** ```bash pytest tests/core/test_location_floor_plan.py -v ``` 기대: 27 (Phase 3) + 5 (Phase 5) = 32 PASS. - [ ] **Step 4.7: Commit** ```bash git add prompts/_base/location_floor_plan/2.* backend/app/modules/pipeline/location_floor_plan.py backend/tests/core/test_location_floor_plan.py git commit -m "$(cat <<'EOF' feat(p5): floor_plan v2 prompt + generate_floor_plan_image ref_paths v2 system + user_template은 multi-turn 컨텍스트(이전 도면 요약 + building group ref) 처리 지시 추가. generate_floor_plan_image는 ref_paths 인자로 multi-image edit (Phase 4 ExitStack 패턴). text-only 회귀 보장. Co-Authored-By: Claude Opus 4.7 (1M context) EOF )" ``` --- ## Task 5: LocationFloorPlanStep Sequential Redesign **Files:** - Modify: `backend/app/core/steps/location_floor_plan_step.py` - Modify: `backend/tests/core/test_location_floor_plan.py` > **위험도**: 가장 높음. ThreadPoolExecutor → sequential loop 전환. 기존 27 tests 회귀 필수. 같은 building_group의 prev PNG 첨부 로직 신규. - [ ] **Step 5.1: 실패 테스트 — sequential 처리 확인** ```python def test_location_floor_plan_processes_in_planner_order(tmp_path, monkeypatch): """planner.floor_plan_order 순서대로 처리, prev 도면 컨텍스트 inject 확인""" # planner cp: floor_plans=[FP1, FP2, FP3], order=[FP3, FP1, FP2] # FP1, FP3은 같은 building_group="g1", FP2는 "g2" # 처리 순서가 FP3 → FP1 → FP2여야 함 # FP1 처리 시 FP3 prompt summary가 user_prompt에 있어야 함 # FP1 처리 시 FP3 PNG가 ref_paths에 있어야 함 # FP2 처리 시 FP3, FP1 prompt summary는 있지만 ref_paths는 다른 group이므로 없음 ... ``` - [ ] **Step 5.2: 새 _execute 구현 (의사코드)** design.md §5의 알고리즘. ThreadPoolExecutor 제거. 단일 thread loop: ```python def _execute(self, mode="resume") -> Dict[str, Any]: if settings.background_mode != "floor_plan_anchored": return {"applicable_count": 0, ...} planner_cp = self._load_prev_checkpoint("background_planner") if not planner_cp or not planner_cp.get("data", {}).get("floor_plans"): return {"applicable_count": 0, "completed_count": 0, "failed_count": 0, "data": {"floor_plans": {}}} fp_specs = {fp["id"]: fp for fp in planner_cp["data"]["floor_plans"]} fp_order = planner_cp["data"]["floor_plan_order"] location_canon_by_short, location_label_by_short = self._load_location_canons() rules_text = self._load_rules_text() image_dir = self._image_dir() image_dir.mkdir(parents=True, exist_ok=True) rendered_summaries: Dict[str, str] = {} rendered_paths: Dict[str, Path] = {} results: Dict[str, Dict] = {} system_prompt = load_prompt("location_floor_plan", "system") user_template = load_prompt("location_floor_plan", "user_template") for fp_id in fp_order: # SEQUENTIAL — 한 번에 한 도면 spec = fp_specs[fp_id] prev_summaries = [(oid, rendered_summaries[oid]) for oid in rendered_summaries if oid != fp_id] same_group_refs = [ rendered_paths[oid] for oid, ospec in fp_specs.items() if oid in rendered_paths and ospec["building_group"] == spec["building_group"] ] result = self._process_floor_plan( fp_id=fp_id, spec=spec, location_canon_by_short=location_canon_by_short, location_label_by_short=location_label_by_short, prev_summaries=prev_summaries, same_group_ref_paths=same_group_refs, system_prompt=system_prompt, user_template=user_template, rules_text=rules_text, image_dir=image_dir, ) results[fp_id] = result if result.get("status") == "ok": rendered_summaries[fp_id] = _summarize_prompt(result["prompt_text"]) rendered_paths[fp_id] = Path(result["png_path"]) self._register_image_assets(results, fp_specs, location_canon_by_short) return { "applicable_count": len(fp_order), "completed_count": sum(1 for r in results.values() if r.get("status") == "ok"), "failed_count": sum(1 for r in results.values() if r.get("status") == "failed"), "data": {"floor_plans": results, "config_hash": self._config_hash()}, } ``` - [ ] **Step 5.3: _process_floor_plan 헬퍼** 기존 `_process_location` 로직 + multi-turn 컨텍스트 inject + ref_paths 전달. - [ ] **Step 5.4: _register_image_assets 수정 (C안 — variant 패턴)** > **NOTE (T1.5에서 발견)**: 기존 `image_asset` 테이블에는 `short_id` 컬럼이 없고 unique constraint도 없습니다. natural key는 `entity_id`(FK to `entity_canon.id`) + 추가 변별자. Phase 5는 `(project_id, asset_type, entity_id, variant_index)` 조합을 application-level UPSERT 키로 사용 (DB constraint 추가 없음). 각 floor_plan을 `(asset_type="floor_plan", entity_id=primary_location_canon_id, variant_index=0, variant_label="v00")`로 UPSERT: ```python def _register_image_assets( self, results: Dict[str, Dict], # fp_id → result fp_specs: Dict[str, Dict], # fp_id → planner spec location_canon_by_short: Dict, ) -> None: for fp_id, result in results.items(): if result.get("status") != "ok": continue spec = fp_specs[fp_id] primary_loc = spec["primary_location_id"] canon = location_canon_by_short.get(primary_loc) if not canon: continue upsert_image_asset( db=self.db, project_id=self.project_id, asset_type="floor_plan", entity_id=canon.id, # FK to entity_canon (location) variant_index=0, variant_label="v00", t2i_guide=None, # Phase 6에서 주입 file_path=str(result["png_path"]), prompt_used=result["prompt_text"], is_primary=1, metadata_inline={ "fp_id": fp_id, "building_group": spec["building_group"], "location_ids": spec["location_ids"], }, ) ``` > **building_group 처리**: primary_location만 row 가짐. 다른 location들(L11)은 자체 floor_plan row 없음. chain_bg_render 시 `planner.floor_plans[].location_ids` reverse lookup으로 그룹 PNG를 ref로 사용 (T7 참조). - [ ] **Step 5.5: 회귀 + 신규 8 tests 통과** ```bash pytest tests/core/test_location_floor_plan.py -v ``` 기대: 35+ tests PASS. - [ ] **Step 5.6: Commit** ```bash git add backend/app/core/steps/location_floor_plan_step.py backend/tests/core/test_location_floor_plan.py git commit -m "$(cat <<'EOF' feat(p5): LocationFloorPlanStep sequential redesign ThreadPoolExecutor 제거 → planner.floor_plan_order 순차 처리. multi-turn 컨텍스트(prev 도면 요약 inject) + 같은 building_group prev PNG ref 첨부. mode=off/chain_only 회귀 보장. Co-Authored-By: Claude Opus 4.7 (1M context) EOF )" ``` --- ## Task 6: chain_bg_planning Group-based Redesign **Files:** - Create: `prompts/_base/background_chain_planning/N+1.{ts}/system.md` + `user_template.md` (parent context 추가) - Modify: `backend/app/core/steps/background_chain_planning_step.py` - Modify: `backend/app/modules/pipeline/background_chain_planning.py` - Modify: `backend/tests/core/test_background_chain_planning.py` - [ ] **Step 6.1: planner cp loader 추가 (BackgroundChainPlanningStep)** ```python def _load_planner_groups(self) -> Optional[Dict]: cp = self._load_prev_checkpoint("background_planner") if not cp or not cp.get("data", {}).get("chain_bg_groups"): return None return { "groups": {g["id"]: g for g in cp["data"]["chain_bg_groups"]}, "order": cp["data"].get("chain_bg_order", []), } ``` - [ ] **Step 6.2: planning.py 분기 추가** ```python def run_background_chain_planning(*, planner_groups: Optional[Dict] = None, ...): if planner_groups is not None: return _run_planner_driven(planner_groups, ...) # 신규 Phase 5 path return _run_legacy_per_location(...) # Phase 4 path 보존 ``` - [ ] **Step 6.3: _run_planner_driven 구현** design.md §6 알고리즘. group_id를 chain_bg_order대로 순차 처리, parent prompt 컨텍스트 inject. floor_plan_prompts는 기존 best-effort 그대로. - [ ] **Step 6.4: prompt 신규 버전 (parent_chain_bg_block placeholder)** 기존 user_template + `{parent_chain_bg_block}` 추가. 빈 문자열 시 자동 생략. - [ ] **Step 6.5: 회귀 + 신규 8 tests** 기존 24 tests + 신규 8 = 32 PASS: - planner cp 없음 → legacy fallback (회귀) - planner cp 있음 → group-based 동작 - parent 의존성 inject - 빈 chain_bg_groups → graceful no-op - [ ] **Step 6.6: Commit** ```bash git add prompts/_base/background_chain_planning/ backend/app/core/steps/background_chain_planning_step.py backend/app/modules/pipeline/background_chain_planning.py backend/tests/core/test_background_chain_planning.py git commit -m "$(cat <<'EOF' feat(p5): chain_bg_planning planner-driven group dispatch + legacy fallback planner.chain_bg_groups 있으면 group_id별 sequential 처리(parent prompt 컨텍스트 inject). 없으면 legacy _phase0_group_by_location fallback (mode=chain_only 회귀). Co-Authored-By: Claude Opus 4.7 (1M context) EOF )" ``` --- ## Task 7: chain_bg_render chain_bg_order Driver **Files:** - Modify: `backend/app/core/steps/background_chain_render_step.py` - Modify: `backend/app/modules/pipeline/background_chain_render.py` - Modify: `backend/tests/core/test_phase4_floor_plan_chain.py` - [ ] **Step 7.1: planner cp loader (Phase 4 패턴 그대로)** ```python def _load_planner_chain_order(self) -> Optional[List[str]]: cp = self._load_prev_checkpoint("background_planner") if not cp: return None return cp.get("data", {}).get("chain_bg_order") or [] ``` - [ ] **Step 7.2: render.py 분기** ```python def run_background_chain_render(*, planner_chain_order: Optional[List[str]] = None, ...): if planner_chain_order is not None: return _run_planner_driven_render(planner_chain_order, ...) return _run_legacy_per_location_render(...) ``` - [ ] **Step 7.3: _run_planner_driven_render 구현 + variant counter** ```python def _run_planner_driven_render( planner_chain_order: List[str], planning_groups: Dict[str, Dict], # group_id → planning result (chain_bg_prompt + shot_guides) floor_plan_specs: Dict[str, Dict], # fp_id → planner spec (location_ids[], building_group) floor_plan_paths: Dict[str, Path], # primary_loc_short_id → PNG (T5에서 저장한 v00) db, project_id: str, image_dir: Path, ... ): rendered_paths: Dict[str, Path] = {} # group_id → PNG location_variant_counter: Dict[str, int] = {} # location_short_id → next variant_index (1+) results: Dict[str, Dict] = {} for group_id in planner_chain_order: group = planning_groups.get(group_id) if not group: continue loc_id = group["location_id"] # variant_label 결정 (chain_bg는 항상 v01 부터) next_idx = location_variant_counter.get(loc_id, 1) # v01 시작 variant_label = f"v{next_idx:02d}" location_variant_counter[loc_id] = next_idx + 1 # ref 우선순위 (Phase 4 그대로 + building_group reverse lookup) ref_paths = [] # 1순위: 같은 building_group의 floor_plan PNG primary_for_group = _find_primary_loc_for_chain_bg(loc_id, floor_plan_specs) if primary_for_group and primary_for_group in floor_plan_paths: ref_paths.append(floor_plan_paths[primary_for_group]) # 2순위: parent chain_bg OR location ref if group.get("parent_id") and group["parent_id"] in rendered_paths: ref_paths.append(rendered_paths[group["parent_id"]]) elif not ref_paths: # floor_plan도 parent도 없을 때만 location ref location_ref = _resolve_location_ref(loc_id, ...) if location_ref: ref_paths.append(location_ref) # 렌더 + 저장 + ImageAsset UPSERT (with t2i_guide) png_bytes = render_node_image(group["chain_bg_prompt"], ref_paths=ref_paths, ...) png_path = image_dir / f"{loc_id}_{variant_label}.png" png_path.write_bytes(png_bytes) rendered_paths[group_id] = png_path upsert_image_asset( db=db, project_id=project_id, asset_type="chain_bg", entity_id=location_canon_id_for(loc_id), # FK to entity_canon (location) variant_index=next_idx, variant_label=variant_label, t2i_guide="\n".join(group.get("shot_guides", [])), # Phase 2의 shot_guides[] 영속 file_path=str(png_path), prompt_used=group["chain_bg_prompt"], is_primary=0, # floor_plan(v00)만 primary metadata_inline={ "group_id": group_id, "parent_id": group.get("parent_id", ""), "scenes": group.get("scenes", []), "shot_ids": group.get("shot_ids", []), "time": group.get("time", ""), }, ) results[group_id] = { "status": "ok", "png_path": str(png_path), "variant_label": variant_label, "ref_used": _classify_ref_used(ref_paths, primary_for_group, group.get("parent_id")), } return {"applicable_count": len(planner_chain_order), "completed_count": len(results), ..., "data": {"groups": results}} def _find_primary_loc_for_chain_bg(loc_id: str, fp_specs: Dict[str, Dict]) -> Optional[str]: """building_group reverse lookup. loc_id가 어떤 floor_plan의 location_ids에 포함되면 그 primary 반환.""" for fp in fp_specs.values(): if loc_id in fp.get("location_ids", []): return fp["primary_location_id"] return None ``` - [ ] **Step 7.4: 회귀 + 신규 7 tests** 기존 17 + 16 (Phase 4) + 7 신규 = 40 PASS: - mode=off → not_applicable - mode=chain_only → legacy path - mode=floor_plan_anchored + planner cp → planner-driven - chain_bg_order 빈 list → graceful - parent_id 누락 → root로 처리 - **NEW**: variant_label 순차 증가 검증 (L05 v01, L05 v02, L11 v01) - **NEW**: building_group reverse lookup (L11의 chain_bg가 L05_v00 PNG ref 사용) - **NEW**: t2i_guide ImageAsset 영속 검증 (shot_guides[] → t2i_guide 컬럼) - [ ] **Step 7.5: Commit** ```bash git commit -m "$(cat <<'EOF' feat(p5): chain_bg_render planner-driven + variant counter + t2i_guide UPSERT planner.chain_bg_order 순차 처리. ImageAsset에 variant_label(v01+)과 t2i_guide 영속 (Phase 2 shot_guides 통합). building_group reverse lookup으로 L05_v00 floor_plan PNG를 L11 chain_bg ref로 사용. legacy 회귀 보장. Co-Authored-By: Claude Opus 4.7 (1M context) EOF )" ``` --- ## Task 8: Manifest depends_on + Integration Tests **Files:** - Modify: `backend/app/core/step_manifest.py` - Create: `backend/tests/core/test_phase5_sequential_pipeline.py` (E2E 통합) - [ ] **Step 8.1: location_floor_plan.depends_on += "background_planner"** `step_manifest.py`: ```python "location_floor_plan": { ..., "depends_on": [..., "background_planner"], # 기존 6개 + 1 } ``` `background_chain_planning`은 manifest 의존 X (best-effort 패턴 — Phase 4 I1 동일). - [ ] **Step 8.2: 통합 테스트 작성** ```python # tests/core/test_phase5_sequential_pipeline.py def test_phase5_full_pipeline_floor_plan_anchored(tmp_path, monkeypatch): """mode=floor_plan_anchored 전체 4-step 흐름 mock E2E""" # 1. fixtures: 6 location, 30 selected shot # 2. mode=floor_plan_anchored # 3. background_planner 실행 → floor_plans=2, chain_bg_groups=4 (mock) # 4. location_floor_plan 실행 → 2 PNG 생성 확인 + 같은 group ref attach 확인 # 5. background_chain_planning 실행 → 4 group prompt 생성 확인 # 6. background_chain_render 실행 → 4 PNG, ref priority 확인 ... def test_phase5_mode_off_skips_all(...): """mode=off → 4 step 모두 not_applicable""" def test_phase5_mode_chain_only_legacy(...): """mode=chain_only → planner+floor_plan skip, chain_bg legacy""" def test_phase5_planner_empty_floor_plans(...): """planner.floor_plans=[] → location_floor_plan no-op""" def test_phase5_planner_partial_failure(...): """floor_plan 1개 실패 → chain_bg_render는 그 group skip + 나머지 진행""" ``` - [ ] **Step 8.3: pytest 통합 + 전체 회귀** ```bash pytest tests/core/test_phase5_sequential_pipeline.py -v pytest tests/core/ -v --tb=short ``` 기대: Phase 5 신규 5 + 기존 ~290 = ~295 PASS. - [ ] **Step 8.4: Commit** ```bash git commit -m "$(cat <<'EOF' feat(p5): manifest depends_on 재조정 + integration tests location_floor_plan.depends_on += background_planner. chain_bg_planning은 on_demand best-effort (Phase 4 I1 패턴 유지). 5 통합 시나리오 테스트. Co-Authored-By: Claude Opus 4.7 (1M context) EOF )" ``` --- ## Task 9: 회귀 + 듀얼 리뷰 + Memory + Push - [ ] **Step 9.1: 전체 pytest 회귀** ```bash cd backend && pytest tests/ -v --tb=short -q ``` 기대: 모든 기존 테스트 PASS + Phase 5 신규 ~50 PASS. 회귀 0건. - [ ] **Step 9.2: 정적 검사** ```bash cd backend && python3 -c "from app.core.step_manifest import STEP_MANIFEST; print(len(STEP_MANIFEST))" cd backend && python3 -c "from app.core.steps import STEP_CLASSES; print('background_planner' in STEP_CLASSES)" ``` - [ ] **Step 9.3: Codex 리뷰** `codex:codex-rescue` agent에 작업 요약 + diff 위치 전달: ``` Phase 5 background planner pipeline 구현. 9 commits. 듀얼 리뷰의 Codex 측. 초점: (a) HARD 제약(별도 LLM call) 위반 없음 확인 (b) 회귀 보장(mode=off/chain_only) (c) sequential 멀티턴 컨텍스트 inject 정확성 (d) ExitStack file handle 안전성 (e) frequency rule validate 완전성. Critical/Important/Minor로 분류. ``` - [ ] **Step 9.4: Claude code-reviewer 리뷰** `feature-dev:code-reviewer` agent. 같은 input. 구조적 코드 quality 초점. - [ ] **Step 9.5: 듀얼 리뷰 fix 통합** Critical/Important 모두 fix. 단일 commit으로 묶거나 task별 분리. Phase 4 패턴 (`8f6d9f1`) 참고. - [ ] **Step 9.6: Memory 저장** `/Users/manta/.claude/projects/-Users-manta-Documents-Projects-TheRoad-I1/memory/session_2026{date}_phase5.md` 작성: - commit list (9 commit + review fix) - 핵심 설계 결정 - frequency rule + building group 로직 - HARD 제약 충족 방식 - 듀얼 리뷰 결과 (Critical/Important fix) - 다음 세션 시작점 (production 검증, 비교 페이지) `MEMORY.md` index에 1줄 추가. - [ ] **Step 9.7: Push** ```bash git log origin/main..HEAD --oneline git push origin main ``` - [ ] **Step 9.8: 사용자에게 다음 작업 제안** production 검증 (PID c00bbe19, mode=floor_plan_anchored 재실행, 비교 페이지 갱신). --- ## Self-Review Checklist (writing-plans skill 요구) ### 1. Spec coverage - [x] Frequency rule (1/2/3+ shot, indoor/outdoor) → T1 prompt + T2 validate - [x] Building group rule → T1 schema + T2 validate - [x] LLM-driven order → T2 schema + T3 step - [x] Sequential multi-turn (텍스트 inject + 이미지 ref) → T4 prompt v2 + T5 step - [x] mode 매트릭스 (off/chain_only/floor_plan_anchored) → 모든 step에 분기 - [x] HARD 제약 (별도 LLM call) → T5/T6/T7 모두 sequential loop - [x] 회귀 보장 → T6/T7 legacy fallback + T8 통합 테스트 - [x] 듀얼 리뷰 → T9 ### 2. Placeholder scan - [x] "TBD" 0건 - [x] "TODO" 0건 (코드 내 의도된 미해결 외) - [x] 모든 step의 코드 스니펫 제공 - [x] 모든 commit 메시지 본문 제공 ### 3. Type consistency - [x] `floor_plan_id` 일관: planner.floor_plans[].id == FP_L05 형식, location_floor_plan.results 키, chain_bg_groups[].floor_plan_id 동일 - [x] `chain_bg_id` 일관: planner.chain_bg_groups[].id == CB_L05_living_day 형식, planning groups dict 키, render rendered_paths dict 키 - [x] `shot_id` 일관: planner.chain_bg_groups[].shot_ids[] == S{scene}_Shot{idx} - [x] `generate_floor_plan_image` 시그니처: T4 정의 → T5 사용 동일 ### 4. 의문점 (사용자 확정 완료) - ~~T5.4 ImageAsset short_id 매핑~~ → **C안 채택**: outlook 패턴(variant_index/label/t2i_guide). T1.5에서 alembic migration. T5.4 floor_plan=v00, T7.3 chain_bg=v01+. building_group secondary location은 자체 row 없이 planner.location_ids reverse lookup. - scene_detail이 t2i_guide를 lookup하여 inject하는 부분은 **Phase 6로 이월** (Phase 5 scope = ImageAsset 영속화까지). --- ## 실행 모드 선택 **플랜 작성 완료 — 두 가지 실행 옵션:** 1. **Subagent-Driven (권장)**: task별 fresh subagent 디스패치, two-stage review (spec compliance + code quality), 빠른 iteration. **모델 task별 차등 (haiku/sonnet/opus)**. 2. **Inline 실행**: 같은 세션에서 batch 실행, checkpoint마다 사용자 review.