# Reference Necessity Phase 0+1+2 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:** 캐릭터/소품 reference 이미지 과생성을 차단한다 — catalog 는 넓게 보존하되, 이번 에피소드의 selected-shot 사용량 기준으로 실제 필요한 reference 만 지연 생성한다.

**Architecture:** 세 Phase 를 한 branch 로 묶는다. (0) read-only audit script 가 broad-union 안전망 제거가 안전한지 차집합 감사로 검증한다 — **fail-closed GATE**. (1) reference 생성 보호 집합을 `scene_detail.required_refs` 단독으로 축소하고, 보호 입력인 `t2i_appearance_count` 를 selected-shot 기준으로 정정한다. (2) `scene_detail` 직전에 신규 deterministic step `episode_reference_policy` 가 immutable manifest 1회 계산 → `render_prompt_card` builder 가 manifest 를 읽어 text-only 대상 subject 의 identity policy 를 `generic_descriptor_allowed` 로 결정론적 다운그레이드 (기존 C10 Phase 1 `apply_screen_presence_downgrade` 와 동일 메커니즘). producer/verify recompute 가 동일 manifest 를 읽으므로 card hash drift 없음.

**Tech Stack:** Python 3 / SQLAlchemy / pytest. backend 디렉토리 = `/Users/manta/Documents/Projects/TheRoad-I1/backend` (이하 모든 `app/...` 경로는 이 디렉토리 기준). pytest 실행 CWD = `backend/`.

**범위 밖 (Non-goals):** entity DB 삭제 없음. background reference 경로 무변경. `entity_filter` 프롬프트 reframe 은 이번 branch 범위 밖 (catalog 과보존은 설계상 허용 — 별도 follow-up). scene_detail user prompt 텍스트 변경 없음 — text-only 지시는 deterministic card 의 `id_policy` (`generic_descriptor_allowed`) 가 이미 담당.

**개정 이력:**
- v2 (2026-05-23) — Codex 리뷰 `NEEDS_REVISION` 반영: BLOCKING1 Phase0 GATE fail-closed 화 (Task 1–2), BLOCKING2 `t2i_appearance_count` selected-shot 정정 신규 Task 5, IMPORTANT1 `required_ref_count` circularity 명시 (Task 7 + self-review), prop manifest 모드 `reference_candidate` 로 명확화 (Task 7).
- v3 (2026-05-23) — Codex v2 재리뷰 `NEEDS_REVISION_NARROW` 반영: Task 9 runtime fail-fast (`build_selected_map_or_raise` + `compute_visible_shot_count_from_checkpoints` 가 selected_map 누락 시 AppError — 전체 shot fallback 제거, fail-fast 테스트 3개 추가), Task 2 DB step_run 검사를 raw SQL 로 구체화 (optional 제거 — 필수 GATE).
- **v3-erratum (2026-05-23, 구현 중 발견 — Codex 검토 반영, commit `c16b1e1`)**: Phase 0 GATE predicate 정정. Task 1 의 audit code 는 `audit_set` 중 prompt-ID 가 등장하는 모든 엔티티(`audit_set_with_t2i_usage`)를 곧바로 GATE FAIL 로 처리했으나, fresh checkpoint (project 76212b49 / ep f44339e6) GATE 실행 시 16건 FAIL — 대부분 out-of-scope location + text-only prop/one-shot character 였다. **prompt-ID occurrence 는 diagnostic 일 뿐, GATE blocking 은 materialization risk 로 좁힌다**: location(L##)/outlook(O##) 는 별도 background 경로라 GATE 제외(`out_of_scope` warning), prop(P##) 은 required_refs(kind=prop) 가 단일 SOT 라 prompt-ID 만으론 blocking 불가(`prompt_id_only_prop` warning), character(C##) 는 recurring(`visible_shot_count>=2`) 이면서 required_refs 밖일 때만 blocking — one-shot 은 Phase 2 text_only 정정 대상(`expected_text_only` warning). `safety_diff` 에 `gate_blocking_ids` + `warnings` 추가, GATE FAIL 판정은 `gate_blocking_ids` 비었는지로. input integrity fail-closed 는 무변경. Task 1 의 GATE predicate 설명·테스트(아래 Step 1/3)는 이 erratum 기준으로 읽을 것. variant pole 보호는 broad union 과 독립이라 audit 의 variant 미판정이 GATE 안전성 gap 아님.

---

## 사전 컨텍스트 — 검증된 현재 구조

이 plan 은 아래 코드 사실 위에 작성됨. 실행 에이전트는 수정 전 해당 파일/함수를 반드시 직접 읽을 것.

- **pipeline step 정의**: `app/core/step_manifest.py` `STEP_MANIFEST` dict. 각 entry = `{label, category, order, default_model, provider, depends_on, fan_out, applicability, step_type, lifecycle, schema_version?, ...}`. step class registry = `app/core/steps/__init__.py` `STEP_CLASSES` dict.
- **step 패턴**: `StepRunner` 상속, `_execute(self, mode="resume") -> Dict` 가 `{completed_count, applicable_count, failed_count, data, schema_version?, config_hash?}` 반환. checkpoint 는 StepRunner 가 `_execute` 반환값으로 자동 저장 → `projects/{pid}/checkpoints/episodes/{eid}/{step_id}/manifest.json`. deterministic step 템플릿 = `app/core/steps/shot_dependency_step.py` (LLM 없음, `self._load_prev_checkpoint(step_id)` 로 upstream 읽음).
- **현재 order**: `shot_selection`=15.5, `shot_director`=16.5, `shot_staging`=19.5, `scene_consistency`=19.9, `background_prompt`=21.60, `scene_detail`=21.70, `shot_dependency_t2i`=21.71.
- **reference 생성 보호** (`app/core/entity_protection.py`): `_collect_required_entity_ids(pid, eid)` 가 4-source union (`scene_director.present_entity_ids` ∪ `shot_validator.character_ids` ∪ `shot_director.visible_entity_ids` ∪ `scene_detail.visible_entities` ∪ `scene_detail.render_prompt_card.asset_requirements.required_refs[].id`). `should_skip_low_freq(e, count, is_base_for_variant, is_variant_self, required_by_pipeline)` = 5-gate cascade (`count > 1` 이면 보호). `compute_variant_pole_ids(entities, relations, participants)` = RO-1 variant pole canon UUID 집합.
- **orchestrator** (`app/services/reference_pipeline_orchestrator.py` line 280–338): `_collect_required_entity_ids` → short_id → canon UUID resolve → `should_skip_low_freq` 의 `required` 인자. `_t2i_count_map` (line 281–286) = `EntityEpisodeLink.t2i_appearance_count`. `_low_freq_skip_ids` 를 `save_low_freq_skip_ids` (= `app/core/low_freq_skip.py`) 로 plain JSON array 저장.
- **`t2i_appearance_count` 산출** (`app/services/checkpoint_sync/episode_projection_service.py` `sync_t2i_appearance_counts` line 227–293): **현재 `SceneStill` 전체를 `is_selected`/`status`/`still_index` 필터 없이 카운트** (line 237–240). 같은 orchestrator 의 world-guide stills 쿼리(line 221–227)는 `is_selected==True, still_index>=0, status!="stale"` 로 좁힘 — 대조. → BLOCKING2, Task 5 에서 정정.
- **`low_freq_skip.py`**: `save_low_freq_skip_ids(pid, eid, ids)` = `json.dumps(sorted(set(ids)))` array. `load_low_freq_skip_ids(pid, eid) -> Set[str]`. 소비처 = `pipeline_gate.py`, `api/v1/images.py`, `core/steps/image_steps.py`.
- **subject_reference_policy** (`app/core/subject_reference_policy.py`): enum `ALLOWED_POLICIES = {id_and_outlook_required, base_id_required, generic_descriptor_allowed}`. `apply_screen_presence_downgrade(raw_items, offscreen_referenced, *, where)` 가 off-screen subject 를 `generic_descriptor_allowed` 로 inject/override (`base_id_required` 보존, `id_and_outlook_required` 다운그레이드). `normalize_subject_reference_policy_items(...) -> dict[base→SubjectReferencePolicy]`.
- **render_prompt_card** (`app/core/steps/render_prompt_card.py`): `build_render_prompt_card(...)` (line ~3452) 가 `staging["subject_reference_policy"]` → `filter_subject_reference_policy_to_visible` → `apply_screen_presence_downgrade` (line ~3556) → `normalize_subject_reference_policy_items` → `policy_map`. `build_asset_requirements(..., policy_map=...)` (line 1939) 에서 `_policy.policy == "generic_descriptor_allowed"` 면 character/character_outlook required_ref **emit 안 함** (line 2060–2061). `compute_card_hash(card)` (line 2231) = canonical JSON sha256[:16].
- **detail_steps** (`app/core/steps/detail_steps.py`): `SCENE_DETAIL_SCHEMA_VERSION = 12` (line 106). `_derive_card_inputs_from_ctx(*, ctx, seg, shot_info)` (line 564) 가 ctx 에서 card input 도출. `_collect_card_inputs(*, ctx, seg, shot_info, ...)` (line 753) 가 single source. card build 3지점 = line ~916 (`_recompute_card_hash_with_upstream_dependency` fallback), ~963 (`verify_completion` recompute), ~2891 (`_analyze_one` producer). 세 지점 모두 `builder_inputs = {k:v for k,v in card_inputs.items() if k != "ctx"}` → `build_render_prompt_card(**builder_inputs)`.
- **ctx loader** (`app/core/steps/scene_context_loader.py`): `SceneContextLoader.load_all()` 가 `runner._load_prev_checkpoint(step_id)` 로 각 checkpoint 로드 → `SceneAnalysisContext` (`app/core/dto/scene_analysis.py`) 채움.
- **StepRunner schema 검증**: `step_runner.py` line 597–611 / 739–769 / 1496–1516 — completed cp 의 schema mismatch 를 검사, non-allowlist mismatch 는 BLOCK. `scene_detail` 은 non-allowlist → 운영자 명시 force 필요.

## File Structure

| 파일 | 종류 | 책임 |
|------|------|------|
| `app/services/reference_necessity_audit.py` | 신규 | Phase 0 — checkpoint 읽어 usage matrix + safety_diff + 입력 무결성 검증 (pure 함수, fail-closed) |
| `backend/scripts/reference_necessity_audit.py` | 신규 | Phase 0 — CLI wrapper (DB 조회 + step_run/SceneStill cross-check + report 출력 + exit code) |
| `app/core/low_freq_skip.py` | 수정 | Phase 1 — reasoned report I/O 추가, dual-shape loader |
| `app/core/entity_protection.py` | 수정 | Phase 1 — `_collect_reference_required_ids` (narrow) 추가 |
| `app/services/checkpoint_sync/episode_projection_service.py` | 수정 | Phase 1 — `sync_t2i_appearance_counts` selected-shot 필터 정정 (BLOCKING2) |
| `app/services/reference_pipeline_orchestrator.py` | 수정 | Phase 1 — narrow 보호 집합 + reasoned report 저장 |
| `app/core/episode_reference_policy.py` | 신규 | Phase 2 — manifest 계산 (`compute_episode_reference_policy`) + `extract_text_only_subjects` |
| `app/core/subject_reference_policy.py` | 수정 | Phase 2 — `apply_episode_reference_policy_downgrade` 헬퍼 |
| `app/core/steps/episode_reference_policy_step.py` | 신규 | Phase 2 — `EpisodeReferencePolicyStep(StepRunner)` |
| `app/core/step_manifest.py` | 수정 | Phase 2 — `episode_reference_policy` entry, `scene_detail` depends_on + schema 12→13 |
| `app/core/steps/__init__.py` | 수정 | Phase 2 — `EpisodeReferencePolicyStep` registry 등록 |
| `app/core/dto/scene_analysis.py` | 수정 | Phase 2 — `SceneAnalysisContext.episode_reference_policy` 필드 |
| `app/core/steps/scene_context_loader.py` | 수정 | Phase 2 — `_load_episode_reference_policy()` + `load_all()` wiring |
| `app/core/steps/render_prompt_card.py` | 수정 | Phase 2 — `episode_reference_policy` param + downgrade 적용 |
| `app/core/steps/detail_steps.py` | 수정 | Phase 2 — manifest threading + `SCENE_DETAIL_SCHEMA_VERSION` 12→13 |

---

## Phase 0 — 관찰 및 안전망 차집합 감사

> **GATE (fail-closed)**: Phase 0 audit 가 PASS 해야 Phase 1/2 코드 머지 가능. PASS 조건 = **(a) 필수 입력 5종 전부 정상** (`shot_selection`/`shot_director`/`scene_director`/`shot_validator`/`scene_detail` — manifest 존재 + parse 성공 + manifest status=completed + `data.scenes` 비어있지 않음 + DB step_run completed·non-stale + selected_map 에 shot_director 의 모든 scene 존재) **AND (b) audit_set 안에 실제 selected-shot T2I 사용 엔티티 0건**. 입력 누락·parse fail·partial·stale·selected_map scene 누락 중 하나라도 있으면 **무조건 FAIL** — 빈 audit_set 으로 인한 false PASS 금지, selected_map 누락 시 전체 shot fallback 금지. (b) FAIL = `required_refs` SOT gap = 버그 → required_refs 보강 먼저, broad union crutch 영구화 금지. gap 크면 bundle 중단.

### Task 1: Phase 0 audit — pure 계산 모듈 (fail-closed)

**Files:**
- Create: `app/services/reference_necessity_audit.py`
- Test: `tests/services/test_reference_necessity_audit.py`

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

`tests/services/test_reference_necessity_audit.py`:

```python
"""Phase 0 reference-necessity audit — pure 계산 + fail-closed GATE 검증."""
import json

from app.services.reference_necessity_audit import compute_reference_usage_report

_ALL = ("shot_selection", "shot_director", "scene_director",
        "shot_validator", "scene_detail")


def _write(root, step_id, data, status="completed"):
    d = root / step_id
    d.mkdir(parents=True, exist_ok=True)
    (d / "manifest.json").write_text(
        json.dumps({"status": status, "data": data}), encoding="utf-8",
    )


def _full_checkpoints(tmp_path, *, scene_detail_overrides=None):
    """5종 필수 입력을 전부 정상 상태로 생성. 루트 반환."""
    root = tmp_path / "episodes" / "ep1"
    _write(root, "scene_director",
           {"scenes": [{"scene_index": 1, "present_entity_ids": ["C04"]}]})
    _write(root, "shot_validator",
           {"scenes": [{"scene_index": 1, "shots": [
               {"shot_index": 0, "character_ids": ["C04"]}]}]})
    _write(root, "shot_director",
           {"scenes": [{"scene_index": 1, "shots": [
               {"shot_index": 0, "visible_entity_ids": ["C04", "C01"]}]}]})
    _write(root, "shot_selection",
           {"scenes": [{"scene_index": 1, "selected_shot_indices": [0]}]})
    sd = scene_detail_overrides or {"scenes": [{"scene_index": 1,
        "_shot_index": 0, "visible_entities": ["C01"],
        "render_prompt_card": {"asset_requirements": {"required_refs": [
            {"kind": "character", "id": "C01"}]}},
        "t2i_variations": [{"t2i_prompt": "C01 stands alone"}]}]}
    _write(root, "scene_detail", sd)
    return root


def test_audit_set_is_broad_union_minus_required_refs(tmp_path):
    root = _full_checkpoints(tmp_path)
    report = compute_reference_usage_report(
        checkpoints_dir=root,
        entity_catalog={
            "C01": {"name": "주인공", "entity_type": "character"},
            "C04": {"name": "김형사", "entity_type": "character"},
        },
        generated_ref_counts={"C01": 1, "C04": 2},
    )
    assert report["gate"]["passed"] is True
    assert set(report["safety_diff"]["required_refs_only"]) == {"C01"}
    assert set(report["safety_diff"]["audit_set"]) == {"C04"}
    assert report["safety_diff"]["audit_set_with_t2i_usage"] == []


def test_gate_fails_when_audit_entity_used_in_t2i(tmp_path):
    root = _full_checkpoints(tmp_path, scene_detail_overrides={"scenes": [{
        "scene_index": 1, "_shot_index": 0, "visible_entities": [],
        "render_prompt_card": {"asset_requirements": {"required_refs": []}},
        "t2i_variations": [{"t2i_prompt": "C04 enters the room"}]}]})
    report = compute_reference_usage_report(
        checkpoints_dir=root,
        entity_catalog={"C04": {"name": "김형사", "entity_type": "character"}},
        generated_ref_counts={},
    )
    assert report["gate"]["passed"] is False
    assert "C04" in report["safety_diff"]["audit_set_with_t2i_usage"]


def test_gate_fails_on_missing_required_input(tmp_path):
    """필수 입력 5종 중 하나라도 manifest 부재 → 무조건 FAIL."""
    root = _full_checkpoints(tmp_path)
    import shutil
    shutil.rmtree(root / "shot_director")
    report = compute_reference_usage_report(
        checkpoints_dir=root, entity_catalog={}, generated_ref_counts={},
    )
    assert report["gate"]["passed"] is False
    assert report["gate"]["input_status"]["shot_director"] == "missing"


def test_gate_fails_on_not_completed_status(tmp_path):
    """manifest status != completed (partial 등) → FAIL."""
    root = _full_checkpoints(tmp_path)
    _write(root, "scene_detail", {"scenes": [{"scene_index": 1}]},
           status="partial")
    report = compute_reference_usage_report(
        checkpoints_dir=root, entity_catalog={}, generated_ref_counts={},
    )
    assert report["gate"]["passed"] is False
    assert report["gate"]["input_status"]["scene_detail"] == "not_completed:partial"


def test_gate_fails_on_selected_map_scene_missing(tmp_path):
    """shot_director 에 있는 scene 이 shot_selection 에 없으면 FAIL (fallback 금지)."""
    root = _full_checkpoints(tmp_path)
    # shot_director 에 scene 2 추가, shot_selection 에는 없음
    _write(root, "shot_director", {"scenes": [
        {"scene_index": 1, "shots": [
            {"shot_index": 0, "visible_entity_ids": ["C01"]}]},
        {"scene_index": 2, "shots": [
            {"shot_index": 0, "visible_entity_ids": ["C07"]}]},
    ]})
    report = compute_reference_usage_report(
        checkpoints_dir=root, entity_catalog={}, generated_ref_counts={},
    )
    assert report["gate"]["passed"] is False
    assert any("scene 2" in r for r in report["gate"]["fail_reasons"])
```

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

Run: `cd backend && python -m pytest tests/services/test_reference_necessity_audit.py -v`
Expected: FAIL — `ModuleNotFoundError: app.services.reference_necessity_audit`

- [ ] **Step 3: 모듈 구현**

`app/services/reference_necessity_audit.py`:

```python
"""Phase 0 — reference-necessity audit (read-only, 동작 변경 0).

broad-union 안전망을 reference 생성에서 제거해도 안전한지 차집합 감사로
검증한다. fail-closed: 필수 입력이 누락/parse fail/non-completed 거나
selected_map 이 불완전하면 GATE 를 무조건 FAIL 처리한다 (false PASS 금지).
checkpoint 만 읽는 pure 함수 — DB step_run / SceneStill cross-check 는
CLI wrapper 책임.
"""
from __future__ import annotations

import json
import re
from pathlib import Path
from typing import Any, Dict, List, Optional, Tuple

_ID_RE = re.compile(r"\b([CPLO]\d{2,3})(?:O\d{2,3})?\b")

# Phase 0 GATE 필수 입력 — 하나라도 비정상이면 무조건 FAIL.
REQUIRED_STEPS = (
    "shot_selection", "shot_director", "scene_director",
    "shot_validator", "scene_detail",
)


def _short_id_base(sid: str) -> str:
    return sid.split("O")[0] if sid and "O" in sid else (sid or "")


def _load_with_status(
    checkpoints_dir: Path, step_id: str,
) -> Tuple[Optional[dict], str]:
    """(manifest dict | None, status). status ∈ {ok, missing, parse_error,
    not_completed:<s>, empty}."""
    p = checkpoints_dir / step_id / "manifest.json"
    if not p.exists():
        return None, "missing"
    try:
        cp = json.loads(p.read_text(encoding="utf-8"))
    except (json.JSONDecodeError, OSError):
        return None, "parse_error"
    status = cp.get("status")
    if status is not None and status != "completed":
        return cp, f"not_completed:{status}"
    if not (cp.get("data", {}).get("scenes")):
        return cp, "empty"
    return cp, "ok"


def compute_reference_usage_report(
    *,
    checkpoints_dir: Path,
    entity_catalog: Dict[str, Dict[str, Any]],
    generated_ref_counts: Dict[str, int],
) -> Dict[str, Any]:
    """usage matrix + safety_diff + fail-closed GATE 계산.

    checkpoints_dir = `.../checkpoints/episodes/{eid}` 디렉토리.
    entity_catalog  = {short_id: {name, entity_type}} — catalog SOT.
    generated_ref_counts = {short_id: 생성된 reference ImageAsset 수}.
    """
    cps: Dict[str, Optional[dict]] = {}
    input_status: Dict[str, str] = {}
    fail_reasons: List[str] = []
    for step in REQUIRED_STEPS:
        cp, st = _load_with_status(checkpoints_dir, step)
        cps[step] = cp
        input_status[step] = st
        if st != "ok":
            fail_reasons.append(f"{step}: {st}")

    # selected_map — shot_director 의 모든 scene 이 존재해야 함 (fallback 금지)
    selected: Dict[Any, set] = {}
    sel_cp = cps["shot_selection"]
    if sel_cp:
        for sc in sel_cp.get("data", {}).get("scenes", []) or []:
            selected[sc.get("scene_index")] = set(
                sc.get("selected_shot_indices", []) or []
            )
    sdir_cp = cps["shot_director"]
    if sdir_cp:
        for sc in sdir_cp.get("data", {}).get("scenes", []) or []:
            si = sc.get("scene_index")
            if si not in selected:
                fail_reasons.append(
                    f"shot_selection: scene {si} 누락 — selected_map "
                    f"불완전 (전체 shot fallback 금지)"
                )

    # broad union 4-source (entity_protection._collect_required_entity_ids 와 동일)
    broad: set = set()
    director_cp = cps["scene_director"]
    if director_cp:
        for sc in director_cp.get("data", {}).get("scenes", []) or []:
            for sid in sc.get("present_entity_ids", []) or []:
                broad.add(_short_id_base(sid))
    sv_cp = cps["shot_validator"]
    if sv_cp:
        for sc in sv_cp.get("data", {}).get("scenes", []) or []:
            for sh in sc.get("shots", []) or []:
                for sid in sh.get("character_ids", []) or []:
                    broad.add(_short_id_base(sid))
    if sdir_cp:
        for sc in sdir_cp.get("data", {}).get("scenes", []) or []:
            for sh in sc.get("shots", []) or []:
                for sid in sh.get("visible_entity_ids", []) or []:
                    broad.add(_short_id_base(sid))

    # visible_shot_count — selected shot 한정. selected 에 없는 scene 은
    # fallback 없이 skip (이미 fail_reasons 에 기록됨).
    visible_shot_count: Dict[str, int] = {}
    if sdir_cp:
        for sc in sdir_cp.get("data", {}).get("scenes", []) or []:
            si = sc.get("scene_index")
            sel = selected.get(si)
            if sel is None:
                continue
            for sh in sc.get("shots", []) or []:
                if sh.get("shot_index") not in sel:
                    continue
                for sid in sh.get("visible_entity_ids", []) or []:
                    b = _short_id_base(sid)
                    visible_shot_count[b] = visible_shot_count.get(b, 0) + 1

    # scene_detail: required_refs + visible_entities + t2i prompt id 사용
    required_refs_only: set = set()
    required_ref_count: Dict[str, int] = {}
    prompt_id_count: Dict[str, int] = {}
    sd_cp = cps["scene_detail"]
    if sd_cp:
        for sh in sd_cp.get("data", {}).get("scenes", []) or []:
            for sid in sh.get("visible_entities", []) or []:
                broad.add(_short_id_base(sid))
            rpc = sh.get("render_prompt_card") or {}
            for ref in (rpc.get("asset_requirements") or {}).get(
                "required_refs", []
            ) or []:
                if not isinstance(ref, dict):
                    continue
                rid = ref.get("id")
                if rid:
                    b = _short_id_base(rid)
                    required_refs_only.add(b)
                    broad.add(b)
                    required_ref_count[b] = required_ref_count.get(b, 0) + 1
            for var in sh.get("t2i_variations", []) or []:
                prompt = var.get("t2i_prompt") or ""
                for m in _ID_RE.finditer(prompt):
                    b = _short_id_base(m.group(1))
                    prompt_id_count[b] = prompt_id_count.get(b, 0) + 1

    audit_set = sorted(broad - required_refs_only)
    audit_set_with_t2i = sorted(
        b for b in audit_set if prompt_id_count.get(b, 0) > 0
    )

    entities: List[Dict[str, Any]] = []
    for sid in sorted(entity_catalog):
        meta = entity_catalog[sid]
        entities.append({
            "short_id": sid,
            "name": meta.get("name"),
            "entity_type": meta.get("entity_type"),
            "visible_shot_count": visible_shot_count.get(sid, 0),
            "prompt_id_count": prompt_id_count.get(sid, 0),
            "required_ref_count": required_ref_count.get(sid, 0),
            "generated_refs": generated_ref_counts.get(sid, 0),
        })

    if audit_set_with_t2i:
        fail_reasons.append(
            f"required_refs SOT gap: {audit_set_with_t2i} 가 selected-shot "
            f"t2i_prompt 사용 중"
        )
    gate_passed = not fail_reasons

    return {
        "entities": entities,
        "safety_diff": {
            "broad_required_union": sorted(broad),
            "required_refs_only": sorted(required_refs_only),
            "audit_set": audit_set,
            "audit_set_with_t2i_usage": audit_set_with_t2i,
        },
        "gate": {
            "passed": gate_passed,
            "input_status": input_status,
            "fail_reasons": fail_reasons,
            "reason": (
                "필수 입력 정상 + audit_set 전부 selected-shot T2I 미사용 — "
                "broad union 제거 안전"
                if gate_passed
                else "; ".join(fail_reasons)
            ),
        },
    }
```

> 실행 에이전트 주의: checkpoint manifest 의 top-level `status` 키 존재/값을 `app/core/step_runner.py` 의 checkpoint 저장 코드로 확인. 만약 manifest 에 `status` top-level 키가 없으면(다른 키명이면) `_load_with_status` 의 status 검사를 실제 키에 맞추고, 그래도 불확실하면 CLI 의 DB step_run 검사(Task 2)가 1차 GATE — 단 pure 함수의 presence/parse/empty 검사는 그대로 유지.

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

Run: `cd backend && python -m pytest tests/services/test_reference_necessity_audit.py -v`
Expected: PASS (6 tests)

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

```bash
git add backend/app/services/reference_necessity_audit.py backend/tests/services/test_reference_necessity_audit.py
git commit -m "Phase0: reference-necessity audit pure module (fail-closed gate)"
```

### Task 2: Phase 0 audit — CLI wrapper (DB cross-check)

**Files:**
- Create: `backend/scripts/reference_necessity_audit.py`

- [ ] **Step 1: 의존 모델 확인**

수정 전 grep 으로 확인:
- `ImageAsset` 의 reference 분류 컬럼 (`asset_type`/`entity_id`/`episode_id`/`project_id`) — `app/models/project.py`.
- `SceneStill` 의 `is_selected`/`still_index`/`status`/`t2i_variations_json`/`t2i_prompt_cinematic` — `app/models/project.py`.
- `step_run` 은 ORM 모델 없이 raw SQL 로 조회 — 기존 패턴 = `app/core/step_runner.py:510-515`, `app/services/analysis_dispatch_service.py:110-119`. 쿼리: `SELECT status FROM step_run WHERE project_id=:pid AND episode_id=:eid AND step_id=:sid`. 별도 `is_stale` 컬럼 없음 — `status == 'completed'` 만 PASS.

- [ ] **Step 2: CLI wrapper 작성**

`backend/scripts/reference_necessity_audit.py`:

```python
"""Phase 0 reference-necessity audit CLI (fail-closed GATE).

사용:
  cd backend && python -m scripts.reference_necessity_audit <project_id> <episode_id>

checkpoint + DB 를 읽어 usage matrix / safety_diff 를 계산하고
`projects/{pid}/checkpoints/episodes/{eid}/entity_reference_usage_report.json`
에 저장한다. GATE FAIL 시 exit code 1.

GATE 는 pure 함수(checkpoint presence/status/selected_map) + CLI 의 DB
step_run completed·non-stale 검사 + SceneStill t2i ID cross-check 를 모두
통과해야 PASS.
"""
from __future__ import annotations

import json
import re
import sys
from pathlib import Path

_ID_RE = re.compile(r"\b([CPLO]\d{2,3})(?:O\d{2,3})?\b")
_REQUIRED_STEPS = (
    "shot_selection", "shot_director", "scene_director",
    "shot_validator", "scene_detail",
)


def main(project_id: str, episode_id: str) -> int:
    from app.core.config import settings
    from app.db.session import SessionLocal
    from app.models.project import EntityCanon, EntityEpisodeLink, ImageAsset, SceneStill
    from app.services.reference_necessity_audit import compute_reference_usage_report

    checkpoints_dir = (
        Path(settings.projects_dir) / project_id
        / "checkpoints" / "episodes" / episode_id
    )
    if not checkpoints_dir.exists():
        print(f"checkpoint 디렉토리 없음: {checkpoints_dir}", file=sys.stderr)
        return 2

    db = SessionLocal()
    try:
        links = db.query(EntityEpisodeLink).filter(
            EntityEpisodeLink.project_id == project_id,
            EntityEpisodeLink.episode_id == episode_id,
        ).all()
        canon_ids = [lnk.canon_id for lnk in links]
        canons = db.query(EntityCanon).filter(
            EntityCanon.id.in_(canon_ids)
        ).all() if canon_ids else []
        entity_catalog = {
            c.short_id.split("O")[0]: {"name": c.name, "entity_type": c.entity_type}
            for c in canons if c.short_id
        }
        canon_by_id = {c.id: c for c in canons}

        generated_ref_counts: dict = {}
        ref_assets = db.query(ImageAsset).filter(
            ImageAsset.project_id == project_id,
            ImageAsset.episode_id == episode_id,
            ImageAsset.asset_type == "reference",
        ).all()
        for a in ref_assets:
            c = canon_by_id.get(a.entity_id)
            if c and c.short_id:
                sid = c.short_id.split("O")[0]
                generated_ref_counts[sid] = generated_ref_counts.get(sid, 0) + 1

        # cross-check: DB 의 selected non-stale SceneStill t2i_prompt 에서
        # short_id 추출 → checkpoint scene_detail 의 prompt_id_count 와 비교.
        db_selected_ids: dict = {}
        sel_stills = db.query(SceneStill).filter(
            SceneStill.project_id == project_id,
            SceneStill.episode_id == episode_id,
            SceneStill.is_selected == True,   # noqa: E712
            SceneStill.still_index >= 0,
            SceneStill.status != "stale",
        ).all()
        for st in sel_stills:
            try:
                variations = json.loads(st.t2i_variations_json or "[]")
            except (json.JSONDecodeError, TypeError):
                variations = []
            texts = [v.get("t2i_prompt", "") for v in variations]
            if not texts:
                texts = [st.t2i_prompt_cinematic or ""]
            for t in texts:
                for m in _ID_RE.finditer(t):
                    b = m.group(1).split("O")[0]
                    db_selected_ids[b] = db_selected_ids.get(b, 0) + 1

        # DB step_run — 필수 5 step 전부 status='completed' 여야 GATE PASS.
        # step_run 은 ORM 모델 없음 — raw SQL (step_runner.py:510 패턴).
        from sqlalchemy import text as _sql_text
        step_run_status: dict = {}
        for sid in _REQUIRED_STEPS:
            row = db.execute(_sql_text(
                "SELECT status FROM step_run WHERE project_id = :pid "
                "AND episode_id = :eid AND step_id = :sid"
            ), {"pid": project_id, "eid": episode_id, "sid": sid}).fetchone()
            step_run_status[sid] = row[0] if row else "missing"
    finally:
        db.close()

    report = compute_reference_usage_report(
        checkpoints_dir=checkpoints_dir,
        entity_catalog=entity_catalog,
        generated_ref_counts=generated_ref_counts,
    )
    report["project_id"] = project_id
    report["episode_id"] = episode_id

    # cross-check 결과 첨부 (checkpoint prompt_id_count vs DB selected stills)
    cp_prompt_ids = {
        r["short_id"]: r["prompt_id_count"]
        for r in report["entities"] if r["prompt_id_count"] > 0
    }
    divergence = sorted(
        set(cp_prompt_ids) ^ set(db_selected_ids)
    )
    report["cross_check"] = {
        "db_selected_still_id_count": db_selected_ids,
        "checkpoint_prompt_id_count": cp_prompt_ids,
        "id_set_divergence": divergence,
    }
    if divergence:
        report["gate"]["fail_reasons"].append(
            f"cross-check divergence: checkpoint t2i ID 와 DB selected "
            f"SceneStill ID 불일치 {divergence}"
        )
        report["gate"]["passed"] = False

    # DB step_run GATE — 필수 5 step 전부 status='completed' 여야 PASS.
    # missing / partial / failed / pending / running / stale 전부 FAIL.
    report["db_step_run_status"] = step_run_status
    _bad = {s: st for s, st in step_run_status.items() if st != "completed"}
    if _bad:
        for s, st in sorted(_bad.items()):
            report["gate"]["fail_reasons"].append(f"step_run {s}: {st}")
        report["gate"]["passed"] = False

    out = checkpoints_dir / "entity_reference_usage_report.json"
    out.write_text(
        json.dumps(report, ensure_ascii=False, indent=2), encoding="utf-8",
    )

    print(f"\n=== reference-necessity audit: {project_id} / {episode_id} ===")
    print("input_status:", report["gate"]["input_status"])
    print(f"{'short_id':<10}{'type':<12}{'visible':<9}{'promptID':<10}"
          f"{'reqRef':<8}{'genRef':<8}name")
    for r in report["entities"]:
        print(f"{r['short_id']:<10}{(r['entity_type'] or ''):<12}"
              f"{r['visible_shot_count']:<9}{r['prompt_id_count']:<10}"
              f"{r['required_ref_count']:<8}{r['generated_refs']:<8}{r['name'] or ''}")
    sd = report["safety_diff"]
    print(f"\naudit_set ({len(sd['audit_set'])}): {sd['audit_set']}")
    print(f"audit_set_with_t2i_usage: {sd['audit_set_with_t2i_usage']}")
    print(f"cross_check divergence: {report['cross_check']['id_set_divergence']}")
    print(f"\nGATE: {'PASS' if report['gate']['passed'] else 'FAIL'}")
    for fr in report["gate"]["fail_reasons"]:
        print(f"  - {fr}")
    print(f"report 저장: {out}")
    return 0 if report["gate"]["passed"] else 1


if __name__ == "__main__":
    if len(sys.argv) != 3:
        print("usage: python -m scripts.reference_necessity_audit "
              "<project_id> <episode_id>", file=sys.stderr)
        sys.exit(2)
    sys.exit(main(sys.argv[1], sys.argv[2]))
```

> DB step_run 검사는 위 CLI 코드에 raw SQL 로 구체화되어 **필수**다 (optional fallback 아님 — Codex v2 IMPORTANT1). 5 step 중 하나라도 `status != 'completed'` (missing/partial/failed/pending/running 포함) 면 GATE FAIL. raw SQL 패턴은 `step_runner.py:510` / `analysis_dispatch_service.py:110` 와 동일. `step_run` 테이블/`status` 컬럼명이 실제와 다르면 그 두 파일에서 정확한 이름을 확인해 맞출 것.

- [ ] **Step 3: 구문 검증**

Run: `cd backend && python -c "import ast; ast.parse(open('scripts/reference_necessity_audit.py').read()); print('syntax OK')"`
Expected: `syntax OK`

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

```bash
git add backend/scripts/reference_necessity_audit.py
git commit -m "Phase0: reference-necessity audit CLI (DB step_run + SceneStill cross-check)"
```

> **GATE 실행 시점**: Phase 1/2 코드 머지 전, 최신 fresh E2E checkpoint 가 있는 project/episode 에 대해 CLI 를 실행한다 (Task 14 Step 1). GATE PASS → Phase 1/2 머지. GATE FAIL → 원인별 대응 (입력 무결성 문제면 fresh E2E 재실행, required_refs gap 이면 required_refs 보강 후 재설계). 결과를 Codex 에 보고 후 진입 승인.

---

## Phase 1 — ref_image_gen 과생성 차단

> Phase 1 단독 머지 금지 — Phase 2 와 같은 branch 에서 함께 E2E 검증. `required_refs` 계약은 FINDING 9 버그 이력 영역.

### Task 3: low_freq_skip — reasoned report I/O

**Files:**
- Modify: `app/core/low_freq_skip.py`
- Test: `tests/core/test_low_freq_skip.py` (없으면 생성)

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

`tests/core/test_low_freq_skip.py` (파일 없으면 생성, 있으면 append):

```python
"""low_freq_skip reasoned report I/O."""
from app.core import low_freq_skip
from app.core.config import settings


def test_save_and_load_reasoned_report(tmp_path, monkeypatch):
    monkeypatch.setattr(settings, "projects_dir", str(tmp_path))
    decisions = [
        {"canon_id": "u-a", "skipped": True, "reason": "no required_ref"},
        {"canon_id": "u-b", "skipped": False, "reason": "required_ref present"},
    ]
    low_freq_skip.save_low_freq_skip_report("p1", "e1", decisions)
    assert low_freq_skip.load_low_freq_skip_ids("p1", "e1") == {"u-a"}
    report = low_freq_skip.load_low_freq_skip_report("p1", "e1")
    assert report["version"] == 2
    assert {d["canon_id"] for d in report["decisions"]} == {"u-a", "u-b"}


def test_load_ids_backcompat_with_legacy_array(tmp_path, monkeypatch):
    """구 plain-array 포맷도 load_low_freq_skip_ids 가 계속 읽는다."""
    monkeypatch.setattr(settings, "projects_dir", str(tmp_path))
    low_freq_skip.save_low_freq_skip_ids("p1", "e1", ["u-x", "u-y"])
    assert low_freq_skip.load_low_freq_skip_ids("p1", "e1") == {"u-x", "u-y"}
```

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

Run: `cd backend && python -m pytest tests/core/test_low_freq_skip.py -v`
Expected: FAIL — `AttributeError: ... 'save_low_freq_skip_report'`

- [ ] **Step 3: 구현**

`app/core/low_freq_skip.py` 의 `load_low_freq_skip_ids` 의 `try` 블록을 아래로 교체:

```python
    try:
        raw = json.loads(skip_file.read_text(encoding="utf-8"))
    except Exception as exc:
        logger.warning(
            "low_freq_skip: %s parse failed: %s", skip_file, exc,
        )
        return set()
    # v1 = plain array of canon_id / v2 = {"version":2,"decisions":[...]}
    if isinstance(raw, dict):
        return {
            d["canon_id"] for d in raw.get("decisions", [])
            if isinstance(d, dict) and d.get("skipped") and d.get("canon_id")
        }
    return set(raw)
```

파일 끝에 추가:

```python
def save_low_freq_skip_report(
    project_id: str, episode_id: str, decisions: Iterable[dict],
) -> None:
    """reasoned object 포맷(v2)으로 저장.

    decisions item = {canon_id, skipped: bool, reason: str, ...}.
    `load_low_freq_skip_ids` 가 v2 dict 도 읽으므로 기존 소비처 무변경.
    """
    skip_file = get_skip_file_path(project_id, episode_id)
    skip_file.parent.mkdir(parents=True, exist_ok=True)
    payload = {
        "version": 2,
        "decisions": sorted(
            (dict(d) for d in decisions),
            key=lambda d: d.get("canon_id", ""),
        ),
    }
    skip_file.write_text(
        json.dumps(payload, ensure_ascii=False), encoding="utf-8",
    )


def load_low_freq_skip_report(project_id: str, episode_id: str) -> dict:
    """reasoned report 반환. v1 array 면 decisions 로 승격, 부재 시 빈 v2."""
    skip_file = get_skip_file_path(project_id, episode_id)
    if not skip_file.exists():
        return {"version": 2, "decisions": []}
    try:
        raw = json.loads(skip_file.read_text(encoding="utf-8"))
    except Exception as exc:
        logger.warning("low_freq_skip: %s parse failed: %s", skip_file, exc)
        return {"version": 2, "decisions": []}
    if isinstance(raw, dict):
        return raw
    return {
        "version": 2,
        "decisions": [
            {"canon_id": cid, "skipped": True, "reason": "(legacy v1)"}
            for cid in raw
        ],
    }
```

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

Run: `cd backend && python -m pytest tests/core/test_low_freq_skip.py -v`
Expected: PASS

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

```bash
git add backend/app/core/low_freq_skip.py backend/tests/core/test_low_freq_skip.py
git commit -m "Phase1: low_freq_skip reasoned report I/O + dual-shape loader"
```

### Task 4: entity_protection — narrow reference-required 집합

**Files:**
- Modify: `app/core/entity_protection.py`
- Test: `tests/core/test_entity_protection.py` (있으면 append, 없으면 생성)

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

`tests/core/test_entity_protection.py` 에 추가:

```python
"""Phase 1 — _collect_reference_required_ids narrow 집합 검증."""
import json
from app.core import entity_protection
from app.core.config import settings


def _write_cp(tmp_path, monkeypatch, pid, eid, step_id, data):
    monkeypatch.setattr(settings, "projects_dir", str(tmp_path))
    d = tmp_path / pid / "checkpoints" / "episodes" / eid / step_id
    d.mkdir(parents=True, exist_ok=True)
    (d / "manifest.json").write_text(json.dumps({"data": data}), encoding="utf-8")


def test_narrow_set_is_only_scene_detail_required_refs(tmp_path, monkeypatch):
    """narrow 집합 = scene_detail.required_refs[].id 만. visible_entities 제외,
    scene_director / shot_director / shot_validator 전부 제외."""
    _write_cp(tmp_path, monkeypatch, "p", "e", "scene_director",
              {"scenes": [{"present_entity_ids": ["C04"]}]})
    _write_cp(tmp_path, monkeypatch, "p", "e", "shot_director",
              {"scenes": [{"shots": [{"visible_entity_ids": ["C05"]}]}]})
    _write_cp(tmp_path, monkeypatch, "p", "e", "scene_detail",
              {"scenes": [{
                  "visible_entities": ["C06"],
                  "render_prompt_card": {"asset_requirements": {"required_refs": [
                      {"kind": "character", "id": "C01"},
                      {"kind": "character_outlook", "id": "C02O03"},
                  ]}}}]})
    got = entity_protection._collect_reference_required_ids("p", "e")
    assert got == {"C01", "C02"}  # composite → base. C04/C05/C06 미포함


def test_narrow_set_empty_when_no_scene_detail(tmp_path, monkeypatch):
    monkeypatch.setattr(settings, "projects_dir", str(tmp_path))
    assert entity_protection._collect_reference_required_ids("p", "e") == set()
```

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

Run: `cd backend && python -m pytest tests/core/test_entity_protection.py -k narrow -v`
Expected: FAIL — `AttributeError: ... '_collect_reference_required_ids'`

- [ ] **Step 3: 구현**

`app/core/entity_protection.py` 의 `_collect_required_entity_ids` 함수 **바로 아래** 에 추가:

```python
def _collect_reference_required_ids(project_id: str, episode_id: str) -> set[str]:
    """Phase 1 — reference 생성 보호의 단일 SOT.

    `scene_detail.render_prompt_card.asset_requirements.required_refs[].id` 만
    수집한다 (`_collect_required_entity_ids` 의 4-source broad union 과 달리
    scene_director.present / shot_validator.character_ids / shot_director.
    visible_entity_ids / scene_detail.visible_entities 전부 **제외**).

    설계 근거: docs/reference-necessity/index.html §6.1 — broad union 은
    catalog/audit 용도로만 남기고, reference generation 보호는 실제 attach
    계약(required_refs)에 가장 가까운 단일 source 로 좁힌다.
    """
    short_ids: set[str] = set()
    sd_cp = _load_cp(project_id, episode_id, "scene_detail")
    if sd_cp:
        for sh in sd_cp.get("data", {}).get("scenes", []) or []:
            rpc = sh.get("render_prompt_card") or {}
            asset_req = rpc.get("asset_requirements") or {}
            for ref in asset_req.get("required_refs", []) or []:
                if not isinstance(ref, dict):
                    continue
                rid = ref.get("id")
                if rid:
                    short_ids.add(_short_id_base(rid))
    return short_ids
```

> `_collect_required_entity_ids` (broad union) 는 **삭제하지 않는다** — Phase 0 audit 가 계속 사용. narrow 함수는 신규 병행 추가.

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

Run: `cd backend && python -m pytest tests/core/test_entity_protection.py -k narrow -v`
Expected: PASS

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

```bash
git add backend/app/core/entity_protection.py backend/tests/core/test_entity_protection.py
git commit -m "Phase1: entity_protection._collect_reference_required_ids narrow SOT"
```

### Task 5: episode_projection_service — t2i_appearance_count selected-shot 정정 (BLOCKING2)

> Codex BLOCKING2: `sync_t2i_appearance_counts` 가 현재 `SceneStill` 전체를 카운트 — stale/비선택 still 의 옛 prompt 가 `count > 1` 보호를 만들어 reference 과생성 fix 를 무력화한다. selected/non-stale/still_index>=0 으로 좁힌다.

**Files:**
- Modify: `app/services/checkpoint_sync/episode_projection_service.py`
- Test: `tests/services/test_episode_projection_service.py` (있으면 append, 없으면 생성 — 기존 projection-service 테스트 패턴 따를 것)

- [ ] **Step 1: 소비처 grep (blast radius 확인)**

Run: `cd backend && grep -rn "t2i_appearance_count" app/`
모든 소비처가 "selected-shot 실사용 횟수" 의미를 기대하는지 확인. 인플레이션된 (stale 포함) 의미에 의존하는 소비처가 발견되면 Codex 에 보고 후 진행.

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

`tests/services/test_episode_projection_service.py` 에 추가. 기존 파일의 DB fixture/setup 패턴(SceneStill / EntityCanon / EntityEpisodeLink row 생성)을 그대로 따른다:

```python
def test_t2i_count_excludes_stale_and_nonselected_stills(db_session, ...):
    """stale / is_selected=False / still_index<0 SceneStill 의 t2i_prompt 는
    t2i_appearance_count 에 기여하지 않는다."""
    # setup: 같은 episode 에 SceneStill 3개 —
    #   (A) is_selected=True, status="ok", still_index=0, t2i="C01 ..."
    #   (B) is_selected=False, status="ok", still_index=1, t2i="C01 ..."
    #   (C) is_selected=True, status="stale", still_index=2, t2i="C01 ..."
    # + EntityCanon(short_id="C01") + EntityEpisodeLink
    # 실행: EpisodeProjectionService(...).sync_t2i_appearance_counts()
    # 검증: C01 의 EntityEpisodeLink.t2i_appearance_count == 1 (A 만)
    ...
```

> 실행 에이전트 주의: `tests/services/` 또는 `tests/services/checkpoint_sync/` 에서 `EpisodeProjectionService` / `sync_t2i_appearance_counts` 를 다루는 기존 테스트를 grep 해 fixture 형태(생성자 인자, DB 세션 fixture)를 정확히 복사할 것. 기존 테스트가 없으면 `SceneStill` 모델 컬럼(`is_selected`, `still_index`, `status`, `t2i_variations_json`, `project_id`, `episode_id`)로 직접 row 를 만든다.

- [ ] **Step 3: 테스트 실패 확인**

Run: `cd backend && python -m pytest tests/services/test_episode_projection_service.py -k t2i_count_excludes -v`
Expected: FAIL — 현재 count=3 (전체 still), 기대 1

- [ ] **Step 4: 구현**

`app/services/checkpoint_sync/episode_projection_service.py` `sync_t2i_appearance_counts` 의 `stills` 쿼리 (line ~237–240) 에 필터 추가 — orchestrator line 221–227 의 world-guide stills 쿼리와 동일 기준:

```python
        stills = self.db.query(SceneStill).filter(
            SceneStill.project_id == self.project_id,
            SceneStill.episode_id == self.episode_id,
            SceneStill.is_selected == True,   # noqa: E712
            SceneStill.still_index >= 0,
            SceneStill.status != "stale",
        ).all()
```

docstring 의 "카운트 단위" 줄에 "selected / non-stale / still_index>=0 still 한정" 명시 추가.

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

Run: `cd backend && python -m pytest tests/services/test_episode_projection_service.py -v`
Expected: PASS (신규 + 기존 회귀 0)

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

```bash
git add backend/app/services/checkpoint_sync/episode_projection_service.py backend/tests/services/test_episode_projection_service.py
git commit -m "Phase1: t2i_appearance_count = selected non-stale stills only (BLOCKING2)"
```

### Task 6: orchestrator — narrow 보호 집합 wiring + reasoned report

**Files:**
- Modify: `app/services/reference_pipeline_orchestrator.py`

- [ ] **Step 1: 수정 대상 읽기**

`app/services/reference_pipeline_orchestrator.py` line 280–344 (저빈도 skip 블록) + import 부 확인.

- [ ] **Step 2: import 교체**

`_collect_required_entity_ids` import 를 narrow 함수로 변경:

```python
from app.core.entity_protection import (
    _collect_reference_required_ids,
    compute_variant_pole_ids,
    should_skip_low_freq,
)
```

(기존 `_collect_required_entity_ids` import 제거 — orchestrator 내 타 사용처 없는지 grep 확인.)

- [ ] **Step 3: 보호 집합 계산 교체**

line 292–305 의 broad union 블록을 narrow 로 교체:

```python
        # Phase 1 — reference 생성 보호 = scene_detail.required_refs 단독 SOT.
        # broad 4-source union (scene_director.present / shot_validator /
        # shot_director.visible / scene_detail.visible_entities) 은 reference
        # 보호에서 제외 — Phase 0 audit 가 차집합 안전성 검증 완료.
        _required_short_ids = _collect_reference_required_ids(
            self._project_id, episode_id,
        )
        _required_canon_uuids: set = set()
        if _required_short_ids:
            _rows = self._db.query(EntityCanon.id).filter(
                EntityCanon.project_id == self._project_id,
                EntityCanon.short_id.in_(_required_short_ids),
            ).all()
            _required_canon_uuids = {r[0] for r in _rows}
```

- [ ] **Step 4: reasoned report 저장으로 교체**

line 315–338 의 skip 결정 루프 + `save_low_freq_skip_ids` 호출을 교체:

```python
        _low_freq_skip_ids: set = set()
        _skip_decisions: list = []
        for e in entities:
            eid = e["id"]
            etype = e.get("entity_type", "")
            if etype in ("location", "outlook"):
                continue
            count = _t2i_count_map.get(eid, 0)
            is_base_for_variant = eid in _reverse_dep_ids
            is_variant_self = eid in _variant_pole_uuids
            required = eid in _required_canon_uuids
            skipped = should_skip_low_freq(
                e, count, is_base_for_variant, is_variant_self, required,
            )
            if skipped:
                _low_freq_skip_ids.add(eid)
                _reason = "no required_ref + low t2i count + not variant"
            elif required:
                _reason = "protected: scene_detail.required_refs"
            elif count > 1:
                _reason = f"protected: t2i_count={count}"
            elif is_base_for_variant:
                _reason = "protected: base_for_variant"
            elif is_variant_self:
                _reason = "protected: variant_self"
            else:
                _reason = "protected"
            _skip_decisions.append({
                "canon_id": eid,
                "name": e.get("name"),
                "entity_type": etype,
                "skipped": skipped,
                "reason": _reason,
                "t2i_count": count,
            })
            logger.info(
                "Low-freq decision: %s (%s, t2i_count=%d, base_for_variant=%s, "
                "variant_self=%s, required=%s, skipped=%s)",
                e.get("name"), etype, count, is_base_for_variant,
                is_variant_self, required, skipped,
            )

        from app.core.low_freq_skip import save_low_freq_skip_report
        save_low_freq_skip_report(
            self._project_id, episode_id, _skip_decisions,
        )
```

기존 `from app.core.low_freq_skip import save_low_freq_skip_ids` 줄(line ~337) 제거. `save_low_freq_skip_ids` 함수 자체는 `low_freq_skip.py` 에 보존 (타 소비처 가능성).

- [ ] **Step 5: 회귀 테스트**

Run: `cd backend && python -m pytest tests/services/ tests/core/test_entity_protection.py tests/core/test_low_freq_skip.py -q`
Expected: PASS (회귀 0). 실패 시 pre-existing vs new 분류 → Codex entry-sanity.

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

```bash
git add backend/app/services/reference_pipeline_orchestrator.py
git commit -m "Phase1: orchestrator narrow reference protection + reasoned skip report"
```

---

## Phase 2 — scene_detail 전 episode_reference_policy overlay

### Task 7: episode_reference_policy core — manifest 계산

> **IMPORTANT1 (Codex)**: 이 step 은 `scene_detail` **전** 에 실행되므로 `required_ref_count` 를 알 수 없다 (circularity). SOT §10.1 게이트 `visible_shot_count >= 2 OR required_ref_count >= 1` 중 `required_ref_count >= 1` 부분은 이 step 이 판정하지 않고, downstream 에서 자연 집행된다: scene_detail 이 required_ref 를 만들면 Phase 1 narrow 보호(`_collect_reference_required_ids`)가 materialization 시점에 보호한다. 따라서 이 step 의 character 게이트는 `visible_shot_count >= 2 OR variant` 만이다.

**Files:**
- Create: `app/core/episode_reference_policy.py`
- Test: `tests/core/test_episode_reference_policy.py`

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

`tests/core/test_episode_reference_policy.py`:

```python
"""episode_reference_policy manifest 계산 + text_only 추출."""
from app.core.episode_reference_policy import (
    compute_episode_reference_policy,
    extract_text_only_subjects,
)


def test_character_two_visible_shots_is_reference_required():
    manifest = compute_episode_reference_policy(
        visible_shot_count={"C01": 3, "C08": 1},
        entity_types={"C01": "character", "C08": "character"},
        variant_pole_short_ids=set(),
    )
    assert manifest["policy"]["C01"]["mode"] == "reference_required"
    assert manifest["policy"]["C08"]["mode"] == "text_only"


def test_variant_pole_protected_even_with_one_shot():
    manifest = compute_episode_reference_policy(
        visible_shot_count={"C12": 1},
        entity_types={"C12": "character"},
        variant_pole_short_ids={"C12"},
    )
    assert manifest["policy"]["C12"]["mode"] == "reference_required"
    assert "variant" in manifest["policy"]["C12"]["reason"]


def test_nonhuman_uses_same_rule_as_character():
    manifest = compute_episode_reference_policy(
        visible_shot_count={"C09": 2},
        entity_types={"C09": "character"},
        variant_pole_short_ids=set(),
    )
    assert manifest["policy"]["C09"]["mode"] == "reference_required"


def test_prop_mode_is_reference_candidate_not_required():
    """prop 은 provisional — mode 가 reference_candidate (생성 확정 아님)."""
    manifest = compute_episode_reference_policy(
        visible_shot_count={"P10": 2, "P11": 1},
        entity_types={"P10": "prop", "P11": "prop"},
        variant_pole_short_ids=set(),
    )
    assert manifest["policy"]["P10"]["mode"] == "reference_candidate"
    assert manifest["policy"]["P10"]["provisional"] is True
    assert manifest["policy"]["P11"]["mode"] == "text_only"


def test_extract_text_only_subjects_chars_only():
    """overlay 는 text_only character subject 만 추출. prop 제외."""
    manifest = compute_episode_reference_policy(
        visible_shot_count={"C08": 1, "P11": 1, "C01": 3},
        entity_types={"C08": "character", "P11": "prop", "C01": "character"},
        variant_pole_short_ids=set(),
    )
    text_only = extract_text_only_subjects(manifest)
    assert set(text_only) == {"C08"}
    assert isinstance(text_only["C08"], str)


def test_extract_handles_none_manifest():
    assert extract_text_only_subjects(None) == {}
    assert extract_text_only_subjects({}) == {}
```

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

Run: `cd backend && python -m pytest tests/core/test_episode_reference_policy.py -v`
Expected: FAIL — `ModuleNotFoundError`

- [ ] **Step 3: 구현**

`app/core/episode_reference_policy.py`:

```python
"""episode_reference_policy — scene_detail 직전 reference necessity manifest.

설계: docs/reference-necessity/index.html (v2 decisions locked).

deterministic 계산 (LLM 없음). 이번 에피소드 selected-shot usage 로
각 catalog 엔티티의 reference necessity 를 1회 확정한다. 이 manifest 가
render_prompt_card producer 와 verify recompute 의 단일 immutable SOT 가 되어
card hash drift 를 막는다.

규칙 (§5.1 / §10):
- character / nonhuman (C##): visible_shot_count >= 2 또는 variant pole →
  reference_required, 아니면 text_only.
- prop (P## 등): provisional — mode 는 reference_candidate / text_only.
  reference_candidate 는 "생성 확정" 이 아니라 후보 표시일 뿐이고, 최종
  게이트는 materialization 단계의 required_ref_count >= 1 이다 (§10.3).

required_ref_count 미사용 사유: 이 step 은 scene_detail 전이라 required_ref_
count 를 알 수 없다 (circularity). §10.1 의 `required_ref_count >= 1` 절은
downstream 에서 집행된다 — scene_detail 이 required_ref 를 만들면 Phase 1
narrow 보호가 materialization 시점에 그 엔티티를 보호한다.
"""
from __future__ import annotations

from typing import Any, Dict, Optional, Set

EPISODE_REFERENCE_POLICY_SCHEMA_VERSION = 1

# character / nonhuman recurring threshold (§10.1).
_RECURRING_VISIBLE_SHOT_THRESHOLD = 2


def _is_character_subject(short_id: str) -> bool:
    """C## prefix = identity 를 가진 depicted subject (동물/비인간 포함)."""
    return short_id.startswith("C")


def compute_episode_reference_policy(
    *,
    visible_shot_count: Dict[str, int],
    entity_types: Dict[str, str],
    variant_pole_short_ids: Set[str],
) -> Dict[str, Any]:
    """reference necessity manifest 계산.

    visible_shot_count = {short_id: selected-shot 등장 횟수}.
    entity_types       = {short_id: entity_type}.
    variant_pole_short_ids = identity/transformation variant pole short_id 집합.
    """
    policy: Dict[str, Dict[str, Any]] = {}
    all_ids = set(visible_shot_count) | set(entity_types)
    for sid in sorted(all_ids):
        etype = entity_types.get(sid, "")
        vsc = visible_shot_count.get(sid, 0)
        is_variant = sid in variant_pole_short_ids

        if _is_character_subject(sid):
            if is_variant:
                mode, reason = "reference_required", (
                    "identity/transformation variant pole — variant 보호 유지"
                )
            elif vsc >= _RECURRING_VISIBLE_SHOT_THRESHOLD:
                mode, reason = "reference_required", (
                    f"visible_shot_count={vsc} >= "
                    f"{_RECURRING_VISIBLE_SHOT_THRESHOLD}"
                )
            else:
                mode, reason = "text_only", (
                    f"visible_shot_count={vsc} — selected-shot 저빈도 "
                    f"주변 subject"
                )
            policy[sid] = {
                "mode": mode,
                "reason": reason,
                "entity_type": etype or "character",
                "visible_shot_count": vsc,
                "provisional": False,
            }
        else:
            # prop / 기타 — provisional. reference_candidate 는 후보 표시일
            # 뿐, 최종 게이트는 materialization 의 required_ref_count >= 1.
            if vsc >= _RECURRING_VISIBLE_SHOT_THRESHOLD:
                mode, reason = "reference_candidate", (
                    f"provisional candidate: visible_shot_count={vsc} — "
                    f"최종 확정은 materialization 의 required_ref_count"
                )
            else:
                mode, reason = "text_only", (
                    f"provisional: visible_shot_count={vsc}"
                )
            policy[sid] = {
                "mode": mode,
                "reason": reason,
                "entity_type": etype or "prop",
                "visible_shot_count": vsc,
                "provisional": True,
            }
    return {
        "schema_version": EPISODE_REFERENCE_POLICY_SCHEMA_VERSION,
        "policy": policy,
    }


def extract_text_only_subjects(
    manifest: Optional[Dict[str, Any]],
) -> Dict[str, str]:
    """manifest 에서 text_only character subject 만 {C##: reason} 으로 추출.

    render_prompt_card 의 downgrade overlay 입력. prop / 기타는 제외 —
    prop 은 render_contracts SOT + materialization 게이트가 담당.
    """
    if not manifest or not isinstance(manifest, dict):
        return {}
    out: Dict[str, str] = {}
    for sid, p in (manifest.get("policy") or {}).items():
        if not isinstance(p, dict):
            continue
        if p.get("mode") != "text_only":
            continue
        if not str(sid).startswith("C"):
            continue
        out[sid] = p.get("reason") or "episode_reference_policy: text_only"
    return out
```

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

Run: `cd backend && python -m pytest tests/core/test_episode_reference_policy.py -v`
Expected: PASS (6 tests)

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

```bash
git add backend/app/core/episode_reference_policy.py backend/tests/core/test_episode_reference_policy.py
git commit -m "Phase2: episode_reference_policy manifest compute + text_only extract"
```

### Task 8: subject_reference_policy — episode policy downgrade 헬퍼

**Files:**
- Modify: `app/core/subject_reference_policy.py`
- Test: `tests/core/test_subject_reference_policy.py` (있으면 append)

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

`tests/core/test_subject_reference_policy.py` 에 추가:

```python
"""apply_episode_reference_policy_downgrade — text_only subject 다운그레이드."""
from app.core.subject_reference_policy import (
    apply_episode_reference_policy_downgrade,
)


def test_downgrade_injects_generic_for_missing_entry():
    out = apply_episode_reference_policy_downgrade(
        [], {"C08": "text_only low freq"}, where="t",
    )
    entry = next(i for i in out if i["subject_id"] == "C08")
    assert entry["policy"] == "generic_descriptor_allowed"
    assert entry["policy_type"] == "identity_reference"


def test_downgrade_overrides_id_and_outlook_required():
    items = [{"subject_id": "C08", "policy_type": "identity_reference",
              "policy": "id_and_outlook_required", "reason": "default"}]
    out = apply_episode_reference_policy_downgrade(
        items, {"C08": "text_only"}, where="t",
    )
    entry = next(i for i in out if i["subject_id"] == "C08")
    assert entry["policy"] == "generic_descriptor_allowed"


def test_downgrade_preserves_base_id_required():
    """base_id_required 는 의도적 partial-frame 정책 — 보존 (materialization
    reconciliation: 이 subject 는 required_ref 를 받아 Phase 1 보호됨)."""
    items = [{"subject_id": "C08", "policy_type": "identity_reference",
              "policy": "base_id_required", "reason": "explicit"}]
    out = apply_episode_reference_policy_downgrade(
        items, {"C08": "text_only"}, where="t",
    )
    entry = next(i for i in out if i["subject_id"] == "C08")
    assert entry["policy"] == "base_id_required"


def test_downgrade_noop_when_no_text_only():
    items = [{"subject_id": "C01", "policy_type": "identity_reference",
              "policy": "id_and_outlook_required", "reason": "x"}]
    assert apply_episode_reference_policy_downgrade(items, {}, where="t") == items


def test_downgrade_passthrough_non_list():
    assert apply_episode_reference_policy_downgrade(
        None, {"C08": "x"}, where="t") is None
```

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

Run: `cd backend && python -m pytest tests/core/test_subject_reference_policy.py -k downgrade -v`
Expected: FAIL — `ImportError: cannot import name 'apply_episode_reference_policy_downgrade'`

- [ ] **Step 3: 구현**

`app/core/subject_reference_policy.py` 의 `apply_screen_presence_downgrade` 함수 **바로 아래** 에 추가:

```python
def apply_episode_reference_policy_downgrade(
    raw_items: Optional[list],
    text_only_subjects: dict,
    *,
    where: str = "",
) -> Optional[list]:
    """Phase 2 — episode_reference_policy manifest 가 text_only 로 판정한
    subject 의 identity policy 를 `generic_descriptor_allowed` 로 다운그레이드.

    `apply_screen_presence_downgrade` 와 동일 override 규칙:
      - 기존 entry 없음            → generic_descriptor_allowed inject;
      - id_and_outlook_required    → generic_descriptor_allowed 다운그레이드;
      - base_id_required           → 보존 (의도적 partial-frame 정책 우선 —
                                     이 subject 는 required_ref 를 받아
                                     Phase 1 materialization 보호 대상);
      - generic_descriptor_allowed → 보존.

    text_only_subjects = {base C## subject_id: reason}.
    `raw_items` 가 list 아니면 그대로 반환 (caller validator 가 처리).
    """
    if not isinstance(raw_items, list):
        return raw_items
    if not text_only_subjects:
        return raw_items
    items = [dict(it) if isinstance(it, dict) else it for it in raw_items]
    by_base: dict = {}
    for it in items:
        if isinstance(it, dict) and isinstance(it.get("subject_id"), str):
            by_base.setdefault(it["subject_id"].split("O")[0], it)
    for sid, reason in sorted(text_only_subjects.items()):
        base = sid.split("O")[0]
        existing = by_base.get(base)
        if existing is None:
            items.append({
                "subject_id": base,
                "policy_type": "identity_reference",
                "policy": "generic_descriptor_allowed",
                "reason": reason or "episode_reference_policy:text_only",
            })
        elif existing.get("policy") == "id_and_outlook_required":
            existing["policy"] = "generic_descriptor_allowed"
            existing["reason"] = (
                f"{reason} | overrode id_and_outlook_required "
                f"(episode_reference_policy)"
            )
    return items
```

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

Run: `cd backend && python -m pytest tests/core/test_subject_reference_policy.py -k downgrade -v`
Expected: PASS (5 tests)

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

```bash
git add backend/app/core/subject_reference_policy.py backend/tests/core/test_subject_reference_policy.py
git commit -m "Phase2: apply_episode_reference_policy_downgrade helper"
```

### Task 9: EpisodeReferencePolicyStep + manifest 등록

**Files:**
- Create: `app/core/steps/episode_reference_policy_step.py`
- Modify: `app/core/step_manifest.py`
- Modify: `app/core/steps/__init__.py`
- Test: `tests/core/steps/test_episode_reference_policy_step.py`

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

`tests/core/steps/test_episode_reference_policy_step.py`:

```python
"""EpisodeReferencePolicyStep — visible_shot_count 집계 + fail-fast 검증."""
import pytest

from app.core.errors import AppError
from app.core.steps.episode_reference_policy_step import (
    build_selected_map_or_raise,
    compute_visible_shot_count_from_checkpoints,
)


def test_visible_shot_count_over_selected_shots_only():
    shot_director_data = {"scenes": [
        {"scene_index": 1, "shots": [
            {"shot_index": 0, "visible_entity_ids": ["C01", "C08"]},
            {"shot_index": 1, "visible_entity_ids": ["C01"]},
            {"shot_index": 2, "visible_entity_ids": ["C01", "C99"]},
        ]},
    ]}
    selected = {1: {0, 1}}  # shot 2 미선택
    got = compute_visible_shot_count_from_checkpoints(
        shot_director_data, selected,
    )
    assert got["C01"] == 2
    assert got["C08"] == 1
    assert "C99" not in got


def test_visible_shot_count_composite_id_normalized_to_base():
    shot_director_data = {"scenes": [
        {"scene_index": 1, "shots": [
            {"shot_index": 0, "visible_entity_ids": ["C01O02"]},
        ]},
    ]}
    got = compute_visible_shot_count_from_checkpoints(
        shot_director_data, {1: {0}},
    )
    assert got.get("C01") == 1


def test_visible_shot_count_fails_when_selected_scene_missing():
    """shot_director scene 이 selected_map 에 없으면 AppError fail-fast
    (전체 shot fallback 금지 — Phase 0 와 동일 fail-closed)."""
    shot_director_data = {"scenes": [
        {"scene_index": 1, "shots": [
            {"shot_index": 0, "visible_entity_ids": ["C01"]}]},
        {"scene_index": 2, "shots": [
            {"shot_index": 0, "visible_entity_ids": ["C07"]}]},
    ]}
    with pytest.raises(AppError):
        compute_visible_shot_count_from_checkpoints(shot_director_data, {1: {0}})


def test_build_selected_map_fails_when_shot_selection_missing():
    """shot_selection checkpoint 부재/empty → AppError fail-fast."""
    with pytest.raises(AppError):
        build_selected_map_or_raise(None)
    with pytest.raises(AppError):
        build_selected_map_or_raise({"data": {"scenes": []}})


def test_build_selected_map_ok():
    got = build_selected_map_or_raise(
        {"data": {"scenes": [
            {"scene_index": 1, "selected_shot_indices": [0, 2]}]}})
    assert got == {1: {0, 2}}
```

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

Run: `cd backend && python -m pytest tests/core/steps/test_episode_reference_policy_step.py -v`
Expected: FAIL — `ModuleNotFoundError`

- [ ] **Step 3: step 구현**

`app/core/steps/episode_reference_policy_step.py`:

```python
"""episode_reference_policy StepRunner — scene_detail 직전 reference
necessity manifest 1회 계산 (deterministic, LLM 없음).

설계: docs/reference-necessity/index.html §6.1 / §7 Phase 2.
"""
from __future__ import annotations

import logging
from typing import Any, Dict, Tuple

from app.core.episode_reference_policy import (
    EPISODE_REFERENCE_POLICY_SCHEMA_VERSION,
    compute_episode_reference_policy,
)
from app.core.step_runner import StepRunner

logger = logging.getLogger(__name__)


def _short_id_base(sid: str) -> str:
    return sid.split("O")[0] if sid and "O" in sid else (sid or "")


def build_selected_map_or_raise(shot_selection_cp) -> Dict[int, set]:
    """shot_selection checkpoint → {scene_index: set(selected shot_index)}.

    checkpoint 부재/empty 면 AppError fail-fast — episode_reference_policy 는
    scene_detail 직전 SOT 라 selected_map 불완전 시 전체 shot fallback 금지
    (Phase 0 audit 와 동일 fail-closed 원칙).
    """
    from app.core.errors import AppError
    scenes = (shot_selection_cp or {}).get("data", {}).get("scenes")
    if not scenes:
        raise AppError(
            code="step.no_input",
            message="shot_selection 결과 없음 — episode_reference_policy 는 "
                    "selected_map 없이 진행 불가",
            status_code=400,
        )
    selected_map: Dict[int, set] = {}
    for s in scenes:
        selected_map[s.get("scene_index")] = set(
            s.get("selected_shot_indices", []) or []
        )
    return selected_map


def compute_visible_shot_count_from_checkpoints(
    shot_director_data: Dict[str, Any],
    selected_map: Dict[int, set],
) -> Dict[str, int]:
    """shot_director 의 selected shot 에서 entity 등장 횟수 집계.

    selected_map = {scene_index: set(selected shot_index)}.
    shot_director 의 어떤 scene 이 selected_map 에 없으면 AppError fail-fast
    — 전체 shot fallback 금지 (false-positive reference_required 방지).
    """
    from app.core.errors import AppError
    counts: Dict[str, int] = {}
    for sc in (shot_director_data or {}).get("scenes", []) or []:
        si = sc.get("scene_index")
        if si not in selected_map:
            raise AppError(
                code="step.episode_reference_policy.selected_scene_missing",
                message=f"shot_selection 에 scene {si} 누락 — selected_map "
                        f"불완전 (fallback 금지)",
                status_code=400,
            )
        sel = selected_map[si]
        for sh in sc.get("shots", []) or []:
            if sh.get("shot_index") not in sel:
                continue
            for sid in sh.get("visible_entity_ids", []) or []:
                b = _short_id_base(sid)
                if b:
                    counts[b] = counts.get(b, 0) + 1
    return counts


class EpisodeReferencePolicyStep(StepRunner):
    """Step 21.65: episode reference necessity manifest 계산."""

    def _execute(self, mode="resume") -> Dict[str, Any]:
        from app.core.errors import AppError

        sdir_cp = self._load_prev_checkpoint("shot_director")
        if not sdir_cp or not sdir_cp.get("data", {}).get("scenes"):
            raise AppError(
                code="step.no_input",
                message="shot_director 결과 없음",
                status_code=400,
            )
        shot_director_data = sdir_cp["data"]

        # shot_selection fail-fast — selected_map 불완전 시 전체 shot
        # fallback 금지 (Codex v2 BLOCKING1).
        sel_cp = self._load_prev_checkpoint("shot_selection")
        selected_map = build_selected_map_or_raise(sel_cp)

        visible_shot_count = compute_visible_shot_count_from_checkpoints(
            shot_director_data, selected_map,
        )
        entity_types, variant_pole_short_ids = self._load_entity_signals()

        manifest = compute_episode_reference_policy(
            visible_shot_count=visible_shot_count,
            entity_types=entity_types,
            variant_pole_short_ids=variant_pole_short_ids,
        )

        policy = manifest["policy"]
        text_only_n = sum(1 for p in policy.values() if p["mode"] == "text_only")
        logger.info(
            "episode_reference_policy: %d entities — %d text_only, %d other",
            len(policy), text_only_n, len(policy) - text_only_n,
        )
        return {
            "completed_count": len(policy),
            "applicable_count": len(policy),
            "failed_count": 0,
            "data": manifest,
            "schema_version": EPISODE_REFERENCE_POLICY_SCHEMA_VERSION,
        }

    def _load_entity_signals(self) -> Tuple[Dict[str, str], set]:
        """{short_id: entity_type} + variant pole short_id 집합."""
        from app.core.entity_protection import compute_variant_pole_ids
        from app.models.project import (
            EntityCanon, EntityEpisodeLink, RelationFact, RelationParticipant,
        )

        links = self.db.query(EntityEpisodeLink).filter(
            EntityEpisodeLink.project_id == self.project_id,
            EntityEpisodeLink.episode_id == self.episode_id,
        ).all()
        canon_ids = [lnk.canon_id for lnk in links]
        canons = self.db.query(EntityCanon).filter(
            EntityCanon.id.in_(canon_ids)
        ).all() if canon_ids else []
        entity_types: Dict[str, str] = {}
        short_by_canon: Dict[str, str] = {}
        for c in canons:
            if c.short_id:
                base = c.short_id.split("O")[0]
                entity_types[base] = c.entity_type
                short_by_canon[c.id] = base

        relations = self.db.query(RelationFact).filter(
            RelationFact.project_id == self.project_id
        ).all()
        rel_ids = [r.id for r in relations]
        participants = self.db.query(RelationParticipant).filter(
            RelationParticipant.relation_id.in_(rel_ids)
        ).all() if rel_ids else []
        variant_pole_uuids = compute_variant_pole_ids(
            [{"id": c.id, "entity_type": c.entity_type} for c in canons],
            [{"id": r.id, "relation_family": r.relation_family}
             for r in relations],
            [{"relation_id": p.relation_id, "canon_id": p.canon_id}
             for p in participants],
        )
        variant_pole_short_ids = {
            short_by_canon[u] for u in variant_pole_uuids
            if u in short_by_canon
        }
        return entity_types, variant_pole_short_ids
```

> 실행 에이전트 주의: `StepRunner` 의 `self.db`/`self.project_id`/`self.episode_id`/`self._load_prev_checkpoint` 존재를 `app/core/step_runner.py` 로 확인. `RelationFact`/`RelationParticipant`/`EntityCanon`/`EntityEpisodeLink` 컬럼명(`relation_family`/`relation_id`/`canon_id`/`short_id`/`entity_type`)을 `app/models/project.py` 로 확인. `compute_variant_pole_ids` 의 인자 dict 키는 `entity_protection.py:125` 시그니처와 일치시킬 것.

- [ ] **Step 4: manifest entry 추가**

`app/core/step_manifest.py` `background_prompt`(21.60) 와 `scene_detail`(21.70) 사이에 추가:

```python
    # 21.65 episode_reference_policy — scene_detail 직전 reference necessity
    # manifest. deterministic (LLM 없음). render_prompt_card producer/verify 가
    # 동일 immutable manifest 를 읽어 text-only subject 의 identity policy 를
    # generic_descriptor_allowed 로 다운그레이드 → reference 과생성 차단.
    # 설계: docs/reference-necessity/index.html.
    "episode_reference_policy": {
        "label": "에피소드 참조 정책",
        "category": "analysis",
        "order": 21.65,
        "default_model": "gpt",   # 미사용 (deterministic) — manifest 형식상 필수
        "provider": "openai",
        "depends_on": [
            "shot_director", "shot_selection", "shot_validator",
            "entity_merge", "entity_relation",
        ],
        "fan_out": False,
        "applicability": "always",
        "step_type": "transform",
        "lifecycle": "active",
        "schema_version": 1,
    },
```

> `shot_dependency` manifest entry 와 비교 — deterministic step 의 `default_model`/`provider` 처리 방식을 동일하게 맞출 것 (`shot_dependency` 가 둘 다 없으면 생략).

- [ ] **Step 5: registry 등록**

`app/core/steps/__init__.py` import 부에 추가:

```python
from app.core.steps.episode_reference_policy_step import EpisodeReferencePolicyStep
```

`STEP_CLASSES` dict 에 추가 (`scene_detail` 근처):

```python
    "episode_reference_policy": EpisodeReferencePolicyStep,
```

- [ ] **Step 6: 테스트 + manifest 정합 확인**

Run: `cd backend && python -m pytest tests/core/steps/test_episode_reference_policy_step.py -v`
Expected: PASS (5 tests — 집계 2 + fail-fast 3)

Run: `cd backend && python -c "from app.core.step_manifest import STEP_MANIFEST; from app.core.steps import STEP_CLASSES; assert 'episode_reference_policy' in STEP_MANIFEST and 'episode_reference_policy' in STEP_CLASSES; print('manifest+registry OK')"`
Expected: `manifest+registry OK`

Run: `cd backend && python -m pytest tests/ -k "manifest" -q`
Expected: PASS (회귀 0)

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

```bash
git add backend/app/core/steps/episode_reference_policy_step.py backend/app/core/step_manifest.py backend/app/core/steps/__init__.py backend/tests/core/steps/test_episode_reference_policy_step.py
git commit -m "Phase2: EpisodeReferencePolicyStep + manifest order 21.65 + registry"
```

### Task 10: render_prompt_card — episode_reference_policy overlay 적용

**Files:**
- Modify: `app/core/steps/render_prompt_card.py`
- Test: `tests/core/steps/test_render_prompt_card_episode_policy.py`

- [ ] **Step 1: 수정 대상 읽기**

`render_prompt_card.py` line 3452–3605 (`build_render_prompt_card` 전체), line 88–92 (subject_reference_policy import) 를 직접 읽는다. 기존 `tests/core/steps/` 의 `build_render_prompt_card` 테스트에서 호출 fixture 형태를 확보.

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

`tests/core/steps/test_render_prompt_card_episode_policy.py`:

```python
"""render_prompt_card episode_reference_policy overlay — text_only subject
required_ref 억제 + base_id_required materialization reconciliation."""
from app.core.steps.render_prompt_card import (
    build_render_prompt_card, compute_card_hash,
)

# _minimal_card_kwargs(visible, srp_items): 기존 build_render_prompt_card
# 테스트 fixture 를 복사해 구성. staging 에 subject_reference_policy=srp_items
# 포함 필수. 아래 helper 는 그 fixture 위에 작성한다.


def _ref_ids(card):
    return [r.get("id") for r in card["asset_requirements"]["required_refs"]]


def test_text_only_char_suppressed_from_required_refs():
    """C08=text_only (staging 기본 id_and_outlook_required) → required_ref 제거."""
    manifest = {"schema_version": 1, "policy": {
        "C08": {"mode": "text_only", "entity_type": "character",
                "reason": "low freq", "visible_shot_count": 1,
                "provisional": False}}}
    common = _minimal_card_kwargs(visible=["C01O00", "C08O00"], srp_items=[])
    card_off = build_render_prompt_card(**common, episode_reference_policy=None)
    card_on = build_render_prompt_card(
        **common, episode_reference_policy=manifest)
    assert any("C08" in str(i) for i in _ref_ids(card_off))
    assert not any("C08" in str(i) for i in _ref_ids(card_on))
    assert any("C01" in str(i) for i in _ref_ids(card_on))


def test_base_id_required_preserved_despite_text_only():
    """materialization reconciliation: text_only 여도 staging 이 C08 을
    base_id_required 로 지정했으면 required_ref(kind=character) 유지."""
    manifest = {"schema_version": 1, "policy": {
        "C08": {"mode": "text_only", "entity_type": "character",
                "reason": "low freq", "visible_shot_count": 1,
                "provisional": False}}}
    srp = [{"subject_id": "C08", "policy_type": "identity_reference",
            "policy": "base_id_required", "reason": "explicit partial-frame"}]
    common = _minimal_card_kwargs(visible=["C08O00"], srp_items=srp)
    card = build_render_prompt_card(**common, episode_reference_policy=manifest)
    assert any(r.get("id") == "C08" and r.get("kind") == "character"
               for r in card["asset_requirements"]["required_refs"])


def test_none_manifest_is_noop():
    common = _minimal_card_kwargs(visible=["C01O00"], srp_items=[])
    h1 = compute_card_hash(build_render_prompt_card(**common))
    h2 = compute_card_hash(
        build_render_prompt_card(**common, episode_reference_policy=None))
    assert h1 == h2
```

> 실행 에이전트 주의: `_minimal_card_kwargs` 는 plan 미제공 — `tests/core/steps/` 의 기존 `build_render_prompt_card` 테스트(FINDING 9 / C10 관련)를 grep 해 실제 필수 인자 전부를 채우는 helper 로 작성. staging dict 에 `subject_reference_policy` 키 필수. `C08` 이 srp_items 에 없으면 기본 `id_and_outlook_required` 로 취급되어 다운그레이드 효과가 보인다.

- [ ] **Step 3: 테스트 실패 확인**

Run: `cd backend && python -m pytest tests/core/steps/test_render_prompt_card_episode_policy.py -v`
Expected: FAIL — `unexpected keyword argument 'episode_reference_policy'`

- [ ] **Step 4: 시그니처 + import**

`build_render_prompt_card` 시그니처 (line ~3452–3481) 마지막 param 다음에 추가:

```python
    episode_reference_policy: Optional[Dict[str, Any]] = None,
) -> Dict[str, Any]:
```

line 88–92 의 `from app.core.subject_reference_policy import (...)` 에 `apply_episode_reference_policy_downgrade` 추가.

- [ ] **Step 5: downgrade 적용**

`build_render_prompt_card` 내부, `apply_screen_presence_downgrade` 호출 블록 (line ~3555–3558) **직후**, `normalize_subject_reference_policy_items` 호출 **직전** 에 추가:

```python
    # Phase 2 — episode_reference_policy overlay: manifest 가 text_only 로
    # 판정한 character subject 의 identity policy 를 generic_descriptor_allowed
    # 로 결정론적 다운그레이드. screen_presence downgrade 와 동일 메커니즘 —
    # builder 내부 실행이므로 producer / verify recompute hash 동일.
    if episode_reference_policy is not None:
        from app.core.episode_reference_policy import extract_text_only_subjects
        _text_only = {
            sid: r for sid, r in
            extract_text_only_subjects(episode_reference_policy).items()
            if sid in visible_bases
        }
        if _text_only:
            raw_srp_items = apply_episode_reference_policy_downgrade(
                raw_srp_items, _text_only, where=where_srp,
            )
```

> `visible_bases` 교집합 필수 — manifest 는 catalog 전체 기반이므로, 다운그레이드 대상을 이 shot 의 visible subject 로 한정해야 `normalize_subject_reference_policy_items` 의 `unknown_subject` raise 를 피한다.

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

Run: `cd backend && python -m pytest tests/core/steps/test_render_prompt_card_episode_policy.py -v`
Expected: PASS (3 tests)

Run: `cd backend && python -m pytest tests/core/steps/ -k "render_prompt_card or card_hash" -q`
Expected: PASS (회귀 0)

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

```bash
git add backend/app/core/steps/render_prompt_card.py backend/tests/core/steps/test_render_prompt_card_episode_policy.py
git commit -m "Phase2: render_prompt_card episode_reference_policy overlay downgrade"
```

### Task 11: detail_steps — manifest threading

**Files:**
- Modify: `app/core/dto/scene_analysis.py`
- Modify: `app/core/steps/scene_context_loader.py`
- Modify: `app/core/steps/detail_steps.py`
- Test: `tests/core/steps/test_detail_steps_episode_policy.py`

- [ ] **Step 1: SceneAnalysisContext 필드 추가**

`app/core/dto/scene_analysis.py` `SceneAnalysisContext` dataclass 마지막 필드 다음에 추가:

```python
    # ── episode_reference_policy (Phase 2) ──────────────────
    episode_reference_policy: Optional[Dict[str, Any]] = None
```

(`Optional`/`Dict`/`Any` 가 파일 상단 typing import 에 있음 — 확인.)

- [ ] **Step 2: SceneContextLoader 로더 추가**

`scene_context_loader.py` `load_all()` 마지막 `ctx.<field> = ...` 다음에:

```python
        ctx.episode_reference_policy = self._load_episode_reference_policy()
```

loader 메서드 추가 (`_load_staging_map` 근처):

```python
    def _load_episode_reference_policy(self) -> Optional[Dict[str, Any]]:
        """episode_reference_policy checkpoint → manifest dict (or None).

        부재 시 None — build_render_prompt_card 가 None 을 no-op 처리.
        """
        cp = self.runner._load_prev_checkpoint("episode_reference_policy")
        if not cp:
            return None
        return cp.get("data")
```

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

`tests/core/steps/test_detail_steps_episode_policy.py`:

```python
"""detail_steps — episode_reference_policy 가 card inputs 로 threading."""
from app.core.dto.scene_analysis import SceneAnalysisContext
from app.core.steps.detail_steps import _derive_card_inputs_from_ctx


def test_derive_card_inputs_includes_episode_reference_policy():
    manifest = {"schema_version": 1, "policy": {
        "C08": {"mode": "text_only", "entity_type": "character",
                "reason": "x", "visible_shot_count": 1, "provisional": False}}}
    ctx = SceneAnalysisContext()
    ctx.episode_reference_policy = manifest
    inputs = _derive_card_inputs_from_ctx(
        ctx=ctx, seg={"scene_index": 1}, shot_info=None)
    assert inputs.get("episode_reference_policy") == manifest


def test_derive_card_inputs_none_when_absent():
    ctx = SceneAnalysisContext()
    inputs = _derive_card_inputs_from_ctx(
        ctx=ctx, seg={"scene_index": 1}, shot_info=None)
    assert inputs.get("episode_reference_policy") is None
```

- [ ] **Step 4: 테스트 실패 확인**

Run: `cd backend && python -m pytest tests/core/steps/test_detail_steps_episode_policy.py -v`
Expected: FAIL — 첫 테스트, 키 없음

- [ ] **Step 5: _derive_card_inputs_from_ctx 에 키 추가**

`detail_steps.py` `_derive_card_inputs_from_ctx` (line 564–750) 반환 dict 에 추가:

```python
        "episode_reference_policy": getattr(
            ctx, "episode_reference_policy", None),
```

- [ ] **Step 6: _collect_card_inputs 통과 확인**

`_collect_card_inputs` (line 753–869) 가 `_derive_card_inputs_from_ctx` 결과 dict 를 통째 보존하는지 확인. 통째 쓰면 추가 작업 없음 — 3 builder 지점(916/963/2891)이 `builder_inputs = {k:v for k,v in card_inputs if k!="ctx"}` 로 자동 전파. legacy/override path 가 키별 명시 조립이면 동일하게 추가:

```python
        "episode_reference_policy": getattr(
            ctx, "episode_reference_policy", None) if ctx is not None else None,
```

- [ ] **Step 7: 3 builder 지점 검증**

line ~916 / ~963 / ~2891 의 card_inputs 가 모두 `_collect_card_inputs` 경유인지 확인. 경유하면 코드 변경 0. 직접 dict 조립 지점이 있으면 `episode_reference_policy` 키 추가. fallback ctx (line 916) 도 `SceneAnalysisContext` 면 dataclass 기본값 `None` 으로 안전.

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

Run: `cd backend && python -m pytest tests/core/steps/test_detail_steps_episode_policy.py -v`
Expected: PASS (2 tests)

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

```bash
git add backend/app/core/dto/scene_analysis.py backend/app/core/steps/scene_context_loader.py backend/app/core/steps/detail_steps.py backend/tests/core/steps/test_detail_steps_episode_policy.py
git commit -m "Phase2: thread episode_reference_policy manifest through ctx to card builders"
```

### Task 12: scene_detail manifest 의존성 + schema bump

**Files:**
- Modify: `app/core/step_manifest.py`
- Modify: `app/core/steps/detail_steps.py`

- [ ] **Step 1: scene_detail depends_on 추가**

`step_manifest.py` `scene_detail` entry `depends_on` 에 `"episode_reference_policy"` 추가.

- [ ] **Step 2: SCENE_DETAIL_SCHEMA_VERSION bump 12 → 13**

`detail_steps.py` line 106:

```python
SCENE_DETAIL_SCHEMA_VERSION = 13  # 2026-05-23 (reference-necessity Phase 2): render_prompt_card 가 episode_reference_policy overlay 반영 — text_only subject 의 id_policy/asset_requirements 변동 + card hash semantics 변경 → 구 cp invalidation.
```

`step_manifest.py` `scene_detail` entry `"schema_version": 12` → `13`, 주석에 bump 사유 한 줄 추가.

- [ ] **Step 3: manifest 정합 확인**

Run: `cd backend && python -c "from app.core.step_manifest import STEP_MANIFEST; from app.core.steps.detail_steps import SCENE_DETAIL_SCHEMA_VERSION; assert STEP_MANIFEST['scene_detail']['schema_version'] == SCENE_DETAIL_SCHEMA_VERSION == 13; assert 'episode_reference_policy' in STEP_MANIFEST['scene_detail']['depends_on']; print('scene_detail manifest OK')"`
Expected: `scene_detail manifest OK`

- [ ] **Step 4: 전체 backend 회귀**

Run: `cd backend && python -m pytest tests/ -q`
Expected: 신규 테스트 전부 PASS. 기존 실패는 pre-existing baseline 과 대조 (회귀 0). 실패 발견 시 정확한 evidence 캡처 → pre-existing vs new 분류 → Codex entry-sanity.

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

```bash
git add backend/app/core/step_manifest.py backend/app/core/steps/detail_steps.py
git commit -m "Phase2: scene_detail depends_on episode_reference_policy + schema 12->13"
```

### Task 13: Codex 코드 리뷰 (구현 완료 후, E2E 전)

- [ ] **Step 1: range diff 준비**

```bash
git log --oneline origin/main..HEAD
git diff origin/main..HEAD --stat
```

- [ ] **Step 2: Codex 리뷰 요청**

tmux-bridge MCP raw workflow 로 Codex 세션에 Phase 0+1+2 전체 range 리뷰 요청 (`tmux_read` → `tmux_type` body-only → `tmux_read` → `tmux_keys ["Enter"]` → `tmux_read`). 메시지에 `claude 에서 보냄` 표기, 끝에 `claude mcp 로 응답해달라` 추가. 리뷰 항목: narrow 보호 집합 안전성, count 정정 blast radius, card hash drift(producer/verify manifest 동일성), schema bump, overlay visible_bases 교집합, scenario-generalized 여부.

- [ ] **Step 3: 리뷰 반영**

Codex 지적을 superpowers:receiving-code-review 로 검증 후 반영. 반영 시 신규 커밋 (amend 금지).

---

## 검증 (E2E)

### Task 14: Phase 0 GATE 실행 + E2E acceptance

- [ ] **Step 1: Phase 0 GATE 실행**

> Phase 1/2 코드 머지 전에 수행 (Task 1–2 의 audit 도구로). fresh E2E checkpoint 가 있는 project/episode 에 대해:

```bash
cd backend && python -m scripts.reference_necessity_audit <project_id> <episode_id>
```

Expected: `GATE: PASS`. FAIL 시 — 입력 무결성 문제면 fresh E2E 재실행, required_refs gap 이면 보강 후 재설계. Codex 보고 후 진행 결정.

- [ ] **Step 2: fresh full E2E 실행**

새 project/episode 로 기획안+1부 분석 → 이미지까지 full E2E. `feedback_e2e_pipeline_monitoring` 모니터링 규칙 준수 (모든 종료 경로 regex + 이미지 step mtime liveness).

- [ ] **Step 3: E2E acceptance 검증**

설계 §8.2:
- reference asset 수 감소 — selected-shot T2I 사용량 0 인 character/prop 의 reference 미생성.
- `scene_image_pipeline` 이 `missing required ref` 없이 완료.
- 한 컷 주변 인물이 ID 오염 없이 자연어로 렌더 (text_only subject 의 t2i_prompt 에 C## 없음).
- `ref_low_freq_skip.json` 이 reasoned object(v2) — skip 결정에 reason 포함.
- `episode_reference_policy/manifest.json` 1회 생성, text_only/reference_required/reference_candidate 분류 포함.

- [ ] **Step 4: Codex 최종 리뷰**

E2E 결과 Codex MCP read-only 리뷰. APPROVED 시 push 승인 요청.

- [ ] **Step 5: closure memory + push**

Codex `APPROVED_FOR_PUSH` 후 origin/main push. closure memory 작성.

---

## Self-Review (작성자 체크리스트)

**1. Spec coverage:**
- 설계 §7 Phase 0 (관찰+차집합 감사, fail-closed GATE) → Task 1–2 ✅
- 설계 §7 Phase 1 (entity_protection split, ref_low_freq_skip reasoned object) → Task 3–4, 6 ✅
- BLOCKING2 `t2i_appearance_count` selected-shot 정정 → Task 5 ✅
- 설계 §7 Phase 2 (episode_reference_policy step, render_prompt_card overlay, scene_detail threading) → Task 7–12 ✅
- 설계 §10.1 character threshold — 이 step 은 scene_detail 전이라 `required_ref_count` 미지(circularity). character 게이트 = `visible_shot_count>=2 OR variant`. `required_ref_count>=1` 절은 downstream 집행: scene_detail → required_ref → Phase 1 narrow 보호(`_collect_reference_required_ids`) → materialization 보호. Task 8 `test_downgrade_preserves_base_id_required` + Task 10 `test_base_id_required_preserved_despite_text_only` 가 reconciliation 검증 ✅
- 설계 §10.2 동물/비인간 동일 규칙 (C## prefix 기준) → Task 7 ✅
- 설계 §10.3 prop provisional — Task 7 에서 prop mode = `reference_candidate`(생성 확정 아님)/`text_only`, 최종 게이트 = required_ref_count (Phase 1 narrow 보호가 집행) ✅
- 설계 §10.4 Phase 0 audit FAIL fallback → Task 14 Step 1 GATE 명시 ✅
- 설계 drift 방지 (immutable manifest, producer/verify 동일 source) → Task 11 (ctx 단일 source threading) ✅

**2. Placeholder scan:** `_minimal_card_kwargs`(Task 10) 는 의도적 미제공 — 기존 fixture 복사 지시 명시. step_run 모델명(Task 2) 도 grep 지시 명시. 그 외 TBD/TODO 없음.

**3. Type consistency:** manifest shape `{"schema_version": int, "policy": {short_id: {"mode", "reason", "entity_type", "visible_shot_count", "provisional"}}}` — Task 7 생산, `extract_text_only_subjects` 소비(`text_only` mode 만), Task 10 render_prompt_card 소비, Task 11 ctx threading 일관. `mode` enum = `text_only`/`reference_required`(char)/`reference_candidate`(prop). `apply_episode_reference_policy_downgrade(raw_items, text_only_subjects: dict, *, where)` — Task 8 정의, Task 10 호출 일관. `compute_episode_reference_policy(*, visible_shot_count, entity_types, variant_pole_short_ids)` — Task 7 정의, Task 9 호출 일관.

**Codex v1 리뷰 반영 완료:** BLOCKING1(fail-closed GATE), BLOCKING2(count 정정 Task 5), IMPORTANT1(required_ref_count circularity 명시 + reconciliation 테스트), Q3(prop `reference_candidate`). 질문 답변 반영: order 21.65 / schema bump 12→13 / entity_filter 제외 — 전부 Codex 확인 완료.

**Codex v2 재리뷰 반영 완료:** BLOCKING1-v2(Task 9 runtime fail-fast — `episode_reference_policy` step 이 selected_map 누락 시 전체 shot fallback 하던 불일치 제거, `build_selected_map_or_raise` + `compute_visible_shot_count_from_checkpoints` AppError fail-fast, 테스트 3개), IMPORTANT1-v2(Task 2 DB step_run 검사 raw SQL 로 구체화 — optional 제거, 필수 GATE).