"""FINDING 8 (e2e-bughunt-v1) — scene_detail render_prompt_card verify dual-source. scene_detail 의 stored ``render_prompt_card`` 는 ``previous_shot_refs`` / ``forward_zoom_targets`` 를 ``ctx.dependencies`` 에서 도출한다. ``SceneContextLoader._load_dependencies()`` 는 downstream ``shot_dependency_t2i`` checkpoint 가 존재하면 그것을 우선 사용한다. ``shot_dependency_t2i`` 는 scene_detail *이후* 순서(21.71 > 21.7)에 실행되므로, scene_detail 이 카드를 저장한 뒤 ``shot_dependency_t2i`` 가 갱신되면 이후의 ``verify_completion()`` card-hash recompute 가 카드 build 당시와 다른 dependency source 를 읽어 **false-positive ``card_drifted`` hash** 가 발생한다. (image generation resume 시 49/64 shot drift → pipeline 중단.) Opt C dual-source verify fix: 1차 recompute 가 hash mismatch 면 upstream ``shot_dependency`` checkpoint 를 source 로 한 fallback recompute 를 1회 수행 — fallback 이 stored hash 와 일치하면 false drift 로 보고 통과시킨다. fallback 은 구조 손상을 절대 구제하지 않는다 (hash-only rescue). LLM/DB 미사용 — ``SceneContextLoader.load_all`` mock. """ from __future__ import annotations import copy import json from pathlib import Path from unittest.mock import MagicMock, patch from app.core.dto.scene_analysis import SceneAnalysisContext from app.core.step_manifest import get_manifest_dict from app.core.steps._owned_helpers import ( OWNED_VALIDATOR_FULL, compute_camera_direction_hash, compute_owned_hash, compute_t2i_prompt_hash, ) from app.core.steps.detail_steps import ( SCENE_DETAIL_SCHEMA_VERSION, SceneDetailStep, _collect_card_inputs, ) from app.core.steps.render_prompt_card import ( build_render_prompt_card, compute_card_hash, ) from app.core.steps.scene_context_loader import SceneContextLoader # 두 dependency variant — 유일한 차이는 location_refs[0].ref_usage. # upstream(shot_dependency): scene_detail producer 가 카드 build 당시 본 source. # t2i(shot_dependency_t2i): scene_detail 이후 갱신돼 verify 시점에 보이는 source. _UPSTREAM_DEPS = [ { "scene_index": 12, "shot_index": 4, "location_refs": [ { "scene_index": 12, "shot_index": 3, "ref_usage": "", "keep_elements": [], } ], "character_refs": [], } ] _T2I_DEPS = [ { "scene_index": 12, "shot_index": 4, "location_refs": [ { "scene_index": 12, "shot_index": 3, "ref_usage": "exact_background", "keep_elements": [], } ], "character_refs": [], } ] _PROMPT_TEXT = "A figure stands by the doorway." _CAM_DIR = "medium shot of doorway" _OWNED = ["door", "window"] def _build_ctx(dependencies: list) -> SceneAnalysisContext: """non-trivial SceneAnalysisContext — dependencies 만 parametrize.""" ctx = SceneAnalysisContext(project_id="P1", episode_id="E1") ctx.segments = [{"scene_index": 12, "text": "scene 12 text"}] ctx.selected_map = {12: {4}} ctx.shot_scenes_map = { 12: [{"shot_index": 4, "camera_direction": _CAM_DIR, "description": "wide doorway shot"}], } ctx.scene_visible = {12: ["C01"]} ctx.summaries = {12: "summary"} ctx.world_rules = {} ctx.planning_context = None ctx.staging_map = { "12_4": { "camera_direction": _CAM_DIR, "framing_scale": "medium", "lighting_mood": "warm dim", "key_bg_elements": [], "subject_reference_policy": [], }, } ctx.chain_bg_owned_by_shot = {(12, 4): list(_OWNED)} ctx.chain_bg_camera_meta_by_shot = { (12, 4): { "camera_position": "southeast doorway", "camera_height": "eye-level", "lens_hint": "35mm", "framing_notes": "wide", }, } ctx.chain_bg_guide_by_shot = {(12, 4): "doorway view"} ctx.chain_bg_id_by_shot = {(12, 4): "cb_main_room"} ctx.shot_director_ve = {(12, 4): ["C01"]} ctx.shot_director_vr = {} ctx.outlook_data = { "outlooks": [{"outlook_id": "O02", "name": "casual_jacket"}], "scene_assignments": [ {"scene_index": 12, "assignments": [ {"character_id": "C01", "outlook_id": "O02"}, ]}, ], } ctx.fixed_elements_by_scene = { 12: [{ "element_id": "body_full_pose", "element_type": "character_state", "character_name": "C01", "description": "the figure stands rigid in the doorway", "applies_to_shots": [4], "element_scope": "full", "source_facts": ["scene narration: 'they stood unmoving'"], "visual_inferences": ["lighting falls from corridor"], "creative_decisions": ["camera respects 180-degree line"], "confidence": "high", }], } ctx.dependencies = copy.deepcopy(dependencies) return ctx def _make_step(tmp_path, monkeypatch, *, bg_mode: str = "on") -> SceneDetailStep: monkeypatch.setattr( "app.core.config.settings.projects_dir", str(tmp_path), raising=False, ) monkeypatch.setattr( "app.core.config.settings.background_mode", bg_mode, raising=False, ) step = SceneDetailStep.__new__(SceneDetailStep) step.project_id = "P1" step.episode_id = "E1" step.project_config = {} step.step_id = "scene_detail" step.run_id = "run-test" step.opik_context = {} step.db = MagicMock() step.db.execute.return_value.fetchall.return_value = [] step.db.execute.return_value.fetchone.return_value = None step.manifest = get_manifest_dict("scene_detail") step._cp_dir = ( Path(tmp_path) / "P1" / "checkpoints" / "episodes" / "E1" / "scene_detail" ) return step def _write_shot_dependency_cp(tmp_path, dependencies: list) -> None: d = ( Path(tmp_path) / "P1" / "checkpoints" / "episodes" / "E1" / "shot_dependency" ) d.mkdir(parents=True, exist_ok=True) (d / "manifest.json").write_text( json.dumps({"data": {"dependencies": dependencies}}), encoding="utf-8", ) def _build_stored_card(ctx: SceneAnalysisContext): """producer 와 동일 ctx-driven single source 로 card + hash 산출.""" inputs = _collect_card_inputs( ctx=ctx, seg={"scene_index": 12}, shot_info={"shot_index": 4}, ) builder_inputs = {k: v for k, v in inputs.items() if k != "ctx"} card = build_render_prompt_card(**builder_inputs) return card, compute_card_hash(card) def _cp_result(card: dict, card_hash: str) -> dict: # owned_validation sentinel — verify_completion 의 4-way drift 검증 # (validator / t2i_prompt_hash / owned_hash / camera_direction_hash) 통과용. # 수동 구성 — owned/owned_object_usage cardinality 검증 우회 (test fixture). sentinel = { "schema_version": 2, "validator": OWNED_VALIDATOR_FULL, "owned_hash": compute_owned_hash(list(_OWNED)), "camera_direction_hash": compute_camera_direction_hash(_CAM_DIR), "t2i_prompt_hash": compute_t2i_prompt_hash(_PROMPT_TEXT), "owned_usage_hash": "0123456789abcdef", "violations": [], } return { "schema_version": SCENE_DETAIL_SCHEMA_VERSION, "data": { "scenes": [{ "scene_index": 12, "_shot_index": 4, "visible_entities": ["C01"], "render_prompt_card": card, "render_prompt_card_hash": card_hash, "t2i_variations": [{ "t2i_prompt": _PROMPT_TEXT, "outfit_assignments": [ {"character_id": "C01", "outlook_id": "O02"}, ], "owned_validation": sentinel, }], }], }, } def _run_verify(step, verify_ctx): with patch.object( SceneContextLoader, "load_all", return_value=verify_ctx, ), patch.object( SceneContextLoader, "_load_chain_bg_owned_by_shot", return_value=verify_ctx.chain_bg_owned_by_shot, ), patch.object( SceneContextLoader, "_load_staging_map", return_value=verify_ctx.staging_map, ): return step.verify_completion() # ────────────────────────────────────────────────────────────────────────── # G1/G2 — card 가 upstream shot_dependency 로 build 됐고 verify 는 # downstream shot_dependency_t2i 를 보는 경우: dual-source fallback 이 # false drift 를 해소해 verify_completion 이 clean 이어야 한다. # fix 전: fallback 없음 → card_drifted hash → partial (RED). # ────────────────────────────────────────────────────────────────────────── def test_upstream_built_card_resolved_by_dual_source_fallback( tmp_path, monkeypatch, ): step = _make_step(tmp_path, monkeypatch, bg_mode="on") # stored card = upstream shot_dependency source. card, card_hash = _build_stored_card(_build_ctx(_UPSTREAM_DEPS)) step._last_execute_result = _cp_result(card, card_hash) # verify 시점 ctx 는 downstream shot_dependency_t2i source (다른 ref_usage). verify_ctx = _build_ctx(_T2I_DEPS) # upstream shot_dependency checkpoint 존재 — fallback source. _write_shot_dependency_cp(tmp_path, _UPSTREAM_DEPS) report = _run_verify(step, verify_ctx) assert report.is_complete is True, ( f"dual-source fallback 이 upstream shot_dependency 로 recompute 해 " f"false drift 를 해소해야 함. missing={list(report.missing)!r} " f"severity={report.severity!r}" ) assert report.severity == "clean", ( f"severity must be 'clean' (got {report.severity!r})" ) assert not report.metadata.get("card_drifted"), ( f"card_drifted 가 비어 있어야 함: {report.metadata.get('card_drifted')!r}" ) # ────────────────────────────────────────────────────────────────────────── # Case-B regression guard — card 가 shot_dependency_t2i 로 build 됐고 # verify 도 동일 source 를 보는 경우: 1차 recompute 가 곧바로 일치 → # fallback 미사용 → clean. (fix 가 역방향 false-drift 를 만들지 않는지 확인.) # ────────────────────────────────────────────────────────────────────────── def test_t2i_built_card_no_false_drift_when_verify_matches( tmp_path, monkeypatch, ): step = _make_step(tmp_path, monkeypatch, bg_mode="on") # stored card = downstream shot_dependency_t2i source. card, card_hash = _build_stored_card(_build_ctx(_T2I_DEPS)) step._last_execute_result = _cp_result(card, card_hash) # verify 시점 ctx 도 동일 t2i source. verify_ctx = _build_ctx(_T2I_DEPS) # upstream cp 가 존재해도 1차 recompute 가 이미 일치하므로 fallback 미진입. _write_shot_dependency_cp(tmp_path, _UPSTREAM_DEPS) report = _run_verify(step, verify_ctx) assert report.is_complete is True, ( f"1차 recompute 가 일치하면 fallback 없이 clean 이어야 함. " f"missing={list(report.missing)!r} severity={report.severity!r}" ) assert report.severity == "clean" assert not report.metadata.get("card_drifted") # ────────────────────────────────────────────────────────────────────────── # G3 canary — render_prompt_card_hash 가 실제로 손상(어떤 dependency source # 로도 재현 불가)된 경우: 1차/fallback recompute 둘 다 mismatch → # card_drifted hash 유지 → partial. fallback 이 진짜 drift 를 구제하면 안 됨. # ────────────────────────────────────────────────────────────────────────── def test_genuine_hash_corruption_still_drifts(tmp_path, monkeypatch): """저장 손상은 차단 유지 — Codex 합의 BLOCK3 1단계 (2026-08-12). 2a3cb8e1 이 live-mismatch hash 를 경고로 낮추면서 이 시나리오(카드 정상 + 저장 hash 오염)까지 clean 으로 삼켰다. 이제 stored-card 자기정합 검사(compute_card_hash(stored) vs stored_hash — live ctx 무관이라 하류 재산출 순환 없음)가 ``self_hash`` 구조 태그로 차단한다. """ step = _make_step(tmp_path, monkeypatch, bg_mode="on") # card payload 는 upstream 기반(shape 정상)이나 hash 는 bogus. card, _real_hash = _build_stored_card(_build_ctx(_UPSTREAM_DEPS)) bogus_hash = "deadbeefdeadbeef" assert bogus_hash != _real_hash step._last_execute_result = _cp_result(card, bogus_hash) verify_ctx = _build_ctx(_T2I_DEPS) _write_shot_dependency_cp(tmp_path, _UPSTREAM_DEPS) report = _run_verify(step, verify_ctx) assert report.is_complete is False, ( "저장 손상(자기정합 불일치)은 차단 유지여야 함" ) assert report.severity == "partial" drifted = report.metadata.get("card_drifted") or [] assert (12, 4, "self_hash") in drifted, ( f"card_drifted 에 (12,4,'self_hash') 가 있어야 함: {drifted!r}" ) # 감사 계약: 자기정합 불일치여도 live recompute 는 계속 진행되어 # 두 태그가 병존한다 — 손상 케이스의 기록이 더 정확해진다. assert (12, 4, "hash") in drifted, ( f"self_hash 와 hash 태그가 병존해야 함(감사 계약): {drifted!r}" ) # ────────────────────────────────────────────────────────────────────────── # G4 — pure verifier/loader-boundary fix: schema/prompt version bump 없음. # ────────────────────────────────────────────────────────────────────────── def test_no_schema_version_bump(): # FINDING 8 자체는 pure verifier fix — bump 없었음 (was 12 at time of fix). # reference-necessity Phase 2 (2026-05-23): render_prompt_card overlay → 12→13. assert SCENE_DETAIL_SCHEMA_VERSION == 13, ( "reference-necessity Phase 2: SCENE_DETAIL_SCHEMA_VERSION bump 12→13" )