# Deterministic bg_id & Cross-Step Catalog Lineage Implementation Plan (D6) **Status**: DRAFT_R2 — Claude audit NEEDS_REVISION_MAJOR 적용 (B1~B7 BLOCKING + I1~I8 IMPORTANT + 추가발견 10건). Codex 재audit 생략 (사용자 결정 — Claude audit 깊이 충분). 다음 세션에서 구현 진입. R1 → R2 변경 history: - **B1 정정**: scene_detail manifest stamp 위치를 `render_prompt_card.py` (helper) → `detail_steps.py:104,798,819` (실제 SceneDetailStep) 로 정정. File Structure + T5 모두. - **B2 patch**: `_SAFE_BG_RE = ^[a-z0-9][a-z0-9_]*$` 가 `background_render_step.py:43,178,562,586` 4 site 에서 새 `L09B01` 거부 — T1 에 sub-step 추가 (`BG_ID_RE` 로 교체). - **B3 patch**: T4 의 flat catalog 가 기존 `data.plans[gid].plan.{...}` per-group shape 를 깨지 않도록 — sibling 추가 정책 (per-group 보존 + flat catalog 추가). 추가발견 #3 (master_plan 의 `plan.backgrounds[].bg_id` slot 에 새 ID 갱신) handshake 도 명시. - **B4 patch**: T2 entity sync chain 4 site 모두 명시 — (a) `ENTITY_DETAIL_SCHEMA.additionalProperties=False` 에 `metadata_json` 추가, (b) `entity_t2i._gen_t2i` source-forward 에 추가, (c) `EntitySyncService.sync_from_checkpoint` UPSERT 에 write, (d) test 가 `sync_entities` 모듈 함수 → `EntitySyncService` instance method 로 변경. - **B5 patch**: T0 alembic 의 sequential file naming (`007_d6_...`) + 006 의 inspector idempotent 패턴 적용 + `_migrations` block dual-track 결정 (alembic-only 채택, `_migrations` 미추가). - **B6 patch**: T3+T4 atomic commit 합치기 + T3 prerequisite 로 `validate_master_plan_output` split (raw_intent validator + assigned validator) 추가. intermediate commit red 차단. - **B7 patch**: `_max_b_for_loc` 의 monotonic next_b 정책 명시 (history 기반 — deletion 후 reset 안 함). spec §4.5 보강 + AC-1 명시. - **I1~I8 patch**: preflight ordering / 배치 endpoint / chain_bg_lookup 분기 / floor_plan_render note / err["upstream"] / T5 split / character branch / intent ordering 모두 cover. > **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking. **Goal:** `bg_id` 를 LLM-generated free-form 에서 code-assigned deterministic ID (`L09B01`) 로 전환 + cross-step catalog lineage hash + dispatcher preflight 의 STALE_UPSTREAM error. master_plan 재실행마다 동일 ID 보장 + drift 감지. **Architecture:** - master_plan 후처리에서 코드가 `(loc_id, space_key, time_phase, state_class)` semantic_key 로 dedup 후 deterministic ID 부여 (`f"L{loc_num:02d}B{var_num:02d}"`). - `space_key` 는 `entity_canon.metadata_json.location.space_profile` SOT (single_space → main 강제 / multi_space → enum strict). - `state_class` 는 controlled vocab enum (master_plan prompt 에 inject + code 검증). - `bg_catalog_hash` (catalog only, metadata 제외) + `shot_binding_hash` (shot_background_map only) 두 hash 분리 — consumer 별 stamp. - dispatcher preflight 가 endpoint entry 에서 hash 정합 검증 → mismatch 시 `STALE_UPSTREAM` HTTP 422. - scene_detail prompt 변경 0 — `ctx.chain_bg_id_by_shot` chain 그대로 (R3 N1). **Tech Stack:** Python 3.12 / FastAPI / SQLAlchemy / pytest / Alembic / OpenAI gpt-5.5 (master_plan) + Gemini (entity_extractor). DB schema 변경 1건 (`entity_canon.metadata_json` column 추가). **Spec:** `docs/superpowers/specs/2026-05-09-deterministic-bg-id-and-catalog-lineage.md` (DRAFT_R3.1 — APPROVED_FOR_PLAN_DRAFTING) **Today's date:** 2026-05-09 / branch: `main` / HEAD: `699a41c` --- > **Non-goals enforcement** (이 plan 의 모든 task 에서 절대 추가 금지): > - [ ] cross-episode ID 일관성 — episode-local 만 보장 > - [ ] legacy compatibility map / dispatcher 분기 — force re-run 으로 처리 > - [ ] embedding/fuzzy clustering — enum strict 만 > - [ ] floor_plan ID 결정론화 — fp_id 기존 longform 그대로 (별도 후속 spec) > - [ ] custom_prompt path 의 ref-builder 개편 — preflight 만 통과, ref schema 별도 후속 > - [ ] floor_plan_render lineage stamp — fp_id D6 외라 stamp 대상 아님 > - [ ] scene_detail prompt 변경 — R3 N1 으로 0 > - [ ] auto-rerun on hash mismatch — preflight = single source, _config_hash 미포함 > - [ ] `data.plans[gid].plan` per-group shape 제거 — **R2 B3 결정: 보존 + sibling 추가** (downstream reader 영향 0) > - [ ] `_migrations` block 추가 — **R2 B5 결정: alembic-only ownership** (006 패턴 반복) --- ## R2 핵심 변경 요약 (Claude audit B1~B7 + I1~I8 cover) R1 의 14 task 그대로 유지하되, 다음 R2 patch 사항이 각 task body 안에 반영됨: | 결함 | task 위치 | R2 patch 방향 | |---|---|---| | **B1** wrong file (scene_detail) | T5 step 6 + File Structure | `render_prompt_card.py` → `detail_steps.py:104,798,819` 정정 | | **B2** `_SAFE_BG_RE` silent reject | **T1 새 sub-step** | 4 site 모두 `BG_ID_RE` 로 교체 + 명시 test | | **B3** per-group vs flat shape | T4 + T5 | flat catalog 를 sibling 으로 추가 (per-group 보존). plan.backgrounds[].bg_id 도 새 ID handshake | | **B4** entity sync 4 site | T2 (전면 재작성) | schema 확장 + entity_t2i forward + EntitySyncService UPSERT + test 의 클래스 호출 | | **B5** alembic 정합 | T0 | sequential prefix `007_` + 006 inspector pattern + `_migrations` 미추가 결정 | | **B6** T3+T4 intermediate red | T3+T4 atomic merge + T3.5 prerequisite | validator split (raw_intent + assigned) 사전, atomic commit | | **B7** _max_b_for_loc determinism | T4 + AC-1 | monotonic next_b 명시 (history 기반, deletion 후 reset 안 함) | | **I1** preflight ordering | T8 | `check_scene_images_ready`/`assert_episode_asset_readiness` 보다 **앞**. STALE_UPSTREAM = first signal | | **I2** 배치 endpoint preflight | T8 | 배치 wiring 전용 sub-step + integration test | | **I3** chain_bg_lookup 분기 | T9 | `BG_ID_RE` match → STALE_UPSTREAM, legacy `bg_*` → RefContractError | | **I4** floor_plan_render note | T11 runbook | non-stamp 명시 + 운영자 force-yes-skip-OK 안내 | | **I5** err["upstream"] | T13 | assertion 에 `err["upstream"] == "background_render"` 명시 | | **I6** T5 incremental fixture | T5 split | 4 sub-task (per-step RED/GREEN) — fixture pre-build 모순 제거 | | **I7** character/prop branch | T2 prompt | location 만 metadata_json 출력, others 미출력 + UPSERT default `'{}'` | | **I8** intent ordering | T4 + AC-1 | `assign_bg_ids` 가 intent sort by `(loc_id, time_phase, state_class)` 후 처리 — 첫 run 결정론 | --- ## R2 새 사전 task — T-pre R1 의 T0 진입 전, R2 가 추가하는 사전 task: ### T-pre-1: `_SAFE_BG_RE` 4 site 교체 (B2) **Files:** - Modify: `backend/app/core/steps/background_render_step.py:43, 178, 562, 586` - Test: `backend/tests/core/test_d6_safe_bg_re_replacement.py` ```python # backend/tests/core/test_d6_safe_bg_re_replacement.py def test_background_render_step_accepts_uppercase_bg_id(): """L09B01 형식이 _SAFE_BG_RE → BG_ID_RE 교체 후 통과.""" from app.core.steps.background_render_step import _BG_ID_FILTER_RE assert _BG_ID_FILTER_RE.match("L09B01") assert _BG_ID_FILTER_RE.match("L113B07") assert not _BG_ID_FILTER_RE.match("Bg_invalid") ``` **Modify** — `background_render_step.py` 의 모듈 레벨 `_SAFE_BG_RE` import → `from app.core.bg_state_vocab import BG_ID_RE as _BG_ID_FILTER_RE` (T1 의 `BG_ID_RE` 와 동일 정규식). 4 site (line 43 정의, 178 `_execute`, 562/586 `_recover_from_self_cp`) 모두 새 변수 사용. T1 dependency 라 T1 후 즉시 진행. ### T-pre-2: master_plan validator split (B6) **Files:** - Modify: `backend/app/modules/pipeline/background_master_plan.py:78-228` - Test: `backend/tests/core/test_d6_master_plan_validator_split.py` 기존 monolithic `validate_master_plan_output` (`bg_id` field hard-require) 를 다음 두 함수로 split: ```python def validate_master_plan_raw_intent(intents: List[Dict]) -> None: """LLM raw intent 검증 — bg_id 없음. fp_id/sub_location_label/state_label_raw 만 검사.""" from app.core.bg_state_vocab import LLM_INTENT_ID_RE for intent in intents: for fp_id in intent.get("depends_on_fp") or []: if not LLM_INTENT_ID_RE.match(fp_id): raise MasterPlanError(f"raw fp_id {fp_id!r} fails LLM_INTENT_ID_RE") # state_class 는 bg_state_vocab.validate_state_class 가 별도 호출. def validate_master_plan_assigned(catalog: Dict[str, Dict]) -> None: """code-assigned bg_id 검증 — `^L\\d{2,3}B\\d{2,3}$` 강제.""" from app.core.bg_state_vocab import BG_ID_RE seen = set() for bg_id, entry in catalog.items(): if bg_id in seen: raise MasterPlanError(f"duplicate bg_id {bg_id}") if not BG_ID_RE.match(bg_id): raise MasterPlanError(f"assigned bg_id {bg_id!r} fails BG_ID_RE") seen.add(bg_id) ``` 기존 `validate_master_plan_output` 호출 (`run_background_master_plan` retry loop, line 260) 도 두 분리 호출로 변경. **T-pre-2 와 T3+T4 atomic merge 가 같은 commit** — intermediate red 차단. --- --- ## File Structure ### 새 파일 | 파일 | 책임 | |---|---| | `backend/app/core/bg_state_vocab.py` | `STATE_CLASS_ENUM` + `BG_ID_RE` + `LLM_INTENT_ID_RE` + `format_bg_id` + helper | | `backend/app/core/bg_catalog.py` | `compute_semantic_key` + `assign_bg_ids` + `compute_bg_catalog_hash` + `compute_shot_binding_hash` + `normalize_space_key` + `build_shot_background_map` | | `backend/alembic/versions/007_d6_entity_canon_metadata_json.py` | migration (sequential prefix — B5) | | `backend/app/services/dispatcher_preflight.py` | endpoint preflight 로직 — hash 비교 + STALE_UPSTREAM raise | | `backend/app/core/errors/stale_upstream.py` | `StaleUpstreamError` exception class | | `docs/runbooks/d6-migration.md` | operator ordered-force runbook | ### 수정 파일 (R2 — B1 정정 + 추가 site 명시) | 파일 | 변경 | |---|---| | `backend/app/models/project.py:28-42` | `EntityCanon.metadata_json` Column 추가 | | `backend/app/modules/pipeline/entity_extractor_v3.py:84-103` | **B4 추가** — `ENTITY_DETAIL_SCHEMA` 에 `metadata_json: {type: object}` 추가 (`additionalProperties=False` 와 정합) | | `backend/app/core/steps/entity_steps.py:600-660, 768-799` | **B4 명시** — `entity_detail_batch` 가 `metadata_json` field forward + `entity_t2i._gen_t2i` source-forward 에 metadata_json 추가 | | `backend/app/services/checkpoint_sync/entity_sync_service.py:25, 120-148` | **B4 명시** — `EntitySyncService.sync_from_checkpoint` 의 UPSERT (`existing = existing_canons.get(name)` 영역) 에 `metadata_json = json.dumps(ent.get("metadata_json") or {})` write | | `backend/app/modules/pipeline/background_master_plan.py:22,78-228,260` | **B6 명시** — 기존 monolithic `validate_master_plan_output` 을 `validate_master_plan_raw_intent` (no bg_id) + `validate_master_plan_assigned` (with bg_id) 로 split. T3+T4 atomic | | `backend/app/core/steps/background_master_plan_step.py:124-183` | **B3 명시** — manifest 의 `data.plans[gid].plan.{...}` per-group shape **보존** + sibling 으로 `data.background_catalog` + `data.shot_background_map` + `data.bg_catalog_hash` + `data.shot_binding_hash` 추가. 추가로 `plan.backgrounds[].bg_id` slot 도 새 `L09B01` 형식으로 갱신 (handshake — 추가발견 #3) | | `prompts/_base/background_master_plan//system.md` | raw intent schema (loc_id/space_key_hint/time_phase/state_class enum) | | `prompts/_base/entity_extractor//system.md` | **I7 명시** — location entity 만 `metadata_json.location.space_profile` 출력. character/prop 은 metadata_json 미출력 (코드가 default `'{}'` 채움) | | `backend/app/core/steps/floor_plan_prompt_step.py:34-92` | `consumed_bg_catalog_hash` stamp + 새 `data.background_catalog` 우선 read (legacy fallback `data.plans[gid].plan`) | | `backend/app/core/steps/background_prompt_step.py` | `consumed_bg_catalog_hash` + `consumed_shot_binding_hash` stamp + 새 catalog 우선 read | | `backend/app/core/steps/background_render_step.py:43, 178, 562, 586, ...` | **B2 명시** — `_SAFE_BG_RE = ^[a-z0-9][a-z0-9_]*$` 4 site 모두 → `BG_ID_RE = ^L\d{2,3}B\d{2,3}$` 교체. `consumed_bg_catalog_hash` stamp + schema_version bump | | **`backend/app/core/steps/detail_steps.py:104, 798, 819`** (B1 정정 — `render_prompt_card.py` 아님) | **scene_detail step 진짜 위치**. `SCENE_DETAIL_SCHEMA_VERSION` (line 104) bump + `_execute` return dict (line 819) 에 `consumed_*_hash` stamp | | `backend/app/api/v1/images.py:387-406, 397, 599-650` | router function entry 에 preflight wiring. **I1 명시** — preflight 가 `check_scene_images_ready` (line 397) **이전** 에 동작 (STALE_UPSTREAM 이 first signal). I2 명시 — 배치 endpoint 도 동일 wiring + `enforce_shot_binding=True` (배치는 custom_prompt 의미 없음) | | `backend/app/core/ref_contract_validator.py:178-318` | error code 분리. **I3 명시** — bg_id 가 `BG_ID_RE` match 시 STALE_UPSTREAM (D6 path), 그 외 (legacy `bg_*`) RefContractError (D5 정합) | | `backend/app/core/step_manifest.py:55-90` | step 별 schema_version bump 정합 (5 steps — master_plan / floor_plan_prompt / background_prompt / background_render / scene_detail). `_LEGACY_SCHEMA_BUMP_ALLOWLIST` 미수정 (모든 D6 step BLOCK on schema bump = 운영자 force 의무 — D6-D 정합) | ### Test 파일 | 파일 | 책임 | |---|---| | `backend/tests/core/test_bg_state_vocab.py` | enum strict / regex / format helper | | `backend/tests/core/test_bg_catalog.py` | semantic_key / assign_bg_ids / hash 분리 / normalize_space_key | | `backend/tests/services/test_dispatcher_preflight.py` | endpoint preflight + STALE_UPSTREAM | | `backend/tests/core/test_d6_master_plan_determinism.py` | 5회 재실행 → bg_id 변동 0 + I8 shuffle + handshake (T7 unit-style pivot) | | `backend/tests/core/test_d6_hash_isolation.py` | catalog 안정 + binding 변경 시 hash 분리 동작 (T7 unit-style pivot) | | `backend/tests/core/test_d6_scene_detail_chain_unchanged.py` | scene_detail bg_id pass-through + R2 폐기 schema 부재 (T6 unit-style pivot) | | `backend/tests/integration/test_d6_force_order.py` | wrong-order force → STALE_UPSTREAM smoke | | `backend/tests/services/test_d6_custom_prompt_preflight.py` | custom_prompt 의 preflight 통과 의무 | --- ## T0: Alembic migration — `entity_canon.metadata_json` (R2 B5 patch) **R2 B5 결정**: - 파일명 sequential prefix `007_d6_entity_canon_metadata_json.py` (기존 `001..006` 패턴 정합). - `006_llm_call_log_metadata_json.py` 의 inspector idempotent 패턴 적용 (restart safety). - `_migrations` block (`database.py:63-218`) **미추가** — alembic-only ownership. 기존 column 들도 historical reasons 로 dual-track 이지만 신규는 alembic 단일 경로로 정리. CLAUDE.md 의 "alembic upgrade head → backend startup" 시퀀스 정합. **구현 함정 (review 발견)**: - **alembic_version VARCHAR(32) 제한**: revision_id 가 32 char 초과 시 alembic_version UPDATE fail (`StringDataRightTruncation`). plan 의 `007_d6_entity_canon_metadata_json` (33 char) 그대로 쓰면 fail — 실제 revision_id 는 27 char 로 단축 (`007_d6_entity_metadata_json`). 파일명은 sequential prefix 보존 위해 `007_d6_entity_canon_metadata_json.py` 유지. - **test DB 는 alembic 미사용**: `init_db()` 의 `Base.metadata.create_all(checkfirst=True)` 가 기존 테이블의 새 column 자동 추가 안 함. Plan B5 (`_migrations` block 미추가) 와 충돌 없이 보강하려면 `tests/conftest.py` 에 **session-scope** autouse fixture (`_ensure_d6_test_db_schema`) 로 `Base.metadata.create_all` + `ALTER TABLE entity_canon ADD COLUMN IF NOT EXISTS metadata_json TEXT NOT NULL DEFAULT '{}'` 1회 보강. 예외는 `ProgrammingError` / `OperationalError` 만 narrow catch + raise (silent skip 시 stale schema 다른 test 까지 전염). cross-order 방어로 `test_d6_metadata_json_migration.py` 에 module-scope autouse `_module_level_d6_schema_fence` 추가 (`safe_drop_all` 을 쓰는 다른 file 이 같은 세션에서 먼저 돌 경우). production 은 여전히 `alembic upgrade head` 의무. - **project_export/import round-trip desync (review I2)**: `_row_to_dict` 는 export 에 새 column 자동 포함하지만, `ProjectExportService` 의 import side `EntityCanon(...)` constructor 는 명시적 keyword 만 받음. metadata_json 누락 시 round-trip 으로 server_default '{}' 로 reset → space_profile 손실. 동시에 `short_id` / `t2i_prompt` 도 같은 desync 카테고리 — 함께 보강. **Files:** - Create: `backend/alembic/versions/007_d6_entity_canon_metadata_json.py` - Test: `backend/tests/core/test_d6_metadata_json_migration.py` - [ ] **Step 1: Write the failing test** ```python # backend/tests/core/test_d6_metadata_json_migration.py """metadata_json column 존재 + default '{}' + JSON parse 가능 검증.""" from sqlalchemy import inspect, text from app.core.database import engine from app.models.project import EntityCanon def test_entity_canon_has_metadata_json_column(): insp = inspect(engine) cols = {c["name"] for c in insp.get_columns("entity_canon")} assert "metadata_json" in cols def test_metadata_json_default_empty_dict(db_session): import uuid, json from datetime import datetime, timezone pid = "test-d6-mig-" + uuid.uuid4().hex[:8] db_session.execute( text( "INSERT INTO project_registry (id, name, created_at) " "VALUES (:id, :name, :ts)" ), {"id": pid, "name": "d6 mig", "ts": datetime.now(timezone.utc).isoformat()}, ) ent = EntityCanon( id=str(uuid.uuid4()), project_id=pid, short_id="L01", entity_type="location", name="test_loc", created_at=datetime.now(timezone.utc).isoformat(), updated_at=datetime.now(timezone.utc).isoformat(), ) db_session.add(ent) db_session.commit() db_session.refresh(ent) assert ent.metadata_json == "{}" assert json.loads(ent.metadata_json) == {} ``` - [ ] **Step 2: Run test to verify it fails** ``` cd backend && .venv/bin/python -m pytest tests/core/test_d6_metadata_json_migration.py -xvs ``` Expected: FAIL — `metadata_json` column 없음 (`AttributeError: 'EntityCanon' object has no attribute 'metadata_json'`) - [ ] **Step 3: alembic revision (sequential prefix + idempotent — R2 B5)** ``` cd backend && .venv/bin/alembic revision -m "d6_entity_canon_metadata_json" ``` 생성된 파일을 `007_d6_entity_canon_metadata_json.py` 로 rename + 다음 내용: ```python # backend/alembic/versions/007_d6_entity_canon_metadata_json.py """d6 entity_canon metadata_json Revision ID: 007 Revises: 006 Create Date: 2026-05-09 D6 prerequisite — EntityCanon.stable_traits 의 list 계약 보존을 위해 별도 metadata_json column 추가. location entity 의 space_profile 저장. inspector idempotent (006 패턴 정합) — restart safety. `_migrations` block (database.py:63-218) 미추가 — alembic-only ownership. """ from alembic import op import sqlalchemy as sa from sqlalchemy import inspect revision = "007" down_revision = "006" branch_labels = None depends_on = None def upgrade() -> None: bind = op.get_bind() insp = inspect(bind) cols = {c["name"] for c in insp.get_columns("entity_canon")} if "metadata_json" not in cols: op.add_column( "entity_canon", sa.Column( "metadata_json", sa.Text(), nullable=False, server_default="{}", ), ) def downgrade() -> None: bind = op.get_bind() insp = inspect(bind) cols = {c["name"] for c in insp.get_columns("entity_canon")} if "metadata_json" in cols: op.drop_column("entity_canon", "metadata_json") ``` - [ ] **Step 4: Model 에 column 추가** `backend/app/models/project.py:28-42` (EntityCanon class): ```python class EntityCanon(Base): __tablename__ = "entity_canon" id = Column(Text, primary_key=True) project_id = Column(Text, ForeignKey("project_registry.id"), nullable=False) short_id = Column(Text) # C01, L03, P05, O01 — LLM 통신용 짧은 ID entity_type = Column(Text, nullable=False) # character|location|prop|outlook name = Column(Text, nullable=False) description = Column(Text) stable_traits = Column(Text, default="{}") metadata_json = Column(Text, nullable=False, default="{}") # D6: location.space_profile 등 t2i_prompt = Column(Text) status = Column(Text, default="active") created_at = Column(Text, nullable=False) updated_at = Column(Text, nullable=False) ``` - [ ] **Step 5: Migration 실행 + 테스트 통과** ``` cd backend && .venv/bin/alembic upgrade head .venv/bin/python -m pytest tests/core/test_d6_metadata_json_migration.py -xvs ``` Expected: PASS - [ ] **Step 6: Commit** ``` git add backend/alembic/versions/*_d6_*.py backend/app/models/project.py backend/tests/core/test_d6_metadata_json_migration.py git commit -m "feat(d6): T0 metadata_json column on entity_canon (alembic + model)" ``` --- ## T1: bg_state_vocab + ID 정규식 + format helper **Files:** - Create: `backend/app/core/bg_state_vocab.py` - Test: `backend/tests/core/test_bg_state_vocab.py` - [ ] **Step 1: Write the failing tests** ```python # backend/tests/core/test_bg_state_vocab.py """bg_state_vocab — STATE_CLASS_ENUM strict + ID regex + format helper.""" import pytest def test_state_class_enum_includes_required_set(): from app.core.bg_state_vocab import STATE_CLASS_ENUM expected = { "normal", "quiet", "busy", "busy_exit", "ransacked", "clean_after", "blood_scene", "intrusion", "arrival", "evidence_display", "dream_or_vision_state", } assert expected.issubset(STATE_CLASS_ENUM) def test_state_class_enum_strict_validates(): from app.core.bg_state_vocab import validate_state_class, StateClassError validate_state_class("normal") # OK with pytest.raises(StateClassError): validate_state_class("unknown") with pytest.raises(StateClassError): validate_state_class("dusk_normal") # 자유 텍스트 reject def test_bg_id_regex_accepts_2_3_digit(): from app.core.bg_state_vocab import BG_ID_RE assert BG_ID_RE.match("L09B01") assert BG_ID_RE.match("L113B07") assert BG_ID_RE.match("L09B100") assert not BG_ID_RE.match("bg_store_sales_floor_dusk") assert not BG_ID_RE.match("L9B01") # 1-digit reject assert not BG_ID_RE.match("l09b01") # lowercase reject def test_llm_intent_id_regex_lowercase_snake(): from app.core.bg_state_vocab import LLM_INTENT_ID_RE assert LLM_INTENT_ID_RE.match("fp_supermarket_sales_floor") assert LLM_INTENT_ID_RE.match("store_sales_floor") assert not LLM_INTENT_ID_RE.match("L09B01") # uppercase reject assert not LLM_INTENT_ID_RE.match("Store Sales Floor") def test_format_bg_id_zero_pad_2_digits(): from app.core.bg_state_vocab import format_bg_id assert format_bg_id(9, 1) == "L09B01" assert format_bg_id(4, 1) == "L04B01" def test_format_bg_id_3_digit_natural_extension(): from app.core.bg_state_vocab import format_bg_id assert format_bg_id(113, 7) == "L113B07" assert format_bg_id(9, 100) == "L09B100" ``` - [ ] **Step 2: Run tests to verify they fail** ``` cd backend && .venv/bin/python -m pytest tests/core/test_bg_state_vocab.py -xvs ``` Expected: FAIL with "No module named 'app.core.bg_state_vocab'" - [ ] **Step 3: Implement `bg_state_vocab.py`** ```python # backend/app/core/bg_state_vocab.py """D6 controlled vocab — bg_id regex + state_class enum + format helper. spec: docs/superpowers/specs/2026-05-09-deterministic-bg-id-and-catalog-lineage.md §4.1, §4.4 LLM 은 ID 를 만들 수 없음. code 가 부여. 두 정규식이 다른 step (raw vs assigned) 에 적용. """ from __future__ import annotations import re from typing import FrozenSet # ────────────────────────────────────────────────────────────────────── # state_class controlled vocab (§4.4) # ────────────────────────────────────────────────────────────────────── STATE_CLASS_ENUM: FrozenSet[str] = frozenset({ "normal", "quiet", "busy", "busy_exit", "ransacked", "clean_after", "blood_scene", "intrusion", "arrival", "evidence_display", "dream_or_vision_state", }) class StateClassError(ValueError): """state_class 가 enum 밖 — fail-fast (nearest-match 안 함).""" def validate_state_class(state_class: str) -> None: if state_class not in STATE_CLASS_ENUM: raise StateClassError( f"state_class {state_class!r} not in enum (size={len(STATE_CLASS_ENUM)}). " f"LLM prompt 의 enum 강제 violation 또는 nearest-match 시도 — fail-fast." ) # ────────────────────────────────────────────────────────────────────── # ID 정규식 (§4.1) # ────────────────────────────────────────────────────────────────────── # code-assigned bg_id — uppercase, 2-3 digit zero-pad BG_ID_RE = re.compile(r"^L\d{2,3}B\d{2,3}$") # LLM raw intent id (fp_id, sub_location_label_canon, state_label_raw_canon 등) # 기존 _SAFE_ID_RE 동일. LLM_INTENT_ID_RE = re.compile(r"^[a-z0-9][a-z0-9_]*$") def format_bg_id(loc_num: int, var_num: int) -> str: """deterministic bg_id 생성. 99 까지 2-digit zero-pad, 100+ 자연 확장.""" return f"L{loc_num:02d}B{var_num:02d}" def parse_loc_num(short_id: str) -> int: """`L09` → 9. invariant: short_id 가 `^L\\d{2,3}$`.""" if not re.match(r"^L\d{2,3}$", short_id): raise ValueError(f"location short_id {short_id!r} invalid — must match ^L\\d{{2,3}}$") return int(short_id[1:]) ``` - [ ] **Step 4: Run tests to verify they pass** ``` cd backend && .venv/bin/python -m pytest tests/core/test_bg_state_vocab.py -xvs ``` Expected: PASS (6 tests) - [ ] **Step 5: Commit** ``` git add backend/app/core/bg_state_vocab.py backend/tests/core/test_bg_state_vocab.py git commit -m "feat(d6): T1 bg_state_vocab — STATE_CLASS_ENUM + BG_ID_RE + format_bg_id" ``` --- ## T2: entity_extractor — `metadata_json.location.space_profile` (R2 B4 + I7 patch) **R2 B4 결정 — 4 site 모두 명시**: 1. `ENTITY_DETAIL_SCHEMA` (`entity_extractor_v3.py:84-103`) 의 `additionalProperties=False` 가 `metadata_json` reject — schema 에 `metadata_json: {type: object}` 추가. 2. `entity_t2i._gen_t2i` source-forward (`entity_steps.py:768-779, 787-799`) — 현재 `description/visual_traits` 만 forward. `metadata_json` 추가. 3. `EntitySyncService.sync_from_checkpoint` UPSERT (`entity_sync_service.py:25-148`) — 현재 `visual_traits` 만 stable_traits 로 sync. `metadata_json` write 추가. 4. T2 test 의 호출 패턴 — R1 의 `from ... import sync_entities` (모듈 함수) 는 미존재. 실제는 `EntitySyncService(db).sync_from_checkpoint()` 인스턴스 메서드. **R2 I7 결정 — character/prop branch**: prompt 변경은 location entity 전용. character/prop 은 `metadata_json` 미출력. EntitySyncService UPSERT 가 entity_type=='location' 만 metadata_json write, 그 외는 default `'{}'` (DB column server_default). **Files:** - Modify: `prompts/_base/entity_extractor//system.md` - Modify: `backend/app/modules/pipeline/entity_extractor_v3.py:84-103` (ENTITY_DETAIL_SCHEMA 확장) - Modify: `backend/app/core/steps/entity_steps.py:768-779, 787-799` (entity_t2i source-forward 에 metadata_json 포함) - Modify: `backend/app/services/checkpoint_sync/entity_sync_service.py:120-148` (UPSERT 에 metadata_json write, location 전용 branch) - Test: `backend/tests/core/test_entity_steps_metadata_json.py`, `backend/tests/core/test_entity_sync_metadata_json.py` - [ ] **Step 1: Write the failing test (R2 B4 정정 — `EntitySyncService` instance method 호출)** ```python # backend/tests/core/test_entity_steps_metadata_json.py """entity_extractor 가 location entity 에 space_profile 채우는지 검증.""" import json def test_location_entity_has_space_profile_after_sync(db_session, project_id, episode_id): """LLM mock 출력에 space_profile 있으면 entity_canon.metadata_json 에 sync. R2 B4: `sync_entities` 모듈 함수 미존재 — 실제는 `EntitySyncService.sync_from_checkpoint()` 인스턴스 메서드. 본 test 는 entity_t2i checkpoint 를 fixture 로 mock 후 sync 호출. """ from app.services.checkpoint_sync.entity_sync_service import EntitySyncService from app.models.project import EntityCanon # entity_t2i checkpoint mock — metadata_json 포함된 location entity # (실제 sync_from_checkpoint 는 entity_t2i cp 를 read 함, line 29) _write_entity_t2i_checkpoint_fixture( project_id, episode_id, entities=[{ "short_id": "L09", "entity_type": "location", "name": "supermarket", "visual_traits": ["bright", "clean"], "metadata_json": { "location": { "space_profile": { "kind": "single_space", "allowed_space_keys": ["main"], } } }, }], ) # sync 호출 svc = EntitySyncService(db_session, project_id, episode_id) svc.sync_from_checkpoint() ent = db_session.query(EntityCanon).filter_by(short_id="L09").first() assert ent is not None parsed = json.loads(ent.metadata_json) assert parsed["location"]["space_profile"]["kind"] == "single_space" assert parsed["location"]["space_profile"]["allowed_space_keys"] == ["main"] def test_character_entity_metadata_json_default_empty(db_session, project_id): """character entity 는 metadata_json={}.""" from app.services.checkpoint_sync.entity_sync_service import sync_entities from app.models.project import EntityCanon sync_entities(db_session, project_id, [{ "short_id": "C01", "entity_type": "character", "name": "수리영", "visual_traits": ["young", "korean"], }]) ent = db_session.query(EntityCanon).filter_by(short_id="C01").first() assert json.loads(ent.metadata_json) == {} def test_multi_space_location_allowed_keys(db_session, project_id): """multi_space location 은 allowed_space_keys 여러 개.""" from app.services.checkpoint_sync.entity_sync_service import sync_entities from app.models.project import EntityCanon import json as _json sync_entities(db_session, project_id, [{ "short_id": "L05", "entity_type": "location", "name": "house", "visual_traits": [], "metadata_json": { "location": { "space_profile": { "kind": "multi_space", "allowed_space_keys": ["main", "kitchen", "rooftop", "stairs"], "default_space_key": "main", } } }, }]) ent = db_session.query(EntityCanon).filter_by(short_id="L05").first() parsed = _json.loads(ent.metadata_json) assert parsed["location"]["space_profile"]["kind"] == "multi_space" assert "kitchen" in parsed["location"]["space_profile"]["allowed_space_keys"] ``` - [ ] **Step 2: Run tests to verify they fail** ``` cd backend && .venv/bin/python -m pytest tests/core/test_entity_steps_metadata_json.py -xvs ``` Expected: FAIL — sync_entities 가 metadata_json field 무시 (현재 visual_traits 만 처리). - [ ] **Step 3: Update entity_sync_service to write metadata_json** `backend/app/services/checkpoint_sync/entity_sync_service.py:120-148` (sync_entities 함수의 EntityCanon insert/update 부분): ```python # 기존 visual_traits → stable_traits 매핑 보존 # 추가: metadata_json 도 ent dict 에서 읽어 sync import json as _json # entity 가 metadata_json key 를 가지면 dict → JSON.dumps. 없으면 default '{}'. metadata_json = _json.dumps(ent.get("metadata_json") or {}) # INSERT/UPDATE 양쪽 set 에 metadata_json=metadata_json 추가 ``` (정확한 patch 는 기존 sync_entities 의 INSERT/UPDATE statement 위치에 metadata_json 컬럼 추가. T0 의 schema 변경 후 ORM 이 자동 인식.) - [ ] **Step 4: entity_extractor prompt 새 버전 디렉토리 생성** ``` prompts/_base/entity_extractor//system.md ``` prompt 변경 — location entity 출력 schema 에 metadata_json 추가: ``` location entity 는 다음 형식으로 출력: { "short_id": "L##", "entity_type": "location", "name": "...", "visual_traits": [...], "metadata_json": { "location": { "space_profile": { "kind": "single_space" | "multi_space", "allowed_space_keys": ["main"] | ["main", "kitchen", ...], "default_space_key": "main" // multi_space 만 } } } } Rules: - single_space: location 안에 단일 시각 공간만 있을 때 (예: 사무실, 작은 매장). allowed_space_keys 는 정확히 ["main"]. - multi_space: 복수 공간 (예: 집의 거실/주방/옥상, 학교의 교실/복도/운동장). allowed_space_keys 는 controlled vocab 안 — main / kitchen / rooftop / stairs / yard / exterior / office. - 위 enum 밖 단어 사용 금지. 모르면 single_space + main 으로 처리. - character / prop / outlook entity 는 metadata_json 출력 안 함 (코드가 default '{}' 채움). ``` prompt_version bump. - [ ] **Step 5: Run tests to verify they pass** ``` cd backend && .venv/bin/python -m pytest tests/core/test_entity_steps_metadata_json.py -xvs ``` Expected: PASS (3 tests) - [ ] **Step 6: Commit** ``` git add prompts/_base/entity_extractor/ backend/app/services/checkpoint_sync/entity_sync_service.py backend/app/core/steps/entity_steps.py backend/tests/core/test_entity_steps_metadata_json.py git commit -m "feat(d6): T2 entity_extractor space_profile output + metadata_json sync" ``` --- ## T3: master_plan prompt redesign — raw intent schema (R2 B6 — T3+T4 atomic, T-pre-2 prerequisite) **R2 B6 atomic merge**: - T3 (prompt) + T4 (post-processing) 가 별도 commit 이면 intermediate red 확실 (R1 audit B6). - 따라서 **T3 + T4 + T-pre-2 (validator split) 는 single atomic commit**. - commit message: `feat(d6): T3+T4 master_plan prompt + post-processing + validator split (atomic)` - 본 task 의 step 5 (commit) 는 T4 의 step 6 (commit) 으로 합쳐짐 — T3 단독 commit 없음. **Files:** - Create: `prompts/_base/background_master_plan//system.md` - Test: `backend/tests/core/test_master_plan_prompt_schema.py` - [ ] **Step 1: Write the failing schema test** ```python # backend/tests/core/test_master_plan_prompt_schema.py """master_plan LLM raw intent schema 검증. prompt 가 enum 강제 + raw intent fields 의무 출력하는지 — schema only test (LLM call 자체는 integration test 에서).""" def test_master_plan_prompt_includes_state_class_enum_and_examples(): """현재 활성 master_plan prompt 가 state_class enum 11 개 모두 명시.""" from pathlib import Path from app.core.bg_state_vocab import STATE_CLASS_ENUM prompt_dirs = sorted(Path("prompts/_base/background_master_plan").glob("*/system.md")) latest = prompt_dirs[-1] text = latest.read_text(encoding="utf-8") for enum_val in STATE_CLASS_ENUM: assert enum_val in text, f"state_class enum {enum_val!r} 누락 in {latest}" def test_master_plan_prompt_demands_raw_intent_fields(): """prompt 가 raw intent 출력 의무 명시 — loc_id/space_key_hint/time_phase/state_class/applies_to_shots/sub_location_label/state_label_raw.""" from pathlib import Path prompt_dirs = sorted(Path("prompts/_base/background_master_plan").glob("*/system.md")) latest = prompt_dirs[-1] text = latest.read_text(encoding="utf-8") required = [ "loc_id", "space_key_hint", "time_phase", "state_class", "applies_to_shots", "sub_location_label", "state_label_raw", ] for field in required: assert field in text, f"raw intent field {field!r} 누락 in {latest}" def test_master_plan_prompt_forbids_bg_id_output(): """prompt 는 LLM 이 bg_id 를 만들지 못하도록 명시.""" from pathlib import Path prompt_dirs = sorted(Path("prompts/_base/background_master_plan").glob("*/system.md")) latest = prompt_dirs[-1] text = latest.read_text(encoding="utf-8") # "bg_id 는 코드가 부여" / "bg_id output 금지" 같은 강제 문구 assert "bg_id" in text assert any(p in text for p in ["코드가 부여", "code-assigned", "do not output bg_id", "출력하지 않"]) ``` - [ ] **Step 2: Run tests to verify they fail** ``` cd backend && .venv/bin/python -m pytest tests/core/test_master_plan_prompt_schema.py -xvs ``` Expected: FAIL — 새 prompt 디렉토리 없음 또는 enum 누락. - [ ] **Step 3: Create new master_plan prompt version** 기존 `prompts/_base/background_master_plan/` 의 latest 버전 디렉토리를 base 로, 새 버전 디렉토리 생성 (버전 형식 `.`). 새 `system.md` 의 요지: ```markdown # Background Master Plan — D6 raw intent schema LLM 출력은 다음 schema 의 raw intent 만. **bg_id 는 절대 출력하지 말 것** — 코드가 후처리에서 부여한다. ## Output schema (per background candidate) { "loc_id": "L##", "space_key_hint": "main" | "kitchen" | "rooftop" | "stairs" | "yard" | "exterior" | "office", "time_phase": "dawn" | "morning" | "day" | "dusk" | "night", "state_class": , "applies_to_shots": ["S_Shot", ...], "sub_location_label": "사람이 읽는 자유 텍스트 (한국어 OK)", "state_label_raw": "사람이 읽는 자유 텍스트 (예: 'dusk busy with employee pointing toward exit')", "depends_on_fp": ["fp_", ...] } ## state_class enum (정확히 한 값. 그 외는 reject): normal / quiet / busy / busy_exit / ransacked / clean_after / blood_scene / intrusion / arrival / evidence_display / dream_or_vision_state ## Rules - bg_id 출력 금지 — 코드가 부여 (`f"L{loc_num:02d}B{var_num:02d}"`). - state_class 는 enum 밖이면 reject. 모르면 `normal` 로 처리하지 말고 진짜 적합한 enum 선택. - space_key_hint 는 single_space location 의 경우 무시되어 `main` 으로 normalize 됨 — 그래도 출력 의무. - time_phase 는 5 enum 중 하나. - sub_location_label / state_label_raw 는 metadata 만, semantic_key 에 안 들어감. - applies_to_shots 는 metadata 만, semantic_key 에 안 들어감. ``` prompt_version bump 의무. - [ ] **Step 4: Run tests to verify they pass** ``` cd backend && .venv/bin/python -m pytest tests/core/test_master_plan_prompt_schema.py -xvs ``` Expected: PASS (3 tests) - [ ] **Step 5: ⚠ T3 단독 commit 금지 (R2 B6) — T4 와 atomic merge** T3 prompt 만 변경하고 commit 하면 LLM 이 다음 run 에서 bg_id 안 출력 → 기존 monolithic `validate_master_plan_output` 가 `_SAFE_ID_RE.match("")` fail → MasterPlanError × 3 retries → production red 확실. → T3 step 5 의 `git add` + `git commit` 은 T4 의 step 6 으로 이동. T3 + T-pre-2 + T4 가 **same commit**. --- ## T4: master_plan post-processing 6 단계 + bg_catalog module (R2 B3 + B7 + I8 patch + T3 atomic merge) **R2 추가 사항 (B3 + B7 + I8)**: - **B3 sibling shape**: `BackgroundMasterPlanStep._execute` (`background_master_plan_step.py:124-183`) 의 `data.plans[gid].plan.{floor_plans, backgrounds, gen_order}` per-group shape 를 **보존**. `data.background_catalog`, `data.shot_background_map`, `data.bg_catalog_hash`, `data.shot_binding_hash` 를 **sibling field** 로 추가. 기존 reader (`floor_plan_prompt_step:78-129`, `background_render_step:121-191`) 은 그대로 동작. 추가발견 #3: `plan.backgrounds[].bg_id` slot 도 새 `L09B01` 로 갱신 (handshake — `background_prompt` 의 reader 가 새 형식 받음). - **B7 monotonic next_b**: `_max_b_for_loc(prev_catalog, loc_id)` 가 prev_catalog 의 max var_num 기반. 삭제된 entry 도 max 계산에 포함 (history 기반 monotonic). spec §4.5 정합. 결과: 새 ID 가 deletion 후에도 reset 안 함. AC-1 의 "5 reruns same input stable" 보장 (다른 input 들어간 후 다시 같은 input 으로 돌아가도 같은 결과 — 단, 그 사이 다른 input 으로 발급된 ID 는 보존). - **I8 intent sort**: `assign_bg_ids` 가 `new_intents` 를 호출 시 `sorted(intents, key=lambda i: (i["loc_id"], i["time_phase"], i["state_class"]))` 후 처리. 첫 run (empty prev_catalog) 에서도 LLM 출력 순서 무관하게 결정론. **Files:** - Create: `backend/app/core/bg_catalog.py` - Modify: `backend/app/modules/pipeline/background_master_plan.py:22,78-260` (T-pre-2 의 validator split 정합) - Modify: `backend/app/core/steps/background_master_plan_step.py:124-183` (sibling shape + handshake) - Test: `backend/tests/core/test_bg_catalog.py` - [ ] **Step 1: Write failing tests for bg_catalog module** ```python # backend/tests/core/test_bg_catalog.py """bg_catalog — semantic_key + assign_bg_ids + hash 분리 + normalize_space_key.""" import json import pytest def _intent(loc_id="L09", space_key_hint="main", time_phase="dusk", state_class="busy_exit", applies_to_shots=None, sub_location_label="store sales floor", state_label_raw="dusk busy", depends_on_fp=None): return { "loc_id": loc_id, "space_key_hint": space_key_hint, "time_phase": time_phase, "state_class": state_class, "applies_to_shots": applies_to_shots or ["S8_Shot4"], "sub_location_label": sub_location_label, "state_label_raw": state_label_raw, "depends_on_fp": depends_on_fp or [], } # ── normalize_space_key ── def test_single_space_location_forces_main_regardless_of_hint(): from app.core.bg_catalog import normalize_space_key profile = {"kind": "single_space", "allowed_space_keys": ["main"]} assert normalize_space_key("L09", "store_sales_floor", profile) == "main" assert normalize_space_key("L09", "main", profile) == "main" assert normalize_space_key("L09", "kitchen", profile) == "main" # ignored def test_multi_space_location_enum_strict(): from app.core.bg_catalog import normalize_space_key, SemanticKeyError profile = { "kind": "multi_space", "allowed_space_keys": ["main", "kitchen", "rooftop"], "default_space_key": "main", } assert normalize_space_key("L05", "main", profile) == "main" assert normalize_space_key("L05", "kitchen", profile) == "kitchen" with pytest.raises(SemanticKeyError): normalize_space_key("L05", "garage", profile) # enum 밖 # ── compute_semantic_key ── def test_semantic_key_excludes_metadata_only_fields(): from app.core.bg_catalog import compute_semantic_key sk1 = compute_semantic_key(loc_id="L09", space_key="main", time_phase="dusk", state_class="busy_exit") sk2 = compute_semantic_key(loc_id="L09", space_key="main", time_phase="dusk", state_class="busy_exit") assert sk1 == sk2 == "L09|main|dusk|busy_exit" def test_semantic_key_state_class_changes_key(): from app.core.bg_catalog import compute_semantic_key sk1 = compute_semantic_key("L09", "main", "dusk", "busy_exit") sk2 = compute_semantic_key("L09", "main", "dusk", "ransacked") assert sk1 != sk2 # ── assign_bg_ids ── def test_assign_bg_ids_reuses_for_same_semantic_key(): from app.core.bg_catalog import assign_bg_ids prev = { "L09B01": { "bg_id": "L09B01", "loc_id": "L09", "space_key": "main", "time_phase": "dusk", "state_class": "busy_exit", "semantic_key": "L09|main|dusk|busy_exit", "applies_to_shots": ["S8_Shot4"], }, } new_intents = [_intent(applies_to_shots=["S8_Shot4", "S8_Shot5"])] # shot 변경 catalog = assign_bg_ids(prev_catalog=prev, new_intents=new_intents, location_profiles={"L09": {"kind": "single_space", "allowed_space_keys": ["main"]}}) assert "L09B01" in catalog # 재사용 assert catalog["L09B01"]["applies_to_shots"] == ["S8_Shot4", "S8_Shot5"] def test_assign_bg_ids_increments_for_new_semantic_key(): from app.core.bg_catalog import assign_bg_ids prev = { "L09B01": {"bg_id": "L09B01", "loc_id": "L09", "semantic_key": "L09|main|dusk|busy_exit"}, } new_intents = [ _intent(state_class="busy_exit"), # 기존 재사용 → L09B01 _intent(state_class="ransacked"), # 새 → L09B02 ] catalog = assign_bg_ids(prev_catalog=prev, new_intents=new_intents, location_profiles={"L09": {"kind": "single_space", "allowed_space_keys": ["main"]}}) assert "L09B01" in catalog assert "L09B02" in catalog def test_assign_bg_ids_5_reruns_stable_id(): from app.core.bg_catalog import assign_bg_ids intents = [_intent(state_class="busy_exit")] profile = {"L09": {"kind": "single_space", "allowed_space_keys": ["main"]}} catalog1 = assign_bg_ids({}, intents, profile) bg_id_1 = next(iter(catalog1.keys())) for i in range(4): catalog_n = assign_bg_ids(catalog1, intents, profile) bg_id_n = next(iter(catalog_n.keys())) assert bg_id_n == bg_id_1, f"run {i+2}: ID drift {bg_id_1} → {bg_id_n}" def test_assign_bg_ids_validates_assigned_format(): from app.core.bg_catalog import assign_bg_ids intents = [_intent()] profile = {"L09": {"kind": "single_space", "allowed_space_keys": ["main"]}} catalog = assign_bg_ids({}, intents, profile) bg_id = next(iter(catalog.keys())) import re assert re.match(r"^L\d{2,3}B\d{2,3}$", bg_id) # ── compute_bg_catalog_hash + compute_shot_binding_hash ── def test_bg_catalog_hash_excludes_applies_to_shots(): """applies_to_shots 만 변동해도 catalog_hash 안정.""" from app.core.bg_catalog import compute_bg_catalog_hash catalog_a = { "L09B01": { "bg_id": "L09B01", "loc_id": "L09", "space_key": "main", "time_phase": "dusk", "state_class": "busy_exit", "semantic_key": "L09|main|dusk|busy_exit", "depends_on_bg": [], "depends_on_fp": ["fp_supermarket"], # metadata "applies_to_shots": ["S8_Shot4"], "sub_location_label": "sales floor", "state_label_raw": "raw a", } } catalog_b = {**catalog_a} catalog_b["L09B01"] = {**catalog_a["L09B01"], "applies_to_shots": ["S8_Shot4", "S8_Shot5"], "sub_location_label": "store interior", "state_label_raw": "raw b"} assert compute_bg_catalog_hash(catalog_a) == compute_bg_catalog_hash(catalog_b) def test_bg_catalog_hash_changes_on_render_relevant_field(): """state_class 변경 → catalog_hash 변동.""" from app.core.bg_catalog import compute_bg_catalog_hash catalog_a = {"L09B01": {"bg_id": "L09B01", "loc_id": "L09", "space_key": "main", "time_phase": "dusk", "state_class": "busy_exit", "semantic_key": "L09|main|dusk|busy_exit", "depends_on_bg": [], "depends_on_fp": []}} catalog_b = {"L09B01": {**catalog_a["L09B01"], "state_class": "ransacked", "semantic_key": "L09|main|dusk|ransacked"}} assert compute_bg_catalog_hash(catalog_a) != compute_bg_catalog_hash(catalog_b) def test_shot_binding_hash_isolated_from_catalog(): """shot_background_map 변경만으로 shot_binding_hash 변동, catalog_hash 안정.""" from app.core.bg_catalog import compute_bg_catalog_hash, compute_shot_binding_hash catalog = {"L09B01": {"bg_id": "L09B01", "loc_id": "L09", "space_key": "main", "time_phase": "dusk", "state_class": "busy_exit", "semantic_key": "L09|main|dusk|busy_exit", "depends_on_bg": [], "depends_on_fp": [], "applies_to_shots": ["S8_Shot4"]}} map_a = {"S8_Shot4": "L09B01"} map_b = {"S8_Shot4": "L09B01", "S8_Shot5": "L09B01"} assert compute_bg_catalog_hash(catalog) == compute_bg_catalog_hash(catalog) assert compute_shot_binding_hash(map_a) != compute_shot_binding_hash(map_b) ``` - [ ] **Step 2: Run tests to verify they fail** ``` cd backend && .venv/bin/python -m pytest tests/core/test_bg_catalog.py -xvs ``` Expected: FAIL with "No module named 'app.core.bg_catalog'" - [ ] **Step 3: Implement `bg_catalog.py`** ```python # backend/app/core/bg_catalog.py """D6 — bg_id catalog: semantic_key dedup + assign_bg_ids + hash 분리. spec: docs/superpowers/specs/2026-05-09-deterministic-bg-id-and-catalog-lineage.md §4.2, §4.5, §4.6 코드 SOT — LLM 은 raw intent 만 출력, 코드가 deterministic ID 부여. hash 두 축 분리: - bg_catalog_hash: catalog 의 render-relevant field (metadata 제외) - shot_binding_hash: shot_background_map 전용 """ from __future__ import annotations import hashlib import json from typing import Any, Dict, List, Optional from app.core.bg_state_vocab import ( BG_ID_RE, STATE_CLASS_ENUM, StateClassError, format_bg_id, parse_loc_num, validate_state_class, ) class SemanticKeyError(ValueError): """semantic_key 구성 실패 — space_key normalize / state_class enum 위반.""" # ────────────────────────────────────────────────────────────────────── # normalize_space_key (§4.3) # ────────────────────────────────────────────────────────────────────── def normalize_space_key(loc_id: str, hint: str, profile: Dict[str, Any]) -> str: """entity_canon.metadata_json.location.space_profile 을 SOT 로 hint normalize. single_space → 항상 main. multi_space → allowed_space_keys 안이면 그대로, 밖이면 raise. """ if not isinstance(profile, dict): raise SemanticKeyError(f"location {loc_id!r} profile not dict: {profile!r}") kind = profile.get("kind") if kind == "single_space": return "main" if kind == "multi_space": allowed = set(profile.get("allowed_space_keys") or []) if hint in allowed: return hint raise SemanticKeyError( f"space_key hint {hint!r} not in {sorted(allowed)} for {loc_id} (multi_space)" ) raise SemanticKeyError(f"location {loc_id!r} unknown space_profile.kind={kind!r}") # ────────────────────────────────────────────────────────────────────── # compute_semantic_key (§4.2) # ────────────────────────────────────────────────────────────────────── def compute_semantic_key( loc_id: str, space_key: str, time_phase: str, state_class: str, ) -> str: """semantic_key = `loc_id|space_key|time_phase|state_class`. metadata-only fields (applies_to_shots / sub_location_label / state_label_raw) 는 절대 들어가지 않음 — drift 회피 (P2/P3/P4). """ validate_state_class(state_class) return f"{loc_id}|{space_key}|{time_phase}|{state_class}" # ────────────────────────────────────────────────────────────────────── # assign_bg_ids (§4.5) # ────────────────────────────────────────────────────────────────────── def _max_b_for_loc(prev_catalog: Dict[str, Dict[str, Any]], loc_id: str) -> int: """주어진 loc_id 안에서 prev_catalog 의 최대 var_num. 없으면 0.""" max_b = 0 for entry in prev_catalog.values(): if entry.get("loc_id") != loc_id: continue bg_id = entry.get("bg_id", "") if not BG_ID_RE.match(bg_id): continue # parse: L{NN}B{NN} 또는 L{NNN}B{NNN} b_part = bg_id.split("B", 1)[1] try: b_num = int(b_part) if b_num > max_b: max_b = b_num except ValueError: continue return max_b def _metadata_from(intent: Dict[str, Any]) -> Dict[str, Any]: """intent 에서 metadata-only field 추출 (semantic_key 안 들어가는 것).""" return { "applies_to_shots": list(intent.get("applies_to_shots") or []), "sub_location_label": intent.get("sub_location_label", ""), "state_label_raw": intent.get("state_label_raw", ""), } def assign_bg_ids( prev_catalog: Dict[str, Dict[str, Any]], new_intents: List[Dict[str, Any]], location_profiles: Dict[str, Dict[str, Any]], ) -> Dict[str, Dict[str, Any]]: """master_plan 후처리 — semantic_key 매칭 + ID 부여. - 기존 catalog 의 같은 semantic_key 가 있으면 ID 재사용 (재실행 안정성). - 새 semantic_key 면 해당 loc 의 next_b 로 새 ID. - 삭제 정책: prev 에 있고 new 에 없는 entry 는 catalog 에서 제거. M4: prev_by_sk index — O(N+M). """ # M4 — index for O(1) semantic_key lookup prev_by_sk = {e["semantic_key"]: e for e in prev_catalog.values()} # next_b per loc — prev_catalog 의 max_b + 1 next_b_per_loc: Dict[str, int] = {} catalog: Dict[str, Dict[str, Any]] = {} for intent in new_intents: loc_id = intent["loc_id"] profile = location_profiles.get(loc_id) if profile is None: raise SemanticKeyError( f"location {loc_id!r} has no space_profile in entity catalog " f"(metadata_json.location.space_profile)" ) space_key = normalize_space_key(loc_id, intent.get("space_key_hint", ""), profile) time_phase = intent["time_phase"] state_class = intent["state_class"] sk = compute_semantic_key(loc_id, space_key, time_phase, state_class) existing = prev_by_sk.get(sk) if existing: # 재사용 + metadata 갱신 bg_id = existing["bg_id"] catalog[bg_id] = { **existing, **_metadata_from(intent), "depends_on_fp": list(intent.get("depends_on_fp") or []), "depends_on_bg": list(intent.get("depends_on_bg") or []), } else: # 새 ID 부여 if loc_id not in next_b_per_loc: next_b_per_loc[loc_id] = _max_b_for_loc(prev_catalog, loc_id) + 1 bg_id = format_bg_id(parse_loc_num(loc_id), next_b_per_loc[loc_id]) next_b_per_loc[loc_id] += 1 if not BG_ID_RE.match(bg_id): raise SemanticKeyError(f"assigned bg_id {bg_id!r} fails BG_ID_RE") catalog[bg_id] = { "bg_id": bg_id, "loc_id": loc_id, "space_key": space_key, "time_phase": time_phase, "state_class": state_class, "semantic_key": sk, "depends_on_bg": list(intent.get("depends_on_bg") or []), "depends_on_fp": list(intent.get("depends_on_fp") or []), **_metadata_from(intent), } return catalog def build_shot_background_map(catalog: Dict[str, Dict[str, Any]]) -> Dict[str, str]: """catalog 의 applies_to_shots 를 shot_id → bg_id 로 펼침. N:1.""" m: Dict[str, str] = {} for bg_id, entry in catalog.items(): for sid in entry.get("applies_to_shots") or []: if sid in m: # 중복 — first-wins (master_plan 의 N:1 강제) continue m[sid] = bg_id return m # ────────────────────────────────────────────────────────────────────── # hash 분리 (§4.6) # ────────────────────────────────────────────────────────────────────── # bg_catalog_hash 가 보호하는 field — render-relevant only. _CATALOG_HASH_FIELDS = ( "bg_id", "loc_id", "space_key", "time_phase", "state_class", "semantic_key", "depends_on_bg", "depends_on_fp", ) def _project_for_catalog_hash(catalog: Dict[str, Dict[str, Any]]) -> Dict[str, Dict[str, Any]]: """metadata-only field 제외 projection.""" return { bg_id: {k: entry.get(k) for k in _CATALOG_HASH_FIELDS} for bg_id, entry in catalog.items() } def compute_bg_catalog_hash(catalog: Dict[str, Dict[str, Any]]) -> str: """catalog 의 render-relevant field projection 의 sorted JSON SHA-256.""" projection = _project_for_catalog_hash(catalog) payload = json.dumps(projection, sort_keys=True, ensure_ascii=False) return hashlib.sha256(payload.encode("utf-8")).hexdigest() def compute_shot_binding_hash(shot_background_map: Dict[str, str]) -> str: """shot_background_map 전용 hash.""" payload = json.dumps(shot_background_map, sort_keys=True, ensure_ascii=False) return hashlib.sha256(payload.encode("utf-8")).hexdigest() ``` - [ ] **Step 4: Wire into background_master_plan.py — 6 단계 ordering** `backend/app/modules/pipeline/background_master_plan.py:22,118-228`: ```python # Top of file — keep _SAFE_ID_RE for raw intent; rename for clarity. from app.core.bg_state_vocab import LLM_INTENT_ID_RE as _LLM_INTENT_ID_RE from app.core.bg_catalog import ( assign_bg_ids, build_shot_background_map, compute_bg_catalog_hash, compute_shot_binding_hash, ) def validate_master_plan_raw_intent(intents: List[Dict[str, Any]]) -> None: """6 단계 ordering — 4a. LLM raw intent 검증. LLM-produced field 만 검사 — fp_id, sub_location_label_canon, state_label_raw_canon 등. bg_id 는 검사 대상 아님 (이 시점엔 아직 없음). """ for intent in intents: for fp_id in intent.get("depends_on_fp") or []: if not _LLM_INTENT_ID_RE.match(fp_id): raise ValueError(f"raw fp_id {fp_id!r} fails _LLM_INTENT_ID_RE") # state_class 는 bg_state_vocab.validate_state_class 가 별도 호출. # sub_location_label 은 한국어 OK 라 정규식 검사 안 함 (metadata only). def validate_master_plan_assigned(catalog: Dict[str, Dict[str, Any]]) -> None: """4e. code-assigned bg_id 검사. _BG_ID_RE 는 assign_bg_ids 안에서 이미 체크.""" seen = set() for bg_id, entry in catalog.items(): if bg_id in seen: raise ValueError(f"duplicate bg_id {bg_id}") seen.add(bg_id) # semantic_key uniqueness sk = entry["semantic_key"] # (이미 assign_bg_ids 가 reuse 보장 — 여기는 invariant 검증) def run_master_plan_post_processing( raw_intents: List[Dict[str, Any]], prev_catalog: Optional[Dict[str, Dict[str, Any]]], location_profiles: Dict[str, Dict[str, Any]], ) -> Dict[str, Any]: """master_plan 의 6 단계 ordering — §4.5. 1. raw intent validation (LLM_INTENT_ID_RE) 2. normalize_space_key (entity profile 보고) 3. compute_semantic_key 4. assign_bg_ids 5. validate_master_plan_assigned (BG_ID_RE) 6. compute_bg_catalog_hash + compute_shot_binding_hash """ validate_master_plan_raw_intent(raw_intents) catalog = assign_bg_ids(prev_catalog or {}, raw_intents, location_profiles) validate_master_plan_assigned(catalog) shot_map = build_shot_background_map(catalog) return { "background_catalog": catalog, "shot_background_map": shot_map, "bg_catalog_hash": compute_bg_catalog_hash(catalog), "shot_binding_hash": compute_shot_binding_hash(shot_map), "bg_catalog_version": 1, } ``` - [ ] **Step 5: Run tests to verify they pass** ``` cd backend && .venv/bin/python -m pytest tests/core/test_bg_catalog.py -xvs ``` Expected: PASS (모든 12 tests) - [ ] **Step 6: Commit** ``` git add backend/app/core/bg_catalog.py backend/app/modules/pipeline/background_master_plan.py backend/tests/core/test_bg_catalog.py git commit -m "feat(d6): T4 bg_catalog module + master_plan post-processing 6 단계" ``` --- ## T5: consumer step manifest stamp (R2 B1 정정 + I6 split) **R2 B1 정정 — scene_detail 위치**: R1 의 `render_prompt_card.py` 는 helper, **NOT a step**. 실제 SceneDetailStep 은 `backend/app/core/steps/detail_steps.py:104, 798, 819`. T5 step 6 도 정정. **R2 I6 split — 4 sub-task RED/GREEN per consumer**: R1 의 단일 step 1 통합 test 는 fixture pre-build 모순. T5 를 4 sub-task 로 split (per consumer step). 각 sub-task 가 자기 step manifest 만 mock + assert. integration test (모두 함께 검증) 는 T7 으로 이동. 각 consumer step 의 manifest 에 `consumed_*_hash` stamp. consumer table (§4.6) 기준: | step | consumed_bg_catalog_hash | consumed_shot_binding_hash | |---|---|---| | `floor_plan_prompt` | ✓ | ✗ | | `background_prompt` | ✓ | ✓ | | `background_render` | ✓ | ✗ | | `scene_detail` (= **detail_steps.py**, NOT render_prompt_card.py — R2 B1) | ✓ | ✓ | **Files** (R2 정정): - Modify: `backend/app/core/steps/floor_plan_prompt_step.py:34-92` - Modify: `backend/app/core/steps/background_prompt_step.py` - Modify: `backend/app/core/steps/background_render_step.py:60-700` (T1 의 BG_ID_RE 교체 정합 + stamp 추가) - Modify: **`backend/app/core/steps/detail_steps.py:104, 798, 819`** (B1 정정 — `render_prompt_card.py` 아님. line 104 = `SCENE_DETAIL_SCHEMA_VERSION`, line 819 = `_execute` return dict) - Modify: `backend/app/core/step_manifest.py:55-90` (5 step schema_version bump) - Test: 4 unit test (per consumer) + integration test 는 T7 - [ ] **Step 1: Write the failing integration test** > **T5-fix iter7 정정 (2026-05-09)**: 아래 초안은 floor_plan_prompt / background_render 가 binding hash 안 찍는다 가정했으나, T5-fix B1+B2 (review iter7) 에서 4 consumer 모두 양쪽 hash stamp 로 정정됨. 이유: floor_plan_prompt 가 applied_shots 를 user prompt 에 inject (`floor_plan_prompt.py:54`), background_render 가 background_prompt 의 applies_to_shots / shot_guides 와 ImageAsset shot_ids 에 transitive 의존. preflight (T8) 가 어느 한쪽 변화도 catch 하려면 모든 consumer 양쪽 stamp 의무. 실제 ship 된 test 는 `tests/core/test_d6_consumer_hash_stamps.py` (plural `stamps` + `tests/core/`). 아래 pseudocode 는 history 보존용 — 최종 assertion 은 ship 된 파일 참조. ```python # backend/tests/integration/test_d6_consumer_hash_stamp.py (drafted; 실제 ship 된 파일은 tests/core/test_d6_consumer_hash_stamps.py) """consumer table — 각 step 이 자기 소비 hash 만 stamp 검증. | step | consumed_bg_catalog_hash | consumed_shot_binding_hash | | floor_plan_prompt | ✓ | ✓ (T5-fix iter7) | | background_prompt | ✓ | ✓ | | background_render | ✓ | ✓ (T5-fix iter7) | | scene_detail | ✓ | ✓ | """ import json def _read_manifest(projects_dir, pid, eid, step): from pathlib import Path p = Path(projects_dir) / pid / "checkpoints" / "episodes" / eid / step / "manifest.json" return json.loads(p.read_text(encoding="utf-8")) def test_floor_plan_prompt_stamps_only_catalog_hash(d6_episode): m = _read_manifest(d6_episode.projects_dir, d6_episode.pid, d6_episode.eid, "floor_plan_prompt") assert "consumed_bg_catalog_hash" in m["data"] assert "consumed_shot_binding_hash" not in m["data"] def test_background_prompt_stamps_both(d6_episode): m = _read_manifest(d6_episode.projects_dir, d6_episode.pid, d6_episode.eid, "background_prompt") assert "consumed_bg_catalog_hash" in m["data"] assert "consumed_shot_binding_hash" in m["data"] def test_background_render_stamps_only_catalog_hash(d6_episode): m = _read_manifest(d6_episode.projects_dir, d6_episode.pid, d6_episode.eid, "background_render") assert "consumed_bg_catalog_hash" in m["data"] assert "consumed_shot_binding_hash" not in m["data"] def test_scene_detail_stamps_both(d6_episode): m = _read_manifest(d6_episode.projects_dir, d6_episode.pid, d6_episode.eid, "scene_detail") assert "consumed_bg_catalog_hash" in m["data"] assert "consumed_shot_binding_hash" in m["data"] def test_consumed_hash_matches_master_plan_latest(d6_episode): """모든 consumer 의 stamp 가 master_plan 의 hash 와 일치.""" mp = _read_manifest(d6_episode.projects_dir, d6_episode.pid, d6_episode.eid, "background_master_plan") expected_catalog = mp["data"]["bg_catalog_hash"] expected_binding = mp["data"]["shot_binding_hash"] for step in ("floor_plan_prompt", "background_prompt", "background_render", "scene_detail"): m = _read_manifest(d6_episode.projects_dir, d6_episode.pid, d6_episode.eid, step) assert m["data"]["consumed_bg_catalog_hash"] == expected_catalog, f"{step} catalog drift" # T5-fix iter7: 4 consumer 모두 binding hash 양쪽 stamp (이전 초안의 이분법 폐기). assert m["data"]["consumed_shot_binding_hash"] == expected_binding, f"{step} binding drift" ``` `d6_episode` fixture 는 본 plan 어디에도 정의 X — T6 이후 unit-style pivot 으로 `d6_episode` 의존 자체가 제거됨 (T6 section 의 unit-style pivot note 참조). T5 의 ship 된 test 는 unittest.mock 으로 manifest cp 를 직접 합성 (`tests/core/test_d6_consumer_hash_stamps.py`). - [ ] **Step 2: Run tests to verify they fail** ``` cd backend && .venv/bin/python -m pytest tests/integration/test_d6_consumer_hash_stamp.py -xvs ``` Expected: FAIL — manifest 의 `consumed_*_hash` 키 없음. - [ ] **Step 3: Modify floor_plan_prompt_step.py** step 의 `_execute` 또는 `_build_manifest_data` 끝부분에 추가: ```python # Read upstream master_plan latest plans_cp = self._load_prev_checkpoint("background_master_plan") bg_catalog_hash = (plans_cp or {}).get("data", {}).get("bg_catalog_hash", "") return { "applicable_count": ..., "data": { ..., "consumed_bg_catalog_hash": bg_catalog_hash, }, ... } ``` - [ ] **Step 4: Modify background_prompt_step.py — both hashes** ```python plans_cp = self._load_prev_checkpoint("background_master_plan") bg_catalog_hash = (plans_cp or {}).get("data", {}).get("bg_catalog_hash", "") shot_binding_hash = (plans_cp or {}).get("data", {}).get("shot_binding_hash", "") return { "data": { ..., "consumed_bg_catalog_hash": bg_catalog_hash, "consumed_shot_binding_hash": shot_binding_hash, }, } ``` - [ ] **Step 5: Modify background_render_step.py** ```python plans_cp = self._load_prev_checkpoint("background_master_plan") bg_catalog_hash = (plans_cp or {}).get("data", {}).get("bg_catalog_hash", "") # 기존 _execute 의 return dict 에 추가: return { ..., "data": { ..., "consumed_bg_catalog_hash": bg_catalog_hash, }, } ``` `SCHEMA_VERSION` 도 bump (consumed_bg_catalog_hash 추가 = manifest 형식 변경). - [ ] **Step 6: Modify scene_detail (render_prompt_card.py) — both hashes** scene_detail 의 manifest data 끝에: ```python plans_cp = self._load_prev_checkpoint("background_master_plan") bg_catalog_hash = (plans_cp or {}).get("data", {}).get("bg_catalog_hash", "") shot_binding_hash = (plans_cp or {}).get("data", {}).get("shot_binding_hash", "") # manifest data return 에 추가: data["consumed_bg_catalog_hash"] = bg_catalog_hash data["consumed_shot_binding_hash"] = shot_binding_hash ``` scene_detail SCHEMA_VERSION bump (background_binding.bg_id 형식 변경 + consumed_*_hash). - [ ] **Step 7: Update step_manifest.py — schema_version 정합** ```python # backend/app/core/step_manifest.py "floor_plan_prompt": { ..., "schema_version": }, "background_prompt": { ..., "schema_version": }, "background_render": { ..., "schema_version": }, "scene_detail": { ..., "schema_version": }, "background_master_plan": { ..., "schema_version": }, ``` - [ ] **Step 8: Run integration tests to verify they pass** ``` cd backend && .venv/bin/python -m pytest tests/integration/test_d6_consumer_hash_stamp.py -xvs ``` Expected: PASS - [ ] **Step 9: Commit** ``` git add backend/app/core/steps/floor_plan_prompt_step.py backend/app/core/steps/background_prompt_step.py backend/app/core/steps/background_render_step.py backend/app/core/steps/render_prompt_card.py backend/app/core/step_manifest.py backend/tests/integration/test_d6_consumer_hash_stamp.py git commit -m "feat(d6): T5 consumer step hash stamp + schema_version bump" ``` --- ## T6: scene_detail chain unchanged 검증 (unit-style) scene_detail 은 prompt 변경 0 (R3 N1). T5 에서 schema_version + stamp 처리됨. 본 task 는 `ctx.chain_bg_id_by_shot[(si, shi)]` → `_derive_card_inputs_from_ctx` → `build_render_prompt_card` → `render_prompt_card.background_binding.bg_id` chain 이 D6 의 새 `L##B##` 형식 그대로 통과하는지 검증. **unit-style pivot (vs R2 R3 의 integration-style 초안)**: - T6 의 본질은 "scene_detail 코드 변경 0 검증" — LLM/E2E 호출 의미 없음. chain 은 `detail_steps.py:432` 의 `bg_id = (ctx.chain_bg_id_by_shot or {}).get((si, shi))` 한 줄 + `build_background_binding(bg_id=...)` 의 `"bg_id": bg_id` 한 줄로 구성. 둘 다 pure pass-through. - 초안의 `d6_episode` fixture 는 plan 어디에도 정의 없음 (line 1403 mention 만). 진짜 fixture 구축은 T6 명세 대비 비용 과다 — T13 wrong-order force smoke 에서 별도로 다룸. - T7 의 5-회 재실행도 같은 unit-style pivot 적용됨 (T7 본문 참조 — 초안의 `run_master_plan_post_processing` 함수 / `d6_episode_factory` fixture 둘 다 미존재로 확인됨, `BackgroundMasterPlanStep._d6_post_process` 직접 호출로 대체). **Files:** - Test: `backend/tests/core/test_d6_scene_detail_chain_unchanged.py` (NEW — `tests/core/` 하위로 T5 와 일치) - [ ] **Step 1: Write the unit test** ```python # backend/tests/core/test_d6_scene_detail_chain_unchanged.py """D6 R3 N1 — scene_detail chain 변경 0 검증 (unit-style). ctx.chain_bg_id_by_shot[(si, shi)] (D6 의 L##B## 형식) 가 → _derive_card_inputs_from_ctx (`detail_steps.py:432`) → build_render_prompt_card (`render_prompt_card.py`) → render_prompt_card.background_binding.bg_id chain 으로 pass-through 되는지 unit-level 검증. scene_detail step 자체에는 D6 코드 변경 0 (T5 의 manifest stamp + schema_version bump 만). bg_id 형식 변경 (D4 의 bg_* → D6 의 L##B##) 은 chain 의 producer (master_plan / chain_bg loader) 변경에서 자동 cascading. """ from __future__ import annotations from types import SimpleNamespace def _make_fake_ctx(bg_id: str = "L09B01"): """detail_steps._derive_card_inputs_from_ctx 가 읽는 모든 ctx attribute 의 minimal 합법 fake. (si=1, shi=1) 한 shot 만 채움.""" return SimpleNamespace( shot_director_ve={(1, 1): ["C01"]}, scene_visible={1: ["C01"]}, outlook_data=None, staging_map={"1_1": {"camera_direction": "medium shot", "lighting_mood": "warm"}}, chain_bg_id_by_shot={(1, 1): bg_id}, chain_bg_owned_by_shot={(1, 1): []}, chain_bg_camera_meta_by_shot={}, chain_bg_guide_by_shot={}, fixed_elements_by_scene={}, dependencies=[], shot_scenes_map={}, ) def test_chain_bg_id_lnnbnn_passes_through_to_background_binding(monkeypatch): """ctx.chain_bg_id_by_shot[(1,1)]='L09B01' → RPC.background_binding.bg_id='L09B01'.""" from app.core.steps.detail_steps import _derive_card_inputs_from_ctx from app.core.steps.render_prompt_card import ( build_render_prompt_card, BG_MODE_REF_ATTACHED, ) monkeypatch.setattr("app.core.config.settings.background_mode", "on") ctx = _make_fake_ctx(bg_id="L09B01") seg = {"index": 1, "scene_index": 1, "text": "scene segment text"} shot_info = {"shot_index": 1, "camera_direction": "medium shot", "primary_subject": "the subject"} inputs = _derive_card_inputs_from_ctx(ctx=ctx, seg=seg, shot_info=shot_info) assert inputs["bg_id"] == "L09B01", "ctx → inputs pass-through 깨짐" inputs.pop("ctx", None) # builder 시그니처 외 — splat 전 strip card = build_render_prompt_card(**inputs) bb = card["background_binding"] assert bb["bg_id"] == "L09B01", f"binding bg_id drift: {bb!r}" assert bb["mode"] == BG_MODE_REF_ATTACHED, ( f"bg_on + bg_id present + non-close framing → ref_attached, got {bb['mode']!r}" ) def test_render_prompt_card_does_not_emit_deprecated_background_binding_intent( monkeypatch, ): """R2 의 폐기된 schema field background_binding_intent 가 RPC 에 없음.""" from app.core.steps.detail_steps import _derive_card_inputs_from_ctx from app.core.steps.render_prompt_card import build_render_prompt_card monkeypatch.setattr("app.core.config.settings.background_mode", "on") ctx = _make_fake_ctx(bg_id="L09B01") seg = {"index": 1, "scene_index": 1, "text": "scene segment text"} shot_info = {"shot_index": 1, "camera_direction": "medium shot", "primary_subject": "the subject"} inputs = _derive_card_inputs_from_ctx(ctx=ctx, seg=seg, shot_info=shot_info) inputs.pop("ctx", None) card = build_render_prompt_card(**inputs) assert "background_binding_intent" not in card, ( "R2 폐기 schema 가 부활됨 (R3 N1 위반)" ) ``` - [ ] **Step 2: Run + commit** ``` cd backend && .venv/bin/python -m pytest tests/core/test_d6_scene_detail_chain_unchanged.py -xvs ``` Expected: PASS (T5 의 변경 + T4 의 master_plan 후처리가 cascading 으로 충족 — 단, 본 unit test 는 producer 의존 없음). ``` git add backend/tests/core/test_d6_scene_detail_chain_unchanged.py docs/superpowers/plans/2026-05-09-d6-deterministic-bg-id-and-catalog-lineage-implementation.md git commit -m "test(d6): T6 scene_detail chain unchanged (R3 N1) — bg_id pass-through + plan unit-style pivot" ``` --- ## T7: master_plan determinism + hash isolation (unit-style) **unit-style pivot (vs R1 R2 의 integration-style 초안)**: - 초안의 `run_master_plan_post_processing(intents, prev, profiles)` helper 는 코드에 존재 X (T4 구현은 `BackgroundMasterPlanStep._d6_post_process(ordered_plans)` instance method 로 결합). `d6_episode_factory` / `d6_episode` fixture 도 plan 어디에도 정의 없음. - AC-1 의 본질 (결정론 + hash 축 분리) 은 atom level (`assign_bg_ids` / `compute_bg_catalog_hash` / `compute_shot_binding_hash`) 에서 이미 T4 가 광범위 cover (`tests/core/test_bg_catalog.py:194,246,310,335,350,373,393`). T7 은 atom 위에 있는 step boundary (`_d6_post_process` 의 4-tuple 결합 + plan.backgrounds[].bg_id handshake) 회귀 차단으로 가야 가치가 추가됨. - 진짜 episode fixture 는 T13 wrong-order force smoke 에서 별도 다룸 (T6 와 같은 결정). **Files:** - Test: `backend/tests/core/test_d6_master_plan_determinism.py`, `backend/tests/core/test_d6_hash_isolation.py` - [ ] **Step 1: Determinism test (5 회 + I8 shuffle + handshake boundary)** ```python # backend/tests/core/test_d6_master_plan_determinism.py """D6 T7 — _d6_post_process 결정론 + handshake boundary. T4 의 atom-level 결정론 위에 step composition (`_d6_post_process` 의 4-tuple 결합 + plan.backgrounds[].bg_id handshake) 회귀 차단. """ from __future__ import annotations from typing import Any, Dict, List _PROFILES = {"L09": {"kind": "single_space", "allowed_space_keys": ["main"]}} def _make_step(monkeypatch, prev_catalog=None): """BackgroundMasterPlanStep instance + DB/cp 의존 monkeypatch.""" from app.core.steps.background_master_plan_step import BackgroundMasterPlanStep step = BackgroundMasterPlanStep.__new__(BackgroundMasterPlanStep) step.project_id = "p-d6-t7" step.episode_id = "ep1" step.step_id = "background_master_plan" step.project_config = {} monkeypatch.setattr(step, "_load_location_profiles", lambda: dict(_PROFILES)) monkeypatch.setattr(step, "_load_prev_background_catalog", lambda: dict(prev_catalog or {})) return step def _make_ordered_plans(intents: List[Dict[str, Any]]) -> Dict[str, Dict[str, Any]]: """단일 group 'g1' 에 raw intents 를 backgrounds 로 묶음.""" return {"g1": {"status": "ok", "plan": { "floor_plans": [], "backgrounds": list(intents), }}} def _intent(**kw): base = { "loc_id": "L09", "space_key_hint": "main", "time_phase": "dusk", "state_class": "busy_exit", "applies_to_shots": ["S8_Shot4"], "sub_location_label": "store sales floor", "state_label_raw": "raw", "depends_on_fp": ["fp_supermarket"], "depends_on_bg": [], } base.update(kw) return base def test_d6_post_process_5_reruns_stable_hashes(monkeypatch): """AC-1 — 같은 ordered_plans 5 회 → catalog/shot_bg_map/2 hash 모두 안정.""" intents = [_intent(state_class="busy_exit"), _intent(state_class="ransacked", applies_to_shots=["S15_Shot2"])] plans_first = _make_ordered_plans(intents) step = _make_step(monkeypatch) cat_1, map_1, ch_1, bh_1 = step._d6_post_process(plans_first) # prev_catalog 를 첫 run 의 catalog 로 고정. shot grouping/순서 그대로. step_n = _make_step(monkeypatch, prev_catalog=cat_1) for i in range(4): plans_n = _make_ordered_plans(intents) cat_n, map_n, ch_n, bh_n = step_n._d6_post_process(plans_n) assert ch_n == ch_1, f"run {i+2}: catalog hash drift" assert bh_n == bh_1, f"run {i+2}: binding hash drift" assert set(cat_n.keys()) == set(cat_1.keys()) assert map_n == map_1 def test_d6_post_process_intent_order_shuffle_stable_hashes(monkeypatch): """I8 — per-group backgrounds[] 순서가 달라도 같은 set 이면 catalog/binding 안정.""" intent_a = _intent(loc_id="L09", time_phase="dusk", state_class="busy_exit", applies_to_shots=["S8_Shot4"]) intent_b = _intent(loc_id="L09", time_phase="dusk", state_class="ransacked", applies_to_shots=["S15_Shot2"]) plans_ab = _make_ordered_plans([intent_a, intent_b]) plans_ba = _make_ordered_plans([intent_b, intent_a]) step1 = _make_step(monkeypatch) cat_ab, map_ab, ch_ab, bh_ab = step1._d6_post_process(plans_ab) step2 = _make_step(monkeypatch) cat_ba, map_ba, ch_ba, bh_ba = step2._d6_post_process(plans_ba) assert ch_ab == ch_ba, "I8 sort drift — catalog hash" assert bh_ab == bh_ba, "I8 sort drift — binding hash" assert set(cat_ab.keys()) == set(cat_ba.keys()) sem_ab = {v["semantic_key"]: bid for bid, v in cat_ab.items()} sem_ba = {v["semantic_key"]: bid for bid, v in cat_ba.items()} assert sem_ab == sem_ba def test_d6_post_process_handshake_updates_plan_backgrounds_bg_id(monkeypatch): """boundary — _d6_post_process 가 plan.backgrounds[].bg_id slot 을 L##B## 로 갱신. catalog helper 는 맞는데 step handshake 가 깨지면 downstream reader (background_prompt / background_render) 가 옛 bg_* 또는 빈 bg_id 받아 cascade. 본 test 가 그 drift 차단. """ import re intents = [_intent(state_class="busy_exit"), _intent(state_class="ransacked", applies_to_shots=["S15_Shot2"])] plans = _make_ordered_plans(intents) step = _make_step(monkeypatch) catalog, _map, _ch, _bh = step._d6_post_process(plans) bg_id_re = re.compile(r"^L\d{2,3}B\d{2,3}$") bg_ids_in_plan = [ b.get("bg_id") for b in plans["g1"]["plan"]["backgrounds"] ] assert all(bg_id_re.match(bid or "") for bid in bg_ids_in_plan), ( f"handshake 깨짐 — plan.backgrounds[].bg_id 가 L##B## 형식 아님: {bg_ids_in_plan}" ) # plan.backgrounds[].bg_id 가 catalog 의 bg_id 와 정합 (semantic_key 기반). catalog_ids = set(catalog.keys()) assert set(bg_ids_in_plan) == catalog_ids, ( f"handshake drift — plan ids {bg_ids_in_plan} vs catalog {sorted(catalog_ids)}" ) ``` - [ ] **Step 2: Hash isolation test (cross-axis 독립)** ```python # backend/tests/core/test_d6_hash_isolation.py """D6 T7 — bg_catalog_hash vs shot_binding_hash 두 축 독립. 축 1 (catalog): render-relevant field (loc_id/space_key/time_phase/state_class/ semantic_key/depends_on_fp/depends_on_bg) 변경 → catalog_hash drift. 축 2 (binding): applies_to_shots 변경만 → binding_hash drift, catalog_hash 안정. """ from __future__ import annotations import pytest _PROFILES = {"L09": {"kind": "single_space", "allowed_space_keys": ["main"]}} def _make_step(monkeypatch, prev_catalog=None): from app.core.steps.background_master_plan_step import BackgroundMasterPlanStep step = BackgroundMasterPlanStep.__new__(BackgroundMasterPlanStep) step.project_id = "p-d6-t7-iso" step.episode_id = "ep1" step.step_id = "background_master_plan" step.project_config = {} monkeypatch.setattr(step, "_load_location_profiles", lambda: dict(_PROFILES)) monkeypatch.setattr(step, "_load_prev_background_catalog", lambda: dict(prev_catalog or {})) return step def _plans(intents): return {"g1": {"status": "ok", "plan": { "floor_plans": [], "backgrounds": list(intents), }}} def _intent(**kw): base = { "loc_id": "L09", "space_key_hint": "main", "time_phase": "dusk", "state_class": "busy_exit", "applies_to_shots": ["S8_Shot4"], "sub_location_label": "store sales floor", "state_label_raw": "raw", "depends_on_fp": ["fp_supermarket"], "depends_on_bg": [], } base.update(kw) return base def test_applies_to_shots_change_only_drifts_binding_hash(monkeypatch): """축 2 — applies_to_shots 변경만 → binding_hash drift, catalog_hash 안정.""" a = _intent(applies_to_shots=["S8_Shot4"]) b = _intent(applies_to_shots=["S8_Shot4", "S8_Shot5"]) step_a = _make_step(monkeypatch) _, _, ch_a, bh_a = step_a._d6_post_process(_plans([a])) step_b = _make_step(monkeypatch, prev_catalog=None) # prev 무관 — 같은 sem_key 라 같은 bg_id. _, _, ch_b, bh_b = step_b._d6_post_process(_plans([b])) assert ch_a == ch_b, f"catalog hash 가 applies_to_shots 변경에 흔들림: {ch_a} vs {ch_b}" assert bh_a != bh_b, "binding hash 가 shot 추가에 안 변함 — 축 분리 깨짐" def test_state_class_change_drifts_catalog_hash(monkeypatch): """축 1 — state_class (render-relevant) 변경 시 catalog_hash drift.""" a = _intent(state_class="busy_exit") b = _intent(state_class="ransacked") step_a = _make_step(monkeypatch) _, _, ch_a, _ = step_a._d6_post_process(_plans([a])) step_b = _make_step(monkeypatch) _, _, ch_b, _ = step_b._d6_post_process(_plans([b])) assert ch_a != ch_b, ( "state_class 변경이 catalog hash 에 안 잡힘 — render-relevant field " "isolation 깨짐 (T8 preflight 가 stale_upstream 못 잡게 됨)." ) def test_depends_on_fp_change_fail_fasts_on_same_sem_key(monkeypatch): """축 1 — 같은 sem_key (=같은 catalog row) 에서 depends_on_fp 가 다르면 fail-fast. T4-fix B3 의 silent first-wins 차단: 같은 sem_key 의 두 intent 가 다른 deps 를 갖는 건 SOT 결손 → SemanticKeyError. catalog hash 가 drift 든 안 하든 fail-fast 가 우선 — 무결성 SOT. """ from app.core.bg_catalog import SemanticKeyError a = _intent(depends_on_fp=["fp_supermarket"]) b = _intent(depends_on_fp=["fp_warehouse"]) # 같은 sem_key, 다른 deps step = _make_step(monkeypatch) with pytest.raises(SemanticKeyError): step._d6_post_process(_plans([a, b])) ``` - [ ] **Step 3: Run + commit** ``` cd backend && .venv/bin/python -m pytest tests/core/test_d6_master_plan_determinism.py tests/core/test_d6_hash_isolation.py -xvs ``` Expected: PASS ``` git add backend/tests/core/test_d6_master_plan_determinism.py backend/tests/core/test_d6_hash_isolation.py docs/superpowers/plans/2026-05-09-d6-deterministic-bg-id-and-catalog-lineage-implementation.md git commit -m "test(d6): T7 master_plan determinism + hash isolation (AC-1) — unit-style pivot" ``` --- ## T8: dispatcher preflight at endpoint entry (R2 I1 + I2 patch) **2-Phase split (production code 진입 — review checkpoint)**: - **Phase A**: `StaleUpstreamError` + `dispatcher_preflight.py` + tmp-manifest unit tests (no endpoint touch). 1차 review. - **Phase B**: `images.py` single + batch endpoint wiring + TestClient/mock smoke. 2차 review. **unit-style pivot — fixture stale 정정**: - 초안의 `d6_episode` / `d6_episode_with_stale_render` / `d6_episode_with_stale_binding` 셋 다 plan 어디에도 정의 없음. T6/T7 와 같은 패턴 채택 — `tmp_path` 에 minimal manifest (`background_master_plan` + 4 consumer) 직접 기록 + `monkeypatch settings.projects_dir`. preflight 가 `_read_manifest` 로 읽으니 그대로 hit. - endpoint integration 은 FastAPI `TestClient` + dependency override (DB query mock) — 진짜 SceneStill row + Episode 까지의 fixture 는 T13 wrong-order force smoke 에서 별도. **T5-fix iter7 결론 정합 — 4-consumer binding stamp**: - 초안의 `_BINDING_CONSUMERS = ("background_prompt", "scene_detail")` 는 stale (T5-fix B1+B2 결론과 모순). 모든 4 consumer (`floor_plan_prompt` / `background_prompt` / `background_render` / `scene_detail`) 가 양쪽 hash stamp 하므로 preflight 도 4 consumer 전체 binding 비교 의무. **fail-fast 강화 (silent skip 차단)**: - `_read_manifest` parse 실패 → `return None` silent skip 폐기. `StaleUpstreamError` raise (운영 신호 손실 차단 — corrupt manifest = catalog freshness 판단 불가). - consumer manifest 가 missing 인데 `master_plan` 에 D6 hash 가 있으면 → `StaleUpstreamError` raise (consumer 가 D6 mode 에서 안 돌았다는 뜻 — silent pass 면 stale chain 으로 dispatch). - `master_plan` 자체 missing 또는 `bg_catalog_hash` 부재 → return (D6 미적용 episode — D5 fallback path). **R2 I1 ordering**: preflight 가 router 의 **first signal**. 순서: 1. SceneStill row fetch (existing — episode_id 도출 의무). 2. **preflight (`check_bg_catalog_freshness`)** — STALE_UPSTREAM 422 if mismatch. 3. `check_scene_images_ready` (router gate, `images.py:401/644`) — pipeline_gate.NOT_READY if stale lifecycle. 4. service entry (`generate_single_scene_image`). 5. service 안 `assert_episode_asset_readiness` (`scene_image_service.py:620`). → STALE_UPSTREAM 이 first. 운영자가 catalog drift 를 다른 gate 에 가려지지 않고 명확히 받음. **R2 I2 batch endpoint**: - `api/v1/images.py:598-661` `generate_images` — `submit_background_job` (line 652) 이전, sync 부분에 preflight wiring (`check_scene_images_ready` 호출 직전). - 배치는 `enforce_shot_binding=True` 강제 (custom_prompt 의미 없음). **error response shape — `AppError` extension (subclass)**: - 현재 `AppError(code, message, status_code)` 만 받음 (`backend/app/core/errors.py:5`). plan 초안의 `AppError(details=...)` 비호환. - 해결: `AppError.__init__` 에 optional kw-only `details: Optional[Dict] = None` 추가 + `app_error_handler` 가 `details` 를 응답 `error` dict 에 merge. `StaleUpstreamError(AppError)` 가 details 를 채워 super 에 전달. 기존 AppError 호출자는 변경 0 (default `details=None` 으로 backward compatible). **Files:** - Modify: `backend/app/core/errors.py` (AppError details + StaleUpstreamError + handler merge — 단일 파일, 별도 모듈 X) - Create: `backend/app/services/dispatcher_preflight.py` - Modify: `backend/app/api/v1/images.py:387-406, 598-650` (preflight wiring 두 endpoint, Phase B) - Test: `backend/tests/services/test_dispatcher_preflight.py` (Phase A, tmp manifest unit-style — 27 case), `backend/tests/api/test_d6_endpoint_preflight.py` (Phase B, TestClient + mock) --- ### Phase A — service module + error class (✅ 구현 완료, commit `e0353fe` + review iter1 fix ``) **구현 정합 (review iter1 후 정합)**: - `app/core/errors.py`: `AppError(*, details=None)` optional kw-only — `_RESERVED_DETAIL_KEYS = ("code", "message")` 거부 (fail-fast). handler 가 details merge 시 base fields 우선 (defense in depth). - `app/core/errors.py`: `StaleUpstreamError(AppError)` — code=`STALE_UPSTREAM`, status=422, upstream/expected_*/observed_*/missing_in_render/remediation 을 details 로 운반. - `app/services/dispatcher_preflight.py`: `_CONSUMERS = (floor_plan_prompt, background_prompt, background_render, scene_detail)` — T5-fix iter7 4-consumer 정합. `check_bg_catalog_freshness(pid, eid, *, enforce_shot_binding)`. parse fail → fail-fast. master_plan D6 hash 있는데 consumer manifest missing → fail-fast (각 consumer 의 empty-input path 도 stamp 한다는 invariant 에 의존 — 모듈 docstring 참조). - D5 fallback 보존: master_plan cp 부재 또는 `bg_catalog_hash` 부재 → silent return. - custom_prompt path: `enforce_shot_binding=False` — catalog freshness 강제, binding-only stale 만 허용. 본 Phase A 의 canonical 코드는 `app/core/errors.py` + `app/services/dispatcher_preflight.py` 본체 — plan 의 옛 pseudocode 참조 금지. **테스트 정합 (`tests/services/test_dispatcher_preflight.py` — 27 unit case)**: - happy path (4 consumer match, enforce True/False). - D5 fallback (master_plan missing / D6 hash 부재). - catalog mismatch × 4 consumer (parametrized). - binding mismatch × 4 consumer (parametrized). - enforce False — binding skip + catalog still strict. - consumer manifest absent × 4 (parametrized). - corrupt consumer / corrupt master_plan. - StaleUpstreamError details 확인 (catalog-only / binding sibling). - handler serialization (StaleUpstreamError + plain AppError backward compat). - IMPORTANT 1 (review iter1): `details` 의 reserved key reject + handler base fields win. - BLOCKING (review iter1) 검증: D6 mode + empty background_catalog → 4 consumer empty-path manifest + valid stamp → preflight 통과. unit-style pivot 사유: `d6_episode_*` fixture 미존재. `tmp_path` 에 manifest 직접 기록 + `monkeypatch settings.projects_dir` 가 정확한 검증 경로 (T6/T7 와 동일 패턴). --- ### Phase B — endpoint wiring (TBD) **`api/v1/images.py:387` `generate_still_image`**: 1. 기존 `still = db.query(_SS).filter(...).first()` (line 399) 결과로 episode_id 도출 후, `check_scene_images_ready` (line 401) 직전에 `check_bg_catalog_freshness(project_id, still.episode_id, enforce_shot_binding=(body.custom_prompt is None))` 호출. 2. `still` 부재면 (404 결과) preflight skip — episode_id 없음. 3. `StaleUpstreamError` 는 `app_error_handler` 가 이미 catch (subclass 라 자동) — 추가 try/except 불필요. **`api/v1/images.py:598` `generate_images` (batch)**: 1. episode 검증 (line 617-640) 통과 후, `check_scene_images_ready` (line 644) 직전에 `check_bg_catalog_freshness(project_id, episode_id, enforce_shot_binding=True)` 호출. 2. 배치는 binding 강제 (custom_prompt 의미 없음). **TestClient mock smoke (`tests/api/test_d6_endpoint_preflight.py`)**: - FastAPI `TestClient` + dependency override 로 `db.query(SceneStill)` mock + `check_bg_catalog_freshness` 가 raise/통과 양쪽 path → HTTP 422 응답 shape 검증 (`{"error": {"code": "STALE_UPSTREAM", "upstream": ..., "remediation": ...}}`). - 검증 순서: `still 조회 OK → preflight raises → check_scene_images_ready 미호출 / service 미호출` (preflight 가 first signal — review iter1 R2 I1 ordering 정합). - batch endpoint 동일 검증. **Phase B commit**: ``` feat(d6): T8b endpoint preflight wiring (single + batch) + TestClient smoke ``` 진짜 production E2E (Episode + SceneStill row + manifest 4 set) 는 T13 wrong-order force smoke 에서 별도. --- ### Phase A 의 옛 초안 (history 보존) > 아래 Step 1~8 은 R2 R3 시점의 초안 — `d6_episode_*` fixture / `app.core.errors.stale_upstream` 모듈 / `_BINDING_CONSUMERS` 2-tuple / `AppError(details=...)` 직접 호출 등 현재 구현과 안 맞음 (review iter1 IMPORTANT 2 — 작업자 잘못 복사 위험). 다음 작업자는 이 부분 무시하고 위 Phase A/B 명세 + 실제 코드 (`backend/app/core/errors.py` / `backend/app/services/dispatcher_preflight.py` / `backend/tests/services/test_dispatcher_preflight.py`) 를 SOT 로 사용. - [ ] **Step 1: Write the failing test** ```python # backend/tests/services/test_dispatcher_preflight.py """dispatcher preflight — endpoint entry hash freshness 검증.""" import pytest def test_preflight_passes_when_hashes_match(d6_episode): from app.services.dispatcher_preflight import check_bg_catalog_freshness, StaleUpstreamError # 정상 시점 — master_plan 의 hash 와 consumer 의 stamp 일치. check_bg_catalog_freshness(d6_episode.pid, d6_episode.eid, enforce_shot_binding=True) def test_preflight_raises_stale_upstream_on_catalog_mismatch(d6_episode_with_stale_render): from app.services.dispatcher_preflight import check_bg_catalog_freshness from app.core.errors.stale_upstream import StaleUpstreamError with pytest.raises(StaleUpstreamError) as exc: check_bg_catalog_freshness( d6_episode_with_stale_render.pid, d6_episode_with_stale_render.eid, enforce_shot_binding=False, ) err = exc.value assert err.upstream == "background_render" assert err.expected_bg_catalog_hash != err.observed_bg_catalog_hash def test_preflight_custom_prompt_skips_shot_binding_check(d6_episode_with_stale_binding): """custom_prompt 는 shot_binding hash 강제 안 함.""" from app.services.dispatcher_preflight import check_bg_catalog_freshness # binding 만 stale, catalog 는 OK → custom_prompt 통과 check_bg_catalog_freshness( d6_episode_with_stale_binding.pid, d6_episode_with_stale_binding.eid, enforce_shot_binding=False, # custom_prompt ) def test_preflight_includes_remediation_metadata(d6_episode_with_stale_render): from app.services.dispatcher_preflight import check_bg_catalog_freshness from app.core.errors.stale_upstream import StaleUpstreamError with pytest.raises(StaleUpstreamError) as exc: check_bg_catalog_freshness( d6_episode_with_stale_render.pid, d6_episode_with_stale_render.eid, enforce_shot_binding=True, ) payload = exc.value.to_response_dict() assert payload["error"]["code"] == "STALE_UPSTREAM" assert "remediation" in payload["error"] assert "background_render" in payload["error"]["remediation"] ``` - [ ] **Step 2: Run tests to verify they fail** ``` cd backend && .venv/bin/python -m pytest tests/services/test_dispatcher_preflight.py -xvs ``` Expected: FAIL — module 없음. - [ ] **Step 3: Implement `StaleUpstreamError`** ```python # backend/app/core/errors/stale_upstream.py """STALE_UPSTREAM error — D6 §4.8.""" from __future__ import annotations from typing import Any, Dict, List, Optional class StaleUpstreamError(Exception): """preflight detection — upstream cp 가 stale. HTTP 422 응답 형식: { "error": { "code": "STALE_UPSTREAM", "upstream": "...", "expected_bg_catalog_hash": "...", "observed_bg_catalog_hash": "...", "expected_shot_binding_hash": "...", "observed_shot_binding_hash": "...", "missing_in_render": [...], "remediation": "..." } } """ def __init__( self, *, upstream: str, expected_bg_catalog_hash: str, observed_bg_catalog_hash: str, expected_shot_binding_hash: Optional[str] = None, observed_shot_binding_hash: Optional[str] = None, missing_in_render: Optional[List[str]] = None, remediation: str = "", ) -> None: self.upstream = upstream self.expected_bg_catalog_hash = expected_bg_catalog_hash self.observed_bg_catalog_hash = observed_bg_catalog_hash self.expected_shot_binding_hash = expected_shot_binding_hash self.observed_shot_binding_hash = observed_shot_binding_hash self.missing_in_render = missing_in_render or [] self.remediation = remediation super().__init__(f"STALE_UPSTREAM upstream={upstream}") def to_response_dict(self) -> Dict[str, Any]: err: Dict[str, Any] = { "code": "STALE_UPSTREAM", "upstream": self.upstream, "expected_bg_catalog_hash": self.expected_bg_catalog_hash, "observed_bg_catalog_hash": self.observed_bg_catalog_hash, "remediation": self.remediation, } if self.expected_shot_binding_hash is not None: err["expected_shot_binding_hash"] = self.expected_shot_binding_hash err["observed_shot_binding_hash"] = self.observed_shot_binding_hash if self.missing_in_render: err["missing_in_render"] = self.missing_in_render return {"error": err} ``` - [ ] **Step 4: Implement `dispatcher_preflight.py`** ```python # backend/app/services/dispatcher_preflight.py """endpoint entry preflight — bg catalog/binding hash freshness. spec: docs/superpowers/specs/2026-05-09-deterministic-bg-id-and-catalog-lineage.md §4.8 """ from __future__ import annotations import json import logging from pathlib import Path from typing import Dict, List, Optional, Tuple from app.core.config import settings from app.core.errors.stale_upstream import StaleUpstreamError logger = logging.getLogger(__name__) _CATALOG_CONSUMERS = ("floor_plan_prompt", "background_prompt", "background_render", "scene_detail") _BINDING_CONSUMERS = ("background_prompt", "scene_detail") def _read_manifest(pid: str, eid: str, step: str) -> Optional[Dict]: p = (Path(settings.projects_dir) / pid / "checkpoints" / "episodes" / eid / step / "manifest.json") if not p.exists(): return None try: return json.loads(p.read_text(encoding="utf-8")) except Exception as exc: logger.warning("preflight: %s manifest parse failed: %s", step, exc) return None def check_bg_catalog_freshness( pid: str, eid: str, enforce_shot_binding: bool = True, ) -> None: """master_plan 의 hash 와 consumer 들의 stamp 비교. enforce_shot_binding=False — custom_prompt path. catalog hash 만 검증. 상위 caller (api/v1/images.py:387 / :599) 가 try/except StaleUpstreamError → HTTP 422 응답. """ mp = _read_manifest(pid, eid, "background_master_plan") if mp is None: # master_plan 없으면 D6 적용 안 된 episode — preflight skip (D5 fallback). return expected_catalog = mp.get("data", {}).get("bg_catalog_hash") expected_binding = mp.get("data", {}).get("shot_binding_hash") if not expected_catalog: # legacy episode — D6 stamp 없음. skip. return for step in _CATALOG_CONSUMERS: cm = _read_manifest(pid, eid, step) if cm is None: continue # step 미실행 — D6 path 외 (e.g., floor_plan_prompt off) observed = cm.get("data", {}).get("consumed_bg_catalog_hash", "") if observed != expected_catalog: raise StaleUpstreamError( upstream=step, expected_bg_catalog_hash=expected_catalog, observed_bg_catalog_hash=observed, remediation=( f"POST /steps/{step}?mode=force " f"(or full ordered re-run — see docs/runbooks/d6-migration.md)" ), ) if enforce_shot_binding and expected_binding: for step in _BINDING_CONSUMERS: cm = _read_manifest(pid, eid, step) if cm is None: continue observed = cm.get("data", {}).get("consumed_shot_binding_hash", "") if observed != expected_binding: raise StaleUpstreamError( upstream=step, expected_bg_catalog_hash=expected_catalog, observed_bg_catalog_hash=expected_catalog, # catalog 는 OK expected_shot_binding_hash=expected_binding, observed_shot_binding_hash=observed, remediation=( f"POST /steps/{step}?mode=force " f"(shot grouping changed — see docs/runbooks/d6-migration.md)" ), ) ``` - [ ] **Step 5: Run preflight unit tests** ``` cd backend && .venv/bin/python -m pytest tests/services/test_dispatcher_preflight.py -xvs ``` Expected: PASS (4 tests) - [ ] **Step 6: Wire into endpoint entry — `api/v1/images.py:387, 599`** ```python # backend/app/api/v1/images.py # 단건 generate-image (line 387 부근) @router.post("/stills/{still_id}/generate-image", response_model=ImageResponse) def generate_still_image( still_id: str, body: Optional[Dict[str, Any]] = Body(None), project_id: str = Depends(verify_project_access), db: OrmSession = Depends(get_db), current_user: UserAccount = Depends(get_current_user), ): from app.services.dispatcher_preflight import check_bg_catalog_freshness from app.core.errors.stale_upstream import StaleUpstreamError from app.core.errors import AppError # D6 preflight — endpoint entry. custom_prompt 도 통과 의무 (catalog freshness). custom_prompt = (body or {}).get("custom_prompt") still = ... # fetch still for episode_id try: check_bg_catalog_freshness( project_id, still.episode_id, enforce_shot_binding=(custom_prompt is None), ) except StaleUpstreamError as exc: raise AppError( code="STALE_UPSTREAM", message=str(exc), status_code=422, details=exc.to_response_dict()["error"], ) # ...기존 service 호출 ``` 배치 endpoint (line 599) 에도 동일 preflight (모든 episode-level still 에 대해 한 번 검사). - [ ] **Step 7: Integration test — endpoint level** ```python # backend/tests/integration/test_d6_endpoint_preflight.py def test_endpoint_returns_422_on_stale_upstream(client, d6_episode_with_stale_render): res = client.post( f"/api/v1/projects/{d6_episode_with_stale_render.pid}" f"/stills/{d6_episode_with_stale_render.sid}/generate-image", json={}, ) assert res.status_code == 422 body = res.json() assert body["error"]["code"] == "STALE_UPSTREAM" def test_endpoint_passes_for_fresh_episode(client, d6_episode): res = client.post( f"/api/v1/projects/{d6_episode.pid}/stills/{d6_episode.sid}/generate-image", json={}, ) assert res.status_code == 200 ``` - [ ] **Step 8: Commit** ``` git add backend/app/core/errors/stale_upstream.py backend/app/services/dispatcher_preflight.py backend/app/api/v1/images.py backend/tests/services/test_dispatcher_preflight.py backend/tests/integration/test_d6_endpoint_preflight.py git commit -m "feat(d6): T8 dispatcher preflight at endpoint entry + STALE_UPSTREAM 422" ``` --- ## T9: validator error code 분리 (fallback safety net) — R2 I3 patch **R2 I3 — chain_bg_lookup 분기 명시**: - 새 `BG_ID_RE` (`^L\d{2,3}B\d{2,3}$`) match 시 → `StaleUpstreamError` (D6 path, render manifest stale). - legacy `bg_*` longform (예: 마이그레이션 전 episode) 시 → 기존 `RefContractError` (D5 정합). - chain_bg_lookup result 는 둘 다 None 가능 (D5 의 passive lookup) — 분기는 bg_id 형식 기준. **Files:** - Modify: `backend/app/core/ref_contract_validator.py` - Test: `backend/tests/services/test_d6_validator_stale_upstream.py` - [ ] **Step 1: Write the failing test** ```python def test_validator_emits_stale_upstream_when_bg_id_missing_from_render(d6_no_preflight_path): """preflight 가 빠진 path 로 호출되면 validator 가 STALE_UPSTREAM 분리 raise.""" from app.core.ref_contract_validator import validate_attached_refs from app.core.errors.stale_upstream import StaleUpstreamError import pytest rpc = {"asset_requirements": { "required_refs": [{"kind": "background", "id": "L09B01", "policy": "required"}] }} # background_chain_bg_map 에 L09B01 없음 → render manifest stale 의 표현 with pytest.raises(StaleUpstreamError): validate_attached_refs( rpc, [], [], "prompt", is_close_framing=False, chain_bg_lookup=lambda x: None, ) ``` - [ ] **Step 2: Modify `validate_attached_refs`** 기존 `RefContractError` raise 부분 (line 273-278) 분기: ```python if not is_close_framing: for bg_id in required.get("background", []): if ("background", bg_id) in attached_set: continue required_loc = chain_bg_lookup(bg_id) if chain_bg_lookup else None if required_loc and ("background_prev_shot", required_loc) in attached_set: continue # D6 — preflight fallback. bg_id 가 render manifest 부재 = STALE_UPSTREAM. # bg_id 형식이 ^L\d{2,3}B\d{2,3}$ 면 D6 path → STALE_UPSTREAM. # legacy bg_* 형식은 기존 RefContractError (D5 정합). from app.core.bg_state_vocab import BG_ID_RE from app.core.errors.stale_upstream import StaleUpstreamError if BG_ID_RE.match(bg_id): raise StaleUpstreamError( upstream="background_render", expected_bg_catalog_hash="(unknown — preflight 미통과 path)", observed_bg_catalog_hash="", missing_in_render=[bg_id], remediation="POST /steps/background_render?mode=force", ) raise RefContractError( f"required background {bg_id!r} missing " f"(expected exact ('background', {bg_id!r}) or " f"('background_prev_shot', {required_loc!r})) — " f"attached={sorted(attached_set)}" ) ``` - [ ] **Step 3: Run test + commit** ``` cd backend && .venv/bin/python -m pytest tests/services/test_d6_validator_stale_upstream.py -xvs ``` Expected: PASS ``` git add backend/app/core/ref_contract_validator.py backend/tests/services/test_d6_validator_stale_upstream.py git commit -m "feat(d6): T9 validator fallback STALE_UPSTREAM 분리 (preflight safety net)" ``` --- ## T10: HTTP response shape integration **unit-style pivot (T6/T7/T8 와 동일 결정)** — `d6_episode_with_stale_*` fixture 미존재. T8 Phase B 의 TestClient 패턴 (real DB project/episode/SceneStill) 위에 T8a 의 tmp manifest 패턴 (settings.projects_dir 의 checkpoint 경로에 master_plan + 4 consumer manifest 직접 기록) 결합. preflight 는 mock 안 하고 real (file I/O), service 는 200 path 검증용 mock. **T8 Phase B 와 차이**: - T8 Phase B (`tests/api/test_d6_endpoint_preflight.py`): preflight 함수 자체를 monkeypatch → endpoint↔preflight wiring + ordering invariant 검증. 422 shape 은 mock raise 시 hardcoded 값. - T10 (`tests/api/test_d6_http_response_shape.py`): real manifest → real preflight 호출 → endpoint 422 shape (full chain). manifest 의 stale 조건 (catalog mismatch / binding mismatch) 별 422 vs 200 boundary 검증. **Files:** - Test: `backend/tests/api/test_d6_http_response_shape.py` (NEW — `tests/api/` 하위로 T8 Phase B 와 일치) - [ ] **Step 1: Write the integration test (unit-style pivot)** ```python def test_stale_upstream_response_shape(client, d6_episode_with_stale_render): res = client.post( f"/api/v1/projects/{d6_episode_with_stale_render.pid}" f"/stills/{d6_episode_with_stale_render.sid}/generate-image", json={}, ) assert res.status_code == 422 body = res.json() err = body["error"] assert err["code"] == "STALE_UPSTREAM" assert err["upstream"] in ("background_render", "background_prompt", "scene_detail", "floor_plan_prompt") assert "expected_bg_catalog_hash" in err assert "observed_bg_catalog_hash" in err assert "remediation" in err assert "force" in err["remediation"] def test_custom_prompt_passes_with_stale_binding(client, d6_episode_with_stale_binding): """custom_prompt 는 binding stale 무시.""" res = client.post( f"/api/v1/projects/{d6_episode_with_stale_binding.pid}" f"/stills/{d6_episode_with_stale_binding.sid}/generate-image", json={"custom_prompt": "test prompt"}, ) # binding 만 stale, catalog OK → 200 assert res.status_code == 200 def test_custom_prompt_blocked_on_stale_catalog(client, d6_episode_with_stale_render): """custom_prompt 도 catalog stale 면 차단.""" res = client.post( f"/api/v1/projects/{d6_episode_with_stale_render.pid}" f"/stills/{d6_episode_with_stale_render.sid}/generate-image", json={"custom_prompt": "test prompt"}, ) assert res.status_code == 422 assert res.json()["error"]["code"] == "STALE_UPSTREAM" ``` - [ ] **Step 2: Run + commit** ``` cd backend && .venv/bin/python -m pytest tests/integration/test_d6_http_response_shape.py -xvs ``` Expected: PASS (3 tests) ``` git add backend/tests/integration/test_d6_http_response_shape.py git commit -m "test(d6): T10 HTTP response shape + custom_prompt branch coverage" ``` --- ## T11: operator runbook — ordered force (R2 I4 patch) **R2 I4 — `floor_plan_render` non-stamp note**: - runbook 에 `floor_plan_render` 가 D6 stamp 대상 아님 (spec §3 Non-goals A3) 명시. - 그러나 force order 에 포함 — `floor_plan_prompt` 갱신 후 fp PNG 재생성 의무 (LLM 의 새 catalog 가 fp 와 정합 안 되면 시각 결함 가능). - preflight 가 fp_render stale 을 catch 안 함 — 운영자 책임으로 명시. **Files:** - Create: `docs/runbooks/d6-migration.md` - [ ] **Step 1: Write runbook** ```markdown # D6 Migration Runbook — ordered force re-run ## When to use - 새 D6 backend deploy 후 기존 episode 의 bg_id 가 longform (`bg_*`) → STALE_UPSTREAM 응답. - master_plan 갱신 후 downstream cascade 가 자동으로 일어나지 않음 — 운영자 명시 force 의무. ## Force order D6 의 cascade 매커니즘은 `_config_hash` mismatch 가 아닌 **operator 명시 force** 로 작동. 다음 순서 그대로: 1. `background_master_plan` — 새 prompt + post-processing 6 단계 적용. ``` POST /api/v1/projects/$PID/episodes/$EID/steps/background_master_plan?mode=force ``` 2. `floor_plan_prompt` — catalog 의 fp 의존성 갱신. ``` POST /api/v1/projects/$PID/episodes/$EID/steps/floor_plan_prompt?mode=force ``` 3. `floor_plan_render` — fp PNG 갱신 (필요 시). 4. `background_prompt` — t2i prompt + shot_guides 갱신. 5. `background_render` — bg PNG 재생성 (gpt-image-2 비용 발생). 6. `scene_detail` — RPC 의 background_binding.bg_id 새 형식 stamp. ## Wrong-order detection 순서 어긋나면 dispatcher preflight 가 STALE_UPSTREAM 422 로 차단: ```json { "error": { "code": "STALE_UPSTREAM", "upstream": "background_render", "expected_bg_catalog_hash": "...", "observed_bg_catalog_hash": "", "remediation": "POST /steps/background_render?mode=force ..." } } ``` ## Cost estimate - gpt-image-2: 24 bg × ~$0.15 ≈ $3.6/episode - 도면: 14 fp × ~$0.10 ≈ $1.4/episode - Total: ~$5/episode. ## Verification 각 step force 후 manifest mtime + status 확인: ```bash PID="..." EID="..." curl -sS -b /tmp/theroad_cookies.txt \ "http://localhost:8000/api/v1/projects/$PID/episodes/$EID/steps" \ | python3 -c "import sys,json; d=json.load(sys.stdin); \ [print(s['step_id'], s['status']) for s in d['steps'] \ if s['step_id'] in ('background_master_plan', 'floor_plan_prompt', \ 'background_prompt', 'background_render', 'scene_detail')]" ``` 모든 step `completed` 확인 후 단건 generate-image 재호출. ``` - [ ] **Step 2: Commit** ``` git add docs/runbooks/d6-migration.md git commit -m "docs(d6): T11 operator ordered-force runbook" ``` --- ## T12: legacy compat note (no code) — **ABSORBED INTO T9, drop** > **Closure note (review iter — T11 후 재평가)**: T12 원안은 "legacy `bg_*` episode → 422 STALE_UPSTREAM 차단" 가정 → **현재 코드 정책과 모순**. 실제는 (a) preflight 가 master_plan 의 `bg_catalog_hash` 부재 시 D5 fallback silent return (`backend/app/services/dispatcher_preflight.py:111`), (b) legacy `bg_*` background missing 은 D6 `StaleUpstreamError` 가 아니라 기존 `RefContractError` 경로 (`backend/app/core/ref_contract_validator.py:282 분기`). T12 원안대로 작성 시 잘못된 모델 고정. > > **흡수 위치**: T9 의 `tests/services/test_d6_validator_stale_upstream.py::test_validator_keeps_ref_contract_error_for_legacy_bg_id` (4 parametrized: bg_supermarket_dusk / bg_kitchen / bg_office_normal / "") 가 정확히 "legacy 는 D6 분리 우회 + 기존 RefContractError 보존" 보장. 추가 test 는 동일 검증 duplication. > > **운영 정책**: T11 runbook (`docs/runbooks/d6-migration.md`) 의 "When to use" case 3 에서 "Pre-D6 episode 는 preflight 차단 안 됨, validator 가 RefContractError. 어느 경우든 ordered force migration 으로 D6 형식 전환" 명시. > > 아래 step 1~2 pseudocode 는 history 보존용 — **무시할 것**. D6 는 compatibility map 안 만듦 — schema_version bump 만으로 자동 reject (T5). **Files:** - Test: `backend/tests/integration/test_d6_legacy_episode_blocked.py` - [ ] **Step 1: Write the integration test** ```python def test_legacy_bg_id_episode_blocked_until_force_rerun(legacy_d5_episode, client): """기존 longform bg_* episode 는 D6 deploy 후 422 STALE_UPSTREAM.""" res = client.post( f"/api/v1/projects/{legacy_d5_episode.pid}/stills/{legacy_d5_episode.sid}/generate-image", json={}, ) assert res.status_code == 422 assert "STALE_UPSTREAM" in res.json().get("error", {}).get("code", "") ``` - [ ] **Step 2: Run + commit** ``` cd backend && .venv/bin/python -m pytest tests/integration/test_d6_legacy_episode_blocked.py -xvs git add backend/tests/integration/test_d6_legacy_episode_blocked.py git commit -m "test(d6): T12 legacy episode blocked → force re-run path verified" ``` --- ## T13: wrong-order force smoke test (R2 I5 patch) — **ABSORBED INTO T10, drop** > **Closure note (review iter — T11 후 재평가)**: T13 원안은 실제 step force endpoint 호출 (async background job 시작 + polling) + LLM/fixture 무거운 E2E. 본질 검증 ("consumer manifest hash 가 master_plan hash 와 다르면 422 with `upstream` field") 은 이미 T10 의 manifest-기반 real preflight 검증으로 cover. > > **흡수 위치**: > - `tests/api/test_d6_http_response_shape.py::test_binding_mismatch_returns_422_with_upstream_per_consumer` (4 parametrized) — 4 consumer 중 어느 stale 이든 `upstream` 필드가 정확한 consumer 식별, R2 I5 의 assertion 강화 의도와 정합. > - `tests/api/test_d6_http_response_shape.py::test_stale_upstream_response_shape_full_payload` — `expected_*/observed_*/remediation` payload shape full 검증. > - `tests/api/test_d6_endpoint_preflight.py` (9 case) — single+batch endpoint preflight wiring + ordering invariant. > > Full E2E 는 step async + polling + Gemini/gpt-image-2 비용으로 flaky risk + maintenance 부담 큼 (T11 runbook §"step 완료 polling 의무" 가 운영자 절차로 별도 가이드). T13-lite (manifest mutation 1 scenario) 도 T10 의 4 parametrized 와 framing 만 다름 — 기각. > > 아래 step 1~2 pseudocode 는 history 보존용 — **무시할 것**. **R2 I5 — assertion 강화**: - test 가 단순 `err["code"] == "STALE_UPSTREAM"` 가 아니라 **`err["upstream"]` 도 명시** assert. - scene_detail 만 force 후 → preflight 가 catch 하는 first stale = `background_render` (또는 `background_prompt`/`floor_plan_prompt`). assertion: `err["upstream"] == "background_render"`. - scene_detail 자체 force 는 200 (별도 step force 는 성공). preflight failure 는 후속 generate-image call. **Files:** - Test: `backend/tests/integration/test_d6_force_order.py` - [ ] **Step 1: Write the smoke test** ```python def test_wrong_order_force_produces_stale_upstream(d6_episode_pending_master_plan, client): """master_plan force 안하고 scene_detail 만 force → preflight STALE_UPSTREAM.""" pid = d6_episode_pending_master_plan.pid eid = d6_episode_pending_master_plan.eid sid = d6_episode_pending_master_plan.sid # scene_detail 만 force client.post(f"/api/v1/projects/{pid}/episodes/{eid}/steps/scene_detail?mode=force") # 단건 generate-image 시도 res = client.post(f"/api/v1/projects/{pid}/stills/{sid}/generate-image", json={}) assert res.status_code == 422 err = res.json()["error"] assert err["code"] == "STALE_UPSTREAM" def test_correct_order_force_succeeds(d6_episode_pending_master_plan, client): """master_plan → floor_plan_prompt → background_prompt → background_render → scene_detail 순서대로 force 후 단건 호출 200.""" pid = d6_episode_pending_master_plan.pid eid = d6_episode_pending_master_plan.eid sid = d6_episode_pending_master_plan.sid for step in ("background_master_plan", "floor_plan_prompt", "background_prompt", "background_render", "scene_detail"): client.post(f"/api/v1/projects/{pid}/episodes/{eid}/steps/{step}?mode=force") # 각 step 완료 대기 (백그라운드 thread) # ... res = client.post(f"/api/v1/projects/{pid}/stills/{sid}/generate-image", json={}) assert res.status_code == 200 ``` - [ ] **Step 2: Run + commit** ``` cd backend && .venv/bin/python -m pytest tests/integration/test_d6_force_order.py -xvs git add backend/tests/integration/test_d6_force_order.py git commit -m "test(d6): T13 wrong-order force smoke + correct-order success" ``` --- ## Acceptance Criteria Final Check 각 AC 가 실제 task 에서 cover 되는지 매핑: - **AC-1 (deterministic ID)** — T4 (assign_bg_ids) + T7 (5 reruns determinism) - **AC-2 (semantic_key key-only)** — T4 (compute_semantic_key + bg_catalog_hash 의 metadata 제외) - **AC-3 (controlled vocab strict)** — T1 (validate_state_class) + T4 (normalize_space_key) - **AC-4 (cross-step lineage)** — T5 (consumer hash stamp) + T7 (hash isolation) - **AC-5 (scene_detail = chain 소비자)** — T6 (chain_bg_id_by_shot 그대로) + scene_detail prompt 변경 0 검증 - **AC-6 (STALE_UPSTREAM error)** — T8 (preflight) + T9 (validator fallback) + T10 (HTTP shape) + custom_prompt T10 branch - **AC-7 (legacy migration)** — T11 (runbook) + T12 (legacy blocked) - **AC-8 (regression)** — T1 (regex 분리) + T6 (chain unchanged) + 전체 broader test suite --- ## Self-Review **1. Spec coverage:** - §4.1 bg_id 형식 → T1 (BG_ID_RE / format_bg_id) ✓ - §4.2 semantic_key → T4 (compute_semantic_key + bg_catalog_hash projection) ✓ - §4.3 space_profile → T2 (entity_extractor) + T4 (normalize_space_key) ✓ - §4.4 state_class enum → T1 (STATE_CLASS_ENUM) + T3 (master_plan prompt) ✓ - §4.5 catalog 빌드 + 6 단계 ordering → T4 ✓ - §4.6 hash 분리 + consumer table → T4 (hash) + T5 (stamp) ✓ - §4.7 scene_detail = chain 소비자 → T6 ✓ - §4.8 STALE_UPSTREAM preflight → T8 + T9 + T10 ✓ - §6 D6-D ordered-force runbook → T11 ✓ - §6 wrong-order smoke → T13 ✓ **2. Placeholder scan:** 모든 task 에 actual code + commands. "TBD"/"implement later"/"add error handling" 없음. **3. Type consistency:** - `BG_ID_RE` (T1) → `assign_bg_ids` (T4) → `validate_master_plan_assigned` (T4) - `STATE_CLASS_ENUM` (T1) → `validate_state_class` (T1) → `compute_semantic_key` (T4) - `compute_bg_catalog_hash` / `compute_shot_binding_hash` (T4) → consumer stamp (T5) → preflight (T8) - `StaleUpstreamError` (T8) → validator fallback (T9) → HTTP response (T10) 이름/시그니처 일관성 확인 ✓ --- ## Plan Status **DRAFT_R2 — APPROVED_FOR_IMPLEMENTATION (다음 세션)** R2 patch 적용 항목 (Claude audit B1~B7 + I1~I8 + 추가발견 10건 cover): - B1 file mapping 정정 / B2 `_SAFE_BG_RE` 4 site 교체 (T-pre-1) / B3 sibling shape + handshake / B4 entity sync 4 site / B5 alembic sequential + idempotent / B6 T3+T4 atomic merge + T-pre-2 validator split / B7 monotonic next_b - I1~I8 모두 task body 안 inline patch - Codex 재audit 생략 (사용자 결정) **다음 세션 진입 시**: - 사용자 권고: **Inline + task별 review checkpoint**. D6 가 cross-step contract 라 작업자 분산하면 signature/manifest shape 어긋날 가능성 큼. - T0~T1 은 작고 독립적 — subagent 도 OK. - T2 이후는 한 흐름으로 inline 실행 + 각 task commit 후 사용자 review checkpoint. - T-pre-1 (B2 _SAFE_BG_RE 교체) 는 T1 직후 즉시 진행. - T-pre-2 (validator split) 는 T3+T4 atomic commit 안에 포함. **Plan complete and saved to `docs/superpowers/plans/2026-05-09-d6-deterministic-bg-id-and-catalog-lineage-implementation.md`.**