"""background_chain_planning StepRunner — set_design 교체용 신규 step (PR #3, 2026-04-27). on_demand: 자동 run-all에서 실행 X. 사용자가 명시적으로 step을 트리거할 때만 동작. Note: shot_validator 체크포인트는 shot_extract와 동일한 `data.scenes[].shots[]` 스키마를 보존한다 (shot_validator_step.py:7-8 contract). 따라서 본 step에서 `shot_extract_data` 인자에 shot_validator 결과를 그대로 전달해도 안전하다. """ import json import logging from pathlib import Path from typing import Any, Dict, Optional from app.core.step_runner import StepRunner logger = logging.getLogger(__name__) # Phase 5.1: chain_planning v3 → v4 prompt (FLOOR PLAN 블록 있으면 outdoor skip 금지) + # floor_plan_prompts dict가 building_group 모든 멤버 location_id 커버. PROMPT_VERSION bump # 으로 stale checkpoint 자동 invalidation. SCHEMA_VERSION = 2 # C4 v1 — v5 prompt: SKIP DECISION indoor/outdoor 닫힌 예시 enumeration + node state # 닫힌 목록 + skip_chain description OUTDOOR 목록을 generic semantic principle 로 추상화 # (Prompt Closed-List Ban). config_hash invalidation. SCHEMA_VERSION 2 유지. PROMPT_VERSION = "5" class BackgroundChainPlanningStep(StepRunner): """Step: location별 배경 이미지 chain 트리 설계.""" def _config_hash(self) -> str: """SCHEMA_VERSION + PROMPT_VERSION 포함한 config_hash. resume 시 stale plan 회피.""" import hashlib, json as _json from app.core.config import settings payload = { "background_mode": settings.background_mode, "model": settings.openai_model, "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(): return json.loads(cp.read_text(encoding="utf-8")) return None def _load_planner_groups(self) -> Optional[Dict[str, Any]]: """background_planner 체크포인트에서 chain_bg_groups + chain_bg_order 로드. Phase 5: planner cp가 있고 chain_bg_groups가 비어있지 않으면 planner-driven path로 진입한다. 없으면 None → legacy `_phase0_group_by_location` fallback (Phase 4 회귀 보장). Returns: None — planner cp가 없거나 chain_bg_groups가 비어있을 때 {"groups": {gid: spec}, "order": [gid, ...]} — planner-driven 입력 """ cp = self._load_prev_checkpoint("background_planner") if not cp: return None data = cp.get("data") or {} chain_bg_groups = data.get("chain_bg_groups") or [] chain_bg_order = data.get("chain_bg_order") or [] if not chain_bg_groups: return None # group_id → spec 인덱스 (order 무관) groups_by_id: Dict[str, Any] = {} for g in chain_bg_groups: gid = g.get("id", "") if gid: groups_by_id[gid] = g return {"groups": groups_by_id, "order": list(chain_bg_order)} def _load_floor_plan_prompts(self) -> Dict[str, str]: """location_floor_plan checkpoint에서 status='ok' + non-empty prompt_text 매핑. Phase 4: chain_bg_planning user_prompt에 도면 prompt를 prepend하여 LLM이 floor plan 인식한 chain bg prompt 생성. mode=off / chain_only 시 빈 dict. Phase 5: building_group의 모든 멤버 location_id에 동일 prompt를 매핑. (fp_rooftop_unit이 L05+L04+L10+L11을 커버하면 4 키 모두 같은 텍스트.) 외부/서브룸 location의 chain_bg_planning에도 [FLOOR PLAN] 블록이 prepend되어 outdoor skip 룰 회피. data.floor_plans (신 shape) 우선, 없으면 data.locations (legacy compat alias)로 fallback. ⚠️ 순서 의존성: chain_bg_planning(order 19.6)이 location_floor_plan(order 21.5) 보다 작은 order이지만, 둘 다 applicability=on_demand이므로 사용자가 명시 트리거 시점에는 location_floor_plan 체크포인트가 이미 존재한다 (run-all → 사용자 트리거 순서). chain_bg_planning의 applicability를 always로 바꾸면 이 prepend가 silently 깨질 수 있다. """ cp = self._load_prev_checkpoint("location_floor_plan") if not cp: return {} data = cp.get("data", {}) or {} out: Dict[str, str] = {} # Phase 5: data.floor_plans dict {fp_id: {location_ids[], prompt_text, status}} floor_plans = data.get("floor_plans") or {} if isinstance(floor_plans, dict): for fp in floor_plans.values(): if not isinstance(fp, dict): continue if fp.get("status") != "ok": continue text = (fp.get("prompt_text") or "").strip() if not text: continue for lid in fp.get("location_ids") or []: if isinstance(lid, str) and lid: out[lid] = text # legacy fallback: data.locations[] {id, prompt_text, status} — Phase 4 compat alias if not out: for loc in data.get("locations", []) or []: if loc.get("status") != "ok": continue text = (loc.get("prompt_text") or "").strip() if not text: continue out[loc["id"]] = text return out def _execute(self, mode: str = "resume") -> Dict[str, Any]: from app.core.errors import AppError # 의존 체크포인트 로드 shot_cp = self._load_prev_checkpoint("shot_validator") sel_cp = self._load_prev_checkpoint("shot_selection") staging_cp = self._load_prev_checkpoint("shot_staging") director_cp = self._load_prev_checkpoint("scene_director") merge_cp = self._load_prev_checkpoint("entity_merge") detail_cp = self._load_prev_checkpoint("entity_detail") rules_cp = self._load_prev_checkpoint("visual_world_rules") # 핵심 의존 4개는 누락 시 silent no-op 방지를 위해 명시 raise. # shot_staging / entity_detail / visual_world_rules는 컨텍스트 보강용이라 # 누락되어도 빈 dict로 fallback (LLM 컨텍스트만 빈약해짐). required = ( ("shot_validator", shot_cp), ("shot_selection", sel_cp), ("scene_director", director_cp), ("entity_merge", merge_cp), ) for name, cp in required: if not cp or not cp.get("data"): raise AppError( code="step.no_input", message=( f"{name} 체크포인트 없음 — background_chain_planning은 " f"shot_validator/shot_selection/scene_director/entity_merge 4개를 " f"필수로 요구합니다. 분석 단계를 먼저 실행하세요." ), status_code=400, ) from app.modules.pipeline.background_chain_planning import ( run_background_chain_planning, ) # Phase 4: floor plan prompt 로드 (mode=off 시 빈 dict — 회귀 보장) floor_plan_prompts = self._load_floor_plan_prompts() # Phase 5: planner cp 있으면 planner-driven path. 없으면 None → legacy fallback. planner_groups = self._load_planner_groups() result = run_background_chain_planning( shot_extract_data=shot_cp.get("data", {}), shot_selection_data=sel_cp.get("data", {}) if sel_cp else {}, shot_staging_data=staging_cp.get("data", {}) if staging_cp else {}, director_data=director_cp.get("data", {}) if director_cp else {}, entity_merge_data=merge_cp.get("data", {}) if merge_cp else {}, entity_detail_data=detail_cp.get("data", {}) if detail_cp else None, world_rules_data=rules_cp.get("data", {}) if rules_cp else None, opik_metadata=self.build_opik_metadata(), floor_plan_prompts=floor_plan_prompts, planner_groups=planner_groups, ) # Phase 5 planner-driven path: data.groups가 primary, data.locations는 compat alias. # Phase 4 legacy path: data.locations만 존재. groups = result.get("groups") if groups is not None: applicable = len(groups) ok_count = sum( 1 for v in groups.values() if v.get("status") in ("ok", "skipped") ) else: locations = result.get("locations", {}) applicable = len(locations) ok_count = sum( 1 for v in locations.values() if v.get("status") in ("ok", "skipped") ) failed = result.pop("_failed_count", 0) return { "completed_count": ok_count, "applicable_count": applicable, "failed_count": failed, "schema_version": SCHEMA_VERSION, "config_hash": self._config_hash(), "data": result, }