"""scene_detail redo-shot service — 단건 shot 의 LLM 재호출 + cp/DB partial update. 2026-05-11 사용자 권장안 #2 축소형 (시간 절약 + cp/DB 무결성 보장): - service + API endpoint + thin CLI (HTTP wrapper). DB 직접 UPDATE 금지. - 운영 복구 도구 — scene_detail step 의 atomic-fail 대비 (runner shot-level typed failure 와 별개). 흐름: 1. SceneDetailStep instance + ctx load (SceneContextLoader.load_all()) 2. system / schema 빌드 (step._build_llm_inputs) 3. seg + sh 구축 (ctx.segments / ctx.shot_scenes_map lookup) 4. step._analyze_one(seg, sh, ctx, system, schema) 호출 5. 기존 manifest 읽고 해당 (scene_index, shot_index) result replace 6. atomic_write_json 으로 manifest 갱신 (in-place) — top-level status / completed_count / failed_count 가 per-shot 실제 상태를 반영. 7. checkpoint_sync.orchestrate_full_sync (per-step 부분 sync 충분) 8. step_run projection promote (W4b-2 + fixup): manifest 가 lightweight verify 를 통과 (per-shot contract_violation 0건 AND per-shot t2i_variations non-empty AND per-shot render_prompt_card + render_prompt_card_hash + sentinel shape valid) AND checkpoint_sync 가 성공한 경우에만 step_run.status='completed' + counts + error_message=NULL 로 갱신. 그 외엔 updated_at 만 touch (이전 status 보존). 이는 redo service 의 정상 projection 책임이며 manual SQL repair 가 아니다 — 이전엔 partial row 가 남아 다음 `scene_detail?mode=resume` 이 60 shot 전부 LLM 재실행 → 신선한 LLM 비결정으로 contract violation 재발하는 회귀 패턴 (이번 wave §7b W4b 핸드오프 + W4b-fixup Codex IMPORTANT 1+2 보강). W4b-fixup 검증 신호 (IMPORTANT 1): ``_verify_manifest_lightweight`` 는 ``SceneDetailStep.verify_completion`` 의 ctx-free 부분집합을 inline 재구현한다 — 4 signal: (a) data.scenes 존재 + non-empty. (b) per-shot t2i_variations non-empty (detail_steps.py:1822-1823). (c) per-shot status != "contract_violation" (detail_steps.py:1825-1826). (d) per-shot render_prompt_card / render_prompt_card_hash 존재 + card shape valid (detail_steps.py:1836-1848). (e) per-variation owned_validation sentinel 존재 + shape valid (detail_steps.py:1961-1972 ; reuse ``_owned_helpers.assert_owned_sentinel_shape``). Heavy ctx-dependent drift signals (camera_direction_hash recompute / owned_hash recompute against loader / card hash recompute against build_render_prompt_card) 는 service helper 가 처리하지 않는다 — 다음 ``scene_detail?mode=resume`` 의 StepRunner SKIP-path 의 verify_completion() 가 standard load_all() 위에서 검증한다. W4b-fixup promote precondition (IMPORTANT 2): ``step_run.status='completed'`` UPDATE 는 다음 셋이 모두 참일 때만 발사: 1. ``manifest_clean`` (per-shot contract_violation 0건 — manifest top-level status 도 'completed' 로 기록). 2. ``verify_clean`` (lightweight verify 의 4 signal 위반 0건). 3. ``checkpoint_sync_success`` (orchestrate_full_sync 가 raise 안 함 AND sync_summary 에 'error' key 부재). UPDATE rowcount != 1 (step_run row 부재 / 다중 매치) 는 promote 실패로 surface — response 에 ``step_run_promoted=False`` + ``step_run_promoted_reason`` 명시. 예외는 swallow 하지 않고 reason string 으로 보고된다. """ from __future__ import annotations import logging from typing import Any, Dict, List, Optional, Tuple from sqlalchemy.orm import Session as OrmSession from app.core.errors import AppError from app.services.checkpoint_sync import orchestrate_full_sync logger = logging.getLogger(__name__) def redo_scene_detail_shot( *, db: OrmSession, project_id: str, episode_id: str, scene_index: int, shot_index: int, mode: str = "force", ) -> Dict[str, Any]: """단건 shot scene_detail re-run. Args: db: 활성 SQLAlchemy session. project_id / episode_id: 대상 PID/EID. scene_index / shot_index: 재실행 대상 shot 의 좌표. mode: 현재 "force" 만 지원 (기존 결과 무관 강제 LLM 재호출). Returns: { "ok": bool, "scene_index": int, "shot_index": int, "replaced": bool, # manifest 내 result replace 성공 여부 "render_prompt_card_hash": str | None, "checkpoint_sync": dict | None, # orchestrate_full_sync 결과 요약 } Raises: AppError(code="scene_detail.redo.manifest_missing") — manifest 부재. AppError(code="scene_detail.redo.shot_not_in_segments") — 대상 shot 부재. AppError(code="scene_detail.redo.analyze_returned_none") — LLM 호출 실패. AppError — _analyze_one 안의 contract violation (Rule X-2 등) 그대로 propagate. """ if mode != "force": raise AppError( code="scene_detail.redo.invalid_mode", message=f"redo_scene_detail_shot: mode must be 'force', got {mode!r}", status_code=400, ) # 1) step instance + ctx from app.services.analysis_dispatch_service import get_step_runner from app.services.step_execution_service import _load_project_config from app.core.steps.detail_steps import SceneDetailStep from app.core.steps.scene_context_loader import SceneContextLoader from app.core.checkpoint_io import atomic_write_json project_config = _load_project_config(db, project_id) step = get_step_runner( "scene_detail", project_id, episode_id, db, project_config, ) if not isinstance(step, SceneDetailStep): raise AppError( code="scene_detail.redo.step_type_mismatch", message=( f"get_step_runner returned {type(step).__name__}, expected " f"SceneDetailStep" ), status_code=500, ) ctx = SceneContextLoader(step).load_all() # 2) system / schema (step._build_llm_inputs — _execute 와 동일 path) system, schema = step._build_llm_inputs(ctx) # 3) seg + sh 구축 seg = next( (s for s in (ctx.segments or []) if isinstance(s, dict) and s.get("scene_index") == scene_index), None, ) if seg is None: raise AppError( code="scene_detail.redo.scene_not_in_segments", message=( f"redo_scene_detail_shot: scene_index={scene_index} not found in " f"ctx.segments (episode={episode_id})" ), status_code=400, ) shots_in_scene = (ctx.shot_scenes_map or {}).get(scene_index) or [] sh = next( (s for s in shots_in_scene if isinstance(s, dict) and s.get("shot_index") == shot_index), None, ) if sh is None: raise AppError( code="scene_detail.redo.shot_not_in_segments", message=( f"redo_scene_detail_shot: shot_index={shot_index} not found in " f"scene={scene_index} (available={sorted({(s or {}).get('shot_index') for s in shots_in_scene})})" ), status_code=400, ) # 4) _analyze_one — contract violation 시 AppError raise (caller 가 전파) logger.info( "scene_detail redo-shot: project=%s episode=%s S%d_Shot%d (mode=%s)", project_id, episode_id, scene_index, shot_index, mode, ) result = step._analyze_one(seg, sh, ctx, system, schema) if result is None: raise AppError( code="scene_detail.redo.analyze_returned_none", message=( f"redo_scene_detail_shot: _analyze_one returned None for " f"S{scene_index}_Shot{shot_index} (LLM call failed or hard-fail " f"before raise — see backend log)" ), status_code=500, ) # 5) 기존 manifest 읽고 해당 shot result replace cp = step.load_checkpoint() if not cp: raise AppError( code="scene_detail.redo.manifest_missing", message=( f"redo_scene_detail_shot: scene_detail manifest 없음 " f"(project={project_id} episode={episode_id}). force_cleared " f"marker 또는 archive 부재. 정식 force re-run 필요." ), status_code=409, ) data = cp.get("data") or {} scenes = data.get("scenes") if not isinstance(scenes, list): raise AppError( code="scene_detail.redo.manifest_shape", message=( "redo_scene_detail_shot: manifest data.scenes not list — " f"got {type(scenes).__name__}" ), status_code=500, ) replaced = False for idx, existing in enumerate(scenes): if ( isinstance(existing, dict) and existing.get("scene_index") == scene_index and existing.get("_shot_index") == shot_index ): scenes[idx] = result replaced = True break if not replaced: # 본 shot 이 cp 에 없으면 append — applicable_count 와 어긋날 수 있으므로 # 로그 + append (manifest 가 stale 일 가능성). logger.warning( "scene_detail redo-shot: S%d_Shot%d not in existing manifest, " "appending (manifest may be stale)", scene_index, shot_index, ) scenes.append(result) # 6) atomic_write_json 으로 manifest 갱신 in-place. W4b-2: top-level # status / failed_count 는 per-shot 실제 상태에서 derive — 이전엔 # 무조건 status='completed', failed_count=0 으로 hard-code 되어 # contract_violation 잔존 shot 이 있는 dirty 상태에서도 manifest 가 # clean 으로 거짓 표시되었다 (downstream / 다음 resume 판단을 흐리는 # silent 거짓신호). dirty_shot_indices = [ (s.get("scene_index"), s.get("_shot_index")) for s in scenes if isinstance(s, dict) and s.get("status") == "contract_violation" ] dirty_count = len(dirty_shot_indices) manifest_clean = dirty_count == 0 manifest_path = step._cp_dir / "manifest.json" cp_to_write = dict(cp) cp_to_write["data"] = data cp_to_write["completed_count"] = len(scenes) cp_to_write["failed_count"] = dirty_count cp_to_write["status"] = "completed" if manifest_clean else "partial" atomic_write_json(manifest_path, cp_to_write) logger.info( "scene_detail redo-shot: manifest replaced (S%d_Shot%d, total=%d, " "dirty=%d, manifest_status=%s)", scene_index, shot_index, len(scenes), dirty_count, cp_to_write["status"], ) # 7) checkpoint_sync — DB 정합 (scene_still UPDATE 등). W4b-fixup # (IMPORTANT 2): 실패는 promote 차단 신호로 surface — sync_summary 에 # 'error' key 가 들어가면 verify_clean 과 무관하게 promote 발사 X. sync_summary: Optional[Dict[str, Any]] = None checkpoint_sync_success = False try: sync_summary = orchestrate_full_sync( project_id, episode_id, db, step_id="scene_detail", ) db.commit() checkpoint_sync_success = True except Exception as exc: db.rollback() logger.error( "scene_detail redo-shot: orchestrate_full_sync failed: %s", exc ) # sync 실패는 cp 변경에 영향 X (file 은 이미 갱신). caller 가 별도 # sync 호출. W4b-fixup: promote 차단을 위해 success=False 유지. sync_summary = {"error": str(exc)} checkpoint_sync_success = False # 7.5) W4b-fixup IMPORTANT 1 — lightweight verify of post-replace # manifest. ctx-free subset of SceneDetailStep.verify_completion (4 # signal). verify_clean=False 면 promote 차단. verify_clean, verify_reasons = _verify_manifest_lightweight( scenes=scenes, result_schema_version=cp_to_write.get( "schema_version", _get_scene_detail_schema_version() ), ) if not verify_clean: logger.warning( "scene_detail redo-shot: lightweight verify failed: %s", "; ".join(verify_reasons[:5]), ) # 8) step_run projection promote (W4b-2 + W4b-fixup). # # W4b-fixup IMPORTANT 2: promote precondition = # manifest_clean AND verify_clean AND checkpoint_sync_success # 위 셋이 모두 참일 때만 step_run.status='completed' UPDATE 발사. # 그 외엔 updated_at 만 touch (이전 status / error_message 보존) — # 다음 `scene_detail?mode=resume` 에서 StepRunner 가 정상 verify_completion # 을 다시 평가하도록 둔다. promote_eligible = ( manifest_clean and verify_clean and checkpoint_sync_success ) if promote_eligible: promote_result = _promote_step_run_projection( db=db, project_id=project_id, episode_id=episode_id, completed_count=cp_to_write["completed_count"], failed_count=cp_to_write["failed_count"], applicable_count=cp_to_write.get("applicable_count"), ) else: # not eligible → only touch updated_at, capture the reason. reason_parts: List[str] = [] if not manifest_clean: reason_parts.append( f"manifest dirty ({dirty_count} contract_violation shots)" ) if not verify_clean: reason_parts.append( "lightweight verify failed: " + "; ".join(verify_reasons[:3]) ) if not checkpoint_sync_success: reason_parts.append("checkpoint_sync failed") promote_result = _touch_step_run_updated_at( db=db, project_id=project_id, episode_id=episode_id, reason="; ".join(reason_parts) or "preconditions unmet", ) return { "ok": True, "scene_index": scene_index, "shot_index": shot_index, "replaced": replaced, "render_prompt_card_hash": result.get("render_prompt_card_hash"), "checkpoint_sync": sync_summary, "checkpoint_sync_success": checkpoint_sync_success, "manifest_clean": manifest_clean, "dirty_count": dirty_count, "dirty_shot_indices": dirty_shot_indices, "verify_clean": verify_clean, "verify_reasons": verify_reasons, "step_run_promoted": promote_result.get("promoted", False), "step_run_promoted_reason": promote_result.get("reason", ""), } def _get_scene_detail_schema_version() -> int: """Lazy SCENE_DETAIL_SCHEMA_VERSION import (avoid module-level cycle). The constant lives in ``app.core.steps.detail_steps`` and the redo service is imported indirectly by FastAPI endpoint wiring; module- level import of detail_steps here creates a cycle risk. Inline- import keeps the dependency cheap. """ from app.core.steps.detail_steps import SCENE_DETAIL_SCHEMA_VERSION return SCENE_DETAIL_SCHEMA_VERSION def _verify_manifest_lightweight( *, scenes: List[Any], result_schema_version: Any, ) -> Tuple[bool, List[str]]: """ctx-free subset of ``SceneDetailStep.verify_completion`` signals. W4b-fixup IMPORTANT 1: redo service can verify these signals against the in-memory manifest WITHOUT rebuilding ``SceneContextLoader`` ctx (no chain_bg_owned, no staging_map, no card-hash recompute, no camera_direction lookup). The heavier ctx-dependent drift checks (camera_direction_hash / owned_hash recompute against loader / card hash recompute via build_render_prompt_card) defer to the next ``scene_detail?mode=resume`` cycle, which runs the full ``verify_completion()`` through ``StepRunner._evaluate_resume_decision``. Signals (clean = all pass): (a) ``scenes`` is a non-empty list (detail_steps.py:1700-1705). (b) every shot result has non-empty ``t2i_variations`` (detail_steps.py:1822-1823). (c) every shot result has ``status != "contract_violation"`` (detail_steps.py:1825-1826). (d) when ``result_schema_version`` equals the current ``SCENE_DETAIL_SCHEMA_VERSION``, each shot has both ``render_prompt_card`` and ``render_prompt_card_hash`` present, and ``assert_card_shape`` does not raise (detail_steps.py:1836-1848). (e) every variation has ``owned_validation`` present and ``assert_owned_sentinel_shape`` does not raise (detail_steps.py:1961-1972). Args: scenes: post-replace manifest's ``data.scenes`` list (already replaced by the redo loop; trusted as up-to-date). result_schema_version: manifest's ``schema_version`` field (top- level). Used to gate card-shape check. Returns: (clean, reasons) — ``clean=True`` iff all signals pass. ``reasons`` is a list of short strings; first 5 are quoted into the log. """ from app.core.steps._owned_helpers import assert_owned_sentinel_shape from app.core.steps.render_prompt_card import assert_card_shape reasons: List[str] = [] # (a) scenes non-empty. if not isinstance(scenes, list) or not scenes: reasons.append("data.scenes empty or non-list") return False, reasons card_check_active = ( result_schema_version == _get_scene_detail_schema_version() ) for r in scenes: if not isinstance(r, dict): reasons.append(f"non-dict scene entry ({type(r).__name__})") continue si = r.get("scene_index") shi = r.get("_shot_index") # (b) t2i_variations non-empty. tvars = r.get("t2i_variations") or [] if not tvars: reasons.append(f"S{si}_Shot{shi}: empty t2i_variations") continue # (c) contract_violation status. if r.get("status") == "contract_violation": reasons.append(f"S{si}_Shot{shi}: status=contract_violation") # don't skip the per-variation sentinel checks — collect all # signals for richer debug surface. # (d) card present + shape valid (gated by schema_version). if card_check_active: stored_card = r.get("render_prompt_card") stored_card_hash = r.get("render_prompt_card_hash") if stored_card is None or stored_card_hash is None: reasons.append( f"S{si}_Shot{shi}: render_prompt_card/card_hash missing" ) else: try: assert_card_shape( stored_card, where=( f"redo_service.verify_lightweight " f"s{si}_sh{shi}" ), ) except Exception as exc: reasons.append( f"S{si}_Shot{shi}: render_prompt_card shape " f"invalid: {exc}" ) # (e) per-variation owned_validation sentinel. for var_idx, var in enumerate(tvars): if not isinstance(var, dict): reasons.append( f"S{si}_Shot{shi} var[{var_idx}]: non-dict variation" ) continue sentinel = var.get("owned_validation") if sentinel is None: reasons.append( f"S{si}_Shot{shi} var[{var_idx}]: owned_validation " "missing" ) continue try: assert_owned_sentinel_shape( sentinel, where=( f"redo_service.verify_lightweight " f"s{si}_sh{shi}_var{var_idx}" ), ) except Exception as exc: reasons.append( f"S{si}_Shot{shi} var[{var_idx}]: owned_validation " f"shape invalid: {exc}" ) return len(reasons) == 0, reasons def _promote_step_run_projection( *, db: OrmSession, project_id: str, episode_id: str, completed_count: int, failed_count: int, applicable_count: Optional[int], ) -> Dict[str, Any]: """step_run projection — promote to status='completed' (clean path). W4b-2 + W4b-fixup IMPORTANT 2: caller has already verified all promote preconditions (manifest_clean AND verify_clean AND checkpoint_sync_success). This function executes the UPDATE and returns a result dict so the caller can populate the response ``step_run_promoted`` field with the actual UPDATE outcome. Returns: { "promoted": bool, # True iff UPDATE affected exactly 1 row. "reason": str, # empty on success; explanation on failure. } Failure modes (promoted=False): - UPDATE rowcount != 1 (row missing or multiple matches — violates the (project_id, episode_id, step_id) unique key). - SQLAlchemy / DB exception during execute. raw SQL — step_run 은 ORM 모델 미정의 (database.py:64 주석 참조). rowcount 는 ``CursorResult.rowcount`` 로 확인. 실패 시 rollback, reason string 으로 surface (warning log 도 함께). """ from datetime import datetime, timezone from sqlalchemy import text from sqlalchemy.exc import SQLAlchemyError now_iso = datetime.now(timezone.utc).isoformat() params: Dict[str, Any] = { "now": now_iso, "completed_count": completed_count, "failed_count": failed_count, "pid": project_id, "eid": episode_id, } sql = ( "UPDATE step_run SET " "status = 'completed', " "updated_at = :now, " "completed_at = :now, " "completed_count = :completed_count, " "failed_count = :failed_count, " "error_message = NULL" ) if applicable_count is not None: sql += ", applicable_count = :applicable_count" params["applicable_count"] = applicable_count # ★도는 중인 scene_detail 은 건드리지 않는다 (2026-08-26 Codex 재리뷰 # BLOCK-4). 주인이 누구든 completed 로 갈아 끼우면, 돌고 있던 주행이 # 자기 결과를 적을 때 이 값을 덮거나 반대로 이 값이 그 주행의 진행을 # 지운다. 한 샷 다시 만들기는 스텝이 **끝난 뒤에** 하는 일이다. sql += ( " WHERE project_id = :pid AND episode_id = :eid " "AND step_id = 'scene_detail' AND status <> 'running'" ) try: cursor = db.execute(text(sql), params) rowcount = getattr(cursor, "rowcount", None) if rowcount != 1: db.rollback() busy = db.execute(text( "SELECT status FROM step_run " "WHERE project_id = :pid AND episode_id = :eid " " AND step_id = 'scene_detail'" ), {"pid": project_id, "eid": episode_id}).fetchone() if busy and busy[0] == "running": reason = ( f"scene_detail 이 실행 중이라 승격하지 않았다 " f"(project={project_id} episode={episode_id}) — " "스텝이 끝난 뒤 다시 시도할 것" ) else: reason = ( f"step_run row not found (UPDATE rowcount={rowcount}) for " f"project={project_id} episode={episode_id} " f"step_id='scene_detail'" ) logger.warning("scene_detail redo-shot: %s", reason) return {"promoted": False, "reason": reason} db.commit() logger.info( "scene_detail redo-shot: step_run promoted to " "status='completed' (completed=%d, failed=%d) project=%s " "episode=%s", completed_count, failed_count, project_id, episode_id, ) return {"promoted": True, "reason": ""} except SQLAlchemyError as exc: db.rollback() reason = f"DB exception during UPDATE: {type(exc).__name__}: {exc}" logger.warning("scene_detail redo-shot: %s", reason) return {"promoted": False, "reason": reason} def _touch_step_run_updated_at( *, db: OrmSession, project_id: str, episode_id: str, reason: str, ) -> Dict[str, Any]: """step_run.updated_at only-touch path (precondition unmet). W4b-fixup IMPORTANT 2: when promote preconditions are not all met (manifest dirty / verify dirty / sync failed), we still touch ``updated_at`` so downstream observability sees the redo attempt, but preserve the prior ``status`` and ``error_message`` so the next ``scene_detail?mode=resume`` will re-evaluate. Returns the same shape as ``_promote_step_run_projection`` with ``promoted=False``. Args: reason: human-readable string explaining why promote was skipped (composed by caller from the unmet preconditions). """ from datetime import datetime, timezone from sqlalchemy import text from sqlalchemy.exc import SQLAlchemyError now_iso = datetime.now(timezone.utc).isoformat() try: # 승격과 같은 이유로 도는 중이면 건드리지 않는다. cursor = db.execute( text( "UPDATE step_run SET updated_at = :now " "WHERE project_id = :pid AND episode_id = :eid " "AND step_id = 'scene_detail' AND status <> 'running'" ), { "now": now_iso, "pid": project_id, "eid": episode_id, }, ) db.commit() # ★결과를 읽어야 한다 (2026-08-26 Codex 2차 재리뷰 IMPORTANT). # 도는 중이면 0줄인데, 안 읽으면 「touched」라고 적고 돌려준다 — # 아무 일도 안 했는데 했다고 보고하는 셈이다. if getattr(cursor, "rowcount", None) == 0: skipped = ( f"{reason}; updated_at touch skipped — scene_detail 이 " f"실행 중이거나 행이 없다" ) logger.warning("scene_detail redo-shot: %s", skipped) return {"promoted": False, "reason": skipped} logger.info( "scene_detail redo-shot: step_run promote skipped (%s) — " "updated_at touched, prior status preserved. project=%s " "episode=%s", reason, project_id, episode_id, ) return {"promoted": False, "reason": reason} except SQLAlchemyError as exc: db.rollback() detail_reason = ( f"{reason}; updated_at touch also failed: " f"{type(exc).__name__}: {exc}" ) logger.warning( "scene_detail redo-shot: step_run touch failed: %s", detail_reason, ) return {"promoted": False, "reason": detail_reason}