# Resume Architecture Fix — Design Spec

**Date**: 2026-05-07
**Status**: DRAFT_REVIEWED_PATCHED_V5 (Plan v1 NEEDS_REVISION → implementation contract clarification — caller pre-capture / `_execute_and_finalize` / verify_completion fixture / script quarantine / AppError contract / `_get_step_run` dict shape)
**Author**: Claude (Opus 4.7) + 사용자 directed review (Codex auxiliary)
**Trigger**: PID `80f62523` E2E 중 t2i_review → scene_detail cascade 폭주 사고

---

## §1. Background

### 1.1 사고 timeline (2026-05-07)

1. PID `80f62523` dispatch (mode=resume, category=all) 시작 (Gemini 회복 후)
2. `scene_detail` (gemini-pro × 30 segments) 8.5분 만에 63/63 완료
3. `shot_dependency_t2i` (gpt-mini, 4분) 63/63 완료
4. `t2i_review` (gpt-5.5) 1/1 완료 — entity_t2i 66 fix + scene_detail 3 fix 적용
5. `world_guide` 완료
6. 사용자가 "t2i_review부터 재실행" 요청 → dispatch 정지 + `step_run.t2i_review.status='stale'` SET
7. dispatch 재시작 → `[WARNING] Step scene_detail stale detected: verify failed: ["3 owned_validation sentinel_drifted: ...", "47 render_prompt_card drift/missing: ..."]`
8. scene_detail 이 force-like 자동 재실행 진입 → 사용자가 다시 정지

### 1.2 사고가 드러낸 architectural defects

사용자 분석 (Codex auxiliary review) 이 짚은 BLOCKING 5건:

| # | Defect | Evidence |
|---|---|---|
| **B1** | 단일 step resume 경로가 StepRunner resume 검증을 우회 | `step_execution_service.py:169-200` cp=None → skip / `step_runner.py:546` cp=None → stale + force-like (정반대 동작) |
| **B2** | DB-level 실행 lock 부재 | `task_registry.py:8` process-local `Dict + threading.Lock` 만 / 직접 script (`dispatch_pid_80f62523.py:96`) 는 task_registry 우회 |
| **B3** | resume mismatch 가 너무 넓은 project_config hash 에 묶여 있음 | `step_runner.py:347` step-local 없으면 project_config 전체 hash. 무관 설정 변경도 cascade |
| **B4** | verify signal 이 자동 force 권한을 가짐 (crash + incomplete 모두) | `step_runner.py:842` `_safe_verify_completion` 이 모든 예외를 `is_complete=False` 로 변환 / `step_runner.py:556-600` 단일 `mismatch + force-like` 경로 |
| **B5** | running/failed/partial/stale/pending 모두 force-like | `step_runner.py:608` 5개 status 동일하게 `mode="force"` |

IMPORTANT 3건:
- **I1** dispatch script wrapper DB session 긴 batch 동안 유지 — **stale evidence**, 이미 short-session 으로 fix 됨 (`dispatch_pid_80f62523.py:71-94`)
- **I2** `resume_sensitive` 사용처 한정 (shot_selection toggle 만) — `step_manifest.py:33,454,566,638` + `shot_selection_service.py:201`. cleanup category, 본 사고 주범 X
- **I3** auto-rerun 이 `invalidate_downstream()` recursive — `step_runner.py:641` cascade 폭주 amplifier

### 1.3 본 사고와 BLOCKING 매핑

오늘 cascade 폭주는 **B4 + B5 + I3 결합 결과**:

```
t2i_review 가 scene_detail.t2i_prompt in-place 수정 (의도된 1-pass policy)
  → owned_validation.t2i_prompt_hash drift (sentinel 미갱신)
  → next resume → scene_detail.verify_completion 이 sentinel drift 감지 → is_complete=False
  → step_runner.py:556-558 mismatch=f"verify failed: {missing}"
  → step_runner.py:600 mode="force" 자동 격상  [B4 + B5]
  → step_runner.py:641 invalidate_downstream() recursive  [I3]
  → scene_detail downstream (shot_dependency_t2i / scene_image_pipeline 등) 모두 stale
  → 다음 resume 에서 또 cascade
```

좁은 fix (t2i_review 가 sentinel 갱신하면 끝) 는 표면 — 진짜 root cause 는 **모든 verify 신호가 자동 force 권한을 보유**하는 것. 다음 새 invariant 추가 시 (G3.2 → G4.1 → Phase 9.2 history) 또 까먹으면 같은 cascade 재발.

### 1.4 반복되는 동일 클래스 버그 history

scene_detail cp invariant 추가 timeline:
- **G3.2** (2026-05-04): `owned_validation` sentinel + `t2i_prompt_hash`, `owned_hash`, `camera_direction_hash`, `validator` 도입
- **G4.1** (2026-05-04): `render_prompt_card` + `render_prompt_card_hash` top-level field 도입
- **Phase 9.2** (2026-04-30): t2i_review v2 (gemini-flash → gpt-5.5 swap, 신규 검증 룰)

매 리팩토링마다 scene_detail cp 에 새 invariant 추가 → t2i_review 의 mutation 후 자동 갱신 메커니즘 없음 → 같은 클래스 cascade 반복. 사용자 짜증의 직접 근원.

### 1.5 step_run schema (참고)

`backend/app/core/database.py:67-94` 의 실제 정의:

```sql
CREATE TABLE step_run (
    id text PRIMARY KEY,                  -- UUID, claim 시점에 발급 필수
    project_id text NOT NULL,
    episode_id text NOT NULL,
    step_id text NOT NULL,
    status text NOT NULL DEFAULT 'pending',
    run_id text,                          -- worker 실행 instance ID, owner check 용
    resolved_model text,
    input_hash text,
    upstream_revision text,
    prompt_version text,
    applicable_count integer,
    completed_count integer DEFAULT 0,
    failed_count integer DEFAULT 0,
    error_message text,
    result_summary text,
    started_at text,                      -- timestamptz string (text type, 캐스팅 필요)
    completed_at text,
    created_at text NOT NULL,             -- text type
    updated_at text NOT NULL,
    sync_status text,
    sync_error text,
    synced_at text,
    recovery_count integer NOT NULL DEFAULT 0,
    last_recovery_reason text,
    UNIQUE (project_id, episode_id, step_id)
)
```

**SQL 작성 시 주의**:
- `id`, `created_at`, `updated_at` 은 NOT NULL. INSERT 시 모두 명시 필수.
- `started_at`, `completed_at` 은 nullable text — `::timestamptz` 캐스팅 필요 (NOW() 비교 시).
- `started_at` parse 실패 가능 — null 또는 invalid format 모두 별도 처리.

---

## §2. Architectural Decisions

### 2.1 Resume signal 분류 (4 origin)

verify_completion / cp_mismatch / crash 가 모두 동일한 `mismatch + force-like` 경로로 합쳐지는 게 root cause. 4 origin 분리:

```python
# integrity_report.py 확장
@dataclass(frozen=True)
class CompletionReport:
    is_complete: bool
    missing: list[str]
    severity: Literal["clean", "partial", "missing"]
    metadata: dict
    origin: Literal[
        "clean",
        "artifact_missing",   # DB row / PNG / cp 파일 부재
        "contract_drift",     # schema_version / config_hash mismatch / loader contract / AppError
        "invariant_drift",    # sentinel/hash drift (mutator/manual/unknown)
        "verify_crashed",     # verifier 의 unexpected exception
    ] = "artifact_missing"   # backward-compatible default
```

### 2.2 Origin별 처리 정책

| origin | 처리 | invalidate_downstream |
|---|---|---|
| `clean` | skip | — |
| `artifact_missing` | **`rerun_self`** (auto-recovery) | **False** |
| `contract_drift` | **block** (`step.resume_invalid` raise, 사용자 명시 force 요구) | — |
| `invariant_drift` | **origin 분기** (Phase 2 #6 — mutator/manual/unknown) | False (mutator 시), block (manual/unknown) |
| `verify_crashed` | **fail-fast** (`step.verify_crashed` raise, 자동 force 금지) | — |

**핵심 차단선**: 자동 recovery 는 `rerun_self` (자기 step 만 rerun, downstream invalidate **없음**). 기존 `mode="force"` 경로 (cleanup_artifacts + invalidate_downstream) 는 사용자 명시 force 일 때만.

### 2.3 ResumeDecision 단일 판정자 (V2 patch I1 — enum 확장)

```python
class ResumeAction(Enum):
    """ResumeDecision.action — claim 후 실행 분기를 결정.

    NEEDS_SPEC_PATCH_V2 I1: pseudo-code 가 STALE_RUNNING_RECOVERY / FORCE_EXPLICIT
    언급하지만 enum 부재 → 구현자가 bool/문자열 분기로 흐를 위험. 명시 enum 으로 고정.
    """
    SKIP = "skip"                            # 이미 완료, cp 일치 — claim 안 함
    RERUN_SELF = "rerun_self"                # auto-recovery — 자기 step 만 rerun, downstream 보존
    FORCE_EXPLICIT = "force_explicit"        # 사용자 명시 force — 기존 cleanup + invalidate_downstream 경로
    STALE_RUNNING_RECOVERY = "stale_running_recovery"  # timeout 초과 running steal
    BLOCK = "block"                          # contract_drift / verify_crashed / unknown invariant_drift / running BLOCK
    NOT_APPLICABLE = "not_applicable"        # applicability 미달

@dataclass(frozen=True)
class ResumeDecision:
    """판정 결과 + 사유 — 로깅/감사 위해 reason 동반.

    V3 patch B1: STALE_RUNNING_RECOVERY 시 expected_started_at + expected_run_id 동반
    — claim SQL 의 atomic steal 조건에서 SQL cast 없이 (text 비교만) 검증.
    """
    action: ResumeAction
    reason: str               # 사유 ("cp clean", "stale started_at=...", "invariant_drift mutator", ...)
    origin: Optional[str] = None  # CompletionReport.origin (verify 결과 시)
    expected_started_at: Optional[str] = None  # STALE_RUNNING_RECOVERY 시 — text exact match (cast X)
    expected_run_id: Optional[str] = None      # STALE_RUNNING_RECOVERY 시 — text exact match
```

`StepRunner.run`, `start_step`, `run_steps_batch` 모두 이 단일 판정자 사용. `start_step` 의 자체 skip 판단 (`step_execution_service.py:169-200`) 제거.

**V5 patch S6 — AppError contract (Codex NEEDS_REVISION BLOCKING #5 반영)**:

ResumeAction.BLOCK / NOT_APPLICABLE 처리 시 raise 하는 AppError 는 현재 signature 만 사용:

```python
# backend/app/core/errors.py:5 — 현재 contract
class AppError(Exception):
    def __init__(self, code: str, message: str, status_code: int = 400):
        ...
```

- `extra` keyword **금지** — `__init__` 가 받지 않음 → `TypeError`. v3/v4 spec 의 `extra={"origin": ...}` 예시는 잔존 결함 (v5 에서 모두 제거).
- origin / decision context 는 `message` 에 inline (예: `f"{reason} (origin={origin})"`).
- AppError 확장 (extra/details/context 필드 도입) 은 본 spec scope 외 — 별도 spec 필요.
- **Plan v2 / implementation verification gate** 에서 모든 AppError raise 자리에 `extra` keyword 사용 0건 검증 (grep `AppError\(.*extra=` → 0 lines). 현 Plan v1 (`docs/superpowers/plans/2026-05-07-resume-architecture-fix-implementation.md:2628`) 에 `extra=...` 잔존 — Plan v2 P3 / 추가 a 에서 제거 의무.

**중요 (B1 patch v1 + I4 patch v2 반영)**:
- ResumeDecision 평가는 **non-mutating** (status='running' 으로 바꾸지 않음).
- atomic claim 은 평가 결과가 `RERUN_SELF` / `FORCE_EXPLICIT` / `STALE_RUNNING_RECOVERY` 일 때만.
- `SKIP` / `BLOCK` 은 claim 안 함 (running leak 차단).
- `NOT_APPLICABLE` 은 claim 안 하지만 `_mark_not_applicable()` 호출 — DB step_run + checkpoint 에 not_applicable 기록 (V2 patch I4: 현재 동작 보존).

### 2.4 modifies_checkpoints semantic 강화 (D4 — 후속 PR)

mutator (= editorial step, 현재 t2i_review 만) 가 victim cp 수정 시:

1. mutator 는 invariant 갱신 helper 의무 호출 (Block A 에서 t2i_prompt_hash refresh 로 시작, D4 에서 일반화)
2. step_runner 가 mutator 종료 시점에 victim cp 의 `verify_completion` auto-invoke
3. **저장 순서**:
   - victim cp 저장 **전** archive 생성 (`_archive_manifest` helper 재사용)
   - 저장 후 verify 실패 시 → archive rollback 또는 victim cp 에 `_mutation_invalid` marker 기록
   - mutator step status='failed' (silent 차단)
   - victim step 자동 stale 금지
4. silent corruption 방지 — verify 실패 시 mutator 가 책임짐 (victim 자동 rerun X)

### 2.5 DB advisory lock semantic — INSERT ON CONFLICT + atomic steal (V2 patch B1 + B2)

**선택**: `step_run` row atomic claim — INSERT ON CONFLICT + **expected-match steal** (V3 patch B1: SQL cast 제거).

**V3 patch B1 핵심 변경**:
- 이전 v3 의 `started_at::timestamptz < NOW() - interval` 캐스팅이 invalid text row 하나에 query 전체 실패 위험.
- timeout 판정은 Python `_evaluate_running_state()` 에서 이미 수행 — SQL 은 단순 **text exact match** 만 (cast 없음).
- `_evaluate_running_state` 가 stale 판정 시 `expected_started_at` + `expected_run_id` (당시 read 값) 를 ResumeDecision 에 담아 claim 으로 전달 → SQL 은 그 값으로 atomic match.
- read 와 claim 사이에 다른 worker 가 갱신했으면 (예: 정상 완료 또는 새 claim) match 실패 → claim 실패 → safer (오래된 stale 정보로 steal 안 함).

```sql
-- claim acquire (first-run 또는 기존 row 모두 처리, expected-match 로 stale steal)
-- 모든 NOT NULL 컬럼 (id, created_at) 명시 (V2 patch B1: schema 위반 방지)
INSERT INTO step_run (
    id, project_id, episode_id, step_id, run_id, status,
    started_at, created_at, updated_at
)
VALUES (
    :new_step_run_id,                      -- new UUID per claim attempt
    :pid, :eid, :sid, :run_id, 'running',
    NOW()::text, NOW()::text, NOW()::text
)
ON CONFLICT (project_id, episode_id, step_id) DO UPDATE
  SET run_id = EXCLUDED.run_id,
      status = 'running',
      started_at = EXCLUDED.started_at,
      updated_at = EXCLUDED.updated_at,
      recovery_count = step_run.recovery_count + CASE
        WHEN step_run.status = 'running' THEN 1 ELSE 0
      END
  WHERE step_run.status != 'running'
     OR (
       -- V3 patch B1: SQL cast 제거 — Python 에서 timeout 판정한 stale row 만
       -- expected_started_at + expected_run_id text 정확 매칭 시 steal.
       :allow_stale_steal IS TRUE
       AND step_run.status = 'running'
       AND step_run.started_at = :expected_started_at
       AND step_run.run_id = :expected_run_id
     )
RETURNING run_id, (xmax = 0) AS is_insert;
```

- `allow_stale_steal`: ResumeAction 이 `STALE_RUNNING_RECOVERY` 일 때 True, 그 외 False
- `expected_started_at` / `expected_run_id`: stale 판정 시 read 한 값 (text), `RERUN_SELF` / `FORCE_EXPLICIT` 은 NULL (steal 발동 안 함)
- 결과 row 1개 → claim 성공 (`is_insert=True` 면 first-run, `False` 면 기존 row 복용 또는 expected-match stale steal)
- 결과 row 0개 → 다른 worker 가 이미 잡았거나 read 후 갱신됨 → `step.already_running` raise
- finally 에서 status='completed' / 'failed' / 'partial' 로 release (단 **WHERE run_id=:self.run_id** 조건 필수 — V1 patch I1)
- connection pool leak 없음 (트랜잭션과 무관, atomic UPSERT)
- recovery_count 는 stale steal 시 +1 (감사 로그)

**started_at parse 실패 / NULL 행 처리**:
- Python `_evaluate_running_state()` 가 parse 실패 시 BLOCK 반환 → claim 시도 자체 안 함 → SQL cast 위험 0.
- audit 도 cast 없이 SELECT 후 application 레벨에서 datetime.fromisoformat 시도 (§7.3).

**expected-match 의미**:
- read 시점의 `(started_at, run_id)` 텍스트가 claim SQL 실행 시점에도 동일 → 진짜 stale (그 worker 가 정상 완료 안 했음)
- 사이에 다른 worker 가 정상 완료 → started_at/run_id 변경 → match 실패 → claim 실패 (안전)

**거부 alternatives**:
- `pg_try_advisory_lock`: connection pool 에서 잡으면 lock leak — dedicated connection 의무 + finally release 의무 (복잡도 높음)
- `pg_try_advisory_xact_lock`: commit 때 풀려서 StepRunner 중간 commit 패턴과 부적합

---

## §3. Block A — t2i_review hotfix (~0.5일)

### 3.1 목적

본 사고 (scene_detail sentinel drift) 의 직접 원인 차단. 다른 BLOCKING 은 잔존 — Block C 까지 후 dispatch 재개.

### 3.2 변경 위치

`backend/app/core/steps/t2i_review_step.py:55-62` 의 mutation 사이트 직후에 sentinel refresh 추가.

추가 변경: `backend/app/modules/pipeline/t2i_review.py:222-360` 의 `_review_scene_detail` + `_apply_scene_fixes` (V1 patch B3).

### 3.3 Pipeline 수정 — deterministic item_id 기반 exact mutated path 반환 (V1 patch B3 + V3 patch I1)

**문제 (현재 코드)**:
- `_apply_scene_fixes()` 가 `scene_index` 만 보고 첫 matching scene 의 첫 matching variation 수정. scene_detail 은 shot fan-out (각 scene 의 t2i_variations 가 shot 단위) → 같은 scene_index 가 여러 entry 일 수 있어 잘못된 shot 의 prompt/sentinel 갱신 위험.
- **V3 patch I1**: 단순히 `scene_index`/`var_index` 만 LLM 이 추론하면 같은 scene_index 가 여러 fan-out row 에 있을 때 모호. LLM 이 scene_list_idx 를 직접 추론하게 두는 것도 불안정 (구조 의존).

**수정 (V3 patch I1)**:

1. **deterministic `item_id` 사용** — t2i_review prompt 에 각 t2i_variation 의 `item_id` (예: `"S{scene_index}_V{var_index}_LIST{scene_list_idx}"` 또는 단순 sequential `"item_{n}"`) 를 넣고, schema `required` 로 `item_id` 받음. LLM 은 단순 echo 만 하면 됨 (추론 부담 0).
2. `_review_scene_detail()` 가 fix 객체에 `item_id` 포함.
3. application 코드에서 `item_id → (scene_list_idx, variation_idx)` map 빌드 (prompt 생성 시점에 확정).
4. `_apply_scene_fixes()` 가 `item_id` 로 정확한 위치 찾아 mutation 수행 + 실제 수정된 `(scene_list_idx, variation_idx)` 튜플 리스트 반환:

```python
# t2i_review.py 의 _apply_scene_fixes 시그니처 변경 (V3 patch I1: item_id 기반)
def _build_item_id_map(
    scene_detail_data: Dict
) -> Dict[str, Tuple[int, int]]:
    """t2i_variations 모두 순회 — item_id (deterministic) → (scene_list_idx, variation_idx) map 생성.

    item_id format: f"S{scene_index}_L{scene_list_idx}_V{variation_idx}"
                    (scene_index 는 scene 의 절대 번호, scene_list_idx 는 scenes[] 위치, V 는 variation index)
    LLM 이 prompt 의 item_id 를 echo 하면 application 이 map 으로 정확 위치 찾음.
    """
    m = {}
    for s_list_idx, scene in enumerate(scene_detail_data["scenes"]):
        scene_index = scene.get("scene_index")  # 절대 번호 (display)
        for v_idx, variation in enumerate(scene.get("t2i_variations", [])):
            item_id = f"S{scene_index}_L{s_list_idx}_V{v_idx}"
            m[item_id] = (s_list_idx, v_idx)
    return m


def _apply_scene_fixes(
    scene_detail_data: Dict, scene_fixes: List[Dict],
    item_id_map: Dict[str, Tuple[int, int]],
) -> Tuple[int, List[Tuple[int, int]]]:
    """
    LLM 이 schema required 로 echo 한 fix["item_id"] 로 정확 위치 매핑.

    Returns: (count, applied_indices) — count 는 실제 적용된 fix 수,
             applied_indices 는 (scene_list_idx, variation_idx) 튜플 리스트.
    """
    applied_indices = []
    count = 0
    for fix in scene_fixes:
        item_id = fix["item_id"]  # required by LLM schema
        if item_id not in item_id_map:
            raise AppError(
                code="t2i_review.unknown_item_id",
                message=f"LLM 이 반환한 item_id={item_id!r} 가 prompt 에 없음 — schema 위반",
            )
        s_list_idx, v_idx = item_id_map[item_id]
        # ... mutation 수행 ...
        applied_indices.append((s_list_idx, v_idx))
        count += 1
    return count, applied_indices
```

3. `run_t2i_review()` 의 반환 dict 에 `scene_applied_indices: List[Tuple[int, int]]` 필드 추가.
4. **schema 변경**: `prompts/_base/t2i_review/scene_schema.json` 의 fix 항목 required 에 `item_id: string` 추가. 기존 `scene_index`, `var_index` 는 보조 정보로만.
5. **prompt 변경**: t2i_review prompt 본문에 각 t2i_variation 의 `item_id` 명시 (예: 각 entry 에 `[item_id: S5_L4_V2]` prefix). LLM 추론 부담 0 — 단순 echo.

### 3.4 sentinel refresh 방식 (V1 보정 1 + V1 patch M2 helper 위치 혼합)

**`build_owned_sentinel()` 전체 재생성 X** — t2i_review 에는 `owned`, `camera_direction`, `is_close_framing` 원천 ctx 가 완전하지 않음. 잘못된 ctx 로 재생성 시 owned_hash / camera_direction_hash false drift 위험.

**helper 분리 (V1 patch M2)**:
- **traversal**: `t2i_review_step.py` local — scene_detail data 구조 의존
- **sentinel 단일 필드 갱신 primitive**: `_owned_helpers.refresh_t2i_prompt_hash(sentinel, t2i_prompt, where)` — D4 재사용 가능

```python
# _owned_helpers.py 신규 primitive
def refresh_t2i_prompt_hash(
    sentinel: Dict[str, Any], t2i_prompt: str, where: str = ""
) -> bool:
    """sentinel 의 t2i_prompt_hash 만 갱신. 다른 hash 필드 보존.

    sentinel shape 검증 후 갱신 — shape 위반 시 raise.
    Returns: True (갱신 발생) / False (이미 일치, no-op).
    """
    assert_owned_sentinel_shape(sentinel, where=where)
    new_hash = compute_t2i_prompt_hash(t2i_prompt)
    if sentinel["t2i_prompt_hash"] == new_hash:
        return False
    sentinel["t2i_prompt_hash"] = new_hash
    return True
```

```python
# t2i_review_step.py local helper (traversal)
def _refresh_scene_detail_sentinels(
    scene_detail_data: Dict, applied_indices: List[Tuple[int, int]]
) -> int:
    """t2i_review 가 수정한 (scene_list_idx, variation_idx) 항목의 owned_validation
    sentinel 의 t2i_prompt_hash 만 갱신.

    applied_indices 외 항목은 건드리지 않음 (보정 1 — 좁게).
    sentinel 누락 / shape 위반 시 raise → t2i_review failed.
    """
    from app.core.steps._owned_helpers import refresh_t2i_prompt_hash
    refreshed = 0
    scenes = scene_detail_data["scenes"]
    for s_list_idx, v_idx in applied_indices:
        scene = scenes[s_list_idx]
        variation = scene["t2i_variations"][v_idx]
        sentinel = variation.get("owned_validation")
        if not sentinel:
            raise AppError(
                code="t2i_review.owned_validation_missing",
                message=(
                    f"scenes[{s_list_idx}].t2i_variations[{v_idx}].owned_validation 없음 — schema 위반"
                ),
            )
        if refresh_t2i_prompt_hash(
            sentinel,
            variation["t2i_prompt"],
            where=f"t2i_review_refresh@s{s_list_idx}_v{v_idx}",
        ):
            refreshed += 1
    return refreshed
```

`t2i_review_step._execute()` 수정:

```python
# t2i_review_step.py:55-62 → 수정
if review_result["scene_applied"] > 0:
    try:
        refreshed = _refresh_scene_detail_sentinels(
            scene_detail_data,
            review_result["scene_applied_indices"],  # V1 patch B3 — exact path
        )
        logger.info("t2i_review: sentinel refreshed (%d entries)", refreshed)
    except (AppError, KeyError, IndexError) as exc:
        # sentinel refresh 실패 → t2i_review failed (silent corruption 차단)
        # KeyError/IndexError 도 catch — applied_indices 가 잘못된 path 가리킬 가능성
        raise AppError(
            code="t2i_review.sentinel_refresh_failed",
            message=f"sentinel refresh 실패: {exc}",
        ) from exc
    self._save_checkpoint_data("scene_detail", scene_detail_data)
```

### 3.5 render_prompt_card_hash (V1 보정 2 — 가볍게)

**이번 hotfix scope 외**.

근거: t2i_review 가 card payload (`render_prompt_card` field) 를 직접 수정하지 않으면 card hash 재계산은 false drift 유발. card 재생성에는 ctx (entity description, scene metadata 등) 필요.

**가벼운 assert 만 추가 (V5 patch S1 — caller 책임 명시)**:

**중요 — pre/post capture 는 caller 책임 (V5 patch S1, Codex NEEDS_REVISION BLOCKING #2 반영)**:

`run_t2i_review()` 는 내부에서 `_apply_entity_fixes` + `_apply_scene_fixes` 로 `scene_detail_data` 를 mutate 한 뒤 반환한다 (`backend/app/modules/pipeline/t2i_review.py:60-65`). 따라서 `pre_card_state` 를 review 결과 받은 **후** capture 하면 post-mutation state 를 pre 로 저장 → assert 무의미. **반드시 caller (`T2iReviewStep._execute`) 가 `run_t2i_review()` 호출 전 capture** 해야 한다.

**Caller 호출 순서 (필수)**:

```
1) pre_card_state = _capture_card_hash_state(scene_detail_data)        # mutation 전, 전체 capture
2) review_result = run_t2i_review(...)                                  # 내부 mutation 발생
3) if review_result["scene_applied"] > 0:
4)     applied_indices = review_result["scene_applied_indices"]         # AC-A6
5)     _refresh_scene_detail_sentinels(scene_detail_data, applied_indices)
6)     _assert_card_hash_unchanged(scene_detail_data, applied_indices, pre_card_state)
```

**Helper signature**:

```python
# _owned_helpers.py 또는 t2i_review_step.py local helper
def _capture_card_hash_state(
    scene_detail_data: Dict,
) -> Dict[Tuple[int, int], Optional[str]]:
    """모든 scene/variation 의 render_prompt_card_hash 를 mutation 전 capture.

    전체 (scene_idx, var_idx) 를 키로 — applied_indices 의 super-set 보장 (under-coverage 위험 0).
    Caller 책임 (V5 patch S1): T2iReviewStep._execute 가 run_t2i_review 호출 직전 호출.
    """
    state: Dict[Tuple[int, int], Optional[str]] = {}
    for s_idx, scene in enumerate(scene_detail_data.get("scenes", [])):
        for v_idx, v in enumerate(scene.get("t2i_variations", [])):
            state[(s_idx, v_idx)] = v.get("render_prompt_card_hash")
    return state


# t2i_review_step.py: mutation 직후 (refresh sentinels 후)
def _assert_card_hash_unchanged(
    scene_detail_data: Dict, applied_indices: List[Tuple[int, int]],
    pre_state: Dict[Tuple[int, int], Optional[str]],
) -> None:
    """t2i_review 는 card payload 를 변경하면 안 됨. 변경 detect 시 raise.

    pre_state: caller 가 mutation 전 _capture_card_hash_state() 로 만든 dict — 전체 매핑.
    applied_indices: review_result["scene_applied_indices"] (AC-A6) — 실제 mutate 된 (s_idx, v_idx) 만.
    """
    for s_idx, v_idx in applied_indices:
        v = scene_detail_data["scenes"][s_idx]["t2i_variations"][v_idx]
        post_hash = v.get("render_prompt_card_hash")
        if pre_state.get((s_idx, v_idx)) != post_hash:
            raise AppError(
                code="t2i_review.card_hash_unexpected_change",
                message=(
                    f"t2i_review 가 card hash 를 변경함 — scope 외 mutation: "
                    f"scenes[{s_idx}].t2i_variations[{v_idx}]"
                ),
            )
```

card hash drift 는 Block D4 에서 정식 처리 (modifies_checkpoints rollback semantic).

### 3.6 entity_t2i sentinel

entity_t2i 는 verify_completion override 없음 → sentinel drift 감지 안 됨 → 본 hotfix scope 외. asymmetric silent corruption 위험은 잔존하지만 이번 cascade 의 trigger 아님. **D5 에서 정식 처리**.

### 3.7 Block A acceptance criteria

- ✅ AC-A1: `_owned_helpers.refresh_t2i_prompt_hash()` primitive — sentinel shape 검증 + t2i_prompt_hash 만 갱신 (다른 hash 보존)
- ✅ AC-A2: `t2i_review_step._execute()` 가 mutation 후 helper 호출, 실패 시 `t2i_review.sentinel_refresh_failed` raise → status='failed'
- ✅ AC-A3: render_prompt_card_hash 변경 없음을 assert 로 검증 — false drift 차단
- ✅ AC-A4: 단위 테스트 — fix 후 next resume 시 scene_detail.verify_completion 통과 (drift 0)
- ✅ AC-A5: 회귀 — t2i_review 가 fix 0건 적용 시 sentinel 은 그대로 (no-op)
- ✅ AC-A6: `_apply_scene_fixes()` 가 실제 수정된 `(scene_list_idx, variation_idx)` 튜플 리스트 반환. fan-out 시나리오 (같은 scene_index 다수) 회귀 테스트 — 잘못된 shot mutation 0건

---

## §4. Block B — StepRunner decision 정리 (~1일)

### 4.1 변경 범위

| 위치 | 변경 |
|---|---|
| `step_runner.py:465-471` (`_get_step_run`) | tuple `(status, completed_count, applicable_count)` → **dict** `{status, run_id, started_at, completed_count, applicable_count, recovery_count}` (V5 patch S7) — atomic claim / stale steal 이 run_id+started_at 의존 |
| `step_runner.py:546-600` | `mismatch` 단일 문자열 → `ResumeDecision` dataclass 반환. force-like 격상 분기 제거 |
| `step_runner.py:608` (running/failed/partial/stale/pending) | running 분리 — `started_at + timeout` 초과만 `STALE_RUNNING_RECOVERY`, 그 외 BLOCK |
| `step_runner.py:641` (force 분기) | auto-recovery 시 `invalidate_downstream=False` (rerun_self 별도 경로) |
| `step_runner.py:652-717` (run() try 블록) | `_execute_and_finalize()` helper 로 추출 (V5 patch S2) — `_execute_rerun_self` / `_execute_force` 양쪽 공유. finalization 6 단계 보존 의무 |
| `step_runner.py:842` (`_safe_verify_completion`) | crash → `verify_crashed` raise (자동 force 금지). entry/exit 양쪽 |
| `step_execution_service.py:169-200` | 자체 skip 판단 제거 → 항상 background worker → StepRunner.run 단일 판정 |
| `integrity_report.py` | `CompletionReport.origin` 추가 (default "artifact_missing") |
| `detail_steps.py:1133, 1276, 1373, 1404` (scene_detail.verify_completion) | sentinel/card drift / loader / card recompute 분류 (§4.3 표 참조) |

### 4.2 rerun_self 별도 실행 경로 (V1 보정 3)

**중요**: 기존 `mode="force"` 경로 재사용 X. force 는 `cleanup_artifacts() + invalidate_downstream(delete_cp=True)` cascade 를 발동 — 이번 사고의 amplifier.

**V5 patch S2 — finalization 공통 보존 (Codex NEEDS_REVISION BLOCKING #3 반영)**:

`_execute_rerun_self` / `_execute_force` 분리는 **사전 정책 (cleanup / invalidate cascade 여부)** 차이만 담당. 그 후의 finalization 블록은 두 경로 모두 동일하게 거쳐야 한다. 현 `step_runner.py:661-693` 의 다음 6 단계는 **반드시 공통 helper `_execute_and_finalize()` 로 보존**:

| # | 단계 | 현 위치 | 보존 의무 |
|---|---|---|---|
| 1 | `final_status` 계산 (failed/partial/completed) | `step_runner.py:665` | 필수 |
| 2 | exit verify (`_safe_verify_completion`) → 실패 시 partial 격상 | `step_runner.py:670-677` | 필수 (AC-B8) |
| 3 | `_update_step_run(final_status, require_owner=True)` | `step_runner.py:679-688` | 필수 (AC-C7) |
| 4 | `save_checkpoint({"status": final_status, **result})` | `step_runner.py:689` | 필수 |
| 5 | `_reset_recovery_counter()` (final_status=='completed' 시) | `step_runner.py:692-693` | 필수 |
| 6 | `modifies_checkpoints` cascade policy (1-pass default) | `step_runner.py:703-713` | 필수 |

**구조**:

```python
# step_runner.py 신규 _execute_rerun_self() — 정책만 담당
def _execute_rerun_self(self) -> Dict[str, Any]:
    """auto-recovery 정책 — cleanup/invalidate cascade 안 함, 그 후 공통 finalize."""
    # 사전 정책: 없음 (cleanup_artifacts / invalidate_downstream 호출 X)
    return self._execute_and_finalize()


# step_runner.py 신규 _execute_force() — cleanup + invalidate 후 공통 finalize
def _execute_force(self) -> Dict[str, Any]:
    """force_explicit 정책 — cleanup_artifacts + invalidate_downstream(delete_cp=True) 후 공통 finalize."""
    # 사전 정책: 사용자 명시 force
    self.cleanup_artifacts()
    self.invalidate_downstream(delete_checkpoints=True)
    return self._execute_and_finalize()


# step_runner.py 신규 _execute_and_finalize() — 두 경로 공통
def _execute_and_finalize(self) -> Dict[str, Any]:
    """_execute() 호출 + finalization 공통. 현 run() line 652-717 의 try 블록 직접 이식.

    포함 (V5 patch S2 — 모두 보존 의무):
      1) result = self._execute(mode)  +  self._last_execute_result = result
      2) final_status 계산
      3) exit verify (final_status='completed' 시) → 실패 시 partial 격상 (AC-B8)
      4) _update_step_run(final_status, require_owner=True, ...) (AC-C7)
      5) save_checkpoint({"status": final_status, **result})
      6) final_status='completed' 시 _reset_recovery_counter()
      7) modifies_checkpoints cascade (cascade_on=False default, 1-pass)
      8) return {"status": final_status, "result": result}
    """
    ...
```

`run()` resume 분기에서 ResumeAction 에 따라:
- `RERUN_SELF` → `_execute_rerun_self()` → 내부 `_execute_and_finalize()`
- `FORCE_EXPLICIT` → `_execute_force()` → 내부 `_execute_and_finalize()`
- `STALE_RUNNING_RECOVERY` → `_execute_rerun_self()` (downstream 보존, 죽은 worker 의 cp 가 남아있을 수 있음 → `_execute()` 가 그대로 사용 가능)

**검증 의무 (Plan v2 P3 보강 체크리스트)**:
- run() 의 기존 try 블록 (line 652-717) 의 모든 finalization 단계가 `_execute_and_finalize` 에 1:1 이식
- `_update_step_run` 호출이 `require_owner=True` 적용
- 단위 테스트 — `_execute_rerun_self` 호출 시 `_update_step_run` / `save_checkpoint` / `_reset_recovery_counter` 모두 호출 검증
- 단위 테스트 — `_execute_force` 호출 시 `cleanup_artifacts` + `invalidate_downstream(delete_checkpoints=True)` 호출 후 `_execute_and_finalize` 호출 검증

### 4.3 verify policy 점진 도입 + 분기 매핑 (V2 patch I2 — 더 세밀히)

`CompletionReport.origin` 4분류 모든 verifier 동시 수정 X — diff 큼. 1차 점진 적용:

| 위치 | 신호 | origin 분류 |
|---|---|---|
| `_safe_verify_completion()` 의 unexpected exception (entry/exit 양쪽) | 모든 비-AppError exception | **`verify_crashed`** (자동 force 금지) |
| `_safe_verify_completion()` 의 AppError | `step.contract_violation` / loader AppError 등 | **`contract_drift`** |
| `scene_detail.verify_completion` `loader_violations` (loader contract) | `_load_chain_bg_*` 의 shape mismatch / required key missing | **`contract_drift`** |
| `scene_detail.verify_completion` 의 `_g41_render_prompt_card()` AppError (recompute contract) | card recompute 시 AppError (예: `step.contract_violation`) | **`contract_drift`** |
| `scene_detail.verify_completion` 의 `_g41_render_prompt_card()` unexpected exception (예: KeyError, ValueError, ctx 누락) | recompute 시 비-AppError exception | **`verify_crashed`** |
| `scene_detail.verify_completion` 의 `sentinel.t2i_prompt_hash != computed` (V2 patch I2 — stored hash mismatch) | sentinel hash drift | **`invariant_drift`** |
| `scene_detail.verify_completion` 의 `render_prompt_card_hash` mismatch (stored vs computed) | card hash drift (stored 와 recomputed 비교) | **`invariant_drift`** |
| `scene_detail.verify_completion` 의 `owned_validation` missing / sentinel shape 위반 (V2 patch I2 — 택일 고정: contract_drift) | sentinel 누락 또는 shape 위반 (예: required field 누락) | **`contract_drift`** |
| `scene_detail.verify_completion` 의 `render_prompt_card` payload missing (top-level field 자체 부재) | stored card missing | **`contract_drift`** |
| `_check_cp_mismatch` (schema/config_hash) | StepRunner 내부 decision | **`contract_drift`** |
| 다른 verifier (image steps 등) | 변경 없음 | default `"artifact_missing"` (D1 점진 migrate) |

다른 verifier 는 D1 (전면 적용) 에서 점진 migrate. 1차 정책으로도 본 사고 차단 가능.

**분류 원칙 (V2 patch I2 명시)**:
- **AppError 계열 + loader contract** → `contract_drift` (구조적 위반, 사용자 force 요구)
- **unexpected exception** → `verify_crashed` (verifier 버그 가능, fail-fast)
- **stored vs computed hash mismatch** → `invariant_drift` (정상 sentinel 이지만 값 불일치)
- **stored sentinel/card 부재 또는 shape 위반** → `contract_drift` (schema/구조 위반)

**테스트 검증 의무 (V5 patch S4 — Codex NEEDS_REVISION BLOCKING #6 반영)**:

`SceneDetailStep.verify_completion()` 의 **실제 분기를 fixture 로 태우는 단위 테스트** 의무. `CompletionReport(origin=...)` 객체를 직접 build 하고 assert 만 하는 테스트는 분류 helper 의 deterministic 동작은 검증하나, `verify_completion` 안의 `loader_violations` / `failed_indices` / `contract_violations` / `sentinel_drifted` / `card_drifted` 우선순위 (`detail_steps.py:1358-1419`) 와 `severity` (missing vs partial) 분류 매핑은 검증하지 않음. **plan 의 Task B8 단위 테스트는 다음을 모두 fixture 로 직접 태움**:

| Branch | Fixture 요구 | 기대 origin |
|---|---|---|
| `loader_violations` 발생 (예: `_load_chain_bg_*` shape mismatch) | loader 가 violation 반환하도록 monkeypatch | `contract_drift` |
| `_g41_render_prompt_card()` AppError | recompute helper 가 AppError raise | `contract_drift` |
| `_g41_render_prompt_card()` unexpected exception (KeyError 등) | recompute helper 가 KeyError raise | `verify_crashed` |
| `sentinel_drifted` t2i_prompt_hash mismatch | scene_detail cp 의 sentinel.t2i_prompt_hash 를 stale 값으로 set | `invariant_drift` |
| `card_drifted` stored vs computed hash mismatch | render_prompt_card_hash 를 stale 값으로 set | `invariant_drift` |
| `owned_validation` missing 또는 sentinel shape 위반 (예: required field 누락) | sentinel 에서 validator 필드 제거 | `contract_drift` |
| `render_prompt_card` payload missing (top-level field 부재) | scene_detail cp 에서 render_prompt_card 자체 제거 | `contract_drift` |

각 fixture 는 실제 `SceneDetailStep.verify_completion()` 호출 → 반환된 `CompletionReport.origin` 검증. 우선순위 충돌 시나리오 (예: loader_violations + sentinel_drifted 동시) 도 1건 추가 — 어느 origin 이 우선되는지 명시적으로 검증.

### 4.4 invariant_drift 분기 (V1 보정 5의 일부, D4까지 임시 정책)

mutator origin 식별 메커니즘은 D4 에서 정식 도입 (modifies_checkpoints rollback semantic). Block B 1차 정책:

```python
# StepRunner._evaluate_invariant_drift()  (B 1차)
def _evaluate_invariant_drift(self, report: CompletionReport) -> ResumeDecision:
    """1차 정책: invariant_drift 는 무조건 BLOCK.

    이유: mutator origin 식별 메커니즘은 D4 에서 도입. 그 전까지는 unknown 분류 →
    safer default 인 BLOCK. 사용자가 force 명시해야 진행.

    D4 도입 후: mutator origin 일 때 mutator step failed 처리, victim BLOCK 해제.
    """
    return ResumeDecision(
        action=ResumeAction.BLOCK,
        reason=f"invariant_drift (D4 까지 임시 BLOCK 정책): {report.missing[:3]}",
        origin="invariant_drift",
    )
```

D4 이후엔 `_owned_helpers` 에 origin tracking 필드 추가 (`_mutation_origin: str` in sentinel) → mutator 분기 가능.

### 4.5 schema_version mismatch 임시 allowlist (V1 보정 6 + V1 patch I3)

**문제**: B 의 `contract_drift=BLOCK` 정책으로 schema bump 자동 cascade 도 막힘. 기존 운영은 entity_t2i `schema_version=1→2` 같은 변경 시 자동 rerun cascade 의도 사용.

**임시 정책 (B → D3 전까지) — V1 patch I3 반영, scene_detail 제거**:

```python
# step_manifest.py 에 임시 allowlist
# 한정적 — entity_t2i 만. scene_detail 제거 사유: rerun_self 시 downstream
# (shot_dependency_t2i / scene_image_pipeline 등) 보존이라 새 scene_detail 결과와
# 기존 downstream 사이 불일치 위험. scene_detail 은 사용자 명시 force 또는
# D3 manifest flag 도입 후 허용.
_LEGACY_SCHEMA_BUMP_ALLOWLIST = frozenset({
    "entity_t2i",   # schema_version 1→2 cascade 의도, downstream 영향 작음
    # expiry: D3 deploy 시점에 제거 (manifest.allow_auto_rerun_on_schema_bump 로 대체)
    # comment: scene_detail 등 downstream 큰 step 추가 금지 (rerun_self 시 drift 유발)
})

# StepRunner._evaluate_contract_drift()
def _evaluate_contract_drift(self, mismatch_reason: str) -> ResumeDecision:
    if "schema_version mismatch" in mismatch_reason:
        if self.step_id in _LEGACY_SCHEMA_BUMP_ALLOWLIST:
            return ResumeDecision(
                action=ResumeAction.RERUN_SELF,
                reason=f"legacy schema bump allowlist: {self.step_id}",
                origin="contract_drift",
            )
    return ResumeDecision(
        action=ResumeAction.BLOCK,
        reason=mismatch_reason,
        origin="contract_drift",
    )
```

**테스트 검증 (V2 질문 답변 #3)**: `_LEGACY_SCHEMA_BUMP_ALLOWLIST == frozenset({"entity_t2i"})` 정확 일치 단위 테스트 — 다른 step 추가 차단.

**D3 진입 후**: `manifest.allow_auto_rerun_on_schema_bump: bool = False` per-step flag. allowlist 제거.

### 4.6 running timeout (V1 보정 5)

```python
# step_runner.py running recovery (B2)
def _evaluate_running_state(self, existing) -> ResumeDecision:
    """running 자동 force 금지. started_at timeout 초과만 별도 recovery.
    started_at 파싱 실패/누락 → BLOCK (자동 rerun 금지)."""
    started_at_raw = existing.get("started_at")
    if started_at_raw is None:
        # 누락 → BLOCK (자동 rerun 금지)
        logger.warning(
            "[STALE_RUNNING] step=%s started_at NULL → BLOCK", self.step_id,
        )
        return ResumeDecision(
            action=ResumeAction.BLOCK,
            reason="running with started_at=NULL (manual investigation required)",
        )

    try:
        started_at = datetime.fromisoformat(started_at_raw.replace("Z", "+00:00"))
    except (ValueError, TypeError) as exc:
        # parse 실패 → BLOCK (수동 investigation)
        logger.warning(
            "[STALE_RUNNING] step=%s started_at parse failed (%r): %s → BLOCK",
            self.step_id, started_at_raw, exc,
        )
        return ResumeDecision(
            action=ResumeAction.BLOCK,
            reason=f"running with started_at parse failed: {started_at_raw!r}",
        )

    from app.core.config import settings
    timeout_seconds = settings.step_running_timeout_seconds  # NEW setting, default 3600s

    elapsed = (datetime.now(timezone.utc) - started_at).total_seconds()
    if elapsed < timeout_seconds:
        return ResumeDecision(
            action=ResumeAction.BLOCK,
            reason=f"running healthy (elapsed={elapsed:.0f}s < timeout={timeout_seconds}s)",
        )

    # timeout 초과 → STALE_RUNNING_RECOVERY 분류 (atomic claim 으로 steal 시도)
    # V3 patch B1: expected_started_at + expected_run_id 동반 (claim SQL text 매칭)
    logger.warning(
        "[STALE_RUNNING] step=%s elapsed=%.0fs timeout=%ds → STALE_RUNNING_RECOVERY",
        self.step_id, elapsed, timeout_seconds,
    )
    return ResumeDecision(
        action=ResumeAction.STALE_RUNNING_RECOVERY,
        reason=f"running stale (elapsed={elapsed:.0f}s > timeout={timeout_seconds}s)",
        expected_started_at=started_at_raw,        # read 시점 text 그대로
        expected_run_id=existing.get("run_id"),    # read 시점 run_id 그대로
    )
```

heartbeat 는 D7 (별도 PR) — 1차는 `started_at` 기준만.

### 4.7 owner-aware status update (V1 patch I1 + V2 patch I3 확장)

claim 후 status transition 은 run_id owner 확인 의무. **V2 patch I3**: success path 뿐 아니라 **모든 transition path** 에서 적용 — exception handler / cleanup failure / partial / failed 포함.

```python
# step_runner.py _update_step_run() 변경 — require_owner=True 옵션
def _update_step_run(
    self, status: str, *, require_owner: bool = False,
    completed_count: int = 0, applicable_count: int = 1, failed_count: int = 0,
    error_message: Optional[str] = None, result_summary: Optional[str] = None,
) -> bool:
    """status transition. require_owner=True 면 WHERE run_id=:self.run_id 추가.

    Returns: True (update 성공) / False (다른 worker 가 row 소유 — race condition).
    """
    where_clause = "WHERE project_id = :pid AND episode_id = :eid AND step_id = :sid"
    if require_owner:
        where_clause += " AND run_id = :run_id"
    # ... UPDATE step_run SET status = :status, ... ${where_clause} ...
    return result.rowcount == 1
```

claim 이후 모든 transition 호출자 — `require_owner=True` 사용:

```python
# success path
self._update_step_run("completed", require_owner=True, ...)
self._update_step_run("partial", require_owner=True, ...)

# failure path (exception handler)
try:
    result = self._execute(...)
except Exception as exc:
    # V2 patch I3 — exception handler 도 owner check
    self._update_step_run("failed", require_owner=True, error_message=str(exc))
    raise

# cleanup failure
try:
    cleanup_report = self.cleanup_artifacts()
except Exception as exc:
    # V2 patch I3 — cleanup failure 도 owner check
    self._update_step_run(
        "failed", require_owner=True,
        error_message=f"cleanup_artifacts crashed: {exc}",
    )
    raise
```

오래된 worker (장기 timeout 후 깨어난) 가 새 worker 의 status 를 덮어쓰는 race 방지.

### 4.8 NOT_APPLICABLE 처리 (V2 patch I4)

`NOT_APPLICABLE` 은 claim 안 함, 그러나 현재 동작 (DB step_run + cp 기록) 은 보존:

```python
# StepRunner.run() — NOT_APPLICABLE 분기
if decision.action == ResumeAction.NOT_APPLICABLE:
    # claim 없이 _mark_not_applicable() 호출 — DB step_run + cp 기록
    self._mark_not_applicable(reason=decision.reason)
    return {"status": "not_applicable", "reason": decision.reason}

# _mark_not_applicable 은 race 가능성 낮음 (applicability 결정은 deterministic).
# 그러나 안전을 위해 require_owner=False (claim 안 했으므로) 로 단순 upsert.
```

### 4.9 Block B acceptance criteria

- ✅ AC-B1: `start_step()` 자체 skip 판단 제거. background worker → StepRunner.run 단일 판정자.
- ✅ AC-B2: `_safe_verify_completion()` unexpected exception → `step.verify_crashed` raise (자동 force 금지). entry side. AppError 계열은 `contract_drift`.
- ✅ AC-B3: scene_detail sentinel/card drift (stored vs computed mismatch) → `origin="invariant_drift"` 명시 → BLOCK (D4 까지 임시).
- ✅ AC-B4: `contract_drift` (schema/config_hash + loader/AppError + sentinel/card 부재 / shape 위반) → BLOCK. legacy schema bump allowlist (entity_t2i 만) 적용. allowlist == frozenset({"entity_t2i"}) 단위 테스트 검증.
- ✅ AC-B5: running 자동 force 금지. `started_at + timeout` 초과만 STALE_RUNNING_RECOVERY, 그 외 BLOCK. started_at 누락/parse 실패 시도 BLOCK.
- ✅ AC-B6: auto-recovery 경로 (`RERUN_SELF` / `STALE_RUNNING_RECOVERY`) 는 `_execute_rerun_self()` — `cleanup_artifacts` + `invalidate_downstream` 호출 안 함.
- ✅ AC-B7: 회귀 테스트 — single-step resume vs run-all resume 동일성 (cp=None / completed / failed 상태에서 모두 동일 결정).
- ✅ AC-B8: exit verify (`_safe_verify_completion` 의 exit-side) unexpected exception → `step.verify_crashed` raise. status='failed', save_checkpoint 미실행, error_message 기록. partial 아님.
- ✅ AC-B9: scene_detail.verify_completion 분류 매핑 §4.3 표 정확 적용 — loader/AppError contract → `contract_drift`, unexpected exception → `verify_crashed`, stored hash mismatch → `invariant_drift`, stored 부재/shape 위반 → `contract_drift`.

---

## §5. Block C — Concurrency lock (~0.5일)

### 5.1 목적

직접 script (`dispatch_pid_80f62523.py`) 와 background worker (api 경유) 가 같은 step 동시 잡는 race 차단. **Block C 완료 후 dispatch 재개**.

### 5.2 step_run row atomic claim — INSERT ON CONFLICT + atomic steal (V2 patch B1 + B2)

**V5 patch S7 — `_get_step_run()` dict shape (Codex NEEDS_REVISION BLOCKING #4 반영)**:

`_evaluate_running_state()` 의 stale 판정 + `_try_claim_running()` 의 expected-match steal 이 모두 `run_id` + `started_at` 텍스트에 의존하므로, 현 tuple `(status, completed_count, applicable_count)` (`step_runner.py:465-471`) 로는 부족. **dict 반환으로 변경**:

```python
def _get_step_run(self, step_id: str) -> Optional[Dict[str, Any]]:
    """step_run row 의 모든 결정 필드를 dict 로 반환.

    V5 patch S7 (Codex NEEDS_REVISION BLOCKING #4):
      atomic claim / stale steal / verify path 가 run_id + started_at + recovery_count 의존.
      tuple 반환은 caller 가 index 로 접근해 새 필드 추가 시 오해석 위험 (BLOCKING #4 의 root cause).
    """
    row = self.db.execute(text("""
        SELECT status, run_id, started_at, completed_count, applicable_count,
               COALESCE(recovery_count, 0) AS recovery_count, updated_at
        FROM step_run
        WHERE project_id = :pid AND episode_id = :eid AND step_id = :sid
    """), {"pid": self.project_id, "eid": self.episode_id, "sid": step_id}).fetchone()
    if row is None:
        return None
    return {
        "status": row.status,
        "run_id": row.run_id,
        "started_at": row.started_at,                # text — cast 없이 그대로 저장
        "completed_count": row.completed_count,
        "applicable_count": row.applicable_count,
        "recovery_count": row.recovery_count,
        "updated_at": row.updated_at,
    }
```

**기존 caller 갱신 의무 (Plan v2 P4 grep 검증)**:

| Caller | 현재 사용 | 갱신 후 |
|---|---|---|
| `run()` resume 분기 (line 534-535) | `existing[0] == "completed"` | `existing["status"] == "completed"` |
| `check_gate()` (있다면) | tuple index | dict key |
| `_evaluate_resume_decision()` (신규) | — | dict key 직접 사용 (`existing["run_id"]`, `existing["started_at"]`) |
| `_evaluate_running_state()` (신규) | — | dict 입력으로 통일 — V5 patch S7 적용 후 `if isinstance(existing, tuple)` fallback 제거 |
| 단위/통합 테스트 fixture | `("completed", 1, 1)` 같은 tuple | `{"status": ..., "run_id": ..., ...}` dict |

**Plan grep 검증 명령 (Plan v2 P4 의 신규 Task B0 끝)**:

```bash
# 기존 tuple-index 접근이 모두 제거됐는지 검증
rg "_get_step_run\([^)]*\)\s*\[\d" backend/app/ backend/tests/   # → 0 lines
rg "existing\[\d\]" backend/app/core/step_runner.py             # → 0 lines (또는 명시 fallback 코멘트)
```

**B1 patch v1 (claim 시점)**: ResumeDecision 평가는 **non-mutating** 으로 먼저. `RERUN_SELF` / `FORCE_EXPLICIT` / `STALE_RUNNING_RECOVERY` 일 때만 atomic claim. SKIP/BLOCK/NOT_APPLICABLE 은 claim 안 함 (running leak 차단).

**B2 patch v1 + V2 patch B1 + B2**: first-run 은 step_run row 가 없으므로 단순 `UPDATE ... RETURNING` 은 row 0 → already_running 오판. 또한 INSERT 시 NOT NULL 컬럼 (id, created_at, updated_at) 모두 명시 필수. 그리고 timeout 초과 running 도 atomic steal 가능해야 함:

```python
# step_runner.py 신규 _try_claim_running()
def _try_claim_running(self) -> bool:
    """atomic INSERT ON CONFLICT — race-free running claim.

    first-run (row 없음) + 재실행 (row 존재) + stale running steal 모두 처리.

    Returns: True (claim 성공, 자기가 실행) / False (다른 worker 이미 잡음, timeout 안 넘김).
    """
    import uuid
    from app.core.config import settings

    new_step_run_id = str(uuid.uuid4())
    timeout_seconds = settings.step_running_timeout_seconds  # default 3600s

    result = self.db.execute(text("""
        INSERT INTO step_run (
            id, project_id, episode_id, step_id, run_id, status,
            started_at, created_at, updated_at
        )
        VALUES (
            :new_step_run_id, :pid, :eid, :sid, :run_id, 'running',
            NOW()::text, NOW()::text, NOW()::text
        )
        ON CONFLICT (project_id, episode_id, step_id) DO UPDATE
          SET run_id = EXCLUDED.run_id,
              status = 'running',
              started_at = EXCLUDED.started_at,
              updated_at = EXCLUDED.updated_at,
              recovery_count = step_run.recovery_count + CASE
                WHEN step_run.status = 'running' THEN 1 ELSE 0
              END
          WHERE step_run.status != 'running'
             OR (
               -- V3 patch B1: SQL cast 제거. expected-match steal 만.
               :allow_stale_steal IS TRUE
               AND step_run.status = 'running'
               AND step_run.started_at = :expected_started_at
               AND step_run.run_id = :expected_run_id
             )
        RETURNING run_id
    """), {
        "new_step_run_id": new_step_run_id,
        "pid": self.project_id,
        "eid": self.episode_id,
        "sid": self.step_id,
        "run_id": self.run_id,
        "allow_stale_steal": allow_stale_steal,
        "expected_started_at": expected_started_at,
        "expected_run_id": expected_run_id,
    })
    self.db.commit()
    # V3 patch I2: rowcount 대신 fetchone() 으로 명시 — 드라이버/SQLAlchemy 차이 회피
    row = result.fetchone()
    return row is not None
```

`_try_claim_running` 시그니처 (V3 patch B1):

```python
def _try_claim_running(
    self,
    *,
    allow_stale_steal: bool = False,
    expected_started_at: Optional[str] = None,
    expected_run_id: Optional[str] = None,
) -> bool:
    """ResumeDecision.action == STALE_RUNNING_RECOVERY 일 때만 expected_* 전달.
    그 외 (RERUN_SELF / FORCE_EXPLICIT) 는 default (allow_stale_steal=False, expected_*=None)
    — 일반 conflict 시 status != 'running' 만 만족하면 claim 성공."""
```

**started_at parse 실패 row 는 steal 안 함** — `started_at::timestamptz` 캐스팅 실패 시 PG 에러 발생 가능. 따라서 timeout/parse 판정은 모두 Python `_evaluate_running_state()` 가 application 레벨에서 수행하고, claim SQL 은 `started_at = :expected_started_at` text exact match 만 수행 (V3 patch B1). parse 실패 row 는 `_evaluate_running_state()` 가 BLOCK 으로 분류 → claim 시도 자체 안 함 → audit (§7.3) 에서 별도 수동 처리.

`StepRunner.run` 변경 — claim 시점 (V1 patch B1 + V2 patch I4):

```python
def run(self, mode: str = "resume") -> Dict[str, Any]:
    # 1. gate / applicability (변경 없음)
    self.check_gate()
    if not self.check_applicability():
        # V2 patch I4: not_applicable 도 _mark_not_applicable() 로 DB+cp 기록
        self._mark_not_applicable(reason="check_applicability=False")
        return {"status": "not_applicable"}

    # 2. NEW: ResumeDecision 평가 (non-mutating, V1 patch B1)
    decision = self._evaluate_resume_decision(mode)

    if decision.action == ResumeAction.SKIP:
        return {"status": "skipped", "reason": decision.reason}

    if decision.action == ResumeAction.BLOCK:
        # V5 patch S6: AppError signature (code, message, status_code) 만. extra 금지.
        # origin 정보는 message 에 inline — backend/app/core/errors.py:5 contract 준수.
        raise AppError(
            code="step.resume_blocked",
            message=f"{decision.reason} (origin={decision.origin})",
            status_code=409,
        )

    if decision.action == ResumeAction.NOT_APPLICABLE:
        # V2 patch I4
        self._mark_not_applicable(reason=decision.reason)
        return {"status": "not_applicable", "reason": decision.reason}

    # 3. NEW: atomic claim (Block C, V1+V2+V3 patch)
    # decision.action in {RERUN_SELF, FORCE_EXPLICIT, STALE_RUNNING_RECOVERY}
    # V3 patch B1: STALE_RUNNING_RECOVERY 일 때만 expected_started_at/expected_run_id 전달
    is_stale_steal = (decision.action == ResumeAction.STALE_RUNNING_RECOVERY)
    if not self._try_claim_running(
        allow_stale_steal=is_stale_steal,
        expected_started_at=decision.expected_started_at if is_stale_steal else None,
        expected_run_id=decision.expected_run_id if is_stale_steal else None,
    ):
        raise AppError(
            code="step.already_running",
            message=f"{self.step_id} 이 이미 다른 worker 에서 실행 중 (또는 stale read 후 갱신됨)",
            status_code=409,
        )

    try:
        # 4. 실행 (action 별 분기)
        if decision.action == ResumeAction.FORCE_EXPLICIT:
            return self._execute_force()  # cleanup + invalidate_downstream
        else:
            # RERUN_SELF / STALE_RUNNING_RECOVERY — downstream 보존
            return self._execute_rerun_self()
    except Exception as exc:
        # V2 patch I3: exception path 도 owner check
        self._update_step_run(
            "failed", require_owner=True, error_message=str(exc),
        )
        raise
    # 정상 완료 시 _execute_* 내부에서 require_owner=True 로 status='completed' 설정
```

### 5.3 직접 script 정리

`dispatch_pid_80f62523.py` 와 같은 직접 script 도 task_registry / `_try_claim_running` 경유. `init_db()` 호출 금지 (이미 startup 전용, `_migrations` race 위험).

```python
# dispatch_pid_80f62523.py 변경 (또는 가이드 patch)
def main():
    from app.core.database import register_models, SessionLocal
    register_models()  # ← OK (model 등록만, DDL 없음)

    # ... preflight / select / context (이미 short-session 으로 완료) ...

    # run_steps_batch 가 내부에서 StepRunner.run 호출 → atomic claim 자동 적용
    run_steps_batch(...)
```

**V5 patch S5 — script quarantine 정책 (Codex NEEDS_REVISION IMPORTANT #3 반영)**:

`backend/scripts/test_scene_detail.py:122` 가 `original_execute(mode)` 를 직접 호출하는 패턴이 잔존 (fan-out 시 일부 scene 만 실행하기 위해 `_load_prev_checkpoint` monkeypatch 후 `original_execute(mode)` 호출). 이 류의 직접 `_execute()` / `original_execute()` 호출 script 는 atomic claim 우회 → AC-C3 위반.

**처리 정책 (택일, plan 에서 명시 의무)**:

| 옵션 | 적용 대상 | 기대 동작 |
|---|---|---|
| **A. StepRunner.run() 경유 전환** | 일반 운영 / production-like 시나리오 | `original_execute(mode)` 호출 자리를 `runner.run(mode=mode)` 로 교체. fan-out 단일 scene 필터는 `_load_prev_checkpoint` monkeypatch 만으로 충분 (StepRunner.run 안에서 `_execute` 호출 시 patched loader 활용). claim 정상 작동. |
| **B. quarantine (개발/디버그 전용)** | 개발자 한정, production 운영 경로 아님 | script 상단에 `# QUARANTINED: dev-only — atomic claim 우회. production dispatch 에서 호출 금지` 주석 + 파일 docstring 명시. 동시 실행 시 race 가능성 있음을 명시. AC-C3 grep 의 allowlist 에 등록. |

**기본**: A (전환). B 는 사용자 명시 결정 시만.

**AC-C3 검증 grep 정의 (재정의)**:

```bash
# 1. 직접 _execute() / original_execute() 호출 검색 — quarantine allowlist 제외
# review M2 정정 (Block C closure): rg --glob 은 cwd-기준 path 가 아닌 basename
# 매칭이라 prefixed 'backend/scripts/_quarantined/**' 는 cwd 가 backend/ 일 때
# 매칭 안 함. '**/_quarantined/**' 가 모든 path 위치에서 안전.
rg "(?:^|[^_])\b(?:original_execute|_execute)\(" backend/scripts/ \
   --glob '!**/__pycache__/**' \
   --glob '!**/_quarantined/**'
# Expected: 0 lines (quarantine 파일은 _quarantined/ 디렉토리 또는 명시 allowlist 로 분리)

# 2. quarantine allowlist 검증 — 명시된 파일만 우회 허용
EXPECTED_QUARANTINED='backend/scripts/test_scene_detail.py'   # 옵션 B 선택 시
# (옵션 A 선택 시 EXPECTED_QUARANTINED='' 빈 문자열)
```

**Plan 책임**: Plan v2 P6 / 추가 c 에서 `backend/scripts/test_scene_detail.py` 처리 task 명시 (옵션 A: StepRunner.run() 경유 전환 / 옵션 B: 명시 quarantine + allowlist 등록). plan 작성 시 사용자에게 옵션 결정 묻기.

코드 search 로 `_execute()` 직접 호출 0건 (또는 quarantine allowlist 만 매치) 확인 (run_steps_batch 외 우회 경로 없음을 spec 검증).

### 5.4 connection pool 안전성 (V1 보정 4)

`pg_try_advisory_lock` 은 **사용 X** — connection pool 에서 잡으면 lock leak 위험. row atomic claim 은 트랜잭션과 무관, atomic UPSERT 만으로 race-free → connection pool 호환.

`pg_try_advisory_xact_lock` 도 사용 X — commit 때 풀려서 StepRunner 의 중간 commit 패턴 (per-step status update 등) 과 부적합.

### 5.5 Block C acceptance criteria

- ✅ AC-C1: `_try_claim_running()` atomic INSERT ON CONFLICT — id/created_at/updated_at 모두 NOT NULL 충족. 동시 worker 두 개가 claim 시 한 쪽만 성공.
- ✅ AC-C2: claim 실패 시 `step.already_running` raise (status_code=409).
- ✅ AC-C3 (V5 patch S5): 직접 script (dispatch_pid_*) 도 `run_steps_batch` → `StepRunner.run` → `_try_claim_running` 경유. **AC-C3 검증 grep 재정의** (§5.3): `rg "(?:original_execute|_execute)\("` 결과가 quarantine allowlist 외 0건 (default: allowlist 0건 → 결과 0건). 현재 `backend/scripts/test_scene_detail.py:122` 의 `original_execute(mode)` 호출은 plan v2 P6 / 추가 c 에서 옵션 A (StepRunner.run() 경유 전환) 또는 옵션 B (명시 quarantine + allowlist 등록) 로 처리.
- ✅ AC-C4: connection pool leak 0 (트랜잭션 long-hold 없음, atomic INSERT ON CONFLICT only). `_try_claim_running` 후 즉시 commit.
- ✅ AC-C5: 동시 claim 회귀 테스트 — 두 thread/process 가 같은 step 잡을 때 LLM/image 호출 1회만 발생.
- ✅ AC-C6: first-run (step_run row 없음) INSERT 경로 — claim 성공, status='running' 신규 row 생성. UPDATE-only 로직의 already_running 오판 없음.
- ✅ AC-C7: claim 후 모든 status transition (`completed` / `partial` / `failed` + exception handler + cleanup failure) 에 `require_owner=True` (`WHERE run_id=:self.run_id`). 오래된 worker 가 새 worker 상태 덮어쓰는 race 차단.
- ✅ AC-C8 (V2 patch B2 + V3 patch B1): timeout 초과 running row 는 expected-match atomic steal — `step_run.started_at = :expected_started_at AND run_id = :expected_run_id` (SQL cast 없음). STALE_RUNNING_RECOVERY 시 claim 성공.
- ✅ AC-C9 (V2 patch B1): SQL 의 INSERT 컬럼 목록 검증 — id, project_id, episode_id, step_id, run_id, status, started_at, created_at, updated_at 모두 명시. NOT NULL violation 0.
- ✅ AC-C10 (V3 patch B1): claim SQL 에 `::timestamptz` cast 0건. timeout 판정은 모두 Python `_evaluate_running_state()` 가 수행. invalid text started_at row 가 있어도 query 정상 작동.
- ✅ AC-C11 (V3 patch I2): `_try_claim_running()` 결과 판정은 `result.fetchone() is not None` 사용 — `rowcount` 의 driver/SQLAlchemy 차이 회피.
- ✅ AC-C12 (V3 patch I3): `settings.step_running_timeout_seconds` 신규 config 추가, default `3600`. config 로드 + 기본값 단위 테스트 — 누락 시 startup 실패.

---

## §6. Block D — 후속 PR (별도)

Block A+B+C dispatch 재개 후 진행. 본 spec scope 외 — Block D 는 별도 spec drafting.

| ID | Item | 설명 |
|---|---|---|
| **D1** | `CompletionReport.origin` 전면 적용 | Block B 점진 도입 → 모든 verifier (image_steps, background_render, floor_plan_render 등) origin 명시 migrate |
| **D2** | `resume_config_keys` per-step manifest | config hash 축소. default `[]` (= hash 없음, skip 우선). `manifest.resume_config_keys: List[str]` 명시한 step 만 hash. project_config 전체 fallback 제거 |
| **D3** | schema bump policy 명시화 | `manifest.allow_auto_rerun_on_schema_bump: bool = False`. Block B 의 임시 allowlist (`_LEGACY_SCHEMA_BUMP_ALLOWLIST`) 제거 |
| **D4** | `modifies_checkpoints` rollback semantic | victim cp archive → verify 후 실패 시 archive rollback. mutator step failed. victim 자동 stale 금지. mutator origin tracking (sentinel `_mutation_origin` 필드) — invariant_drift 분기 정식화 |
| **D5** | `entity_t2i.verify_completion` override | asymmetric silent corruption 해소. t2i_review 가 entity_t2i 수정 시 sentinel/hash refresh + verify entry |
| **D6** | `resume_sensitive` 이름 정리 | shot_selection toggle 한정 사용처 — `stale_on_shot_selection_change` 또는 StepRunner 직접 소비. cleanup category |
| **D7** | heartbeat 컬럼/업데이트 | step_run 에 `heartbeat_at` 컬럼 + 주기 update. running timeout 정확도 향상 (현재 `started_at` 기준만) |
| **D8** | 회귀 테스트 전면 | concurrent worker / single-vs-run-all 동일성 / config 무관 변경 skip / sentinel drift 시나리오 / cp=None 회복 / verify crash fail-fast / contract_drift block |

---

## §7. Migration / Rollback

### 7.1 Backward compatibility

- `CompletionReport.origin`: optional default `"artifact_missing"` — 기존 verifier 변경 없이도 동작 (점진 도입).
- `_LEGACY_SCHEMA_BUMP_ALLOWLIST` (entity_t2i 만): 기존 운영 패턴 (schema_version 1→2 자동 cascade) 보존.
- `step_running_timeout_seconds`: 신규 setting, default 3600s (1시간). 기존 step 무영향.
- step_run 스키마 변경 없음 (Block A/B/C 모두). D7 (heartbeat) 에서만 컬럼 추가.

### 7.2 Block 단위 deploy 가능성

| Block | 단독 deploy 가능 | 의존성 |
|---|---|---|
| A | ✅ | 없음 |
| B | ✅ (A 없어도 deploy 가능) | A 권장 (sentinel refresh 로 invariant_drift trigger 제거) |
| C | ✅ | A+B 권장 (lock 만 추가, 다른 race 잔존) |

**dispatch 재개는 A+B+C 모두 완료 후**. 단독 deploy 가능하지만 race 차단 부족 (A 만 / A+B 만 = race 잔존).

### 7.3 배포 전 running row audit (V1 patch I5 + V2 patch B3 — SQL 정정)

Block B 배포는 기존 `running` status row 들을 새 `_evaluate_running_state` 로직으로 평가시킴. `started_at + timeout` 초과 안 한 row 는 BLOCK 처리 → 운영 incident 가능.

**배포 전 audit 절차 — V3 patch B1 (SQL cast 완전 제거, app-level parse 통일)**:

SQL 은 단순 SELECT 만 (cast 없음). elapsed/parse 판정은 모두 application 레벨.

```sql
-- 1. 모든 running row 식별 (cast 없음 — invalid text 도 안전하게 회수)
SELECT
  sr.project_id, sr.episode_id, sr.step_id, sr.run_id, sr.id,
  sr.started_at, sr.updated_at, sr.recovery_count
FROM step_run sr
WHERE sr.status = 'running'
ORDER BY sr.started_at ASC NULLS FIRST;
```

```python
# scripts/audit_stale_running.py (배포 전 1회)
# V3 patch B1: SQL cast 0 — 모든 timeout/parse 판정 application 레벨
import os
from datetime import datetime, timezone
import sqlalchemy as sa

TIMEOUT_SECONDS = int(os.environ.get("STEP_RUNNING_TIMEOUT_SECONDS", "3600"))

with engine.connect() as conn:
    rows = conn.execute(sa.text(
        "SELECT id, project_id, episode_id, step_id, run_id, started_at, updated_at, recovery_count "
        "FROM step_run "
        "WHERE status = 'running' "
        "ORDER BY started_at ASC NULLS FIRST"
    )).fetchall()

now = datetime.now(timezone.utc)
healthy, stale, parse_failed, null_started = [], [], [], []

for r in rows:
    if r.started_at is None:
        null_started.append(r)
        continue
    try:
        st = datetime.fromisoformat(r.started_at.replace("Z", "+00:00"))
        elapsed = (now - st).total_seconds()
        if elapsed > TIMEOUT_SECONDS:
            stale.append((r, elapsed))
        else:
            healthy.append((r, elapsed))
    except (ValueError, TypeError) as exc:
        parse_failed.append((r, str(exc)))

# 출력
print(f"[HEALTHY] {len(healthy)} rows (elapsed < {TIMEOUT_SECONDS}s)")
for r, elapsed in healthy:
    print(f"  step={r.step_id} run_id={r.run_id} elapsed={elapsed:.0f}s")
print(f"\n[STALE] {len(stale)} rows (elapsed > {TIMEOUT_SECONDS}s — STALE_RUNNING_RECOVERY 대상)")
for r, elapsed in stale:
    print(f"  step={r.step_id} run_id={r.run_id} elapsed={elapsed:.0f}s started_at={r.started_at}")
print(f"\n[PARSE_FAILED] {len(parse_failed)} rows (수동 investigation 필요)")
for r, err in parse_failed:
    print(f"  step={r.step_id} started_at={r.started_at!r} err={err}")
print(f"\n[NULL_STARTED] {len(null_started)} rows (수동 investigation 필요)")
for r in null_started:
    print(f"  step={r.step_id} run_id={r.run_id}")
```

```python
# scripts/audit_stale_running.py (배포 전 1회)
from datetime import datetime
import sqlalchemy as sa

with engine.connect() as conn:
    rows = conn.execute(sa.text(
        "SELECT id, project_id, episode_id, step_id, started_at FROM step_run "
        "WHERE status = 'running' AND started_at IS NOT NULL"
    )).fetchall()
    for r in rows:
        try:
            datetime.fromisoformat(r.started_at.replace("Z", "+00:00"))
        except (ValueError, TypeError) as exc:
            print(f"PARSE_FAILED step={r.step_id} started_at={r.started_at!r} err={exc}")
```

**처리 기준**:
- 실제 worker 가 살아있는 row → 배포 후에도 정상 완료 → 그대로
- 죽은 worker 의 stale row → 수동 reset:
  ```sql
  UPDATE step_run
  SET status = 'failed',
      error_message = 'pre-B deploy stale running reset',
      updated_at = NOW()::text
  WHERE id = :step_run_id
    AND status = 'running'
    AND started_at = :original_started_at;  -- 우연한 동시 update 방지
  ```
- 판단 모호한 row → `started_at` 시점이 오래됐으면 (24h+) reset, 최근이면 그대로 (정상 worker 가능성)
- started_at NULL 또는 parse 실패 row → 수동 investigation 우선, 무조건 reset 금지

### 7.4 배포 후 모니터링 (V2 질문 답변 #2 확장)

배포 직후 24시간 동안 다음 신호 모니터링:

| 신호 | 임계값 | 조치 |
|---|---|---|
| `step.resume_blocked` 신규 incident | 시간당 5건 이상 | timeout 값 조정 또는 수동 reset 검토 |
| `step.verify_crashed` 신규 incident | 시간당 1건 이상 | verifier 코드 버그 가능성 — 즉시 stack trace 분석 |
| `step.already_running` 신규 incident | 시간당 3건 이상 | claim race 또는 stale running steal 미작동 — Block C 동작 확인 |
| `running` status row 가 timeout 초과한 채 재발 | 시간당 1건 이상 | atomic steal SQL 동작 안 함 가능 — `_try_claim_running` 로그 확인 |

### 7.5 Rollback 절차

각 Block 은 단일 PR. 문제 발생 시 PR revert.
- A revert → `t2i_review` 가 다시 sentinel 미갱신. invariant_drift cascade 재발 (B 미배포 시).
- B revert → `start_step` 자체 skip 복원, force-like 자동 격상 복원. 사용자가 다시 수동 force 필요.
- C revert → atomic claim 제거. race 재발 가능.

A → B → C 순서로 deploy → 역순 (C → B → A) 으로 revert.

---

## §8. Risks + Mitigations

| Risk | Mitigation |
|---|---|
| Block A 의 sentinel refresh 가 schema 변화 (G3.2 → G4.1 → ...) 때 또 누락 | helper 단위 테스트 + assert_owned_sentinel_shape 으로 shape 위반 raise. D4 에서 일반화 (mutator 자동 invariant 갱신 의무) |
| Block B 의 `invariant_drift = BLOCK` 임시 정책으로 운영 중단 | Block A 로 mutator drift 사전 차단 → 일반 운영에서 invariant_drift 발생 0 예상. 발생 시 사용자 명시 force 로 진행 |
| Block C 의 row atomic claim 가 wait/timeout 발생 (V1 patch M3) | deadlock risk low (단일 row UPSERT). 단 외부 트랜잭션 순서와 엮일 시 wait/timeout 가능. `lock_timeout` 설정 (현재 `database.py` 마이그레이션 로직 참조) + 회귀 테스트로 검증 |
| 직접 script (dispatch_pid_*) 가 atomic claim 우회 시도 | run_steps_batch 가 내부에서 StepRunner.run 호출 → 자동 적용. 직접 `_execute()` 호출하는 script 는 코드 search 로 0건 확인 |
| schema_version mismatch 가 allowlist 외 step 에서 발생 | Block B 에서 BLOCK → 운영자 확인 후 force 명시. 운영 incident 가능성 — D3 (manifest flag) 빠르게 진입 |
| Block A 의 `applied_indices` 반환이 pipeline 변경 큼 (V1 patch B3) | `run_t2i_review()` 이 이미 `entity_applied`/`scene_applied` count 반환 — indices 추가는 dict 1 field 확장. 회귀 작음. 단 fan-out 시나리오 (같은 scene_index 다수) 가 충분히 covered 되었는지 회귀 테스트 |
| 사용자가 dispatch 재개 시점을 잘못 판단 (A 만 / A+B 만 deploy 후 재개) | spec 에 "Block A+B+C 모두 완료 후 dispatch 재개" 명시. CI / pre-deploy gate 에 Block 묶음 marker 추가 가능 |
| Block B 배포 후 기존 running row 가 BLOCK 처리됨 (V1 patch I5) | §7.3 audit 절차 — 배포 전 running row 식별 + 처리 기준 + 수동 reset SQL 준비 |
| owner-aware update 누락된 호출자가 silent race | `_update_step_run(require_owner=True)` 가 default 가 아니므로 명시 누락 가능. 회귀 테스트 — claim 후 모든 status transition (success/partial/failed/exception/cleanup) 이 owner 검증 통과하는지 verify (V2 patch I3 강화) |
| started_at parse 실패 row 가 stale steal 안 됨 (V2 patch B3) | application 레벨 audit script 로 별도 detection. 배포 전 수동 reset. timeout 초과 row 자동 steal 은 정상 ISO 8601 format 만 처리 |
| INSERT SQL 의 NOT NULL 컬럼 누락으로 first-run 실패 (V2 patch B1) | AC-C9 — INSERT 컬럼 목록 단위 테스트. NOT NULL violation 0 검증 |

---

## §9. Acceptance Criteria 요약

### Block A (6건)
- AC-A1: `_owned_helpers.refresh_t2i_prompt_hash()` primitive — sentinel shape 검증 + t2i_prompt_hash 만 갱신 (다른 hash 보존)
- AC-A2: helper 실패 시 t2i_review status='failed' (silent 차단)
- AC-A3: render_prompt_card_hash 변경 없음을 assert 로 검증
- AC-A4: 단위 테스트 — fix 후 next resume 시 verify 통과
- AC-A5: 회귀 — fix 0건 시 sentinel no-op
- AC-A6: `_apply_scene_fixes()` 가 실제 수정된 `(scene_list_idx, variation_idx)` 반환. fan-out 회귀 테스트

### Block B (9건)
- AC-B1: `start_step()` 자체 skip 제거 — 단일 판정자 (ResumeDecision dataclass)
- AC-B2: verify unexpected exception → `step.verify_crashed` (자동 force 금지) — entry side. AppError 는 `contract_drift`
- AC-B3: invariant_drift (stored hash mismatch) → BLOCK (D4까지 임시)
- AC-B4: contract_drift (schema/config_hash + loader/AppError + sentinel/card 부재/shape 위반) → BLOCK. legacy allowlist `frozenset({"entity_t2i"})` 단위 테스트 검증
- AC-B5: running 자동 force 금지 — `started_at + timeout` 초과만 STALE_RUNNING_RECOVERY, 누락/parse 실패 시 BLOCK
- AC-B6: auto-recovery (RERUN_SELF + STALE_RUNNING_RECOVERY) 는 `_execute_rerun_self` (cleanup + invalidate 안 함)
- AC-B7 (V5 patch S3): single-step vs run-all 동일성 회귀 — **실제 두 path 검증 의무**. 같은 runner 의 `_evaluate_resume_decision()` 두 번 호출은 미충족. 두 path 명시:
  - **path 1**: `step_execution_service.start_step()` (background_worker → `StepRunner.run` 단일 호출)
  - **path 2**: `run_steps_batch()` (직접 또는 dispatch script 경유 → 각 step 마다 `StepRunner.run` 호출)
  테스트는 두 path 각각 stub 또는 fixture 로 진입 → 같은 (step_id, mode, 사전 step_run 상태) 에서 같은 ResumeDecision/ResumeAction 도달 + 같은 claim 동작 (또는 SKIP 조기 return) 검증.
- AC-B8: exit verify crash → `step.verify_crashed` raise. status='failed', save_checkpoint 미실행, error_message 기록
- AC-B9 (V5 patch S4): scene_detail.verify_completion 분류 매핑 §4.3 표 정확 적용 (loader/AppError → contract_drift, unexpected → verify_crashed, stored mismatch → invariant_drift, 부재/shape 위반 → contract_drift). **실제 `SceneDetailStep.verify_completion()` 을 fixture 로 직접 태운 단위 테스트** 의무 — `CompletionReport(origin=...)` 를 직접 build 한 helper-only 테스트는 미충족. §4.3 의 7개 branch 별 fixture 작성, severity (missing vs partial) 분류도 함께 검증.

### Block C (9건)
- AC-C1: atomic INSERT ON CONFLICT — id/created_at/updated_at NOT NULL 충족 — 동시 두 worker 한 쪽만 성공
- AC-C2: claim 실패 시 409 raise
- AC-C3: 직접 script 도 atomic claim 경유 (우회 0)
- AC-C4: connection pool leak 0
- AC-C5: 동시 worker 회귀 테스트 — LLM 호출 1회만
- AC-C6: first-run (step_run row 없음) INSERT 경로 — claim 성공, UPDATE-only 오판 없음
- AC-C7: claim 후 모든 status transition (success/partial/failed/exception handler/cleanup failure) `require_owner=True`
- AC-C8: timeout 초과 running expected-match atomic steal — SQL cast 없이 `started_at = :expected_started_at AND run_id = :expected_run_id`
- AC-C9: INSERT SQL 컬럼 목록 검증 (id, project_id, episode_id, step_id, run_id, status, started_at, created_at, updated_at 모두 명시) — NOT NULL violation 0
- AC-C10: claim/audit SQL 에 `::timestamptz` cast 0건. invalid text row 가 있어도 query 정상
- AC-C11: `result.fetchone() is not None` 으로 claim 결과 판정 (rowcount X)
- AC-C12: `settings.step_running_timeout_seconds` config 추가 (default 3600). 단위 테스트로 default 검증

### Plan v1 NEEDS_REVISION → V5 patch → AC 매핑

| V5 patch | Codex finding | 영향 AC | spec section |
|---|---|---|---|
| S1 | BLOCKING #2 (card hash pre-capture timing) | AC-A3 | §3.5 (caller 책임 명시 + `_capture_card_hash_state`) |
| S2 | BLOCKING #3 (run() finalization 손실) | AC-B6 | §4.1 (변경 범위 row 추가) + §4.2 (`_execute_and_finalize` 6 단계 보존) |
| S3 | IMPORTANT #2 (동일성 테스트 두 path 미검증) | AC-B7 | §9 AC-B7 (start_step path + run_steps_batch path 명시) |
| S4 | BLOCKING #6 (B8 verify_completion fixture 부재) | AC-B9 | §4.3 (테스트 검증 의무 표) + §9 AC-B9 |
| S5 | IMPORTANT #3 (script 우회 grep) | AC-C3 | §5.3 (옵션 A 전환 / 옵션 B quarantine 정책) + §9 AC-C3 (grep 재정의) |
| S6 | BLOCKING #5 (AppError extra TypeError) | AC-B1 / §2.3 / §5.2 run() | §2.3 (AppError contract NOTE) + §5.2 (BLOCK 분기 코드 정정) |
| S7 | BLOCKING #4 (`_get_step_run` tuple shape 오해석) | AC-B5 / AC-C1 / AC-C8 | §4.1 (변경 범위 row 추가) + §5.2 (helper signature + caller 갱신 의무 + grep 검증) |

NEEDS_REVISION 의 BLOCKING #1 (prompt path) + IMPORTANT #1 (A4 stub return shape) + IMPORTANT #4 (owner-aware false 처리) + MINOR 2건 은 spec patch 불필요 — plan v2 에서 직접 정정 (§3.2 변경 위치는 spec 가 path 명시 안 했으므로 plan 의 자체 결함).

### NEEDS_SPEC_PATCH_V2 BLOCKING/IMPORTANT → AC 매핑

| Patch | AC |
|---|---|
| V2 B1 (INSERT NOT NULL 컬럼 명시) | AC-C9 |
| V2 B2 (timeout-based atomic steal) | AC-C8 |
| V2 B3 (audit SQL 정정) | §7.3 |
| V2 I1 (ResumeDecision dataclass + enum 확장) | AC-B1 (ResumeDecision dataclass), §2.3 enum 확장 |
| V2 I2 (verify origin 분기 더 세밀히) | AC-B9, §4.3 표 |
| V2 I3 (require_owner exception path 포함) | AC-C7 (모든 transition) |
| V2 I4 (NOT_APPLICABLE _mark_not_applicable 호출) | §4.8, §5.2 run() 흐름 |

### NEEDS_SPEC_PATCH (V1) BLOCKING/IMPORTANT → AC 매핑

| Patch | AC |
|---|---|
| V1 B1 (claim 시점, ResumeDecision 평가 후) | AC-C1 (Block C run() 흐름 명시) |
| V1 B2 (first-run 처리 INSERT ON CONFLICT) | AC-C6 |
| V1 B3 (exact mutated path 반환) | AC-A6 |
| V1 I1 (run_id owner update) | AC-C7 |
| V1 I2 (loader/card recompute origin) | AC-B9 |
| V1 I3 (scene_detail allowlist 제거) | AC-B4 (allowlist = entity_t2i 만) |
| V1 I4 (exit verify crash) | AC-B8 |
| V1 I5 (Migration running row audit) | §7.3 |

### 사용자 보정 6 → AC 매핑 (이전)

| 보정 | AC |
|---|---|
| 1. sentinel refresh 좁히기 (build_owned_sentinel 전체 X) | AC-A1 |
| 2. render_prompt_card_hash 가볍게 (assert만) | AC-A3 |
| 3. rerun_self 별도 경로 (force 재사용 X) | AC-B6 |
| 4. row atomic claim (advisory lock X) | AC-C1, AC-C4 |
| 5. running timeout — started_at 기준만 (heartbeat 보류) | AC-B5 |
| 6. schema bump 임시 allowlist | AC-B4 |
| 7. modifies_checkpoints rollback (Phase D4) | D4 (별도 spec) |

---

## §10. Out of scope

- Image step verify (image_steps.py, background_render_step.py 등) — D1 점진 migrate
- `resume_config_keys` per-step — D2
- heartbeat — D7
- entity_t2i verify — D5
- modifies_checkpoints rollback — D4
- frontend UI 변경 (resume status 표시 등) — 별도 PR

---

## §11. References

- `backend/app/core/database.py:67-94` (step_run schema)
- `backend/app/core/step_runner.py:546-600` (verify path), `:608` (status switch), `:641` (invalidate cascade), `:842` (`_safe_verify_completion`), `:679` (exception handler — V2 patch I3)
- `backend/app/services/step_execution_service.py:169-200` (자체 skip 판단)
- `backend/app/core/integrity_report.py:10` (CompletionReport)
- `backend/app/core/steps/_owned_helpers.py:112` (`compute_t2i_prompt_hash`), `:130` (`build_owned_sentinel`), `:158` (`assert_owned_sentinel_shape`)
- `backend/app/core/steps/t2i_review_step.py:55-62` (mutation site)
- `backend/app/core/steps/detail_steps.py:1068` (scene_detail.verify_completion), `:1133` (loader 호출 시점), `:1276` (card recompute), `:1373, 1388, 1404` (drift trigger), `:2380` (sentinel build call site)
- `backend/app/core/step_manifest.py:715` (t2i_review.modifies_checkpoints), `:33,454,566,638` (resume_sensitive)
- `backend/app/services/shot_selection_service.py:201` (resume_sensitive consumer)
- `backend/app/core/task_registry.py` (process-local registry)
- `backend/app/modules/pipeline/t2i_review.py:222-360` (`_review_scene_detail`, `_apply_scene_fixes`)
- `backend/scripts/dispatch_pid_80f62523.py:71-94` (short-session pattern, 이미 fix 됨)

---

## §12. Patch History

| Date | Version | Patch | Source |
|---|---|---|---|
| 2026-05-07 | v1 | Initial draft | Claude (Opus 4.7) — 사용자 brainstorm 결정 반영 |
| 2026-05-07 | v2 | NEEDS_SPEC_PATCH 반영 — V1 B1+B2+B3 BLOCKING + V1 I1~I5 + V1 M1~M3 | 사용자 directed review (Codex auxiliary) |
| 2026-05-07 | v3 | NEEDS_SPEC_PATCH_V2 반영 — V2 B1 (INSERT NOT NULL) + V2 B2 (atomic steal) + V2 B3 (audit SQL) + V2 I1 (ResumeDecision dataclass) + V2 I2 (verify 분기) + V2 I3 (exception owner) + V2 I4 (NOT_APPLICABLE) | 사용자 directed review (Codex auxiliary) |
| 2026-05-07 | v4 | NEEDS_SMALL_SPEC_PATCH 반영 — V3 B1 (SQL cast 제거 + expected-match steal) + V3 I1 (item_id deterministic) + V3 I2 (fetchone) + V3 I3 (config 추가) | 사용자 directed review (Codex auxiliary) |
| 2026-05-08 | v5 | Plan v1 NEEDS_REVISION (Codex BLOCKING 6 + IMPORTANT 4) → implementation contract clarification patch — S1 (AC-A3 caller 책임 + `_capture_card_hash_state`) + S2 (`_execute_and_finalize` 공통 finalize) + S3 (AC-B7 두 path 명시) + S4 (AC-B9 verify_completion fixture 의무) + S5 (AC-C3 script quarantine 정책) + S6 (AppError contract — extra 금지) + S7 (`_get_step_run` dict shape) | 사용자 directed review (Codex auxiliary) |