"""StepRunner 기반 분석 dispatch 서비스.

Phase 4.1 (architecture-refactor-final/02-final-roadmap.md §Phase 4).
기존 `AnalysisService.run_analysis/reanalyze_scenes` 경로와 `steps.run_all_steps`의
`_run_all_bg` 내부 로직을 단일 dispatch 진입점으로 통합.

공식 경로: Frontend / API / 내부 호출 전부 StepRunner (→ StepCatalog) 경유.
"""
from __future__ import annotations

import json
import logging
import uuid
from datetime import datetime, timezone
from typing import Any, Dict, List, Optional, Tuple

from sqlalchemy.orm import Session as OrmSession

from app.core.database import SessionLocal
from app.core.errors import AppError
from app.core.image_call_budget import (
    ImageCallBudget,
    install_budget,
    uninstall_budget,
)
from app.core.job_manager import submit_background_job
from app.core.step_catalog import (
    STEP_CATALOG,
    get_active_entries,
    get_all_downstream_recursive,
)
from app.services.checkpoint_sync import orchestrate_full_sync

logger = logging.getLogger(__name__)


# W1-F12: projection 사전 sync 필요 여부는 manifest 필드 `requires_projection_sync_before_run`
# 로 승격. `_MID_SYNC_TRIGGER_STEPS` 하드코딩은 제거되고 step_manifest 조회로 전환.
# (이전 값: ("scene_director",))


# run_all 병행 차단에 쓰는 category 토큰 — run_step, dispatch_category_run,
# dispatch_scene_reanalysis가 공유. "reanalyze"는 /reanalyze-scenes 경로의 job key.
RUN_ALL_CATEGORY_TOKENS: Tuple[str, ...] = ("analysis", "image", "all", "reanalyze")


def build_opik_context(
    db: OrmSession, project_id: str, episode_id: str
) -> Dict[str, str]:
    """Opik thread 그룹핑용 context 생성."""
    from app.models.catalog import ProjectRegistry
    from app.models.project import Episode

    project = (
        db.query(ProjectRegistry).filter(ProjectRegistry.id == project_id).first()
    )
    episode = (
        db.query(Episode)
        .filter(Episode.id == episode_id, Episode.project_id == project_id)
        .first()
    )
    project_name = project.name if project else project_id[:8]
    episode_title = episode.title if episode else episode_id[:8]
    ts = datetime.now(timezone.utc).strftime("%m%d-%H%M%S")
    run_tag = f"{project_name}_{episode_title}_{ts}_{str(uuid.uuid4())[:8]}"
    return {
        "project_name": project_name,
        "episode_title": episode_title,
        "run_tag": run_tag,
    }


def load_project_llm_config(db: OrmSession, project_id: str) -> Dict[str, Any]:
    """ProjectSettings.llm_config_json 로드 (없거나 손상 시 빈 dict)."""
    from app.models.project import ProjectSettings

    ps = (
        db.query(ProjectSettings)
        .filter(ProjectSettings.project_id == project_id)
        .first()
    )
    if not ps or not ps.llm_config_json:
        return {}
    try:
        return json.loads(ps.llm_config_json)
    except json.JSONDecodeError:
        logger.warning(
            "Malformed llm_config_json for project=%s — falling back to defaults",
            project_id,
        )
        return {}


def _project_has_planning_doc(db: OrmSession, project_id: str) -> bool:
    """text/PDF/checkpoint 중 하나라도 있으면 True (단일 source 위임).

    PDF-only 업로드 후 text 추출 짧은 경우에도 True 반환되어 dispatcher 가
    planning_doc_analysis step 을 cascade 에 포함시킴 (review IMPORTANT 통일).
    """
    from app.services.planning_doc_analysis_service import project_has_planning_doc

    return project_has_planning_doc(project_id, db=db)


def _is_step_completed(
    db: OrmSession, project_id: str, episode_id: str, step_id: str
) -> bool:
    """step_run.status == 'completed' 여부 판정.

    Fix 3.3 prerequisite 검증용. partial은 아직 완료 미달로 처리하여 사용자에게
    명확한 에러를 노출 — silent transitive invalidate 보다 안전한 정책.
    """
    from sqlalchemy import text as _sql_text

    row = db.execute(
        _sql_text(
            "SELECT status FROM step_run "
            "WHERE project_id = :pid AND episode_id = :eid AND step_id = :sid"
        ),
        {"pid": project_id, "eid": episode_id, "sid": step_id},
    ).fetchone()
    if not row:
        return False
    return row[0] == "completed"


def _is_step_satisfied(
    db: OrmSession,
    project_id: str,
    episode_id: str,
    step_id: str,
) -> bool:
    """prerequisite 검증용: step이 완료되었거나 applicability로 skip 가능한지.

    Codex P1-1: image preflight applicability 무시 사고 fix.
        background_mode=='off' (default) 환경에서 background_classify /
        floor_plan_prompt / background_prompt 등은 applicability='if_background_mode'
        → 정적 평가 시 not_applicable. StepRunner는 자동 skip 처리.
        prerequisite 검증이 이를 무시하고 _is_step_completed=False로만 판정하면
        category='image' force가 dispatch.deps_incomplete로 차단되는데, 실제 실행
        시에는 StepRunner가 이를 skip하고 downstream gate를 통과시키므로 차단은 잘못.

    판정 로직:
        1) step_run.status='completed'  → True (이미 완료)
        2) applicability 정적 평가 결과 not_applicable → True (StepRunner skip 가능)
        3) 그 외 → False (prerequisite 미충족 — 사용자에게 명확한 에러 안내)
    """
    if _is_step_completed(db, project_id, episode_id, step_id):
        return True
    from app.core.applicability import evaluate_step_applicability
    return evaluate_step_applicability(step_id, project_id, episode_id) == "not_applicable"


def _check_force_cascade_risk(
    direct: List[str],
    direct_set: set,
    active_by_id: Dict[str, Any],
    mode: str,
    category: str,
) -> None:
    """Fix D (cascade fix 2026-05-05): mode=force 시 invalidate_downstream 이
    cross-category step 무효화할 risk 를 시작 시점에 거절.

    StepRunner.run() 의 force 동작이 invalidate_downstream(delete_cp=True) 호출 →
    transitive downstream manifest 모두 delete. batch 안에 없는 cross-category
    step 도 delete 됨 → batch 진행 중 mid-cascade missing 발생 (e.g.
    floor_plan_render(image) force → background_prompt(analysis) manifest delete →
    background_render 시점 fail; scene_detail(analysis) force → scene_image_pipeline
    (image) manifest delete → 후속 image batch 가 stale).

    검증: 각 batch step 의 transitive downstream 중 batch 외 active step 의 category
    가 현재 category 와 다르면 risky. 시작 시점 fail-fast + 사용자에게 `category=all
    &mode=force` 또는 `category=<X>&mode=resume` 안내.

    Args:
        direct: batch 에 포함된 step_id list.
        direct_set: direct 의 set (transitive downstream 검사 시 batch 내 step skip).
        active_by_id: step_id → entry. caller 에서 한 번 build 후 재사용.
        mode: "resume" 또는 "force". "resume" 면 검증 skip.
        category: 현재 batch category ("analysis" / "image"). non-matching downstream 만 risky.

    Raises:
        AppError(dispatch.force_cascade_cross_category) — risky 발견 시.
    """
    if mode != "force":
        return
    cross_downstream: List[Tuple[str, str]] = []
    for sid in direct:
        for dsid in get_all_downstream_recursive(sid):
            if dsid in direct_set:
                continue  # batch 안 — 같이 재실행됨
            ds_entry = active_by_id.get(dsid)
            if not ds_entry:
                continue  # 비활성/replaced — 적용 대상 아님
            if ds_entry.category != category:
                cross_downstream.append((sid, dsid))
    if cross_downstream:
        cross_steps = sorted({dsid for _, dsid in cross_downstream})
        raise AppError(
            code="dispatch.force_cascade_cross_category",
            message=(
                f"category={category!r} mode=force 가 cross-category step "
                f"무효화 위험: {cross_steps}. `category=all&mode=force` 또는 "
                f"`category={category}&mode=resume` 사용 권장."
            ),
            status_code=400,
        )


def select_steps_for_category(
    db: OrmSession,
    project_id: str,
    category: str,
    *,
    episode_id: Optional[str] = None,
    mode: str = "resume",
) -> List[str]:
    """카테고리(analysis/image/all)에 해당하는 실행 가능 step_id 목록.

    applicability(on_demand/disabled/if_planning_doc)를 정적 필터만 수행.
    이외 applicability는 StepRunner 런타임(resolve_applicability)에서 처리.

    Fix 3.3 (M4 해소): category="image" 동작 변경.
    이전: image step의 transitive deps(analysis 포함) 자동 포함 — 사용자가 의도한 "image만 force"
          가 실제로 거의 모든 step force로 폭발 (오늘 13:13 사고 trigger).
    현재: image step만 직접 반환. analysis dep이 미완료/stale이면 prerequisite 검증으로 거부
          (AppError code="dispatch.deps_incomplete"). 사용자가 다음 행동(category=all 또는 해당
          analysis step 먼저 실행)을 명시적으로 선택. silent miss는 prerequisite 에러로 대체.
    prerequisite 검증은 episode_id가 제공된 경우에만 적용 (preview/list 호출은 episode 없음).

    Fix D (cascade fix 2026-05-05): `mode="force"` 시 force-invalidate 가
    cross-category downstream step 을 무효화할 risk 를 시작 시점에 거절.
    StepRunner.run() 의 force 동작이 invalidate_downstream(delete_cp=True) 를
    호출 — 만약 image step 의 transitive downstream 이 active analysis step 을
    포함하면 batch 진행 중 mid-cascade missing 발생 (e.g. floor_plan_render force
    → background_prompt manifest delete → background_render 시점 fail).
    """
    if category not in ("analysis", "image", "all"):
        raise AppError(
            code="dispatch.invalid_category",
            message=f"Unknown category: {category}",
            status_code=400,
        )
    # Fix D iter 1 (Claude I1): mode enum strict check — typo/None silent skip 차단.
    if mode not in ("resume", "force"):
        raise AppError(
            code="dispatch.invalid_mode",
            message=f"mode must be 'resume' or 'force' (got {mode!r})",
            status_code=400,
        )

    has_planning = _project_has_planning_doc(db, project_id)

    # 직접 매칭 (해당 category + applicable)
    direct: List[str] = []
    for entry in get_active_entries():
        if category != "all" and entry.category != category:
            continue
        appl = entry.applicability
        if appl in ("on_demand", "disabled"):
            continue
        if appl == "if_planning_doc" and not has_planning:
            continue
        direct.append(entry.step_id)

    from app.core.step_manifest import get_manifest_dict

    if category != "image":
        # category="all": 모든 step 직접 포함 — auto-include 불필요.
        # category="analysis": cross-category(image) prerequisite 검증 (Phase 7 cascade fix).
        #   Phase 7 도입 후 analysis step (background_prompt, scene_detail 등) 이
        #   image step (floor_plan_render, background_render) 에 depend → analysis
        #   단독 force 시 cascade skip + episode.status='analyzing' stuck 사고 발생.
        #   사용자에게 category='all' 명시적 안내.
        if category == "analysis" and episode_id is not None:
            active_by_id_a = {e.step_id: e for e in get_active_entries()}
            direct_set_a = set(direct)
            cross_deps: List[Tuple[str, str]] = []
            for sid in direct:
                meta = get_manifest_dict(sid) or {}
                for dep in meta.get("depends_on", []):
                    if dep in direct_set_a:
                        continue
                    dep_entry = active_by_id_a.get(dep)
                    if not dep_entry:
                        continue  # 비활성/replaced — 적용 대상 아님
                    if dep_entry.category != "analysis":
                        if not _is_step_satisfied(db, project_id, episode_id, dep):
                            cross_deps.append((sid, dep))
            if cross_deps:
                missing = sorted({d for _, d in cross_deps})
                raise AppError(
                    code="dispatch.deps_incomplete",
                    message=(
                        f"analysis step 실행 전 cross-category prerequisite step "
                        f"미완료입니다: {missing}. `category=all` 사용 권장."
                    ),
                    status_code=400,
                )
            # Fix D iter 1 (Codex BLOCKING / Claude I2): analysis 분기 force-cascade 대칭.
            # `mode='force'` 시 analysis step 의 transitive downstream 이 image
            # category 를 포함하면 invalidate cascade 가 image manifest 무효화 →
            # batch 끝난 뒤 image step 이 stale state 로 남음.
            _check_force_cascade_risk(
                direct, direct_set_a, active_by_id_a, mode, "analysis"
            )
        return sorted(direct, key=lambda s: (get_manifest_dict(s) or {}).get("order", 99))

    # category="image": prerequisite 검증 — image step의 cross-category(analysis) dep가
    # 모두 완료여야 진행. 같은 image category dep은 batch 내에서 순차 실행되므로 제외.
    # episode_id 미제공(과거 호출자/preview) 시 검증 skip — 호출 경로별 책임 분리.
    #
    # Codex P1-1 fix: dep의 applicability 정적 평가 결과 not_applicable이면 통과.
    # 이전: dep_appl 문자열만 검사 (`if_planning_doc` 만 special-case)
    #         → background_mode=='off' 환경에서 if_background_mode dep이 검증을 통과하지 못함.
    # 현재: _is_step_satisfied가 (1) status=completed 또는 (2) 정적 applicability=not_applicable
    #         둘 중 하나면 통과. StepRunner의 실제 skip 동작과 일관.
    if episode_id is not None:
        active_by_id = {e.step_id: e for e in get_active_entries()}
        direct_set = set(direct)
        incomplete_deps: List[Tuple[str, str]] = []
        for sid in direct:
            meta = get_manifest_dict(sid) or {}
            for dep in meta.get("depends_on", []):
                # 같은 dispatch에 포함된 image dep은 batch 순차 실행 — 제외.
                if dep in direct_set:
                    continue
                dep_entry = active_by_id.get(dep)
                if not dep_entry:
                    continue  # 비활성/replaced — 적용 대상 아님
                # 정적 applicability 평가 + DB 완료 여부 통합 판정.
                if not _is_step_satisfied(db, project_id, episode_id, dep):
                    incomplete_deps.append((sid, dep))
        if incomplete_deps:
            missing = sorted({d for _, d in incomplete_deps})
            raise AppError(
                code="dispatch.deps_incomplete",
                message=(
                    f"image step 실행 전 다음 prerequisite step들이 미완료입니다: {missing}. "
                    f"`category=all` 사용 또는 해당 step을 먼저 실행하세요."
                ),
                status_code=400,
            )

    # Fix D (cascade fix 2026-05-05): mode=force + image batch 시 force-invalidate
    # 가 cross-category(analysis) downstream step 을 무효화할 risk 를 사전 거절.
    # 사용자에게 `category=all&mode=force` 또는 `category=image&mode=resume` 사용 안내.
    if mode == "force" and episode_id is not None:
        # M1 (Claude): active_by_id 통합 — image 분기 prereq 검증 시 이미 build,
        # 이번 force-cascade 검증에서 재사용. 한 번만 build.
        _check_force_cascade_risk(direct, direct_set, active_by_id, mode, "image")

    # 토폴로지 정렬 (image step끼리 order 기준).
    ordered = sorted(direct, key=lambda s: (get_manifest_dict(s) or {}).get("order", 99))
    return ordered


def select_scene_reanalysis_steps(db: OrmSession, project_id: str) -> List[str]:
    """씬 재분석(reanalyze-scenes) 대상 step 목록.

    엔티티(entity_* / outlook_*)는 보존하고 씬 관련 step만 force 재실행.
    `scene_segmentation`부터 시작 — baseline `AnalysisService.reanalyze_scenes`가
    `segments.json`을 삭제하여 재추출한 효과를 StepRunner force 모드로 재현.
    (Codex Review Important #2)
    """
    # scene_segmentation을 루트로 하여 downstream 전체를 재실행
    downstream = set(get_all_downstream_recursive("scene_segmentation"))
    downstream.add("scene_segmentation")

    has_planning = _project_has_planning_doc(db, project_id)

    # get_active_entries(category="analysis")는 이미 lifecycle=active + category=analysis 필터.
    # scene_segmentation∪downstream 교집합 + applicability 필터만 추가.
    result: List[str] = []
    for entry in get_active_entries(category="analysis"):
        if entry.step_id not in downstream:
            continue
        if entry.applicability in ("on_demand", "disabled"):
            continue
        if entry.applicability == "if_planning_doc" and not has_planning:
            continue
        result.append(entry.step_id)
    return result


def get_step_runner(
    step_id: str,
    project_id: str,
    episode_id: str,
    db: OrmSession,
    project_config: Dict[str, Any],
    opik_context: Optional[Dict[str, str]] = None,
):
    """Step ID → StepRunner 인스턴스."""
    from app.core.steps import STEP_CLASSES as V3_CLASSES

    try:
        from app.core.steps.image_steps import IMAGE_STEP_CLASSES
    except ImportError:
        IMAGE_STEP_CLASSES = {}

    step_classes = {**IMAGE_STEP_CLASSES, **V3_CLASSES}

    cls = step_classes.get(step_id)
    if not cls:
        raise AppError(
            code="step.not_found",
            message=f"Step class not found: {step_id}",
            status_code=404,
        )

    return cls(
        step_id=step_id,
        project_id=project_id,
        episode_id=episode_id,
        db=db,
        project_config=project_config,
        opik_context=opik_context,
    )


def _recover_episode_status_on_failure(
    db: OrmSession, episode_id: str, error_message: str
) -> None:
    """파이프라인 실패 시 episode.status 복구 — 'analyzing' 잠금 해제.

    Review Critical #1: run_steps_batch 실패 시 status가 'analyzing'에 영구 잠기면
    UI에서 분석 진행 중으로 오인되며 재실행이 차단된다.
    baseline AnalysisService.run_analysis의 except 블록과 동일한 의도.
    """
    from app.models.project import Episode

    try:
        ep = db.query(Episode).filter(Episode.id == episode_id).first()
        if ep and ep.status == "analyzing":
            ep.status = "error"
            ep.analysis_error = error_message[:2000]
            db.commit()
    except Exception as rec_exc:
        logger.error("Failed to recover episode status: %s", rec_exc)
        try:
            db.rollback()
        except Exception:
            # 의도적: recovery 중 rollback 실패는 best-effort. 세션이 이미 끊어졌어도 상위 호출자는 계속 진행.
            pass


def evaluate_image_call_budget_gate(
    *,
    category: str,
    image_call_cap: Optional[int],
    approve_image_generation: bool,
) -> Optional[ImageCallBudget]:
    """W20E5 — decide whether the worker thread needs an image-call budget.

    Returns the :class:`ImageCallBudget` instance the worker should install
    for the duration of the run, or ``None`` when no budget is required.

    Behaviour:

    * ``category`` not in ``{"image", "all"}`` → ``None`` (no image steps).
    * ``settings.background_render_reference_mode != "shot_aware_plan"``
      (i.e. legacy / w18j_overlap) → ``None`` unless the caller explicitly
      passes a ``image_call_cap``; in that case the cap is honoured as an
      advisory opt-in (no approval required).
    * Shot-aware image path → fail closed. ``approve_image_generation``
      must be ``True`` and ``image_call_cap`` must be ``>= 1``; otherwise
      :class:`AppError` (``step.image_generation_not_approved`` /
      ``step.image_call_cap_required``) is raised before any worker is
      submitted.
    """
    if category not in ("image", "all"):
        return None

    from app.core.config import settings as _settings

    render_mode = getattr(_settings, "background_render_reference_mode", "legacy")
    if render_mode != "shot_aware_plan":
        # Non-W20 image path — preserve the existing pre-W20E5 contract.
        # Caller may still opt in to a cap if they want; otherwise the
        # call sites remain uncapped (legacy behaviour, no regression).
        if image_call_cap is None:
            return None
        cap_int = max(0, int(image_call_cap))
        return ImageCallBudget(cap=cap_int)

    # W20 shot-aware image path — fail closed.
    if not approve_image_generation:
        raise AppError(
            code="step.image_generation_not_approved",
            message=(
                "W20 shot-aware image phase requires explicit "
                "approve_image_generation=true."
            ),
            status_code=400,
        )
    if image_call_cap is None or int(image_call_cap) < 1:
        raise AppError(
            code="step.image_call_cap_required",
            message=(
                "W20 shot-aware image phase requires image_call_cap >= 1."
            ),
            status_code=400,
        )
    return ImageCallBudget(cap=int(image_call_cap))


def run_steps_batch(
    project_id: str,
    episode_id: str,
    step_ids: List[str],
    run_mode: str,
    project_config: Dict[str, Any],
    opik_context: Dict[str, str],
    budget: Optional[ImageCallBudget] = None,
) -> None:
    """step_ids 순차 실행 — background job 내부에서 호출되는 실제 워커.

    각 step 완료 후 sync projection(체크포인트→DB). 실패/partial 처리 포함.
    실패 시 episode.status를 'analyzing' → 'error'로 복구.

    W20E5 — ``budget`` (optional) is installed on the worker thread for the
    duration of the run and cleared in ``finally`` so it cannot leak to
    other jobs. When ``None``, the existing uncapped behaviour applies.
    """
    if budget is not None:
        install_budget(budget)
    db = SessionLocal()
    failure_message: Optional[str] = None
    try:
        from app.core.step_manifest import get_manifest_dict

        all_ok = True
        # 회귀 가드 (Phase 9.2 — scene_image_pipeline 배경 ref 없이 60장 생성 사고):
        # blocked step silent skip이 downstream을 그대로 진행시켜 잘못된 결과.
        # blocked_steps 추적 → downstream의 depends_on에 포함되면 자동 cascade skip.
        #
        # 주의 — failed/partial는 별도 cascade 안 함 (의도된 동작):
        #   failed: 즉시 break로 batch 중단 → 다음 step 실행 안 됨
        #   partial: continue 진행 — step_runner gate(applicability)가 partial을
        #            통과로 인정 (downstream이 부분 결과 사용 가능 가정)
        blocked_steps: set[str] = set()
        for sid in step_ids:
            try:
                step_meta = get_manifest_dict(sid) or {}

                # Cascade 검사: 선행 step blocked면 자동 skip (silent miss 방지).
                blocked_deps = [d for d in step_meta.get("depends_on", []) if d in blocked_steps]
                if blocked_deps:
                    logger.warning(
                        "Step %s skipped (cascade) — upstream blocked: %s",
                        sid, blocked_deps,
                    )
                    blocked_steps.add(sid)
                    continue

                # W1-F12: manifest 필드 조회. 각 step이 DB projection 선행 필요 시 pre-sync.
                if step_meta.get("requires_projection_sync_before_run", False):
                    try:
                        orchestrate_full_sync(project_id, episode_id, db)
                    except Exception as sync_exc:
                        # Codex W1 High 2: flagged step presync 실패는 non-fatal이 아님.
                        # 잘못된 DB 상태로 step을 실행하면 조용히 잘못된 결과가 나온다.
                        logger.error(
                            "Required pre-sync failed before %s: %s — aborting batch",
                            sid,
                            sync_exc,
                        )
                        db.rollback()
                        all_ok = False
                        failure_message = f"Pre-sync failed before {sid}: {sync_exc}"
                        break

                runner = get_step_runner(
                    sid, project_id, episode_id, db, project_config, opik_context
                )
                result = runner.run(mode=run_mode)
                status = result.get("status", "done")
                logger.info("Step %s: %s", sid, status)

                # post-sync (2026-08-04): projection 원천 step 은 **실행 후에도**
                # DB 에 반영한다. 단일 step 경로는 이미 그렇게 한다
                # (`step_execution_service.py:111`) — 배치 경로에만 없어서 구멍이
                # 났다.
                #
                # 실측: resume 시작 시점의 pre-sync 때 `scene_detail` CP 가 직전
                # 실패 상태라 `is_cp_syncable` False 로 스킵됐고, 그 뒤 255/255
                # completed 됐지만 flagged step 이 모두 소진돼 sync 가 다시 돌
                # 기회가 없었다 → `scene_still` 영구 0행 (하류 이미지가 붙을
                # 대상 없음). status 무관하게 돌린다 — failed 여도 partial 로
                # 보존된 성공분은 반영되어야 한다.
                if step_meta.get("requires_projection_sync_before_run", False):
                    try:
                        orchestrate_full_sync(project_id, episode_id, db, step_id=sid)
                    except Exception as sync_exc:
                        # pre-sync 와 달리 abort 하지 않는다 — step 은 이미 끝났고
                        # 단일 step 경로도 로그만 남긴다. 다만 조용히 넘기면 같은
                        # 구멍이 되므로 error 로 드러낸다.
                        logger.error(
                            "Post-step sync failed after %s — DB projection may "
                            "be stale: %s", sid, sync_exc,
                        )
                        db.rollback()
                if status == "failed":
                    logger.error("Step %s failed, stopping pipeline", sid)
                    all_ok = False
                    failure_message = f"Step {sid} failed"
                    break
                elif status == "partial":
                    # partial cascade contract (problems.md #11):
                    # step 별 allow_partial_downstream 명시 False 면 batch 중단.
                    # default True 는 backward-compat — 기존 동작 유지.
                    if step_meta.get("allow_partial_downstream", True) is False:
                        logger.error(
                            "Step %s partial and allow_partial_downstream=False — stopping pipeline",
                            sid,
                        )
                        all_ok = False
                        failure_message = (
                            f"Step {sid} partial (allow_partial_downstream=False)"
                        )
                        break
                    logger.warning(
                        "Step %s partial — continuing (downstream may use partial results)",
                        sid,
                    )
            except AppError as ae:
                if ae.code == "gate.blocked":
                    # 회귀 가드: blocked → blocked_steps 추적 → downstream cascade.
                    logger.warning(
                        "Step %s blocked: %s — downstream depending on this will cascade-skip",
                        sid, ae.message,
                    )
                    blocked_steps.add(sid)
                    continue
                logger.error("Step %s error: %s", sid, ae.message)
                all_ok = False
                failure_message = f"Step {sid}: {ae.message}"
                break
            except Exception as exc:
                logger.error("Step %s crashed: %s", sid, exc)
                all_ok = False
                failure_message = f"Step {sid} crashed: {exc}"
                break

        # 후처리: blocked가 발생했으면 사용자에게 명시 알림 (silent miss 방지).
        if blocked_steps:
            logger.error(
                "Pipeline finished with %d blocked step(s): %s. Downstream skipped via cascade. "
                "Hint: missing dependencies may be in another category — try category='all'.",
                len(blocked_steps), sorted(blocked_steps),
            )

        if all_ok:
            try:
                orchestrate_full_sync(project_id, episode_id, db)
            except Exception as sync_exc:
                logger.error("Final sync failed: %s", sync_exc)
                db.rollback()
                _recover_episode_status_on_failure(
                    db, episode_id, f"Final sync failed: {sync_exc}"
                )
        else:
            _recover_episode_status_on_failure(
                db, episode_id, failure_message or "Pipeline failed"
            )
    except Exception as outer_exc:
        logger.error("run_steps_batch outer error: %s", outer_exc)
        try:
            db.rollback()
        except Exception:
            # 의도적: outer error 복구 rollback 실패는 best-effort. _recover가 자체적으로 새 세션을 쓰므로 계속 진행.
            pass
        _recover_episode_status_on_failure(db, episode_id, str(outer_exc))
    finally:
        db.close()
        if budget is not None:
            uninstall_budget()


_ANALYSIS_CATEGORIES = ("analysis", "all")


def preflight_analysis_start(
    db: OrmSession,
    project_id: str,
    episode_id: str,
) -> Optional[str]:
    """분석 시작 preflight — /analyze와 /steps/run-all 공용 (Codex W1 Critical 수정).

    1. Episode row lock
    2. status != "analyzing" 검증 (이미 실행 중이면 409)
    3. fulltext 존재 검증 (없으면 400)
    4. OPENAI_API_KEY 설정 검증 (없으면 400)
    5. status='analyzing', analysis_error=None set + commit (race window 종료)

    Returns:
        prior_status — dispatch 예외 시 복구에 사용. 호출자가 rollback 책임.
    """
    from app.core.config import settings as _settings
    from app.i18n.loader import t
    from app.models.project import Episode

    episode = (
        db.query(Episode)
        .filter(Episode.id == episode_id, Episode.project_id == project_id)
        .with_for_update()
        .first()
    )
    if not episode:
        raise AppError(code="episode.not_found", message=t("episode.not_found"), status_code=404)

    stale_recovered = False
    if episode.status == "analyzing":
        # Stale 'analyzing' lock 자동 회복: task_registry 에 active background job 이
        # 없으면 직전 실행이 비정상 종료(server kill / cascade strict-projection block 등)된
        # 상태이므로 새 trigger 를 통과시킴. 진짜 진행 중이라면 task_registry 에 job 이 있어
        # 정상적으로 409 거절. (cascade fix 2026-05-05)
        from app.core.task_registry import is_task_running

        has_active_job = any(
            is_task_running(f"run_all:{project_id}:{episode_id}:{cat}")
            for cat in RUN_ALL_CATEGORY_TOKENS
        )
        if has_active_job:
            raise AppError(
                code="analysis.already_running",
                message=t("analysis.already_running"),
                status_code=409,
            )
        logger.warning(
            "preflight_analysis_start: episode %s status='analyzing' but no active "
            "task_registry entry — auto-recovering stale lock from previous abnormal exit",
            episode_id,
        )
        stale_recovered = True
    if not episode.fulltext:
        raise AppError(code="analysis.no_text", message=t("analysis.no_text"), status_code=400)
    from app.core.openai_keys import has_openai_key
    if not has_openai_key():
        raise AppError(
            code="analysis.openai_key_missing",
            message=t("analysis.openai_key_missing"),
            status_code=400,
        )

    # Stale recovery 시 prior_status 를 'error' 로 normalize — 후속 dispatch 가 fail
    # 하더라도 rollback 시 다시 'analyzing' 으로 복원되지 않도록 보장 (review I1).
    prior_status = "error" if stale_recovered else episode.status
    episode.status = "analyzing"
    episode.analysis_error = None
    db.commit()
    return prior_status


def rollback_episode_status(
    db: OrmSession,
    episode_id: str,
    prior_status: Optional[str],
) -> None:
    """preflight 이후 dispatch 실패 시 episode.status 복구."""
    if prior_status is None:
        return
    from app.models.project import Episode

    ep = db.query(Episode).filter(Episode.id == episode_id).first()
    if ep:
        ep.status = prior_status
        db.commit()


def dispatch_category_run(
    project_id: str,
    episode_id: str,
    category: str,
    mode: str,
    db: OrmSession,
    *,
    image_call_cap: Optional[int] = None,
    approve_image_generation: bool = False,
) -> Dict[str, Any]:
    """카테고리 단위 run-all — 단일 진입점.

    API 엔드포인트(run_all_steps, /analyze) 공용 dispatch 헬퍼.
    category가 analysis/all이면 preflight 포함 (Codex W1 Critical 수정).

    W20E5: when ``category in {"image", "all"}`` and
    ``settings.background_render_reference_mode == "shot_aware_plan"``,
    the caller must explicitly pass ``approve_image_generation=True`` and
    a positive ``image_call_cap`` or the dispatch raises before any
    background job is submitted. The budget is then installed on the
    worker thread so every provider call site reserves against it.
    """
    from app.core.task_registry import is_task_running

    for running_cat in RUN_ALL_CATEGORY_TOKENS:
        if is_task_running(f"run_all:{project_id}:{episode_id}:{running_cat}"):
            raise AppError(
                code="step.already_running",
                message="파이프라인 이미 실행 중",
                status_code=409,
            )

    # analysis/all 경로는 episode 상태 전이 + 선검증 필수.
    prior_status: Optional[str] = None
    if category in _ANALYSIS_CATEGORIES:
        prior_status = preflight_analysis_start(db, project_id, episode_id)

    try:
        # W20E5 — evaluate the image-call-budget gate BEFORE doing any work
        # so a fail-closed decision aborts before steps are computed or any
        # background job is submitted.
        budget = evaluate_image_call_budget_gate(
            category=category,
            image_call_cap=image_call_cap,
            approve_image_generation=approve_image_generation,
        )

        project_config = load_project_llm_config(db, project_id)
        # Fix 3.3: episode_id 전달 — category=image의 prerequisite 검증에 필요.
        # Fix D: mode 전달 — category=image&mode=force 의 cross-category cascade 검증.
        step_ids = select_steps_for_category(
            db, project_id, category, episode_id=episode_id, mode=mode
        )
        opik_context = build_opik_context(db, project_id, episode_id)

        job_key = f"run_all:{project_id}:{episode_id}:{category}"
        started = submit_background_job(
            job_key=job_key,
            target=run_steps_batch,
            args=(
                project_id,
                episode_id,
                step_ids,
                mode,
                project_config,
                opik_context,
                budget,
            ),
            description=f"Run all {category} steps for {episode_id}",
        )
        if not started:
            raise AppError(
                code="step.already_running",
                message="파이프라인 이미 실행 중",
                status_code=409,
            )
    except Exception:
        # 실패 시 preflight로 전환된 status를 prior로 복구
        rollback_episode_status(db, episode_id, prior_status)
        raise

    return {
        "ok": True,
        "status": "started",
        "steps": step_ids,
        "job_key": job_key,
    }


def dispatch_scene_reanalysis(
    project_id: str,
    episode_id: str,
    db: OrmSession,
) -> Dict[str, Any]:
    """씬 재분석(reanalyze-scenes) dispatch — scene_save + downstream force 실행."""
    from app.core.task_registry import is_task_running

    for running_cat in RUN_ALL_CATEGORY_TOKENS:
        if is_task_running(f"run_all:{project_id}:{episode_id}:{running_cat}"):
            raise AppError(
                code="step.already_running",
                message="파이프라인 이미 실행 중",
                status_code=409,
            )

    project_config = load_project_llm_config(db, project_id)
    step_ids = select_scene_reanalysis_steps(db, project_id)
    opik_context = build_opik_context(db, project_id, episode_id)

    job_key = f"run_all:{project_id}:{episode_id}:reanalyze"
    started = submit_background_job(
        job_key=job_key,
        target=run_steps_batch,
        args=(project_id, episode_id, step_ids, "force", project_config, opik_context),
        description=f"Reanalyze scenes for episode {episode_id}",
    )
    if not started:
        raise AppError(
            code="step.already_running",
            message="파이프라인 이미 실행 중",
            status_code=409,
        )

    return {
        "ok": True,
        "status": "started",
        "steps": step_ids,
        "job_key": job_key,
    }