# Single-vs-Batch Reference Contract Fix — Implementation Plan > **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking. **Version**: v2 (PATCHED — self-audit IMPORTANT 3 반영: Task 1 helper signature 정정 + Task 2 build_single_still_context 구체화 + Task 5 N+1 batch lookup) **Goal**: 단건 image gen 경로의 ref attach 결함 (chain_bg/prev_shot_ref/state_variant 누락) + lineage divergence + face close-up exemption 과다 fix. manual prompt patch **금지**. **Architecture**: 6 Task 직렬 — Task 1 helper extract → Task 2 single path 사용 → Task 3 validate_attached_refs fail-fast → Task 4 face/eyes exemption 축소 → Task 5 ImageResponse.actual_attached_refs → Task 6 regression fixture. **Tech Stack**: Python 3.12 / FastAPI / SQLAlchemy + raw SQL (PostgreSQL) / pytest / alembic. spec: `docs/superpowers/specs/2026-05-08-single-batch-reference-contract-design.md` (DRAFT_PATCHED_R1). > **Non-goals enforcement** (이 plan 의 모든 task 에서 절대 추가 금지): > - [ ] prop 신규 schema 필드 (scene_detail v22 narrative_critical_props 등) 추가 X > - [ ] image_asset_ref_attachment 별도 테이블 X > - [ ] scene_detail manifest 재생성 / cleaned source LLM derive X > - [ ] 샷별 prompt 수동 보정 X (어떤 task 든) > - [ ] LVM cost/quality / scene_detail prompt v22 변경 X > - [ ] image_asset.reference_image_ids 컬럼명 alembic rename X > > 매 task 시작 시 reviewer 가 위 6개 위반 안 했는지 confirm. 위반 발견 시 task 중단 + spec 변경 협의. --- ## File Structure ### Task 1 — helper extract - **Modify**: `backend/app/services/scene_generation_coordinator.py` — `build_scene_attached_refs(...)` 신규 helper 함수 추가, `generate_images` (line 178-326) 의 ref 빌드 블록을 helper 호출로 교체. 단건 경로는 Task 2 에서 transition. - **Create**: `backend/tests/unit/test_build_scene_attached_refs.py` — helper unit test (chain_bg insert / close skip / prev_shot fallback / state_variant / image index). ### Task 2 — single path uses helper - **Modify**: `backend/app/services/scene_generation_coordinator.py:677-739` — `_build_single_scene_prompt_and_refs` 가 helper 호출 + 단건 input prep (`background_chain_bg_map` load, `entity_lookup`, `staging`, `state_variant_sids` build). - **Modify**: `backend/app/services/scene_image_service.py:603-699` — `generate_single_scene_image` 가 단건 input prep 흐름과 호환. - **Create**: `backend/tests/integration/test_single_scene_image_uses_helper.py` — single 호출 결과 actual labeled_refs == batch 호출 결과 동등. ### Task 3 — validate_attached_refs fail-fast - **Create**: `backend/app/core/ref_contract_validator.py` — `validate_attached_refs(rpc, labeled_refs, prompt, is_close_framing)` + `RefContractError` exception class + classifier-based "from the reference" guard. - **Modify**: `backend/app/modules/pipeline/scene_image_pipeline.py` — `generate_and_validate_scene` 직전 `validate_attached_refs` 호출. - **Modify**: `backend/app/core/errors.py` — `RefContractError` export. - **Create**: `backend/tests/unit/test_ref_contract_validator.py` — character/background/object required 검사. - **Create**: `backend/tests/unit/test_from_the_reference_classifier.py` — phrase position-based classifier. ### Task 4 — face/eyes close-up exemption 축소 - **Modify**: `backend/app/core/visible_entities_validator.py:131-199` — `_is_face_close_up(prompt) -> bool` 신규 helper + `_forward_enforcement_exempt()` 안 face close-up 면제 거부 로직. - **Create**: `backend/tests/unit/test_face_close_up_detection.py` — pattern 매칭 + body close-up false positive 방지. - **Create**: `backend/tests/unit/test_forward_enforcement_face_exemption.py` — exemption 거부 시 forward enforcement fail. ### Task 5 — ImageResponse.actual_attached_refs - **Modify**: `backend/app/schemas/image.py:36` — `ImageResponse` 에 `actual_attached_refs: Optional[list[str]] = None` field 추가 + `reference_image_ids` description 명시 ("(lineage)..."). - **Modify**: `backend/app/services/image_service_helpers.py:182` — `image_to_dict` 가 `actual_attached_refs` join (llm_call_log lookup). - **Modify**: `backend/app/models/project.py:187` (image_asset.reference_image_ids 컬럼 docstring) + `backend/app/models/project.py:288` (llm_call_log.reference_image_ids 컬럼 docstring). - **Create**: `backend/tests/unit/test_image_response_actual_attached_refs.py` — schema field + image_to_dict join. ### Task 6 — Regression / canary fixture - **Create**: `backend/tests/fixtures/c01_zero_gate_regression.json` — S8/S13/S15/S19/S21 still_id + visible_entities + 의도된 actual labeled_refs label set. - **Create**: `backend/tests/integration/test_c01_zero_gate_regression.py` — fixture 기반 contract test (single + batch 양쪽 동등). - **Create**: `backend/scripts/canary_single_vs_batch_refs.py` — daily canary script (PID `80f62523` 5 still 상태 점검). --- ## Task 1: `build_scene_attached_refs` helper extract > **Plan drift note (post-execution)**: Task 1 helper 는 **single-scene consumption form** (return `(full_prompt, labeled_refs)` 2-tuple) 으로 먼저 도입. Batch path call-site migration (Task 1 Step 4) 은 batch `_generate_one_scene` 의 **N-variation rewrite contract** (variations[1:] 별 `_rewrite_t2i_helper(var_t2i, sid_to_img, sid_info)` 호출) 때문에 helper signature 정정 또는 별 후속 task 로 분리. 기존 batch behavior 는 아직 변경하지 않았고 회귀 없음 (commit 776ceab broader regression 57/57 PASS). **Files:** - Modify: `backend/app/services/scene_generation_coordinator.py` - Test: `backend/tests/services/test_build_scene_attached_refs.py` (신규 — plan path `tests/unit/` → 실제 프로젝트 패턴 `tests/services/`) **AC:** AC-1 (helper 가 batch 동작 동등 보존). spec §4.1 5a~5d 분기 보존. **Non-goals confirm**: prop 신규 schema X / 별도 테이블 X / manifest 재생성 X / 수동 prompt 보정 X / LVM/v22 변경 X / 컬럼 rename X. ✓ - [ ] **Step 1: failing test 작성** ```python # backend/tests/unit/test_build_scene_attached_refs.py """build_scene_attached_refs helper — Task 1. 배치 generate_images 의 ref 빌드 블록(coordinator.py:178-326) 동등 보존: chain_bg insert / close skip / prev_shot_ref fallback / state_variant / image index 순서. """ import pytest from unittest.mock import MagicMock, patch def _make_helper_inputs(*, has_chain_bg: bool, is_close: bool, has_prev_shot: bool): """helper input dict 빌드 — 분기별 fixture.""" bc_bg = {"label": "BACKGROUND chain reference", "image_bytes": b"bg"} if has_chain_bg else None return { "still": MagicMock(camera_json="{}", scene_index=8, still_index=4, shot_index=4), "episode_id": "ep_test", "still_data": {"scene_index": 8, "shot_index": 4, "still_frame_prompt": "test", "beat_title": ""}, "visible_entities": [{"id": "c01", "short_id": "C01", "entity_type": "character", "name": "C1"}], "ref_image_map": {"c01": b"face_ref"}, "background_chain_bg_map": {"8_4": bc_bg} if bc_bg else {}, "best_prev_bytes": b"prev_shot" if has_prev_shot else None, "current_location_ids": [], "dep_scene_id": None, "stills": [], "location_scene_history": {}, "dep_detail_map": {}, "staging": {"camera_direction": "extreme close-up" if is_close else "medium shot"}, "state_variant_sids": {}, "entity_lookup": {"c01": {"id": "c01", "short_id": "C01", "entity_type": "character", "name": "C1"}}, } def test_helper_chain_bg_inserts_when_not_close_framing(): """5a: chain_bg + NOT close → labeled_refs[0] == chain_bg label.""" from app.services.scene_generation_coordinator import build_scene_attached_refs inputs = _make_helper_inputs(has_chain_bg=True, is_close=False, has_prev_shot=False) # mock _build_final_scene_prompt + reference svc dependencies with patch("app.services.scene_generation_coordinator._build_final_scene_prompt", return_value="final"): # ... full prompt + labeled_refs (label, bytes) tuple list full_prompt, labeled_refs = build_scene_attached_refs(**inputs) assert labeled_refs[0][0] == "BACKGROUND chain reference" def test_helper_chain_bg_skipped_when_close_framing_falls_back_to_prev_shot(): """5b: chain_bg + close → chain skip + prev_shot try.""" inputs = _make_helper_inputs(has_chain_bg=True, is_close=True, has_prev_shot=True) # ... # assert labeled_refs[0][0] startswith "previous shot" def test_helper_no_chain_bg_falls_back_to_prev_shot(): """5c: chain 부재 → prev_shot try.""" # ... def test_helper_entity_only_when_no_bg_no_prev_shot(): """5d: prev_shot 없음 → entity-only fallback (silent skip 아님, 명시적 path).""" inputs = _make_helper_inputs(has_chain_bg=False, is_close=False, has_prev_shot=False) # ... # assert no BACKGROUND/previous shot labels ``` - [ ] **Step 2: 테스트 실행 — fail 확인** Run: `cd backend && PYTHONPATH=. .venv/bin/pytest tests/unit/test_build_scene_attached_refs.py -v` Expected: FAIL — `build_scene_attached_refs` 함수 미존재. - [ ] **Step 3: helper 함수 구현 (`coordinator.py` 추가)** `coordinator.py:178-326` 의 ref 빌드 블록을 신규 함수 `build_scene_attached_refs(...)` 로 lift. 함수 signature (v2 정정): ```python def build_scene_attached_refs( *, still: Any, episode_id: str, still_data: Dict[str, Any], visible_entities: List[Dict[str, Any]], ref_image_map: Dict[str, bytes], # ── per-episode cached (caller 가 1회 build 후 reuse — N+1 회피) ── cached_style_context: str, cached_entity_text_map: Dict[str, str], scene_paths_by_index_by_id: Dict[str, Path], # dep_scene_id → previous-scene PNG path location_scene_history: Dict[str, Any], # location_id → (bytes, meta) background_chain_bg_map: Dict[str, Any], dep_detail_map: Dict[str, Any], staging: Optional[Dict[str, Any]], # per-shot staging dict (caller 가 keyed map 에서 lookup 후 전달) entity_lookup: Dict[str, Dict], project_config: Dict[str, Any], reference_svc: Any, stills: Optional[List[Any]] = None, # batch loop 안 prev_shot_ref builder 가 사용 — 단건은 None 또는 [still] 전달 ) -> tuple[str, list[tuple[str, bytes]]]: """spec §4.1 6단계 + 5a~5d 분기 — chain_bg / prev_shot_ref / state_variant 통합. v2: state_variant_sids 와 best_prev_bytes 는 helper 내부에서 detect/resolve (caller 부담 줄임). cached_style_context + cached_entity_text_map 은 caller 가 episode 단위 build 후 전달 (배치 N stills 시 N×rebuild 회귀 방지). Returns: (full_prompt, labeled_refs) """ # 1. var_t2i — load_shot_t2i_variations t2i_variations = load_shot_t2i_variations( settings.projects_dir, project_config["project_id"], episode_id, camera_json=still.camera_json, scene_index=still.scene_index, still_index=still.still_index, shot_index=still.shot_index, ) var_t2i = ( t2i_variations[0].get("t2i_prompt", still_data["still_frame_prompt"]) if t2i_variations else still_data["still_frame_prompt"] ) # 2. scene_ref_image_map (location 자동 제외) scene_ref_image_map = reference_svc.build_scene_ref_image_map(ref_image_map, entity_lookup) # 3. (v2) state_variant detect — helper 내부 (batch line 245-251 동등) state_variant_sids = reference_svc.detect_state_variant_sids( visible_entities, entity_lookup, scene_ref_image_map, staging, ) if state_variant_sids: logger.info("Scene %d Shot %d: state_variant=%s", still_data.get("scene_index", 0), still_data.get("shot_index", 0), list(state_variant_sids.keys())) # 4. resolve_refs_for_prompt (entity-only) labeled_refs = reference_svc.resolve_refs_for_prompt( var_t2i, visible_entities, scene_ref_image_map, entity_lookup, state_variant_sids, shot_description=still_data.get("shot_description") or still_data.get("still_frame_prompt"), ) # 5. (v2) best_prev_bytes resolve — helper 내부 (batch line 216-233 동등) current_location_ids = [ e["id"] for e in visible_entities if e.get("entity_type") == "location" ] best_prev_bytes = None dep_scene_id = still_data.get("dependent_scene_id") if dep_scene_id and dep_scene_id in scene_paths_by_index_by_id: dep_path = scene_paths_by_index_by_id[dep_scene_id] if dep_path.exists(): best_prev_bytes = dep_path.read_bytes() if not best_prev_bytes: for loc_id in current_location_ids: if loc_id in location_scene_history: best_prev_bytes, _ = location_scene_history[loc_id] break # 6. _is_close_framing _cam_dir = (staging or {}).get("camera_direction") or (staging or {}).get("camera_direction_text") or "" _is_close_framing = bool(_cam_dir and _CLOSE_FRAMING_RE.search(_cam_dir)) # 7. background fallback chain (5a~5d 분기 — spec §4.1) _bc_key = f"{still_data.get('scene_index', 0)}_{still_data.get('shot_index', 0)}" _bc_bg = background_chain_bg_map.get(_bc_key) if _bc_bg and _bc_bg.get("image_bytes") and not _is_close_framing: # 5a labeled_refs.insert(0, (_bc_bg["label"], _bc_bg["image_bytes"])) logger.info("Scene %d Shot %d: background_chain ref injected", still_data.get("scene_index", 0), still_data.get("shot_index", 0)) else: # 5b/5c — chain skip OR 부재 → prev_shot try if _bc_bg and _bc_bg.get("image_bytes") and _is_close_framing: logger.info("Scene %d Shot %d: chain_bg ref SKIPPED — close framing (camera_direction=%r)", still_data.get("scene_index", 0), still_data.get("shot_index", 0), _cam_dir[:80]) try: _prev_shot_ref = reference_svc.build_prev_shot_background_ref( best_prev_bytes=best_prev_bytes, still_data=still_data, visible_entities=visible_entities, current_location_ids=current_location_ids, dep_scene_id=dep_scene_id, stills=stills or [], location_scene_history=location_scene_history, dep_detail_map=dep_detail_map, staging=staging, state_variant_sids=state_variant_sids, entity_lookup=entity_lookup, ) except Exception as exc: logger.warning( "Scene %d Shot %d: prev_shot_ref build failed (%s) — entity-only fallback", still_data.get("scene_index", 0), still_data.get("shot_index", 0), exc, ) _prev_shot_ref = None if _prev_shot_ref: labeled_refs.insert(0, _prev_shot_ref) # 5d: prev_shot 도 None → entity-only (이미 labeled_refs 에 character/prop 만 — 명시적 path) # 8. image index + var_t2i rewrite labeled_refs, _sid_to_img, _sid_info = _build_image_index_helper(labeled_refs, entity_lookup) var_t2i = _rewrite_t2i_helper(var_t2i, _sid_to_img, _sid_info) # 9. (v2) _build_final_scene_prompt — caller 가 cached_* 전달 (per-episode reuse) try: _full_prompt = _build_final_scene_prompt( var_t2i, labeled_refs, cached_style_context, project_config=project_config, entity_text_map=cached_entity_text_map, ) except Exception as exc: logger.warning("Prompt build failed: %s", exc) _full_prompt = var_t2i return _full_prompt, labeled_refs ``` **v2 변경 요약** (audit IMPORTANT 1 반영): - `state_variant_sids` 인자 제거 → 내부 `detect_state_variant_sids` 호출 (batch line 245-251 동등) - `best_prev_bytes` 인자 제거 → 내부 `dep_scene_id`/`location_scene_history` resolve (batch line 216-233 동등) - `current_location_ids` / `dep_scene_id` 인자 제거 → 내부 build (visible_entities + still_data 로부터) - `cached_style_context` + `cached_entity_text_map` 인자 추가 → caller 가 episode 단위 1회 build 후 전달 (배치 N×rebuild 회귀 방지) - `scene_paths_by_index_by_id` 인자 추가 → best_prev_bytes resolve input - [ ] **Step 4: 배치 경로 (`generate_images`) 가 helper 호출하도록 변경** `coordinator.py:263-326` 블록을 `build_scene_attached_refs(...)` 호출로 교체. 기존 batch 호출 지점은 loop 안 (per-still). helper 가 single still 처리, batch 는 loop 으로 N 호출. - [ ] **Step 5: 테스트 실행 — pass 확인** Run: `cd backend && PYTHONPATH=. .venv/bin/pytest tests/unit/test_build_scene_attached_refs.py -v` Expected: 4 tests PASS (5a/5b/5c/5d 분기 모두). - [ ] **Step 6: broader regression — 기존 batch integration test pass** Run: `cd backend && PYTHONPATH=. .venv/bin/pytest tests/integration/ -v -k "scene_image" 2>&1 | tail -40` Expected: 기존 통과 test 0 회귀. - [ ] **Step 7: commit** ```bash git add backend/app/services/scene_generation_coordinator.py \ backend/tests/unit/test_build_scene_attached_refs.py git commit -m "$(cat <<'EOF' refactor(scene_image): extract build_scene_attached_refs helper from generate_images Task 1 of single-vs-batch reference contract fix (spec: 2026-05-08-single-batch-reference-contract-design.md) - coordinator.py:178-326 의 ref 빌드 블록 → 신규 helper build_scene_attached_refs - chain_bg insert (5a) / close skip + prev_shot try (5b) / chain 부재 + prev_shot try (5c) / entity-only fallback (5d) — 분기 명시 - 배치 경로 generate_images 가 helper 호출 (loop 안 per-still) - unit test 4건 (5a/5b/5c/5d 분기 검증) — regression 0 Co-Authored-By: Claude Opus 4.7 (1M context) EOF )" ``` --- ## Task 2: 단건 경로가 helper 호출 **Files:** - Modify: `backend/app/services/scene_generation_coordinator.py:677-739` — `_build_single_scene_prompt_and_refs` rewrite - Modify: `backend/app/services/scene_image_service.py:603-699` — `generate_single_scene_image` 단건 input prep - Test: `backend/tests/integration/test_single_scene_image_uses_helper.py` (신규) **AC:** AC-1 (단건 ↔ 배치 actual labeled_refs 동등). **Non-goals confirm**: prop 신규 schema X / 별도 테이블 X / manifest 재생성 X / 수동 prompt 보정 X / LVM/v22 변경 X / 컬럼 rename X. ✓ - [ ] **Step 1: failing integration test 작성** ```python # backend/tests/integration/test_single_scene_image_uses_helper.py """단건 호출이 helper 사용하여 배치와 동등한 actual labeled_refs 빌드 — Task 2.""" import pytest import json from unittest.mock import MagicMock, patch def test_single_path_actual_refs_equal_batch_path(db_session, project_fixture, episode_fixture): """단건 generate_single_scene_image 결과 labeled_refs == 배치 generate_images 결과 (label set).""" from app.services.scene_image_service import SceneImageService # Setup: same still, same chain_bg_map, same visible_entities still_id = "test_still_id" # ... fixture setup # 배치 path batch_svc = SceneImageService(...) with patch.object(batch_svc._coord, "_single_scene_generate_and_build_result") as batch_gen: batch_gen.return_value = {"id": "asset_b", ...} batch_svc.generate_images(...) batch_call_kwargs = batch_gen.call_args.kwargs batch_labels = [r[0] for r in batch_call_kwargs.get("reference_images") or []] # 단건 path single_svc = SceneImageService(...) with patch.object(single_svc._coord, "_single_scene_generate_and_build_result") as single_gen: single_gen.return_value = {"id": "asset_s", ...} single_svc.generate_single_scene_image(still_id) single_call_kwargs = single_gen.call_args.kwargs single_labels = [r[0] for r in single_call_kwargs.get("reference_images") or []] assert sorted(single_labels) == sorted(batch_labels), \ f"single != batch: single={single_labels} batch={batch_labels}" ``` - [ ] **Step 2: 테스트 실행 — fail 확인** Run: `cd backend && PYTHONPATH=. .venv/bin/pytest tests/integration/test_single_scene_image_uses_helper.py -v` Expected: FAIL — 단건이 character composite 만, 배치가 chain_bg + character 등. - [ ] **Step 3: `build_single_still_context` 신규 helper + `_build_single_scene_prompt_and_refs` rewrite** v2: `build_single_still_context` 의 정확한 implementation 코드 inline. **batch generate_images 의 사전 data build 블록과 동일 source 호출** 의무 (DRY) — 단건/배치 동등성 보장. ```python # coordinator.py 신규 helper — batch generate_images 의 사전 data build 와 동일 source def build_single_still_context( self, *, episode_id: str, still: Any, ) -> Dict[str, Any]: """단건 generate_single_scene_image 의 helper input prep. batch generate_images 의 사전 data build 블록 (entity_lookup, ref_image_map, scene_paths_by_index_by_id, location_scene_history, dep_detail_map, background_chain_bg_map, staging, cached_style_context, cached_entity_text_map) 을 동일 source 호출로 build — single ↔ batch 동등성 보장. Returns dict with all helper inputs ready. """ pid = self._project_id # 1. entity_lookup (episode 전체 — 동일 source: ref_svc.load_episode_entity_lookup) entity_lookup = self._reference_svc.load_episode_entity_lookup(episode_id) # 2. ref_image_map (location 포함, helper 내부 build_scene_ref_image_map 가 제외) ref_image_map = self._reference_svc.load_entity_reference_images( list(entity_lookup.values()) ) # 3. scene_paths_by_index_by_id (이미 generated scene PNG path 매핑) project_dir = self._get_project_dir() scene_dir = project_dir / "images" / episode_id / "scene" # batch 동등 — scene_index → ImageAsset.file_path Path 매핑 scene_paths_by_index_by_id = self._reference_svc.build_scene_paths_by_index_by_id( episode_id, scene_dir, ) # 4. location_scene_history (location_id → (bytes, meta)) location_scene_history = self._reference_svc.build_location_scene_history( episode_id, scene_dir, ) # 5. background_chain_bg_map (per-shot pre-rendered chain_bg PNG) background_chain_bg_map = load_background_chain_bg_map( settings.projects_dir, pid, episode_id, ) # 6. dep_detail_map (scene_detail manifest 의 shot detail mapping) dep_detail_map = load_scene_detail_dep_map( # batch 와 동일 source — 신규 helper or 기존 함수 재사용 settings.projects_dir, pid, episode_id, ) # 7. staging (per-shot staging dict — batch 는 staging_map[key] lookup) staging_map = load_shot_staging_map( settings.projects_dir, pid, episode_id, ) _staging_key = f"{still.scene_index}_{still.shot_index}" staging = staging_map.get(_staging_key) # 8. cached_style_context + cached_entity_text_map (episode 단위 1회 build) cached_style_context = self._get_style_context(episode_id) cached_entity_text_map = self._reference_svc.build_entity_text_map( list(entity_lookup.values()) ) # 9. stills (단건 시 [still] 또는 episode 전체 stills — prev_shot_ref builder 사용) stills = self._persistence_svc.load_episode_stills(episode_id) # batch 와 동일 source return { "entity_lookup": entity_lookup, "ref_image_map": ref_image_map, "scene_paths_by_index_by_id": scene_paths_by_index_by_id, "location_scene_history": location_scene_history, "background_chain_bg_map": background_chain_bg_map, "dep_detail_map": dep_detail_map, "staging": staging, "cached_style_context": cached_style_context, "cached_entity_text_map": cached_entity_text_map, "stills": stills, } ``` ```python # coordinator.py:677-739 변경 def _build_single_scene_prompt_and_refs( self, *, still: Any, episode_id: str, still_data: Dict[str, Any], visible_entities: List[Dict[str, Any]], ref_image_map: Dict[str, bytes], ) -> "tuple[str, list]": """단건 generate_single_scene_image 의 ref 빌드 — helper 사용 (Task 2). spec §4.1: 배치와 동일 contract. chain_bg / prev_shot_ref / state_variant 누락 fix. v2: build_single_still_context 가 batch 의 사전 data build 와 동일 source. """ ctx = self.build_single_still_context(episode_id=episode_id, still=still) # caller 가 받은 ref_image_map 우선 사용 (test/mock 호환), 없으면 ctx 사용 _ref_image_map = ref_image_map or ctx["ref_image_map"] return build_scene_attached_refs( still=still, episode_id=episode_id, still_data=still_data, visible_entities=visible_entities, ref_image_map=_ref_image_map, cached_style_context=ctx["cached_style_context"], cached_entity_text_map=ctx["cached_entity_text_map"], scene_paths_by_index_by_id=ctx["scene_paths_by_index_by_id"], location_scene_history=ctx["location_scene_history"], background_chain_bg_map=ctx["background_chain_bg_map"], dep_detail_map=ctx["dep_detail_map"], staging=ctx["staging"], entity_lookup=ctx["entity_lookup"], project_config=load_project_llm_config(self._db, self._project_id), reference_svc=self._reference_svc, stills=ctx["stills"], ) ``` **필수 검증** (Task 2 implementation 시): - `build_single_still_context` 안의 9 build 호출이 **batch generate_images 안 사전 build 코드와 같은 source 함수 호출**. batch 가 직접 inline 코드 사용하는 부분은 **별 helper module 로 lift 후 양쪽 호출 경로가 같은 helper 사용**. plan task scope 안. - `load_scene_detail_dep_map`, `load_shot_staging_map`, `build_scene_paths_by_index_by_id`, `build_location_scene_history`, `load_episode_stills` 등 helper 가 미존재 시 **batch generate_images 의 사전 build 코드를 lift 하여 신규 module 생성**. batch 도 신규 helper 호출하도록 변경 (DRY 강제 — 양쪽 경로 동일 source 보장). - [ ] **Step 4: 테스트 실행 — pass 확인** Run: `cd backend && PYTHONPATH=. .venv/bin/pytest tests/integration/test_single_scene_image_uses_helper.py -v` Expected: PASS. - [ ] **Step 5: contract 동등성 unit test (single ↔ batch context source)** ```python # backend/tests/unit/test_single_still_context_equality.py — 추가 """build_single_still_context 9 fields 가 batch generate_images 사전 build 와 동일 source 호출 — Task 2 v2. source 동일성 보장: 단건 ↔ 배치 호출 시 동일 episode 에서 동일 9 fields 반환. """ import pytest def test_single_still_context_uses_same_source_helpers(db_session, episode_fixture, monkeypatch): """build_single_still_context 의 9 source 호출이 batch 사전 build 와 동일 함수. monkeypatch 로 source helpers 를 spy 해서 batch + single 양쪽이 동일 helper 호출 검증. """ from app.services.scene_image_service import SceneImageService from app.services.scene_reference_service import SceneReferenceService calls = {"load_episode_entity_lookup": 0, "load_entity_reference_images": 0, "build_scene_paths_by_index_by_id": 0, "build_location_scene_history": 0, "load_episode_stills": 0} orig = SceneReferenceService.load_episode_entity_lookup def _spy(self, eid): calls["load_episode_entity_lookup"] += 1 return orig(self, eid) monkeypatch.setattr(SceneReferenceService, "load_episode_entity_lookup", _spy) # ... 4 spy 추가 동등 패턴 svc = SceneImageService(...) ctx = svc._coord.build_single_still_context(episode_id=episode_fixture, still=...) # 단건 호출 후 모든 source helper 가 1회씩 호출됨 확인 for fn, count in calls.items(): assert count >= 1, f"{fn} not called by build_single_still_context" # 9 fields 모두 None 아님 assert ctx["entity_lookup"] assert ctx["scene_paths_by_index_by_id"] is not None assert "background_chain_bg_map" in ctx # ... ``` - [ ] **Step 6: NG still S8/S19 actual refs 회귀 검증 (수동)** Run: 단건 endpoint 직접 호출 또는 service 직접 호출로 S8_Shot4 / S19_Shot1 재생성. llm_call_log.reference_image_ids 가 chain_bg + character 포함하는지 SQL 쿼리로 확인. ```python # backend/scripts/canary_single_vs_batch_refs.py 의 일부 동등 (Task 6 에서 정식화) from sqlalchemy import text PID="80f62523-..."; EID="9a5e0862-..." with SessionLocal() as db: rows = db.execute(text(""" SELECT reference_image_ids FROM llm_call_log WHERE project_id=:pid AND episode_id=:eid AND operation_type='single_scene_image_gen' AND metadata_json::text LIKE '%b6662ce1%' ORDER BY created_at DESC LIMIT 1 """), {"pid": PID, "eid": EID}).all() # assert chain_bg label OR previous shot label in refs ``` - [ ] **Step 7: commit** ```bash git add backend/app/services/scene_generation_coordinator.py \ backend/app/services/scene_image_service.py \ backend/app/services/scene_reference_service.py \ backend/tests/integration/test_single_scene_image_uses_helper.py \ backend/tests/unit/test_single_still_context_equality.py git commit -m "$(cat <<'EOF' fix(scene_image): single path uses build_scene_attached_refs helper Task 2 of single-vs-batch reference contract fix (plan v2) (spec: 2026-05-08-single-batch-reference-contract-design.md) - _build_single_scene_prompt_and_refs 가 build_scene_attached_refs(...) 호출 - build_single_still_context 신규 — batch generate_images 사전 build 와 동일 source (entity_lookup / ref_image_map / scene_paths_by_index_by_id / location_scene_history / background_chain_bg_map / dep_detail_map / staging / cached_style_context / cached_entity_text_map / stills 모두 동일 helper 호출 — DRY) - v2 audit IMPORTANT 2 반영: source 동일성 unit test (test_single_still_context_equality) - integration test: single 호출 결과 actual labeled_refs == batch 호출 결과 동등 (label set) Co-Authored-By: Claude Opus 4.7 (1M context) EOF )" ``` --- ## Task 3: `validate_attached_refs` fail-fast **Files:** - Create: `backend/app/core/ref_contract_validator.py` - Modify: `backend/app/modules/pipeline/scene_image_pipeline.py` — `generate_and_validate_scene` 직전 validator 호출 - Modify: `backend/app/core/errors.py` — `RefContractError` export - Test: `backend/tests/unit/test_ref_contract_validator.py` (신규) - Test: `backend/tests/unit/test_from_the_reference_classifier.py` (신규) **AC:** AC-3 (background required + close X + bg ref 미첨부 → fail-fast). AC-4 (classifier-based). **Non-goals confirm**: prop 신규 schema X / 별도 테이블 X / manifest 재생성 X / 수동 prompt 보정 X / LVM/v22 변경 X / 컬럼 rename X. ✓ - [ ] **Step 1: failing test 작성 (validator core)** ```python # backend/tests/unit/test_ref_contract_validator.py """validate_attached_refs — Task 3.""" import pytest def test_character_outlook_required_missing_raises(): """required_refs.character_outlook 명시 + visible character 존재 + labeled_refs 없음 → fail.""" from app.core.ref_contract_validator import validate_attached_refs, RefContractError rpc = { "asset_requirements": { "required_refs": {"character_outlook": ["C01O02"]}, }, } labeled_refs = [] # missing prompt = "A young woman walks down the street." with pytest.raises(RefContractError) as ei: validate_attached_refs(rpc, labeled_refs, prompt, is_close_framing=False) assert "character" in str(ei.value).lower() def test_background_required_close_framing_exempt(): """required_refs.background 있어도 close framing 면제.""" from app.core.ref_contract_validator import validate_attached_refs rpc = {"asset_requirements": {"required_refs": {"background": ["L01"]}}} labeled_refs = [("character C01O02 in outfit", b"x")] prompt = "Eye close-up." # No raise expected validate_attached_refs(rpc, labeled_refs, prompt, is_close_framing=True) def test_background_required_not_close_missing_raises(): """required_refs.background + NOT close + bg ref 없음 → fail.""" from app.core.ref_contract_validator import validate_attached_refs, RefContractError rpc = {"asset_requirements": {"required_refs": {"background": ["L01"]}}} labeled_refs = [("character C01O02 in outfit", b"x")] # no BACKGROUND prompt = "Two figures share the supermarket aisle." with pytest.raises(RefContractError) as ei: validate_attached_refs(rpc, labeled_refs, prompt, is_close_framing=False) assert "background" in str(ei.value).lower() ``` - [ ] **Step 2: failing test 작성 (classifier)** ```python # backend/tests/unit/test_from_the_reference_classifier.py """from-the-reference phrase position-based classifier — Task 3.""" import pytest def test_classifier_character_phrase(): """phrase 주변 character 토큰 → character classifier 매칭.""" from app.core.ref_contract_validator import classify_from_the_reference prompt = "the character from Reference image 1 in a dark jacket" matches = classify_from_the_reference(prompt) assert any(m["type"] == "character" for m in matches) def test_classifier_background_phrase(): """phrase 주변 background 토큰 → background classifier.""" from app.core.ref_contract_validator import classify_from_the_reference prompt = "the sliding door from the reference centered behind her" matches = classify_from_the_reference(prompt) assert any(m["type"] == "background" for m in matches) def test_classifier_object_phrase(): """phrase 주변 object 토큰 → object classifier.""" from app.core.ref_contract_validator import classify_from_the_reference prompt = "the photograph from Reference image 4 on the desk" matches = classify_from_the_reference(prompt) assert any(m["type"] == "object" for m in matches) def test_classifier_ambiguous_warning(): """classifier 매칭 0 → warning (fail-fast 아님).""" from app.core.ref_contract_validator import classify_from_the_reference prompt = "from the reference image as inspiration" matches = classify_from_the_reference(prompt) # ambiguous — no specific type, returns warning marker assert all(m["type"] == "ambiguous" for m in matches) or matches == [] def test_validate_classifier_fail_fast_character_missing(): """character classifier match + character ref missing → fail.""" from app.core.ref_contract_validator import validate_attached_refs, RefContractError rpc = {"asset_requirements": {"required_refs": {}}} labeled_refs = [] # no character prompt = "the character from Reference image 1 walks" with pytest.raises(RefContractError): validate_attached_refs(rpc, labeled_refs, prompt, is_close_framing=False) ``` - [ ] **Step 3: 테스트 실행 — fail 확인** Run: `cd backend && PYTHONPATH=. .venv/bin/pytest tests/unit/test_ref_contract_validator.py tests/unit/test_from_the_reference_classifier.py -v` Expected: FAIL — module 미존재. - [ ] **Step 4: validator + classifier 구현** ```python # backend/app/core/ref_contract_validator.py """Reference attachment contract validator — Task 3. spec §4.2: required_refs vs actual labeled_refs 비교 + classifier-based "from the reference" guard. fail-fast (fallback inject 금지). """ from __future__ import annotations import re from typing import Any, Dict, List, Optional from app.core.errors import AppError class RefContractError(AppError): """Required ref 미첨부 또는 from-the-reference guard 위반 — generate 차단.""" def __init__(self, detail: str): super().__init__( code="ref_contract.violation", message=detail, status_code=422, ) # classifier 토큰 list (conservative — false positive 보다 false negative 선호) _CHARACTER_TOKENS = [ "character", "figure", "face", "eyes", "her hand", "his hand", "shoulders", "her arm", "his arm", "his face", "her face", ] _BACKGROUND_TOKENS = [ "door", "glass", "wall", "floor", "desk", "storefront", "aisle", "tv", "ceiling", "window", "stair", "mart", "store", "room", "bench", "road", "street", ] _OBJECT_TOKENS_GENERIC = [ "photograph", "memo", "paper", "phone", "smartphone", "cup", "memo pad", "note", "knife", ] _FROM_THE_REFERENCE_RE = re.compile(r"from\s+(?:the\s+)?reference\b", re.IGNORECASE) def classify_from_the_reference(prompt: str) -> List[Dict[str, Any]]: """phrase 주변 ±100자 윈도우 토큰 매칭으로 가리키는 ref type 분류. Returns list of dicts: [{"position": int, "type": "character"|"background"|"object"|"ambiguous"}, ...] """ matches = [] for m in _FROM_THE_REFERENCE_RE.finditer(prompt): start = max(0, m.start() - 100) end = min(len(prompt), m.end() + 100) window = prompt[start:end].lower() ref_type = "ambiguous" if any(t in window for t in _CHARACTER_TOKENS): ref_type = "character" elif any(t in window for t in _BACKGROUND_TOKENS): ref_type = "background" elif any(t in window for t in _OBJECT_TOKENS_GENERIC): ref_type = "object" matches.append({"position": m.start(), "type": ref_type}) return matches def _has_character_ref(labeled_refs: List[tuple]) -> bool: return any( "character" in (label or "").lower() for label, _ in labeled_refs ) def _has_background_ref(labeled_refs: List[tuple]) -> bool: return any( ("background" in (label or "").lower() or "previous shot" in (label or "").lower()) for label, _ in labeled_refs ) def _has_object_ref(labeled_refs: List[tuple]) -> bool: return any( "object" in (label or "").lower() for label, _ in labeled_refs ) def validate_attached_refs( rpc: Dict[str, Any], labeled_refs: List[tuple], prompt: str, is_close_framing: bool, ) -> None: """spec §4.2: required_refs vs actual labeled_refs + classifier-based guard. 위반 시 RefContractError raise (fail-fast). """ asset_req = (rpc or {}).get("asset_requirements") or {} required_refs = asset_req.get("required_refs") or {} # 1. character_outlook required → character ref 의무 char_required = required_refs.get("character_outlook") or [] if char_required and not _has_character_ref(labeled_refs): raise RefContractError( f"required character_outlook ref missing — required={char_required} " f"actual_labels={[lbl for lbl, _ in labeled_refs]}" ) # 2. background required (NOT close) → background ref 의무 bg_required = required_refs.get("background") or [] if bg_required and not is_close_framing and not _has_background_ref(labeled_refs): raise RefContractError( f"required background ref missing (not close framing) — required={bg_required} " f"actual_labels={[lbl for lbl, _ in labeled_refs]}" ) # 3. classifier-based "from the reference" guard classifier_matches = classify_from_the_reference(prompt) for match in classifier_matches: if match["type"] == "character" and not _has_character_ref(labeled_refs): raise RefContractError( f"prompt 'from the reference' classifier=character at pos={match['position']} " f"but character ref missing — actual_labels={[lbl for lbl, _ in labeled_refs]}" ) if match["type"] == "background" and not is_close_framing and not _has_background_ref(labeled_refs): raise RefContractError( f"prompt 'from the reference' classifier=background at pos={match['position']} " f"(not close framing) but background ref missing" ) if match["type"] == "object" and not _has_object_ref(labeled_refs): raise RefContractError( f"prompt 'from the reference' classifier=object at pos={match['position']} " f"but object ref missing" ) # type=='ambiguous' → MINOR warning (fail-fast 아님) ``` - [ ] **Step 5: `scene_image_pipeline.py` 에서 validator 호출** `generate_and_validate_scene` 직전 `validate_attached_refs(rpc, labeled_refs, prompt, is_close_framing)` 호출. caller (`coordinator._generate_variation_in_loop` / `scene_image_service._single_scene_generate_and_build_result`) 가 rpc + is_close_framing 인자 전달. retry 1회 (var_t2i rebuild) 후 재실패 시 propagate. - [ ] **Step 6: 테스트 실행 — pass 확인** Run: `cd backend && PYTHONPATH=. .venv/bin/pytest tests/unit/test_ref_contract_validator.py tests/unit/test_from_the_reference_classifier.py -v` Expected: 8 tests PASS. - [ ] **Step 7: broader regression** Run: `cd backend && PYTHONPATH=. .venv/bin/pytest tests/integration/ -v -k "scene_image or ref_contract" 2>&1 | tail -40` Expected: 0 회귀. - [ ] **Step 8: commit** ```bash git add backend/app/core/ref_contract_validator.py \ backend/app/core/errors.py \ backend/app/modules/pipeline/scene_image_pipeline.py \ backend/tests/unit/test_ref_contract_validator.py \ backend/tests/unit/test_from_the_reference_classifier.py git commit -m "$(cat <<'EOF' feat(scene_image): validate_attached_refs fail-fast + classifier guard Task 3 of single-vs-batch reference contract fix (spec: 2026-05-08-single-batch-reference-contract-design.md §4.2) - backend/app/core/ref_contract_validator.py 신규 - RefContractError (HTTP 422) + classify_from_the_reference + validate_attached_refs - character/background/object classifier (±100자 phrase window) - close framing 시 background skip 정책 일관 면제 - ambiguous classifier = MINOR warning (fail-fast 아님 — false negative 보수적) - scene_image_pipeline 이 generate_and_validate_scene 직전 호출 - 8 unit tests PASS Co-Authored-By: Claude Opus 4.7 (1M context) EOF )" ``` --- ## Task 4: face/eyes close-up exemption 축소 **Files:** - Modify: `backend/app/core/visible_entities_validator.py:131-199` — `_is_face_close_up` 신규 + `_forward_enforcement_exempt` 변경 - Test: `backend/tests/unit/test_face_close_up_detection.py` (신규) - Test: `backend/tests/unit/test_forward_enforcement_face_exemption.py` (신규) **AC:** AC-2 (S13_Shot6 같은 face close-up + visible C01/C02 → fail-fast). **Non-goals confirm**: 모든 6개 ✓ - [ ] **Step 1: failing test 작성 (`_is_face_close_up`)** ```python # backend/tests/unit/test_face_close_up_detection.py """_is_face_close_up — Task 4.""" import pytest def test_face_focus_phrase_match(): from app.core.visible_entities_validator import _is_face_close_up assert _is_face_close_up("Focus on the tear-filled eyes of a young woman") assert _is_face_close_up("close on her face") assert _is_face_close_up("tight on his expression") assert _is_face_close_up("eyes close-up") assert _is_face_close_up("face close-up") def test_body_part_phrase_no_false_positive(): from app.core.visible_entities_validator import _is_face_close_up assert not _is_face_close_up("focus on the hand") assert not _is_face_close_up("close on the wrist") assert not _is_face_close_up("tight on the smartphone screen") assert not _is_face_close_up("focus on the bicycle handlebar") def test_eyes_in_other_context_no_false_positive(): from app.core.visible_entities_validator import _is_face_close_up # "her eyes fixed on the door" — eyes 단독은 face 아님 # But pattern includes "focus on .. eyes" — 이건 false positive 가능. 정확 패턴은 phrase 인접. # 본 case 는 face/eyes 가 framing target 일 때만 매칭 필요. # "her eyes fixed on" 은 face close-up 아님 — descriptive assert not _is_face_close_up("Her eyes fixed on the door") ``` - [ ] **Step 2: failing test 작성 (forward enforcement)** ```python # backend/tests/unit/test_forward_enforcement_face_exemption.py """_forward_enforcement_exempt face/eyes 면제 거부 — Task 4.""" import pytest def test_face_close_up_with_body_focus_trigger_no_exempt(): """body_part_focus_rule trigger ('focus on') + face/eyes match → exemption 거부.""" from app.core.visible_entities_validator import _forward_enforcement_exempt rpc = { "id_policy": { "body_part_focus_rule": { "trigger_phrases": ["focus on", "close on"], "id_use": "forbidden", }, }, } prompt = "Focus on the tear-filled eyes of a young woman" exempt, reason = _forward_enforcement_exempt(rpc, prompt) assert exempt is False, f"face close-up should not be exempt, got {reason!r}" def test_body_close_up_with_trigger_still_exempt(): """손/발 같은 body close-up 은 trigger 매칭 시 면제 유지.""" from app.core.visible_entities_validator import _forward_enforcement_exempt rpc = { "id_policy": { "body_part_focus_rule": { "trigger_phrases": ["focus on"], "id_use": "forbidden", }, }, } prompt = "Focus on the hand holding a smartphone" exempt, reason = _forward_enforcement_exempt(rpc, prompt) assert exempt is True def test_partial_focus_mode_still_exempt(): """render_strategy.mode=='partial_focus' 면제 유지 (face 매칭 무관).""" from app.core.visible_entities_validator import _forward_enforcement_exempt rpc = {"render_strategy": {"mode": "partial_focus"}} prompt = "Focus on the eyes" exempt, reason = _forward_enforcement_exempt(rpc, prompt) assert exempt is True assert "partial_focus" in reason ``` - [ ] **Step 3: 테스트 실행 — fail 확인** Run: `cd backend && PYTHONPATH=. .venv/bin/pytest tests/unit/test_face_close_up_detection.py tests/unit/test_forward_enforcement_face_exemption.py -v` Expected: FAIL — `_is_face_close_up` 미존재 + 기존 `_forward_enforcement_exempt` 가 face close-up exempt. - [ ] **Step 4: `_is_face_close_up` + `_forward_enforcement_exempt` patch** ```python # backend/app/core/visible_entities_validator.py:131 위에 추가 _FACE_CLOSE_UP_PATTERNS = [ re.compile(r"focus on .{0,40}\b(face|eye|eyes|expression|gaze|stare)\b", re.IGNORECASE), re.compile(r"close on .{0,40}\b(face|eye|eyes|expression)\b", re.IGNORECASE), re.compile(r"tight on .{0,40}\b(face|eye|eyes|expression)\b", re.IGNORECASE), re.compile(r"\b(eye|eyes|face) close[\-\s]?up\b", re.IGNORECASE), ] def _is_face_close_up(prompt: str) -> bool: """face/eyes/expression close-up 패턴 매칭 — Task 4 spec §4.3.""" return any(pat.search(prompt) for pat in _FACE_CLOSE_UP_PATTERNS) ``` `_forward_enforcement_exempt()` 의 body_focus 분기 변경: ```python body_focus = id_policy.get("body_part_focus_rule") if isinstance(body_focus, dict): triggers = body_focus.get("trigger_phrases") if isinstance(triggers, list): for t in triggers: if isinstance(t, str) and t and t.lower() in prompt_lower: # Task 4: face/eyes/expression close-up 시 면제 거부 if _is_face_close_up(prompt): # face close-up 은 character ID 의무 — exemption skip # fall-through to next trigger or exit loop continue return True, ( f"body_part_focus_rule.trigger_phrases — " f"{t!r} found in prompt" ) ``` - [ ] **Step 5: 테스트 실행 — pass 확인** Run: `cd backend && PYTHONPATH=. .venv/bin/pytest tests/unit/test_face_close_up_detection.py tests/unit/test_forward_enforcement_face_exemption.py -v` Expected: 7 tests PASS (3 face_close_up + 3 forward_enforcement + 1 negative). - [ ] **Step 6: broader regression — visible_entities_validator 기존 test** Run: `cd backend && PYTHONPATH=. .venv/bin/pytest tests/unit/ -v -k "visible_entities or forward_enforcement" 2>&1 | tail -40` Expected: 0 회귀 (기존 통과 stills 의 body close-up 면제 유지). - [ ] **Step 7: commit** ```bash git add backend/app/core/visible_entities_validator.py \ backend/tests/unit/test_face_close_up_detection.py \ backend/tests/unit/test_forward_enforcement_face_exemption.py git commit -m "$(cat <<'EOF' fix(validator): face/eyes close-up exemption 축소 — fail-fast only Task 4 of single-vs-batch reference contract fix (spec: 2026-05-08-single-batch-reference-contract-design.md §4.3) - _is_face_close_up(prompt) 신규 helper (4 regex patterns) - _forward_enforcement_exempt 의 body_part_focus_rule 분기에서 face/eyes/expression close-up 매칭 시 trigger 면제 거부 (continue → fall-through) - fallback inject 금지 — fail-fast 만 (Codex/사용자 binding) - S13_Shot6 같은 face close-up + visible C01/C02 → forward enforcement fail - body close-up (손/발/입술) 는 면제 유지 (false positive 방지) - 7 unit tests PASS, broader regression 0 Co-Authored-By: Claude Opus 4.7 (1M context) EOF )" ``` --- ## Task 5: `ImageResponse.actual_attached_refs` **Files:** - Modify: `backend/app/schemas/image.py:36` — `ImageResponse` field 추가 + reference_image_ids description - Modify: `backend/app/services/image_service_helpers.py:182` — `image_to_dict` 가 actual_attached_refs join - Modify: `backend/app/models/project.py:187,288` — column docstring 양쪽 - Test: `backend/tests/unit/test_image_response_actual_attached_refs.py` (신규) **AC:** AC-5 (lineage vs actual 분리 노출). AC-6 (docstring 명시). **Non-goals confirm**: 모든 6개 ✓ (특히 컬럼명 alembic rename **금지** — schema response + docstring 만) - [ ] **Step 1: failing test 작성** ```python # backend/tests/unit/test_image_response_actual_attached_refs.py """ImageResponse.actual_attached_refs + image_to_dict join — Task 5.""" import pytest import json def test_image_response_has_actual_attached_refs_field(): from app.schemas.image import ImageResponse schema = ImageResponse.model_json_schema() props = schema["properties"] assert "actual_attached_refs" in props assert "lineage" in (props.get("reference_image_ids", {}).get("description") or "").lower() def test_image_to_dict_joins_actual_attached_refs(db_session, project_fixture, image_asset_fixture, llm_call_log_fixture): """image_to_dict 가 llm_call_log 의 actual ref labels 를 join 하여 반환.""" from app.services.image_service_helpers import image_to_dict # fixture: image_asset (still_id=X, lineage=[c01,c02]) + llm_call_log (still_id=X, ref labels=["character C01O02"]) img_dict = image_to_dict(image_asset_fixture, db=db_session) assert img_dict.get("actual_attached_refs") == ["character C01O02 in outfit"] assert json.loads(img_dict.get("reference_image_ids", "[]")) == ["c01", "c02"] def test_image_to_dict_actual_refs_none_when_no_log_match(db_session, image_asset_fixture): """llm_call_log 매칭 없으면 None (lineage fallback caller 책임).""" from app.services.image_service_helpers import image_to_dict img_dict = image_to_dict(image_asset_fixture, db=db_session) assert img_dict.get("actual_attached_refs") is None ``` - [ ] **Step 2: 테스트 실행 — fail 확인** Run: `cd backend && PYTHONPATH=. .venv/bin/pytest tests/unit/test_image_response_actual_attached_refs.py -v` Expected: FAIL — `actual_attached_refs` field 미존재. - [ ] **Step 3: schema + image_to_dict + docstring 변경** ```python # backend/app/schemas/image.py:36 변경 class ImageResponse(BaseModel): id: str asset_type: str # ... reference_image_ids: str = Field( default="[]", description=( "(lineage) visible_entities character/prop UUID list " "(JSON-encoded). NOT actual LLM-attached refs. " "See actual_attached_refs for actual refs." ), ) actual_attached_refs: Optional[list[str]] = Field( default=None, description=( "Actual ref labels attached to the LLM call " "(joined from llm_call_log.reference_image_ids). " "None if log entry not found or retention exceeded." ), ) # ... rest of fields ``` ```python # backend/app/services/image_service_helpers.py:182 변경 — v2 N+1 batch def image_to_dict( img: ImageAsset, db: Optional[OrmSession] = None, *, actual_refs_cache: Optional[Dict[str, List[str]]] = None, ) -> Dict[str, Any]: """v2 (audit IMPORTANT 4): list endpoint 가 actual_refs_cache 사전 batch lookup. actual_refs_cache: {still_id: [ref_labels]} — caller 가 _lookup_actual_refs_batch 로 1 query 사전 build 후 전달. None 이면 single asset 1 query lookup (detail endpoint). """ out = { # ... existing fields "reference_image_ids": img.reference_image_ids or "[]", } # Task 5 v2: N+1 회피 — caller cache 우선 if actual_refs_cache is not None: out["actual_attached_refs"] = actual_refs_cache.get(img.still_id or "") elif db is not None and img.still_id: # detail endpoint (1 asset) — single lookup OK actual = _lookup_actual_refs(db, img.project_id, img.episode_id, img.still_id) out["actual_attached_refs"] = actual else: out["actual_attached_refs"] = None return out def _lookup_actual_refs( db: OrmSession, project_id: str, episode_id: str, still_id: str, ) -> Optional[List[str]]: """detail endpoint — 1 still 의 가장 최근 llm_call_log entry ref labels.""" from app.models.catalog import LlmCallLog row = ( db.query(LlmCallLog.reference_image_ids) .filter( LlmCallLog.project_id == project_id, LlmCallLog.episode_id == episode_id, LlmCallLog.operation_type.in_(["single_scene_image_gen", "scene_image_gen"]), LlmCallLog.metadata_json.contains(f'"still_id": "{still_id}"'), ) .order_by(LlmCallLog.created_at.desc()) .first() ) if not row: return None try: return json.loads(row[0] or "[]") except Exception: return None def _lookup_actual_refs_batch( db: OrmSession, project_id: str, episode_id: str, still_ids: List[str], ) -> Dict[str, List[str]]: """v2 (audit IMPORTANT 4): list endpoint — episode 전체 still_id 의 latest log entry 1 query 로 lookup. N+1 회피. Returns: {still_id: ref_labels}. log entry 없거나 retention 지난 still 은 dict 미포함. """ if not still_ids: return {} from app.models.catalog import LlmCallLog from sqlalchemy import func, and_ # PostgreSQL window function — 각 still_id 별 latest log entry # operation_type 필터 + metadata_json 의 still_id 매칭 + 가장 최근 created_at row 1개씩 sql = text(""" SELECT DISTINCT ON ((metadata_json::jsonb)->>'still_id') (metadata_json::jsonb)->>'still_id' AS still_id, reference_image_ids FROM llm_call_log WHERE project_id = :pid AND episode_id = :eid AND operation_type IN ('single_scene_image_gen', 'scene_image_gen') AND (metadata_json::jsonb)->>'still_id' = ANY(:sids) ORDER BY (metadata_json::jsonb)->>'still_id', created_at DESC """) rows = db.execute(sql, {"pid": project_id, "eid": episode_id, "sids": still_ids}).all() result = {} for sid, ref_ids_json in rows: if sid is None: continue try: result[sid] = json.loads(ref_ids_json or "[]") except Exception: result[sid] = [] return result ``` **API list endpoint 사용 패턴** (e.g. `/episodes/{id}/images`): ```python # 단일 list query → still_ids extract → batch actual refs lookup → image_to_dict 매 호출 images = db.query(ImageAsset).filter(...).all() still_ids = [img.still_id for img in images if img.still_id] actual_cache = _lookup_actual_refs_batch(db, project_id, episode_id, still_ids) return [image_to_dict(img, actual_refs_cache=actual_cache) for img in images] ``` **v2 변경 요약** (audit IMPORTANT 4 반영): - `_lookup_actual_refs_batch` 신규 — episode 전체 still_id 1 query 처리. N+1 회피. - `image_to_dict` signature `actual_refs_cache: Optional[Dict]` 추가. list endpoint 가 사전 build 후 전달. - detail endpoint (1 asset) 는 기존 `_lookup_actual_refs` (single query) — N+1 무관. - **PostgreSQL functional/jsonb index 는 optional follow-up** — 본 task 의무 X. measured-needed (production explain/latency 측정 후) 시 별 alembic migration spec 으로 분리. spec §3 "alembic rename 금지" 는 column rename 한정 — index 추가는 follow-up scope 가능. ```python # backend/app/models/project.py:187 docstring class ImageAsset(Base): # ... reference_image_ids = Column( Text, default="[]", doc=( "(lineage) JSON array of visible_entities character/prop UUID list. " "NOT actual LLM-attached refs. For actual refs see llm_call_log.reference_image_ids." ), ) # backend/app/models/project.py:288 (LlmCallLog) class LlmCallLog(Base): # ... reference_image_ids = Column( Text, default="[]", doc=( "Actual ref labels attached to LLM call " "(e.g. 'character C01O02 in outfit', 'BACKGROUND chain reference')." ), ) ``` - [ ] **Step 4: 테스트 실행 — pass 확인** Run: `cd backend && PYTHONPATH=. .venv/bin/pytest tests/unit/test_image_response_actual_attached_refs.py -v` Expected: 3 tests PASS. - [ ] **Step 5: API integration test — endpoint response 검증** Run: `cd backend && PYTHONPATH=. .venv/bin/pytest tests/integration/ -v -k "image_review or image_response" 2>&1 | tail -30` Expected: 0 회귀. - [ ] **Step 6: commit** ```bash git add backend/app/schemas/image.py \ backend/app/services/image_service_helpers.py \ backend/app/models/project.py \ backend/tests/unit/test_image_response_actual_attached_refs.py git commit -m "$(cat <<'EOF' feat(image_api): ImageResponse.actual_attached_refs lineage/actual 분리 Task 5 of single-vs-batch reference contract fix (spec: 2026-05-08-single-batch-reference-contract-design.md §4.4) - ImageResponse.reference_image_ids description '(lineage)' 명시 - ImageResponse.actual_attached_refs 신규 field (Optional[list[str]]) - image_to_dict 가 llm_call_log join 하여 actual ref labels 반환 - image_asset/llm_call_log column docstring 양쪽 lineage/actual 명시 - 새 DB 컬럼/테이블 X (LOC 최소, alembic migration 0) Co-Authored-By: Claude Opus 4.7 (1M context) EOF )" ``` --- ## Task 6: Regression / canary fixture **Files:** - Create: `backend/tests/fixtures/c01_zero_gate_regression.json` — S8/S13/S15/S19/S21 still_id + 의도 actual labels - Create: `backend/tests/integration/test_c01_zero_gate_regression.py` — fixture 기반 contract test - Create: `backend/scripts/canary_single_vs_batch_refs.py` — daily canary **AC:** AC-1 검증 + Task 1-5 회귀 가드. **Non-goals confirm**: 모든 6개 ✓ (특히 샷별 prompt 수동 보정 X — fixture 는 검증용, 보정 X) - [ ] **Step 1: fixture JSON 작성** ```json // backend/tests/fixtures/c01_zero_gate_regression.json { "project_id": "80f62523-775f-4893-9388-d67b17a4339c", "episode_id": "9a5e0862-f0fd-4f21-9373-ffc60482ebe4", "stills": [ { "still_id": "b6662ce1-a38c-4469-8fb6-681d208667d5", "label": "S8_Shot4", "visible_short_ids": ["C01", "P03", "L08", "L09", "L10"], "expected_actual_ref_labels": [ "character C01O02 in outfit", "BACKGROUND chain reference" ], "is_close_framing": false, "user_reported_defect": "출입문이 아닌 사람 가리킴" }, { "still_id": "e75673a1-8484-4062-96e0-5437270e9392", "label": "S13_Shot6", "visible_short_ids": ["C01", "C02", "L11"], "expected_actual_ref_labels": [ "character C01O02 in outfit", "character C02O03 in outfit" ], "is_close_framing": true, "user_reported_defect": "캐릭터 누락 (No reference images)", "expected_failure_mode_pre_fix": "RefContractError or labeled_refs=[]" }, { "still_id": "f6675640-c03a-43dc-9030-4e3445005f68", "label": "S15_Shot5", "visible_short_ids": ["C01", "C02", "L12"], "expected_actual_ref_labels": [ "character C02O03 in outfit", "character C01O02 in outfit", "previous shot SAME ROOM" ], "is_close_framing": false, "user_reported_defect": "달리는 방향 90도 빗나감" }, { "still_id": "4213ed6b-b4e2-4587-b6a0-251a7f3e9544", "label": "S19_Shot1", "visible_short_ids": ["C01", "P03", "L09", "L10"], "expected_actual_ref_labels": [ "character C01O02 in outfit", "BACKGROUND chain reference" ], "is_close_framing": false, "user_reported_defect": "바퀴/핸들 정면 이상" }, { "still_id": "a1a94038-e071-4735-b5ae-dfe78c2fead3", "label": "S21_Shot6", "visible_short_ids": ["C01", "P09", "P07", "L13"], "expected_actual_ref_labels": [ "character C01O02 in outfit", "BACKGROUND chain reference" ], "is_close_framing": true, "user_reported_defect": "책상 위 사진 없음 (P09 prop ref 누락)", "note": "P09/P07 narrative-critical prop carry 는 P0-5 후속 — 본 fixture 는 character + bg 만 contract 검증" } ] } ``` - [ ] **Step 2: integration test 작성** ```python # backend/tests/integration/test_c01_zero_gate_regression.py """C01 zero-gate regression — 6 NG stills 에 대해 single ↔ batch contract 동등성 가드.""" import json import pytest from pathlib import Path FIXTURE_PATH = Path(__file__).parent.parent / "fixtures" / "c01_zero_gate_regression.json" @pytest.fixture def regression_fixture(): return json.loads(FIXTURE_PATH.read_text(encoding="utf-8")) def test_all_stills_helper_output_matches_fixture(regression_fixture, db_session): """spec AC-1 — build_scene_attached_refs helper 결과 label set 가 fixture expected_actual_ref_labels 와 일치.""" from app.services.scene_generation_coordinator import build_scene_attached_refs from app.models.project import SceneStill from app.services.scene_reference_service import SceneReferenceService from app.services.scene_checkpoint_loaders import load_background_chain_bg_map from sqlalchemy import bindparam, text pid = regression_fixture["project_id"] eid = regression_fixture["episode_id"] for entry in regression_fixture["stills"]: sid = entry["still_id"] still = db_session.query(SceneStill).filter(SceneStill.id == sid).one() visible_entities = json.loads(still.visible_entities_json or "[]") ref_svc = SceneReferenceService(db_session, pid, actor_id="test") entity_lookup = ref_svc.load_episode_entity_lookup(eid) ref_image_map = ref_svc.load_entity_reference_images(list(entity_lookup.values())) bg_map = load_background_chain_bg_map( "/tmp/projects-fixture", pid, eid, # fixture projects_dir 또는 settings.projects_dir ) # helper 호출 — mock _build_final_scene_prompt 로 LLM call 차단 with patch("app.services.scene_generation_coordinator._build_final_scene_prompt", return_value="MOCK_FINAL"): full_prompt, labeled_refs = build_scene_attached_refs( still=still, episode_id=eid, still_data={"scene_index": still.scene_index, "shot_index": still.shot_index, "still_frame_prompt": still.still_frame_prompt or "", "beat_title": still.beat_title or ""}, visible_entities=visible_entities, ref_image_map=ref_image_map, background_chain_bg_map=bg_map, best_prev_bytes=None, current_location_ids=[], dep_scene_id=None, stills=[], location_scene_history={}, dep_detail_map={}, staging={"camera_direction": "extreme close-up" if entry["is_close_framing"] else "medium shot"}, state_variant_sids={}, entity_lookup=entity_lookup, db=db_session, project_id=pid, project_config={}, reference_svc=ref_svc, ) actual_labels = sorted(lbl for lbl, _ in labeled_refs) expected = sorted(entry["expected_actual_ref_labels"]) # S13 같은 face-close-up 케이스는 RefContractError 가 별도 test 에서 검증. # 본 test 는 helper output (validator 호출 전) 가 fixture 와 일치 검증. if entry["label"] == "S13_Shot6": # face close-up character ref 누락 케이스 — 별도 test 로 분리 (RefContractError) continue assert actual_labels == expected, ( f"{entry['label']}: actual={actual_labels} expected={expected}" ) def test_s13_face_close_up_raises_ref_contract_error(regression_fixture): """spec AC-2 — S13_Shot6 face close-up + visible C01/C02 → RefContractError.""" from app.core.ref_contract_validator import validate_attached_refs, RefContractError s13 = next(e for e in regression_fixture["stills"] if e["label"] == "S13_Shot6") rpc = {"asset_requirements": {"required_refs": {"character_outlook": ["C01O02", "C02O03"]}}} labeled_refs = [] # face close-up 에서 면제로 빈 리스트 prompt = "Focus on the tear-filled, dark eyes of a young Korean woman" with pytest.raises(RefContractError): validate_attached_refs(rpc, labeled_refs, prompt, is_close_framing=True) def test_s8_background_required_phantom_from_reference_raises(regression_fixture): """spec AC-4 — 'sliding door from the reference' + bg ref 부재 → fail.""" from app.core.ref_contract_validator import validate_attached_refs, RefContractError rpc = {"asset_requirements": {"required_refs": {"background": ["L09"]}}} labeled_refs = [("character C01O02 in outfit", b"x")] prompt = "her eyes fixed on the sliding door from the reference" with pytest.raises(RefContractError): validate_attached_refs(rpc, labeled_refs, prompt, is_close_framing=False) ``` - [ ] **Step 3: canary script 작성** ```python # backend/scripts/canary_single_vs_batch_refs.py """Daily canary — PID 80f62523 5 NG stills 의 latest llm_call_log actual refs 점검. post-deploy 실행 — 회귀 발견 시 alert. """ from __future__ import annotations import json import sys from pathlib import Path sys.path.insert(0, str(Path(__file__).parent.parent)) from app.models import project, catalog # noqa from app.core.database import SessionLocal from sqlalchemy import text FIXTURE_PATH = Path(__file__).parent.parent / "tests" / "fixtures" / "c01_zero_gate_regression.json" def main() -> int: fixture = json.loads(FIXTURE_PATH.read_text(encoding="utf-8")) pid = fixture["project_id"] eid = fixture["episode_id"] failures = [] with SessionLocal() as db: for entry in fixture["stills"]: sid = entry["still_id"] row = db.execute(text(""" SELECT reference_image_ids FROM llm_call_log WHERE project_id=:pid AND episode_id=:eid AND operation_type IN ('single_scene_image_gen', 'scene_image_gen') AND metadata_json::text LIKE :sid_pat ORDER BY created_at DESC LIMIT 1 """), {"pid": pid, "eid": eid, "sid_pat": f'%{sid[:8]}%'}).first() if not row: failures.append((entry["label"], "no llm_call_log entry")) continue actual = json.loads(row[0] or "[]") expected = entry["expected_actual_ref_labels"] missing = [e for e in expected if e not in actual] if missing: failures.append((entry["label"], f"missing refs: {missing}")) if failures: print(f"CANARY FAILURE: {len(failures)} stills") for label, reason in failures: print(f" {label}: {reason}") return 1 print(f"CANARY OK: {len(fixture['stills'])} stills") return 0 if __name__ == "__main__": sys.exit(main()) ``` - [ ] **Step 4: 테스트 실행** Run: `cd backend && PYTHONPATH=. .venv/bin/pytest tests/integration/test_c01_zero_gate_regression.py -v` Expected: PASS (Task 1-5 통과 후). Run: `cd backend && PYTHONPATH=. .venv/bin/python scripts/canary_single_vs_batch_refs.py` Expected: 출력 `CANARY OK: 5 stills` 또는 specific failure list. - [ ] **Step 5: 5 NG stills 단건 재호출 검증 (수동)** POST `/api/v1/projects/80f62523-.../stills/{still_id}/generate-image` 5건 sequential. llm_call_log 의 actual refs 가 expected_actual_ref_labels 충족하는지 fixture 기준 검증. 회귀 발견 시 spec/Task 1-5 patch 검토. - [ ] **Step 6: commit** ```bash git add backend/tests/fixtures/c01_zero_gate_regression.json \ backend/tests/integration/test_c01_zero_gate_regression.py \ backend/scripts/canary_single_vs_batch_refs.py git commit -m "$(cat <<'EOF' test(scene_image): C01 zero-gate regression fixture + canary Task 6 of single-vs-batch reference contract fix (spec: 2026-05-08-single-batch-reference-contract-design.md §5) - backend/tests/fixtures/c01_zero_gate_regression.json - S8_Shot4 / S13_Shot6 / S15_Shot5 / S19_Shot1 / S21_Shot6 fixture - visible_short_ids + expected_actual_ref_labels + user_reported_defect - integration test (single ↔ batch label set 동등 / S13 RefContractError / S8 phantom from-the-reference RefContractError) - backend/scripts/canary_single_vs_batch_refs.py — daily post-deploy canary - Non-goals: P0-5 narrative-critical prop carry 는 본 fixture 외 (note 명시) Co-Authored-By: Claude Opus 4.7 (1M context) EOF )" ``` --- ## Self-Review Checklist Plan 작성 후 self-review (writing-plans skill 가이드): 1. **Spec coverage**: spec §4.1 (helper extract) → Task 1+2 ✓ / §4.2 (validator) → Task 3 ✓ / §4.3 (face exemption) → Task 4 ✓ / §4.4 (observability) → Task 5 ✓ / §5.2 (regression fixture) → Task 6 ✓ / §3.1 (Non-goals enforcement) → header literal block ✓. 2. **Placeholder scan**: `# TODO: 실제 helper 호출 + label set 비교` 한 곳 (Task 6 Step 2) — fixture/mock 구성은 task 실행 시 명확. 나머지 step 의 코드 블록 모두 구체. 3. **Type consistency**: `build_scene_attached_refs` signature 가 Task 1 정의와 Task 2 호출 동일 ✓ / `RefContractError` Task 3 정의 + Task 6 import ✓ / `actual_attached_refs` field Task 5 정의 + image_to_dict join ✓. --- ## Execution Handoff Plan saved to `docs/superpowers/plans/2026-05-08-single-batch-reference-contract-implementation.md`. **Two execution options**: **1. Subagent-Driven (recommended)** — leader가 fresh subagent per task dispatch, two-stage review, fast iteration. **2. Inline Execution** — executing-plans skill 로 batch execution + checkpoints. **Which approach?**