"""OutdoorPlaceSpecStep — 야외 장소 마커 스펙 저작 (W22 직행 체인 ①, order 21.77). building_group(실외 멤버 보유)당 1스펙. 출력 checkpoint: data.groups[] = { "spec": {layout_narration_en, zone_labels_en[], items[...]}, "attempts": n, "outdoor_loc_ids": [...], "scene_indices": [...], } 실패 그룹 = {"error": str, "outdoor_loc_ids": [...]} (그룹 단위 격리) 씬 매핑 불가 그룹 = {"skipped": str, "outdoor_loc_ids": [...]} (근거 0 저작 금지) 게이트: background_mode ∈ {on, floor_plan_anchored} AND settings.outdoor_direct_compose_enabled (opt-in, default False — applicability=if_outdoor_direct_compose 로 run-all 에서도 제외). 설계: docs/w22-outdoor-canon-direct-compose-design-20260709/design.md """ import json import logging from pathlib import Path from typing import Any, Dict, List, Optional, Tuple from app.core.step_runner import StepRunner logger = logging.getLogger(__name__) SCHEMA_VERSION = 2 # v2: temporal_scope+excluded_transient_elements (2026-07-11) # 2026-07-25 사용자 지적("맵이 틀렸다 — 정류장이 도로 옆이 아니고 # 표식도 없다"): v3=허용 추정을 기능적 성립까지 확장(차량이 서는 # 시설↔노면 접함, 승하차 시설↔식별 표지, 지역 관행 배치). PROMPT_VERSION = "3" # 2026-07-11 시간 불변 + 07-25 기능적 성립 def outdoor_groups_of(classify_cp: Optional[Dict[str, Any]]) -> List[Tuple[Dict[str, Any], List[Dict[str, Any]]]]: """`background_classify` CP → 실외 멤버가 있는 그룹 `[(group, outdoor_members), …]`. ★술어는 여기 **한 곳** — 스텝(구매 단위)과 canary 의 사전 문(`assert_outdoor_group_cap_covers`)이 같은 함수를 부른다(Codex BLOCK 2026-09-03: 상한이 producer 산출을 provider 앞에서 대조해야 한다).""" groups = ((classify_cp or {}).get("data") or {}).get("building_groups") or [] out: List[Tuple[Dict[str, Any], List[Dict[str, Any]]]] = [] for g in groups: outdoor_members = [m for m in (g.get("members") or []) if m.get("is_indoor") is False] if outdoor_members: out.append((g, outdoor_members)) return out def outdoor_loc_ids_of(classify_cp: Optional[Dict[str, Any]]) -> List[str]: """실외 멤버가 가리키는 **unique** location id — 보충 행의 분모(그룹 수는 계측 분모).""" ids = set() for _g, members in outdoor_groups_of(classify_cp): ids |= {str(m.get("loc_id") or "") for m in members} - {""} return sorted(ids) class OutdoorPlaceSpecStep(StepRunner): """야외 장소 마커 스펙 (W22 ①) — evidence-bound 저작 + 결정론 검증.""" def _config_hash(self) -> str: import hashlib import 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, # direct ON 경로에선 기존 상수 True 와 byte-identical. 레시피 # (Codex 1차 리뷰 BLOCKING-4): map 게이트로도 실행 가능해져 # 실제 게이트 상태를 스탬프 — direct OFF+map ON 은 새 hash. "outdoor_direct_compose_enabled": bool( getattr(settings, "outdoor_direct_compose_enabled", False) ), } if bool(getattr(settings, "outdoor_map_conti_enabled", False)): payload["outdoor_map_conti_enabled"] = True # Stage D BLOCKING-1: lane pipe 게이트로도 실행 가능 — ON 시만 # 스탬프 (기존 조합 byte-identical) from app.core.applicability import outdoor_lane_pipe_on if outdoor_lane_pipe_on(): payload["outdoor_lane_pipe_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_place_spec: %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() from app.core.applicability import outdoor_lane_pipe_on if not ( getattr(settings, "outdoor_direct_compose_enabled", False) or getattr(settings, "outdoor_map_conti_enabled", False) # Stage D BLOCKING-1: lane pipe 도 spec 소비자 — manifest 게이트 # (if_outdoor_direct_or_map_or_lane)와 동일 predicate or outdoor_lane_pipe_on() ): return _empty() classify_cp = self._load_prev_checkpoint("background_classify") outdoor_groups = outdoor_groups_of(classify_cp) if not outdoor_groups: logger.info("outdoor_place_spec: 실외 멤버 그룹 없음 — no-op") return _empty() # ── 입력 cp 로드 ────────────────────────────────────────── 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 # 선택 샷 키 — 미선택 샷의 location 이 씬 매핑/샷 블록을 만들지 않게 # 필터 (Codex NARROW_1). 규칙은 W22 공용 헬퍼가 SOT (grounding 과 공유). from app.modules.pipeline.outdoor_direct_common import ( build_selected_keys, build_shot_loc_map, filter_group_shots, scene_indices_for_locs, ) selected_keys = build_selected_keys( self._load_prev_checkpoint("shot_selection") ) shot_loc_by_key = build_shot_loc_map( self._load_prev_checkpoint("shot_validator") ) staging_cp = self._load_prev_checkpoint("shot_staging") staging_shots = (staging_cp or {}).get("data", {}).get("shots", []) or [] merge_cp = self._load_prev_checkpoint("entity_merge") entity_locations = ( (merge_cp or {}).get("data", {}).get("locations", []) or [] ) # 2026-07-25 사용자 지적 실측 결함: 아래 두 키(rules_text/text)는 # visual_world_rules CP 에 **존재하지 않는다**(실제 키=region/era/ # rules/t2i_context) — 즉 장소 스펙은 지금까지 세계관을 한 글자도 # 받지 못했다. 공용 조립기로 교체(entity/beat 스텝 관례와 동조). rules_cp = self._load_prev_checkpoint("visual_world_rules") from app.core.world_context import build_world_rules_text rules_text = build_world_rules_text(rules_cp) from app.core.creator_corrections import project_corrections_block corrections_block = project_corrections_block(self.project_id) from app.modules.pipeline.outdoor_place_spec import run_outdoor_place_spec # ── 그룹별 저작 (그룹 단위 실패 격리) ───────────────────── results: Dict[str, Any] = {} completed = 0 failed = 0 for g, outdoor_members in outdoor_groups: gid = g.get("group_id", "") loc_ids = {m.get("loc_id", "") for m in outdoor_members} - {""} scene_indices = scene_indices_for_locs( loc_ids, scene_primary, shot_loc_by_key, selected_keys ) if not scene_indices: # 근거 씬 0 — 원문 근거 없는 저작은 발명이므로 스킵 (비실패) logger.info( "outdoor_place_spec: group=%s 매핑 씬 없음 — skip", gid ) results[gid] = { "skipped": "no scenes mapped to outdoor locs", "outdoor_loc_ids": sorted(loc_ids), } completed += 1 continue scene_texts = [ (si, scene_texts_all.get(si, "")) for si in scene_indices ] group_shots = filter_group_shots( staging_shots, scene_indices=scene_indices, loc_ids=loc_ids, scene_primary=scene_primary, shot_loc_by_key=shot_loc_by_key, selected_keys=selected_keys, ) try: out = run_outdoor_place_spec( group=g, outdoor_members=outdoor_members, entity_locations=entity_locations, scene_texts=scene_texts, shots=group_shots, rules_text=rules_text, creator_corrections_block=corrections_block, prompt_version=PROMPT_VERSION, project_config=self.project_config, opik_metadata=self.build_opik_metadata(), ) results[gid] = { "spec": out["spec"], "attempts": out["attempts"], "outdoor_loc_ids": sorted(loc_ids), "scene_indices": scene_indices, } completed += 1 except Exception as exc: # noqa: BLE001 logger.exception( "outdoor_place_spec: group=%s 저작 실패", gid ) results[gid] = { "error": str(exc), "outdoor_loc_ids": sorted(loc_ids), # Codex Stage D 재재리뷰: 실패 그룹에도 구조키 보존 — # 하류 parity 가 '현재 선택 샷 존재' 를 재구성해 # 실패를 skip 으로 세탁하는 fail-open 차단 "scene_indices": scene_indices, } failed += 1 return { "applicable_count": len(outdoor_groups), "completed_count": completed, "failed_count": failed, "schema_version": SCHEMA_VERSION, "config_hash": self._config_hash(), "data": {"groups": results}, }