# Background Pipeline Slice Experiment (W1) — Implementation Plan

> **For agentic workers:** REQUIRED SUB-SKILL: Use `superpowers:subagent-driven-development` (Opus 4.7) for implementation. Steps use checkbox (`- [ ]`) syntax. **NO commits, NO pushes** in this plan — user has explicitly forbidden them. Steps end at "tests pass".

**Goal:** 기존 production background/image/reference/prompt 파이프라인에 연결 가능한 **최소 dry-run slice** 를 만든다 — 한 episode 의 selected shot 을 입력받아 LLM 이 raw background intent 만 emit 하고, deterministic adapter 가 production helper (`bg_catalog`) 로 `L##B##` unit_id 를 부여하여 production master_plan-호환 shape 의 adapter plan + dry-run render payload + reference attach plan + compatibility report 를 산출한다. production code 0 수정, DB write 0, image API 호출 0.

**Architecture:** 2-layer plan:
- `llm_raw_plan.json` = LLM structured output. building_groups / raw background intents (loc_id, space_key_hint, time_phase, state_class (production STATE_CLASS_ENUM 강제), sub_location_label, state_label_raw, parent_intent_ref, fp_intent_key) / **shot_binding_intents (shot_key SOT — `applies_to_shots` 는 LLM raw 에 없음, code 가 derive)** / prompt intents / open_risks. **bg_id 는 emit 하지 않음.**
- `production_adapter_plan.json` = code 가 `bg_catalog.assign_bg_ids` + `build_shot_background_map` 등 production helper 로 빌드한 master_plan-호환 shape. **`background_catalog: Dict[bg_id, entry]` (SOT)**, `shot_background_map: Dict[shot_key, bg_id]`, `bg_catalog_hash`, `shot_binding_hash`, `gen_order: List[fp_id | bg_id]` (topo), `floor_plans: List`.

deterministic code 는 ID 형식 / coverage / DAG / shape / source verbatim / no-write/no-call sentinel 만 검증. semantic regex/substring/boundary 금지. 사람 승인/결정 보류 enum 금지 — fallback 으로도 못 만들면 `validation_failed` + exit 1.

**Tech Stack:** Python 3 / SQLAlchemy (read-only) / litellm (gemini-3.5-flash) / jsonschema / pytest. 실행 CWD = `backend/`.

**범위 밖 (Non-goals):** 완벽한 background planner 만들기, 도면/문/창/가구 개수 확정, scene-image 직접 호출, production code 수정, write-back to production checkpoints, 사람 승인 flow, sample-specific scenario rule (specific location label / character name / room label / prop name 을 methodology 로 박는 것), sibling experiment 결과 import.

**개정 이력:**
- v1 (2026-05-24): 초안. Codex design 의논 Q1-Q8 합의 (APPROVED_FOR_PLAN with constraints) 및 사용자 W1 범위 축소 지시 (minimal slice / ~25 test / ~10 touchpoint / 필수 validator 만) 반영.
- v2 (2026-05-24): Codex 1차 plan 리뷰 NEEDS_REVISION 4 BLOCKING + 3 IMPORTANT + 3 MINOR 반영.
- v5 (2026-05-24): Codex 4차 plan 리뷰 NEEDS_NARROW_REVISION 1 BLOCKING (+ 1 hallucinated BLOCKING). **BLOCKING 1 (Codex misread)**: `--model` 중복 등록 — grep 결과 line 220 단일 occurrence (다른 2 곳은 prompt text + run command 인용). 코드 무변경. **BLOCKING 2**: test import path 가 `from backend.scripts.experiment_background_pipeline_slice import ...` 인데 pytest CWD=`backend/` 이라 `backend.scripts.*` resolve 실패. 기존 `backend/tests/scripts/test_experiment_*.py` convention 따라 `_SCRIPTS = _REPO_ROOT / "backend" / "scripts"; sys.path.insert(0, str(_SCRIPTS))` + bare `from experiment_background_pipeline_slice import ...` 로 변경. 13 import site 일괄 patch.
- v4 (2026-05-24): Codex 3차 plan 리뷰 NEEDS_NARROW_REVISION 1 BLOCKING (+ 1 hallucinated BLOCKING). **BLOCKING 1**: parent dedup conflict 에 None-vs-parent 케이스 누락 — `parent_by_bg` 에 parent 있을 때만 add 했음. 새 `parent_candidates_by_bg` 가 None 도 candidate 로 기록, set 크기 2+ 면 `dedup_parent_conflict`. 기존 test 에 case B (None vs k1) 추가 — 별도 test 추가 X, count 32 유지. **BLOCKING 2 (Codex misread)**: plan 의 topo loop 에 중복 `raise ValueError(` 없음 (grep 검증: 5 occurrences 모두 단독, line 1247-1249 단일 raise). 코드 무변경.
- v3 (2026-05-24): Codex 2차 plan 리뷰 NEEDS_NARROW_REVISION 2 BLOCKING + 2 IMPORTANT 반영. **BLOCKING 1**: `state_class` enum 이 production `STATE_CLASS_ENUM` (`bg_state_vocab.py:26-38` — 11 값: `normal/quiet/busy/busy_exit/ransacked/clean_after/blood_scene/intrusion/arrival/evidence_display/dream_or_vision_state`) 과 불일치. LLM schema enum 으로 강제 + fake fixture state_class `clean`→`normal`, `dirty`→`ransacked`. **BLOCKING 2**: `_build_adapter_plan` 의 parent_intent_ref 누적 시 `set` 1개 초과면 `dedup_parent_conflict` fail-fast, self-cycle 도 fail. graph 검증 위임 금지. **IMPORTANT 1**: stale `background_catalog[]` list 표기 / `applies_to_shots` LLM raw 잔재 wording 정리. **IMPORTANT 2**: plan 본문 (revision history 자체 제외) 의 sample literal 예시 generic 화 (자세한 단어는 자체 검토 통한 fixture data 보존만). **BLOCKING 1**: `bg_catalog.assign_bg_ids` / `build_shot_background_map` / `compute_bg_catalog_hash` 의 catalog 는 `Dict[str, Dict]` (bg_id → entry) — list 아님. plan 전체 dict 통일. **BLOCKING 2**: location profile shape — production 은 `kind` (single_space / multi_space) + `allowed_space_keys`, `space_kind` 아님. real loader 는 `validate_location_space_profile` 사용. **BLOCKING 3**: `readiness_policy` enum = `block_if_missing` / `skipped_by_policy` / `not_applicable` (production render_prompt_card.py:462-466). `use_attached` invalid → `block_if_missing` 으로 교체. **BLOCKING 4**: prompt snippet 의 `L05/옥탑방` literal 제거 → generic phrasing (자기 fixture-swap test 와 충돌). **IMPORTANT 1**: shot id SOT = `shot_key` (composite). row UUID = `row_id`. LLM schema / shot_binding_intents / shot_background_map / reference_input_plan / coverage 모두 `shot_key`. **IMPORTANT 2**: `shot_binding_intents` 가 SOT — code 가 `applies_to_shots` derive. `raw_background_intents.applies_to_shots` 필드 제거. **IMPORTANT 3**: DB write sentinel = static guard + SQLAlchemy Session-event 병행. **MINOR**: touchpoint observed_contract dict 표기 정리, HTML preview slice 주석, implementation 크기 hard cap 제거.

---

## 사전 컨텍스트 — 4개 Opus 4.7 explorer 결과 (요약)

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

### Production background pipeline 흐름
1. `app/core/steps/background_classify_step.py:34 BackgroundClassifyStep` (order 19.51) — `data.building_groups[{group_id, members[], anchor_loc, kind, rationale}]`
2. `app/core/steps/background_master_plan_step.py:43 BackgroundMasterPlanStep` (order 19.52, SCHEMA_VERSION=3, PROMPT_VERSION="5.202605201759") — per chain_bg group: 1 LLM → `data.plans[group_id]={floor_plans[{fp_id, loc_id, space_key_hint, depends_on_fp, ...}], backgrounds[{bg_id, loc_id, space_key_hint, time_phase, state_class, depends_on_fp, depends_on_bg, applies_to_shots, sub_location_label, state_label_raw}], gen_order[]}` + sibling `data.background_catalog`, `data.shot_background_map`, `data.bg_catalog_hash`, `data.shot_binding_hash`. post-processing `_d6_post_process` 호출 `bg_catalog.assign_bg_ids`.
3. `app/core/steps/floor_plan_prompt_step.py:34 FloorPlanPromptStep` (order 19.61) — per fp t2i_prompt + camera_recommendations
4. `app/core/steps/floor_plan_render_step.py:91 FloorPlanRenderStep` (order 21.55) — PNG via gpt-image-2, `ImageAsset(asset_type='floor_plan', variant_type=fp_id)`
5. `app/core/steps/background_prompt_step.py BackgroundPromptStep` (order 21.60) — per bg t2i_prompt + shot_guides
6. `app/core/steps/background_render_step.py:89 BackgroundRenderStep` (order 24.72) — PNG via gpt-image-2, `ImageAsset(asset_type='chain_bg', variant_type=bg_id)`
7. `app/core/steps/episode_reference_policy_step.py:99 EpisodeReferencePolicyStep` (order 21.65) deterministic policy manifest
8. `app/core/steps/detail_steps.py:1175 SceneDetailStep` (order 21.70, SCHEMA_VERSION=13) — emits `t2i_variations[].reference_phrase_kinds` + `render_prompt_card.asset_requirements.required_refs[]`
9. `app/core/steps/image_steps.py:508 SceneImagePipelineStep` (order 25) — consumes `render_prompt_card` + `background_chain_bg_map`, fires `generate_and_validate_scene` (gemini-3.1-flash-image-preview)

### Pure deterministic helpers (실험 import 가능, side effect 0)
- `app/core/bg_catalog.py`: `assign_bg_ids(prev_catalog, new_intents, location_profiles)` (line 135), `build_shot_background_map(catalog)` (line 307), `compute_semantic_key(loc_id, space_key, time_phase, state_class)` (line 80), `normalize_space_key(loc_id, hint, profile)` (line 49), `compute_bg_catalog_hash(catalog)` (line 274), `compute_shot_binding_hash(shot_background_map)` (line 291). 예외: `SemanticKeyError, FpLinkMismatchError, ShotBindingError`.
- `app/core/bg_state_vocab.py`: `BG_ID_RE = re.compile(r'^L\d{2,3}B\d{2,3}$')` (line 64), `STATE_CLASS_ENUM` (line 26), `LOCATION_SPACE_KEY_VOCAB`, `LOCATION_SPACE_KIND_VOCAB`, `validate_location_space_profile(profile)` (line 145).
- `app/core/ref_contract_validator.py`: `validate_attached_refs(rpc, labeled_refs, attached_meta, prompt, is_close_framing, *, chain_bg_lookup, reference_phrase_kinds)` (line 148), `RefContractError` (line 39, HTTP 422). 6 sequential checks (length / rpc shape / character_outlook strict / prop strict / character base strict / background lineage + close-framing skip + readiness / phantom guard).

### LLM client convention (실험)
- 모든 prior experiment (`experiment_background_semantic_extractor.py`, `experiment_background_topology_planner.py`, `experiment_background_place_grouping.py`, `experiment_background_spatial_decision.py`) 는 `litellm` 직접 호출 — `gemini/<model>` prefix, `response_format={"type":"json_object"}`. 기본 모델 `gemini-3.5-flash`. `GEMINI_API_KEY` 또는 `GOOGLE_API_KEY` env 필요. production wrapper `app.modules.llm.llm_client.call_structured` **사용 안 함**.

### Run-dir / run_meta convention (실험 기존)
- run dir = `scripts_output/<experiment_name>/<KST YYYYMMDD_HHMM>_<6hex>/`
- `run_meta.json` 필드: `run_id, plan_version, generated_at, model, args, outputs[]`
- `_load_backend_env()` = backend/.env 를 os.environ 으로 manual parse (SessionLocal 연결용)
- `_REPO_ROOT = Path(__file__).resolve().parents[2]`, `_BACKEND_ROOT = _REPO_ROOT / "backend"`, sys.path prepend

### DB read-only source data 후보
- project `6cb862d9-590c-4dce-86e6-d10c2977db19` / episode `08ad2cd3-3e96-4d84-808f-869ee628473c` — 61 selected shots, 2026-05-22 reference-necessity E2E clean (**default fixture**)
- project `76212b49-bf96-45fb-8c56-cd8d8ae02dda` / episode `f44339e6-1bd1-4d10-8a8f-2e30e5a1d36d` — 61 selected (alternate fixture for swap test)
- project `8d56bc5d-89eb-4733-9890-cbec35dd358b` / episode `1458fcc5-fb7d-407c-aa45-bc41bf98bba7` — 66 selected
- 읽을 모델: `ProjectRegistry`, `Episode`, `SceneStill` (is_selected=True), `EntityCanon` (location entries, metadata_json.location.space_profile)
- checkpoint dir = `projects/{pid}/checkpoints/episodes/{eid}/{step_id}/manifest.json` — **diagnostic only**, SOT 아님

### Import whitelist (W1)
| 허용 | 금지 |
|------|------|
| `app.core.database.SessionLocal` | step `_execute()` 호출 |
| ORM models read-only (ProjectRegistry/Episode/SceneStill/EntityCanon) | `background_*_step` / `floor_plan_*_step` / `scene_*_step` 인스턴스화·실행 |
| `app.core.bg_catalog.*` | image client / fal helpers / gpt-image / `gemini_image_client` import |
| `app.core.bg_state_vocab.*` (`BG_ID_RE` 포함 — ID-format regex 는 허용) | `app.modules.llm.llm_client.call_structured` |
| `app.core.ref_contract_validator.validate_attached_refs`, `RefContractError` | semantic regex/substring/boundary matching (글자 단위) |
| `litellm` (generate mode 한정) | DB write (INSERT/UPDATE/DELETE) |

`BG_ID_RE` 같은 production ID-format regex import 는 ID 형식 검증 용도면 허용. 금지는 LLM 결정해야 할 **semantic extraction/grouping** 을 글자 패턴으로 하는 것.

---

## File Structure

| 파일 | 종류 | 책임 |
|------|------|------|
| `backend/scripts/experiment_background_pipeline_slice.py` | 신규 | CLI / run_dir / 모든 phase 로직 (~1500 lines 목표 이내) |
| `backend/tests/scripts/test_experiment_background_pipeline_slice.py` | 신규 | TDD test (~25개) |
| `scripts_output/background_pipeline_slice_experiment/<run_id>/source_bundle.json` | 산출 | 선택 shot + entity_canon location verbatim |
| `…/production_pipeline_map.json` | 산출 | 핵심 ~10 touchpoint 정적 dict (path+symbol verified) |
| `…/llm_raw_plan.json` | 산출 | LLM structured output (또는 placeholder dry-run) |
| `…/production_adapter_plan.json` | 산출 | code-built master_plan-호환 shape |
| `…/generation_payload_dry_run.json` | 산출 | per-shot 최소 dry-run payload + unmapped_fields |
| `…/reference_input_plan.json` | 산출 | per-shot `('background', bg_id)` attach plan |
| `…/pipeline_compatibility_report.json` | 산출 | 9 필수 invariant 결과 |
| `…/index.html` | 산출 | diagnostic (unit count / shot coverage / ref chain / validation / payload sample) |
| `…/run_meta.json` | 산출 | run_id / plan_version / model / args / outputs / status |
| `…/llm_raw_plan_quarantined.json` | 산출 (조건부) | validation fail 시 격리본 |

---

## Acceptance Gates (W1 필수 9 invariant)

1. **production_diff_zero**: `git diff --stat backend/app backend/alembic` 빈 출력.
2. **db_write_zero**: SQLAlchemy event listener (after_insert/update/delete) 카운트 = 0.
3. **image_call_zero**: `gemini_image_client` / `fal_angle_helpers` / `gpt-image-2` 모듈 import 자체 없음 (import sentinel 테스트).
4. **source_verbatim**: `source_bundle.json` 안의 scene_still field SHA256 가 원본 DB row 와 일치.
5. **all_selected_shots_covered**: 모든 `scene_still.is_selected=True` shot 이 `shot_bindings` 에 정확히 1번 등장.
6. **bg_id_assigned_by_production_helper**: `production_adapter_plan.background_catalog` (Dict[bg_id, entry]) 의 모든 key (bg_id) 가 `BG_ID_RE` 통과, `assign_bg_ids` deterministic 재실행 시 동일 `bg_catalog_hash`.
7. **shot_background_map_n_to_1**: `build_shot_background_map` 호출 성공 (N:1 위반 없음).
8. **reference_chain_acyclic**: `production_adapter_plan.gen_order` 가 unit DAG topo order, 모든 unit_id 존재 + cycle 없음.
9. **ref_contract_dry_run_pass**: 모든 shot 에 대해 `validate_attached_refs(minimal_rpc, [(label, b"")], [('background', bg_id)], prompt, is_close_framing=False, chain_bg_lookup=lambda x: loc_id, reference_phrase_kinds=['background'])` 가 `RefContractError` 없이 통과. `minimal_rpc.asset_requirements.readiness_policy='block_if_missing'` (production enum: `block_if_missing` / `skipped_by_policy` / `not_applicable` — `render_prompt_card.py:462-466` SOT) + `forbidden_refs=[]`.

**합격**: 9 invariant 전부 PASS + production_pipeline_map 의 모든 entry path/symbol 존재 + selected shots > 0.
**실패**: 1개라도 FAIL → `run_status="validation_failed"`, exit code 1, `llm_raw_plan_quarantined.json` 생성, run_meta `failed_invariants[]` 채움.

---

## Phase 1 — Script Skeleton (3 tests)

### Task 1: CLI + run_dir + run_meta 초기화

**Files:**
- Create: `backend/scripts/experiment_background_pipeline_slice.py`
- Create: `backend/tests/scripts/test_experiment_background_pipeline_slice.py`

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

```python
# backend/tests/scripts/test_experiment_background_pipeline_slice.py
"""W1 experiment_background_pipeline_slice — minimal slice TDD.

기존 backend/tests/scripts/* convention 과 동일:
- _SCRIPTS sys.path prepend → bare module import (`from experiment_background_pipeline_slice import ...`)
- pytest 실행 CWD = `backend/` (sys.path 에는 backend 자체가 들어가지 repo root 가 들어가지 않음)
"""
import json
import subprocess
import sys
from pathlib import Path

_REPO_ROOT = Path(__file__).resolve().parents[3]
_SCRIPT = _REPO_ROOT / "backend" / "scripts" / "experiment_background_pipeline_slice.py"
_SCRIPTS = _REPO_ROOT / "backend" / "scripts"
if str(_SCRIPTS) not in sys.path:
    sys.path.insert(0, str(_SCRIPTS))


def _run(args, *, cwd=None):
    return subprocess.run(
        [sys.executable, str(_SCRIPT), *args],
        cwd=str(cwd or _REPO_ROOT),
        capture_output=True,
        text=True,
        timeout=120,
    )


def test_skeleton_emits_run_meta_with_required_fields(tmp_path):
    out_root = tmp_path / "out"
    result = _run([
        "--dry-run",
        "--skip-db",
        "--output-root", str(out_root),
    ])
    assert result.returncode in (0, 1), result.stderr
    run_dirs = sorted(out_root.glob("*/"))
    assert len(run_dirs) == 1
    run_meta = json.loads((run_dirs[0] / "run_meta.json").read_text())
    for key in ("run_id", "plan_version", "generated_at", "model", "args", "outputs", "run_status", "exit_code"):
        assert key in run_meta, f"run_meta missing {key}"
    assert run_meta["plan_version"] == "bps_w1"
```

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

Run: `cd backend && pytest tests/scripts/test_experiment_background_pipeline_slice.py::test_skeleton_emits_run_meta_with_required_fields -v`
Expected: FAIL — script 파일 없음.

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

```python
# backend/scripts/experiment_background_pipeline_slice.py
"""background_pipeline_slice experiment (W1, plan bps_w1).

Goal: production background/image/reference pipeline 연결 가능한 minimal dry-run
slice. LLM 은 raw intent 만, deterministic code 가 bg_catalog helper 로
production-shape adapter plan 빌드. production code 0 수정 / DB write 0 / image
call 0. 9 acceptance gate 통과 시 exit 0, 아니면 validation_failed + exit 1.
"""
from __future__ import annotations

import argparse
import json
import os
import sys
import uuid
from datetime import datetime, timezone, timedelta
from pathlib import Path

PLAN_VERSION = "bps_w1"
DEFAULT_MODEL = "gemini-3.5-flash"
DEFAULT_PROJECT_ID = "6cb862d9-590c-4dce-86e6-d10c2977db19"
DEFAULT_EPISODE_ID = "08ad2cd3-3e96-4d84-808f-869ee628473c"

_REPO_ROOT = Path(__file__).resolve().parents[2]
_BACKEND_ROOT = _REPO_ROOT / "backend"
_DEFAULT_OUTPUT_ROOT = _REPO_ROOT / "scripts_output" / "background_pipeline_slice_experiment"

KST = timezone(timedelta(hours=9))


def _run_id() -> str:
    return f"{datetime.now(KST).strftime('%Y%m%d_%H%M')}_{uuid.uuid4().hex[:6]}"


def _load_backend_env() -> None:
    env_path = _BACKEND_ROOT / ".env"
    if not env_path.exists():
        return
    for line in env_path.read_text().splitlines():
        line = line.strip()
        if not line or line.startswith("#") or "=" not in line:
            continue
        key, _, value = line.partition("=")
        os.environ.setdefault(key.strip(), value.strip().strip('"').strip("'"))


def _parse_args(argv):
    p = argparse.ArgumentParser(description="background_pipeline_slice experiment (W1)")
    p.add_argument("--project-id", default=DEFAULT_PROJECT_ID)
    p.add_argument("--episode-id", default=DEFAULT_EPISODE_ID)
    p.add_argument("--output-root", default=str(_DEFAULT_OUTPUT_ROOT))
    p.add_argument("--dry-run", action="store_true", help="default; no LLM call")
    p.add_argument("--generate", action="store_true", help="actual LLM call (gemini-3.5-flash)")
    p.add_argument("--model", default=DEFAULT_MODEL)
    p.add_argument("--skip-db", action="store_true", help="skeleton test only; skip DB load")
    return p.parse_args(argv)


def main(argv=None) -> int:
    args = _parse_args(argv)
    run_id = _run_id()
    out_root = Path(args.output_root)
    run_dir = out_root / run_id
    run_dir.mkdir(parents=True, exist_ok=True)
    outputs = []
    failed_invariants = []
    run_status = "succeeded"
    exit_code = 0
    # Phase 1 skeleton ends here; later phases populate outputs/failed_invariants.
    run_meta = {
        "run_id": run_id,
        "plan_version": PLAN_VERSION,
        "generated_at": datetime.now(KST).isoformat(),
        "model": args.model if args.generate else None,
        "args": vars(args),
        "outputs": outputs,
        "run_status": run_status,
        "exit_code": exit_code,
        "failed_invariants": failed_invariants,
    }
    (run_dir / "run_meta.json").write_text(json.dumps(run_meta, ensure_ascii=False, indent=2))
    return exit_code


if __name__ == "__main__":
    sys.exit(main())
```

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

Run: `cd backend && pytest tests/scripts/test_experiment_background_pipeline_slice.py::test_skeleton_emits_run_meta_with_required_fields -v`
Expected: PASS.

### Task 2: KST run_id 형식 + output_root 분리 테스트

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

```python
def test_run_id_format_kst_and_unique(tmp_path):
    out_root = tmp_path / "out"
    r1 = _run(["--dry-run", "--skip-db", "--output-root", str(out_root)])
    r2 = _run(["--dry-run", "--skip-db", "--output-root", str(out_root)])
    assert r1.returncode in (0, 1) and r2.returncode in (0, 1)
    runs = sorted(out_root.glob("*/"))
    assert len(runs) == 2
    for d in runs:
        name = d.name
        assert len(name) == 13 + 1 + 6, f"unexpected run_id: {name}"  # YYYYMMDD_HHMM_xxxxxx
        assert name[8] == "_" and name[13] == "_"
    assert runs[0].name != runs[1].name


def test_dry_run_default_no_model_in_meta(tmp_path):
    out_root = tmp_path / "out"
    _run(["--dry-run", "--skip-db", "--output-root", str(out_root)])
    run_meta = json.loads(next(out_root.glob("*/run_meta.json")).read_text())
    assert run_meta["model"] is None
    assert run_meta["args"]["generate"] is False
```

- [ ] **Step 2/3: 실패 확인 → 위 skeleton 으로 이미 PASS (확인용).**

Run: `cd backend && pytest tests/scripts/test_experiment_background_pipeline_slice.py -v`
Expected: 3 PASS.

---

## Phase 2 — SourceBundle Loader (3 tests)

### Task 3: DB read-only loader (verbatim)

**Files:**
- Modify: `backend/scripts/experiment_background_pipeline_slice.py`
- Modify: `backend/tests/scripts/test_experiment_background_pipeline_slice.py`

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

```python
def test_source_bundle_loads_selected_shots_verbatim(tmp_path):
    out_root = tmp_path / "out"
    result = _run([
        "--dry-run",
        "--output-root", str(out_root),
        "--project-id", "6cb862d9-590c-4dce-86e6-d10c2977db19",
        "--episode-id", "08ad2cd3-3e96-4d84-808f-869ee628473c",
        "--stop-after", "source_bundle",
    ])
    assert result.returncode in (0, 1), result.stderr
    run_dir = next(out_root.glob("*/"))
    bundle = json.loads((run_dir / "source_bundle.json").read_text())
    assert bundle["project_id"] == "6cb862d9-590c-4dce-86e6-d10c2977db19"
    assert bundle["episode_id"] == "08ad2cd3-3e96-4d84-808f-869ee628473c"
    assert bundle["selected_shot_count"] > 0
    for shot in bundle["selected_shots"]:
        assert shot["is_selected"] is True
        # row_id = UUID (DB), shot_key = composite SOT
        for required in ("scene_index", "shot_index", "shot_description",
                         "t2i_variations_json", "visible_entities_json",
                         "row_id", "shot_key"):
            assert required in shot, f"shot row missing {required}"
        # shot_key format
        assert shot["shot_key"].startswith("S")
        # verbatim — no truncate
        assert shot.get("t2i_variations_truncated") is None
    assert bundle["source_hash"]
    # location catalog included
    assert isinstance(bundle["locations"], list)
    for loc in bundle["locations"]:
        assert loc["entity_type"] == "location"
        assert "short_id" in loc and "metadata_json" in loc
    # location_profiles (validate_location_space_profile 통과만)
    assert isinstance(bundle["location_profiles"], dict)
    assert isinstance(bundle["location_profile_errors"], list)


def test_source_bundle_fails_when_no_selected_shots(tmp_path):
    out_root = tmp_path / "out"
    # use a project_id that doesn't exist
    result = _run([
        "--dry-run",
        "--output-root", str(out_root),
        "--project-id", "ZZZZZZZZ-0000-0000-0000-000000000000",
        "--episode-id", "ZZZZZZZZ-0000-0000-0000-000000000000",
        "--stop-after", "source_bundle",
    ])
    assert result.returncode == 1
    run_meta = json.loads(next(out_root.glob("*/run_meta.json")).read_text())
    assert run_meta["run_status"] == "validation_failed"
    assert "no_selected_shots" in run_meta["failed_invariants"]


def test_source_verbatim_hash_matches_db(tmp_path):
    """Bundle 의 verbatim hash 가 DB row 와 일치해야 함."""
    import hashlib
    out_root = tmp_path / "out"
    _run([
        "--dry-run",
        "--output-root", str(out_root),
        "--project-id", "6cb862d9-590c-4dce-86e6-d10c2977db19",
        "--episode-id", "08ad2cd3-3e96-4d84-808f-869ee628473c",
        "--stop-after", "source_bundle",
    ])
    bundle = json.loads(next(out_root.glob("*/source_bundle.json")).read_text())
    canon = json.dumps(bundle["selected_shots"], sort_keys=True, ensure_ascii=False).encode("utf-8")
    assert bundle["source_hash"] == hashlib.sha256(canon).hexdigest()[:16]
```

- [ ] **Step 2: 실패 확인**

Run: `cd backend && pytest tests/scripts/test_experiment_background_pipeline_slice.py::test_source_bundle_loads_selected_shots_verbatim -v`
Expected: FAIL (`--stop-after`, source_bundle 미구현).

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

`experiment_background_pipeline_slice.py` 에 아래 추가:

```python
import hashlib

# argparse 에 추가
p.add_argument("--stop-after", choices=[
    "source_bundle", "pipeline_map", "llm_raw_plan",
    "adapter_plan", "payload", "reference_plan", "compatibility_report"
], default=None)


def _load_source_bundle(project_id: str, episode_id: str) -> dict:
    """DB read-only. SessionLocal + scope 안에서 모든 데이터 fetch 후 dict 변환.
    READ-ONLY guard: session.flush() / session.commit() 호출 금지.
    """
    _load_backend_env()
    sys.path.insert(0, str(_BACKEND_ROOT))
    from app.core.database import SessionLocal  # type: ignore
    from app.models.project import ProjectRegistry, Episode, SceneStill, EntityCanon  # type: ignore

    with SessionLocal() as session:
        proj = session.query(ProjectRegistry).filter_by(id=project_id).one_or_none()
        ep = session.query(Episode).filter_by(id=episode_id, project_id=project_id).one_or_none()
        if proj is None or ep is None:
            return {"project_id": project_id, "episode_id": episode_id,
                    "selected_shot_count": 0, "selected_shots": [], "locations": [],
                    "source_hash": ""}
        rows = (session.query(SceneStill)
                .filter_by(project_id=project_id, episode_id=episode_id, is_selected=True)
                .order_by(SceneStill.scene_index, SceneStill.shot_index, SceneStill.still_index)
                .all())
        shots = []
        for r in rows:
            scene_idx = r.scene_index if r.scene_index is not None else 0
            shot_idx = r.shot_index if r.shot_index is not None else 0
            shot_key = f"S{scene_idx:02d}_Shot{shot_idx}"  # composite SOT
            shots.append({
                "row_id": r.id,  # UUID — DB row 식별용만, binding 에 사용 금지
                "shot_key": shot_key,  # SOT — LLM/binding/coverage 전부 이 키
                "project_id": r.project_id, "episode_id": r.episode_id,
                "still_index": r.still_index, "scene_index": r.scene_index,
                "shot_index": r.shot_index, "shot_description": r.shot_description,
                "screenplay_scene_heading": r.screenplay_scene_heading,
                "beat_title": r.beat_title, "still_frame_prompt": r.still_frame_prompt,
                "camera_json": r.camera_json, "lighting_json": r.lighting_json,
                "visible_entities_json": r.visible_entities_json,
                "t2i_prompt_cinematic": r.t2i_prompt_cinematic,
                "t2i_prompt_closeup": r.t2i_prompt_closeup,
                "t2i_variations_json": r.t2i_variations_json,
                "dependent_scene_id": r.dependent_scene_id,
                "shot_type_1": r.shot_type_1, "shot_type_2": r.shot_type_2,
                "scene_type": r.scene_type, "scene_summary": r.scene_summary,
                "is_selected": bool(r.is_selected),
            })
        locs = (session.query(EntityCanon)
                .filter_by(project_id=project_id, entity_type="location")
                .all())
        location_rows = []
        location_profiles = {}  # short_id → production-shape profile (validated)
        # validate_location_space_profile 은 metadata_json 전체를 받고
        # metadata_json.location.space_profile 을 검사 + 반환.
        # 실패 시 raise → bundle 안 location_profile_errors 에 기록 후 계속.
        from app.core.bg_state_vocab import validate_location_space_profile  # type: ignore
        profile_errors = []
        for L in locs:
            row = {
                "id": L.id, "short_id": L.short_id, "entity_type": L.entity_type,
                "name": L.name, "description": L.description,
                "metadata_json": L.metadata_json,
            }
            location_rows.append(row)
            try:
                # metadata_json 이 str 이면 parse
                meta = L.metadata_json
                if isinstance(meta, str):
                    meta = json.loads(meta) if meta else {}
                profile = validate_location_space_profile(meta, short_id=L.short_id)
                location_profiles[L.short_id] = profile
            except Exception as exc:
                profile_errors.append({"short_id": L.short_id, "error": str(exc)})

    bundle = {
        "project_id": project_id,
        "episode_id": episode_id,
        "project_name": proj.name,
        "episode_title": ep.title,
        "episode_number": ep.episode_number,
        "selected_shot_count": len(shots),
        "selected_shots": shots,
        "locations": location_rows,
        "location_profiles": location_profiles,
        "location_profile_errors": profile_errors,
    }
    canon = json.dumps(shots, sort_keys=True, ensure_ascii=False).encode("utf-8")
    bundle["source_hash"] = hashlib.sha256(canon).hexdigest()[:16]
    return bundle


# main() 안에 phase wiring (skip-db 분기 후):
if not args.skip_db:
    # IMPORTANT 3 (Codex): DB write sentinel install BEFORE 첫 SessionLocal use
    _install_db_write_sentinel()
    bundle = _load_source_bundle(args.project_id, args.episode_id)
    bundle_path = run_dir / "source_bundle.json"
    bundle_path.write_text(json.dumps(bundle, ensure_ascii=False, indent=2))
    outputs.append("source_bundle.json")
    if bundle["selected_shot_count"] == 0:
        failed_invariants.append("no_selected_shots")
        run_status = "validation_failed"
        exit_code = 1
    if args.stop_after == "source_bundle":
        # write run_meta + return
        ...
```

main 의 control flow 를 phase-by-phase 로 재구성. `--stop-after` 가 일치하면 run_meta 쓰고 return.

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

Run: `cd backend && pytest tests/scripts/test_experiment_background_pipeline_slice.py -v`
Expected: 6 PASS (3 skeleton + 3 source_bundle).

---

## Phase 3 — production_pipeline_map (3 tests)

### Task 4: 핵심 10 touchpoint 정적 dict + path/symbol 존재 검증

**Files:**
- Modify: `experiment_background_pipeline_slice.py`
- Modify: test file

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

```python
def test_production_pipeline_map_has_required_touchpoints(tmp_path):
    out_root = tmp_path / "out"
    _run([
        "--dry-run", "--skip-db",
        "--output-root", str(out_root),
        "--stop-after", "pipeline_map",
    ])
    pmap = json.loads(next(out_root.glob("*/production_pipeline_map.json")).read_text())
    assert isinstance(pmap["touchpoints"], list)
    assert 8 <= len(pmap["touchpoints"]) <= 12, "10 ± 2 touchpoints"
    for tp in pmap["touchpoints"]:
        for k in ("file", "symbol", "purpose", "observed_contract"):
            assert k in tp, f"touchpoint missing {k}"
    names = {tp["symbol"] for tp in pmap["touchpoints"]}
    assert "assign_bg_ids" in names
    assert "build_shot_background_map" in names
    assert "BG_ID_RE" in names
    assert "validate_attached_refs" in names
    assert "BackgroundMasterPlanStep" in names


def test_pipeline_map_paths_and_symbols_exist():
    """Static map 의 모든 file path 가 존재하고 symbol text 가 file 안에 있어야 함."""
    from experiment_background_pipeline_slice import PRODUCTION_PIPELINE_MAP
    for tp in PRODUCTION_PIPELINE_MAP["touchpoints"]:
        path = _REPO_ROOT / tp["file"]
        assert path.exists(), f"missing file: {tp['file']}"
        text = path.read_text()
        assert tp["symbol"] in text, f"symbol {tp['symbol']} not found in {tp['file']}"


def test_pipeline_map_report_path_existence_into_run_meta(tmp_path):
    out_root = tmp_path / "out"
    _run([
        "--dry-run", "--skip-db",
        "--output-root", str(out_root),
        "--stop-after", "pipeline_map",
    ])
    run_meta = json.loads(next(out_root.glob("*/run_meta.json")).read_text())
    assert "production_pipeline_map_check" in run_meta
    assert run_meta["production_pipeline_map_check"]["all_paths_exist"] is True
    assert run_meta["production_pipeline_map_check"]["all_symbols_found"] is True
```

- [ ] **Step 2: 실패 확인**

Run: `cd backend && pytest tests/scripts/test_experiment_background_pipeline_slice.py::test_production_pipeline_map_has_required_touchpoints -v`
Expected: FAIL.

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

`experiment_background_pipeline_slice.py` 에 추가:

```python
PRODUCTION_PIPELINE_MAP = {
    "touchpoints": [
        {
            "file": "backend/app/core/bg_catalog.py",
            "symbol": "assign_bg_ids",
            "purpose": "Deterministic L##B## bg_id assignment from LLM raw intents",
            "observed_contract": "assign_bg_ids(prev_catalog: Dict[str, Dict[str, Any]], new_intents: List[Dict[str, Any]], location_profiles: Dict[loc_id, profile]) -> Dict[bg_id, entry]. entry = {bg_id, loc_id, space_key, time_phase, state_class, semantic_key, depends_on_fp[], depends_on_bg[], applies_to_shots[], sub_location_label, state_label_raw}. Raises SemanticKeyError/FpLinkMismatchError. Same semantic_key with differing render-relevant deps → SemanticKeyError fail-fast.",
        },
        {
            "file": "backend/app/core/bg_catalog.py",
            "symbol": "build_shot_background_map",
            "purpose": "Flatten catalog applies_to_shots into shot_id->bg_id; N:1 invariant",
            "observed_contract": "build_shot_background_map(catalog: Dict[bg_id, entry]) -> Dict[shot_id, bg_id]. Raises ShotBindingError if same shot maps to >1 bg_id.",
        },
        {
            "file": "backend/app/core/bg_catalog.py",
            "symbol": "compute_bg_catalog_hash",
            "purpose": "SHA256[:16] over render-relevant catalog fields for drift detection",
            "observed_contract": "compute_bg_catalog_hash(catalog: Dict[bg_id, entry]) -> str. Excludes applies_to_shots/sub_location_label/state_label_raw.",
        },
        {
            "file": "backend/app/core/bg_state_vocab.py",
            "symbol": "BG_ID_RE",
            "purpose": "Production bg_id format regex (^L\\d{2,3}B\\d{2,3}$) — ID-format validation only, not semantic",
            "observed_contract": "re.compile(r'^L\\d{2,3}B\\d{2,3}$'). Used by background_render_step strict marker filter.",
        },
        {
            "file": "backend/app/core/bg_state_vocab.py",
            "symbol": "STATE_CLASS_ENUM",
            "purpose": "11-value state_class vocabulary for background variants",
            "observed_contract": "Frozenset of allowed state_class strings emitted by master_plan LLM and validated by assign_bg_ids.",
        },
        {
            "file": "backend/app/core/ref_contract_validator.py",
            "symbol": "validate_attached_refs",
            "purpose": "Tier-3 attached-reference identity contract; RefContractError HTTP422 on violation",
            "observed_contract": "validate_attached_refs(rpc: dict, labeled_refs: List[(label, bytes)], attached_meta: List[(kind, id)], prompt: str, is_close_framing: bool, *, chain_bg_lookup: Callable, reference_phrase_kinds: List[str]) -> None. 6 sequential checks (length, rpc shape, character_outlook strict, prop strict, character base strict, background lineage + close-framing skip + readiness, phantom guard).",
        },
        {
            "file": "backend/app/core/steps/background_master_plan_step.py",
            "symbol": "BackgroundMasterPlanStep",
            "purpose": "Production producer that emits plans[group_id]={floor_plans, backgrounds, gen_order} + shot_background_map (mirror target for production_adapter_plan)",
            "observed_contract": "SCHEMA_VERSION=3, PROMPT_VERSION='5.202605201759'. _d6_post_process calls bg_catalog.assign_bg_ids. Output checkpoint at projects/{pid}/checkpoints/episodes/{eid}/background_master_plan/manifest.json.",
        },
        {
            "file": "backend/app/core/steps/background_render_step.py",
            "symbol": "BackgroundRenderStep",
            "purpose": "PNG producer; ImageAsset(asset_type='chain_bg', variant_type=bg_id) (do NOT call from experiment)",
            "observed_contract": "Consumes background_master_plan + background_prompt + floor_plan_render. UPSERT pattern. gpt-image-2 model. Strict BG_ID_RE marker filter.",
        },
        {
            "file": "backend/app/services/scene_checkpoint_loaders.py",
            "symbol": "load_background_chain_bg_map",
            "purpose": "Image-phase consumer that produces {scene_idx_shot_idx: {bg_id, location_id, label, image_bytes}}",
            "observed_contract": "load_background_chain_bg_map(project_id, episode_id) -> Dict[shot_key, dict]. Reads background_render checkpoint; respects settings.background_chain_enabled.",
        },
        {
            "file": "backend/app/services/scene_generation_coordinator.py",
            "symbol": "build_chain_bg_lookup",
            "purpose": "Build callable bg_id->location_id for ref_contract_validator chain_bg_lookup",
            "observed_contract": "build_chain_bg_lookup(background_chain_bg_map) -> Callable[[bg_id], Optional[loc_id]].",
        },
    ],
}


def _verify_pipeline_map(run_dir: Path) -> dict:
    out_path = run_dir / "production_pipeline_map.json"
    out_path.write_text(json.dumps(PRODUCTION_PIPELINE_MAP, ensure_ascii=False, indent=2))
    all_paths_exist = True
    all_symbols_found = True
    missing = []
    for tp in PRODUCTION_PIPELINE_MAP["touchpoints"]:
        p = _REPO_ROOT / tp["file"]
        if not p.exists():
            all_paths_exist = False
            missing.append({"path": tp["file"], "kind": "path_missing"})
            continue
        if tp["symbol"] not in p.read_text():
            all_symbols_found = False
            missing.append({"path": tp["file"], "symbol": tp["symbol"], "kind": "symbol_missing"})
    return {"all_paths_exist": all_paths_exist, "all_symbols_found": all_symbols_found, "missing": missing}


# main() 안에 phase wiring (source_bundle 후):
pmap_check = _verify_pipeline_map(run_dir)
outputs.append("production_pipeline_map.json")
if not (pmap_check["all_paths_exist"] and pmap_check["all_symbols_found"]):
    failed_invariants.append("production_pipeline_map_stale")
    run_status = "validation_failed"; exit_code = 1
# run_meta 에 production_pipeline_map_check 키 추가
```

- [ ] **Step 4: PASS 확인**

Run: `cd backend && pytest tests/scripts/test_experiment_background_pipeline_slice.py -v`
Expected: 9 PASS.

---

## Phase 4 — LLM Raw Plan (schema + dry-run placeholder + generate) (4 tests)

### Task 5: LLM raw plan schema + dry-run placeholder

**Schema decision (Codex 합의):** LLM 은 raw intent 만 emit. bg_id 절대 emit 하지 않음. depends_on_fp 는 `fp_intent_key` (semantic key) 만 emit. depends_on_bg 는 `parent_intent_ref` 만 emit (raw intent key 참조). code 가 후처리로 bg_id 부여 + resolve. **state_class 는 production `STATE_CLASS_ENUM` (`app.core.bg_state_vocab`) 강제** — fail-fast `StateClassError` 방지.

```python
# script 상단 module-level (sys.path 후, settings 불필요):
sys.path.insert(0, str(_BACKEND_ROOT))
from app.core.bg_state_vocab import STATE_CLASS_ENUM as _STATE_CLASS_ENUM  # type: ignore

LLM_RAW_PLAN_SCHEMA = {
    "type": "object",
    "additionalProperties": False,
    "required": ["building_groups", "raw_background_intents", "floor_plan_intents",
                 "shot_binding_intents", "prompt_payload_intents", "open_risks"],
    "properties": {
        "building_groups": {
            "type": "array",
            "items": {
                "type": "object",
                "additionalProperties": False,
                "required": ["group_id", "member_loc_ids", "anchor_loc_id", "kind", "rationale"],
                "properties": {
                    "group_id": {"type": "string"},
                    "member_loc_ids": {"type": "array", "items": {"type": "string"}},
                    "anchor_loc_id": {"type": "string"},
                    "kind": {"type": "string", "enum": ["chain_bg", "prev_shot_ref", "skip"]},
                    "rationale": {"type": "string"},
                },
            },
        },
        "raw_background_intents": {
            "type": "array",
            "items": {
                "type": "object",
                "additionalProperties": False,
                # NOTE: applies_to_shots 필드 없음 — shot_binding_intents 가 SOT.
                # code 가 binding 으로부터 derive 후 catalog.applies_to_shots 에 채워넣음.
                "required": ["intent_key", "group_id", "loc_id", "space_key_hint",
                             "time_phase", "state_class", "sub_location_label",
                             "state_label_raw", "fp_intent_key",
                             "parent_intent_ref", "evidence_basis"],
                "properties": {
                    "intent_key": {"type": "string", "description": "stable key (lowercase a-z0-9_) for cross-reference; NOT bg_id"},
                    "group_id": {"type": "string"},
                    "loc_id": {"type": "string"},
                    "space_key_hint": {"type": "string"},
                    "time_phase": {"type": "string"},
                    "state_class": {"type": "string", "enum": sorted(_STATE_CLASS_ENUM)},
                    "sub_location_label": {"type": ["string", "null"]},
                    "state_label_raw": {"type": ["string", "null"]},
                    "fp_intent_key": {"type": ["string", "null"]},
                    "parent_intent_ref": {"type": ["string", "null"]},
                    "evidence_basis": {
                        "type": "array",
                        "items": {
                            "type": "object",
                            "additionalProperties": False,
                            "required": ["quote", "source_ref", "confidence"],
                            "properties": {
                                "quote": {"type": "string"},
                                "source_ref": {"type": "string"},
                                "confidence": {"type": "string", "enum": ["trusted", "plausible", "weak"]},
                            },
                        },
                    },
                },
            },
        },
        "floor_plan_intents": {
            "type": "array",
            "items": {
                "type": "object",
                "additionalProperties": False,
                "required": ["fp_intent_key", "loc_id", "space_key_hint", "rationale"],
                "properties": {
                    "fp_intent_key": {"type": "string"},
                    "loc_id": {"type": "string"},
                    "space_key_hint": {"type": "string"},
                    "rationale": {"type": "string"},
                },
            },
        },
        "shot_binding_intents": {
            "type": "array",
            "items": {
                "type": "object",
                "additionalProperties": False,
                "required": ["shot_key", "intent_key"],
                "properties": {
                    "shot_key": {"type": "string", "description": "composite shot id (e.g. 'S{scene_index}_Shot{shot_index}'). SOT for all binding."},
                    "intent_key": {"type": "string"},
                },
            },
        },
        "prompt_payload_intents": {
            "type": "array",
            "items": {
                "type": "object",
                "additionalProperties": False,
                "required": ["shot_key", "bg_intent_text"],
                "properties": {
                    "shot_key": {"type": "string"},
                    "bg_intent_text": {"type": "string"},
                },
            },
        },
        "open_risks": {
            "type": "array",
            "maxItems": 5,
            "items": {
                "type": "object",
                "additionalProperties": False,
                "required": ["type", "severity", "description", "conservative_fallback"],
                "properties": {
                    "type": {"type": "string"},
                    "severity": {"type": "string", "enum": ["low", "med", "high"]},
                    "description": {"type": "string"},
                    "conservative_fallback": {"type": "string"},
                },
            },
        },
    },
}
```

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

```python
def test_llm_raw_plan_placeholder_when_dry_run(tmp_path):
    out_root = tmp_path / "out"
    _run([
        "--dry-run",
        "--output-root", str(out_root),
        "--project-id", "6cb862d9-590c-4dce-86e6-d10c2977db19",
        "--episode-id", "08ad2cd3-3e96-4d84-808f-869ee628473c",
        "--stop-after", "llm_raw_plan",
    ])
    plan = json.loads(next(out_root.glob("*/llm_raw_plan.json")).read_text())
    assert plan["status"] == "placeholder_dry_run"
    for key in ("building_groups", "raw_background_intents", "floor_plan_intents",
                "shot_binding_intents", "prompt_payload_intents", "open_risks"):
        assert key in plan["plan"]


def test_llm_raw_plan_schema_validates_against_jsonschema():
    from experiment_background_pipeline_slice import (
        LLM_RAW_PLAN_SCHEMA, _build_placeholder_raw_plan,
    )
    import jsonschema
    sample = {
        "selected_shots": [
            {"id": "s1", "scene_index": 1, "shot_index": 1, "is_selected": True,
             "shot_description": "x"}
        ],
        "locations": [],
    }
    placeholder = _build_placeholder_raw_plan(sample)
    jsonschema.validate(placeholder["plan"], LLM_RAW_PLAN_SCHEMA)


def test_dry_run_does_not_import_litellm(tmp_path):
    """sentinel — dry-run 에서는 litellm 이 import 되지 않아야 함."""
    out_root = tmp_path / "out"
    result = _run([
        "--dry-run",
        "--output-root", str(out_root),
        "--project-id", "6cb862d9-590c-4dce-86e6-d10c2977db19",
        "--episode-id", "08ad2cd3-3e96-4d84-808f-869ee628473c",
        "--stop-after", "llm_raw_plan",
        "--diag-print-imports",
    ])
    # script 가 --diag-print-imports 일 때 sys.modules 키를 stderr 로 출력
    assert "litellm" not in result.stderr


def test_open_risks_max_five_and_have_fallback():
    from experiment_background_pipeline_slice import LLM_RAW_PLAN_SCHEMA
    risks_schema = LLM_RAW_PLAN_SCHEMA["properties"]["open_risks"]
    assert risks_schema["maxItems"] == 5
    item = risks_schema["items"]
    assert "conservative_fallback" in item["required"]


def test_llm_raw_plan_schema_state_class_enum_matches_production():
    """state_class enum 이 production STATE_CLASS_ENUM 과 동일해야 함 (Codex BLOCKING1)."""
    from experiment_background_pipeline_slice import LLM_RAW_PLAN_SCHEMA
    import sys as _sys
    _sys.path.insert(0, str(_REPO_ROOT / "backend"))
    from app.core.bg_state_vocab import STATE_CLASS_ENUM
    intents_schema = LLM_RAW_PLAN_SCHEMA["properties"]["raw_background_intents"]["items"]
    enum_in_schema = set(intents_schema["properties"]["state_class"]["enum"])
    assert enum_in_schema == set(STATE_CLASS_ENUM), (
        f"state_class enum drift: schema={sorted(enum_in_schema)} "
        f"vs production={sorted(STATE_CLASS_ENUM)}"
    )
```

- [ ] **Step 2: 실패 확인**

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

```python
def _build_placeholder_raw_plan(bundle: dict) -> dict:
    """dry-run 기본값. 실제 LLM call 없이 schema-valid skeleton 만 emit."""
    selected = bundle.get("selected_shots", [])
    locs = bundle.get("locations", [])
    plan = {
        "building_groups": [],
        "raw_background_intents": [],
        "floor_plan_intents": [],
        "shot_binding_intents": [],
        "prompt_payload_intents": [],
        "open_risks": [
            {
                "type": "placeholder_dry_run",
                "severity": "low",
                "description": "No LLM call performed; --generate not specified.",
                "conservative_fallback": "Run with --generate --model gemini-3.5-flash to populate plan.",
            }
        ],
    }
    return {"status": "placeholder_dry_run", "plan": plan, "model_used": None}


def _generate_llm_raw_plan(bundle: dict, *, model: str) -> dict:
    """--generate 시 호출. litellm 직접 + response_format json_object."""
    import litellm  # lazy import — dry-run 에서는 absent
    # NOTE: system prompt 안에 sample-specific 이름 (방, 인물, 소품, episode id) literal
    # 0건. 자기 fixture-swap test 가 script source 의 banned literal 을 검사하므로 이
    # prompt 자체에도 들어가면 안 됨. methodology 는 generic 표현으로만.
    system_prompt = (
        "You plan production-compatible background generation units for a film scene "
        "pipeline. Emit raw semantic intents only — do NOT assign bg_id, floor_plan id, "
        "or any deterministic identifier; downstream code assigns those via the production "
        "helper. Each background intent must include loc_id, space_key_hint, time_phase, "
        "state_class, sub_location_label, state_label_raw, and evidence_basis quoting the "
        "source bundle. shot-to-intent mapping belongs to shot_binding_intents (one entry "
        "per selected shot_key), not in raw_background_intents. Use intent_key for cross-"
        "reference and parent_intent_ref for depends_on_bg semantics. Constraints: do not "
        "introduce sample-specific location labels, named characters, room labels, props, "
        "or episode identifiers as methodology rules; source quotes may contain fixture "
        "data, but your own schema/rules must stay generic. Max 5 open_risks, each with a "
        "conservative_fallback. Output must match the provided JSON schema."
    )
    user_prompt = json.dumps({
        "source_bundle": {
            "selected_shots": bundle["selected_shots"],
            "locations": bundle["locations"],
        },
        "json_schema": LLM_RAW_PLAN_SCHEMA,
    }, ensure_ascii=False)
    routed = model if model.startswith("gemini/") or not model.lower().startswith("gemini") else f"gemini/{model}"
    if not (os.environ.get("GEMINI_API_KEY") or os.environ.get("GOOGLE_API_KEY")):
        raise RuntimeError("missing GEMINI_API_KEY / GOOGLE_API_KEY env var")
    resp = litellm.completion(
        model=routed,
        messages=[{"role": "system", "content": system_prompt},
                  {"role": "user", "content": user_prompt}],
        response_format={"type": "json_object"},
    )
    raw_text = resp.choices[0].message.content
    parsed = json.loads(raw_text)
    return {"status": "generated", "plan": parsed, "model_used": model}


# main() 안:
if args.generate:
    try:
        raw_plan = _generate_llm_raw_plan(bundle, model=args.model)
        # schema validate
        import jsonschema
        jsonschema.validate(raw_plan["plan"], LLM_RAW_PLAN_SCHEMA)
    except Exception as exc:
        # 1회 deterministic correction retry — schema/id-only (semantic 미수정)
        try:
            raw_plan = _generate_llm_raw_plan(bundle, model=args.model)
            jsonschema.validate(raw_plan["plan"], LLM_RAW_PLAN_SCHEMA)
            raw_plan["correction_attempted"] = True
        except Exception as exc2:
            raw_plan = {"status": "generate_failed", "plan": None, "model_used": args.model,
                        "error": str(exc2)}
            failed_invariants.append("llm_raw_plan_generate_failed")
            run_status = "validation_failed"; exit_code = 1
else:
    raw_plan = _build_placeholder_raw_plan(bundle)
(run_dir / "llm_raw_plan.json").write_text(json.dumps(raw_plan, ensure_ascii=False, indent=2))
outputs.append("llm_raw_plan.json")
```

`--diag-print-imports` 추가:
```python
p.add_argument("--diag-print-imports", action="store_true")
# main 끝에:
if args.diag_print_imports:
    sys.stderr.write(",".join(sorted(sys.modules.keys())) + "\n")
```

- [ ] **Step 4: PASS 확인**

Run: `cd backend && pytest tests/scripts/test_experiment_background_pipeline_slice.py -v`
Expected: 13 PASS.

---

## Phase 5 — production_adapter_plan (bg_catalog adapter) (4 tests)

### Task 6: raw intent → assign_bg_ids → master_plan-호환 shape

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

```python
def _fake_raw_plan_minimal():
    """test fixture — 3 raw intent, 2 fp intent, 4 shot binding.

    shot_key (composite, SOT) 만 사용. raw intent 에는 applies_to_shots 없음 —
    shot_binding_intents 가 SOT. fp_intent_key 와 parent_intent_ref 로 deps semantic 만 LLM.
    """
    return {
        "building_groups": [
            {"group_id": "G01", "member_loc_ids": ["L01"], "anchor_loc_id": "L01",
             "kind": "chain_bg", "rationale": "single-room scene"},
            {"group_id": "G02", "member_loc_ids": ["L02"], "anchor_loc_id": "L02",
             "kind": "chain_bg", "rationale": "outdoor"},
        ],
        "raw_background_intents": [
            # state_class 는 production STATE_CLASS_ENUM (bg_state_vocab.py:26-38) 강제.
            # 'normal' → 'ransacked' 는 production enum 안에서 의미상 자연스러운 state 전이.
            {"intent_key": "k1", "group_id": "G01", "loc_id": "L01",
             "space_key_hint": "main", "time_phase": "day", "state_class": "normal",
             "sub_location_label": None, "state_label_raw": None,
             "fp_intent_key": "fp1", "parent_intent_ref": None,
             "evidence_basis": [{"quote": "...", "source_ref": "shot.S01_Shot1", "confidence": "trusted"}]},
            {"intent_key": "k2", "group_id": "G01", "loc_id": "L01",
             "space_key_hint": "main", "time_phase": "day", "state_class": "ransacked",
             "sub_location_label": None, "state_label_raw": "after_disturbance",
             "fp_intent_key": "fp1", "parent_intent_ref": "k1",
             "evidence_basis": [{"quote": "...", "source_ref": "shot.S02_Shot3", "confidence": "plausible"}]},
            {"intent_key": "k3", "group_id": "G02", "loc_id": "L02",
             "space_key_hint": "main", "time_phase": "day", "state_class": "normal",
             "sub_location_label": None, "state_label_raw": None,
             "fp_intent_key": "fp2", "parent_intent_ref": None,
             "evidence_basis": [{"quote": "...", "source_ref": "shot.S03_Shot4", "confidence": "trusted"}]},
        ],
        "floor_plan_intents": [
            {"fp_intent_key": "fp1", "loc_id": "L01", "space_key_hint": "main", "rationale": "indoor"},
            {"fp_intent_key": "fp2", "loc_id": "L02", "space_key_hint": "main", "rationale": "indoor"},
        ],
        "shot_binding_intents": [
            {"shot_key": "S01_Shot1", "intent_key": "k1"},
            {"shot_key": "S01_Shot2", "intent_key": "k1"},
            {"shot_key": "S02_Shot3", "intent_key": "k2"},
            {"shot_key": "S03_Shot4", "intent_key": "k3"},
        ],
        "prompt_payload_intents": [
            {"shot_key": sk, "bg_intent_text": "background description"}
            for sk in ["S01_Shot1", "S01_Shot2", "S02_Shot3", "S03_Shot4"]
        ],
        "open_risks": [],
    }


def _fake_location_profiles():
    # production shape — kind (not space_kind), single_space → space_key 'main' 강제
    return {
        "L01": {"kind": "single_space", "allowed_space_keys": ["main"]},
        "L02": {"kind": "single_space", "allowed_space_keys": ["main"]},
    }


def test_adapter_assigns_bg_ids_with_production_helper():
    from experiment_background_pipeline_slice import _build_adapter_plan
    plan = _fake_raw_plan_minimal()
    adapter = _build_adapter_plan(plan, _fake_location_profiles())
    catalog = adapter["background_catalog"]  # Dict[bg_id, entry]
    assert isinstance(catalog, dict)
    import re
    bg_re = re.compile(r"^L\d{2,3}B\d{2,3}$")
    for bid in catalog.keys():
        assert bg_re.match(bid), f"invalid bg_id: {bid}"
    # 3 intents → 3 catalog entries (no semantic dedup expected in this fake)
    assert len(catalog) == 3


def test_adapter_assign_bg_ids_deterministic_rerun():
    from experiment_background_pipeline_slice import _build_adapter_plan
    plan = _fake_raw_plan_minimal()
    p1 = _build_adapter_plan(plan, _fake_location_profiles())
    p2 = _build_adapter_plan(plan, _fake_location_profiles())
    assert p1["bg_catalog_hash"] == p2["bg_catalog_hash"]
    assert p1["shot_binding_hash"] == p2["shot_binding_hash"]


def test_adapter_shot_background_map_n_to_1_invariant():
    """동일 shot_key 이 두 intent_key 에 매핑되면 ShotBindingError → validation_failed."""
    from experiment_background_pipeline_slice import _build_adapter_plan
    plan = _fake_raw_plan_minimal()
    plan["shot_binding_intents"].append({"shot_key": "S01_Shot1", "intent_key": "k2"})
    import pytest
    with pytest.raises(Exception):  # ShotBindingError 또는 wrapped (catalog applies_to_shots derive 후 helper 가 raise)
        _build_adapter_plan(plan, _fake_location_profiles())


def test_adapter_gen_order_topo_resolves_parent_intent_ref():
    """k2 는 parent_intent_ref=k1 → bg_id(k2) 의 depends_on_bg 에 bg_id(k1) 가 들어가야 함.
    gen_order 는 k1 의 bg_id 가 k2 의 bg_id 보다 먼저."""
    from experiment_background_pipeline_slice import _build_adapter_plan
    plan = _fake_raw_plan_minimal()
    adapter = _build_adapter_plan(plan, _fake_location_profiles())
    intent_to_bg = adapter["intent_key_to_bg_id"]
    k1_bg = intent_to_bg["k1"]
    k2_bg = intent_to_bg["k2"]
    k2_entry = adapter["background_catalog"][k2_bg]
    assert k1_bg in k2_entry["depends_on_bg"]
    order = adapter["gen_order"]
    assert order.index(k1_bg) < order.index(k2_bg)


def test_adapter_dedup_parent_conflict_fail_fast():
    """Codex v3 BLOCKING 2 + v4 BLOCKING 1 — 같은 semantic_key 로 dedup 된 intent
    들의 parent semantics 가 inconsistent (parent_bg vs different parent_bg, OR
    None vs some parent_bg) 면 dedup_parent_conflict 로 fail-fast.
    (단일 test 안에 2 case — overengineering 우려로 별도 test 추가 X)."""
    from experiment_background_pipeline_slice import _build_adapter_plan
    import pytest

    # Case A: parent_bg vs different parent_bg
    plan_a = _fake_raw_plan_minimal()
    # k_dup 도 (L01, main, day, ransacked) — k2 와 같은 semantic_key
    # k2 의 parent=k1, k_dup 의 parent=k3 → 같은 bg 로 dedup 시 conflict
    plan_a["raw_background_intents"].append({
        "intent_key": "k_dup", "group_id": "G01", "loc_id": "L01",
        "space_key_hint": "main", "time_phase": "day", "state_class": "ransacked",
        "sub_location_label": None, "state_label_raw": "after_disturbance",
        "fp_intent_key": "fp1", "parent_intent_ref": "k3",
        "evidence_basis": [{"quote": "...", "source_ref": "shot.S_dup", "confidence": "weak"}],
    })
    plan_a["shot_binding_intents"].append({"shot_key": "S_dup_Shot1", "intent_key": "k_dup"})
    with pytest.raises(ValueError, match="dedup_parent_conflict"):
        _build_adapter_plan(plan_a, _fake_location_profiles())

    # Case B: None vs some_bg (v4 BLOCKING 1)
    plan_b = _fake_raw_plan_minimal()
    # k_dup2 가 k2 와 같은 semantic_key 지만 parent=None — k2 는 parent=k1 → None vs k1 conflict
    plan_b["raw_background_intents"].append({
        "intent_key": "k_dup2", "group_id": "G01", "loc_id": "L01",
        "space_key_hint": "main", "time_phase": "day", "state_class": "ransacked",
        "sub_location_label": None, "state_label_raw": "after_disturbance",
        "fp_intent_key": "fp1", "parent_intent_ref": None,
        "evidence_basis": [{"quote": "...", "source_ref": "shot.S_dup2", "confidence": "weak"}],
    })
    plan_b["shot_binding_intents"].append({"shot_key": "S_dup2_Shot1", "intent_key": "k_dup2"})
    with pytest.raises(ValueError, match="dedup_parent_conflict"):
        _build_adapter_plan(plan_b, _fake_location_profiles())


def test_adapter_dedup_parent_self_cycle_fail_fast():
    """self-cycle (intent 가 자기 자신을 parent 로 가리킴) fail-fast."""
    from experiment_background_pipeline_slice import _build_adapter_plan
    plan = _fake_raw_plan_minimal()
    # k1 의 parent 를 k1 자기 자신으로 바꿈
    plan["raw_background_intents"][0]["parent_intent_ref"] = "k1"
    import pytest
    with pytest.raises(ValueError, match="dedup_parent_self_cycle"):
        _build_adapter_plan(plan, _fake_location_profiles())
```

- [ ] **Step 2: 실패 확인**

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

```python
def _build_adapter_plan(llm_raw_plan: dict, location_profiles: dict) -> dict:
    """Code-built master_plan-호환 shape. bg_catalog.assign_bg_ids 사용.

    SOT 규약 (Codex 합의):
    - background_catalog 는 production mirror Dict[bg_id, entry]. SOT.
    - shot_binding_intents 가 shot→intent_key SOT. raw_background_intents 에는
      applies_to_shots 필드 없음. code 가 binding 으로부터 applies_to_shots 를 derive.
    - assign_bg_ids 가 같은 semantic_key 로 2 intent_key 를 dedup 하고 deps 가
      다르면 SemanticKeyError raise — fail-fast (best-effort 금지).
    - HTML 등 view 가 필요하면 background_catalog_list 별도 산출, 단 SOT 는 dict.
    """
    sys.path.insert(0, str(_BACKEND_ROOT))
    from app.core import bg_catalog  # type: ignore
    from app.core.bg_catalog import ShotBindingError, SemanticKeyError  # type: ignore

    intents = llm_raw_plan.get("raw_background_intents", [])
    floor_plan_intents = llm_raw_plan.get("floor_plan_intents", [])
    shot_bindings = llm_raw_plan.get("shot_binding_intents", [])

    # fp_id 부여 (deterministic): fp_intent_key 정렬 후 loc_id 별 시퀀스.
    fp_id_by_intent_key = {}
    fp_seq_by_loc = {}
    for fp in sorted(floor_plan_intents, key=lambda x: (x["loc_id"], x["fp_intent_key"])):
        loc_num = int(fp["loc_id"].lstrip("L"))
        fp_seq_by_loc.setdefault(loc_num, 0)
        fp_seq_by_loc[loc_num] += 1
        fp_id = f"fp_l{loc_num:02d}_{fp_seq_by_loc[loc_num]:02d}"
        fp_id_by_intent_key[fp["fp_intent_key"]] = fp_id

    # IMPORTANT 2: shot_binding_intents SOT → intent_key 별 shot_key 목록 derive.
    shots_by_intent: Dict[str, list] = {}
    for sb in shot_bindings:
        shots_by_intent.setdefault(sb["intent_key"], []).append(sb["shot_key"])

    # raw intent → assign_bg_ids 입력. depends_on_bg 는 ID assign 후 resolve,
    # 일단 빈 list 로 (raw 의 parent_intent_ref 만 보존).
    new_intents = []
    for it in intents:
        depends_fp = [fp_id_by_intent_key[it["fp_intent_key"]]] if it.get("fp_intent_key") else []
        new_intents.append({
            "loc_id": it["loc_id"],
            "space_key_hint": it["space_key_hint"],
            "time_phase": it["time_phase"],
            "state_class": it["state_class"],
            "sub_location_label": it.get("sub_location_label"),
            "state_label_raw": it.get("state_label_raw"),
            "depends_on_fp": depends_fp,
            "depends_on_bg": [],
            # derived from shot_binding_intents (SOT)
            "applies_to_shots": list(shots_by_intent.get(it["intent_key"], [])),
        })

    # 1차 pass: bg_id 부여 (depends_on_bg 빈 상태).
    catalog: Dict[str, Dict] = bg_catalog.assign_bg_ids(
        prev_catalog={}, new_intents=new_intents, location_profiles=location_profiles)

    # intent_key → bg_id 매핑. assign_bg_ids 가 같은 semantic_key 를 dedup 하므로
    # semantic_key 기준 역매핑. dedup 으로 intent_key 1개 ↔ bg_id 1개를 보장하는 게
    # 아니라, 여러 intent_key 가 같은 bg_id 에 매핑될 수 있음 → Q1 fail-fast 처리는
    # 아래 parent_intent_ref resolve 시 깐깐히 확인.
    intent_key_to_bg = {}
    for it in intents:
        sem_key = bg_catalog.compute_semantic_key(
            it["loc_id"],
            bg_catalog.normalize_space_key(it["loc_id"], it["space_key_hint"],
                                           location_profiles[it["loc_id"]]),
            it["time_phase"], it["state_class"])
        match_bg = None
        for bg_id, entry in catalog.items():
            if entry["semantic_key"] == sem_key:
                match_bg = bg_id
                break
        if match_bg is None:
            raise ValueError(f"intent_key {it['intent_key']!r} did not match any catalog entry")
        intent_key_to_bg[it["intent_key"]] = match_bg

    # 2차 pass: parent_intent_ref → depends_on_bg resolve. dict 갱신.
    # Codex Q1 / v3 BLOCKING 2 + v4 BLOCKING 1 — fail-fast:
    #   - 같은 bg 로 dedup 된 multi intent 의 parent_intent_ref 가 서로 다른 bg 면
    #     dedup_parent_conflict (best-effort 흡수 금지).
    #   - None vs some_bg 인 경우도 inconsistent semantics — 마찬가지 conflict.
    #     한 쪽이 parent 없고 다른 쪽이 있으면 silent winner 선택 금지.
    #   - 자기 자신이 parent 면 dedup_parent_self_cycle.
    #   - assign_bg_ids 가 이미 internal same-run dedup 시 deps invariant 확인하지만,
    #     parent_intent_ref 는 1차 pass 후 채우므로 그 helper 가 못 보는 영역 — 직접 fail-fast.
    parent_candidates_by_bg: Dict[str, set] = {}
    for it in intents:
        bg = intent_key_to_bg[it["intent_key"]]
        raw_parent = it.get("parent_intent_ref")
        if raw_parent:
            parent_bg = intent_key_to_bg.get(raw_parent)
            if parent_bg is None:
                raise ValueError(f"parent_intent_ref {raw_parent!r} unresolved")
            if parent_bg == bg:
                raise ValueError(
                    f"dedup_parent_self_cycle: intent_key={it['intent_key']!r} "
                    f"parent_intent_ref={raw_parent!r} → same bg {bg}"
                )
            parent_candidates_by_bg.setdefault(bg, set()).add(parent_bg)
        else:
            # parent 없음 도 명시적으로 candidate 에 기록 — None vs some_bg conflict 감지.
            parent_candidates_by_bg.setdefault(bg, set()).add(None)
    for bg_id, candidates in parent_candidates_by_bg.items():
        if len(candidates) > 1:
            labels = sorted(c if c is not None else "<none>" for c in candidates)
            raise ValueError(
                f"dedup_parent_conflict: bg_id={bg_id} has inconsistent parent "
                f"candidates {labels} from dedup'd intent group"
            )
        only = next(iter(candidates))
        catalog[bg_id]["depends_on_bg"] = [only] if only is not None else []

    # shot_background_map: production helper SOT.
    shot_bg = bg_catalog.build_shot_background_map(catalog)

    # gen_order topo (depends_on_fp 먼저, 다음 bg DAG topo).
    fp_ids = sorted({fp for e in catalog.values() for fp in e["depends_on_fp"]})
    bg_ids_ordered = []
    pending = list(catalog.items())  # (bg_id, entry)
    placed = set()
    while pending:
        progressed = False
        for bg_id, entry in list(pending):
            if all(d in placed for d in entry["depends_on_bg"]):
                bg_ids_ordered.append(bg_id)
                placed.add(bg_id)
                pending.remove((bg_id, entry))
                progressed = True
        if not progressed:
            raise ValueError(
                f"reference_chain cycle: remaining={[bg for bg, _ in pending]}")
    gen_order = fp_ids + bg_ids_ordered

    return {
        "schema_version": 1,
        "plan_version": PLAN_VERSION,
        "building_groups": llm_raw_plan.get("building_groups", []),
        "floor_plans": [{"fp_id": fp_id_by_intent_key[fp["fp_intent_key"]],
                         "loc_id": fp["loc_id"],
                         "space_key_hint": fp["space_key_hint"]}
                        for fp in floor_plan_intents],
        "background_catalog": catalog,  # Dict[bg_id, entry] — SOT
        "shot_background_map": shot_bg,  # Dict[shot_key, bg_id]
        "bg_catalog_hash": bg_catalog.compute_bg_catalog_hash(catalog),
        "shot_binding_hash": bg_catalog.compute_shot_binding_hash(shot_bg),
        "gen_order": gen_order,
        "intent_key_to_bg_id": intent_key_to_bg,
    }
```

- [ ] **Step 4: PASS 확인**

Run: `cd backend && pytest tests/scripts/test_experiment_background_pipeline_slice.py -v`
Expected: 17 PASS.

---

## Phase 6 — generation_payload_dry_run + reference_input_plan (3 tests)

### Task 7: per-shot dry-run payload + ref attach plan

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

```python
def test_generation_payload_has_minimum_required_fields():
    from experiment_background_pipeline_slice import (
        _build_generation_payload, _build_adapter_plan,
    )
    plan = _fake_raw_plan_minimal()
    adapter = _build_adapter_plan(plan, _fake_location_profiles())
    payload = _build_generation_payload(plan, adapter)
    assert len(payload["per_shot"]) == 4
    for entry in payload["per_shot"]:
        for k in ("shot_key", "bg_id", "loc_id", "bg_intent_text",
                  "expected_attached_meta", "expected_reference_phrase_kinds",
                  "expected_label_placeholder"):
            assert k in entry
        assert entry["expected_attached_meta"] == [{"kind": "background", "id": entry["bg_id"]}]
        assert entry["expected_reference_phrase_kinds"] == ["background"]
    assert isinstance(payload["unmapped_fields"], list)


def test_reference_input_plan_covers_all_selected_shots():
    from experiment_background_pipeline_slice import (
        _build_reference_input_plan, _build_adapter_plan,
    )
    plan = _fake_raw_plan_minimal()
    adapter = _build_adapter_plan(plan, _fake_location_profiles())
    rip = _build_reference_input_plan(plan, adapter)
    shot_keys = {e["shot_key"] for e in rip["entries"]}
    assert shot_keys == {"S01_Shot1", "S01_Shot2", "S02_Shot3", "S03_Shot4"}
    for e in rip["entries"]:
        assert e["ref_contract"] == {"kind": "background", "id": e["bg_id"], "policy": "required"}


def test_ref_contract_dry_run_pass_for_all_shots():
    """validate_attached_refs 가 모든 shot 에 대해 통과해야 함."""
    from experiment_background_pipeline_slice import (
        _build_adapter_plan, _build_reference_input_plan, _ref_contract_dry_run,
    )
    plan = _fake_raw_plan_minimal()
    adapter = _build_adapter_plan(plan, _fake_location_profiles())
    rip = _build_reference_input_plan(plan, adapter)
    result = _ref_contract_dry_run(rip, adapter)
    assert result["all_pass"] is True
    assert len(result["failures"]) == 0
```

- [ ] **Step 2: 실패 확인**

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

```python
def _build_generation_payload(raw_plan: dict, adapter: dict) -> dict:
    """background_render_step 이 받을 contract 의 최소 mirror.
    실제 호출 0. unmapped_fields 는 production contract 에서 우리가 emit 못 한 필드.
    background_catalog 는 Dict[bg_id, entry] (SOT). shot_key 가 shot id SOT.
    """
    intent_to_bg = adapter["intent_key_to_bg_id"]
    bg_to_entry = adapter["background_catalog"]  # 이미 Dict[bg_id, entry]
    prompt_by_shot = {p["shot_key"]: p["bg_intent_text"]
                      for p in raw_plan.get("prompt_payload_intents", [])}
    per_shot = []
    for sb in raw_plan.get("shot_binding_intents", []):
        bg = intent_to_bg.get(sb["intent_key"])
        if bg is None:
            continue
        entry = bg_to_entry[bg]
        per_shot.append({
            "shot_key": sb["shot_key"],
            "bg_id": bg,
            "loc_id": entry["loc_id"],
            "bg_intent_text": prompt_by_shot.get(sb["shot_key"], ""),
            "expected_attached_meta": [{"kind": "background", "id": bg}],
            "expected_reference_phrase_kinds": ["background"],
            "expected_label_placeholder": f"chain_bg:{bg}",
        })
    # production contract 와 비교 — unmapped fields 명시
    unmapped = [
        "image_bytes",  # 실제 PNG (실험에서는 placeholder)
        "camera_recommendations",  # floor_plan_prompt 산출 (이번 W1 미생성)
        "shot_guides",  # background_prompt 산출 (이번 W1 미생성)
        "ref_used",  # background_render 산출 (실제 PNG 소비 후)
    ]
    return {"per_shot": per_shot, "unmapped_fields": unmapped}


def _build_reference_input_plan(raw_plan: dict, adapter: dict) -> dict:
    intent_to_bg = adapter["intent_key_to_bg_id"]
    bg_to_entry = adapter["background_catalog"]  # Dict[bg_id, entry]
    entries = []
    for sb in raw_plan.get("shot_binding_intents", []):
        bg = intent_to_bg.get(sb["intent_key"])
        if bg is None:
            continue
        entry = bg_to_entry[bg]
        entries.append({
            "shot_key": sb["shot_key"],
            "bg_id": bg,
            "loc_id": entry["loc_id"],
            "ref_contract": {"kind": "background", "id": bg, "policy": "required"},
            "expected_attached_meta": [{"kind": "background", "id": bg}],
            "reference_phrase_kinds": ["background"],
        })
    return {"entries": entries}


def _ref_contract_dry_run(rip: dict, adapter: dict) -> dict:
    """production helper validate_attached_refs 로 각 shot dry-run.
    background_catalog 는 Dict[bg_id, entry] (SOT).
    """
    sys.path.insert(0, str(_BACKEND_ROOT))
    from app.core.ref_contract_validator import validate_attached_refs, RefContractError  # type: ignore

    unit_to_loc = {bg: entry["loc_id"]
                   for bg, entry in adapter["background_catalog"].items()}
    chain_bg_lookup = lambda bg: unit_to_loc.get(bg)

    failures = []
    for e in rip["entries"]:
        bg = e["bg_id"]
        # production enum SOT (render_prompt_card.py:462-466):
        # readiness_policy ∈ {block_if_missing, skipped_by_policy, not_applicable}
        minimal_rpc = {
            "asset_requirements": {
                "required_refs": [{"kind": "background", "id": bg, "policy": "required"}],
                "forbidden_refs": [],
                "readiness_policy": "block_if_missing",
            },
        }
        labeled_refs = [(f"chain_bg:{bg}", b"")]
        attached_meta = [("background", bg)]
        try:
            validate_attached_refs(
                minimal_rpc, labeled_refs, attached_meta,
                prompt="dry-run", is_close_framing=False,
                chain_bg_lookup=chain_bg_lookup,
                reference_phrase_kinds=["background"],
            )
        except RefContractError as exc:
            failures.append({"shot_key": e["shot_key"], "bg_id": bg, "error": str(exc)})
    return {"all_pass": len(failures) == 0, "failures": failures}
```

- [ ] **Step 4: PASS 확인**

Run: `cd backend && pytest tests/scripts/test_experiment_background_pipeline_slice.py -v`
Expected: 20 PASS.

---

## Phase 7 — pipeline_compatibility_report (9 invariant aggregator) (3 tests)

### Task 8: compatibility report + acceptance gates

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

```python
def test_compatibility_report_aggregates_nine_invariants():
    """fake bundle 의 selected_shots 가 fake_raw_plan_minimal 의 shot_key 와 일치."""
    import hashlib
    shot_keys = ["S01_Shot1", "S01_Shot2", "S02_Shot3", "S03_Shot4"]
    selected = [{"row_id": f"u{i}", "shot_key": sk, "scene_index": int(sk[1:3]),
                 "shot_index": int(sk.split("Shot")[1]), "is_selected": True,
                 "shot_description": ""}
                for i, sk in enumerate(shot_keys)]
    canon = json.dumps(selected, sort_keys=True, ensure_ascii=False).encode("utf-8")
    bundle = {"selected_shot_count": len(selected), "selected_shots": selected,
              "source_hash": hashlib.sha256(canon).hexdigest()[:16]}
    from experiment_background_pipeline_slice import (
        _build_compatibility_report, _build_adapter_plan,
        _build_reference_input_plan, _ref_contract_dry_run,
    )
    plan = _fake_raw_plan_minimal()
    adapter = _build_adapter_plan(plan, _fake_location_profiles())
    rip = _build_reference_input_plan(plan, adapter)
    ref_dry = _ref_contract_dry_run(rip, adapter)
    report = _build_compatibility_report(
        bundle=bundle, raw_plan=plan, adapter=adapter, rip=rip,
        ref_dry=ref_dry, pmap_check={"all_paths_exist": True, "all_symbols_found": True, "missing": []},
        production_diff_empty=True, db_write_count=0, image_import_seen=False,
    )
    for inv in (
        "production_diff_zero", "db_write_zero", "image_call_zero",
        "source_verbatim", "all_selected_shots_covered",
        "bg_id_assigned_by_production_helper", "shot_background_map_n_to_1",
        "reference_chain_acyclic", "ref_contract_dry_run_pass",
    ):
        assert inv in report["invariants"]


def test_compatibility_report_fails_when_unmatched_shot():
    import hashlib
    selected = [{"row_id": f"u{i}", "shot_key": f"S99_Shot{i}", "scene_index": 99,
                 "shot_index": i, "is_selected": True} for i in range(1, 6)]
    canon = json.dumps(selected, sort_keys=True, ensure_ascii=False).encode("utf-8")
    bundle = {"selected_shot_count": 5, "selected_shots": selected,
              "source_hash": hashlib.sha256(canon).hexdigest()[:16]}
    from experiment_background_pipeline_slice import (
        _build_compatibility_report, _build_adapter_plan,
        _build_reference_input_plan, _ref_contract_dry_run,
    )
    plan = _fake_raw_plan_minimal()
    adapter = _build_adapter_plan(plan, _fake_location_profiles())
    rip = _build_reference_input_plan(plan, adapter)
    ref_dry = _ref_contract_dry_run(rip, adapter)
    report = _build_compatibility_report(
        bundle=bundle, raw_plan=plan, adapter=adapter, rip=rip, ref_dry=ref_dry,
        pmap_check={"all_paths_exist": True, "all_symbols_found": True, "missing": []},
        production_diff_empty=True, db_write_count=0, image_import_seen=False,
    )
    assert report["invariants"]["all_selected_shots_covered"]["pass"] is False
    assert report["all_pass"] is False


def test_runner_exits_1_on_any_invariant_fail(tmp_path):
    out_root = tmp_path / "out"
    # 강제 fail mode — DB 가 없는 가짜 project_id
    result = _run([
        "--dry-run",
        "--output-root", str(out_root),
        "--project-id", "ZZZZZZZZ-0000-0000-0000-000000000000",
        "--episode-id", "ZZZZZZZZ-0000-0000-0000-000000000000",
    ])
    assert result.returncode == 1
    rm = json.loads(next(out_root.glob("*/run_meta.json")).read_text())
    assert rm["run_status"] == "validation_failed"
    assert rm["exit_code"] == 1
    assert len(rm["failed_invariants"]) > 0
```

- [ ] **Step 2: 실패 확인**

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

```python
def _build_compatibility_report(*, bundle, raw_plan, adapter, rip, ref_dry,
                                pmap_check, production_diff_empty, db_write_count, image_import_seen):
    inv = {}

    # 1
    inv["production_diff_zero"] = {"pass": production_diff_empty,
                                   "detail": "git diff backend/app backend/alembic empty"}
    # 2
    inv["db_write_zero"] = {"pass": db_write_count == 0, "detail": f"writes={db_write_count}"}
    # 3
    inv["image_call_zero"] = {"pass": not image_import_seen,
                              "detail": "no gemini_image_client / fal / gpt-image module import"}
    # 4
    import hashlib
    canon = json.dumps(bundle["selected_shots"], sort_keys=True, ensure_ascii=False).encode("utf-8")
    inv["source_verbatim"] = {"pass": bundle["source_hash"] == hashlib.sha256(canon).hexdigest()[:16],
                              "detail": f"hash={bundle['source_hash']}"}
    # 5 — shot_key SOT (composite). row_id 는 사용 안 함.
    shot_keys = {s["shot_key"] for s in bundle["selected_shots"]}
    binding_keys = {e["shot_key"] for e in rip["entries"]}
    missing = shot_keys - binding_keys
    inv["all_selected_shots_covered"] = {"pass": len(missing) == 0,
                                         "detail": {"missing": sorted(missing)[:10]}}
    # 6
    sys.path.insert(0, str(_BACKEND_ROOT))
    from app.core.bg_state_vocab import BG_ID_RE  # type: ignore
    # background_catalog 는 Dict[bg_id, entry] — keys 가 곧 bg_id 목록
    bg_ok = all(BG_ID_RE.match(bg) for bg in adapter["background_catalog"].keys())
    inv["bg_id_assigned_by_production_helper"] = {"pass": bg_ok,
                                                  "detail": "all bg_id match BG_ID_RE"}
    # 7 — shot_background_map already raised on N:1; if we got here, it passed
    inv["shot_background_map_n_to_1"] = {"pass": True, "detail": "build_shot_background_map success"}
    # 8 — gen_order acyclic check (already raises on cycle); cross-check coverage
    cat_bg_ids = set(adapter["background_catalog"].keys())
    order_bg_ids = set(adapter["gen_order"]) - {fp for fp in adapter["gen_order"] if fp.startswith("fp_")}
    inv["reference_chain_acyclic"] = {"pass": order_bg_ids == cat_bg_ids,
                                      "detail": "gen_order covers all bg_id"}
    # 9
    inv["ref_contract_dry_run_pass"] = {"pass": ref_dry["all_pass"],
                                        "detail": {"failures": ref_dry["failures"][:5]}}

    all_pass = all(v["pass"] for v in inv.values()) and pmap_check["all_paths_exist"] and pmap_check["all_symbols_found"]
    return {"invariants": inv, "all_pass": all_pass,
            "production_pipeline_map_check": pmap_check}


# main() 안에 통합:
# adapter, rip, ref_dry 계산 완료 후
report = _build_compatibility_report(
    bundle=bundle, raw_plan=raw_plan["plan"], adapter=adapter, rip=rip,
    ref_dry=ref_dry, pmap_check=pmap_check,
    production_diff_empty=_check_production_diff_empty(),
    db_write_count=_DB_WRITE_COUNT,
    image_import_seen=_check_image_imports_present(),
)
(run_dir / "pipeline_compatibility_report.json").write_text(
    json.dumps(report, ensure_ascii=False, indent=2))
outputs.append("pipeline_compatibility_report.json")
for name, v in report["invariants"].items():
    if not v["pass"]:
        failed_invariants.append(name)
if failed_invariants:
    run_status = "validation_failed"; exit_code = 1
    # quarantine raw plan
    if raw_plan.get("status") == "generated":
        (run_dir / "llm_raw_plan_quarantined.json").write_text(
            json.dumps(raw_plan, ensure_ascii=False, indent=2))
        outputs.append("llm_raw_plan_quarantined.json")
```

DB write sentinel — SessionLocal wrap 시점에 SQLAlchemy event:

```python
_DB_WRITE_COUNT = 0

def _install_db_write_sentinel():
    """SessionLocal 사용 직전 호출. session 의 flush 시 new/dirty/deleted 카운트.

    Runtime sentinel — SQLAlchemy event listener.
    추가 static guard: 이 script source 에 'session.commit'/'session.add'/
    'session.delete'/'session.merge'/'session.flush' literal 이 등장하지 않는지
    test 가 별도 검사 (test_static_guard_no_db_write_method_calls).
    """
    from sqlalchemy import event  # type: ignore
    from app.core.database import SessionLocal  # type: ignore
    from sqlalchemy.orm import Session as _SASession  # type: ignore

    # SessionLocal 은 sessionmaker — 안전하게 Session 클래스에 listen.
    @event.listens_for(_SASession, "after_flush")
    def _after_flush(session, flush_context):
        global _DB_WRITE_COUNT
        _DB_WRITE_COUNT += len(session.new) + len(session.dirty) + len(session.deleted)


def _check_production_diff_empty() -> bool:
    import subprocess
    r = subprocess.run(["git", "diff", "--stat", "backend/app", "backend/alembic"],
                       cwd=str(_REPO_ROOT), capture_output=True, text=True)
    return r.returncode == 0 and r.stdout.strip() == ""


def _check_image_imports_present() -> bool:
    banned = ("app.modules.llm.gemini_image_client",
              "app.services.fal_angle_helpers",
              "app.modules.gemini_i2i_editor")
    return any(b in sys.modules for b in banned)
```

- [ ] **Step 4: PASS 확인**

Run: `cd backend && pytest tests/scripts/test_experiment_background_pipeline_slice.py -v`
Expected: 23 PASS.

---

## Phase 8 — HTML diagnostic + run_meta finalization (2 tests)

### Task 9: minimal HTML

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

```python
def test_index_html_shows_key_metrics(tmp_path):
    out_root = tmp_path / "out"
    _run([
        "--dry-run",
        "--output-root", str(out_root),
        "--project-id", "6cb862d9-590c-4dce-86e6-d10c2977db19",
        "--episode-id", "08ad2cd3-3e96-4d84-808f-869ee628473c",
    ])
    html = next(out_root.glob("*/index.html")).read_text()
    # 첫 화면 metric
    for needle in ("unit count", "shot coverage", "reference chain",
                   "validation", "payload sample"):
        assert needle in html.lower(), f"index.html missing '{needle}'"
    # raw JSON 은 접혀있어야 함 — <details> 사용
    assert "<details" in html


def test_index_html_renders_without_breaking_on_empty_bundle(tmp_path):
    """fail mode 에서도 HTML 이 깨지지 않고 validation_failed 상태를 보여줘야 함."""
    out_root = tmp_path / "out"
    _run([
        "--dry-run",
        "--output-root", str(out_root),
        "--project-id", "ZZZZZZZZ-0000-0000-0000-000000000000",
        "--episode-id", "ZZZZZZZZ-0000-0000-0000-000000000000",
    ])
    html = next(out_root.glob("*/index.html")).read_text()
    assert "validation_failed" in html.lower() or "failed_invariants" in html.lower()
```

- [ ] **Step 2: 실패 확인**

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

```python
def _render_html(run_meta: dict, bundle, adapter, rip, report, run_dir: Path) -> None:
    def esc(x): return (str(x).replace("&", "&amp;").replace("<", "&lt;").replace(">", "&gt;"))
    unit_count = len(adapter.get("background_catalog", [])) if adapter else 0
    shot_total = bundle.get("selected_shot_count", 0) if bundle else 0
    shot_covered = len(rip.get("entries", [])) if rip else 0
    chain_len = len(adapter.get("gen_order", [])) if adapter else 0
    inv = (report or {}).get("invariants", {}) or {}
    inv_rows = "".join(
        f"<tr><td>{esc(k)}</td><td>{'PASS' if v['pass'] else 'FAIL'}</td></tr>"
        for k, v in inv.items()
    )
    payload_sample = ""
    payload_path = run_dir / "generation_payload_dry_run.json"
    if payload_path.exists():
        sample = json.loads(payload_path.read_text())
        first = sample.get("per_shot", [])[:2]
        payload_sample = esc(json.dumps(first, ensure_ascii=False, indent=2))

    html = f"""<!doctype html><html><head><meta charset="utf-8">
<title>background_pipeline_slice {esc(run_meta.get('run_id'))}</title>
<style>body{{font-family:sans-serif;margin:1.5em;}}
table{{border-collapse:collapse}} td,th{{border:1px solid #ccc;padding:4px 8px}}
.metric{{display:inline-block;margin:0 1em 1em 0;padding:1em;border:1px solid #ddd;border-radius:6px}}
.fail{{color:#b00}} .pass{{color:#080}}</style></head>
<body>
<h1>background_pipeline_slice — {esc(run_meta.get('run_id'))}</h1>
<p>plan_version: <b>{esc(run_meta.get('plan_version'))}</b> | run_status:
<b class="{'fail' if run_meta.get('run_status')!='succeeded' else 'pass'}">{esc(run_meta.get('run_status'))}</b>
| exit_code: {esc(run_meta.get('exit_code'))}</p>
<div>
<div class="metric"><div>unit count</div><b>{unit_count}</b></div>
<div class="metric"><div>shot coverage</div><b>{shot_covered} / {shot_total}</b></div>
<div class="metric"><div>reference chain</div><b>{chain_len} steps</b></div>
<div class="metric"><div>validation</div><b>{'PASS' if report and report.get('all_pass') else 'FAIL'}</b></div>
</div>
<h2>Acceptance gates</h2>
<table><tr><th>invariant</th><th>status</th></tr>{inv_rows}</table>
<h2>Failed invariants</h2><pre>{esc(json.dumps(run_meta.get('failed_invariants', []), ensure_ascii=False, indent=2))}</pre>
<h2>payload sample (first 2)</h2><pre>{payload_sample}</pre>
<details><summary>raw run_meta.json</summary><pre>{esc(json.dumps(run_meta, ensure_ascii=False, indent=2))}</pre></details>
<details><summary>raw adapter plan (HTML preview only, slice 50000 chars — 원본 JSON 은 production_adapter_plan.json 전체 보존, slice/truncate 0)</summary><pre>{esc(json.dumps(adapter, ensure_ascii=False, indent=2))[:50000]}</pre></details>
</body></html>"""
    (run_dir / "index.html").write_text(html)


# main() 끝에:
_render_html(run_meta, bundle if not args.skip_db else None,
             adapter if 'adapter' in locals() else None,
             rip if 'rip' in locals() else None,
             report if 'report' in locals() else None,
             run_dir)
outputs.append("index.html")
```

- [ ] **Step 4: PASS 확인**

Run: `cd backend && pytest tests/scripts/test_experiment_background_pipeline_slice.py -v`
Expected: 25 PASS.

---

## Phase 9 — Fixture-swap test (1 test) + sentinel test (2 tests)

### Task 10: fixture-swap + no-image-import sentinel + production diff sentinel

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

```python
def test_fixture_swap_alternate_episode(tmp_path):
    """동일 코드가 다른 episode 에서도 동작해야 함 — 특정 location/캐릭터 literal 없음.
    alternate 는 source_bundle 단계까지만 (Codex Q4 합의 — generate 비용 절감)."""
    out_root = tmp_path / "out"
    result = _run([
        "--dry-run",
        "--output-root", str(out_root),
        "--project-id", "76212b49-bf96-45fb-8c56-cd8d8ae02dda",
        "--episode-id", "f44339e6-1bd1-4d10-8a8f-2e30e5a1d36d",
        "--stop-after", "source_bundle",
    ])
    assert result.returncode in (0, 1), result.stderr
    bundle = json.loads(next(out_root.glob("*/source_bundle.json")).read_text())
    assert bundle["selected_shot_count"] > 0
    # script source 안에 sample-specific literal 0 확인 (Codex 추가 list 반영)
    src = (_REPO_ROOT / "backend/scripts/experiment_background_pipeline_slice.py").read_text()
    for banned in ("옥탑방", "L05", "rooftop", "수리영", "민숙", "석창포"):
        assert banned not in src, f"banned literal '{banned}' in script source"


def test_static_guard_no_db_write_method_calls():
    """script source 안에 session 의 write method 호출 literal 0건.
    DB read-only 보장 — IMPORTANT 3 (Codex)."""
    src = (_REPO_ROOT / "backend/scripts/experiment_background_pipeline_slice.py").read_text()
    for banned in ("session.commit", "session.add", "session.delete",
                   "session.merge", "session.flush", ".commit(", ".add("):
        # comment 안에서 written-about 은 허용 — 실제 callable 일 때만 잡으려면 AST 가 더 정확.
        # 최소 보호 단계로 literal 검사.
        if banned in src:
            # 정밀히 — 같은 줄이 comment 'NOTE: never call session.commit' 이면 허용.
            for line in src.splitlines():
                if banned in line and not line.lstrip().startswith("#"):
                    raise AssertionError(f"banned write call '{banned}' in non-comment line: {line.strip()}")


def test_no_banned_image_modules_imported_at_runtime(tmp_path):
    """sentinel — 어떤 모드에서도 image 모듈 import 가 일어나면 안 됨."""
    out_root = tmp_path / "out"
    result = _run([
        "--dry-run",
        "--output-root", str(out_root),
        "--project-id", "6cb862d9-590c-4dce-86e6-d10c2977db19",
        "--episode-id", "08ad2cd3-3e96-4d84-808f-869ee628473c",
        "--diag-print-imports",
    ])
    banned = ("gemini_image_client", "fal_angle_helpers", "gemini_i2i_editor")
    for b in banned:
        assert b not in result.stderr, f"banned module imported: {b}"


def test_production_diff_empty_invariant_recorded(tmp_path):
    """compatibility report 에 production_diff_zero invariant 가 PASS 로 기록되어야 함
    (이 PR 가 production code 를 만지지 않으므로)."""
    out_root = tmp_path / "out"
    _run([
        "--dry-run",
        "--output-root", str(out_root),
        "--project-id", "6cb862d9-590c-4dce-86e6-d10c2977db19",
        "--episode-id", "08ad2cd3-3e96-4d84-808f-869ee628473c",
    ])
    report = json.loads(next(out_root.glob("*/pipeline_compatibility_report.json")).read_text())
    assert report["invariants"]["production_diff_zero"]["pass"] is True
```

- [ ] **Step 2: 실패 확인**

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

위 Phase 들의 코드로 이미 대부분 보장. fixture swap test 가 통과하려면 `--stop-after source_bundle` 이 다른 episode 에서도 잘 동작해야 함. 이미 그렇게 구현됨.

- [ ] **Step 4: PASS 확인**

Run: `cd backend && pytest tests/scripts/test_experiment_background_pipeline_slice.py -v`
Expected: 32 PASS (총 32개, v3 BLOCKING 1/2 추가 test 3개로 29→32).

| Phase | 신규 test 수 | 누적 |
|------|------|------|
| 1 skeleton | 3 | 3 |
| 2 source_bundle | 3 | 6 |
| 3 pipeline_map | 3 | 9 |
| 4 llm_raw_plan | 5 (placeholder/schema/no-litellm-in-dry/open_risks/state_class_enum_drift) | 14 |
| 5 adapter | 6 (helper/deterministic/n_to_1/topo + v3 dedup_conflict + self_cycle) | 20 |
| 6 payload+rip | 3 | 23 |
| 7 compatibility_report | 3 | 26 |
| 8 html | 2 | 28 |
| 9 fixture-swap + sentinels | 4 (fixture_swap + no_banned_image + production_diff_invariant + static_guard_no_db_write) | 32 |

---

## Run sequence (after all tests pass)

### Dry-run (LLM 없음)

```bash
cd /Users/manta/Documents/Projects/TheRoad-I1
.venv/bin/python backend/scripts/experiment_background_pipeline_slice.py \
  --project-id 6cb862d9-590c-4dce-86e6-d10c2977db19 \
  --episode-id 08ad2cd3-3e96-4d84-808f-869ee628473c
```

기대: source_bundle / production_pipeline_map / llm_raw_plan(placeholder) / adapter (빈 catalog) / payload (빈) / reference_input_plan (빈) / compatibility_report (대부분 PASS, 단 unit count=0 으로 인한 acceptance 실패 가능 — 이 경우 dry-run 에서는 expected, run_status=`validation_failed` exit 1).

### Generate (LLM 1회)

```bash
.venv/bin/python backend/scripts/experiment_background_pipeline_slice.py \
  --generate --model gemini-3.5-flash \
  --project-id 6cb862d9-590c-4dce-86e6-d10c2977db19 \
  --episode-id 08ad2cd3-3e96-4d84-808f-869ee628473c
```

성공 시 LLM raw plan 가 채워지고 9 invariant 모두 PASS → exit 0. LLM 가 schema fail 시 1회 retry, 그래도 실패면 quarantine + exit 1.

### Regression
```bash
cd backend && pytest tests/scripts/ -v
```

---

## Self-Review Checklist

1. **Spec coverage**: 사용자가 요구한 필수 산출물 (source_bundle / production_pipeline_map / llm_raw_plan ↔ background_unit_plan / shot_bindings / reference_chain / generation_payload_dry_run / reference_input_plan / pipeline_compatibility_report / index.html / run_meta) 모두 phase 안에 들어감. ✓
2. **Codex 제약**: 2-layer (raw + adapter), LLM 절대 bg_id 안 만듦, gen_order code 산출, depends_on_fp code derive (fp_intent_key→fp_id), depends_on_bg LLM intent → code resolve, ref_contract dry-run exact `('background', bg_id)`, open_risks max 5 + conservative_fallback, validation_failed exit 1. ✓
3. **사용자 W1 축소 지시**: ~28 test, ~10 touchpoint, 9 필수 invariant, HTML 첫 화면 metric+payload sample+folded raw. ✓
4. **금지 사항**: commit/push step 0, sample-specific scenario literal 0 (script 자체에 검사, Phase 9 banned-literals test), regex semantic 0 (BG_ID_RE 는 ID-format 예외), DB write 0 (sentinel + static guard), image call 0 (sentinel + module import sentinel), 사람 결정 enum 0 (open_risks 만, conservative_fallback 필수). ✓
5. **type consistency**: `intent_key` (LLM) ↔ `bg_id` (code) ↔ `shot_id` 일관. `fp_intent_key` ↔ `fp_id` 일관. `chain_bg_lookup` 시그니처 `lambda bg_id: loc_id` 일관. ✓
6. **placeholder scan**: 모든 step 에 코드 또는 명령 포함, "TBD"/"implement later" 0. ✓

---

## Execution Handoff

Plan 완료. 저장 위치: `docs/superpowers/plans/2026-05-24-background-pipeline-slice-experiment.md`.

**Codex 리뷰 요청 → APPROVED_FOR_IMPLEMENTATION 받은 후** subagent-driven (Opus 4.7) 으로 phase 1→9 순차 실행. 각 phase 끝에서 target test PASS 확인. 모든 phase 통과 후 dry-run + generate 1회 실행 → 결과로 Codex 최종 리뷰 요청.

**commit/push 금지** — 사용자/Codex 승인 후 별도 turn 에서만.