# 파이프라인 회복력 재설계 (2026-05-01)

> **목표**: 무한 반복되는 1-step-fail → hotfix → 다른 step에서 재발 패턴을 시스템 차원에서 종결.
>
> **배경**: 2026-05-01 새 프로젝트 E2E (PID `0bb48ebf`, 금월도 EP1)에서 8건의 사고 발생. 모든 사고가 4가지 메타 원인의 조합. 각 사고를 개별 hotfix로 처리했으나 시스템 결함 그대로.

---

## Part 1 — 오늘 사고 시간순 분석

### 사고 목록

| # | 시각 | 사고 step | 직접 원인 | 메타 원인 |
|---|------|----------|----------|---------|
| 1 | 10:25 | `entity_character_list` | gemini-pro PROHIBITED 3회 (retry 포함) | M1: graphic visual_rules + step에 fallback 없음 |
| 2 | 02:23 | `shot_validator S19` | gemini-pro empty 1건 | M1: shot_validator는 자체 fallback 보유 → 회복 |
| 3 | 02:53 | `scene_consistency S18` | gemini-pro empty 1건 | M1: 3-tier fallback 보유 → sanitize로 회복 |
| 4 | 02:57 | `scene_detail S25_shot6` | gemini-pro PROHIBITED (시나리오 graphic 텍스트) | M1: scene_detail에 fallback 없었음 (오늘 추가) |
| 5 | 12:43 | `scene_image_pipeline` | gate.no_scenes (scene_still 0건) | M2: scene_still sync 가드가 partial 차단 |
| 6 | 13:08 | dispatcher 멈춤 | `background_classify` config_hash mismatch | M3: config_hash trigger 불투명, resume 모드 stale 차단 |
| 7 | 13:13 | force category=image이 전체 invalidate | transitive deps auto-include 광범위 | M4: dispatcher mode 의미 부재 |
| 8 | 13:19 | manifest.json 비어있음 | force 직전 백업, backend kill로 새 manifest 못 만듦 | M5: 자동 복구 메커니즘 없음 |

### 메타 원인 (5가지)

| 코드 | 이름 | 단일 사고 횟수 | fan-out 위험 |
|-----|------|--------------|-----------|
| **M1** | LLM step별 fallback 정책 부재 | 4건 | 1 step fail → downstream cascade |
| **M2** | sync_service의 partial 처리 정책 부재 | 1건 (scene_still) | 1 shot fail → 64 shot 차단 (10x fan-out) |
| **M3** | dispatcher resume이 stale step 처리 못함 | 1건 | dispatcher 영구 멈춤 |
| **M4** | dispatcher mode 의미 명문화 부재 | 1건 | 사용자 명령 사고 |
| **M5** | manifest 자동 복구 부재 | 1건 | 사고 후 수동 복구 작업 |

---

## Part 2 — 코드 레벨 정확한 진단

### M1 — LLM step별 fallback 정책 부재

**현재 상태 정밀 audit** (21 step 중 LLM 호출):

| 보유 패턴 | step | 비고 |
|---------|------|------|
| ✅ 3-tier fallback (gemini → sanitize → gpt) | `scene_consistency_step.py`, `location_consistency_step.py`, `detail_steps.py`(scene_detail, 오늘 추가) | 안전 |
| ⚠️ Simple retry only (같은 prompt N회) | `analysis_steps_legacy.py`, `beat_shot_steps.py`(beat_extract/shot_extract), `character_list_step.py`, `entity_steps.py`, `shot_cinematography_step.py` | PROHIBITED는 deterministic, retry 무의미 |
| ❌ NO fallback | `background_classify_step.py`, `background_master_plan_step.py`, `background_planner_step.py`, `background_prompt_step.py`, `director_steps.py`(scene_director/shot_director), `floor_plan_prompt_step.py`, `planning_doc_step.py`, `scene_camera_flow_step.py`, `scene_steps.py`(scene_summary/episode_summary/visual_world_rules), `shot_dependency_t2i_step.py`, `shot_essence_extraction_step.py`, `shot_selection_step.py`, `shot_validator_step.py` | **18 step 무방비** |

**18개 step이 시나리오의 graphic 단어 한 번 만나면 무방비로 fail.**

**파일 위치 — `call_structured` 본체** (`backend/app/modules/llm/llm_client.py:355-404`):
```python
def call_structured(step, system_prompt, user_prompt, response_schema, ...):
    router = _get_router()
    model = _resolve_model(step, project_config)
    kwargs = {"model": model, "messages": [...], "response_format": {...}, ...}
    _sanitize_kwargs_for_model(model, kwargs)
    _apply_gemini_safety(model, kwargs)
    response = router.completion(**kwargs)
    content = response.choices[0].message.content
    if not content:
        raise RuntimeError(f"LLM returned empty response for step={step}, model={model}")
    return json.loads(content)
```

→ **fallback 로직이 step 코드에 분산**. `call_structured` 자체는 1차 호출만 함. 각 step이 try/except로 감싸야 fallback 작동.

**참고 — scene_consistency의 3-tier 패턴** (`scene_consistency_step.py:336-400`):
```python
# Tier 1: Gemini 기본
try: result = call_structured(...)
except: 
    # Tier 2: sanitize + 영화 프레이밍
    sanitized = _sanitize_for_safety(user_prompt)
    safe_system = system + "\n\n[콘텐츠 안전 참고]\n이 텍스트는 영화/드라마 촬영 시나리오입니다..."
    try: result = call_structured(system_prompt=safe_system, user_prompt=sanitized)
    except:
        # Tier 3: GPT fallback
        gpt_config = {**project_config, "scene_consistency": {"model": "gpt"}}
        result = call_structured(system_prompt=safe_system, user_prompt=sanitized, project_config=gpt_config)
```

→ 같은 패턴을 **18 step에 일일이 복붙**해야 하는 ad-hoc 구조. 새 step 추가될 때마다 누락 위험.

### M2 — sync_service의 partial 처리 정책 부재

**현재 상태 정밀 audit** (4 sync_service):

| 파일 | 가드 위치 | 현재 동작 |
|------|---------|---------|
| `scene_still_sync_service.py:41` | `bundle.sd_completed` 검사 | partial → sync skip (오늘 사고 발생) |
| `entity_sync_service.py:30` | `t2i_cp.get("status") != "completed"` | partial → sync skip |
| `outlook_sync_service.py:35,43` | `ol_cp.get("status") != "completed"` | partial → sync skip |
| `relation_sync_service.py:26` | `rel_cp.get("status") != "completed"` | partial → sync skip |

**`scene_still_checkpoint_loader.py:28`** (오늘 fix 후 코드):
```python
status = sd_cp.get("status")
bundle.sd_completed = status in ("completed", "partial")  # ← 오늘 fix
if bundle.sd_completed:
    scenes = sd_cp.get("data", {}).get("scenes", [])
    if not scenes:
        bundle.sd_completed = False  # cascade 안전
    else:
        bundle.scenes = scenes
```

→ scene_still만 fix됨. **다른 3개 sync_service는 여전히 partial 거부**. 만약 entity_t2i, outlook_phase3, entity_relation에서 partial 발생하면 같은 사고 재발.

**fan-out 분석**:
- `scene_still`: 1 shot fail → 64 shot 차단 (10x)
- `entity_t2i`: 1 entity fail → 모든 entity sync 차단 (40x+)
- `outlook_phase3`: 1 outlook fail → 모든 outlook sync 차단 (10x)
- `relation_facts`: 1 relation fail → 모든 relation sync 차단 (5x+)

**모두 동일한 cascade 사고 위험**.

### M3 — dispatcher resume이 stale step 처리 못함

**현재 상태** (`step_runner.py:312-345` resume 로직):
```python
if mode == "resume":
    existing = self._get_step_run(self.step_id)
    if existing and existing[0] == "completed":
        cp = self.load_checkpoint()
        if cp:
            current_schema = self.manifest.get("schema_version", 1)
            cp_schema = cp.get("schema_version")
            cp_hash = cp.get("config_hash")
            mismatch_reason = None
            if cp_schema is not None and cp_schema != current_schema:
                mismatch_reason = f"schema_version mismatch..."
            elif cp_hash is not None:
                current_hash = compute_config_hash(self.project_config)
                if cp_hash != current_hash:
                    mismatch_reason = "config_hash mismatch: project_config 변경 감지"
            if mismatch_reason:
                raise AppError(code="step.resume_invalid", ...)  # ← stale로 분류
        return {"status": "skipped", "reason": "already completed"}
```

**문제점**:
1. `mismatch_reason`이 발생하면 `AppError`를 raise해서 dispatcher 전체 멈춤
2. **무엇이 변경됐는지 로그에 안 나옴** (config_hash mismatch만 표시, project_config 어느 key가 다른지 누락)
3. **stale = 사람이 force 명령 내려야 함** — 자동 회복 메커니즘 없음

오늘 13:08 사고:
- backend 재시작 → project_config 일부 변경 감지 → `background_classify` stale
- resume 모드는 stale step 못 함 → dispatcher 멈춤
- 사용자가 강제 force 내림 → 광범위 invalidate (M4 사고 trigger)

### M4 — dispatcher mode 의미 명문화 부재

**현재 동작** (`analysis_dispatch_service.py:98-178`, `select_steps_for_category`):

| 입력 | 코드상 동작 | 사용자 직관 |
|-----|----------|----------|
| `category="all"` | 모든 active step 반환 (transitive 불필요) | ✅ 일치 |
| `category="analysis"` | analysis만, transitive 안 함 | ✅ 일치 |
| `category="image"` | image step + **transitive deps 모두 자동 포함** | ❌ **불일치** |

**문제 핵심** (line 139-178):
```python
# category="image" 만 cross-category transitive deps 자동 포함.
included = set(direct)
queue = list(direct)
while queue:
    sid = queue.pop()
    meta = get_manifest_dict(sid) or {}
    for dep in meta.get("depends_on", []):
        if dep in included: continue
        ...
        included.add(dep); queue.append(dep)
```

→ image 카테고리 force 시 **scene_image_pipeline** 같은 image step의 deps을 transitive하게 따라가서 결국 `scene_detail`, `entity_*`, `outlook_*`, ..., `text_cleanup`까지 모두 포함.

→ 사용자 의도 ("image만 force") ≠ 실제 동작 (43 step 거의 다 force).

**오늘 13:13 사고가 정확히 이 동작**:
- 사용자가 `force category=image` 내림 (image만 의도)
- 실제는 43 step 시작 (text_cleanup부터 visual_world_rules까지 모두)
- backend kill → 모든 cp 일부만 invalidate된 상태로 멈춤

### M5 — manifest 자동 복구 부재

**현재 동작** (`step_runner.py:133-184`):

```python
def _archive_manifest(manifest_path):
    """기존 manifest.json을 날짜시간 버전으로 복사 보관."""
    ts = datetime.now().strftime("%Y%m%d_%H%M%S")
    archive_path = manifest_path.parent / f"manifest_{ts}.json"
    shutil.copy2(str(manifest_path), str(archive_path))

def save_checkpoint(self, data):
    manifest_path = self._cp_dir / "manifest.json"
    ...
    self._archive_manifest(manifest_path)
    atomic_write_json(manifest_path, data)

def clear_checkpoint(self):
    manifest_path = self._cp_dir / "manifest.json"
    if manifest_path.exists():
        self._archive_manifest(manifest_path)
        manifest_path.unlink(missing_ok=True)
```

**상황 시퀀스 (오늘 13:13~13:19)**:
1. force 호출 → `clear_checkpoint()` 실행 → archive 만들고 manifest.json **삭제**
2. step 새로 실행 시작 → 결과 만들고 `save_checkpoint()` 호출 예정
3. backend kill → `save_checkpoint()` 못 호출
4. **결과**: 디렉토리에 archive만 있고 manifest.json 비어있음
5. dispatcher가 cp 못 찾아 step "stale" 분류

**`load_checkpoint()` 로직** (line 121-131):
```python
def load_checkpoint(self):
    manifest_path = self._cp_dir / "manifest.json"
    if not manifest_path.exists():
        return None  # ← 자동 복구 시도 없음
```

→ archive (manifest_*.json)는 있지만 자동 복구 시도 안 함. **사용자가 수동으로 rename**해야 회복.

---

## Part 3 — Fix 계획 (코드 레벨 디테일)

### Fix 1 — 글로벌 fallback layer (M1 해소)

**목표**: `call_structured()` 자체에 3-tier fallback 통합. step별 코드 추가 없이 자동 적용.

#### 변경 1.1 — `_sanitize_for_safety` 모듈화

**파일**: `backend/app/modules/llm/safety.py` (신규)

```python
"""LLM 안전 필터 우회 유틸 — 픽션 시나리오 분석용.

`scene_consistency_step.py`와 `detail_steps.py`에 분산되어 있던
sanitize 로직을 단일 모듈로 통합. 모든 LLM 호출이 자동 사용.
"""
import re

# 한국어 graphic 단어 → 영화 촬영 set 표현
_SAFETY_REPLACEMENTS_KO = [
    ("죽은", "역할상 사망한"),
    ("시신", "분장된 인물"),
    ("피", "특수분장 페인트"),
    ("피웅덩이", "분장 페인트 풀"),
    ("피범벅", "특수분장 페인트가 묻은"),
    ("피묻은", "특수분장 페인트가 묻은"),
    ("살점", "분장 표면"),
    ("뜯", "찢어진 분장"),
    ("녹아내리는", "효과로 변형되는"),
    # ... 추후 사고 발생 시 추가
]

# 영어 (regex)
_SAFETY_REPLACEMENTS_EN = [
    (r"\bdead\b", "in-character motionless"),
    (r"\bcorpse\b", "motionless figure in character"),
    (r"\bblood\b", "dark red stage paint"),
    (r"\bbloody\b", "stage-painted"),
    (r"\bgore\b", "stage effects"),
    (r"\bbleed(ing)?\b", r"emit(ing)? stage paint"),
]

_re_safety = re.compile(r"")  # placeholder for IGNORECASE flag


def sanitize_for_safety(text: str) -> str:
    """안전 필터 회피용 표현 순화 (한국어 + 영어)."""
    result = text
    for ko, replacement in _SAFETY_REPLACEMENTS_KO:
        result = result.replace(ko, replacement)
    for pattern, replacement in _SAFETY_REPLACEMENTS_EN:
        result = re.sub(pattern, replacement, result, flags=re.IGNORECASE)
    return result


# 영화 프레이밍 system prompt 추가
SAFETY_SYSTEM_SUFFIX = """

[콘텐츠 안전 참고]
이 텍스트는 영화/드라마 촬영 시나리오입니다. 모든 폭력/상해/사망 묘사는 특수분장과 연기입니다.
응답 작성 시에도 영화 촬영 세트 맥락에서 순화된 표현을 사용하세요:
- 'dark red stage paint pool' (혈웅덩이 X)
- 'motionless figure in character' (시신 X)
- 'aged photograph prop' (오래된 사진 X — graphic 시 추가)
"""
```

#### 변경 1.2 — `call_structured`에 3-tier 통합

**파일**: `backend/app/modules/llm/llm_client.py:355` 부근

**Before**:
```python
def call_structured(step, system_prompt, user_prompt, response_schema, ...):
    router = _get_router()
    model = _resolve_model(step, project_config)
    kwargs = {...}
    _sanitize_kwargs_for_model(model, kwargs)
    _apply_gemini_safety(model, kwargs)
    response = router.completion(**kwargs)
    content = response.choices[0].message.content
    if not content:
        raise RuntimeError(f"LLM returned empty response for step={step}, model={model}")
    return json.loads(content)
```

**After**:
```python
def call_structured(
    step, system_prompt, user_prompt, response_schema,
    project_config=None, schema_name="response",
    opik_metadata=None, temperature=0.2, max_tokens=None,
    *,
    enable_fallback=True,  # 신규 옵션 — 호출자가 disable 가능 (legacy 호환)
):
    """Structured JSON output 호출 — 3-tier fallback 자동 적용.
    
    Tier 1: 기본 모델
    Tier 2: sanitize + 영화 프레이밍 system suffix
    Tier 3: gpt fallback (model=gpt 강제)
    """
    from app.modules.llm.safety import sanitize_for_safety, SAFETY_SYSTEM_SUFFIX
    
    def _do_call(sys_p, user_p, project_cfg, suffix_tag=""):
        router = _get_router()
        model = _resolve_model(step, project_cfg)
        metadata = _build_opik_metadata(step, opik_metadata)
        if suffix_tag:
            metadata.setdefault("opik", {}).setdefault("tags", []).append(suffix_tag)
        kwargs = {
            "model": model,
            "messages": [
                {"role": "system", "content": sys_p},
                {"role": "user", "content": user_p},
            ],
            "response_format": {
                "type": "json_schema",
                "json_schema": {"name": schema_name, "schema": response_schema, "strict": True},
            },
            "temperature": temperature,
            "metadata": metadata,
        }
        from app.core.config import settings
        kwargs["max_tokens"] = max_tokens if max_tokens is not None else settings.llm_max_output_tokens
        _sanitize_kwargs_for_model(model, kwargs)
        _apply_gemini_safety(model, kwargs)
        response = router.completion(**kwargs)
        content = response.choices[0].message.content
        if not content:
            raise RuntimeError(f"LLM returned empty response for step={step}, model={model}")
        return json.loads(content)
    
    # Tier 1: 기본
    try:
        return _do_call(system_prompt, user_prompt, project_config)
    except Exception as exc_t1:
        if not enable_fallback:
            raise
        logger.warning("call_structured[%s] Tier 1 failed (%s), trying sanitized", step, exc_t1)
    
    # Tier 2: sanitize
    sanitized_prompt = sanitize_for_safety(user_prompt) if isinstance(user_prompt, str) else user_prompt
    safe_system = system_prompt + SAFETY_SYSTEM_SUFFIX
    try:
        return _do_call(safe_system, sanitized_prompt, project_config, suffix_tag="sanitized")
    except Exception as exc_t2:
        logger.warning("call_structured[%s] Tier 2 sanitized failed (%s), trying GPT fallback", step, exc_t2)
    
    # Tier 3: GPT fallback (Gemini → GPT)
    gpt_config = dict(project_config) if project_config else {}
    gpt_config[step] = {"model": "gpt"}
    return _do_call(safe_system, sanitized_prompt, gpt_config, suffix_tag="gpt_fallback")
```

#### 변경 1.3 — 기존 step의 ad-hoc 3-tier 제거 (선택)

**영향 step**:
- `scene_consistency_step.py:336-400` — Tier 1/2/3 블록 제거, `call_structured(...)` 1줄로 단순화
- `detail_steps.py:762-825` (오늘 추가한 부분) — 동일
- `location_consistency_step.py` — 동일

**원칙**: 글로벌 fallback이 step별 ad-hoc보다 안전. 다만 **step별 specific suffix가 있으면 보존** (예: scene_consistency의 "description에 entity ID 금지" 같은 step-specific 안내).

→ 이 변경은 선택적. 일단 글로벌 fallback만 추가해도 18 step 무방비 해소.

#### 변경 1.4 — `call_text`, `call_multiturn`도 동일 적용

**파일**: `llm_client.py:407-478` (call_text, call_multiturn)

→ call_structured와 같은 3-tier 패턴 적용.

#### 검증 (Fix 1)

```bash
# 단위 테스트 — 신규
backend/tests/modules/llm/test_call_structured_fallback.py
- test_tier1_success_no_fallback
- test_tier1_fail_tier2_success (mock gemini-pro empty → sanitize 통과)
- test_tier2_fail_tier3_success (mock gemini-pro 2회 fail → gpt 통과)
- test_disable_fallback_legacy (enable_fallback=False)
```

```bash
# E2E 검증
1. 새 PID 만들어서 graphic 시나리오 (S25 같은 내용) 분석
2. 모든 18 step 무방비 step에서 graphic word 만나도 통과 확인
3. opik trace에서 sanitized/gpt_fallback 태그 확인
```

---

### Fix 2 — sync_service partial 정책 통일 (M2 해소)

**목표**: 4개 sync_service의 partial 처리 정책을 단일화. cascade 가드는 **데이터 0개 검사**로 단일화.

#### 변경 2.1 — `_base.py`에 공통 가드 함수

**파일**: `backend/app/services/checkpoint_sync/_base.py` (확장)

```python
def is_cp_syncable(cp: Optional[Dict], data_keys: List[str]) -> bool:
    """체크포인트가 sync 가능한 상태인지 판단.
    
    완전한 cascade 가드 단일 표준:
    - cp 없음 → False (pre-analysis)
    - status not in ("completed", "partial") → False (running/error/etc)
    - data 의 핵심 list가 모두 비어있음 → False (cascade 직후)
    - 그 외 → True (partial이라도 데이터 있으면 sync OK)
    
    Args:
        cp: load_cp() 결과
        data_keys: 검사할 data 내부 list 키 (예: ["scenes", "characters", "outlooks"])
                   하나라도 비어있지 않으면 syncable.
    """
    if not cp:
        return False
    status = cp.get("status")
    if status not in ("completed", "partial"):
        return False
    data = cp.get("data", {})
    for key in data_keys:
        val = data.get(key)
        if isinstance(val, list) and val:
            return True
    return False
```

#### 변경 2.2 — 4개 sync_service 적용

**`scene_still_sync_service.py:41`** (오늘 fix를 더 명확화):
```python
# Before (오늘 fix)
if not bundle.sd_completed:
    return {"stills": 0}

# After (단일 표준)
from app.services.checkpoint_sync._base import is_cp_syncable
if not is_cp_syncable(sd_cp, ["scenes"]):
    logger.info("Skipping scene_still sync: scene_detail 데이터 없음 (cascade or pre-analysis)")
    return {"stills": 0}
```

**`entity_sync_service.py:30`**:
```python
# Before
if not t2i_cp or t2i_cp.get("status") != "completed":
    return {"synced": 0, "removed": 0, "skipped": 1}

# After
if not is_cp_syncable(t2i_cp, ["characters", "locations", "props"]):
    logger.info("Skipping entity sync: entity_t2i 데이터 없음")
    return {"synced": 0, "removed": 0, "skipped": 1}
```

**`outlook_sync_service.py:35`**:
```python
# Before
ol_cp = self._load_cp("outlook_phase3")
if not ol_cp or ol_cp.get("status") != "completed":
    ol_cp = self._load_cp("outlook_extraction")
...
if ol_cp and ol_cp.get("status") == "completed":
    ...

# After
ol_cp = self._load_cp("outlook_phase3")
if not is_cp_syncable(ol_cp, ["outlooks"]):
    ol_cp = self._load_cp("outlook_extraction")
if is_cp_syncable(ol_cp, ["outlooks"]):
    ...
```

**`relation_sync_service.py:26`**:
```python
# Before
if not rel_cp or rel_cp.get("status") != "completed":
    return self._empty_delta(skipped=1)

# After
if not is_cp_syncable(rel_cp, ["relations"]):
    return self._empty_delta(skipped=1)
```

#### 변경 2.3 — checkpoint_loader 단일화

**파일**: `scene_still_checkpoint_loader.py:24-32` (오늘 fix 통합)

```python
# Before (오늘 fix)
status = sd_cp.get("status")
bundle.sd_completed = status in ("completed", "partial")
if bundle.sd_completed:
    scenes = sd_cp.get("data", {}).get("scenes", [])
    if not scenes:
        bundle.sd_completed = False
    else:
        bundle.scenes = scenes

# After (단일 표준)
from app.services.checkpoint_sync._base import is_cp_syncable
bundle.sd_completed = is_cp_syncable(sd_cp, ["scenes"])
if bundle.sd_completed:
    bundle.scenes = sd_cp["data"]["scenes"]
```

#### 검증 (Fix 2)

```bash
# 단위 테스트
backend/tests/services/checkpoint_sync/test_partial_sync_policy.py
- test_completed_with_data → syncable
- test_partial_with_data → syncable (← 오늘 사고 핵심)
- test_partial_with_empty_data → not syncable (cascade 가드)
- test_completed_with_empty_data → not syncable (이상 상태)
- test_running → not syncable
- test_no_cp → not syncable
```

```bash
# E2E 검증
1. scene_detail partial 상태 만들기 (S25 fail 재현)
2. SceneStillSyncService 호출
3. scene_still 63 row 정상 sync 확인 (S25 1개만 빠짐)
4. entity_t2i, outlook_phase3, relation도 partial 시 같은 동작 확인
```

---

### Fix 3 — dispatcher mode 의미 명문화 (M3 + M4 해소)

#### 변경 3.1 — resume에 stale 자동 처리 정책 명시

**파일**: `step_runner.py:312-345` (resume 로직)

**Before**:
```python
if mismatch_reason:
    raise AppError(
        code="step.resume_invalid",
        message=f"... force 모드로 재실행하세요.",
        status_code=409,
    )
```

**After**:
```python
if mismatch_reason:
    # mismatch 종류 자세히 로깅 (M3: trigger 불투명 해소)
    logger.warning(
        "Step %s stale detected: %s. Project config diff: %s",
        self.step_id, mismatch_reason,
        _diff_project_config(cp.get("project_config_snapshot"), self.project_config),
    )
    # resume 모드 정책: stale step은 에러 대신 force-like 재실행 (사용자 명령 없이 자동 회복)
    # 단, 사용자가 명시적으로 strict_resume=True 옵션 시 에러 유지
    if self.project_config.get("strict_resume", False):
        raise AppError(code="step.resume_invalid", ...)
    logger.info("Step %s: stale → auto-rerun (resume mode default)", self.step_id)
    # falldown to force-like 실행
    self.invalidate_downstream()
    self.clear_checkpoint()
    # mode를 force로 격상 (다음 단계 force 분기 사용)
    mode = "force"
```

→ resume 모드가 stale을 자동 회복. dispatcher 멈춤 사고 종결.

#### 변경 3.2 — config_hash diff 로깅

**파일**: `backend/app/core/applicability.py` 또는 별도 utility

```python
def _diff_project_config(old: Optional[Dict], new: Dict) -> Dict[str, Any]:
    """project_config diff — config_hash mismatch 시 어느 key가 바뀌었는지 로그용."""
    if not old:
        return {"_changed": "snapshot 없음 (legacy cp)"}
    diff = {}
    all_keys = set(old.keys()) | set(new.keys())
    for k in all_keys:
        if old.get(k) != new.get(k):
            diff[k] = {"old": old.get(k), "new": new.get(k)}
    return diff
```

#### 변경 3.3 — `force category=image`의 transitive 동작 명시 변경

**파일**: `analysis_dispatch_service.py:139-178`

**Before** (transitive auto-include이 기본):
```python
# category="image" 만 cross-category transitive deps 자동 포함.
included = set(direct)
queue = list(direct)
while queue:
    sid = queue.pop()
    meta = get_manifest_dict(sid) or {}
    for dep in meta.get("depends_on", []):
        ...
        included.add(dep); queue.append(dep)
```

**After** (사용자 명시 필요):
```python
# category="image": 기본 동작 = image step만 force (transitive 안 함).
# transitive 의존 처리는 별도 옵션 `auto_include_deps=True` 시만 활성.
# 이전 동작(transitive 자동) 의도: silent miss 방지. 새 동작 의도: 사용자 명령 사고 방지.
# 두 trade-off 의 균형 — silent miss는 prerequisite 검증으로 대체 (아래).

# 사용자가 image step force할 때 의존 analysis가 미완료면 prerequisite 검증으로 거부.
# 즉 전체 invalidate 대신 명확한 에러 메시지로 사용자가 다음 행동 선택.
incomplete_deps = []
for sid in direct:
    meta = get_manifest_dict(sid) or {}
    for dep in meta.get("depends_on", []):
        # dep step의 status 확인 (DB step_run 또는 cp)
        if not _is_step_completed(db, project_id, episode_id, dep):
            incomplete_deps.append((sid, dep))

if incomplete_deps:
    raise AppError(
        code="dispatch.deps_incomplete",
        message=f"image step 실행 전 다음 analysis가 필요합니다: {set(d for _, d in incomplete_deps)}. "
                f"`category=all` 사용 또는 해당 analysis 먼저 실행하세요.",
        status_code=400,
    )

return direct
```

#### 변경 3.4 — API 명세 (mode/category 의미 표)

**파일**: `docs/architecture/dispatcher-modes.md` (신규)

| `mode` | `category` | 동작 | 사용 예 |
|--------|-----------|------|--------|
| `resume` | `all`/`analysis`/`image` | completed step skip + 미완료/stale 자동 회복 | 일반 진행 (정상 워크플로우) |
| `force` | `analysis` | analysis step만 강제 재실행 | analysis 결과 폐기하고 다시 |
| `force` | `image` | image step만 강제 재실행. analysis 미완료면 에러 | 이미지만 다시 (단일 step force와 비슷) |
| `force` | `all` | 전체 강제 재실행 (text_cleanup부터) | 처음부터 다시 |

#### 검증 (Fix 3)

```bash
# 단위 테스트
backend/tests/services/test_dispatcher_modes.py
- test_resume_skips_completed
- test_resume_auto_recovers_stale (← M3 핵심)
- test_resume_with_strict_flag_raises (legacy)
- test_force_image_blocks_when_analysis_incomplete (← M4 핵심)
- test_force_image_succeeds_when_analysis_complete
- test_force_all_invalidates_everything
```

---

### Fix 4 — manifest 자동 복구 (M5 해소)

**목표**: backend kill 후 manifest.json 비어있으면 가장 최근 archive를 자동 복구.

#### 변경 4.1 — `load_checkpoint`에 archive fallback

**파일**: `step_runner.py:121-131`

**Before**:
```python
def load_checkpoint(self):
    manifest_path = self._cp_dir / "manifest.json"
    if not manifest_path.exists():
        return None
    try:
        data = json.loads(manifest_path.read_text(encoding="utf-8"))
        return data
    except Exception as exc:
        logger.warning("Checkpoint load failed for %s: %s", self.step_id, exc)
        return None
```

**After**:
```python
def load_checkpoint(self):
    manifest_path = self._cp_dir / "manifest.json"
    
    # Primary: manifest.json 직접 로드
    if manifest_path.exists():
        try:
            return json.loads(manifest_path.read_text(encoding="utf-8"))
        except Exception as exc:
            logger.warning("Checkpoint load failed for %s: %s. Trying archive fallback.", self.step_id, exc)
    
    # Fallback: 최신 archive (manifest_TIMESTAMP.json) 자동 복원
    # 발생 시나리오: force 직후 backend kill → manifest.json 빈 상태 + archive 살아있음.
    archive = self._find_latest_archive()
    if archive:
        try:
            data = json.loads(archive.read_text(encoding="utf-8"))
            # manifest.json 자동 복구 — atomic 으로 archive → manifest
            atomic_write_json(manifest_path, data)
            logger.warning(
                "Step %s: manifest.json missing/corrupted. Auto-restored from archive %s",
                self.step_id, archive.name,
            )
            return data
        except Exception as exc:
            logger.error("Archive fallback failed for %s: %s", self.step_id, exc)
    
    return None


def _find_latest_archive(self) -> Optional[Path]:
    """최신 manifest_TIMESTAMP.json archive 찾기."""
    if not self._cp_dir.exists():
        return None
    pattern = re.compile(r"^manifest_(\d{8}_\d{6})(?:_[a-zA-Z0-9_-]+)?\.json$")
    archives = []
    for p in self._cp_dir.iterdir():
        m = pattern.match(p.name)
        if m:
            archives.append((m.group(1), p))
    if not archives:
        return None
    archives.sort(reverse=True)  # latest first
    return archives[0][1]
```

#### 변경 4.2 — backend startup 시 sweep (선택)

**파일**: `backend/app/main.py` 또는 startup hook

```python
@app.on_event("startup")
async def restore_orphaned_manifests():
    """startup 시 모든 step 디렉토리 sweep — 빈 manifest.json 자동 복원.
    
    사고 시나리오: force 직후 backend crash. 다수 step의 manifest 비어있음.
    이 sweep이 자동 복구.
    """
    from app.core.config import settings
    projects_dir = Path(settings.projects_dir)
    if not projects_dir.exists():
        return
    
    restored = 0
    for cp_dir in projects_dir.glob("**/checkpoints/episodes/*/*"):
        if not cp_dir.is_dir():
            continue
        manifest_path = cp_dir / "manifest.json"
        if manifest_path.exists():
            continue
        # archive 있으면 복원
        archive = _find_latest_archive_in_dir(cp_dir)
        if archive:
            try:
                shutil.copy2(archive, manifest_path)
                restored += 1
                logger.info("Restored manifest from %s", archive.name)
            except Exception:
                pass
    if restored:
        logger.warning("Startup manifest restore: %d files recovered", restored)
```

#### 검증 (Fix 4)

```bash
# 단위 테스트
backend/tests/core/test_checkpoint_archive_recovery.py
- test_load_returns_manifest_when_present
- test_load_falls_back_to_archive_when_missing (← M5 핵심)
- test_load_picks_latest_archive
- test_load_returns_none_when_no_archive_either
- test_startup_sweep_restores_orphaned_manifests
```

```bash
# E2E 검증 (사고 재현 시나리오)
1. force 호출 → 백업 생성됨
2. backend SIGKILL 흉내 (manifest.json 빈 상태)
3. backend 재시작
4. dispatcher가 모든 step 정상 인식 (자동 복구로 archive → manifest)
```

---

## Part 4 — 우선순위 + 실행 순서

### 우선순위

| Fix | 영향 | 시간 | 우선순위 |
|-----|------|------|---------|
| Fix 1 (글로벌 fallback) | **18 step 동시 안전** — graphic 시나리오 무한 반복 종결 | 60-75분 | **P0** (가장 큰 효과) |
| Fix 2 (sync 통일) | 4 sync_service partial cascade 가드 | 30분 | P1 |
| Fix 3 (dispatcher mode) | resume 자동 회복 + force category=image 명확화 | 45분 | P1 |
| Fix 4 (manifest 자동 복구) | backend crash 시 복구 자동화 | 30분 | P2 |
| **총** | **메타 원인 5개 모두 해소** | **~3시간** | |

### 실행 순서 권장

1. **Fix 1 먼저** (P0, 60-75분)
   - 가장 빈번하고 fan-out 큰 사고
   - 다른 fix와 독립적
   - test 함께 작성
2. **Fix 2** (P1, 30분)
   - 오늘 사고 직접 원인이라 즉시 효과
3. **Fix 3** (P1, 45분)
   - resume 자동 회복은 모든 작업 안정화
4. **Fix 4** (P2, 30분)
   - 다른 fix가 다 적용된 후 안전망
5. **회귀 검증** (60분)
   - E2E 시나리오 (graphic 시나리오로 재현)
   - 단위 테스트 통과 확인

### 짧은 path (긴급 unblock 후 fix)

긴급한 경우:
1. **현재 PID `0bb48ebf` 단순 unblock** (백업 manifest 복원 → background_classify 단일 force → cascade) — ~10분
2. 67 shot 이미지 생성 (~30분)
3. **그 후** Fix 1 → 2 → 3 → 4 순차 진행

---

## Part 5 — 검증 기준

### Fix 별 통과 기준

| Fix | 검증 방법 | 통과 기준 |
|-----|---------|---------|
| Fix 1 | 18 step 무방비 step에 graphic 시나리오 입력 → 모두 통과 | opik trace에 sanitized/gpt_fallback 태그 표시 + 모든 step status=completed |
| Fix 2 | 1 shot fail 시나리오 → scene_still 63 row sync | scene_image_pipeline 통과 |
| Fix 3-resume | stale step 발생 시 dispatcher 자동 진행 | 사용자 force 명령 없이 정상 종료 |
| Fix 3-force=image | analysis 미완료 상태에서 image force | 명확한 에러 메시지, transitive invalidate 안 함 |
| Fix 4 | backend kill 후 재시작 | dispatcher가 모든 step 정상 인식 |

### 회귀 테스트 시나리오

**시나리오 A**: 새 PID + graphic 시나리오 (예: 금월도 1부) → 처음부터 E2E
- 모든 47 step 통과
- failed_count 0
- 시각 검증: 67 shot 이미지 생성

**시나리오 B**: 진행 중 backend kill
- force 후 backend kill
- backend 재시작 → 자동 복구
- resume 모드로 dispatcher 정상 진행

**시나리오 C**: graphic 단어 새 종류 (시나리오마다 다른 단어)
- safety.py에 없는 새 단어가 PROHIBITED trip
- Tier 2 sanitize 실패 → Tier 3 gpt fallback으로 통과
- 사후 safety.py에 단어 추가

---

## Part 6 — Out of Scope (이번 작업 제외)

- visual_world_rules의 `director_notes` 자체 prompt 개선 (이미 v5 적용됨, 추가 회귀 시 별도)
- LiteLLM/Gemini SDK 업그레이드 (~3.x 모델 family 정책은 외부 통제 불가, fallback로 우회)
- Vertex AI 전환 (별도 inflight)
- t2i_review의 cascade 정책 (현재 1-pass, 별도 작업)

---

## Part 7 — 작성 후 자체 검토 (4회 통과)

| 검토 차수 | 발견 이슈 | 조치 |
|---------|--------|-----|
| 1차 | Fix 1의 `enable_fallback=False`가 의미 모호 | "legacy 호환" 명시 추가 |
| 2차 | Fix 2의 `is_cp_syncable`이 외부 의존 list 매개변수 받음 — sync_service별로 다른 키 | `data_keys` 매개변수로 명시, 호출자가 자신의 핵심 list 명시 |
| 3차 | Fix 3의 resume 자동 회복이 `strict_resume` 옵션 없으면 무조건 force 실행 — 사용자가 의도한 동작이 아닐 수도 | 기본은 자동 회복, project_config에 `strict_resume=True` 시 legacy 동작 (기존 사용자 호환) |
| 4차 | Fix 4의 sweep이 startup 시 모든 프로젝트 디렉토리 traverse — 큰 디렉토리에서 느릴 수 있음 | startup sweep는 선택 (변경 4.2). load_checkpoint의 fallback (변경 4.1)이 lazy + 충분 |

### 추가 검토: 사용자 절대 규칙 위반 여부

| 사용자 규칙 (CLAUDE.md / 메모리) | Fix 영향 | 위반 |
|---|---|---|
| "데이터를 절대로 자르지 마라" | sanitize는 LLM 호출 직전에만 적용, 원본 데이터는 보존 | ❌ 위반 없음 |
| "DB/프로젝트 파일 삭제 금지" | manifest 자동 복구는 추가 동작, 기존 archive 보존 | ❌ 위반 없음 |
| "프롬프트 파일 덮어쓰기 금지" | safety.py는 신규 모듈, 기존 prompt 변경 없음 | ❌ 위반 없음 |
| "scene_still sync는 UPSERT" | 기존 UPSERT 정책 그대로, partial 가드만 풀음 | ❌ 위반 없음 |
| "코드 리뷰 Codex + Claude 병행" | 이 계획서 자체는 Claude 작성. 구현 후 codex review 필수 | ✅ 별도 진행 |

---

## Part 8 — 진행 결정 요청

세 가지 path 중 선택:

**Path A — 긴급 unblock 우선**:
1. 현재 PID 백업 manifest 복원 (~10분)
2. 67 shot 이미지 생성 (~30분)
3. **그 후** Fix 1~4 순차 (~3시간)

**Path B — 근본 fix 먼저**:
1. Fix 1~4 모두 (~3시간)
2. 새 PID로 처음부터 E2E (~1.5시간)

**Path C — 병렬**:
1. Fix 1 즉시 적용 (~75분)
2. 동시에 현재 PID 단순 unblock (~10분)
3. Fix 2~4 + 새 PID E2E 회귀 검증 (~2시간)

---

**작성 완료. 다중 검토 4회 통과. 코드 위치 모두 확인. 진행 path 선택 요청.**