"""야외 3레인 게이트 스텝 (설계 v2 Stage A) — 세그먼트/레인/바인딩 저작. outdoor_place_spec 그룹(스펙 있는 그룹만)마다 place_segment 분할+선택 샷 레인 바인딩을 LLM 저작한다. 이미지 생성 없음 — canon/맵 자산 불요. opt-in: settings.outdoor_lane_plan_enabled (default False). """ 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__) SCHEMA_VERSION = 1 def _prompt_version() -> str: """팩 selector — settings opt-in (2026-07-19 재설계 C: "2"=lane none 신설 팩). default "1"=기존 프로젝트 byte-identical.""" from app.core.config import settings return str(getattr(settings, "outdoor_lane_plan_prompt_version", "1")) # evidence 무결성 게이트 의미 버전 — 출력 스키마는 불변이지만 validator 계약이 # 바뀌면 완료 CP 가 cp-clean SKIP 으로 잔존하지 않도록 hash 에 스탬프 (Codex # 재리뷰 NARROW_1). v2=인용 원문 실재(whitespace 정규화)+segment_id 중복 reject. # 주의: bump 시 기존 CP 는 config_hash mismatch BLOCK — force 명시 실행 필요. # v3 (2026-07-25 지적②): 장소 단위 복잡 구조물 게이트 — site 필드 # 실재 검증(require_site)+결정론 강등(apply_site_complexity_gate)이 # validator 계약에 편입. EVIDENCE_VALIDATION_VERSION = 3 class OutdoorLanePlanStep(StepRunner): """야외 lane plan (3레인 ①) — 스펙+씬 전문+선택 샷 → 세그먼트/레인.""" def _config_hash(self) -> str: import hashlib import json as _json from app.core.config import settings payload = { "model": settings.openai_model, "schema_version": SCHEMA_VERSION, "prompt_version": _prompt_version(), "evidence_validation_version": EVIDENCE_VALIDATION_VERSION, "outdoor_lane_plan_enabled": True, } 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: # noqa: BLE001 logger.warning( "outdoor_lane_plan: %s 로드 실패: %s", step_id, exc ) return None def _execute(self, mode: str = "resume") -> Dict[str, Any]: from app.core.config import settings def _empty() -> Dict[str, Any]: return { "applicable_count": 0, "completed_count": 0, "failed_count": 0, "schema_version": SCHEMA_VERSION, "config_hash": self._config_hash(), "data": {"groups": {}}, } if settings.background_mode not in {"on", "floor_plan_anchored"}: return _empty() if not getattr(settings, "outdoor_lane_plan_enabled", False): return _empty() spec_cp = self._load_prev_checkpoint("outdoor_place_spec") spec_groups = (spec_cp or {}).get("data", {}).get("groups", {}) or {} target: Dict[str, Dict[str, Any]] = {} results: Dict[str, Any] = {} for gid, entry in spec_groups.items(): if not isinstance(entry, dict) or not entry.get("spec"): # 스펙 결측(상류 skip/실패) = lane plan 대상 아님 — 비실패 results[gid] = {"skipped": "place spec missing"} continue target[gid] = { "spec": entry["spec"], "outdoor_loc_ids": entry.get("outdoor_loc_ids") or [], "scene_indices": entry.get("scene_indices") or [], } if not target and not results: logger.info("outdoor_lane_plan: 대상 그룹 없음 — no-op") return _empty() # ── 공통 입력 로드 ──────────────────────────────────────── scene_save_cp = self._load_prev_checkpoint("scene_save") scene_texts_all: Dict[int, str] = {} for seg in (scene_save_cp or {}).get("data", {}).get("segments", []) or []: si = seg.get("scene_index") if isinstance(si, int): # 씬 원문 전문 — 절대 자르지 않는다 (CLAUDE.md 절대 규칙) scene_texts_all[si] = seg.get("text") or "" director_cp = self._load_prev_checkpoint("scene_director") scene_primary: Dict[int, str] = {} for sc in (director_cp or {}).get("data", {}).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 from app.modules.pipeline.outdoor_direct_common import ( build_selected_keys, build_shot_loc_map, filter_group_shots, ) selected_keys = build_selected_keys( self._load_prev_checkpoint("shot_selection") ) validator_cp = self._load_prev_checkpoint("shot_validator") shot_loc_by_key = build_shot_loc_map(validator_cp) # 샷 서술 SOT=shot_validator — staging 샷에는 description/characters 가 # 없다(camera_direction 등 연출 필드만). 서술 공백은 lane 판정 근거를 # 무너뜨림 (5회차 실측: 전 샷 "쇼트 설명 없음" low confidence 양산). shot_desc_by_key: Dict[tuple, Dict[str, Any]] = {} for sc in (validator_cp or {}).get("data", {}).get("scenes", []) or []: v_si = sc.get("scene_index") if v_si is None: continue for sh in sc.get("shots", []) or []: v_shi = sh.get("shot_index") if v_shi is None: continue extra: Dict[str, Any] = {} if sh.get("description"): extra["description"] = sh["description"] if sh.get("characters"): extra["characters"] = sh["characters"] if extra: shot_desc_by_key[(int(v_si), int(v_shi))] = extra staging_cp = self._load_prev_checkpoint("shot_staging") staging_shots = (staging_cp or {}).get("data", {}).get("shots", []) or [] from app.modules.pipeline.outdoor_lane_plan import ( run_outdoor_lane_plan_group, ) # ── 그룹별 lane plan (그룹 단위 실패 격리) ──────────────── applicable = 0 completed = 0 failed = 0 skipped_count = len(results) for gid in sorted(target): info = target[gid] loc_ids = set(info["outdoor_loc_ids"]) group_shots = filter_group_shots( staging_shots, scene_indices=info["scene_indices"], loc_ids=loc_ids, scene_primary=scene_primary, shot_loc_by_key=shot_loc_by_key, selected_keys=selected_keys, ) if not group_shots: # 바인딩할 선택 샷 0 — LLM 호출 무의미 (비실패) results[gid] = {"skipped": "no selected shots for group"} skipped_count += 1 continue group_shots = [ {**sh, **shot_desc_by_key.get( (int(sh["scene_index"]), int(sh["shot_index"])), {})} for sh in group_shots ] applicable += 1 scene_texts = { si: scene_texts_all.get(si, "") for si in sorted({ int(sh["scene_index"]) for sh in group_shots }) } try: out = run_outdoor_lane_plan_group( spec=info["spec"], group_shots=group_shots, scene_texts=scene_texts, prompt_version=_prompt_version(), project_config=self.project_config, opik_metadata=self.build_opik_metadata(), ) results[gid] = { "status": "ok", "plan": out["plan"], "attempts": out["attempts"], "outdoor_loc_ids": info["outdoor_loc_ids"], "scene_indices": info["scene_indices"], } completed += 1 except Exception as exc: # noqa: BLE001 logger.exception("outdoor_lane_plan: group=%s 실패", gid) results[gid] = {"status": "failed", "error": str(exc)} failed += 1 # 카운트 불변식 completed<=applicable — skip 을 완료로 세므로 # applicable 에도 포함 (image_steps 관례, Codex NARROW) return { "applicable_count": applicable + skipped_count, "completed_count": completed + skipped_count, "failed_count": failed, "schema_version": SCHEMA_VERSION, "config_hash": self._config_hash(), "data": {"groups": results}, }