"""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 functools
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 _release_episode_status_on_cancel(
    db: OrmSession, episode_id: str, reason: str
) -> None:
    """정지로 멈췄을 때 `analyzing` 잠금을 푼다 — **`error` 로는 안 적는다.**

    `analyzing` 은 「지금 돌고 있다」는 뜻이라, 착수 검사가 그것을 보고
    재실행을 409로 막는다. 세워 놓고 다시 못 돌리게 되는 것이다.

    ★그렇다고 `error` 도 아니다. 운영자가 멈춘 것이지 깨진 것이 아니다.
     상태는 `stopped` 로 적고, 사유는 `analysis_error` 칸에 남긴다
     (그 칸이 「무슨 일이 있었나」를 담는 유일한 자리다).
    """
    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 = "stopped"
            ep.analysis_error = f"주행 정지: {reason or '(사유 없음)'}"[:2000]
            db.commit()
    except Exception as exc:
        logger.error("정지 뒤 에피소드 상태 해제 실패: %s", 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,
) -> Dict[str, Any]:
    """step_ids 순차 실행 — background job 내부에서 호출되는 실제 워커.

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

    Returns:
        ``{"ok": bool, "outcome": "completed|cancelled|failed", "reason": str}``

        ★★★2026-09-04 — 종전에는 **아무것도 안 돌려줬다**(`None`). 실패를
         `all_ok=False` 로 접고 episode.status 만 고친 뒤 조용히 끝났다.
         그래서 프로젝트 대기열이 「앞 화가 실패했나」를 알 길이 없어
         **다음 화를 그대로 시작했다** (Codex BLOCK 2026-09-04).
         배경 작업 경로는 이 값을 안 쓰므로 계약은 뒤로 호환된다.

    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
        # 협조적 정지는 실패와 **다른 결말**이다. 아래 마무리에서 에피소드를
        # `error` 로 바꾸는 복구 경로를 타지 않게 따로 표시한다.
        cancelled = False
        # 회귀 가드 (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:
            # ★스텝 사이가 **주행을 세우는 자리**다. `step_run` 의 정지 표식은
            #  스텝 하나만 세우므로, 그 스텝이 멈춰도 이 반복문은 다음 sid 로
            #  넘어간다. 주행 전체를 덮는 표를 여기서 읽어야 배치가 선다.
            #  (`app/core/run_control.py` 머리말)
            from app.core.run_control import read_cancel_state

            cancel_state = read_cancel_state(db, project_id, episode_id)
            if cancel_state.requested:
                logger.warning(
                    "[RUN-CANCEL] 배치 정지 — %s 앞에서 멈춘다. %s",
                    sid, cancel_state.describe(),
                )
                # ★`all_ok=False` 가 아니다 — 정지는 실패가 아니다.
                cancelled = True
                failure_message = f"주행 정지 요청 ({cancel_state.describe()})"
                break

            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 == "step.cancelled":
                    # ★협조적 정지는 **실패가 아니다.** 여기서 all_ok=False 로
                    #  흘려보내면 아래 `_recover_episode_status_on_failure` 가
                    #  에피소드를 `error` 로 바꾼다 — 운영자가 세운 것을
                    #  시스템이 깨진 것으로 적는 셈이다. StepRunner 가 애써
                    #  `cancelled` 로 갈라 놓은 것이 여기서 다시 뭉개진다.
                    logger.warning("주행 정지 — %s 에서 멈춘다: %s", sid, ae.message)
                    cancelled = True
                    failure_message = ae.message
                    break
                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 cancelled:
            # ★정지는 **에피소드를 error 로 만들지 않는다.** 운영자가 세운 것을
            #  깨진 것으로 적으면, 다시 주행하려 할 때 사람이 먼저 「무엇이
            #  고장났나」를 찾게 된다. 여기까지 한 일은 그대로 남고, 정지 표를
            #  내린 뒤 resume 하면 이어서 간다.
            #
            # ★그렇다고 `analyzing` 으로 **두면 안 된다.** 그 상태는 「지금
            #  돌고 있다」는 뜻이고, 착수 검사가 그것을 보고 재실행을 409로
            #  막는다 — 세워 놓고 다시 못 돌리는 셈이다. 잠금을 푼다.
            logger.warning(
                "[RUN-CANCEL] 주행이 정지 요청으로 멈췄다 — %s. "
                "다시 돌리려면 POST .../steps/cancel/clear 로 표를 내려라.",
                failure_message or "(사유 없음)",
            )
            _release_episode_status_on_cancel(db, episode_id, failure_message)
            return {"ok": False, "outcome": "cancelled",
                    "reason": failure_message or "정지 요청"}
        elif 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}"
                )
                # ★마지막 sync 실패도 **실패**다. 여기서 ok 를 돌려주면
                #  대기열이 「앞 화가 끝났다」로 읽고 다음 화를 시작한다.
                return {"ok": False, "outcome": "failed",
                        "reason": f"Final sync failed: {sync_exc}"}
            return {"ok": True, "outcome": "completed", "reason": ""}
        else:
            _recover_episode_status_on_failure(
                db, episode_id, failure_message or "Pipeline failed"
            )
            return {"ok": False, "outcome": "failed",
                    "reason": 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))
        return {"ok": False, "outcome": "failed", "reason": 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 _rollback_episode_status_fresh(
    episode_id: str,
    prior_status: Optional[str],
) -> None:
    """세션이 죽었을 때 **새 세션으로** 상태를 되돌린다 (best-effort).

    실패를 일으킨 세션이 DB 오류로 못 쓰게 됐고 `db.rollback()` 마저 실패한
    경우다. 그 세션으로는 아무것도 못 하므로 새 세션을 연다.
    """
    if prior_status is None:
        return
    session = SessionLocal()
    try:
        rollback_episode_status(session, episode_id, prior_status)
    finally:
        session.close()


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,
    run_inline: bool = False,
    claim_held_by: Optional[str] = None,
) -> Dict[str, Any]:
    """카테고리 단위 run-all — 단일 진입점.

    ``run_inline`` — 배치를 **이 스레드에서 끝까지** 돌린다. 프로젝트 대기열이
    여러 편을 순서대로 돌릴 때 쓴다. 배경 작업으로 던지면 바로 돌아오므로
    「앞 화가 끝난 뒤 다음 화」를 만들 수 없다.

    ``claim_held_by`` — 호출자가 **프로젝트 자리를 이미 들고 있다**는 표시.
    ★대기열이 화마다 자리를 놓았다 잡으면 그 틈에 단일 실행이 끼어들어
    순서를 깨고 canon·장부를 갱신한다 (Codex BLOCK 2026-09-04). 대기열은
    자리를 **처음부터 끝까지** 들고, 안쪽 화 실행은 그 자리를 물려 쓴다.

    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,
            )

    # ★위 훑기와 아래 등록 사이가 길다 — preflight·설정 읽기·스텝 고르기가
    #  끼어 있다. 요청 둘이 그 틈에 나란히 훑으면 category 가 달라 **둘 다**
    #  통과해 각자 등록된다. 그러면 한 에피소드에서 배치 두 개가 같이 돌고
    #  정지 표 하나로는 둘 다 못 세운다. 에피소드 자리를 먼저 잡는다.
    from app.core.task_registry import claim_episode_run, release_episode_run

    # ★★★자리는 **프로젝트 하나**다 (2026-09-04). 종전에는 에피소드마다
    #  따로라 같은 프로젝트의 두 화가 나란히 돌 수 있었다. 그런데
    #  `entity_sync` 는 프로젝트 범위 `entity_canon` 을 읽고 쓰고, `short_id`
    #  장부도 프로젝트 것이다 — 둘이 같이 돌면 같은 번호를 두고 부딪힌다.
    #
    #  ★그리고 **순차가 요구사항**이다: 앞 화가 끝나야 canon 이 서고, 그래야
    #   다음 화가 앞 화 명부를 보고 이어 붙인다. 나란히 돌면 둘 다 백지에서
    #   시작해 같은 것을 두 신원으로 만든다.
    episode_key = f"run_all:{project_id}"
    # ★자리를 **내가 잡았나**. 대기열이 물려준 것이면 놓지도 않는다.
    _own_claim = claim_held_by is None
    if _own_claim and not claim_episode_run(
            episode_key, holder=f"{episode_id}:{category}"):
        from app.core.task_registry import episode_run_holder
        raise AppError(
            code="step.already_running",
            message=(
                f"이 프로젝트에서 이미 다른 화가 돌고 있습니다 "
                f"(먼저 잡은 쪽: {episode_run_holder(episode_key) or '?'}). "
                f"여러 편은 순서대로 돕니다."
            ),
            status_code=409,
        )
    if not _own_claim:
        from app.core.task_registry import episode_run_holder

        _holder = episode_run_holder(episode_key)
        if _holder != claim_held_by:
            # ★★물려받았다고 **말만** 하고 실제로는 아무도 안 잡고 있으면,
            #  그 순간부터 문이 없는 것과 같다. 확인하고 아니면 선다.
            raise AppError(
                code="step.claim_not_held",
                message=(f"자리를 물려받았다고 했는데 실제 주인은 "
                         f"{_holder or '없음'} 이다"),
                status_code=409)

    # ★자리를 잡은 **뒤의 모든 코드**가 try 안에 있어야 한다.
    #  종전에는 `preflight_analysis_start` 가 try 밖이라, 그것이 던지면
    #  자리가 그대로 남아 **그 에피소드가 프로세스를 다시 띄우기 전까지
    #  영영 막혔다.** 잡았으면 놓는 길이 예외 갈래에도 있어야 한다.
    prior_status: Optional[str] = None
    try:
        # analysis/all 경로는 episode 상태 전이 + 선검증 필수.
        if category in _ANALYSIS_CATEGORIES:
            prior_status = preflight_analysis_start(db, project_id, episode_id)

        # 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}"

        @functools.wraps(run_steps_batch)
        def _run_then_release(*args):
            """배치가 끝나면 에피소드 자리를 놓는다.

            ★놓는 자리가 여기여야 한다 — 아래 `submit_background_job` 은
             제출만 하고 바로 돌아온다. 제출 직후에 놓으면 배치가 도는 동안
             자리가 비어 두 번째 요청이 그대로 들어온다.
            """
            try:
                return run_steps_batch(*args)
            finally:
                # ★자리를 놓는 것은 **이 배치가 잡았을 때만**이다. 대기열이
                #  프로젝트 자리를 통째로 들고 있으면 화 사이에 놓으면 안 된다
                #  — 그 틈에 단일 실행이 들어와 순서를 깬다 (Codex BLOCK).
                if _own_claim:
                    release_episode_run(episode_key)

        _args = (
                project_id,
                episode_id,
                step_ids,
                mode,
                project_config,
                opik_context,
                budget,
        )
        if run_inline:
            # ★이 스레드에서 끝까지 돈다 — 대기열이 「앞 화가 끝난 뒤」를
            #  만들 수 있는 유일한 길이다. 자리 놓기는 `_run_then_release`
            #  안에 있으므로 여기서 또 놓지 않는다.
            #  ★★배치의 **끝 결과를 그대로 올린다.** 무조건 completed 를
            #   돌려주면 앞 화가 실패해도 대기열이 다음 화를 시작한다
            #   (Codex BLOCK 2026-09-04).
            outcome = _run_then_release(*_args) or {
                "ok": False, "outcome": "failed",
                "reason": "run_steps_batch 가 결과를 안 돌려줬다"}
            return {"ok": bool(outcome.get("ok")),
                    "status": outcome.get("outcome", "failed"),
                    "reason": outcome.get("reason", ""),
                    "steps": step_ids, "job_key": job_key}
        started = submit_background_job(
            job_key=job_key,
            target=_run_then_release,
            args=_args,
            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로 복구.
        # ★에피소드 자리도 놓는다 — 안 놓으면 실패 한 번에 그 에피소드가
        #  영영 막힌다(프로세스를 다시 띄우기 전에는 풀 길이 없다).
        #
        # ★순서가 중요하다 (2026-08-26 Codex 재리뷰 BLOCK-1). 종전에는 자리를
        #  **먼저** 놓고 상태를 되돌렸다. 그 틈에 두 번째 요청이 자리를 잡고
        #  preflight 로 'analyzing' 을 적으면, 뒤늦은 되돌리기가 **살아서
        #  도는 남의 상태를 옛 값으로 덮는다.** 상태를 다 되돌린 뒤에 놓는다.
        #
        # ★되돌리기 전에 세션부터 정리한다. 여기 온 예외가 DB 예외였다면 이
        #  세션은 이미 못 쓰는 상태다. 그대로 조회하면 PendingRollbackError 가
        #  나서 **원래 예외를 가리고**, 상태도 안 돌아간 채 남는다.
        session_usable = True
        try:
            db.rollback()
        except Exception:  # noqa: BLE001
            # 세션 정리 실패가 원래 예외를 가리면 안 된다 — 적기만 한다.
            session_usable = False
            logger.warning(
                "dispatch 실패 정리 중 세션 rollback 실패 (episode=%s)",
                episode_id, exc_info=True)
        try:
            if session_usable:
                rollback_episode_status(db, episode_id, prior_status)
            else:
                # 이 세션으로는 아무것도 못 한다 — 새 세션으로 해 본다.
                _rollback_episode_status_fresh(episode_id, prior_status)
        except Exception:  # noqa: BLE001
            # ★되돌리기가 실패해도 **원래 예외를 가리면 안 된다** (2026-08-26
            #  Codex 2차 재리뷰 BLOCK-4). 종전에는 이 예외가 그대로 올라가
            #  아래 `raise` 에 도달하지 못했고, 호출자는 진짜 실패 사유 대신
            #  「되돌리다 실패했다」만 봤다. 적기만 하고 넘어간다.
            logger.error(
                "dispatch 실패 정리 중 episode.status 되돌리기 실패 "
                "(episode=%s, 되돌릴 값=%s) — 원래 예외를 그대로 올린다",
                episode_id, prior_status, exc_info=True)
        finally:
            # 되돌리기가 어떻게 끝나든 **내가 잡은** 자리는 반드시 놓는다.
            if _own_claim:
                release_episode_run(episode_key)
        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,
            )

    # ★재분석도 **같은 에피소드 자리**를 잡는다. 위 훑기는 category run-all
    #  키만 보므로, run-all 과 재분석이 훑기→등록 사이를 나란히 통과해 둘 다
    #  돌 수 있었다. 자리는 하나다.
    from app.core.task_registry import (
        claim_episode_run, episode_run_holder, release_episode_run,
    )

    # ★위와 **같은 자리**다 — 프로젝트 하나에 배치 하나.
    episode_key = f"run_all:{project_id}"
    if not claim_episode_run(episode_key, holder="reanalyze"):
        raise AppError(
            code="step.already_running",
            message=(
                f"이 에피소드는 이미 실행 중입니다 "
                f"(먼저 잡은 쪽: {episode_run_holder(episode_key) or '?'})"
            ),
            status_code=409,
        )

    try:
        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"

        @functools.wraps(run_steps_batch)
        def _run_then_release(*args):
            try:
                run_steps_batch(*args)
            finally:
                release_episode_run(episode_key)

        started = submit_background_job(
            job_key=job_key,
            target=_run_then_release,
            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,
            )
    except Exception:
        release_episode_run(episode_key)
        raise

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


# ── 프로젝트 대기열 — 여러 편을 **순서대로** ────────────────────────────

#: 대기 중 상태. `Episode.status` 에 durable 하게 남아 프로세스를 다시 띄워도
#: 화면에서 보인다. 메모리 대기열은 사라지지만 「무엇을 대기시켰는지」는 남는다.
EPISODE_STATUS_QUEUED = "queued"
#: 앞 화가 실패해서 서 있는 상태. 자동으로 넘어가지 않는다 — 사람이 다시
#: 돌리거나 건너뛴다고 정해야 한다.
EPISODE_STATUS_BLOCKED = "blocked_by_previous"


def dispatch_project_queue(
    project_id: str,
    episode_ids: List[str],
    category: str,
    mode: str,
    db: OrmSession,
    image_call_cap: Optional[int] = None,
    approve_image_generation: bool = False,
) -> Dict[str, Any]:
    """여러 편을 **화수 순서대로 하나씩** 돌린다.

    ★``image_call_cap``·``approve_image_generation`` — 단일 실행과 **같은
    인자**를 받는다. 없으면 `category="all"` 은 착수 검사에서 선다
    (이미지 생성은 사람이 명시로 켜야 한다). 대기열이라고 그 문을 비켜
    가면 안 된다.

    ★★``image_call_cap`` 은 **화당** 상한이다 — 대기열 전체 합이 아니다.
     화마다 새 예산이 깔린다. N편을 넣으면 최대 `N × cap` 이다. 이 말을
     안 적으면 승인한 수와 실제로 살 수 있는 수가 갈린다.

    ★왜 순차인가 — 앞 화가 끝나야 `entity_canon` 이 서고, 그래야 다음 화가
     앞 화 명부를 보고 같은 신원으로 이어 붙인다(`episode_carry`). 나란히
     돌리면 둘 다 백지에서 시작해 같은 것을 두 신원으로 만든다. 게다가
     `short_id` 장부와 canon 이 프로젝트 범위라 동시 쓰기가 부딪힌다.

    ★다른 프로젝트끼리는 그대로 나란히 돈다 — 자리는 프로젝트마다 하나다.

    ★앞 화가 실패하면 뒤 화는 `blocked_by_previous` 로 **선다.** 자동으로
     넘어가면 앞 화 없이 만든 신원이 그대로 굳는다.
    """
    # ★`submit_background_job` 은 **모듈 위**에서 이미 가져온다. 여기서 다시
    #  import 하면 그 이름이 지역으로 가려져, 시험이 모듈 전역을 바꿔도
    #  **진짜 배경 스레드**가 돈다 — 실측 2026-09-04: 그래서 시험이 대기열이
    #  끝나기 전에 DB 를 읽고 「안 섰다」로 보였다.
    from app.models.project import Episode

    if not episode_ids:
        raise AppError(code="queue.empty", message="넣을 에피소드가 없습니다",
                       status_code=400)

    rows = (
        db.query(Episode)
        .filter(Episode.project_id == project_id, Episode.id.in_(episode_ids))
        .order_by(Episode.episode_number)
        .all()
    )
    found = {e.id for e in rows}
    missing = [e for e in episode_ids if e not in found]
    if missing:
        raise AppError(
            code="episode.not_found",
            message=f"이 프로젝트에 없는 에피소드: {missing[:3]}",
            status_code=404)
    # ★화수가 겹치면 순서가 안 정해진다 — 시작 전에 선다 (Codex 추가 확인).
    _nums = [e.episode_number for e in rows]
    _dupes = sorted({n for n in _nums if _nums.count(n) > 1})
    if _dupes:
        raise AppError(
            code="queue.duplicate_episode_number",
            message=(f"화수가 겹칩니다 {_dupes} — 순서를 정할 수 없습니다. "
                     f"화수를 고친 뒤 다시 넣어 주세요."),
            status_code=400)
    running = [e for e in rows if e.status == "analyzing"]
    if running:
        raise AppError(
            code="step.already_running",
            message=f"이미 도는 중인 화가 있습니다 ({running[0].episode_number}화)",
            status_code=409)

    ordered = [(e.id, e.episode_number) for e in rows]
    now = datetime.now(timezone.utc).isoformat()
    for e in rows:
        e.status = EPISODE_STATUS_QUEUED
        e.updated_at = now
    db.commit()

    queue_key = f"run_queue:{project_id}"
    #: 대기열이 프로젝트 자리를 들고 있다는 표식. 안쪽 화 실행이 이것을 물려 쓴다.
    claim_owner = f"queue:{project_id}"

    def _drain():
        """앞 화가 **끝난 것을 확인하고** 다음 화. ★배경 작업은 하나다."""
        from app.core.database import SessionLocal
        from app.core.task_registry import (claim_episode_run,
                                            release_episode_run)

        project_key = f"run_all:{project_id}"
        # ★★자리를 **처음부터 끝까지** 든다. 화마다 놓았다 잡으면 그 틈에
        #  단일 실행이 끼어들어 순서를 깨고 canon·장부를 갱신한다.
        if not claim_episode_run(project_key, holder=claim_owner):
            logger.error("프로젝트 대기열 %s: 자리를 못 잡아 시작하지 못했다",
                         project_id)
            for eid, _n in ordered:
                session = SessionLocal()
                try:
                    _mark_status(session, eid, EPISODE_STATUS_BLOCKED)
                finally:
                    session.close()
            return
        stopped_at: Optional[int] = None
        try:
            for eid, number in ordered:
                session = SessionLocal()
                try:
                    if stopped_at is not None:
                        _mark_status(session, eid, EPISODE_STATUS_BLOCKED)
                        continue
                    out = dispatch_category_run(
                        project_id=project_id, episode_id=eid,
                        category=category, mode=mode, db=session,
                        run_inline=True, claim_held_by=claim_owner,
                        image_call_cap=image_call_cap,
                        approve_image_generation=approve_image_generation,
                    )
                    # ★★★**결과를 읽는다.** `run_steps_batch` 는 스텝 실패에
                    #  예외를 안 올리고 episode.status 만 고친 뒤 조용히 끝난다.
                    #  그래서 예외만 보면 실패한 화 뒤가 그대로 시작된다
                    #  (Codex BLOCK 2026-09-04).
                    if not (out or {}).get("ok"):
                        stopped_at = number
                        logger.error(
                            "프로젝트 대기열 %s: %d화가 %s 로 끝났다 — 남은 화는 "
                            "%s 로 둔다: %s", project_id, number,
                            (out or {}).get("status", "?"),
                            EPISODE_STATUS_BLOCKED, (out or {}).get("reason", ""))
                except Exception as exc:  # noqa: BLE001
                    stopped_at = number
                    logger.error(
                        "프로젝트 대기열 %s: %d화에서 멈춤 — 남은 화는 %s 로 둔다: %s",
                        project_id, number, EPISODE_STATUS_BLOCKED, exc,
                        exc_info=True)
                    # ★★실패한 **그 화**도 표를 고친다. 착수 전에 터지면
                    #  preflight 되돌리기가 상태를 `queued` 로 돌려 놓는데,
                    #  대기열은 이미 끝났으니 그 화는 영영 「대기 중」으로
                    #  보인다 — 사람이 무엇을 할지 모른다 (실측 2026-09-04 E2E).
                    _mark_failed_if_still_queued(session, eid, str(exc))
                finally:
                    session.close()
        finally:
            release_episode_run(project_key)

    started = submit_background_job(
        job_key=queue_key, target=_drain,
        description=f"Run {len(ordered)} episodes in order for {project_id}")
    if not started:
        # ★★위에서 이미 `queued` 로 적고 commit 했다. 여기서 안 되돌리면
        #  그 화들이 **아무도 안 돌릴 대기 중**으로 남는다 (Codex 2026-09-04).
        for e in rows:
            e.status = "uploaded"
            e.updated_at = datetime.now(timezone.utc).isoformat()
        db.commit()
        raise AppError(code="step.already_running",
                       message="이 프로젝트의 대기열이 이미 돌고 있습니다",
                       status_code=409)
    return {"ok": True, "status": "queued",
            "episodes": [{"episode_id": eid, "episode_number": n}
                         for eid, n in ordered]}


def _mark_failed_if_still_queued(session: OrmSession, episode_id: str,
                                 reason: str) -> None:
    """착수 전에 터진 화를 `error` 로. ★이미 다른 표가 있으면 안 건드린다."""
    from app.models.project import Episode

    try:
        row = session.query(Episode).filter(Episode.id == episode_id).first()
        if row is None or row.status != EPISODE_STATUS_QUEUED:
            return
        row.status = "error"
        row.analysis_error = reason[:500]
        row.updated_at = datetime.now(timezone.utc).isoformat()
        session.commit()
    except Exception:  # noqa: BLE001
        session.rollback()
        logger.warning("실패 표 기록 실패 (%s)", episode_id, exc_info=True)


def _mark_status(session: OrmSession, episode_id: str, status: str,
                 now: Optional[str] = None) -> None:
    """상태 한 칸만 바꾼다. 실패해도 대기열을 멈추지 않는다."""
    from app.models.project import Episode

    try:
        row = session.query(Episode).filter(Episode.id == episode_id).first()
        if row is None:
            # ★조용히 넘어가지 않는다 — 「막혔다」를 못 적으면 화면에는
            #  「대기 중」으로 남아 사람이 무엇을 할지 모른다.
            logger.error("에피소드 상태를 못 적었다 — 행이 없다 (%s → %s)",
                         episode_id, status)
            return
        row.status = status
        row.updated_at = now or datetime.now(timezone.utc).isoformat()
        session.commit()
        logger.info("에피소드 상태 %s → %s", episode_id, status)
    except Exception:  # noqa: BLE001
        session.rollback()
        logger.warning("에피소드 상태 기록 실패 (%s → %s)", episode_id, status,
                       exc_info=True)