"""BackgroundMasterPlanStep — Phase 7 Step 2. 각 chain_bg group에 대해 LLM 1회 호출 → master plan 산출. 그룹간 ThreadPool 병렬. prev_shot_ref 그룹은 skip. 흐름: 1. background_classify + scene_save + shot_validator + shot_selection + visual_world_rules 체크포인트 로드 2. classify의 chain_bg group만 추려 그룹별 LLM 호출 (ThreadPool 병렬) 3. prev_shot_ref 그룹은 plan=null로 기록 (호출 X) 체크포인트 data: data.plans = {group_id: {status, plan | error}} """ 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 from app.modules.pipeline.background_master_plan import ( run_background_master_plan, ) logger = logging.getLogger(__name__) # D6 T4 (B3 + B6 + B7 + I8) + T4-fix3 — schema_version 3: # - LLM raw intent (no bg_id), code assigns deterministic L##B## via bg_catalog. # - Per-group `plan.{floor_plans, backgrounds, gen_order}` 보존 (B3 — downstream # reader 영향 0). plan.backgrounds[].bg_id slot 도 새 ID 로 갱신 (handshake). # - sibling field 추가: data.background_catalog / data.shot_background_map / # data.bg_catalog_hash / data.shot_binding_hash. # - T4-fix3 (v3 prompt pack): floor_plans[] 에 loc_id + space_key_hint required. # raw validator 가 fp ↔ bg link cross-check (FpLinkMismatchError raise). # checkpoint contract 변경 — SCHEMA bump 의무. # - W21B-wave-2 (v5): backgrounds[].surface_role raw intent 추가. exterior / # transition / site plate 는 fp-less raw intent 허용. bg_catalog_hash 가 # surface_role 을 render-relevant field 로 반영. SCHEMA_VERSION = 5 PROMPT_VERSION = "6.202605290248" # W21B-wave-2: background surface_role enum + exterior/transition/site fp-less raw intent. def chain_groups_of(groups) -> List[Dict[str, Any]]: """`background_classify` 의 building_groups → **chain_bg 그룹**(이 스텝의 구매 단위). ★술어는 여기 한 곳 — 스텝과 canary 의 사전 문(`assert_chain_group_cap_covers`)이 같은 함수를 부른다.""" return [g for g in (groups or []) if isinstance(g, dict) and g.get("kind") == "chain_bg"] 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": 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(): 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"}: logger.info( "background_master_plan: 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": {}, } 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") scene_director_cp = self._load_prev_checkpoint("scene_director") if shot_validator_cp: from app.core.steps.shot_validator_step import assert_no_failed_scenes assert_no_failed_scenes(shot_validator_cp, self.project_config, consumer_step="background_master_plan") # scene → primary_location fallback (shot에 location_id 없으면, Phase 5 패턴) scene_primary: Dict[int, str] = {} if scene_director_cp: for sc in (scene_director_cp.get("data", {}) or {}).get("scenes", []) or []: si = sc.get("scene_index") primary = sc.get("primary_location", "") or "" if si is not None and primary: scene_primary[int(si)] = primary groups = ((classify_cp or {}).get("data", {}) or {}).get("building_groups", []) or [] if not groups: logger.info("background_master_plan: no groups from classify — skip") return { "applicable_count": 0, "completed_count": 0, "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 [] from app.modules.pipeline.background_master_plan import ( build_master_plan_user_prompt, MasterPlanError, ) from app.modules.llm.llm_client import call_structured chain_groups = chain_groups_of(groups) plans: Dict[str, Any] = {} failed = 0 opik_meta = self.build_opik_metadata() project_config = self.project_config # T4-fix3: location_profiles build (raw validator + post_process 공용 SOT). # chain_groups 비어있으면 LLM 호출 자체가 없으므로 lazy build (DB query # 절감 + prev_shot_only test isolation 보존). location_profiles: Dict[str, Dict[str, Any]] = ( self._load_location_profiles() if chain_groups else {} ) def _process(g: Dict[str, Any]) -> Tuple[str, Dict[str, Any]]: gid = g["group_id"] # A-fix (W21B-w5): LLM 호출 전 group member 를 location_profiles 로 pre-filter. # entity_filter 가 저빈도 generic location (예: 0~1 씬 마당/도로) 을 제거하면 # 그 loc_id 의 location.space_profile 이 없다. 이런 member 가 그룹에 남아 있으면 # LLM 이 그 loc_id 의 floor_plan 을 생성 → validate_master_plan_raw_intent 의 # SemanticKeyError fail-fast 에 걸려 그룹 전체가 3-retry 소진 후 유실됐다 # (같은 그룹의 anchor 실내까지 cascade). pre-filter 로 profile 없는 member 만 # prompt / group_loc_ids / scene selection 에서 빼고 diagnostic 으로 남긴다. # validator fail-fast 는 그대로 유지 (정상 입력만 통과시켜 raise 를 사전 제거할 # 뿐, profile-less 를 fp-less plate 로 부활시키지 않는다 — entity_filter SOT 존중). raw_members = g.get("members", []) or [] kept_members: List[Dict[str, Any]] = [] filtered_members: List[Dict[str, Any]] = [] for m in raw_members: loc = m.get("loc_id", "") if loc and loc in location_profiles: kept_members.append(m) else: filtered_members.append({ "loc_id": loc, "label": m.get("label", ""), "shot_count": int(m.get("shot_count", 0) or 0), "reason": "no_location_profile", }) pre_filter = { "raw_member_count": len(raw_members), "kept_member_count": len(kept_members), "filtered_members": filtered_members, } if not kept_members: # 모든 member 가 profile 부재 — 그룹 전체 skip (failure 아님). logger.warning( "master_plan: group %s skipped — all members lack location_profile (%s)", gid, [fm["loc_id"] for fm in filtered_members], ) return gid, { "status": "all_filtered", "plan": None, "pre_filter": pre_filter, } if filtered_members: logger.info( "master_plan: group %s pre-filtered %d member(s) w/o location_profile: %s", gid, len(filtered_members), [fm["loc_id"] for fm in filtered_members], ) g_eff = {**g, "members": kept_members} if filtered_members else g group_loc_ids = [m["loc_id"] for m in kept_members] relevant_scenes, group_shot_ids = _select_scenes_for_group( g_eff, scene_segments, shot_validator_cp, shot_selection_cp, scene_primary=scene_primary, ) user_prompt = build_master_plan_user_prompt( group=g_eff, scenes=relevant_scenes, visual_world_rules=rules_text, location_profiles=location_profiles, ) 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, location_profiles=location_profiles, call_structured_fn=call_structured, project_config=project_config, opik_metadata=opik_meta, ) return gid, {"status": "ok", "plan": plan, "pre_filter": pre_filter} except MasterPlanError as exc: logger.error("master_plan: group %s failed: %s", gid, exc) return gid, { "status": "failed", "error": str(exc)[:200], "plan": None, "pre_filter": pre_filter, } all_filtered = 0 pre_filter_by_group: Dict[str, Any] = {} 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} # pre_filter diagnostic 은 plan entry 밖으로 빼서 diagnostics 로 집계 # (plan entry shape 은 downstream consumer 계약대로 유지). pf = res.pop("pre_filter", None) if pf and pf.get("filtered_members"): pre_filter_by_group[gid] = pf plans[gid] = res status = res.get("status") if status == "failed": failed += 1 elif status == "all_filtered": all_filtered += 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} # ───────────────────────────────────────────────────────────── # W21B-wave-2 — surface-role scope filter (replaces W20F6 hard indoor filter) # ───────────────────────────────────────────────────────────── # eligibility = background.surface_role aware. # - interior_room remains floor-plan-bound. # - exterior/transition/site plates may be fp-less. # cascade transitive: dropped BG 를 depends_on_bg 로 참조하는 BG 도 # drop (fail-safe). diagnostic 은 raw identity (bg_id 는 # _d6_post_process 후 부여). scope_diag = self._apply_w21b_surface_role_scope_filter( ordered_plans, groups, ) # ───────────────────────────────────────────────────────────── # D6 T4 post-processing — assign deterministic bg_id + sibling fields # ───────────────────────────────────────────────────────────── # B3: per-group plans 그대로 보존 + sibling 으로 background_catalog / # shot_background_map / hashes 추가. plan.backgrounds[].bg_id slot # 도 새 L##B## 형식으로 갱신 (handshake — downstream reader 호환). # B6: validator split (raw_intent 는 LLM 단계에서, assigned 는 본 step 에서). # B7: monotonic next_b — bg_catalog._max_b_for_loc. # I8: intent sort — assign_bg_ids 가 (loc_id, time_phase, state_class) 정렬. background_catalog, shot_background_map, bg_catalog_hash, shot_binding_hash = \ self._d6_post_process(ordered_plans) # gen_order 도 post-processing 에서 build (이전: LLM 출력. D6: 코드 산출). for gid, gentry in ordered_plans.items(): if gentry.get("status") != "ok": continue plan = gentry.get("plan") or {} plan["gen_order"] = self._build_gen_order(plan) # all_filtered 그룹 (profiled member 0) 은 실제 시도 대상이 아니므로 applicable 에서 # 제외 — failure 가 아니며 completed 로도 집계하지 않는다. attempted = max(1, len(chain_groups) - all_filtered) return { "applicable_count": attempted, "completed_count": max(0, attempted - failed), "failed_count": failed, "schema_version": SCHEMA_VERSION, "config_hash": self._config_hash(), "data": { "plans": ordered_plans, # D6 sibling fields — consumer (floor_plan_prompt / background_prompt / # background_render / scene_detail) 가 hash stamp 시 read. "background_catalog": background_catalog, "shot_background_map": shot_background_map, "bg_catalog_hash": bg_catalog_hash, "shot_binding_hash": shot_binding_hash, "diagnostics": { "w21b_surface_role_scope_filter": scope_diag, # A-fix: profile 없는 group member 를 LLM 호출 전 제외한 내역 # (gid → {raw_member_count, kept_member_count, filtered_members[]}). "location_profile_pre_filter": pre_filter_by_group, }, }, } def _load_location_profiles(self) -> Dict[str, Dict[str, Any]]: """entity_canon.metadata_json.location.space_profile 을 loc_id → profile 로 build. D6 SOT: location.space_profile 은 entity_extractor (T2) 가 생성. master_plan 후처리의 normalize_space_key 가 본 dict 를 단일 source 로 사용. """ from app.core.database import SessionLocal from app.models.project import EntityCanon profiles: Dict[str, Dict[str, Any]] = {} # NOTE: 본 step 이 자체 db 세션 보유 안 함 — short-lived SessionLocal. db = SessionLocal() try: rows = db.query(EntityCanon).filter( EntityCanon.project_id == self.project_id, EntityCanon.entity_type == "location", EntityCanon.short_id.isnot(None), ).all() for row in rows: try: md = json.loads(row.metadata_json or "{}") except (TypeError, ValueError): logger.warning( "background_master_plan: EntityCanon %s metadata_json parse fail — skip", row.short_id, ) continue loc_block = (md or {}).get("location") if not isinstance(loc_block, dict): continue profile = loc_block.get("space_profile") if isinstance(profile, dict): profiles[row.short_id] = profile finally: db.close() return profiles def _apply_w21b_surface_role_scope_filter( self, ordered_plans: Dict[str, Dict[str, Any]], groups: List[Dict[str, Any]], ) -> Dict[str, Any]: """W21B-wave-2 — surface-role scope filter + transitive BG cascade. This replaces the W20F6 hard filter ``indoor AND shot_count>=2 AND dep_BG_count>=1``. The old policy made exterior/transition/site surfaces disappear before rendering. The new policy is background-first: - ``interior_room`` backgrounds remain floor-plan-bound. - ``exterior_plate`` / ``transition_zone`` / ``site_surface`` backgrounds may be fp-less and are not dropped merely because the location is outdoor. - floor_plans are retained only when a kept background still references them. - dropped backgrounds cascade through exact raw ``depends_on_bg`` keys. bg_id 부재 (_d6_post_process 호출 전 trim) — diagnostic 은 raw identity (loc_id, space_key_hint, time_phase, state_class, applies_to_shots, reasons) carry. plan.floor_plans / plan.backgrounds 둘 다 in-place trim. group 의 backgrounds[] 가 0 으로 비게 되면 ``dropped_groups[]`` 에 reason ``all_filtered_after_trim`` carry (gen_order 단에서 빈 plan 은 자연 skip; defensive). Returns: diagnostics dict (manifest.data.diagnostics .w21b_surface_role_scope_filter 안으로 carry). """ from app.modules.pipeline.background_master_plan import SURFACE_ROLE_ENUM # ────────────────────────────────────────────────────────────── # 1. groups 별 member lookup (gid → loc_id → {is_indoor, shot_count}). # ────────────────────────────────────────────────────────────── member_by_gid_loc: Dict[str, Dict[str, Dict[str, Any]]] = {} for g in groups: gid = g.get("group_id", "") members = g.get("members", []) or [] member_by_gid_loc[gid] = { m.get("loc_id", ""): { "is_indoor": bool(m.get("is_indoor", False)), "shot_count": int(m.get("shot_count", 0) or 0), } for m in members if isinstance(m, dict) and m.get("loc_id") } # ────────────────────────────────────────────────────────────── # 2. raw background identity helper (bg_id 부재 단계). raw key 는 # 같은 plan 내 unique 하다는 가정 — assign_bg_ids 가 sem_key # 중복 시 fail-fast 라 LLM 단에서 unique 보장. raw cascade 의 # string 매칭은 양쪽 보수적 처리 (Codex Q2 의견). # ────────────────────────────────────────────────────────────── def _raw_bg_identity(bg: Dict[str, Any]) -> Tuple[str, str, str, str]: return ( str(bg.get("loc_id", "")), str(bg.get("space_key_hint", "")), str(bg.get("time_phase", "")), str(bg.get("state_class", "")), ) def _raw_bg_key_string(bg: Dict[str, Any]) -> str: return "|".join(_raw_bg_identity(bg)) def _bg_identity_dict(bg: Dict[str, Any], reasons: List[str]) -> Dict[str, Any]: return { "loc_id": bg.get("loc_id", ""), "space_key_hint": bg.get("space_key_hint", ""), "time_phase": bg.get("time_phase", ""), "state_class": bg.get("state_class", ""), "surface_role": bg.get("surface_role", ""), "applies_to_shots": list(bg.get("applies_to_shots") or []), "reasons": list(reasons), } dropped_floor_plans: List[Dict[str, Any]] = [] dropped_backgrounds: List[Dict[str, Any]] = [] dropped_groups: List[Dict[str, Any]] = [] # ────────────────────────────────────────────────────────────── # 3. per-group trim (raw eligibility first pass). # ────────────────────────────────────────────────────────────── for gid, gentry in ordered_plans.items(): if not isinstance(gentry, dict): continue if gentry.get("status") != "ok": continue plan = gentry.get("plan") or {} if not isinstance(plan, dict): continue floor_plans = list(plan.get("floor_plans") or []) backgrounds = list(plan.get("backgrounds") or []) loc_map = member_by_gid_loc.get(gid, {}) fp_ids = { fp.get("fp_id", "") for fp in floor_plans if isinstance(fp.get("fp_id", ""), str) and fp.get("fp_id", "") } # 3a. raw background eligibility. kept_backgrounds_pass1: List[Dict[str, Any]] = [] dropped_raw_keys_in_group: set = set() dropped_raw_key_strings_in_group: set = set() for bg in backgrounds: role = bg.get("surface_role", "") applies_to_shots = list(bg.get("applies_to_shots") or []) dep_fps = list(bg.get("depends_on_fp") or []) reasons: List[str] = [] if role not in SURFACE_ROLE_ENUM: reasons.append("invalid_surface_role") if not applies_to_shots: reasons.append("no_consuming_shot") if role == "interior_room" and not dep_fps: reasons.append("missing_floor_plan_dependency") missing_fp_refs = [ f for f in dep_fps if isinstance(f, str) and f and f not in fp_ids ] if missing_fp_refs: reasons.append("depends_on_missing_fp") if reasons: dropped_backgrounds.append(_bg_identity_dict( bg, reasons, )) dropped_raw_keys_in_group.add(_raw_bg_identity(bg)) dropped_raw_key_strings_in_group.add(_raw_bg_key_string(bg)) else: kept_backgrounds_pass1.append(bg) # 3b. transitive depends_on_bg cascade (fail-safe, fixed point). # raw depends_on_bg 는 bg_id 부재 단계에서 canonical raw key # string (loc|space|time|state) 을 참조할 수 있다. Meaning # 판단을 substring/keyword 에 맡기지 않고 exact raw key 만 본다. kept_backgrounds = list(kept_backgrounds_pass1) changed = True while changed: changed = False still_kept: List[Dict[str, Any]] = [] for bg in kept_backgrounds: dep_bgs = list(bg.get("depends_on_bg") or []) matched = False for ref in dep_bgs: if not isinstance(ref, str) or not ref: continue if ref in dropped_raw_key_strings_in_group: matched = True break if matched: dropped_backgrounds.append(_bg_identity_dict( bg, ["depends_on_dropped_bg"], )) dropped_raw_keys_in_group.add(_raw_bg_identity(bg)) dropped_raw_key_strings_in_group.add(_raw_bg_key_string(bg)) changed = True else: still_kept.append(bg) kept_backgrounds = still_kept # 3c. floor_plan retention: keep only plans that still feed a kept # background. fp-less exterior/transition/site plates do not force # fake floor_plan retention. referenced_fp_ids = { fp_ref for bg in kept_backgrounds for fp_ref in (bg.get("depends_on_fp") or []) if isinstance(fp_ref, str) and fp_ref } kept_floor_plans: List[Dict[str, Any]] = [] for fp in floor_plans: fp_id = fp.get("fp_id", "") loc_id = fp.get("loc_id", "") member = loc_map.get(loc_id) or {} if fp_id in referenced_fp_ids: kept_floor_plans.append(fp) continue dropped_floor_plans.append({ "group_id": gid, "fp_id": fp_id, "loc_id": loc_id, "is_indoor": bool(member.get("is_indoor", False)), "shot_count": int(member.get("shot_count", 0) or 0), "referenced_by_kept_background": False, "reasons": ["no_kept_background_ref"], }) # 3d. write back trimmed lists. group-empty defensive. plan["floor_plans"] = kept_floor_plans plan["backgrounds"] = kept_backgrounds if not kept_backgrounds: dropped_groups.append({ "group_id": gid, "reason": "all_filtered_after_trim", }) # plan = None 처리로 _d6_post_process 가 skip (현행 분기: # status != ok → skip). 보존 위해 status 만 변경. gentry["status"] = "all_filtered" gentry["plan"] = None # ────────────────────────────────────────────────────────────── # 4. acceptance hard assertion — orphan depends_on_fp 검증. # ────────────────────────────────────────────────────────────── for gid, gentry in ordered_plans.items(): if gentry.get("status") != "ok": continue plan = gentry.get("plan") or {} kept_fp_ids = {fp.get("fp_id", "") for fp in (plan.get("floor_plans") or [])} for bg in (plan.get("backgrounds") or []): for fp_ref in (bg.get("depends_on_fp") or []): if isinstance(fp_ref, str) and fp_ref and fp_ref not in kept_fp_ids: raise RuntimeError( f"W21B surface-role filter assertion: group={gid} background " f"loc_id={bg.get('loc_id')!r} space={bg.get('space_key_hint')!r} " f"still references dropped fp_id={fp_ref!r} after trim." ) diagnostics = { "policy": ( "surface_role-aware background-first filter; interior_room remains " "floor-plan-bound; exterior/transition/site may be fp-less; " "transitive cascade depends_on_bg exact raw key." ), "kept_floor_plans_count": sum( len((g.get("plan") or {}).get("floor_plans") or []) for g in ordered_plans.values() if isinstance(g, dict) and g.get("status") == "ok" ), "kept_backgrounds_count": sum( len((g.get("plan") or {}).get("backgrounds") or []) for g in ordered_plans.values() if isinstance(g, dict) and g.get("status") == "ok" ), "dropped_floor_plans_count": len(dropped_floor_plans), "dropped_backgrounds_count": len(dropped_backgrounds), "dropped_groups_count": len(dropped_groups), "dropped_floor_plans": dropped_floor_plans, "dropped_backgrounds": dropped_backgrounds, "dropped_groups": dropped_groups, } return diagnostics def _d6_post_process( self, ordered_plans: Dict[str, Dict[str, Any]], ) -> Tuple[Dict[str, Dict[str, Any]], Dict[str, str], str, str]: """LLM raw intent → deterministic bg_id catalog + sibling fields. Returns: (background_catalog, shot_background_map, bg_catalog_hash, shot_binding_hash). 실패 시 raise — downstream (background_render) 가 빈 dict 로 silent 진행 안 함. I8 intent sort + B7 monotonic next_b 는 `assign_bg_ids` 가 처리. B3 handshake: 각 plan.backgrounds[].bg_id slot 에 새 L##B## 갱신. """ from app.core.bg_catalog import ( assign_bg_ids, build_shot_background_map, compute_bg_catalog_hash, compute_shot_binding_hash, ) from app.modules.pipeline.background_master_plan import ( validate_master_plan_assigned, ) # 1. location_profiles 로드 — entity_canon.metadata_json.location.space_profile location_profiles = self._load_location_profiles() # 2. 모든 group 의 raw intents 평탄화. 같은 group 안 backgrounds[] 의 인덱스 # 보존 (handshake 용 — 후에 group/index 로 backref). flat_intents: List[Dict[str, Any]] = [] # backref: (gid, idx_in_group_backgrounds) → flat index backref_to_flat: Dict[Tuple[str, int], int] = {} for gid, gentry in ordered_plans.items(): if gentry.get("status") != "ok": continue plan = gentry.get("plan") or {} for idx, bg in enumerate(plan.get("backgrounds") or []): # raw intent shape (D6): bg 자체가 raw intent. # legacy compat (옛 cp 가 들어와도): 코드가 raw intent 필수 field 부재 # 시 assign_bg_ids 가 SemanticKeyError. force re-run 의무. backref_to_flat[(gid, idx)] = len(flat_intents) flat_intents.append(bg) # 3. assign_bg_ids — semantic_key dedup + monotonic next_b. # T4-fix B2: 자기 cp (own load_checkpoint — archive 복원 포함) 에서 # 이전 background_catalog 를 prev 로 전달. B7 monotonic next_b 가 deletion # 후에도 reset 안 함 — 운영 중 master_plan 재실행 시 ID 보존. prev_catalog = self._load_prev_background_catalog() catalog = assign_bg_ids(prev_catalog, flat_intents, location_profiles) # 4. validate_master_plan_assigned — BG_ID_RE 패턴 + unique. validate_master_plan_assigned(catalog) # 5. handshake — per-group plan.backgrounds[].bg_id slot 에 새 ID 갱신. # semantic_key 기반 lookup (assign_bg_ids 가 같은 sem_key 면 같은 bg_id). from app.core.bg_catalog import ( compute_semantic_key, normalize_space_key, SemanticKeyError, ) sem_to_bgid = {entry["semantic_key"]: bg_id for bg_id, entry in catalog.items()} for (gid, idx), _flat_idx in backref_to_flat.items(): bg = ordered_plans[gid]["plan"]["backgrounds"][idx] try: profile = location_profiles[bg["loc_id"]] space_key = normalize_space_key(bg["loc_id"], bg["space_key_hint"], profile) sem_key = compute_semantic_key( bg["loc_id"], space_key, bg["time_phase"], bg["state_class"] ) except (KeyError, SemanticKeyError) as exc: # assign_bg_ids 가 이미 같은 검증 통과 → 여기 도달 시 invariant 위반. raise RuntimeError( f"D6 handshake: group={gid} idx={idx} sem_key 재계산 실패 — {exc}" ) from exc bg["bg_id"] = sem_to_bgid[sem_key] # 6. shot_background_map + hashes shot_bg_map = build_shot_background_map(catalog) catalog_hash = compute_bg_catalog_hash(catalog) binding_hash = compute_shot_binding_hash(shot_bg_map) return catalog, shot_bg_map, catalog_hash, binding_hash def _load_prev_background_catalog(self) -> Dict[str, Dict[str, Any]]: """T4-fix B2 + T4-fix2 B1: 자기 cp 또는 archive 에서 이전 background_catalog 로드. force re-run / 운영 중 master_plan 재실행 시 prev_catalog 가 비어있으면 next_b counter reset 가능 → ID 변경 → 모든 downstream chain_bg 재렌더 발생. 이전 catalog 를 prev 로 넘기면 assign_bg_ids 가 same-semantic_key 는 ID 재사용 + 새 sem_key 만 monotonic next_b 부여. T4-fix2 B1 (review iter6): explicit force (`.force_cleared` marker) 환경에서도 prev catalog 보존 의무. `load_checkpoint()` 는 marker 있으면 archive 복원 차단 → prev={} → ID reset. 본 helper 는 archive 를 직접 read (manifest restore 안 함, force 의도 보존). 조회 우선순위: 1. manifest.json (정상 resume — load_checkpoint 와 동일 결과). 2. .force_cleared marker 있어도 latest archive 직접 read (`_find_latest_archive`). 3. 둘 다 없으면 빈 dict (fresh — 첫 실행). """ # 1. manifest.json 우선 try: manifest_path = self._cp_dir / "manifest.json" if manifest_path.exists(): text_data = manifest_path.read_text(encoding="utf-8").strip() if text_data: own_cp = json.loads(text_data) data = (own_cp or {}).get("data", {}) or {} catalog = data.get("background_catalog") if isinstance(catalog, dict) and catalog: return catalog except Exception as exc: logger.warning( "background_master_plan: manifest.json read failed (try archive): %s", exc, ) # 2. archive 직접 read (force_cleared marker 무시 — D6 ID 보존 우선) try: archive_path = self._find_latest_archive() if archive_path is not None: archive_data = json.loads(archive_path.read_text(encoding="utf-8")) data = (archive_data or {}).get("data", {}) or {} catalog = data.get("background_catalog") if isinstance(catalog, dict): logger.info( "background_master_plan: prev catalog read from archive %s " "(D6 ID preservation under explicit force).", archive_path.name, ) return catalog except Exception as exc: logger.warning( "background_master_plan: archive prev catalog read failed: %s", exc ) # 3. fresh return {} @staticmethod def _build_gen_order(plan: Dict[str, Any]) -> List[str]: """floor_plans + backgrounds 에서 topological gen_order build. floor_plans 가 먼저 (DAG levels), 그 후 backgrounds. depends_on_fp / depends_on_bg 모두 앞에 등장하도록 정렬. fp_id 와 bg_id 같이 unique key 로. T4-fix I3: missing dependency / cycle 검출 시 raise. - missing: depends_on_bg/depends_on_fp 가 plan 안 fp/bg 가 아님 → ValueError. - cycle: DFS 진행 중 같은 노드 재진입 → ValueError. """ # nodes fp_nodes = { fp["fp_id"]: list(fp.get("depends_on_fp") or []) for fp in plan.get("floor_plans") or [] } bg_nodes = { bg["bg_id"]: list((bg.get("depends_on_fp") or []) + (bg.get("depends_on_bg") or [])) for bg in plan.get("backgrounds") or [] if bg.get("bg_id") } all_nodes = {**fp_nodes, **bg_nodes} order: List[str] = [] seen: set = set() in_progress: set = set() def _visit(node: str, parent_chain: Tuple[str, ...] = ()) -> None: if node in seen: return if node not in all_nodes: raise ValueError( f"_build_gen_order: missing dependency {node!r} " f"(referenced from {parent_chain[-1] if parent_chain else 'root'!r}). " f"available: {sorted(all_nodes.keys())}" ) if node in in_progress: cycle_path = " → ".join(parent_chain + (node,)) raise ValueError( f"_build_gen_order: cycle detected — {cycle_path}" ) in_progress.add(node) for dep in all_nodes[node]: _visit(dep, parent_chain + (node,)) in_progress.discard(node) seen.add(node) order.append(node) # floor_plans 먼저 — deterministic for fp_id in fp_nodes: _visit(fp_id) for bg_id in bg_nodes: _visit(bg_id) return order 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]], scene_primary: Optional[Dict[int, str]] = None, ) -> Tuple[List[Dict[str, Any]], List[str]]: """그룹 멤버 loc_id가 등장하는 씬 + 그 씬의 shots만 추출. shot에 location_id가 없으면 scene_primary[scene_index]로 fallback (Phase 5 패턴). Returns: (relevant_scenes [{scene_index, heading, text, shots[]}], group_shot_ids ["S{si}_Shot{shi}", ...]) """ member_locs = {m["loc_id"] for m in group.get("members", []) or []} primary_map = scene_primary or {} 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 []) 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 primary_map.get(si_int, "") 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