"""실내 shared-model pose 가이드 context + attach (Wave5). ★Codex 정렬 — 2단 분리: - build_indoor_shared_pose_context: **coordinator main thread, episode당 1회**. candidate → judge(VLM) → admit → 멤버별 guide+QC precompute(group당 1회) → immutable guide_by_shot map. DB/LLM/VLM 호출은 전부 여기(main thread). - attach_indoor_pose_guide_ref: **worker thread**. precompute 된 guide bytes 를 lookup 만 하고 4-list 를 mutate. DB/LLM/VLM/생성 0. flag OFF → byte-identical no-op. """ import logging from typing import Any, Dict, List, Optional, Tuple from app.core.framing_scale import FRAMING_CLOSE, FRAMING_INSERT logger = logging.getLogger(__name__) INDOOR_POSE_GUIDE_LABEL = "[INDOOR POSE GUIDE]" ShotKey = Tuple[int, int] # ── attach (worker, lookup-only) ──────────────────────────────────────── # W-B (2026-07-03) 프레이밍 게이트 — 전신 pose 가이드가 구조적으로 불일치하는 # framing_scale enum (shot_staging SOT — app.core.framing_scale 상수 재사용). _POSE_GUIDE_SKIP_FRAMINGS = frozenset({FRAMING_CLOSE, FRAMING_INSERT}) def attach_indoor_pose_guide_ref( labeled_refs: List[Any], ref_roles: List[str], ref_role_metadata: List[Dict[str, Any]], attached_meta: List[Any], *, scene_index: Optional[int], shot_index: Optional[int], indoor_pose_ctx: Optional[Dict[str, Any]], framing: Optional[str] = None, ) -> bool: """precompute 된 guide 를 indoor_pose_guide ref 로 append (4-list parallel mutate). flag(indoor_shared_pose_guide_enabled) OFF → no-op(byte-identical). 해당 샷 guide 없으면(미admit/QC fail/생성실패) no-op. worker thread 안전 — lookup + list mutate 만. ``framing`` (W-B 2026-07-03, shot_staging.framing_scale enum SOT): 전신 pose 가이드는 close/insert 샷과 구조적으로 불일치(전수 육안 (e) — 손목/얼굴 클로즈업에 전신 스탠딩 가이드 부착) — close류면 attach skip(진단 로그). None(구 호출부)이면 게이트 비활성(하위호환). """ from app.core.config import settings if not bool(getattr(settings, "indoor_shared_pose_guide_enabled", False)): return False entry = ((indoor_pose_ctx or {}).get("guide_by_shot") or {}).get((scene_index, shot_index)) if not entry or not entry.get("png"): return False # W-B 프레이밍 게이트 (정확 enum 비교, 글자패턴 0) — close류 샷 attach skip. if framing in _POSE_GUIDE_SKIP_FRAMINGS: logger.info( "indoor_pose_guide: S%ssh%s attach SKIPPED — framing_scale=%r " "(close류 샷에 전신 pose 가이드 부착 금지, W-B 게이트)", scene_index, shot_index, framing) return False labeled_refs.append((INDOOR_POSE_GUIDE_LABEL, entry["png"])) ref_roles.append("indoor_pose_guide") ref_role_metadata.append({ "group_id": entry.get("group_id"), "visible_focus": entry.get("visible_focus", ""), # goal#3 (Codex 설계 B): guide ImageAsset UUID 를 구조 lineage 로 실어 # final scene input_image_ids 로 복원(라벨파싱 금지·UUID SOT). None 이면 # coordinator 가 lineage 에서 제외(엣지 미생성). "asset_id": entry.get("asset_id"), "pipeline_role": "indoor_pose_guide", }) attached_meta.append(("indoor_pose_guide", entry.get("group_id"))) return True # ── context build (main thread, episode당 1회 precompute) ──────────────── def _structural_signal_names(sig: Dict[str, Any]) -> List[str]: """complexity_signals 딕셔너리에서 '존재하는 구조 신호' 이름만 뽑는다(의미판정 0). judge 가 어떤 구조 근거가 있는지 audit 하도록 seed prefilter 메타에 싣는다. """ names: List[str] = [] if int(sig.get("fsc_constraint_count") or 0) > 0: names.append("frame_spatial_contract") if int(sig.get("entity_count") or 0) > 0: names.append("character_targets") if int(sig.get("depth_plane_bucket") or 0) >= 2: names.append("multi_depth") if sig.get("gesture_present"): names.append("gesture") if str(sig.get("framing") or "") not in ("", "unknown"): names.append("framing_scale") return names def _judge_payload(cand: Dict[str, Any], shot_by_key: Dict[ShotKey, Dict[str, Any]], name_by_id: Optional[Dict[str, Any]] = None, *, bg_plate_present: Optional[bool] = None) -> Dict[str, Any]: """name-free judge 입력 — 구조 신호만(이름/ID/문구 0). name_by_id 는 brief 내부 character_angle 매칭에만 쓰이고 payload 엔 figure_count/framing 등 집계만 흐른다. ★4b(Codex): single_shot_complexity lane 이면 group-level seed prefilter 메타를 싣는다(seed_lane/why_candidate/figure_count/bg_plate_present/framing_scale/ character_angle_count/frame_spatial_contract_present/available_structural_signal_ names). judge 가 단일 복잡샷 근거를 구조적으로 평가하도록(코드는 의미판정 0). """ from app.modules.pipeline.indoor_shared_pose_plan import build_pose_brief shots = [] for si, shi in cand["member_keys"]: sig = cand["signals_by_shot"].get(f"{si}_{shi}", {}) brief = build_pose_brief(shot_by_key[(si, shi)], name_by_id=name_by_id) shots.append({ "shot_key": f"{si}_{shi}", "signals": sig, "framing": brief.get("framing"), "figure_count": len(brief.get("figures") or []), "contact_locked": brief.get("contact_locked"), }) payload: Dict[str, Any] = { "scene_index": cand["scene_index"], "group_signals": cand["group_signals"], "zoom_members": [f"{si}_{shi}" for si, shi in cand.get("zoom_members", [])], "shots": shots, } lane = cand.get("lane") if lane == "single_shot_complexity": a_si, a_shi = cand["anchor_key"] a_sig = cand["signals_by_shot"].get(f"{a_si}_{a_shi}", {}) payload["seed_lane"] = "single_shot_complexity" payload["seed_prefilter_meta"] = { "why_candidate": "single_shot_indoor_figure_present", "figure_count": int(a_sig.get("entity_count") or 0), "character_angle_count": int(a_sig.get("entity_count") or 0), "bg_plate_present": bg_plate_present, "framing_scale": a_sig.get("framing"), "frame_spatial_contract_present": int(a_sig.get("fsc_constraint_count") or 0) > 0, "available_structural_signal_names": _structural_signal_names(a_sig), } else: payload["seed_lane"] = "cross_shot_continuity" return payload def _resolve_guide_asset_ids( db: Any, project_id: str, episode_id: str, guide_by_shot: Dict[ShotKey, Dict[str, Any]], diagnostics: List[Dict[str, Any]], ) -> None: """goal#3 (Codex 설계 A) — accepted indoor_pose_guide ImageAsset UUID 를 구조키로 회수해 guide_by_shot[k]["asset_id"] 채움(라벨파싱 금지·UUID SOT). cache_hit 경로도 prior accepted row 를 같은 resolver 로 찾는다. resolve 우선순위: 1) (group_id, bg_key, guide_hash) 정확 매칭 2) legacy fallback (group_id, bg_key) created_at desc miss → asset_id 미설정(None) + diagnostic(엣지 생성 안 함). DB read-only. """ if not guide_by_shot: return import json by_triple: Dict[Tuple[str, str, str], str] = {} # ★Codex NARROW: pair fallback 은 (group_id,bg_key) accepted row 가 정확히 1개일 # 때만 허용(여러 개면 어느 게 이 shot 의 guide 인지 구조 증거 없음 → wrong edge 가 # missing edge 보다 나쁨). pair → asset_id 리스트로 모아 len 으로 판정. pair_rows: Dict[Tuple[str, str], List[str]] = {} try: from sqlalchemy import text rows = db.execute(text( "SELECT id, pipeline_metadata_json FROM image_asset " "WHERE project_id = :pid AND episode_id = :eid " "AND pipeline_role = 'indoor_pose_guide' AND disposition = 'accepted' " "AND is_intermediate = true " "ORDER BY created_at DESC" ), {"pid": project_id, "eid": episode_id}).fetchall() except Exception as exc: # noqa: BLE001 logger.warning("indoor pose guide: asset_id resolve 쿼리 실패: %s", exc) return # created_at desc 정렬 → 최신이 먼저. triple 은 같은 키 첫 등장(최신) 보존. for asset_id, meta_json in rows: try: meta = json.loads(meta_json) if meta_json else {} except Exception: # noqa: BLE001 meta = {} gid = str(meta.get("group_id") or "") bgk = str(meta.get("bg_key") or "") gh = str(meta.get("guide_hash") or "") if gid and bgk and gh: by_triple.setdefault((gid, bgk, gh), str(asset_id)) if gid and bgk: pair_rows.setdefault((gid, bgk), []).append(str(asset_id)) for key, entry in guide_by_shot.items(): gid = str(entry.get("group_id") or "") bgk = str(entry.get("bg_key") or "") gh = str(entry.get("guide_hash") or "") aid = by_triple.get((gid, bgk, gh)) if aid: entry["asset_id"] = aid continue # triple miss → pair fallback (unambiguous 일 때만). plist = pair_rows.get((gid, bgk)) or [] if len(plist) == 1: entry["asset_id"] = plist[0] elif len(plist) >= 2: diagnostics.append({ "group_id": gid, "shot": f"{key[0]}_{key[1]}", "status": "guide_asset_unresolved_ambiguous_pair", "reason": f"{len(plist)} accepted indoor_pose_guide rows for " f"(group_id={gid}, bg_key={bgk}) and no guide_hash match " f"(hash={gh}) — ambiguous, edge 미생성(wrong>missing)"}) else: diagnostics.append({ "group_id": gid, "shot": f"{key[0]}_{key[1]}", "status": "guide_asset_unresolved", "reason": f"no accepted indoor_pose_guide row for " f"(group_id={gid}, bg_key={bgk}, guide_hash={gh}) — " f"stale cache without DB row? (cache_rehydrate 미구현)"}) def build_indoor_shared_pose_context( *, selected_keys: List[ShotKey], bg_id_by_shot: Dict[ShotKey, str], shot_by_key: Dict[ShotKey, Dict[str, Any]], zoom_member_keys: set, background_chain_bg_map: Dict[str, Any], cache_dir: Any, judge_fn: Any, guide_fn: Any, qc_fn: Any, judge_enabled: bool, bg_loc_by_id: Optional[Dict[str, str]] = None, indoor_locs: Optional[set] = None, outdoor_locs: Optional[set] = None, bg_asset_by_id: Optional[Dict[str, str]] = None, skip_shot_keys: Optional[set] = None, name_by_id: Optional[Dict[str, str]] = None, openai_client: Any = None, model: str = "gpt-image-2.5-sunburst", single_shot_lane_enabled: bool = False, log_context: Optional[Dict[str, Any]] = None, ) -> Dict[str, Any]: """candidate → (indoor 게이트) → judge → admit → 멤버별 guide+QC precompute → ctx. judge_fn/guide_fn/qc_fn 은 DI(테스트 override). guide_fn = build_indoor_pose_guide 형태(→(png|None, diag)). admit + guide 성공(QC pass)한 멤버만 guide_by_shot 등록. deny/QC fail/bg 부재는 diagnostics(캔버스 노출용). judge_enabled=False → admit 0. ★indoor/outdoor 게이트(Codex 정렬): indoor_locs 가 주어지면 candidate 를 filter_indoor_groups(구조키 bg_loc_by_id + is_indoor 집합)로 순수 indoor 만 남기고 outdoor/mixed/unresolved 는 default-deny diagnostic. indoor_locs=None 이면 게이트 skip(테스트/하위호환). """ from app.core.steps.indoor_pose_guide_judge import ( evaluate_indoor_pose_guide_judge, resolve_single_shot_targets, ) from app.modules.pipeline.indoor_shared_pose_plan import ( build_pose_brief, candidate_groups, filter_indoor_groups, ) guide_by_shot: Dict[ShotKey, Dict[str, Any]] = {} diagnostics: List[Dict[str, Any]] = [] candidates = candidate_groups( selected_keys=selected_keys, bg_id_by_shot=bg_id_by_shot, shot_by_key=shot_by_key, zoom_member_keys=zoom_member_keys, single_shot_lane_enabled=single_shot_lane_enabled) if indoor_locs is not None: candidates, gate_excluded = filter_indoor_groups( candidates, bg_loc_by_id=bg_loc_by_id or {}, indoor_locs=indoor_locs, outdoor_locs=outdoor_locs or set()) for ex in gate_excluded: diagnostics.append({ "group_id": f"indoor-s{ex.get('scene_index')}-{ex.get('bg_id')}", "status": "gate_excluded", "reason": ex.get("reason"), "location_id": ex.get("location_id")}) skip_keys = skip_shot_keys or set() for cand in candidates: # ★4b(Codex): single_shot lane 은 group_id 를 cross-shot 과 분리(안정키). is_single = cand.get("lane") == "single_shot_complexity" if is_single: _a_si, _a_shi = cand["anchor_key"] gid = f"isp-single-{cand['bg_id']}-s{cand['scene_index']}-sh{_a_shi}" _a_entry = background_chain_bg_map.get(f"{_a_si}_{_a_shi}") or {} bg_plate_present = bool(_a_entry.get("image_bytes")) else: gid = f"indoor-s{cand['scene_index']}-{cand['bg_id']}" bg_plate_present = None # resume 가드(Codex): 그룹 전 멤버가 이미 완료(skip) → judge/guide 비용 0. # 그룹 형성·judge 는 전체 멤버 staging 으로 하되(연속성 신호 보존), 미완료 # 멤버가 1개라도 있을 때만 진행한다. live_members = [k for k in cand["member_keys"] if k not in skip_keys] if not live_members: diagnostics.append({"group_id": gid, "status": "all_members_done_skip"}) continue if not judge_enabled: diagnostics.append({"group_id": gid, "status": "judge_disabled_diagnostic_only"}) continue try: verdict = judge_fn( _judge_payload(cand, shot_by_key, name_by_id, bg_plate_present=bg_plate_present), log_context=log_context) except Exception as exc: # noqa: BLE001 — 비차단 diagnostics.append({"group_id": gid, "status": "judge_error", "reason": str(exc)[:160]}) continue attach, deny = evaluate_indoor_pose_guide_judge(verdict) if not attach: diagnostics.append({"group_id": gid, "status": "denied", "reason": deny}) continue # ★shot-scoping(Codex B'): single_shot lane 은 judge 가 지정한 target shot 만 # (member ∧ live ∧ figure>=1). 0-figure establishing/insert 는 절대 target 불가. # cross_shot/both 는 기존처럼 live 멤버 전체. decision = verdict.get("decision_type") if decision == "single_shot_complexity": targets = resolve_single_shot_targets(verdict) gen_members = [ k for k in live_members if k in targets and len((build_pose_brief(shot_by_key[k], name_by_id=name_by_id).get("figures") or [])) >= 1 ] if not gen_members: diagnostics.append({"group_id": gid, "status": "single_shot_no_valid_target"}) continue else: gen_members = live_members # guide 생성은 미완료(live) target 멤버만 — 완료 멤버는 batch 에서 어차피 skip. for si, shi in gen_members: entry = background_chain_bg_map.get(f"{si}_{shi}") or {} env_bytes = entry.get("image_bytes") # ★4b(Codex): single-shot lane 은 actual bg plate underlay 필수. # plate 없으면 guide_fn 호출 0 + diagnostic(no-guide degrade). cross-shot # 은 기존 동작 유지(byte-identical) — single lane 에만 적용. if is_single and not env_bytes: diagnostics.append({ "group_id": gid, "shot": f"{si}_{shi}", "status": "single_shot_no_bg_plate", "reason": "single-shot lane requires actual bg plate underlay; " "guide_fn skipped (no-guide degrade)"}) continue bg_key = str(entry.get("bg_id") or cand["bg_id"]) pose_brief = build_pose_brief(shot_by_key[(si, shi)], name_by_id=name_by_id) png, gdiag = guide_fn( group_id=gid, pose_brief=pose_brief, env_bg_bytes=env_bytes, bg_key=bg_key, cache_dir=cache_dir, qc_fn=qc_fn, openai_client=openai_client, model=model, bg_asset_id=(bg_asset_by_id or {}).get(bg_key)) if png: # goal#3 (Codex 설계): bg_key+guide_hash 를 entry 에 실어 loader 의 # DB resolver 가 accepted guide ImageAsset UUID 를 구조키로 회수 → # asset_id(아래 loader 에서 채움) → final scene input_image_ids lineage. guide_by_shot[(si, shi)] = { "png": png, "group_id": gid, "visible_focus": pose_brief.get("framing", ""), "bg_key": bg_key, "guide_hash": gdiag.get("hash"), "asset_id": None} else: diagnostics.append({ "group_id": gid, "shot": f"{si}_{shi}", "status": gdiag.get("status"), "reason": gdiag.get("reason") or gdiag.get("qc_reason")}) return {"guide_by_shot": guide_by_shot, "diagnostics": diagnostics} # ── orchestration loader (scene_image_service 단일 진입점, episode당 1회) ── def _load_bg_loc_by_id(projects_dir: str, project_id: str, episode_id: str) -> Dict[str, str]: """background_render groups[bg_id].location_id 구조키 맵 (gate loc resolve SOT).""" import json from app.services.scene_checkpoint_loaders import _ep_checkpoint_path out: Dict[str, str] = {} p = _ep_checkpoint_path(projects_dir, project_id, episode_id, "background_render") if not p.exists(): return out try: groups = (json.loads(p.read_text(encoding="utf-8")).get("data") or {}).get("groups") or {} for bg_id, gr in groups.items(): loc = gr.get("location_id") if isinstance(bg_id, str) and bg_id and isinstance(loc, str) and loc: out[bg_id] = loc except Exception as exc: # noqa: BLE001 logger.warning("indoor pose guide: background_render groups 로드 실패: %s", exc) return out def _load_indoor_outdoor_locs(projects_dir: str, project_id: str, episode_id: str): """background_classify building_groups → (indoor_locs, outdoor_locs).""" import json from app.modules.pipeline.indoor_shared_pose_plan import classify_loc_sets from app.services.scene_checkpoint_loaders import _ep_checkpoint_path p = _ep_checkpoint_path(projects_dir, project_id, episode_id, "background_classify") if not p.exists(): return set(), set() try: bg = (json.loads(p.read_text(encoding="utf-8")).get("data") or {}).get("building_groups") or [] return classify_loc_sets(bg) except Exception as exc: # noqa: BLE001 logger.warning("indoor pose guide: background_classify 로드 실패: %s", exc) return set(), set() def _load_bg_asset_by_id(db: Any, project_id: str, episode_id: str) -> Dict[str, str]: """bg_id(variant_type) → background_render ImageAsset.id (lineage resolve, 구조키). is_primary desc / created_at desc tie-breaker (Wave1 패턴). DB read-only. """ out: Dict[str, str] = {} try: from sqlalchemy import text rows = db.execute(text( "SELECT variant_type, id FROM image_asset " "WHERE project_id = :pid AND episode_id = :eid " "AND pipeline_role = 'background_render' AND variant_type IS NOT NULL " "ORDER BY is_primary DESC, created_at DESC" ), {"pid": project_id, "eid": episode_id}).fetchall() for variant_type, asset_id in rows: if variant_type and variant_type not in out: out[str(variant_type)] = str(asset_id) except Exception as exc: # noqa: BLE001 logger.warning("indoor pose guide: bg_asset resolve 실패: %s", exc) return out def _load_name_by_short_id(db: Any, project_id: str, episode_id: str) -> Dict[str, str]: """short_id(C##/P##/L##) → entity name 맵. build_pose_brief 의 target_id→name character_angle 매칭용(읽기 전용). reference_svc.load_episode_entity_lookup 재사용. """ out: Dict[str, str] = {} try: from app.services.scene_reference_service import SceneReferenceService svc = SceneReferenceService(db, project_id) lk = svc.load_episode_entity_lookup(episode_id) for v in (lk or {}).values(): sid = str((v or {}).get("short_id") or "").strip() nm = str((v or {}).get("name") or "").strip() if sid and nm: out[sid] = nm except Exception as exc: # noqa: BLE001 logger.warning("indoor pose guide: name_by_id 로드 실패: %s", exc) return out def load_indoor_shared_pose_guides( *, project_id: str, episode_id: str, stills: List[Dict[str, Any]], staging_map: Dict[str, Any], background_chain_bg_map: Dict[str, Any], zoom_ctx: Optional[Dict[str, Any]], db: Any, openai_client: Any = None, already_done_still_ids: Optional[set] = None, execution_allowlist_still_ids: Optional[set] = None, ) -> Dict[str, Any]: """flag ON 일 때만 episode당 1회 main-thread precompute. OFF → {} (no-op byte-identical). scene_image_service.generate_images 의 단일 진입점. gate 체크포인트 + bg asset lineage + shot 맵 + provider(judge/qc) DI 를 조립해 build_indoor_shared_pose_context 를 generation_context(capture scope) 안에서 1회 호출한다. worker 는 결과 ctx 를 attach_indoor_pose_guide_ref 로 lookup 만. ★Codex 정렬(resume 비용/attach0 가드): ``already_done_still_ids`` 에 든 still 은 precompute 대상에서 제외한다 — resume 에서 이미 완료된 still 까지 guide/judge/QC 비용을 태우고도 batch 에서 걸러져 실제 attach 0 이 되는 것을 막는다. 단 candidate group 은 **선택된(미완료) 멤버가 1개라도 있으면** 그 그룹의 모든 멤버 staging 으로 그룹 신호를 구성해야 cross-shot 연속성이 성립하므로, selected_keys 는 미완료 still 로 좁히되 그룹 형성을 깨지 않게 fresh run(이 set 이 비었거나 None)은 전체 사용. """ from pathlib import Path from app.core.config import settings if not bool(getattr(settings, "indoor_shared_pose_guide_enabled", False)): return {} from app.modules.pipeline.indoor_shared_pose_provider import ( judge_indoor_pose_guide, make_indoor_qc_fn, ) from app.services.image_capture.context import generation_context from app.services.indoor_shared_pose_guide_service import build_indoor_pose_guide # shot 맵 — selected = 전체 still(그룹 형성·연속성 신호 보존), bg_id/staging 구조키 # lookup. skip_shot_keys = 이미 완료된 still(resume) → guide 생성만 제외(Codex 가드). done_ids = already_done_still_ids or set() selected_keys: List[ShotKey] = [] skip_shot_keys: set = set() bg_id_by_shot: Dict[ShotKey, str] = {} shot_by_key: Dict[ShotKey, Dict[str, Any]] = {} for s in stills: si, shi = s.get("scene_index"), s.get("shot_index") if not isinstance(si, int) or not isinstance(shi, int): continue key = (si, shi) selected_keys.append(key) if s.get("id") in done_ids: skip_shot_keys.add(key) elif (execution_allowlist_still_ids is not None and s.get("id") not in execution_allowlist_still_ids): # 표적 씬 슬라이스 (Codex 재리뷰 HIGH-4): 표적 밖 still 은 # guide/underlay 생성 대상에서 제외 — already_done 과 동일한 # skip 채널(그룹 형성·연속성 신호는 selected_keys 로 유지). # None=기존 byte-identical. skip_shot_keys.add(key) entry = (background_chain_bg_map or {}).get(f"{si}_{shi}") or {} bg_id_by_shot[key] = str(entry.get("bg_id") or "") shot_by_key[key] = (staging_map or {}).get(f"{si}_{shi}") or {} zoom_member_keys = set() for zkey in ((zoom_ctx or {}).get("zoom_targets") or {}): if isinstance(zkey, tuple) and len(zkey) == 2: zoom_member_keys.add(zkey) bg_loc_by_id = _load_bg_loc_by_id(settings.projects_dir, project_id, episode_id) indoor_locs, outdoor_locs = _load_indoor_outdoor_locs( settings.projects_dir, project_id, episode_id) bg_asset_by_id = _load_bg_asset_by_id(db, project_id, episode_id) # target_id(short_id C##) → entity name 매핑 — build_pose_brief 가 character_angles # 의 body_pose/gaze 를 FSC target_id 기준으로 robust 매칭(영문 label↔한글 name 회피). name_by_id = _load_name_by_short_id(db, project_id, episode_id) judge_model = getattr(settings, "indoor_shared_pose_guide_judge_model", "gemini-3.1-pro-preview") qc_model = getattr(settings, "indoor_shared_pose_guide_qc_model", "openai/gpt-6-astra") judge_enabled = bool(getattr(settings, "indoor_shared_pose_guide_judge_enabled", False)) guide_model = getattr(settings, "indoor_shared_pose_guide_model", "gpt-image-2.5-sunburst") # 4b(Codex): single-shot lane 은 상위 indoor_shared_pose_guide_enabled(위에서 이미 # 통과) AND 이 flag 둘 다 True 일 때만 broad 1-멤버 후보를 연다. single_shot_lane_enabled = bool(getattr( settings, "indoor_single_shot_pose_lane_enabled", False)) def _judge_fn(payload, log_context=None): return judge_indoor_pose_guide(payload, model=judge_model, log_context=log_context) qc_fn = make_indoor_qc_fn(model=qc_model) if judge_enabled else None cache_dir = (Path(settings.projects_dir) / project_id / "episodes" / episode_id / "images" / "indoor_pose_guide") with generation_context(project_id, episode_id, stage="indoor_pose_guide"): ctx = build_indoor_shared_pose_context( selected_keys=selected_keys, bg_id_by_shot=bg_id_by_shot, shot_by_key=shot_by_key, zoom_member_keys=zoom_member_keys, background_chain_bg_map=background_chain_bg_map, cache_dir=cache_dir, judge_fn=_judge_fn, guide_fn=build_indoor_pose_guide, qc_fn=qc_fn, judge_enabled=judge_enabled, bg_loc_by_id=bg_loc_by_id, indoor_locs=indoor_locs, outdoor_locs=outdoor_locs, bg_asset_by_id=bg_asset_by_id, skip_shot_keys=skip_shot_keys, name_by_id=name_by_id, openai_client=openai_client, model=guide_model, single_shot_lane_enabled=single_shot_lane_enabled, log_context={"project_id": project_id, "episode_id": episode_id, "step_name": "indoor_shared_pose_guide"}) # goal#3 (Codex 설계 A): generation_context flush 후 accepted guide row 의 # asset_id 를 구조키로 회수해 guide_by_shot 에 채운다(final scene input_image_ids # lineage 의 SOT). cache_hit 경로도 prior accepted row 를 같은 resolver 로 찾는다. _resolve_guide_asset_ids( db, project_id, episode_id, ctx.get("guide_by_shot") or {}, ctx.get("diagnostics") or []) n_guides = len(ctx.get("guide_by_shot") or {}) # 관측성 (2026-07-02): diag 가 ctx 안에서만 살다 버려져 guide 전멸(클라이언트 # 배선 오류 등)이 로그에 안 남던 결함 수정 — 실패류는 WARNING, 정책 deny 는 INFO. for _d in ctx.get("diagnostics") or []: _st = str(_d.get("status") or "") _line = ("indoor_shared_pose_guide diag: group=%s shot=%s status=%s reason=%s" % (_d.get("group_id"), _d.get("shot"), _st, _d.get("reason") or _d.get("qc_reason") or "")) if _st in ("failed", "qc_failed", "judge_error"): logger.warning(_line) else: logger.info(_line) logger.info( "indoor_shared_pose_guide: precompute 완료 — guides=%d, diagnostics=%d, " "candidates_indoor_gate(indoor=%d/outdoor=%d locs)", n_guides, len(ctx.get("diagnostics") or []), len(indoor_locs), len(outdoor_locs)) return ctx