"""OutdoorFrameModeStep — (DORMANT, 2026-07-16 복잡 구조물=A/B 재설계) structure_plate 샷의 스케치 경로가 소멸해 frame_mode 판정의 소비처가 없다 — applicability.outdoor_frame_mode_on() 이 항상 False(스텝=항상 not_applicable), shot_conti_light/still_recipe/scene hash 의 소비·스탬프 전부 제거됨. 파일·모듈은 삭제 금지 원칙으로 보존. 이하 원 설계: (E2E6 피드백 ③, order 21.78) outdoor_lane_plan 의 structure_plate 바인딩 샷 전부에 대해 frame_mode (structure_dominant/layout_dominant)를 LLM 판정한다. 소비자= shot_conti_light(structure_dominant=스케치 생략 → direct_seed) + still_recipe(direct_seed 샷 refs=seed 직참조, prev 는 classify SOT 로 location_ref_mode=prev_only/seed_only 결정). 계약: evidence literal 인용 실재+대상 샷 완전성 parity(모듈 검증, 재시도 소진=AppError fail-closed). persisted lane plan 은 소비 직전 revalidate(seed 스텝 관례). data.shots[tag]={status, frame_mode, confidence, evidence, rationale_ko, group_id}. """ 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 PROMPT_VERSION = "1" class OutdoorFrameModeStep(StepRunner): """레인2 frame_mode 판정 (분류 전용 — 이미지 생성 없음).""" def _config_hash(self) -> str: import hashlib import json as _json from app.core.step_runner import compute_config_hash from app.modules.pipeline.outdoor_frame_mode import ( resolve_prompt_version, ) from app.core.config import settings payload = { "schema_version": SCHEMA_VERSION, "prompt_version": resolve_prompt_version(PROMPT_VERSION), # Codex f7c81336 HIGH-4: alias 가 아니라 물리 모델 — 모델 # 교체가 hash 드리프트로 감지되도록 (lane plan 스텝 관례) "model": settings.openai_model, "outdoor_frame_mode_enabled": True, "project_config_hash": compute_config_hash(self.project_config), } 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_frame_mode: %s 로드 실패: %s", step_id, exc) return None def _execute(self, mode: str = "resume") -> Dict[str, Any]: from app.core.applicability import outdoor_frame_mode_on from app.core.errors import AppError from app.modules.pipeline.outdoor_frame_mode import ( run_outdoor_frame_mode, ) from app.modules.pipeline.outdoor_lane_plan import ( build_group_shot_reconstructor, revalidate_persisted_plan, validate_group_parity, ) 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": {"shots": {}}, } # 이중 게이트 (outdoor_lane_pipe_on 관례 — applicability 와 동일 SOT) if not outdoor_frame_mode_on(): return _empty() lane_cp = self._load_prev_checkpoint("outdoor_lane_plan") lane_data = (lane_cp or {}).get("data", {}) or {} spec_cp = self._load_prev_checkpoint("outdoor_place_spec") spec_groups = (spec_cp or {}).get("data", {}).get("groups", {}) or {} scene_cp = self._load_prev_checkpoint("scene_save") scene_texts: Dict[int, str] = {} scene_headings: Dict[int, str] = {} for seg in (scene_cp or {}).get("data", {}).get("segments", []) or []: si = seg.get("scene_index") if isinstance(si, int): # 씬 원문 전문 — 절대 자르지 않는다 scene_texts[si] = seg.get("text") or "" scene_headings[si] = seg.get("heading") or "" validator_cp = self._load_prev_checkpoint("shot_validator") desc_by_key: Dict[str, str] = {} for sc in (validator_cp or {}).get("data", {}).get("scenes", []) or []: si = sc.get("scene_index") for sh in sc.get("shots", []) or []: shi = sh.get("shot_index") if si is not None and shi is not None: desc_by_key[f"S{int(si)}sh{int(shi)}"] = ( sh.get("description") or "") staging_cp = self._load_prev_checkpoint("shot_staging") cam_by_key: Dict[str, str] = {} for sh in (staging_cp or {}).get("data", {}).get("shots", []) or []: si, shi = sh.get("scene_index"), sh.get("shot_index") if si is not None and shi is not None: cam_by_key[f"S{int(si)}sh{int(shi)}"] = ( sh.get("camera_direction") or "") reconstruct = build_group_shot_reconstructor( staging_cp=staging_cp, validator_cp=validator_cp, selection_cp=self._load_prev_checkpoint("shot_selection"), director_cp=self._load_prev_checkpoint("scene_director"), ) # Codex f7c81336 HIGH-3: 그룹 parity 선행 — lane plan 의 누락/ # failed 그룹이 '대상 없음'으로 세탁돼 빈 completed CP 가 되는 # 창 차단 (conti 단계와 동일 계약) parity = validate_group_parity( lane_data=lane_data, spec_groups=spec_groups, reconstruct=reconstruct, ) if parity: raise AppError( code="step.missing_input.outdoor_frame_mode", message=( "lane plan 그룹 parity 위반 — outdoor_lane_plan " "재실행 필요: " + "; ".join(parity[:8]) ), status_code=422, ) # ── structure_plate 대상 샷 수집 (persisted plan 재검증 포함) ── shots: list = [] group_by_tag: Dict[str, str] = {} for gid in sorted(lane_data.get("groups", {}) or {}): entry = lane_data["groups"][gid] if not isinstance(entry, dict) or entry.get("status") != "ok": continue plan = entry.get("plan") or {} sp = [b for b in plan.get("shot_bindings") or [] if (b or {}).get("lane") == "structure_plate"] if not sp: continue spec_entry = spec_groups.get(gid) or {} violations = revalidate_persisted_plan( plan, scene_texts, group_shots=reconstruct(spec_entry), ) if violations: raise AppError( code="step.contract_violation.outdoor_frame_mode", message=( f"persisted lane plan 재검증 실패 (group={gid}): " + "; ".join(violations[:8]) ), status_code=422, ) for b in sp: si, shi = int(b["scene_index"]), int(b["shot_index"]) tag = f"S{si}sh{shi}" shots.append({ "scene_index": si, "shot_index": shi, "description": desc_by_key.get(tag, ""), "camera_direction": cam_by_key.get(tag, ""), }) group_by_tag[tag] = gid if not shots: logger.info("outdoor_frame_mode: structure_plate 샷 없음 — no-op") return _empty() data = run_outdoor_frame_mode( shots=shots, scene_texts=scene_texts, scene_headings=scene_headings, prompt_version=PROMPT_VERSION, project_config=self.project_config, opik_metadata=self.build_opik_metadata(), ) for tag, entry in data["shots"].items(): entry["group_id"] = group_by_tag.get(tag) return { "applicable_count": len(shots), "completed_count": len(data["shots"]), "failed_count": 0, "schema_version": SCHEMA_VERSION, "config_hash": self._config_hash(), "data": data, }