# Resume 무결성 재설계 Implementation Plan

> **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking.

**Goal:** PID 0bb48ebf 사고(status=completed인데 산출물 없음 → silent fallback 캐릭터 random) 재발 방지를 위한 verify + cleanup framework 도입. 양방향 verify_completion(entry/exit) + auto-rerun + status별 force-like + 부분 재생성 보호 + runtime fail-soft.

**Architecture:** `StepRunner` 베이스에 `verify_completion()`/`cleanup_artifacts()` hook 추가 (default 안전), `run()`에서 entry/exit 양방향 호출. 검증 실패 시 자동 force-rerun + recovery_count 추적, 3회 후 차단. 이미지 step의 cleanup_artifacts는 거의 모두 default noop 유지 — 부분 재생성 endpoint 보호.

**Tech Stack:** Python 3.12 / SQLAlchemy / Alembic / FastAPI (backend), pytest, PostgreSQL. Spec: `docs/superpowers/specs/2026-05-01-resume-integrity-redesign.md`.

---

## File Structure

### 신규 파일 (Phase 1)
- `backend/app/core/integrity_report.py` — `CompletionReport` / `CleanupReport` dataclass
- `backend/alembic/versions/003_resume_integrity.py` — recovery_count + last_recovery_reason + variant_label String(32)
- `backend/tests/core/test_integrity_report.py` — dataclass + mask 유틸 단위 테스트
- `backend/tests/core/test_step_runner_resilience.py` — 통합 테스트 (status별 분기, recovery loop)
- `backend/tests/core/steps/test_verify_completion.py` — Phase A 6 step verify 테스트
- `backend/tests/services/test_scene_reference_failsoft.py` — D3 runtime fail-soft
- `backend/tests/test_pipeline_gate_enhanced.py` — D4 gate 강화

### 수정 파일
- `backend/app/core/step_runner.py:289-360` — save_checkpoint snapshot 기록 (D2)
- `backend/app/core/step_runner.py:420-426` — `_get_step_run` 시그니처 확장
- `backend/app/core/step_runner.py:473-615` — `run()` 분기 통합 (entry/exit verify + status별 force-like + recovery)
- `backend/app/core/database.py:33-144` — step_run에 recovery_count + last_recovery_reason 컬럼 (raw SQL ALTER)
- `backend/app/models/project.py:181` — variant_label String(32) (현재 uncommitted 동기화)
- `backend/app/core/steps/image_steps.py:130-578` — Phase A 5 step verify_completion (RefImageGen / Composite / OutlookStandalone 등)
- `backend/app/core/steps/background_chain_render_step.py` — verify_completion
- `backend/app/core/steps/floor_plan_render_step.py` — verify_completion
- `backend/app/services/scene_reference_service.py:254-391` — `resolve_refs_for_prompt`이 missing_refs 반환 (D3)
- `backend/app/core/pipeline_gate.py:51-197` — outlook/state_variant/chain_bg/floor_plan 검증 추가 (D4)
- `backend/app/core/steps/scene_steps.py` 또는 `scene_image_pipeline_step.py` — verify_completion + missing_refs 처리

### PR 단위
- **PR 1**: Tasks 1–18 (Phase 0 + 1 framework + migration)
- **PR 2**: Tasks 19–28 (Phase 2 per-step verify 6건)
- **PR 3**: Tasks 29–37 (Phase 3 + 4 — D3 runtime + D4 gate)
- **PR 4**: Tasks 38–42 (Phase 5 UI)

각 PR 끝에 듀얼 리뷰(Codex + Claude) + 회귀 테스트 통과.

---

# PR 1 — Framework + Migration (Phase 0 + 1)

## Task 1: alembic 003 작성 (recovery_count + last_recovery_reason + variant_label 32)

**Files:**
- Create: `backend/alembic/versions/003_resume_integrity.py`

- [ ] **Step 1: 신규 파일 작성**

```python
"""Phase Resume Integrity: step_run.recovery_count + last_recovery_reason +
image_asset.variant_label String(8) → String(32).

Revision ID: 003_resume_integrity
Revises: 002_phase5_image_asset_variants
Create Date: 2026-05-01
"""
from typing import Sequence, Union

from alembic import op
import sqlalchemy as sa


revision: str = "003_resume_integrity"
down_revision: Union[str, None] = "002_phase5_image_asset_variants"
branch_labels: Union[str, Sequence[str], None] = None
depends_on: Union[str, Sequence[str], None] = None


def upgrade() -> None:
    """Add recovery tracking columns to step_run + extend variant_label length."""
    op.add_column(
        "step_run",
        sa.Column("recovery_count", sa.Integer(), nullable=False, server_default="0"),
    )
    op.add_column(
        "step_run",
        sa.Column("last_recovery_reason", sa.Text(), nullable=True),
    )
    op.alter_column(
        "image_asset", "variant_label",
        type_=sa.String(length=32),
        existing_type=sa.String(length=8),
        existing_nullable=False,
        existing_server_default="v00",
    )


def downgrade() -> None:
    op.alter_column(
        "image_asset", "variant_label",
        type_=sa.String(length=8),
        existing_type=sa.String(length=32),
        existing_nullable=False,
        existing_server_default="v00",
    )
    op.drop_column("step_run", "last_recovery_reason")
    op.drop_column("step_run", "recovery_count")
```

- [ ] **Step 2: alembic upgrade dry-run**

Run: `cd backend && .venv/bin/alembic upgrade head --sql | head -40`
Expected: `ALTER TABLE step_run ADD COLUMN recovery_count` + `ALTER TABLE step_run ADD COLUMN last_recovery_reason` + `ALTER TABLE image_asset ALTER COLUMN variant_label TYPE VARCHAR(32)` SQL 출력. 에러 없음.

- [ ] **Step 3: alembic upgrade 실제 적용**

Run: `cd backend && .venv/bin/alembic upgrade head`
Expected: 정상 종료. 새 마이그레이션 한 줄 로그.

- [ ] **Step 4: PostgreSQL 검증**

Run:
```bash
.venv/bin/python -c "
from sqlalchemy import create_engine, text
eng = create_engine('postgresql://theroad:theroad_dev_2026@localhost:5432/theroad')
with eng.connect() as c:
    res = c.execute(text(\"SELECT column_name, data_type, character_maximum_length FROM information_schema.columns WHERE table_name='step_run' AND column_name IN ('recovery_count', 'last_recovery_reason') ORDER BY column_name\"))
    for row in res: print(row)
    res = c.execute(text(\"SELECT character_maximum_length FROM information_schema.columns WHERE table_name='image_asset' AND column_name='variant_label'\"))
    print('variant_label len:', list(res)[0][0])
"
```
Expected:
```
('last_recovery_reason', 'text', None)
('recovery_count', 'integer', None)
variant_label len: 32
```

## Task 2: database.py raw SQL `_migrations` 동기화

**Files:**
- Modify: `backend/app/core/database.py:33-144`

- [ ] **Step 1: `_migrations` 리스트에 신규 ALTER 3종 추가**

`backend/app/core/database.py` 의 `_migrations` 리스트(라인 33–144) 끝에 추가:
```python
        # Resume integrity: recovery 추적 (alembic 003과 동일, idempotent)
        "ALTER TABLE step_run ADD COLUMN IF NOT EXISTS recovery_count INTEGER NOT NULL DEFAULT 0",
        "ALTER TABLE step_run ADD COLUMN IF NOT EXISTS last_recovery_reason TEXT",
        # variant_label String(8) → String(32) — chain_bg/state_variant/composite long label 수용
        # 이미 32라면 ALTER TYPE이 무동작. 더 짧으면 expand.
        "ALTER TABLE image_asset ALTER COLUMN variant_label TYPE VARCHAR(32)",
```

- [ ] **Step 2: `step_run` CREATE TABLE에도 컬럼 포함 (test PG)**

같은 파일의 `CREATE TABLE IF NOT EXISTS step_run` 블록(라인 37–62)에 컬럼 추가:
```python
        """CREATE TABLE IF NOT EXISTS step_run (
            ...  (기존 그대로)
            sync_status text,
            sync_error text,
            synced_at text,
            recovery_count integer NOT NULL DEFAULT 0,
            last_recovery_reason text,
            CONSTRAINT step_run_project_id_episode_id_step_id_key
                UNIQUE(project_id, episode_id, step_id)
        )""",
```

기존 라인은 그대로 두고 `synced_at text,` 다음에 두 컬럼 추가만.

- [ ] **Step 3: backend init_db 실행해서 마이그레이션 idempotent 동작 검증**

Run:
```bash
.venv/bin/python -c "from app.core.database import init_db; init_db(); print('OK')"
```
Expected: `OK` 출력. 에러 없음. (이미 alembic으로 컬럼이 추가된 상태라 ALTER가 no-op)

## Task 3: ImageAsset.variant_label String(32) 모델 동기화

**Files:**
- Modify: `backend/app/models/project.py:181`

- [ ] **Step 1: 현재 uncommitted 변경 확인**

Run: `git diff backend/app/models/project.py | head -10`
Expected: variant_label `String(8)` → `String(32)` 변경이 보임. 이미 적용되어 있어야 함 (memory `next_session_integrity_validation.md`).

- [ ] **Step 2: 만약 적용 안 되었으면 직접 수정**

`backend/app/models/project.py:181` 라인을:
```python
    variant_label = Column(String(32), nullable=False, default="v00", server_default="v00")
```
로 확인/수정.

## Task 4: `integrity_report.py` 작성 (CompletionReport + CleanupReport)

**Files:**
- Create: `backend/app/core/integrity_report.py`
- Test: `backend/tests/core/test_integrity_report.py`

- [ ] **Step 1: 실패 테스트 작성**

`backend/tests/core/test_integrity_report.py`:
```python
"""CompletionReport / CleanupReport dataclass 단위 테스트."""
import pytest

from app.core.integrity_report import CompletionReport, CleanupReport


def test_completion_report_clean_default():
    r = CompletionReport(is_complete=True, missing=[], severity="clean", metadata={})
    assert r.is_complete is True
    assert r.severity == "clean"


def test_completion_report_missing_severity():
    r = CompletionReport(is_complete=False, missing=["5/12 ref missing"],
                         severity="missing", metadata={"expected": 12, "found": 7})
    assert not r.is_complete
    assert r.severity == "missing"
    assert r.metadata["expected"] == 12


def test_completion_report_frozen():
    r = CompletionReport(is_complete=True, missing=[], severity="clean", metadata={})
    with pytest.raises(Exception):
        r.is_complete = False  # frozen=True 이므로 변경 불가


def test_cleanup_report_zero_default():
    r = CleanupReport(deleted_db_rows=0, deleted_files=0, targets=[], skipped=[])
    assert r.deleted_db_rows == 0
    assert r.targets == []


def test_cleanup_report_with_targets():
    r = CleanupReport(deleted_db_rows=12, deleted_files=8,
                      targets=["image_asset(reference)"], skipped=["scene_still(other_step)"])
    assert r.deleted_db_rows == 12
    assert "scene_still(other_step)" in r.skipped
```

- [ ] **Step 2: 테스트 실행 (실패 확인)**

Run: `cd backend && .venv/bin/pytest tests/core/test_integrity_report.py -v`
Expected: ImportError — `app.core.integrity_report` 모듈 없음.

- [ ] **Step 3: `integrity_report.py` 작성**

`backend/app/core/integrity_report.py`:
```python
"""Step 산출물 무결성 검증 리포트 dataclass.

verify_completion()이 반환하는 CompletionReport와 cleanup_artifacts()가 반환하는
CleanupReport를 정의한다. 모두 frozen — 호출자는 로깅·결정에만 사용.
"""
from dataclasses import dataclass, field
from typing import Literal


@dataclass(frozen=True)
class CompletionReport:
    """Step 산출물 무결성 검증 결과.

    is_complete=True → resume entry verify에서 skipped 결정 + exit verify에서 completed 마킹.
    is_complete=False → entry: mode='force' 격상, exit: status='partial' 마킹.
    """
    is_complete: bool
    missing: list[str]                 # 사람이 읽는 결손 목록
    severity: Literal["clean", "partial", "missing"]
    metadata: dict                     # 진단용 — count/file paths


@dataclass(frozen=True)
class CleanupReport:
    """force/auto-rerun 시 stale artifact 정리 결과.

    deleted_db_rows + deleted_files == 0 이면 noop (override 안 한 step의 default).
    """
    deleted_db_rows: int
    deleted_files: int
    targets: list[str]                 # 정리된 대상 (감사 로그)
    skipped: list[str]                 # self-origin 외 보호된 대상 (drift detection)
```

- [ ] **Step 4: 테스트 통과 확인**

Run: `cd backend && .venv/bin/pytest tests/core/test_integrity_report.py -v`
Expected: 5 passed.

- [ ] **Step 5: 커밋**

```bash
git add backend/app/core/integrity_report.py backend/tests/core/test_integrity_report.py
git commit -m "feat(integrity): CompletionReport + CleanupReport dataclass"
```

## Task 5: StepRunner에 verify_completion / cleanup_artifacts default 메서드 + _safe_verify_completion

**Files:**
- Modify: `backend/app/core/step_runner.py` (run() 메서드 위 + import)
- Test: `backend/tests/core/test_step_runner_resilience.py`

- [ ] **Step 1: 실패 테스트 작성**

`backend/tests/core/test_step_runner_resilience.py` (신규 파일):
```python
"""StepRunner resume 무결성 framework 통합 테스트."""
from app.core.integrity_report import CompletionReport, CleanupReport
from app.core.step_runner import StepRunner


def test_base_verify_completion_returns_clean(make_step_runner):
    """베이스 verify_completion은 항상 is_complete=True (override 안 한 step 보호)."""
    runner = make_step_runner("text_cleanup")
    report = runner.verify_completion()
    assert report.is_complete is True
    assert report.severity == "clean"
    assert report.missing == []


def test_base_cleanup_artifacts_returns_noop(make_step_runner):
    """베이스 cleanup_artifacts는 항상 deleted=0 (이미지 step 보호)."""
    runner = make_step_runner("text_cleanup")
    report = runner.cleanup_artifacts()
    assert report.deleted_db_rows == 0
    assert report.deleted_files == 0
    assert report.targets == []


def test_safe_verify_completion_catches_exception(make_step_runner):
    """verify_completion이 예외 던지면 _safe wrapper가 is_complete=False로 안전 처리."""
    runner = make_step_runner("text_cleanup")

    def crash(self):
        raise RuntimeError("boom")
    runner.verify_completion = crash.__get__(runner, type(runner))

    report = runner._safe_verify_completion()
    assert report.is_complete is False
    assert "verify_crashed" in report.missing[0]
    assert "boom" in report.metadata["crash"]
```

`backend/tests/core/conftest.py`(또는 적절한 conftest)에 fixture가 없으면 추가:
```python
import pytest

@pytest.fixture
def make_step_runner(pg_session, tmp_project_id, tmp_episode_id):
    """StepRunner 인스턴스 팩토리. step_id 임의 변경 가능."""
    def _make(step_id):
        from app.core.step_runner import StepRunner
        return StepRunner(
            step_id=step_id,
            project_id=tmp_project_id,
            episode_id=tmp_episode_id,
            db=pg_session,
            project_config={},
        )
    return _make
```

(`tmp_project_id` / `tmp_episode_id` / `pg_session` fixture는 기존 conftest_pg.py 패턴 활용. 필요 시 작성. 기존 테스트 conftest 확인.)

- [ ] **Step 2: 테스트 실행 (실패 확인)**

Run: `cd backend && .venv/bin/pytest tests/core/test_step_runner_resilience.py -v`
Expected: AttributeError — `verify_completion` / `cleanup_artifacts` / `_safe_verify_completion` 메서드 없음.

- [ ] **Step 3: StepRunner에 메서드 추가**

`backend/app/core/step_runner.py` 의 `_execute()` (라인 617) 바로 위에 추가:
```python
    # ── 무결성 검증 (Phase Resume Integrity) ──

    def verify_completion(self):
        """산출물 무결성 검증. 베이스는 항상 complete — override 안 한 step은 검증 없음.

        Resume entry + execute exit 양방향에서 호출됨. is_complete=False 반환 시:
          - entry: mode='force' 격상 (auto-rerun)
          - exit: status='partial' 마킹

        Override 시 자기 step 산출물(DB row + 파일 stat)만 검증한다. cross-step
        dep verify는 후속 spec.
        """
        from app.core.integrity_report import CompletionReport
        return CompletionReport(is_complete=True, missing=[], severity="clean", metadata={})

    def cleanup_artifacts(self):
        """force/auto-rerun 시 stale artifact 정리. 베이스는 noop.

        Override 작성 기준 (매우 보수적):
        - 전체 force = delete-and-rebuild가 항상 안전한 step만
        - 부분 재생성을 endpoint로 지원하는 step은 절대 override 금지
        - 이미지 step은 거의 모두 default noop 유지 (사용자 caveat)
        """
        from app.core.integrity_report import CleanupReport
        return CleanupReport(deleted_db_rows=0, deleted_files=0, targets=[], skipped=[])

    def _safe_verify_completion(self):
        """verify_completion이 예외를 던지면 안전하게 is_complete=False로 처리."""
        from app.core.integrity_report import CompletionReport
        try:
            return self.verify_completion()
        except Exception as exc:
            logger.error("verify_completion crashed for %s: %s", self.step_id, exc)
            return CompletionReport(
                is_complete=False, missing=[f"verify_crashed: {exc}"],
                severity="missing", metadata={"crash": str(exc)},
            )
```

- [ ] **Step 4: 테스트 통과 확인**

Run: `cd backend && .venv/bin/pytest tests/core/test_step_runner_resilience.py -v`
Expected: 3 passed.

- [ ] **Step 5: 커밋**

```bash
git add backend/app/core/step_runner.py backend/tests/core/test_step_runner_resilience.py
git commit -m "feat(integrity): verify_completion + cleanup_artifacts hooks (default safe)"
```

## Task 6: StepRunner에 recovery counter 메서드 추가

**Files:**
- Modify: `backend/app/core/step_runner.py`
- Modify: `backend/tests/core/test_step_runner_resilience.py`

- [ ] **Step 1: 실패 테스트 추가**

기존 `test_step_runner_resilience.py`에 추가:
```python
def test_record_recovery_increments_count(make_step_runner, pg_session):
    """_record_recovery()는 recovery_count++ + last_recovery_reason 기록."""
    runner = make_step_runner("text_cleanup")
    runner._update_step_run("completed")  # row 생성

    new_count = runner._record_recovery("verify failed: 3 missing")

    assert new_count == 1
    assert runner._get_recovery_count() == 1
    assert runner._get_last_recovery_reason() == "verify failed: 3 missing"


def test_record_recovery_repeats(make_step_runner):
    runner = make_step_runner("text_cleanup")
    runner._update_step_run("completed")
    runner._record_recovery("first")
    runner._record_recovery("second")
    assert runner._get_recovery_count() == 2
    assert runner._get_last_recovery_reason() == "second"


def test_reset_recovery_counter(make_step_runner):
    runner = make_step_runner("text_cleanup")
    runner._update_step_run("completed")
    runner._record_recovery("oops")
    assert runner._get_recovery_count() == 1
    runner._reset_recovery_counter()
    assert runner._get_recovery_count() == 0
    assert runner._get_last_recovery_reason() == "(없음)"


def test_get_recovery_count_no_row_returns_zero(make_step_runner):
    runner = make_step_runner("text_cleanup")
    # step_run row 미존재
    assert runner._get_recovery_count() == 0
    assert runner._get_last_recovery_reason() == "(없음)"
```

- [ ] **Step 2: 테스트 실행 (실패 확인)**

Run: `cd backend && .venv/bin/pytest tests/core/test_step_runner_resilience.py -v -k recovery`
Expected: AttributeError 4건.

- [ ] **Step 3: 메서드 4개 추가**

`backend/app/core/step_runner.py` 의 `cleanup_artifacts()` 바로 다음에 추가:
```python
    def _record_recovery(self, reason: str) -> int:
        """recovery_count++ + last_recovery_reason 적재. 새 카운터값 반환.

        호출 site: entry verify 실패 시. mode='force' 격상 직전.
        """
        now = self._now()
        self.db.execute(text("""
            UPDATE step_run SET
                recovery_count = recovery_count + 1,
                last_recovery_reason = :reason,
                updated_at = :now
            WHERE project_id = :pid AND episode_id = :eid AND step_id = :sid
        """), {"reason": reason[:2000], "pid": self.project_id,
               "eid": self.episode_id, "sid": self.step_id, "now": now})
        self.db.commit()
        return self._get_recovery_count()

    def _reset_recovery_counter(self) -> None:
        """status='completed' 정상 진행 시 카운터 reset (다음 sweep 깨끗한 상태)."""
        now = self._now()
        self.db.execute(text("""
            UPDATE step_run SET recovery_count = 0, last_recovery_reason = NULL, updated_at = :now
            WHERE project_id = :pid AND episode_id = :eid AND step_id = :sid
        """), {"pid": self.project_id, "eid": self.episode_id,
               "sid": self.step_id, "now": now})
        self.db.commit()

    def _get_recovery_count(self) -> int:
        row = self.db.execute(text(
            "SELECT recovery_count FROM step_run "
            "WHERE project_id = :pid AND episode_id = :eid AND step_id = :sid"
        ), {"pid": self.project_id, "eid": self.episode_id, "sid": self.step_id}).fetchone()
        return row[0] if row else 0

    def _get_last_recovery_reason(self) -> str:
        row = self.db.execute(text(
            "SELECT last_recovery_reason FROM step_run "
            "WHERE project_id = :pid AND episode_id = :eid AND step_id = :sid"
        ), {"pid": self.project_id, "eid": self.episode_id, "sid": self.step_id}).fetchone()
        return (row[0] if row and row[0] else "(없음)")
```

- [ ] **Step 4: 테스트 통과 확인**

Run: `cd backend && .venv/bin/pytest tests/core/test_step_runner_resilience.py -v -k recovery`
Expected: 4 passed.

- [ ] **Step 5: 커밋**

```bash
git add backend/app/core/step_runner.py backend/tests/core/test_step_runner_resilience.py
git commit -m "feat(integrity): recovery counter persistence (record/reset/get)"
```

## Task 7: `_mask_sensitive_keys` 추가

**Files:**
- Modify: `backend/app/core/step_runner.py`
- Modify: `backend/tests/core/test_step_runner_resilience.py`

- [ ] **Step 1: 실패 테스트 추가**

```python
def test_mask_sensitive_keys_basic(make_step_runner):
    runner = make_step_runner("text_cleanup")
    cfg = {
        "model": "gpt-5.5",
        "openai_api_key": "sk-xyz",
        "anthropic_api_key": "sk-ant-aaa",
        "password": "p@ss",
        "user_token": "abc",
        "secret_value": "shh",
        "credential": "raw",
        "nested": {"api_key": "deep"},
    }
    masked = runner._mask_sensitive_keys(cfg)
    assert masked["model"] == "gpt-5.5"
    assert masked["openai_api_key"] == "***"
    assert masked["anthropic_api_key"] == "***"
    assert masked["password"] == "***"
    assert masked["user_token"] == "***"
    assert masked["secret_value"] == "***"
    assert masked["credential"] == "***"
    assert masked["nested"]["api_key"] == "***"


def test_mask_sensitive_keys_empty_input(make_step_runner):
    runner = make_step_runner("text_cleanup")
    assert runner._mask_sensitive_keys({}) == {}
    assert runner._mask_sensitive_keys(None) == {}
```

- [ ] **Step 2: 테스트 실행 (실패 확인)**

Run: `cd backend && .venv/bin/pytest tests/core/test_step_runner_resilience.py -v -k mask`
Expected: AttributeError.

- [ ] **Step 3: 메서드 추가**

`backend/app/core/step_runner.py`의 `_get_last_recovery_reason()` 다음에 추가:
```python
    @staticmethod
    def _mask_sensitive_keys(config):
        """project_config 저장 전 민감 키 마스킹. api_key/password/secret/token/credential 패턴.

        nested dict는 재귀. None → 빈 dict.
        D2 fix — save_checkpoint이 project_config_snapshot 기록 시 사용.
        """
        SENSITIVE = ("api_key", "password", "secret", "token", "credential")
        if not config:
            return {}
        masked = {}
        for k, v in config.items():
            if any(s in str(k).lower() for s in SENSITIVE):
                masked[k] = "***"
            elif isinstance(v, dict):
                masked[k] = StepRunner._mask_sensitive_keys(v)
            else:
                masked[k] = v
        return masked
```

- [ ] **Step 4: 테스트 통과 확인**

Run: `cd backend && .venv/bin/pytest tests/core/test_step_runner_resilience.py -v -k mask`
Expected: 2 passed.

- [ ] **Step 5: 커밋**

```bash
git add backend/app/core/step_runner.py backend/tests/core/test_step_runner_resilience.py
git commit -m "feat(integrity): _mask_sensitive_keys for snapshot save"
```

## Task 8: save_checkpoint이 project_config_snapshot 기록 (D2)

**Files:**
- Modify: `backend/app/core/step_runner.py:289-335` (save_checkpoint)
- Modify: `backend/tests/core/test_step_runner_resilience.py`

- [ ] **Step 1: 실패 테스트 추가**

```python
def test_save_checkpoint_records_project_config_snapshot(make_step_runner):
    """D2 fix: save_checkpoint이 project_config_snapshot을 기록 (마스킹 포함)."""
    runner = make_step_runner("text_cleanup")
    runner.project_config = {"model": "gpt-5.5", "openai_api_key": "sk-xyz"}

    runner.save_checkpoint({"status": "completed", "data": {"foo": "bar"}})

    cp = runner.load_checkpoint()
    assert cp is not None
    assert "project_config_snapshot" in cp
    assert cp["project_config_snapshot"]["model"] == "gpt-5.5"
    assert cp["project_config_snapshot"]["openai_api_key"] == "***"
```

- [ ] **Step 2: 테스트 실행 (실패 확인)**

Run: `cd backend && .venv/bin/pytest tests/core/test_step_runner_resilience.py -v -k snapshot`
Expected: KeyError 또는 assertion 실패.

- [ ] **Step 3: save_checkpoint 수정**

`backend/app/core/step_runner.py:289-320` (save_checkpoint 메서드) — `data["config_hash"]` 라인 다음에 추가:
```python
        if "schema_version" not in data:
            data["schema_version"] = self.manifest.get("schema_version", 1)
        if "config_hash" not in data:
            data["config_hash"] = compute_config_hash(self.project_config)
        # D2 fix: project_config_snapshot 기록 — _diff_project_config가 None reference로
        # 폴백되어 매번 stale auto-rerun trigger되는 패턴 종결.
        # 민감 키(api_key/password/token/secret/credential)는 마스킹.
        if "project_config_snapshot" not in data:
            data["project_config_snapshot"] = self._mask_sensitive_keys(self.project_config)
```

- [ ] **Step 4: 테스트 통과 + 회귀 확인**

Run: `cd backend && .venv/bin/pytest tests/core/test_step_runner_resilience.py -v -k snapshot`
Expected: 1 passed.

Run: `cd backend && .venv/bin/pytest tests/core/ -v --tb=short`
Expected: 모든 step_runner 관련 테스트 PASS, 회귀 0.

- [ ] **Step 5: 커밋**

```bash
git add backend/app/core/step_runner.py backend/tests/core/test_step_runner_resilience.py
git commit -m "fix(integrity): D2 — save_checkpoint records project_config_snapshot"
```

## Task 9: `_check_cp_mismatch` 헬퍼 추출 (현행 mismatch 로직 정리)

**Files:**
- Modify: `backend/app/core/step_runner.py` run() 분기 (라인 487-547)

- [ ] **Step 1: 헬퍼 메서드 추출**

현재 `run()` 안의 mismatch 검증 로직을 별도 메서드로 추출. `_safe_verify_completion()` 다음에 추가:
```python
    def _check_cp_mismatch(self, cp: dict):
        """cp의 schema_version + config_hash와 현재 manifest/project_config 비교.

        Returns: mismatch reason str (None이면 일치).
        run() resume 분기의 P0-3 mismatch 검증을 헬퍼로 이관.
        """
        current_schema = self.manifest.get("schema_version", 1)
        cp_schema = cp.get("schema_version")
        cp_hash = cp.get("config_hash")

        if cp_schema is not None and cp_schema != current_schema:
            return f"schema_version mismatch: 체크포인트={cp_schema}, 현재={current_schema}"

        if cp_hash is not None:
            current_hash = compute_config_hash(self.project_config)
            if cp_hash != current_hash:
                return "config_hash mismatch: project_config 변경 감지"

        return None
```

- [ ] **Step 2: run()의 inline mismatch 검증을 헬퍼 호출로 교체**

`backend/app/core/step_runner.py:498-548` 의 inline 검증 블록을 다음으로 교체. **현행 동작 100% 보존** (단, 헬퍼 호출 형태로):
```python
            cp = self.load_checkpoint()
            if cp:
                mismatch_reason = self._check_cp_mismatch(cp)
                if mismatch_reason:
                    # diff 로깅 — config_hash mismatch 시 어느 key가 바뀌었는지 명시.
                    from app.core.applicability import _diff_project_config
                    cp_snapshot = cp.get("project_config_snapshot")
                    try:
                        config_diff = _diff_project_config(cp_snapshot, self.project_config)
                    except Exception as diff_exc:
                        config_diff = {"_diff_error": str(diff_exc)}
                    logger.warning(
                        "Step %s stale detected: %s. Project config diff: %s",
                        self.step_id, mismatch_reason, config_diff,
                    )

                    if self.project_config.get("strict_resume", False):
                        raise AppError(
                            code="step.resume_invalid",
                            message=(
                                f"{self.step_id} 체크포인트가 stale 입니다 "
                                f"({mismatch_reason}). strict_resume=True 이므로 "
                                "force 모드로 재실행하세요."
                            ),
                            status_code=409,
                        )

                    logger.info("Step %s: stale → auto-rerun (resume mode default)", self.step_id)
                    mode = "force"
                else:
                    return {"status": "skipped", "reason": "already completed"}
            else:
                return {"status": "skipped", "reason": "already completed"}
```

(이건 **헬퍼 추출만**이고 동작 동일. 다음 task에서 verify_completion/cp=None 분기 추가)

- [ ] **Step 3: 회귀 확인**

Run: `cd backend && .venv/bin/pytest tests/ -v --tb=short -k step_runner`
Expected: 회귀 0. 기존 테스트 모두 PASS.

- [ ] **Step 4: 커밋**

```bash
git add backend/app/core/step_runner.py
git commit -m "refactor(step_runner): _check_cp_mismatch 헬퍼 추출 (동작 변경 없음)"
```

## Task 10: run() resume 분기에 verify_completion 통합 + cp=None 처리 (D1 핵심)

**Files:**
- Modify: `backend/app/core/step_runner.py:487-548` (run() resume 분기)
- Modify: `backend/tests/core/test_step_runner_resilience.py`

- [ ] **Step 1: 실패 테스트 작성**

```python
def test_resume_with_cp_and_verify_pass_skips(make_step_runner, monkeypatch):
    """status=completed + cp 있음 + verify pass → skipped 반환."""
    runner = make_step_runner("text_cleanup")
    runner._update_step_run("completed")
    runner.save_checkpoint({"status": "completed", "data": {}})

    result = runner.run(mode="resume")
    assert result["status"] == "skipped"


def test_resume_with_cp_none_triggers_force(make_step_runner, monkeypatch):
    """D1 핵심: status=completed + cp=None (manifest 폐기) → force 격상."""
    runner = make_step_runner("text_cleanup")
    runner._update_step_run("completed")
    # 의도적으로 cp 미저장 (또는 .force_cleared marker 설정)
    runner.clear_checkpoint()  # marker 작성됨
    # _execute가 1회 호출되어야 정상 (force 격상 검증)
    executed = []

    def stub_execute(self, mode):
        executed.append(mode)
        return {"completed_count": 1, "applicable_count": 1, "failed_count": 0, "data": {}}
    runner._execute = stub_execute.__get__(runner, type(runner))

    runner.run(mode="resume")
    assert executed == ["force"], "cp=None이면 force-rerun으로 격상되어야 함"
    assert runner._get_recovery_count() == 1
    assert "checkpoint missing" in runner._get_last_recovery_reason()


def test_resume_with_verify_fail_triggers_force(make_step_runner, monkeypatch):
    """status=completed + cp 있음 + verify_completion 실패 → force 격상 + recovery_count 기록."""
    runner = make_step_runner("text_cleanup")
    runner._update_step_run("completed")
    runner.save_checkpoint({"status": "completed", "data": {}})

    from app.core.integrity_report import CompletionReport

    def stub_verify(self):
        return CompletionReport(is_complete=False, missing=["X"], severity="missing", metadata={})
    runner.verify_completion = stub_verify.__get__(runner, type(runner))

    executed = []
    def stub_execute(self, mode):
        executed.append(mode)
        return {"completed_count": 1, "applicable_count": 1, "failed_count": 0, "data": {}}
    runner._execute = stub_execute.__get__(runner, type(runner))

    runner.run(mode="resume")
    assert executed == ["force"]
    assert runner._get_recovery_count() == 1
    assert "verify failed" in runner._get_last_recovery_reason()
```

- [ ] **Step 2: 테스트 실행 (실패 확인)**

Run: `cd backend && .venv/bin/pytest tests/core/test_step_runner_resilience.py -v -k "resume_with"`
Expected: assertion 실패 (현재는 cp=None이면 skipped, verify는 호출 안 됨).

- [ ] **Step 3: run() resume 분기 통합**

`backend/app/core/step_runner.py` 의 run() resume 분기(라인 487-548)를 다음으로 교체:
```python
        # 3. resume: 이미 완료면 verify_completion + cp 무결성 검증
        if mode == "resume":
            existing = self._get_step_run(self.step_id)
            if existing and existing[0] == "completed":
                cp = self.load_checkpoint()
                mismatch = None

                if cp is None:
                    # D1: status=completed + cp=None = projection mismatch.
                    # PID 0bb48ebf 사고 패턴(manifest 폐기 + step_run.completed) 재발 가드.
                    mismatch = "checkpoint missing but step_run.status=completed"
                else:
                    # 현행 schema/config_hash mismatch 검증
                    mismatch = self._check_cp_mismatch(cp)
                    if not mismatch:
                        # NEW: verify_completion entry 호출 (양방향 verify)
                        report = self._safe_verify_completion()
                        if not report.is_complete:
                            mismatch = f"verify failed: {report.missing}"

                if mismatch:
                    # diff 로깅 (config_hash mismatch인 경우 어느 key가 바뀌었는지)
                    if cp:
                        from app.core.applicability import _diff_project_config
                        cp_snapshot = cp.get("project_config_snapshot")
                        try:
                            config_diff = _diff_project_config(cp_snapshot, self.project_config)
                        except Exception as diff_exc:
                            config_diff = {"_diff_error": str(diff_exc)}
                        logger.warning(
                            "Step %s stale detected: %s. Project config diff: %s",
                            self.step_id, mismatch, config_diff,
                        )

                    if self.project_config.get("strict_resume", False) and cp:
                        raise AppError(
                            code="step.resume_invalid",
                            message=(
                                f"{self.step_id} 체크포인트가 stale 입니다 "
                                f"({mismatch}). strict_resume=True 이므로 "
                                "force 모드로 재실행하세요."
                            ),
                            status_code=409,
                        )

                    new_count = self._record_recovery(mismatch)
                    logger.warning("[RECOVERY] step=%s reason=%s cycle=%d",
                                   self.step_id, mismatch, new_count)
                    mode = "force"
                else:
                    return {"status": "skipped", "reason": "already completed"}
```

- [ ] **Step 4: 테스트 통과 확인 + 회귀 확인**

Run: `cd backend && .venv/bin/pytest tests/core/test_step_runner_resilience.py -v -k "resume_with"`
Expected: 3 passed.

Run: `cd backend && .venv/bin/pytest tests/ --tb=short`
Expected: 회귀 0.

- [ ] **Step 5: 커밋**

```bash
git add backend/app/core/step_runner.py backend/tests/core/test_step_runner_resilience.py
git commit -m "fix(integrity): D1 — resume entry verify_completion + cp=None handling"
```

## Task 11: status별 force-like 분기 (running/failed/partial/stale/pending)

**Files:**
- Modify: `backend/app/core/step_runner.py:run()`
- Modify: `backend/tests/core/test_step_runner_resilience.py`

- [ ] **Step 1: 실패 테스트 작성**

```python
import pytest

@pytest.mark.parametrize("status", ["running", "failed", "partial", "stale", "pending"])
def test_resume_non_completed_status_force_like(make_step_runner, status):
    """추천 1: completed 외 모든 상태는 force-like 처리 (cleanup → invalidate → execute)."""
    runner = make_step_runner("text_cleanup")
    runner._update_step_run(status)

    cleanup_called = []
    def stub_cleanup(self):
        from app.core.integrity_report import CleanupReport
        cleanup_called.append(True)
        return CleanupReport(0, 0, [], [])
    runner.cleanup_artifacts = stub_cleanup.__get__(runner, type(runner))

    executed = []
    def stub_execute(self, mode):
        executed.append(mode)
        return {"completed_count": 1, "applicable_count": 1, "failed_count": 0, "data": {}}
    runner._execute = stub_execute.__get__(runner, type(runner))

    runner.run(mode="resume")
    assert executed == ["force"], f"status={status}는 force-like 처리 (mode=force)"
    assert cleanup_called == [True], f"status={status}는 cleanup_artifacts 호출"
```

- [ ] **Step 2: 테스트 실행 (실패 확인)**

Run: `cd backend && .venv/bin/pytest tests/core/test_step_runner_resilience.py -v -k force_like`
Expected: 5 case 중 일부 또는 전부 실패.

- [ ] **Step 3: run()에 status별 분기 추가**

resume 분기 다음에 추가 (Task 10에서 추가한 if 분기 직후):
```python
        # 추천 1: status='completed' 외 모든 상태는 force-like 처리.
        # running = backend kill로 멈춤 / failed = 이전 실패 / partial = 부분 성공 /
        # stale = downstream invalidate 결과 / pending = step_run row만 있고 미실행.
        # 모두 force(cleanup + invalidate + execute) 흐름 필요.
        elif mode == "resume" and existing and existing[0] in ("running", "failed", "partial", "stale", "pending"):
            logger.warning("[RECOVERY] step=%s status=%s → force-like",
                           self.step_id, existing[0])
            mode = "force"
```

(`elif`로 이전 `if existing and existing[0] == "completed":` 분기와 연결.)

- [ ] **Step 4: 테스트 통과 확인 + 회귀 확인**

Run: `cd backend && .venv/bin/pytest tests/core/test_step_runner_resilience.py -v -k force_like`
Expected: 5 passed.

Run: `cd backend && .venv/bin/pytest tests/ --tb=short`
Expected: 회귀 0.

- [ ] **Step 5: 커밋**

```bash
git add backend/app/core/step_runner.py backend/tests/core/test_step_runner_resilience.py
git commit -m "feat(integrity): status별 force-like 분기 (추천 1)"
```

## Task 12: cleanup_artifacts 호출 + 예외 처리 (force path)

**Files:**
- Modify: `backend/app/core/step_runner.py:run()` force 분기

- [ ] **Step 1: 실패 테스트 작성**

```python
def test_force_calls_cleanup_artifacts(make_step_runner):
    runner = make_step_runner("text_cleanup")
    runner._update_step_run("completed")
    runner.save_checkpoint({"status": "completed", "data": {}})

    cleanup_calls = []
    def stub_cleanup(self):
        from app.core.integrity_report import CleanupReport
        cleanup_calls.append(True)
        return CleanupReport(deleted_db_rows=5, deleted_files=2, targets=["x"], skipped=[])
    runner.cleanup_artifacts = stub_cleanup.__get__(runner, type(runner))

    def stub_execute(self, mode):
        return {"completed_count": 1, "applicable_count": 1, "failed_count": 0, "data": {}}
    runner._execute = stub_execute.__get__(runner, type(runner))

    runner.run(mode="force")
    assert cleanup_calls == [True]


def test_cleanup_artifacts_exception_marks_failed(make_step_runner):
    """cleanup_artifacts 예외 시 즉시 raise + step_run='failed' 마킹."""
    runner = make_step_runner("text_cleanup")
    runner._update_step_run("completed")

    def stub_cleanup(self):
        raise RuntimeError("cleanup boom")
    runner.cleanup_artifacts = stub_cleanup.__get__(runner, type(runner))

    with pytest.raises(RuntimeError, match="cleanup boom"):
        runner.run(mode="force")

    row = runner._get_step_run("text_cleanup")
    assert row[0] == "failed"
```

- [ ] **Step 2: 테스트 실행 (실패 확인)**

Run: `cd backend && .venv/bin/pytest tests/core/test_step_runner_resilience.py -v -k cleanup`
Expected: 일부 실패.

- [ ] **Step 3: run() force 분기 수정**

`backend/app/core/step_runner.py` 의 force 분기 (라인 552-555):
```python
        # 4. force: cleanup → invalidate → clear_checkpoint
        if mode == "force":
            try:
                cleanup_report = self.cleanup_artifacts()
            except Exception as exc:
                # cleanup 실패 = DB inconsistent 위험 → 즉시 raise + failed 마킹.
                self._update_step_run("failed", error_message=f"cleanup_artifacts crashed: {exc}")
                logger.error("Step %s cleanup_artifacts crashed: %s", self.step_id, exc)
                raise
            if cleanup_report.deleted_db_rows or cleanup_report.deleted_files:
                logger.warning("[CLEANUP] step=%s rows=%d files=%d targets=%s",
                               self.step_id, cleanup_report.deleted_db_rows,
                               cleanup_report.deleted_files, cleanup_report.targets)
            self.invalidate_downstream()
            self.clear_checkpoint()
```

- [ ] **Step 4: 테스트 통과 확인 + 회귀 확인**

Run: `cd backend && .venv/bin/pytest tests/core/test_step_runner_resilience.py -v -k cleanup`
Expected: 2 passed.

Run: `cd backend && .venv/bin/pytest tests/ --tb=short`
Expected: 회귀 0.

- [ ] **Step 5: 커밋**

```bash
git add backend/app/core/step_runner.py backend/tests/core/test_step_runner_resilience.py
git commit -m "feat(integrity): cleanup_artifacts 호출 + 예외 처리 (force path)"
```

## Task 13: exit verify + recovery counter reset (run() 후반부)

**Files:**
- Modify: `backend/app/core/step_runner.py:run()` 실행 후반부 (라인 565-607)
- Modify: `backend/tests/core/test_step_runner_resilience.py`

- [ ] **Step 1: 실패 테스트 작성**

```python
def test_exit_verify_pass_completes(make_step_runner):
    runner = make_step_runner("text_cleanup")

    def stub_execute(self, mode):
        return {"completed_count": 5, "applicable_count": 5, "failed_count": 0, "data": {}}
    runner._execute = stub_execute.__get__(runner, type(runner))

    result = runner.run(mode="force")
    assert result["status"] == "completed"


def test_exit_verify_fail_marks_partial(make_step_runner):
    """exit verify 실패 → final_status='partial' 마킹."""
    runner = make_step_runner("text_cleanup")

    def stub_execute(self, mode):
        return {"completed_count": 5, "applicable_count": 5, "failed_count": 0, "data": {}}
    runner._execute = stub_execute.__get__(runner, type(runner))

    from app.core.integrity_report import CompletionReport
    def stub_verify(self):
        return CompletionReport(is_complete=False, missing=["x"], severity="partial", metadata={})
    runner.verify_completion = stub_verify.__get__(runner, type(runner))

    result = runner.run(mode="force")
    assert result["status"] == "partial"


def test_completed_resets_recovery_counter(make_step_runner):
    """status=completed 정상 종료 시 recovery_count → 0 reset."""
    runner = make_step_runner("text_cleanup")
    runner._update_step_run("completed")
    runner._record_recovery("prior")
    assert runner._get_recovery_count() == 1

    def stub_execute(self, mode):
        return {"completed_count": 5, "applicable_count": 5, "failed_count": 0, "data": {}}
    runner._execute = stub_execute.__get__(runner, type(runner))

    runner.run(mode="force")
    assert runner._get_recovery_count() == 0
    assert runner._get_last_recovery_reason() == "(없음)"
```

- [ ] **Step 2: 테스트 실행 (실패 확인)**

Run: `cd backend && .venv/bin/pytest tests/core/test_step_runner_resilience.py -v -k "exit_verify or resets_recovery"`
Expected: 일부 실패.

- [ ] **Step 3: run() 후반부 수정**

`backend/app/core/step_runner.py:566-607` 의 try 블록:
```python
        try:
            result = self._execute(mode)

            completed = result.get("completed_count", 1)
            total = result.get("applicable_count", 1)
            failed = result.get("failed_count", 0)

            final_status = "completed" if failed == 0 else ("partial" if completed > 0 else "failed")

            # NEW: Exit verify (양방향 — entry는 resume 분기에서 호출됨).
            # final_status='completed' 일 때만 verify 호출 (failed/partial은 이미 자기 카운터로 마킹됨).
            if final_status == "completed":
                exit_report = self._safe_verify_completion()
                if not exit_report.is_complete:
                    logger.warning("[VERIFY-EXIT] step=%s failed → partial: %s",
                                   self.step_id, exit_report.missing)
                    final_status = "partial"

            self._update_step_run(
                final_status,
                completed_count=completed,
                applicable_count=total,
                failed_count=failed,
                result_summary=json.dumps(
                    {k: v for k, v in result.items() if k != "data" and not isinstance(v, bytes)},
                    ensure_ascii=False, default=str,
                )[:2000],
            )
            self.save_checkpoint({"status": final_status, **result})

            # NEW: 정상 완료 시 recovery counter reset
            if final_status == "completed":
                self._reset_recovery_counter()

            # 이하 editorial cascade 등 기존 로직 그대로...
```

(`mods = self.manifest.get(...)` 이하 editorial cascade는 변경 없음.)

- [ ] **Step 4: 테스트 통과 확인 + 회귀 확인**

Run: `cd backend && .venv/bin/pytest tests/core/test_step_runner_resilience.py -v -k "exit_verify or resets_recovery"`
Expected: 3 passed.

Run: `cd backend && .venv/bin/pytest tests/ --tb=short`
Expected: 회귀 0.

- [ ] **Step 5: 커밋**

```bash
git add backend/app/core/step_runner.py backend/tests/core/test_step_runner_resilience.py
git commit -m "feat(integrity): exit verify + recovery counter reset"
```

## Task 14: recovery_exhausted 차단 (3회 후)

**Files:**
- Modify: `backend/app/core/step_runner.py:run()` 시작부
- Modify: `backend/tests/core/test_step_runner_resilience.py`

- [ ] **Step 1: 실패 테스트 작성**

```python
def test_recovery_exhausted_after_3_attempts(make_step_runner):
    """recovery_count >= 3 + mode=force → step.recovery_exhausted AppError."""
    from app.core.errors import AppError

    runner = make_step_runner("text_cleanup")
    runner._update_step_run("completed")
    runner._record_recovery("a")
    runner._record_recovery("b")
    runner._record_recovery("c")
    assert runner._get_recovery_count() == 3

    # cp 없는 상태 → mode=force 격상 → recovery_exhausted
    runner.clear_checkpoint()
    with pytest.raises(AppError) as exc:
        runner.run(mode="resume")
    assert exc.value.code == "step.recovery_exhausted"
    assert "auto-recovery 3회" in exc.value.message or "3" in exc.value.message
```

- [ ] **Step 2: 테스트 실행 (실패 확인)**

Run: `cd backend && .venv/bin/pytest tests/core/test_step_runner_resilience.py -v -k recovery_exhausted`
Expected: 1 failure.

- [ ] **Step 3: run() force 분기 직전 차단 추가**

`backend/app/core/step_runner.py` 의 force 분기(`if mode == "force":`) 직전에 추가:
```python
        # 추천 2: recovery loop 무한 차단.
        # mode=force로 격상되었거나 사용자가 force 호출했고, 누적 recovery_count가 한계면 차단.
        # 사용자가 진단해야 할 시점 — silent recovery로 묻히면 안 됨.
        MAX_RECOVERY_ATTEMPTS = 3
        if mode == "force" and self._get_recovery_count() >= MAX_RECOVERY_ATTEMPTS:
            raise AppError(
                code="step.recovery_exhausted",
                message=(
                    f"{self.step_id} auto-recovery {self._get_recovery_count()}회 실패. "
                    f"마지막 사유: {self._get_last_recovery_reason()}. 수동 진단 필요."
                ),
                status_code=409,
            )
```

- [ ] **Step 4: 테스트 통과 + 회귀 확인**

Run: `cd backend && .venv/bin/pytest tests/core/test_step_runner_resilience.py -v -k recovery_exhausted`
Expected: 1 passed.

Run: `cd backend && .venv/bin/pytest tests/ --tb=short`
Expected: 회귀 0.

- [ ] **Step 5: 커밋**

```bash
git add backend/app/core/step_runner.py backend/tests/core/test_step_runner_resilience.py
git commit -m "feat(integrity): recovery_exhausted 차단 (3회 후, 수동 진단)"
```

## Task 15: integration test — entry/exit 양방향 cycle

**Files:**
- Modify: `backend/tests/core/test_step_runner_resilience.py`

- [ ] **Step 1: 통합 시나리오 테스트 추가**

```python
def test_full_cycle_entry_exit_recovery(make_step_runner):
    """End-to-end: 정상 → entry verify fail → force → execute → exit verify pass → recovery reset."""
    runner = make_step_runner("text_cleanup")
    runner._update_step_run("completed")
    runner.save_checkpoint({"status": "completed", "data": {}})

    # cycle 1 — entry verify 실패하도록 stub
    from app.core.integrity_report import CompletionReport
    fail_count = [0]

    def stub_verify(self):
        fail_count[0] += 1
        if fail_count[0] == 1:
            return CompletionReport(is_complete=False, missing=["x"], severity="missing", metadata={})
        return CompletionReport(is_complete=True, missing=[], severity="clean", metadata={})
    runner.verify_completion = stub_verify.__get__(runner, type(runner))

    def stub_execute(self, mode):
        return {"completed_count": 5, "applicable_count": 5, "failed_count": 0, "data": {}}
    runner._execute = stub_execute.__get__(runner, type(runner))

    result = runner.run(mode="resume")
    assert result["status"] == "completed"
    assert runner._get_recovery_count() == 0  # exit pass로 reset
    # entry 1회 + exit 1회 = 2회 호출
    assert fail_count[0] == 2
```

- [ ] **Step 2: 테스트 실행**

Run: `cd backend && .venv/bin/pytest tests/core/test_step_runner_resilience.py -v -k full_cycle`
Expected: 1 passed.

- [ ] **Step 3: 커밋**

```bash
git add backend/tests/core/test_step_runner_resilience.py
git commit -m "test(integrity): full entry-exit recovery cycle integration"
```

## Task 16: integration test — `.force_cleared` marker 상호작용

**Files:**
- Modify: `backend/tests/core/test_step_runner_resilience.py`

- [ ] **Step 1: 마커 상호작용 테스트 추가**

```python
def test_force_cleared_marker_with_auto_rerun(make_step_runner):
    """auto-rerun으로 force 격상 → clear_checkpoint이 marker 작성 → 정상 save 시 marker 제거."""
    runner = make_step_runner("text_cleanup")
    runner._update_step_run("completed")
    runner.save_checkpoint({"status": "completed", "data": {}})

    from app.core.integrity_report import CompletionReport
    def stub_verify(self):
        # 1회 entry fail → force 후 exit pass 만들기 위해 호출 카운트 분기
        if not hasattr(self, '_v_call_count'):
            self._v_call_count = 0
        self._v_call_count += 1
        if self._v_call_count == 1:
            return CompletionReport(is_complete=False, missing=["x"], severity="missing", metadata={})
        return CompletionReport(is_complete=True, missing=[], severity="clean", metadata={})
    runner.verify_completion = stub_verify.__get__(runner, type(runner))

    def stub_execute(self, mode):
        return {"completed_count": 1, "applicable_count": 1, "failed_count": 0, "data": {}}
    runner._execute = stub_execute.__get__(runner, type(runner))

    runner.run(mode="resume")
    # 정상 save 후 marker 제거 확인
    marker = runner._cp_dir / runner._FORCE_CLEARED_MARKER
    assert not marker.exists(), "정상 save 후 .force_cleared marker는 제거되어야 함"
```

- [ ] **Step 2: 테스트 실행**

Run: `cd backend && .venv/bin/pytest tests/core/test_step_runner_resilience.py -v -k force_cleared`
Expected: 1 passed.

- [ ] **Step 3: 커밋**

```bash
git add backend/tests/core/test_step_runner_resilience.py
git commit -m "test(integrity): .force_cleared marker auto-rerun integration"
```

## Task 17: PR 1 — 전체 회귀 테스트 + Codex/Claude 듀얼 리뷰 준비

**Files:**
- 없음 (검증 단계)

- [ ] **Step 1: 전체 테스트 swept**

Run: `cd backend && .venv/bin/pytest tests/ --tb=short -q`
Expected: 1440(기존) + ~25(신규) = ~1465 passed, 0 failed.

- [ ] **Step 2: 새 컬럼/메서드 git status 확인**

Run: `git log --oneline 98e2645..HEAD`
Expected: ~14개 commit (Task 1–16). 메시지 형식 일관 (feat/fix/refactor/test).

- [ ] **Step 3: PID 0bb48ebf로 시나리오 B 실측**

Run:
```bash
.venv/bin/python -c "
from sqlalchemy import create_engine, text
eng = create_engine('postgresql://theroad:theroad_dev_2026@localhost:5432/theroad')
pid = '0bb48ebf'  # 기존 사고 PID (실제 UUID로 교체)
with eng.connect() as c:
    c.execute(text('SELECT step_id, status, recovery_count, last_recovery_reason FROM step_run WHERE project_id LIKE :pid'), {'pid': f'%{pid}%'})
"
```
(실제 PID UUID는 backend logs/error.log에서 grep)
Expected: recovery_count + last_recovery_reason 컬럼이 query 가능. 데이터는 모두 0/NULL.

- [ ] **Step 4: PR 1 commit 메시지 sanity check**

Run: `git log --oneline 98e2645..HEAD | head -20`
Expected: 명확한 commit history. 사고 시나리오 추적 가능.

- [ ] **Step 5: 사용자에게 PR 1 듀얼 리뷰 요청 (Codex + Claude)**

스킵 — subagent-driven-development의 review 단계에서 처리.

## Task 18: PR 1 commit + push

**Files:**
- 없음 (배포)

- [ ] **Step 1: 최종 차이 확인**

Run: `git log --oneline 98e2645..HEAD`

- [ ] **Step 2: 사용자 확인 후 push (사용자 명시 후에만)**

Run: `git push origin main` — **사용자 명시 승인 후에만**.

---

# PR 2 — Phase A 6 step verify_completion 본체 (Phase 2)

## Task 19: `RefImageGenStep.verify_completion`

**Files:**
- Modify: `backend/app/core/steps/image_steps.py:130-178` (RefImageGenStep)
- Test: `backend/tests/core/steps/test_verify_completion.py` (신규)

- [ ] **Step 1: 실패 테스트 작성**

`backend/tests/core/steps/test_verify_completion.py`:
```python
"""Phase A 6 step verify_completion 본체 테스트."""
import pytest
from pathlib import Path

from app.models.project import EntityCanon, EntityEpisodeLink, ImageAsset


def test_ref_image_gen_verify_clean(seed_episode_with_chars, tmp_path):
    """active character 3 + low_freq_skip 0 → ImageAsset 3 + 파일 3 → is_complete=True."""
    runner = seed_episode_with_chars(["C01", "C02", "C03"], skip=[])
    # ImageAsset 3 row + file 존재
    for sid in ["C01", "C02", "C03"]:
        png = tmp_path / f"{sid}.png"
        png.write_bytes(b"\x89PNG")
        runner._add_ref_asset(short_id=sid, file_path=str(png))

    report = runner.verify_completion()
    assert report.is_complete is True
    assert report.severity == "clean"
    assert report.metadata["found"] == 3


def test_ref_image_gen_verify_missing(seed_episode_with_chars, tmp_path):
    """active 3, ImageAsset 1 → is_complete=False, severity='missing'."""
    runner = seed_episode_with_chars(["C01", "C02", "C03"], skip=[])
    png = tmp_path / "C01.png"
    png.write_bytes(b"\x89PNG")
    runner._add_ref_asset(short_id="C01", file_path=str(png))

    report = runner.verify_completion()
    assert not report.is_complete
    assert report.severity == "partial"  # found=1 > 0
    assert report.metadata["expected"] == 3
    assert report.metadata["found"] == 1


def test_ref_image_gen_verify_low_freq_excluded(seed_episode_with_chars, tmp_path):
    """active 3, low_freq_skip [C03], ImageAsset 2 → expected=2, is_complete=True."""
    runner = seed_episode_with_chars(["C01", "C02", "C03"], skip=["C03"])
    for sid in ["C01", "C02"]:
        png = tmp_path / f"{sid}.png"
        png.write_bytes(b"\x89PNG")
        runner._add_ref_asset(short_id=sid, file_path=str(png))

    report = runner.verify_completion()
    assert report.is_complete is True


def test_ref_image_gen_verify_file_missing(seed_episode_with_chars, tmp_path):
    """ImageAsset row 3개 있는데 파일 1개 missing → is_complete=False."""
    runner = seed_episode_with_chars(["C01", "C02", "C03"], skip=[])
    for sid in ["C01", "C02"]:
        png = tmp_path / f"{sid}.png"
        png.write_bytes(b"\x89PNG")
        runner._add_ref_asset(short_id=sid, file_path=str(png))
    # C03는 row만 있고 파일 없음
    runner._add_ref_asset(short_id="C03", file_path=str(tmp_path / "missing.png"))

    report = runner.verify_completion()
    assert not report.is_complete
    assert report.metadata["found"] == 2  # 파일 존재만 카운트
    assert report.metadata["rows"] == 3
```

(`seed_episode_with_chars` fixture는 conftest에 작성 필요 — entity_canon + entity_episode_link + RefImageGenStep 인스턴스 생성. `_add_ref_asset` helper.)

- [ ] **Step 2: 테스트 실행 (실패 확인)**

Run: `cd backend && .venv/bin/pytest tests/core/steps/test_verify_completion.py -v -k ref_image_gen`
Expected: AttributeError — `RefImageGenStep.verify_completion` 미존재.

- [ ] **Step 3: `RefImageGenStep`에 verify_completion 추가**

`backend/app/core/steps/image_steps.py:130-178` (RefImageGenStep 클래스 안, _execute 다음에) 추가:
```python
    def verify_completion(self):
        """active character entity 수 == ImageAsset(reference, primary, character) row + 파일 존재.

        low_freq_skip된 character는 expected에서 제외.
        """
        from app.core.integrity_report import CompletionReport
        from app.core.low_freq_skip import load_low_freq_skip_ids
        from app.models.project import EntityCanon, EntityEpisodeLink, ImageAsset
        from pathlib import Path

        active_char_ids = [
            r[0] for r in self.db.query(EntityEpisodeLink.canon_id).join(
                EntityCanon, EntityCanon.id == EntityEpisodeLink.canon_id
            ).filter(
                EntityEpisodeLink.project_id == self.project_id,
                EntityEpisodeLink.episode_id == self.episode_id,
                EntityCanon.entity_type == "character",
            ).all()
        ]
        skipped = load_low_freq_skip_ids(self.project_id, self.episode_id)
        expected_ids = [cid for cid in active_char_ids if cid not in skipped]
        expected = len(expected_ids)

        if expected == 0:
            return CompletionReport(is_complete=True, missing=[], severity="clean",
                                    metadata={"expected": 0, "found": 0, "rows": 0})

        rows = self.db.query(ImageAsset).filter(
            ImageAsset.project_id == self.project_id,
            ImageAsset.episode_id == self.episode_id,
            ImageAsset.asset_type == "reference",
            ImageAsset.is_primary == 1,
            ImageAsset.entity_id.in_(expected_ids),
        ).all()
        found = sum(1 for r in rows if r.file_path and Path(r.file_path).exists())

        if found < expected:
            return CompletionReport(
                is_complete=False,
                missing=[f"{expected - found} character ref images missing (expected={expected}, found={found})"],
                severity="missing" if found == 0 else "partial",
                metadata={"expected": expected, "found": found, "rows": len(rows)},
            )
        return CompletionReport(is_complete=True, missing=[], severity="clean",
                                metadata={"expected": expected, "found": found, "rows": len(rows)})
```

- [ ] **Step 4: 테스트 통과 확인**

Run: `cd backend && .venv/bin/pytest tests/core/steps/test_verify_completion.py -v -k ref_image_gen`
Expected: 4 passed.

- [ ] **Step 5: 커밋**

```bash
git add backend/app/core/steps/image_steps.py backend/tests/core/steps/test_verify_completion.py
git commit -m "feat(integrity): RefImageGenStep.verify_completion (D1 Phase A.1)"
```

## Task 20: `CompositeImageGenStep.verify_completion`

**Files:**
- Modify: `backend/app/core/steps/image_steps.py:182+` (CompositeImageGenStep)
- Test: `backend/tests/core/steps/test_verify_completion.py`

- [ ] **Step 1: 실패 테스트 작성**

```python
def test_composite_verify_clean(seed_episode_with_combos, tmp_path):
    """char×outlook 4 pair (O00 1개 제외) → composite 3개 + 파일 → is_complete=True."""
    runner = seed_episode_with_combos(combos=[
        ("C01", "O01"), ("C01", "O02"), ("C02", "O01"), ("C02", "O00")
    ])
    for char, outlook in [("C01", "O01"), ("C01", "O02"), ("C02", "O01")]:
        png = tmp_path / f"{char}_{outlook}.png"
        png.write_bytes(b"\x89PNG")
        runner._add_composite_asset(char_sid=char, outlook_sid=outlook, file_path=str(png))

    report = runner.verify_completion()
    assert report.is_complete is True
    assert report.metadata["found"] == 3


def test_composite_verify_missing(seed_episode_with_combos, tmp_path):
    """4 pair 중 O00 제외 3, 1개만 생성 → is_complete=False, severity='partial'."""
    runner = seed_episode_with_combos(combos=[
        ("C01", "O01"), ("C01", "O02"), ("C02", "O01"), ("C02", "O00")
    ])
    png = tmp_path / "C01_O01.png"
    png.write_bytes(b"\x89PNG")
    runner._add_composite_asset(char_sid="C01", outlook_sid="O01", file_path=str(png))

    report = runner.verify_completion()
    assert not report.is_complete
    assert report.severity == "partial"
    assert report.metadata["expected"] == 3
    assert report.metadata["found"] == 1
```

- [ ] **Step 2: 실행 (실패 확인)**

Run: `cd backend && .venv/bin/pytest tests/core/steps/test_verify_completion.py -v -k composite_verify`
Expected: AttributeError.

- [ ] **Step 3: 메서드 추가**

`backend/app/core/steps/image_steps.py:CompositeImageGenStep` 클래스에 추가:
```python
    def verify_completion(self):
        """O00 제외 + low_freq skip 제외한 모든 (char, outlook) pair에 대해
        ImageAsset(prompt_used~'composite:{char}:{outlook}') row + 파일 존재.
        """
        from app.core.integrity_report import CompletionReport
        from app.core.low_freq_skip import load_low_freq_skip_ids
        from app.models.project import (
            EntityCanon, EntityEpisodeLink, CharacterOutlook, ImageAsset,
        )
        from pathlib import Path
        import re as _re

        # 에피소드의 character ids
        char_ids = {r[0] for r in self.db.query(EntityEpisodeLink.canon_id).join(
            EntityCanon, EntityCanon.id == EntityEpisodeLink.canon_id
        ).filter(
            EntityEpisodeLink.project_id == self.project_id,
            EntityEpisodeLink.episode_id == self.episode_id,
            EntityCanon.entity_type == "character",
        ).all()}

        # O00 ids 제외
        o00_ids = {e.id for e in self.db.query(EntityCanon).filter(
            EntityCanon.project_id == self.project_id,
            EntityCanon.short_id == "O00",
            EntityCanon.entity_type == "outlook",
        ).all()}

        skipped = load_low_freq_skip_ids(self.project_id, self.episode_id)

        all_combos = self.db.query(CharacterOutlook).filter(
            CharacterOutlook.project_id == self.project_id,
            CharacterOutlook.character_id.in_(char_ids),
        ).all() if char_ids else []
        expected_pairs = [
            (c.character_id, c.outlook_id) for c in all_combos
            if c.outlook_id not in o00_ids and c.character_id not in skipped
        ]
        expected = len(expected_pairs)
        if expected == 0:
            return CompletionReport(is_complete=True, missing=[], severity="clean",
                                    metadata={"expected": 0, "found": 0})

        composite_rows = self.db.query(ImageAsset).filter(
            ImageAsset.project_id == self.project_id,
            ImageAsset.asset_type == "reference",
            ImageAsset.prompt_used.like("%composite:%"),
        ).all()
        existing_keys = set()
        for r in composite_rows:
            m = _re.search(r'composite:([a-f0-9-]+):([a-f0-9-]+)', r.prompt_used or "")
            if m and r.file_path and Path(r.file_path).exists():
                existing_keys.add((m.group(1), m.group(2)))

        found = sum(1 for pair in expected_pairs if pair in existing_keys)
        if found < expected:
            return CompletionReport(
                is_complete=False,
                missing=[f"{expected - found} composites missing (expected={expected}, found={found})"],
                severity="missing" if found == 0 else "partial",
                metadata={"expected": expected, "found": found},
            )
        return CompletionReport(is_complete=True, missing=[], severity="clean",
                                metadata={"expected": expected, "found": found})
```

- [ ] **Step 4: 테스트 통과 + 회귀**

Run: `cd backend && .venv/bin/pytest tests/core/steps/test_verify_completion.py -v -k composite`
Expected: 2 passed.

- [ ] **Step 5: 커밋**

```bash
git add backend/app/core/steps/image_steps.py backend/tests/core/steps/test_verify_completion.py
git commit -m "feat(integrity): CompositeImageGenStep.verify_completion (D1 Phase A.2)"
```

## Task 21: `CharacterStateVariantStep.verify_completion`

**Files:**
- Modify: 적절한 step 파일 (`grep -rn "class CharacterStateVariantStep" backend/app/core/steps/`)
- Test: `backend/tests/core/steps/test_verify_completion.py`

- [ ] **Step 1: 클래스 위치 확인**

Run: `grep -rn "class CharacterStateVariantStep" backend/app/core/steps/`
Expected: 정확한 파일/라인 위치.

- [ ] **Step 2: 실패 테스트 작성**

```python
def test_state_variant_verify_clean(seed_episode_with_staging, tmp_path):
    """staging의 dead/severely_injured/unconscious 인물 (char, state) 쌍 수 ==
    ImageAsset(prompt_used~'state_variant:char:state') row + 파일 존재."""
    runner = seed_episode_with_staging(states=[
        ("C01", "dead"), ("C02", "severely_injured")
    ])
    for char, state in [("C01", "dead"), ("C02", "severely_injured")]:
        png = tmp_path / f"{char}_{state}.png"
        png.write_bytes(b"\x89PNG")
        runner._add_state_variant_asset(char_sid=char, state=state, file_path=str(png))

    report = runner.verify_completion()
    assert report.is_complete is True


def test_state_variant_verify_missing(seed_episode_with_staging, tmp_path):
    """state pair 2 → row 0 → severity='missing'."""
    runner = seed_episode_with_staging(states=[("C01", "dead")])
    report = runner.verify_completion()
    assert not report.is_complete
    assert report.severity == "missing"
    assert report.metadata["expected"] == 1
    assert report.metadata["found"] == 0
```

- [ ] **Step 3: 메서드 추가**

step 파일의 `CharacterStateVariantStep` 클래스에 추가:
```python
    def verify_completion(self):
        """shot_staging의 dead/severely_injured/unconscious 인물 (char_id, state) 쌍 수 ==
        ImageAsset(prompt_used~'state_variant:{char}:{state}') row + 파일 존재.
        """
        from app.core.integrity_report import CompletionReport
        from app.models.project import ImageAsset
        from pathlib import Path
        import re as _re
        from sqlalchemy import text as _sql_text

        # shot_staging에서 dead/severely_injured/unconscious 상태 인물 추출
        rows = self.db.execute(_sql_text("""
            SELECT DISTINCT ss.character_id, ss.state
            FROM shot_staging ss
            JOIN scene_still s ON s.id = ss.scene_still_id
            WHERE s.project_id = :pid AND s.episode_id = :eid
              AND ss.state IN ('dead', 'severely_injured', 'unconscious')
        """), {"pid": self.project_id, "eid": self.episode_id}).fetchall()
        expected_pairs = [(r[0], r[1]) for r in rows]
        expected = len(expected_pairs)
        if expected == 0:
            return CompletionReport(is_complete=True, missing=[], severity="clean",
                                    metadata={"expected": 0, "found": 0})

        sv_rows = self.db.query(ImageAsset).filter(
            ImageAsset.project_id == self.project_id,
            ImageAsset.asset_type == "reference",
            ImageAsset.prompt_used.like("%state_variant:%"),
        ).all()
        existing = set()
        for r in sv_rows:
            m = _re.search(r'state_variant:([a-f0-9-]+):(\w+)', r.prompt_used or "")
            if m and r.file_path and Path(r.file_path).exists():
                existing.add((m.group(1), m.group(2)))

        found = sum(1 for pair in expected_pairs if pair in existing)
        if found < expected:
            return CompletionReport(
                is_complete=False,
                missing=[f"{expected - found} state variants missing"],
                severity="missing" if found == 0 else "partial",
                metadata={"expected": expected, "found": found},
            )
        return CompletionReport(is_complete=True, missing=[], severity="clean",
                                metadata={"expected": expected, "found": found})
```

- [ ] **Step 4: 테스트 통과 확인**

Run: `cd backend && .venv/bin/pytest tests/core/steps/test_verify_completion.py -v -k state_variant`
Expected: 2 passed.

- [ ] **Step 5: 커밋**

```bash
git add <step file> backend/tests/core/steps/test_verify_completion.py
git commit -m "feat(integrity): CharacterStateVariantStep.verify_completion (D1 Phase A.3)"
```

## Task 22: `OutlookStandaloneStep.verify_completion`

**Files:**
- Modify: `backend/app/core/steps/outlook_steps.py` 또는 적절한 위치 (`grep -rn "class OutlookStandalone"`)
- Test: `backend/tests/core/steps/test_verify_completion.py`

- [ ] **Step 1: 클래스 위치 확인 + 실패 테스트 작성**

Run: `grep -rn "class.*Outlook.*Step\|outlook_standalone" backend/app/core/steps/`

테스트:
```python
def test_outlook_standalone_verify_clean(seed_episode_with_outlooks, tmp_path):
    """O00 제외 + low_freq skip 제외한 outlook entity 수 == ImageAsset row + 파일."""
    runner = seed_episode_with_outlooks(outlooks=["O01", "O02", "O00"], skip=[])
    for sid in ["O01", "O02"]:
        png = tmp_path / f"{sid}.png"
        png.write_bytes(b"\x89PNG")
        runner._add_outlook_asset(short_id=sid, file_path=str(png))

    report = runner.verify_completion()
    assert report.is_complete is True


def test_outlook_standalone_verify_missing(seed_episode_with_outlooks, tmp_path):
    runner = seed_episode_with_outlooks(outlooks=["O01", "O02"], skip=[])
    report = runner.verify_completion()
    assert not report.is_complete
    assert report.metadata["expected"] == 2
    assert report.metadata["found"] == 0
```

- [ ] **Step 2: 메서드 추가**

OutlookStandaloneStep 클래스에 추가:
```python
    def verify_completion(self):
        """O00 제외 + low_freq skip 제외한 outlook entity 수 ==
        ImageAsset(asset_type='reference', entity_type='outlook') row + 파일 존재.
        """
        from app.core.integrity_report import CompletionReport
        from app.core.low_freq_skip import load_low_freq_skip_ids
        from app.models.project import EntityCanon, EntityEpisodeLink, ImageAsset
        from pathlib import Path

        # 에피소드의 outlook entities (O00 제외)
        outlook_ids = [r[0] for r in self.db.query(EntityEpisodeLink.canon_id).join(
            EntityCanon, EntityCanon.id == EntityEpisodeLink.canon_id
        ).filter(
            EntityEpisodeLink.project_id == self.project_id,
            EntityEpisodeLink.episode_id == self.episode_id,
            EntityCanon.entity_type == "outlook",
            EntityCanon.short_id != "O00",
        ).all()]

        skipped = load_low_freq_skip_ids(self.project_id, self.episode_id)
        expected_ids = [oid for oid in outlook_ids if oid not in skipped]
        expected = len(expected_ids)
        if expected == 0:
            return CompletionReport(is_complete=True, missing=[], severity="clean",
                                    metadata={"expected": 0, "found": 0})

        rows = self.db.query(ImageAsset).filter(
            ImageAsset.project_id == self.project_id,
            ImageAsset.asset_type == "reference",
            ImageAsset.entity_id.in_(expected_ids),
        ).all()
        found = sum(1 for r in rows if r.file_path and Path(r.file_path).exists())
        if found < expected:
            return CompletionReport(
                is_complete=False,
                missing=[f"{expected - found} outlook standalones missing"],
                severity="missing" if found == 0 else "partial",
                metadata={"expected": expected, "found": found},
            )
        return CompletionReport(is_complete=True, missing=[], severity="clean",
                                metadata={"expected": expected, "found": found})
```

- [ ] **Step 3: 테스트 통과 + 커밋**

Run: `cd backend && .venv/bin/pytest tests/core/steps/test_verify_completion.py -v -k outlook_standalone`
Expected: 2 passed.

```bash
git add <step file> backend/tests/core/steps/test_verify_completion.py
git commit -m "feat(integrity): OutlookStandaloneStep.verify_completion (D1 Phase A.4)"
```

## Task 23: `BackgroundChainRenderStep.verify_completion`

> **Production-grounded 정정 (commit `dcea265`)**: 본 Task의 1차 초안은 production code와 5건 mismatch가 있었음. 아래는 실제 production 동작을 반영한 정정판이다.
>
> | 정정 | 1차 초안 (잘못된 가정) | Production (실제) |
> | --- | --- | --- |
> | (1) asset_type | `"reference"` | planner 분기 `"chain_bg"` / legacy 분기 `"background_chain_node"` |
> | (2) traceability key | `prompt_used like '%chain_bg:%'` regex | `variant_type` 컬럼 직접 매칭 (`in_(group_ids)` / `in_(node_ids)`) |
> | (3) planner 입력 | `background_chain_planning/manifest.json + data.groups` | `background_planner/manifest.json + data.chain_bg_order` (list) |
> | (4) docstring | "background_chain_planner data.groups" | planner=`background_planner.data.chain_bg_order` / legacy=`background_chain_planning.data.locations[loc_id].nodes[]` |
> | (5) 분기 누락 | 단일 path | planner cp 부재 또는 chain_bg_order 비었을 때 legacy fallback 명시 |

**Files:**
- Modify: `backend/app/core/steps/background_chain_render_step.py`
- Test: `backend/tests/core/steps/test_verify_completion.py`

- [ ] **Step 1: 실패 테스트 작성**

```python
def test_chain_bg_render_verify_clean_planner(seed_episode_with_bg_planner, tmp_path):
    """planner-driven path: chain_bg_order 의 group_id 수 ==
    ImageAsset(asset_type='chain_bg', variant_type=group_id) row + 파일."""
    runner = seed_episode_with_bg_planner(chain_bg_order=["g1", "g2", "g3"])
    for gid in ["g1", "g2", "g3"]:
        png = tmp_path / f"{gid}.png"
        png.write_bytes(b"\x89PNG")
        runner._add_chain_bg_asset(group_id=gid, file_path=str(png))  # asset_type='chain_bg'
    assert runner.verify_completion().is_complete


def test_chain_bg_render_verify_missing_planner(seed_episode_with_bg_planner, tmp_path):
    runner = seed_episode_with_bg_planner(chain_bg_order=["g1", "g2", "g3"])
    png = tmp_path / "g1.png"
    png.write_bytes(b"\x89PNG")
    runner._add_chain_bg_asset(group_id="g1", file_path=str(png))
    report = runner.verify_completion()
    assert not report.is_complete
    assert report.metadata["chain_bg_expected"] == 3
    assert report.metadata["chain_bg_found"] == 1
    assert report.metadata["path"] == "planner"


def test_chain_bg_render_verify_legacy_fallback(seed_episode_with_legacy_chain_plan, tmp_path):
    """planner cp 없으면 background_chain_planning.data.locations[loc].nodes[]
    노드 id 수만큼 asset_type='background_chain_node', variant_type=node_id 검증."""
    runner = seed_episode_with_legacy_chain_plan(
        locations={"loc1": {"nodes": [{"id": "n1"}, {"id": "n2"}]}},
    )
    for nid in ["n1", "n2"]:
        png = tmp_path / f"{nid}.png"
        png.write_bytes(b"\x89PNG")
        runner._add_chain_bg_node_asset(node_id=nid, file_path=str(png))
    report = runner.verify_completion()
    assert report.is_complete
    assert report.metadata["path"] == "legacy"
```

- [ ] **Step 2: 메서드 추가**

`BackgroundChainRenderStep` 클래스에 추가:
```python
    def verify_completion(self):
        """chain_bg / background_chain_node 산출물 무결성 검증 (planner+legacy dual path).

        - planner-driven: ``background_planner.data.chain_bg_order`` (list[str]) 가
          비어있지 않으면 ImageAsset(asset_type='chain_bg', variant_type=group_id) 매칭.
        - legacy: planner cp 없거나 chain_bg_order 비어있으면
          ``background_chain_planning.data.locations[loc_id].nodes[]`` 노드 id 를 기대치로
          사용해 ImageAsset(asset_type='background_chain_node', variant_type=node_id) 매칭.
        - traceability는 prompt_used regex가 아니라 variant_type 컬럼 직접 매칭.
        - cleanup_artifacts override 안 함 → default noop (사용자 caveat: 부분 재생성 보호).
        """
        from pathlib import Path
        from app.core.integrity_report import CompletionReport
        from app.models.project import ImageAsset

        # ── Planner-driven path (Phase 5+) ──
        chain_order = self._load_planner_chain_order()
        if chain_order:
            expected_groups = list(chain_order)
            expected = len(expected_groups)
            rows = self.db.query(ImageAsset).filter(
                ImageAsset.project_id == self.project_id,
                ImageAsset.episode_id == self.episode_id,
                ImageAsset.asset_type == "chain_bg",
                ImageAsset.variant_type.in_(expected_groups),
            ).all()
            groups_with_file = {
                r.variant_type for r in rows
                if r.file_path and Path(r.file_path).exists()
            }
            found_set = groups_with_file & set(expected_groups)
            found = len(found_set)
            missing_groups = sorted(set(expected_groups) - found_set)
            severity = (
                "clean" if found == expected
                else "missing" if found == 0
                else "partial"
            )
            return CompletionReport(
                is_complete=(severity == "clean"),
                missing=(
                    [f"{len(missing_groups)} chain_bg groups missing "
                     f"(expected={expected}, found={found}): {missing_groups}"]
                    if missing_groups else []
                ),
                severity=severity,
                metadata={"chain_bg_expected": expected, "chain_bg_found": found,
                          "path": "planner"},
            )

        # ── Legacy path (Phase 4 또는 planner cp 없음) ──
        plan_cp = self._load_prev_checkpoint("background_chain_planning")
        if not plan_cp:
            return CompletionReport(
                is_complete=True, missing=[], severity="clean",
                metadata={"chain_bg_expected": 0, "chain_bg_found": 0, "path": "none"},
            )
        locations = (plan_cp.get("data") or {}).get("locations") or {}
        node_ids = []
        for loc in locations.values():
            for node in loc.get("nodes") or []:
                nid = node.get("id") or node.get("node_id")
                if nid:
                    node_ids.append(nid)
        if not node_ids:
            return CompletionReport(
                is_complete=True, missing=[], severity="clean",
                metadata={"chain_bg_expected": 0, "chain_bg_found": 0, "path": "legacy"},
            )

        expected = len(node_ids)
        rows = self.db.query(ImageAsset).filter(
            ImageAsset.project_id == self.project_id,
            ImageAsset.episode_id == self.episode_id,
            ImageAsset.asset_type == "background_chain_node",
            ImageAsset.variant_type.in_(node_ids),
        ).all()
        nodes_with_file = {
            r.variant_type for r in rows
            if r.file_path and Path(r.file_path).exists()
        }
        found_set = nodes_with_file & set(node_ids)
        found = len(found_set)
        missing_nodes = sorted(set(node_ids) - found_set)
        severity = (
            "clean" if found == expected
            else "missing" if found == 0
            else "partial"
        )
        return CompletionReport(
            is_complete=(severity == "clean"),
            missing=(
                [f"{len(missing_nodes)} chain_bg_node images missing "
                 f"(expected={expected}, found={found}): {missing_nodes}"]
                if missing_nodes else []
            ),
            severity=severity,
            metadata={"chain_bg_expected": expected, "chain_bg_found": found,
                      "path": "legacy"},
        )
```

- [ ] **Step 3: 테스트 통과 + 커밋**

Run: `cd backend && .venv/bin/pytest tests/core/steps/test_verify_completion.py -v -k chain_bg_render`
Expected: 3 passed (planner clean / planner missing / legacy fallback).

```bash
git add backend/app/core/steps/background_chain_render_step.py backend/tests/core/steps/test_verify_completion.py
git commit -m "feat(integrity): BackgroundChainRenderStep.verify_completion (Task 23, planner+legacy)"
```

## Task 24: `FloorPlanRenderStep.verify_completion`

> **Production-grounded 정정 (commit `763944f`)**: 본 Task의 1차 초안은 production code와 6건 mismatch가 있었음. 아래는 실제 production 동작을 반영한 정정판이다.
>
> | 정정 | 1차 초안 (잘못된 가정) | Production (실제) |
> | --- | --- | --- |
> | (6) asset_type | `"reference"` | `"floor_plan"` |
> | (7) traceability key | `prompt_used like '%floor_plan:%'` regex | `variant_type` 컬럼 직접 매칭 (`in_(renderable_fp_ids)`) |
> | (8) expected source | `master_plan + data.groups[].needs_floor_plan` | `background_master_plan.data.plans[gid].plan.floor_plans[].fp_id` (production return shape에 `needs_floor_plan` 필드는 존재하지 않음) |
> | (9) background_mode gate | 없음 | `settings.background_mode ∈ {"on","floor_plan_anchored"}` 일 때만 검증, 그 외 expected=0 clean |
> | (10) renderable filter | 없음 | `floor_plan_prompt.data.floor_plans[fid].status=='ok'` 인 fp만 expected (실패 prompt fp는 production이 애초에 렌더 안 함) |
> | (11) test seed | `needs_fp_groups=[...]` parameter | `master_plan_fps=[(loc, fp_id), ...]` + `prompt_status` dict (status=ok 만 expected) |

**Files:**
- Modify: `backend/app/core/steps/floor_plan_render_step.py`
- Test: `backend/tests/core/steps/test_verify_completion.py`

- [ ] **Step 1: 실패 테스트 작성**

```python
def test_floor_plan_render_verify_clean(seed_episode_with_fp_planner, tmp_path):
    """master_plan.data.plans[gid].plan.floor_plans[].fp_id (status=ok) 수 ==
    ImageAsset(asset_type='floor_plan', variant_type=fp_id) row + 파일."""
    runner = seed_episode_with_fp_planner(
        master_plan_fps=[("loc1", "FP01"), ("loc2", "FP02")],
        prompt_status={"FP01": "ok", "FP02": "ok"},
    )
    for fid in ["FP01", "FP02"]:
        png = tmp_path / f"{fid}.png"
        png.write_bytes(b"\x89PNG")
        runner._add_floor_plan_asset(fp_id=fid, file_path=str(png))  # asset_type='floor_plan'
    assert runner.verify_completion().is_complete


def test_floor_plan_render_verify_skip_when_no_needs(seed_episode_with_fp_planner):
    """master_plan에 floor_plans 없음 → expected=0 → clean (skip)."""
    runner = seed_episode_with_fp_planner(master_plan_fps=[], prompt_status={})
    assert runner.verify_completion().is_complete


def test_floor_plan_render_verify_renderable_filter(seed_episode_with_fp_planner, tmp_path):
    """floor_plan_prompt.status=='ok' 인 fp만 expected. status='fail' fp는 제외."""
    runner = seed_episode_with_fp_planner(
        master_plan_fps=[("loc1", "FP01"), ("loc2", "FP02")],
        prompt_status={"FP01": "ok", "FP02": "fail"},
    )
    png = tmp_path / "FP01.png"
    png.write_bytes(b"\x89PNG")
    runner._add_floor_plan_asset(fp_id="FP01", file_path=str(png))
    report = runner.verify_completion()
    assert report.is_complete
    assert report.metadata["floor_plan_expected"] == 1  # FP02 제외
    assert report.metadata["floor_plan_found"] == 1


def test_floor_plan_render_verify_skip_background_mode_off(seed_episode_with_fp_planner):
    """settings.background_mode='off' → step 자체가 skip되므로 expected=0 clean."""
    runner = seed_episode_with_fp_planner(
        master_plan_fps=[("loc1", "FP01")], prompt_status={"FP01": "ok"},
        background_mode="off",
    )
    report = runner.verify_completion()
    assert report.is_complete
    assert report.metadata.get("skipped_reason") == "background_mode_off"
```

- [ ] **Step 2: 메서드 추가**

`FloorPlanRenderStep` 클래스에 추가:
```python
    def verify_completion(self):
        """Phase 7 Step 4 산출물 무결성 검증.

        gate: settings.background_mode ∈ {"on","floor_plan_anchored"} 일 때만 검증.
        그 외엔 step 자체가 skip되므로 expected=0 clean.

        expected source: ``background_master_plan.data.plans[gid].plan.floor_plans[].fp_id``
        (production return shape — plan §4.4 1차 초안의 ``data.groups[].needs_floor_plan``
        필드는 production code에 존재하지 않음).

        renderable filter: ``floor_plan_prompt.data.floor_plans[fid].status=='ok'`` 인
        fp만 expected에 포함. prompt 단계 실패 fp는 production이 애초에 렌더하지 않으므로
        verify에서도 제외해야 false-missing 방지.

        match key: ImageAsset(asset_type='floor_plan', variant_type=fp_id) + 파일 stat —
        production ``_register_image_assets`` UPSERT와 동일.

        cleanup_artifacts override 안 함 → default noop (사용자 caveat: 부분 재생성 보호).
        """
        from pathlib import Path
        from app.core.config import settings
        from app.core.integrity_report import CompletionReport
        from app.models.project import ImageAsset

        # background_mode gate
        if settings.background_mode not in {"on", "floor_plan_anchored"}:
            return CompletionReport(
                is_complete=True, missing=[], severity="clean",
                metadata={"floor_plan_expected": 0, "floor_plan_found": 0,
                          "skipped_reason": "background_mode_off"},
            )

        # expected: master_plan.data.plans[gid].plan.floor_plans[].fp_id (status=ok 만)
        plans_cp = self._load_prev_checkpoint("background_master_plan")
        plans_map = ((plans_cp or {}).get("data", {}) or {}).get("plans", {}) or {}
        expected_fp_ids: set[str] = set()
        for entry in plans_map.values():
            if entry.get("status") != "ok":
                continue
            for fp in (entry.get("plan") or {}).get("floor_plans") or []:
                fid = fp.get("fp_id")
                if fid:
                    expected_fp_ids.add(fid)

        # renderable filter: floor_plan_prompt.data.floor_plans[fid].status=='ok'
        prompt_cp = self._load_prev_checkpoint("floor_plan_prompt")
        prompt_fps = (
            ((prompt_cp or {}).get("data", {}) or {}).get("floor_plans", {}) or {}
        )
        renderable_fp_ids = {
            fid for fid in expected_fp_ids
            if (prompt_fps.get(fid) or {}).get("status") == "ok"
        }
        expected = len(renderable_fp_ids)

        if expected == 0:
            return CompletionReport(
                is_complete=True, missing=[], severity="clean",
                metadata={"floor_plan_expected": 0, "floor_plan_found": 0},
            )

        rows = self.db.query(ImageAsset).filter(
            ImageAsset.project_id == self.project_id,
            ImageAsset.episode_id == self.episode_id,
            ImageAsset.asset_type == "floor_plan",
            ImageAsset.variant_type.in_(renderable_fp_ids),
        ).all()
        fp_ids_with_file = {
            r.variant_type for r in rows
            if r.file_path and Path(r.file_path).exists()
        }
        found = len(fp_ids_with_file)
        missing = sorted(renderable_fp_ids - fp_ids_with_file)

        if found < expected:
            return CompletionReport(
                is_complete=False, missing=missing,
                severity="missing" if found == 0 else "partial",
                metadata={"floor_plan_expected": expected,
                          "floor_plan_found": found, "rows": len(rows)},
            )
        return CompletionReport(
            is_complete=True, missing=[], severity="clean",
            metadata={"floor_plan_expected": expected,
                      "floor_plan_found": found, "rows": len(rows)},
        )
```

- [ ] **Step 3: 테스트 통과 + 커밋**

Run: `cd backend && .venv/bin/pytest tests/core/steps/test_verify_completion.py -v -k floor_plan`
Expected: 4 passed (clean / skip-when-no-needs / renderable-filter / background_mode-off).

```bash
git add backend/app/core/steps/floor_plan_render_step.py backend/tests/core/steps/test_verify_completion.py
git commit -m "feat(integrity): FloorPlanRenderStep.verify_completion (Task 24, background_mode gate + renderable filter)"
```

## Task 25: 부분 재생성 회귀 가드 테스트 (사용자 caveat — Phase A 6 step)

**Files:**
- Create: `backend/tests/core/steps/test_partial_regen_protection.py`

- [ ] **Step 1: 부분 재생성 보호 테스트 작성**

```python
"""Phase A 6 step의 cleanup_artifacts default noop 보호 — 부분 재생성 endpoint 안전.

사용자 caveat: 이미지는 거의 대부분 재생성 가능해야 한다.
검증: cleanup_artifacts 기본 동작이 noop이라 entity Y/Z 데이터가 보존되는지.
"""
import pytest
from app.core.integrity_report import CleanupReport


@pytest.mark.parametrize("step_id", [
    "ref_image_gen", "composite_image_gen", "character_state_variant",
    "outlook_standalone", "chain_bg_render", "floor_plan_render",
])
def test_cleanup_artifacts_is_noop_for_image_steps(make_step_runner, step_id):
    """이미지 step은 모두 default noop 유지 — 부분 재생성 endpoint 보호."""
    runner = make_step_runner(step_id)
    report = runner.cleanup_artifacts()
    assert isinstance(report, CleanupReport)
    assert report.deleted_db_rows == 0, f"{step_id} cleanup 동작 — 부분 재생성 위험"
    assert report.deleted_files == 0


def test_full_force_preserves_other_entity_data(seed_episode_with_chars, tmp_path):
    """전체 force 시 cleanup_artifacts noop이라 다른 entity 데이터 보존."""
    runner = seed_episode_with_chars(["C01", "C02", "C03"], skip=[])
    # 3 entity 모두 ImageAsset row + 파일 있음
    for sid in ["C01", "C02", "C03"]:
        png = tmp_path / f"{sid}.png"
        png.write_bytes(b"\x89PNG")
        runner._add_ref_asset(short_id=sid, file_path=str(png))

    # cleanup_artifacts 호출 (force path 시뮬레이션)
    report = runner.cleanup_artifacts()
    assert report.deleted_db_rows == 0, "force 시도 default noop이라 row 보존"
    assert report.deleted_files == 0

    # 모든 row + 파일 보존됐는지 확인
    from app.models.project import ImageAsset
    rows = runner.db.query(ImageAsset).filter(
        ImageAsset.project_id == runner.project_id,
    ).all()
    assert len(rows) == 3, "cleanup noop이라 모든 row 보존"
```

- [ ] **Step 2: 테스트 실행**

Run: `cd backend && .venv/bin/pytest tests/core/steps/test_partial_regen_protection.py -v`
Expected: 7 passed (6 parametrize + 1 standalone).

- [ ] **Step 3: 커밋**

```bash
git add backend/tests/core/steps/test_partial_regen_protection.py
git commit -m "test(integrity): 부분 재생성 보호 회귀 가드 (사용자 caveat)"
```

## Task 26: PR 2 — 회귀 + 듀얼 리뷰 준비

- [ ] **Step 1: 전체 swept**

Run: `cd backend && .venv/bin/pytest tests/ --tb=short -q`
Expected: 1465 + ~25 = ~1490 passed.

- [ ] **Step 2: PR 2 commit history 정리**

Run: `git log --oneline 98e2645..HEAD | head -25`
Expected: PR 1 + PR 2 commit. 메시지 일관.

- [ ] **Step 3: 사용자 push 승인 후 push**

---

# PR 3 — D3 Runtime fail-soft + D4 Gate 강화 (Phase 3 + 4)

## Task 27: `resolve_refs_for_prompt`이 missing_refs 반환 (D3)

**Files:**
- Modify: `backend/app/services/scene_reference_service.py:254-391`
- Test: `backend/tests/services/test_scene_reference_failsoft.py`

- [ ] **Step 1: 실패 테스트 작성**

```python
"""scene_reference_service runtime fail-soft 테스트 (D3)."""
import pytest


def test_resolve_refs_returns_missing_when_composite_absent(make_ref_service, seed_image_map):
    """C01O02 prompt 등장 + composite 없음 + face only → missing_refs에 사유."""
    svc = make_ref_service()
    image_map = seed_image_map(char_only=["C01"], composites=[], outlooks=[])
    visible_entities = [
        {"id": "uuid_c01", "short_id": "C01", "name": "주인공", "entity_type": "character"},
        {"id": "uuid_o02", "short_id": "O02", "name": "한복", "entity_type": "outlook"},
    ]

    labeled, missing = svc.resolve_refs_for_prompt(
        t2i_prompt="A C01O02 walking through a market.",
        scene_ref_image_map=image_map,
        visible_entities=visible_entities,
    )
    assert any("composite:uuid_c01:uuid_o02" in m or "C01O02" in m for m in missing), \
        f"composite 누락 시 missing_refs 적재 필요: {missing}"


def test_resolve_refs_returns_empty_missing_when_all_present(make_ref_service, seed_image_map):
    """모든 ref 정상 → missing_refs == []."""
    svc = make_ref_service()
    image_map = seed_image_map(
        char_only=["C01"],
        composites=[("uuid_c01", "uuid_o02")],
        outlooks=["O02"],
    )
    visible_entities = [
        {"id": "uuid_c01", "short_id": "C01", "name": "주인공", "entity_type": "character"},
        {"id": "uuid_o02", "short_id": "O02", "name": "한복", "entity_type": "outlook"},
    ]
    labeled, missing = svc.resolve_refs_for_prompt(
        t2i_prompt="A C01O02 walking.",
        scene_ref_image_map=image_map,
        visible_entities=visible_entities,
    )
    assert missing == [], f"모든 ref 있는데 missing 적재: {missing}"
```

- [ ] **Step 2: 실행 (실패 확인)**

Run: `cd backend && .venv/bin/pytest tests/services/test_scene_reference_failsoft.py -v`
Expected: TypeError — return signature 불일치.

- [ ] **Step 3: `resolve_refs_for_prompt` 시그니처 변경**

`backend/app/services/scene_reference_service.py:254` 메서드 본체 변경:

기존 return 부분(`return labeled_refs`)을 `return labeled_refs, missing_refs`로 변경하고, 누락 감지 로직 추가:
```python
    def resolve_refs_for_prompt(
        self,
        t2i_prompt,
        scene_ref_image_map,
        visible_entities,
        entity_lookup=None,
        state_variant_sids=None,
    ):
        """T2I 프롬프트 ref 매칭. composite/state_variant/face 누락 시 missing_refs 반환.

        Returns: (labeled_refs, missing_refs)
        - labeled_refs: list[(label, path)] — Gemini에 첨부할 ref 이미지
        - missing_refs: list[str] — 누락된 ref 사유 (caller가 shot skip 결정)
        """
        labeled_refs = []
        missing_refs: list[str] = []
        # ... (기존 로직 그대로) ...

        # 1) C##O## 패턴 처리
        for match in _re.finditer(r'\b([CL]\d+)([O]\d+)\b', t2i_prompt):
            char_sid = match.group(1)
            outlook_sid = match.group(2)
            # ... (기존 char_id, outlook_id resolve 로직) ...
            composite_key = f"composite:{char_id}:{outlook_id}" if outlook_id else None
            if composite_key and composite_key in scene_ref_image_map and composite_key not in _used_ref_ids:
                labeled_refs.append((f"character {sid} in outfit", scene_ref_image_map[composite_key]))
                _used_ref_ids.add(composite_key)
                # ... (기존)
                continue
            # composite 없으면 face만 — 단, outlook이 prompt에 있는데 composite 없으면 missing 적재
            if outlook_sid and (not composite_key or composite_key not in scene_ref_image_map):
                missing_refs.append(f"composite missing for {char_sid}{outlook_sid} (composite_key={composite_key})")
            if char_id in scene_ref_image_map and char_id not in _used_ref_ids:
                labeled_refs.append((f"character {char_sid} identity", scene_ref_image_map[char_id]))
                _used_ref_ids.add(char_id)
            elif char_id and char_id not in scene_ref_image_map:
                missing_refs.append(f"face ref missing for {char_sid}")

        # 2) [[name]+[outlook]] 레거시 패턴 — 동일 missing 감지
        for match in _re.finditer(r'\[\[([^\]]+)\]\+\[([^\]]+)\]\]', t2i_prompt):
            char_name, outlook_name = match.group(1), match.group(2)
            char_id, outlook_id = _find_char_outlook_ids(char_name, outlook_name)
            if not char_id:
                missing_refs.append(f"unknown character: {char_name}")
                continue
            if outlook_name == "미지정":
                # face only OK
                ...
                continue
            composite_key = f"composite:{char_id}:{outlook_id}" if outlook_id else None
            if composite_key and composite_key in scene_ref_image_map:
                # 정상 — labeled에 추가
                ...
                continue
            # composite 누락 — face fallback + missing 적재
            missing_refs.append(f"composite missing for [[{char_name}]+[{outlook_name}]]")
            if char_id in scene_ref_image_map:
                ...

        # 3) prop 처리 (변경 없음)
        # ...

        return labeled_refs, missing_refs
```

(원본 코드의 흐름 보존하면서 missing_refs append 부분만 추가. 정확한 indentation/변수명은 원본과 일치시킬 것.)

- [ ] **Step 4: 모든 호출 site 업데이트**

Run: `grep -rn "resolve_refs_for_prompt" backend/app/`
Expected: 호출 site 1-3개 위치 출력. 각각 unpacking 변경:
```python
labeled_refs, missing_refs = ref_service.resolve_refs_for_prompt(...)
```

- [ ] **Step 5: 테스트 통과 + 회귀 확인**

Run: `cd backend && .venv/bin/pytest tests/services/test_scene_reference_failsoft.py -v`
Expected: 2 passed.

Run: `cd backend && .venv/bin/pytest tests/ --tb=short`
Expected: 회귀 0.

- [ ] **Step 6: 커밋**

```bash
git add backend/app/services/scene_reference_service.py backend/tests/services/test_scene_reference_failsoft.py
git commit -m "fix(integrity): D3 — resolve_refs_for_prompt returns missing_refs"
```

## Task 28: scene_image_pipeline shot skip on missing_refs

**Files:**
- Modify: `backend/app/core/steps/scene_steps.py` 또는 scene_image_pipeline 위치
- Test: 위 test_scene_reference_failsoft.py 확장

- [ ] **Step 1: 클래스 위치 확인**

Run: `grep -rn "class SceneImagePipelineStep\|class SceneImage" backend/app/core/steps/`

- [ ] **Step 2: 실패 테스트 추가**

```python
def test_scene_image_pipeline_skips_shot_on_missing_refs(seed_scene_pipeline_with_missing):
    """missing_refs 있으면 해당 shot skip + failed_count++ + scene_still 안 만듦."""
    runner = seed_scene_pipeline_with_missing(shots=[
        {"id": "S01_shot1", "missing": ["composite for C01O02"]},
        {"id": "S01_shot2", "missing": []},
    ])
    result = runner._execute(mode="resume")
    assert result["failed_count"] == 1
    # 정상 shot 1개만 scene_still에 들어감
    from app.models.project import SceneStill
    rows = runner.db.query(SceneStill).filter(
        SceneStill.project_id == runner.project_id,
    ).all()
    assert len(rows) == 1
```

- [ ] **Step 3: scene_image_pipeline 수정**

shot 루프에서 `resolve_refs_for_prompt` unpacking + missing 시 skip:
```python
for shot in shots:
    labeled_refs, missing_refs = ref_service.resolve_refs_for_prompt(
        shot.t2i_prompt, scene_ref_image_map, visible_entities, ...
    )
    if missing_refs:
        logger.warning("[RUNTIME-FAILSOFT] shot=%s skipped — missing refs: %s",
                       shot.id, missing_refs)
        failed_count += 1
        continue  # scene_still row 생성 안 함
    # 정상 진행 (Gemini 호출, scene_still 생성 등)
    ...
```

- [ ] **Step 4: 테스트 통과 확인**

Run: `cd backend && .venv/bin/pytest tests/services/test_scene_reference_failsoft.py -v -k pipeline_skips`
Expected: 1 passed.

- [ ] **Step 5: 커밋**

```bash
git add <scene_pipeline file> backend/tests/services/test_scene_reference_failsoft.py
git commit -m "fix(integrity): D3 — scene_image_pipeline skips shot on missing_refs"
```

## Task 29: `SceneImagePipelineStep.verify_completion`

**Files:**
- Modify: scene_image_pipeline step 파일
- Test: `backend/tests/core/steps/test_verify_completion.py` 확장

- [ ] **Step 1: 실패 테스트 작성**

```python
def test_scene_image_pipeline_verify_clean(seed_scene_pipeline_episode, tmp_path):
    """applicable shot 5 → scene_still 5 row + 각 shot PNG 존재 → is_complete=True."""
    runner = seed_scene_pipeline_episode(shot_count=5, all_have_png=True)
    assert runner.verify_completion().is_complete


def test_scene_image_pipeline_verify_missing_png(seed_scene_pipeline_episode):
    """applicable 5, scene_still 5 row 있는데 PNG 1개 없음 → is_complete=False."""
    runner = seed_scene_pipeline_episode(shot_count=5, all_have_png=False, missing_count=1)
    report = runner.verify_completion()
    assert not report.is_complete
    assert report.metadata["expected"] == 5
    assert report.metadata["found"] == 4
```

- [ ] **Step 2: 메서드 추가**

```python
    def verify_completion(self):
        """applicable shot 수 == scene_still rows + 각 shot의 PNG 파일 존재."""
        from app.core.integrity_report import CompletionReport
        from app.models.project import SceneStill, ImageAsset
        from pathlib import Path

        # applicable shot 수 (is_selected=True)
        expected_rows = self.db.query(SceneStill).filter(
            SceneStill.project_id == self.project_id,
            SceneStill.episode_id == self.episode_id,
            SceneStill.is_selected.is_(True),
        ).all()
        expected = len(expected_rows)
        if expected == 0:
            return CompletionReport(is_complete=True, missing=[], severity="clean",
                                    metadata={"expected": 0, "found": 0})

        # 각 still에 대응하는 ImageAsset (asset_type='scene', is_primary=1) + 파일 존재
        asset_map = {}
        assets = self.db.query(ImageAsset).filter(
            ImageAsset.project_id == self.project_id,
            ImageAsset.episode_id == self.episode_id,
            ImageAsset.asset_type == "scene",
            ImageAsset.is_primary == 1,
        ).all()
        for a in assets:
            if a.still_id and a.file_path and Path(a.file_path).exists():
                asset_map[a.still_id] = a.file_path

        found = sum(1 for s in expected_rows if s.id in asset_map)
        if found < expected:
            return CompletionReport(
                is_complete=False,
                missing=[f"{expected - found} scene images missing (expected={expected}, found={found})"],
                severity="missing" if found == 0 else "partial",
                metadata={"expected": expected, "found": found},
            )
        return CompletionReport(is_complete=True, missing=[], severity="clean",
                                metadata={"expected": expected, "found": found})
```

- [ ] **Step 3: 테스트 통과 + 커밋**

Run: `cd backend && .venv/bin/pytest tests/core/steps/test_verify_completion.py -v -k scene_image_pipeline`
Expected: 2 passed.

```bash
git add <scene_pipeline file> backend/tests/core/steps/test_verify_completion.py
git commit -m "feat(integrity): SceneImagePipelineStep.verify_completion (D1 Phase 3.1)"
```

## Task 30: pipeline_gate D4 강화 — outlook_standalone 검증

**Files:**
- Modify: `backend/app/core/pipeline_gate.py:51-197` (check_scene_images_ready)
- Test: `backend/tests/test_pipeline_gate_enhanced.py` (신규)

- [ ] **Step 1: 실패 테스트 작성**

```python
"""pipeline_gate D4 강화 테스트 — 4종 자산 검증."""
import pytest
from app.core.errors import AppError


def test_check_scene_images_ready_blocks_on_outlook_missing(seed_episode_outlook_missing):
    """outlook 1개인데 standalone ref 0 → gate.incomplete_outlook_standalone."""
    db, project_id, episode_id = seed_episode_outlook_missing(outlooks=["O01"], standalones=[])
    from app.core.pipeline_gate import check_scene_images_ready
    with pytest.raises(AppError) as exc:
        check_scene_images_ready(db, project_id, episode_id)
    assert exc.value.code == "gate.incomplete_outlook_standalone"
```

- [ ] **Step 2: 메서드 추가**

`backend/app/core/pipeline_gate.py:check_scene_images_ready` 마지막 `return {...}` 직전에 추가:
```python
    # 4) outlook standalone 검증 (D4)
    from app.core.low_freq_skip import load_low_freq_skip_ids
    skipped_ids_low = load_low_freq_skip_ids(project_id, episode_id)
    outlook_entity_rows = db.query(EntityCanon).filter(
        EntityCanon.id.in_(linked_ids),
        EntityCanon.entity_type == "outlook",
        EntityCanon.short_id != "O00",
    ).all()
    outlook_expected_ids = [e.id for e in outlook_entity_rows if e.id not in skipped_ids_low]
    if outlook_expected_ids:
        outlook_assets = db.query(ImageAsset).filter(
            ImageAsset.project_id == project_id,
            ImageAsset.episode_id == episode_id,
            ImageAsset.asset_type == "reference",
            ImageAsset.entity_id.in_(outlook_expected_ids),
        ).all()
        outlook_existing = set()
        for oa in outlook_assets:
            from pathlib import Path
            if oa.file_path and Path(oa.file_path).exists():
                outlook_existing.add(oa.entity_id)
        missing_outlook = [oid for oid in outlook_expected_ids if oid not in outlook_existing]
        if missing_outlook:
            raise AppError(
                code="gate.incomplete_outlook_standalone",
                message=f"아웃룩 단독 이미지 미완성: {len(missing_outlook)}개 누락. '참조 이미지 생성'을 먼저 완료하세요.",
                status_code=400,
            )
```

- [ ] **Step 3: 테스트 통과 + 커밋**

Run: `cd backend && .venv/bin/pytest tests/test_pipeline_gate_enhanced.py -v -k outlook`
Expected: 1 passed.

```bash
git add backend/app/core/pipeline_gate.py backend/tests/test_pipeline_gate_enhanced.py
git commit -m "feat(integrity): D4 — outlook_standalone gate"
```

## Task 31: D4 추가 — character_state_variant gate

**Files:**
- Modify: `backend/app/core/pipeline_gate.py`
- Modify: `backend/tests/test_pipeline_gate_enhanced.py`

- [ ] **Step 1: 실패 테스트 + 메서드 추가**

테스트:
```python
def test_check_scene_images_ready_blocks_on_state_variant_missing(seed_episode_dead_char):
    """staging의 dead 인물 1 + state_variant ref 0 → gate.incomplete_state_variant."""
    db, project_id, episode_id = seed_episode_dead_char(states=[("C01", "dead")], variants=[])
    from app.core.pipeline_gate import check_scene_images_ready
    with pytest.raises(AppError) as exc:
        check_scene_images_ready(db, project_id, episode_id)
    assert exc.value.code == "gate.incomplete_state_variant"
```

`pipeline_gate.py`에 추가 (outlook 검증 다음):
```python
    # 5) state_variant 검증 (D4)
    state_pairs_rows = db.execute(text("""
        SELECT DISTINCT ss.character_id, ss.state
        FROM shot_staging ss
        JOIN scene_still s ON s.id = ss.scene_still_id
        WHERE s.project_id = :pid AND s.episode_id = :eid
          AND ss.state IN ('dead', 'severely_injured', 'unconscious')
    """), {"pid": project_id, "eid": episode_id}).fetchall()
    expected_pairs = [(r[0], r[1]) for r in state_pairs_rows]
    if expected_pairs:
        sv_assets = db.query(ImageAsset).filter(
            ImageAsset.project_id == project_id,
            ImageAsset.asset_type == "reference",
            ImageAsset.prompt_used.like("%state_variant:%"),
        ).all()
        existing = set()
        for r in sv_assets:
            m = re.search(r'state_variant:([a-f0-9-]+):(\w+)', r.prompt_used or "")
            if m and r.file_path and Path(r.file_path).exists():
                existing.add((m.group(1), m.group(2)))
        missing = [p for p in expected_pairs if p not in existing]
        if missing:
            raise AppError(
                code="gate.incomplete_state_variant",
                message=f"인물 상태 변형 이미지 미완성: {len(missing)}개 누락. '참조 이미지 생성'을 완료하세요.",
                status_code=400,
            )
```

- [ ] **Step 2: 테스트 + 커밋**

Run: `cd backend && .venv/bin/pytest tests/test_pipeline_gate_enhanced.py -v -k state_variant`
Expected: 1 passed.

```bash
git add backend/app/core/pipeline_gate.py backend/tests/test_pipeline_gate_enhanced.py
git commit -m "feat(integrity): D4 — character_state_variant gate"
```

## Task 32: D4 추가 — chain_bg + floor_plan gate

**Files:**
- Modify: `backend/app/core/pipeline_gate.py`
- Modify: `backend/tests/test_pipeline_gate_enhanced.py`

- [ ] **Step 1: 실패 테스트 작성**

```python
def test_check_scene_images_ready_blocks_on_chain_bg_missing(seed_episode_with_planner):
    """planner 3 group + chain_bg row 1 → gate.incomplete_chain_bg."""
    db, project_id, episode_id = seed_episode_with_planner(groups=3, chain_bg_existing=1)
    from app.core.pipeline_gate import check_scene_images_ready
    with pytest.raises(AppError) as exc:
        check_scene_images_ready(db, project_id, episode_id)
    assert exc.value.code == "gate.incomplete_chain_bg"


def test_check_scene_images_ready_blocks_on_floor_plan_missing(seed_episode_with_fp_planner):
    db, project_id, episode_id = seed_episode_with_fp_planner(needs_fp_groups=2, fp_existing=0)
    from app.core.pipeline_gate import check_scene_images_ready
    with pytest.raises(AppError) as exc:
        check_scene_images_ready(db, project_id, episode_id)
    assert exc.value.code == "gate.incomplete_floor_plan"
```

- [ ] **Step 2: 메서드 추가**

`pipeline_gate.py`에 검증 6번/7번 추가:
```python
    # 6) chain_bg 검증 (D4)
    from pathlib import Path as _PathBg
    from app.core.config import settings as _settings
    planner_cp = (_PathBg(_settings.projects_dir) / project_id
                  / "checkpoints" / "episodes" / episode_id
                  / "background_chain_planning" / "manifest.json")
    if planner_cp.exists():
        import json as _json
        try:
            planner_data = _json.loads(planner_cp.read_text())
            groups = (planner_data.get("data") or {}).get("groups") or {}
            expected_groups = set(groups.keys())
        except Exception:
            expected_groups = set()
        if expected_groups:
            cb_assets = db.query(ImageAsset).filter(
                ImageAsset.project_id == project_id,
                ImageAsset.episode_id == episode_id,
                ImageAsset.asset_type == "reference",
                ImageAsset.prompt_used.like("%chain_bg:%"),
            ).all()
            existing_cb = set()
            for r in cb_assets:
                m = re.search(r'chain_bg:([\w-]+)', r.prompt_used or "")
                if m and r.file_path and _PathBg(r.file_path).exists():
                    existing_cb.add(m.group(1))
            missing_cb = expected_groups - existing_cb
            if missing_cb:
                raise AppError(
                    code="gate.incomplete_chain_bg",
                    message=f"체인 배경 이미지 미완성: {len(missing_cb)}개 그룹 누락.",
                    status_code=400,
                )

    # 7) floor_plan 검증 (D4)
    fp_cp = (_PathBg(_settings.projects_dir) / project_id
             / "checkpoints" / "episodes" / episode_id
             / "background_master_plan" / "manifest.json")
    if fp_cp.exists():
        try:
            fp_data = _json.loads(fp_cp.read_text())
            fp_groups = (fp_data.get("data") or {}).get("groups") or {}
            expected_fp = {gid for gid, g in fp_groups.items() if g.get("needs_floor_plan")}
        except Exception:
            expected_fp = set()
        if expected_fp:
            fp_assets = db.query(ImageAsset).filter(
                ImageAsset.project_id == project_id,
                ImageAsset.episode_id == episode_id,
                ImageAsset.asset_type == "reference",
                ImageAsset.prompt_used.like("%floor_plan:%"),
            ).all()
            existing_fp = set()
            for r in fp_assets:
                m = re.search(r'floor_plan:([\w-]+)', r.prompt_used or "")
                if m and r.file_path and _PathBg(r.file_path).exists():
                    existing_fp.add(m.group(1))
            missing_fp = expected_fp - existing_fp
            if missing_fp:
                raise AppError(
                    code="gate.incomplete_floor_plan",
                    message=f"도면 이미지 미완성: {len(missing_fp)}개 그룹 누락.",
                    status_code=400,
                )
```

- [ ] **Step 3: 테스트 통과 + 커밋**

Run: `cd backend && .venv/bin/pytest tests/test_pipeline_gate_enhanced.py -v`
Expected: 4 passed (outlook + state_variant + chain_bg + floor_plan).

```bash
git add backend/app/core/pipeline_gate.py backend/tests/test_pipeline_gate_enhanced.py
git commit -m "feat(integrity): D4 — chain_bg + floor_plan gate"
```

## Task 33: PR 3 — 회귀 + 듀얼 리뷰 준비

- [ ] **Step 1: 전체 swept**

Run: `cd backend && .venv/bin/pytest tests/ --tb=short -q`
Expected: ~1490 + ~12 = ~1502 passed.

- [ ] **Step 2: PR 3 push (사용자 승인 후)**

---

# PR 4 — UI 가시성 (Phase 5)

## Task 34: backend API에 recovery_count + last_recovery_reason 노출

**Files:**
- Modify: backend pipeline status API (위치: `grep -rn "step_run\|pipeline.*status" backend/app/api/`)
- Test: 기존 API 테스트 확장

- [ ] **Step 1: API 응답에 컬럼 추가**

`pipeline_progress` 엔드포인트 또는 `step_run_summary` 엔드포인트의 SELECT에 `recovery_count, last_recovery_reason` 추가. 응답 스키마(Pydantic)에 두 필드 추가.

- [ ] **Step 2: API 테스트 확장**

```python
def test_step_run_response_includes_recovery_fields(client, seed_step_run_with_recovery):
    res = client.get(f"/api/projects/{pid}/episodes/{eid}/pipeline/status")
    data = res.json()
    assert "recovery_count" in data["steps"][0]
    assert "last_recovery_reason" in data["steps"][0]
```

- [ ] **Step 3: 테스트 통과 + 커밋**

```bash
git add <api file> <schema file> <test file>
git commit -m "feat(ui): API exposes recovery_count + last_recovery_reason"
```

## Task 35: Frontend recovery_count 뱃지

**Files:**
- Modify: 적절한 React 컴포넌트 (`grep -rn "step_run\|step_id\|status" frontend/src/components/`)

- [ ] **Step 1: 뱃지 컴포넌트 작성**

step status 표시 위치에 `RecoveryBadge` 컴포넌트 추가:
```tsx
{step.recovery_count > 0 && (
  <span
    className="inline-flex items-center px-1.5 py-0.5 rounded text-xs bg-yellow-100 text-yellow-800"
    title={`Auto-recovery ${step.recovery_count}회 (마지막 사유: ${step.last_recovery_reason})`}
  >
    🔄 {step.recovery_count}
  </span>
)}
```

- [ ] **Step 2: vitest 테스트**

```tsx
test("RecoveryBadge shows count when > 0", () => {
  render(<RecoveryBadge step={{ recovery_count: 2, last_recovery_reason: "verify failed" }} />);
  expect(screen.getByText("🔄 2")).toBeInTheDocument();
});

test("RecoveryBadge hidden when count = 0", () => {
  const { container } = render(<RecoveryBadge step={{ recovery_count: 0 }} />);
  expect(container.firstChild).toBeNull();
});
```

- [ ] **Step 3: 통합 + 커밋**

```bash
git add frontend/src/components/.../RecoveryBadge.tsx frontend/src/components/.../*.test.tsx
git commit -m "feat(ui): RecoveryBadge for step auto-recovery count"
```

## Task 36: Frontend recovery_exhausted 모달

**Files:**
- Modify: 에러 처리 위치 (axios interceptor 또는 react-query onError)

- [ ] **Step 1: AppError code 분기 추가**

```tsx
// Error handler
if (error.code === "step.recovery_exhausted") {
  showModal({
    title: "Auto-recovery 한계 도달",
    body: error.message,  // "{step_id} auto-recovery 3회 실패. 마지막 사유: ..."
    actions: ["수동 진단", "닫기"],
  });
}
```

- [ ] **Step 2: 모달 컴포넌트 vitest**

```tsx
test("recovery_exhausted error shows modal", async () => {
  ...
  await waitFor(() => {
    expect(screen.getByText("Auto-recovery 한계 도달")).toBeInTheDocument();
  });
});
```

- [ ] **Step 3: 통합 + 커밋**

```bash
git add frontend/src/...
git commit -m "feat(ui): recovery_exhausted error modal"
```

## Task 37: PR 4 — 회귀 + push

- [ ] **Step 1: 전체 frontend + backend 테스트**

Run: `cd backend && .venv/bin/pytest tests/ -q && cd ../frontend && npm test --run`
Expected: 회귀 0.

- [ ] **Step 2: 사용자 승인 후 push**

---

# Self-review

## Spec coverage 체크

| Spec 섹션 | Plan task |
|---|---|
| §3 Architecture (4 layer) | Tasks 5–13 (verify/cleanup/recovery counter/run() 분기) |
| §4.1 dataclass | Task 4 |
| §4.2 base hook | Task 5 |
| §4.3 run() 통합 | Tasks 9–14 |
| §4.4 per-step verify (Phase A 6 step) | Tasks 19–24 |
| §4.5 D3 runtime fail-soft | Tasks 27–28 |
| §4.6 D4 gate 강화 | Tasks 30–32 |
| §4.7 schema migration | Tasks 1–3 |
| §5 시나리오 A–D | Tasks 14–16, 25 |
| §6 Error handling | Tasks 5(safe), 12(cleanup raise), 14(exhausted) |
| §7 Testing strategy | 모든 task에 TDD |
| §8 Phase 도입 | PR 1–4 분리 |
| §10 결정 8건 | 모두 task로 매핑 |

## Placeholder scan

검색 패턴:
- "TBD", "TODO", "implement later" → 없음
- "appropriate error handling" → 없음
- "similar to Task N" → 없음
- step_runner.py 라인 위치는 정확한 라인 번호 명시 (가능한 경우)

미상이거나 후속 task에 위임된 부분:
- Task 21–24의 "클래스 위치 확인" — 실행 시점에 grep으로 확인
- Task 27 step 3의 코드는 패턴 가이드 — 정확한 변수명/indentation은 원본 코드 일치 (실행자가 read 후 적용)
- conftest fixture 미작성 — Task 5 step 1에 명시했지만 실제 작성은 첫 테스트 통과 시도 시점에 추가

## Type consistency

- `CompletionReport` / `CleanupReport` 인터페이스: Task 4에서 정의, Tasks 5/19–24/29에서 일관 사용
- `verify_completion()` / `cleanup_artifacts()` 시그니처: Task 5에서 base 정의, Tasks 19–24/29에서 override (return type 일치)
- `_record_recovery(reason: str) -> int`: Task 6 정의, Tasks 10–13에서 호출 (시그니처 일치)
- `MAX_RECOVERY_ATTEMPTS = 3`: Task 14에서 inline 정의 (모듈 상수로 추출하지 않음 — 본 task scope)
- `resolve_refs_for_prompt`의 return type: Task 27에서 `(labeled_refs, missing_refs)` 튜플로 변경, Task 28에서 unpacking 일치

---

**Plan 작성 완료.** 사용자 검토 후 실행 옵션 선택:

**1. Subagent-Driven (recommended)** — task별 fresh subagent + 사이 review, fast iteration
**2. Inline Execution** — 이 세션에서 batch + checkpoint review

어느 방식으로 진행할까요?