"""W1 experiment_background_pipeline_slice — minimal slice TDD. 기존 backend/tests/scripts/* convention 과 동일: - _SCRIPTS sys.path prepend → bare module import (`from experiment_background_pipeline_slice import ...`) - pytest 실행 CWD = `backend/` (sys.path 에는 backend 자체가 들어가지 repo root 가 들어가지 않음) """ import json import os import subprocess import sys from pathlib import Path _REPO_ROOT = Path(__file__).resolve().parents[3] _SCRIPT = _REPO_ROOT / "backend" / "scripts" / "experiment_background_pipeline_slice.py" _SCRIPTS = _REPO_ROOT / "backend" / "scripts" if str(_SCRIPTS) not in sys.path: sys.path.insert(0, str(_SCRIPTS)) def _subprocess_env(): """Strip pytest-conftest test-DB env so subprocess loads production DATABASE_URL from backend/.env. conftest.py overrides DATABASE_URL to the test PG database for in-process tests, but this experiment script must read from the real PostgreSQL production fixture (project/episode IDs live there). """ env = os.environ.copy() # Conftest pins DATABASE_URL to test DB; drop so .env (production) wins. env.pop("DATABASE_URL", None) env.pop("DATABASE_URL_TEST", None) env.pop("ENVIRONMENT", None) env.pop("PROJECTS_DIR", None) return env def _run(args, *, cwd=None): return subprocess.run( [sys.executable, str(_SCRIPT), *args], cwd=str(cwd or _REPO_ROOT), capture_output=True, text=True, timeout=120, env=_subprocess_env(), ) # ────────────────────────────────────────────────────────────────────── # Phase 1 — Skeleton (3 tests) # ────────────────────────────────────────────────────────────────────── def test_skeleton_emits_run_meta_with_required_fields(tmp_path): out_root = tmp_path / "out" result = _run([ "--dry-run", "--skip-db", "--output-root", str(out_root), ]) assert result.returncode in (0, 1), result.stderr run_dirs = sorted(out_root.glob("*/")) assert len(run_dirs) == 1 run_meta = json.loads((run_dirs[0] / "run_meta.json").read_text()) for key in ("run_id", "plan_version", "generated_at", "model", "args", "outputs", "run_status", "exit_code"): assert key in run_meta, f"run_meta missing {key}" assert run_meta["plan_version"] == "bps_w1" def test_run_id_format_kst_and_unique(tmp_path): out_root = tmp_path / "out" r1 = _run(["--dry-run", "--skip-db", "--output-root", str(out_root)]) r2 = _run(["--dry-run", "--skip-db", "--output-root", str(out_root)]) assert r1.returncode in (0, 1) and r2.returncode in (0, 1) runs = sorted(out_root.glob("*/")) assert len(runs) == 2 for d in runs: name = d.name assert len(name) == 13 + 1 + 6, f"unexpected run_id: {name}" # YYYYMMDD_HHMM_xxxxxx assert name[8] == "_" and name[13] == "_" assert runs[0].name != runs[1].name def test_dry_run_default_no_model_in_meta(tmp_path): out_root = tmp_path / "out" _run(["--dry-run", "--skip-db", "--output-root", str(out_root)]) run_meta = json.loads(next(out_root.glob("*/run_meta.json")).read_text()) assert run_meta["model"] is None assert run_meta["args"]["generate"] is False # ────────────────────────────────────────────────────────────────────── # Phase 2 — SourceBundle (3 tests) # ────────────────────────────────────────────────────────────────────── def test_source_bundle_loads_selected_shots_verbatim(tmp_path): out_root = tmp_path / "out" result = _run([ "--dry-run", "--output-root", str(out_root), "--project-id", "6cb862d9-590c-4dce-86e6-d10c2977db19", "--episode-id", "08ad2cd3-3e96-4d84-808f-869ee628473c", "--stop-after", "source_bundle", ]) assert result.returncode in (0, 1), result.stderr run_dir = next(out_root.glob("*/")) bundle = json.loads((run_dir / "source_bundle.json").read_text()) assert bundle["project_id"] == "6cb862d9-590c-4dce-86e6-d10c2977db19" assert bundle["episode_id"] == "08ad2cd3-3e96-4d84-808f-869ee628473c" assert bundle["selected_shot_count"] > 0 for shot in bundle["selected_shots"]: assert shot["is_selected"] is True # row_id = UUID (DB), shot_key = composite SOT for required in ("scene_index", "shot_index", "shot_description", "t2i_variations_json", "visible_entities_json", "row_id", "shot_key"): assert required in shot, f"shot row missing {required}" # shot_key format assert shot["shot_key"].startswith("S") # verbatim — no truncate assert shot.get("t2i_variations_truncated") is None assert bundle["source_hash"] # location catalog included assert isinstance(bundle["locations"], list) for loc in bundle["locations"]: assert loc["entity_type"] == "location" assert "short_id" in loc and "metadata_json" in loc # location_profiles (validate_location_space_profile 통과만) assert isinstance(bundle["location_profiles"], dict) assert isinstance(bundle["location_profile_errors"], list) def test_source_bundle_fails_when_no_selected_shots(tmp_path): out_root = tmp_path / "out" # use a project_id that doesn't exist result = _run([ "--dry-run", "--output-root", str(out_root), "--project-id", "ZZZZZZZZ-0000-0000-0000-000000000000", "--episode-id", "ZZZZZZZZ-0000-0000-0000-000000000000", "--stop-after", "source_bundle", ]) assert result.returncode == 1 run_meta = json.loads(next(out_root.glob("*/run_meta.json")).read_text()) assert run_meta["run_status"] == "validation_failed" assert "no_selected_shots" in run_meta["failed_invariants"] def test_source_verbatim_hash_matches_db(tmp_path): """Bundle 의 verbatim hash 가 DB row 와 일치해야 함.""" import hashlib out_root = tmp_path / "out" _run([ "--dry-run", "--output-root", str(out_root), "--project-id", "6cb862d9-590c-4dce-86e6-d10c2977db19", "--episode-id", "08ad2cd3-3e96-4d84-808f-869ee628473c", "--stop-after", "source_bundle", ]) bundle = json.loads(next(out_root.glob("*/source_bundle.json")).read_text()) canon = json.dumps(bundle["selected_shots"], sort_keys=True, ensure_ascii=False).encode("utf-8") assert bundle["source_hash"] == hashlib.sha256(canon).hexdigest()[:16] # ────────────────────────────────────────────────────────────────────── # Phase 3 — production_pipeline_map (3 tests) # ────────────────────────────────────────────────────────────────────── def test_production_pipeline_map_has_required_touchpoints(tmp_path): out_root = tmp_path / "out" _run([ "--dry-run", "--skip-db", "--output-root", str(out_root), "--stop-after", "pipeline_map", ]) pmap = json.loads(next(out_root.glob("*/production_pipeline_map.json")).read_text()) assert isinstance(pmap["touchpoints"], list) assert 8 <= len(pmap["touchpoints"]) <= 12, "10 ± 2 touchpoints" for tp in pmap["touchpoints"]: for k in ("file", "symbol", "purpose", "observed_contract"): assert k in tp, f"touchpoint missing {k}" names = {tp["symbol"] for tp in pmap["touchpoints"]} assert "assign_bg_ids" in names assert "build_shot_background_map" in names assert "BG_ID_RE" in names assert "validate_attached_refs" in names assert "BackgroundMasterPlanStep" in names def test_pipeline_map_paths_and_symbols_exist(): """Static map 의 모든 file path 가 존재하고 symbol text 가 file 안에 있어야 함.""" from experiment_background_pipeline_slice import PRODUCTION_PIPELINE_MAP for tp in PRODUCTION_PIPELINE_MAP["touchpoints"]: path = _REPO_ROOT / tp["file"] assert path.exists(), f"missing file: {tp['file']}" text = path.read_text() assert tp["symbol"] in text, f"symbol {tp['symbol']} not found in {tp['file']}" def test_pipeline_map_report_path_existence_into_run_meta(tmp_path): out_root = tmp_path / "out" _run([ "--dry-run", "--skip-db", "--output-root", str(out_root), "--stop-after", "pipeline_map", ]) run_meta = json.loads(next(out_root.glob("*/run_meta.json")).read_text()) assert "production_pipeline_map_check" in run_meta assert run_meta["production_pipeline_map_check"]["all_paths_exist"] is True assert run_meta["production_pipeline_map_check"]["all_symbols_found"] is True # ────────────────────────────────────────────────────────────────────── # Phase 4 — LLM Raw Plan (5 tests) # ────────────────────────────────────────────────────────────────────── def test_llm_raw_plan_placeholder_when_dry_run(tmp_path): out_root = tmp_path / "out" _run([ "--dry-run", "--output-root", str(out_root), "--project-id", "6cb862d9-590c-4dce-86e6-d10c2977db19", "--episode-id", "08ad2cd3-3e96-4d84-808f-869ee628473c", "--stop-after", "llm_raw_plan", ]) plan = json.loads(next(out_root.glob("*/llm_raw_plan.json")).read_text()) assert plan["status"] == "placeholder_dry_run" for key in ("building_groups", "raw_background_intents", "floor_plan_intents", "shot_binding_intents", "prompt_payload_intents", "open_risks"): assert key in plan["plan"] def test_llm_raw_plan_schema_validates_against_jsonschema(): from experiment_background_pipeline_slice import ( LLM_RAW_PLAN_SCHEMA, _build_placeholder_raw_plan, ) import jsonschema sample = { "selected_shots": [ {"id": "s1", "scene_index": 1, "shot_index": 1, "is_selected": True, "shot_description": "x"} ], "locations": [], } placeholder = _build_placeholder_raw_plan(sample) jsonschema.validate(placeholder["plan"], LLM_RAW_PLAN_SCHEMA) def test_dry_run_does_not_import_litellm(tmp_path): """sentinel — dry-run 에서는 litellm 이 import 되지 않아야 함.""" out_root = tmp_path / "out" result = _run([ "--dry-run", "--output-root", str(out_root), "--project-id", "6cb862d9-590c-4dce-86e6-d10c2977db19", "--episode-id", "08ad2cd3-3e96-4d84-808f-869ee628473c", "--stop-after", "llm_raw_plan", "--diag-print-imports", ]) # script 가 --diag-print-imports 일 때 sys.modules 키를 stderr 로 출력 assert "litellm" not in result.stderr def test_open_risks_max_five_and_have_fallback(): from experiment_background_pipeline_slice import LLM_RAW_PLAN_SCHEMA risks_schema = LLM_RAW_PLAN_SCHEMA["properties"]["open_risks"] assert risks_schema["maxItems"] == 5 item = risks_schema["items"] assert "conservative_fallback" in item["required"] def test_llm_raw_plan_schema_state_class_enum_matches_production(): """state_class enum 이 production STATE_CLASS_ENUM 과 동일해야 함 (Codex BLOCKING1).""" from experiment_background_pipeline_slice import LLM_RAW_PLAN_SCHEMA import sys as _sys _sys.path.insert(0, str(_REPO_ROOT / "backend")) from app.core.bg_state_vocab import STATE_CLASS_ENUM intents_schema = LLM_RAW_PLAN_SCHEMA["properties"]["raw_background_intents"]["items"] enum_in_schema = set(intents_schema["properties"]["state_class"]["enum"]) assert enum_in_schema == set(STATE_CLASS_ENUM), ( f"state_class enum drift: schema={sorted(enum_in_schema)} " f"vs production={sorted(STATE_CLASS_ENUM)}" ) # ────────────────────────────────────────────────────────────────────── # Phase 5 — production_adapter_plan (6 tests) # ────────────────────────────────────────────────────────────────────── def _fake_raw_plan_minimal(): """test fixture — 3 raw intent, 2 fp intent, 4 shot binding. shot_key (composite, SOT) 만 사용. raw intent 에는 applies_to_shots 없음 — shot_binding_intents 가 SOT. fp_intent_key 와 parent_intent_ref 로 deps semantic 만 LLM. """ return { "building_groups": [ {"group_id": "G01", "member_loc_ids": ["L01"], "anchor_loc_id": "L01", "kind": "chain_bg", "rationale": "single-room scene"}, {"group_id": "G02", "member_loc_ids": ["L02"], "anchor_loc_id": "L02", "kind": "chain_bg", "rationale": "outdoor"}, ], "raw_background_intents": [ {"intent_key": "k1", "group_id": "G01", "loc_id": "L01", "space_key_hint": "main", "time_phase": "day", "state_class": "normal", "sub_location_label": None, "state_label_raw": None, "fp_intent_key": "fp1", "parent_intent_ref": None, "evidence_basis": [{"quote": "...", "source_ref": "shot.S01_Shot1", "confidence": "trusted"}]}, {"intent_key": "k2", "group_id": "G01", "loc_id": "L01", "space_key_hint": "main", "time_phase": "day", "state_class": "ransacked", "sub_location_label": None, "state_label_raw": "after_disturbance", "fp_intent_key": "fp1", "parent_intent_ref": "k1", "evidence_basis": [{"quote": "...", "source_ref": "shot.S02_Shot3", "confidence": "plausible"}]}, {"intent_key": "k3", "group_id": "G02", "loc_id": "L02", "space_key_hint": "main", "time_phase": "day", "state_class": "normal", "sub_location_label": None, "state_label_raw": None, "fp_intent_key": "fp2", "parent_intent_ref": None, "evidence_basis": [{"quote": "...", "source_ref": "shot.S03_Shot4", "confidence": "trusted"}]}, ], "floor_plan_intents": [ {"fp_intent_key": "fp1", "loc_id": "L01", "space_key_hint": "main", "rationale": "indoor"}, {"fp_intent_key": "fp2", "loc_id": "L02", "space_key_hint": "main", "rationale": "indoor"}, ], "shot_binding_intents": [ {"shot_key": "S01_Shot1", "intent_key": "k1"}, {"shot_key": "S01_Shot2", "intent_key": "k1"}, {"shot_key": "S02_Shot3", "intent_key": "k2"}, {"shot_key": "S03_Shot4", "intent_key": "k3"}, ], "prompt_payload_intents": [ {"shot_key": sk, "bg_intent_text": "background description"} for sk in ["S01_Shot1", "S01_Shot2", "S02_Shot3", "S03_Shot4"] ], "open_risks": [], } def _fake_location_profiles(): # production shape — kind (not space_kind), single_space -> space_key 'main' 강제 return { "L01": {"kind": "single_space", "allowed_space_keys": ["main"]}, "L02": {"kind": "single_space", "allowed_space_keys": ["main"]}, } def test_adapter_assigns_bg_ids_with_production_helper(): from experiment_background_pipeline_slice import _build_adapter_plan plan = _fake_raw_plan_minimal() adapter = _build_adapter_plan(plan, _fake_location_profiles()) catalog = adapter["background_catalog"] # Dict[bg_id, entry] assert isinstance(catalog, dict) import re bg_re = re.compile(r"^L\d{2,3}B\d{2,3}$") for bid in catalog.keys(): assert bg_re.match(bid), f"invalid bg_id: {bid}" # 3 intents -> 3 catalog entries (no semantic dedup expected in this fake) assert len(catalog) == 3 def test_adapter_assign_bg_ids_deterministic_rerun(): from experiment_background_pipeline_slice import _build_adapter_plan plan = _fake_raw_plan_minimal() p1 = _build_adapter_plan(plan, _fake_location_profiles()) p2 = _build_adapter_plan(plan, _fake_location_profiles()) assert p1["bg_catalog_hash"] == p2["bg_catalog_hash"] assert p1["shot_binding_hash"] == p2["shot_binding_hash"] def test_adapter_shot_background_map_n_to_1_invariant(): """동일 shot_key 이 두 intent_key 에 매핑되면 ShotBindingError -> validation_failed.""" from experiment_background_pipeline_slice import _build_adapter_plan plan = _fake_raw_plan_minimal() plan["shot_binding_intents"].append({"shot_key": "S01_Shot1", "intent_key": "k2"}) import pytest with pytest.raises(Exception): # ShotBindingError or wrapped _build_adapter_plan(plan, _fake_location_profiles()) def test_adapter_gen_order_topo_resolves_parent_intent_ref(): """k2 는 parent_intent_ref=k1 -> bg_id(k2) 의 depends_on_bg 에 bg_id(k1) 들어가야. gen_order 는 k1 의 bg_id 가 k2 의 bg_id 보다 먼저.""" from experiment_background_pipeline_slice import _build_adapter_plan plan = _fake_raw_plan_minimal() adapter = _build_adapter_plan(plan, _fake_location_profiles()) intent_to_bg = adapter["intent_key_to_bg_id"] k1_bg = intent_to_bg["k1"] k2_bg = intent_to_bg["k2"] k2_entry = adapter["background_catalog"][k2_bg] assert k1_bg in k2_entry["depends_on_bg"] order = adapter["gen_order"] assert order.index(k1_bg) < order.index(k2_bg) def test_adapter_dedup_parent_conflict_fail_fast(): """Codex v3 BLOCKING 2 + v4 BLOCKING 1 — 같은 semantic_key 로 dedup 된 intent 들의 parent semantics 가 inconsistent (parent_bg vs different parent_bg, OR None vs some parent_bg) 면 dedup_parent_conflict 로 fail-fast. """ from experiment_background_pipeline_slice import _build_adapter_plan import pytest # Case A: parent_bg vs different parent_bg plan_a = _fake_raw_plan_minimal() plan_a["raw_background_intents"].append({ "intent_key": "k_dup", "group_id": "G01", "loc_id": "L01", "space_key_hint": "main", "time_phase": "day", "state_class": "ransacked", "sub_location_label": None, "state_label_raw": "after_disturbance", "fp_intent_key": "fp1", "parent_intent_ref": "k3", "evidence_basis": [{"quote": "...", "source_ref": "shot.S_dup", "confidence": "weak"}], }) plan_a["shot_binding_intents"].append({"shot_key": "S_dup_Shot1", "intent_key": "k_dup"}) with pytest.raises(ValueError, match="dedup_parent_conflict"): _build_adapter_plan(plan_a, _fake_location_profiles()) # Case B: None vs some_bg (v4 BLOCKING 1) plan_b = _fake_raw_plan_minimal() plan_b["raw_background_intents"].append({ "intent_key": "k_dup2", "group_id": "G01", "loc_id": "L01", "space_key_hint": "main", "time_phase": "day", "state_class": "ransacked", "sub_location_label": None, "state_label_raw": "after_disturbance", "fp_intent_key": "fp1", "parent_intent_ref": None, "evidence_basis": [{"quote": "...", "source_ref": "shot.S_dup2", "confidence": "weak"}], }) plan_b["shot_binding_intents"].append({"shot_key": "S_dup2_Shot1", "intent_key": "k_dup2"}) with pytest.raises(ValueError, match="dedup_parent_conflict"): _build_adapter_plan(plan_b, _fake_location_profiles()) def test_adapter_dedup_parent_self_cycle_fail_fast(): """self-cycle (intent 가 자기 자신을 parent 로 가리킴) fail-fast.""" from experiment_background_pipeline_slice import _build_adapter_plan plan = _fake_raw_plan_minimal() plan["raw_background_intents"][0]["parent_intent_ref"] = "k1" import pytest with pytest.raises(ValueError, match="dedup_parent_self_cycle"): _build_adapter_plan(plan, _fake_location_profiles()) # ────────────────────────────────────────────────────────────────────── # Phase 6 — generation_payload_dry_run + reference_input_plan (3 tests) # ────────────────────────────────────────────────────────────────────── def test_generation_payload_has_minimum_required_fields(): from experiment_background_pipeline_slice import ( _build_generation_payload, _build_adapter_plan, ) plan = _fake_raw_plan_minimal() adapter = _build_adapter_plan(plan, _fake_location_profiles()) payload = _build_generation_payload(plan, adapter) assert len(payload["per_shot"]) == 4 for entry in payload["per_shot"]: for k in ("shot_key", "bg_id", "loc_id", "bg_intent_text", "expected_attached_meta", "expected_reference_phrase_kinds", "expected_label_placeholder"): assert k in entry assert entry["expected_attached_meta"] == [{"kind": "background", "id": entry["bg_id"]}] assert entry["expected_reference_phrase_kinds"] == ["background"] assert isinstance(payload["unmapped_fields"], list) def test_reference_input_plan_covers_all_selected_shots(): from experiment_background_pipeline_slice import ( _build_reference_input_plan, _build_adapter_plan, ) plan = _fake_raw_plan_minimal() adapter = _build_adapter_plan(plan, _fake_location_profiles()) rip = _build_reference_input_plan(plan, adapter) shot_keys = {e["shot_key"] for e in rip["entries"]} assert shot_keys == {"S01_Shot1", "S01_Shot2", "S02_Shot3", "S03_Shot4"} for e in rip["entries"]: assert e["ref_contract"] == {"kind": "background", "id": e["bg_id"], "policy": "required"} def test_ref_contract_dry_run_pass_for_all_shots(): """validate_attached_refs 가 모든 shot 에 대해 통과해야 함.""" from experiment_background_pipeline_slice import ( _build_adapter_plan, _build_reference_input_plan, _ref_contract_dry_run, ) plan = _fake_raw_plan_minimal() adapter = _build_adapter_plan(plan, _fake_location_profiles()) rip = _build_reference_input_plan(plan, adapter) result = _ref_contract_dry_run(rip, adapter) assert result["all_pass"] is True assert len(result["failures"]) == 0 # ────────────────────────────────────────────────────────────────────── # Phase 7 — pipeline_compatibility_report (3 tests) # ────────────────────────────────────────────────────────────────────── def test_compatibility_report_aggregates_nine_invariants(): """fake bundle 의 selected_shots 가 fake_raw_plan_minimal 의 shot_key 와 일치.""" import hashlib shot_keys = ["S01_Shot1", "S01_Shot2", "S02_Shot3", "S03_Shot4"] selected = [{"row_id": f"u{i}", "shot_key": sk, "scene_index": int(sk[1:3]), "shot_index": int(sk.split("Shot")[1]), "is_selected": True, "shot_description": ""} for i, sk in enumerate(shot_keys)] canon = json.dumps(selected, sort_keys=True, ensure_ascii=False).encode("utf-8") bundle = {"selected_shot_count": len(selected), "selected_shots": selected, "source_hash": hashlib.sha256(canon).hexdigest()[:16]} from experiment_background_pipeline_slice import ( _build_compatibility_report, _build_adapter_plan, _build_reference_input_plan, _ref_contract_dry_run, ) plan = _fake_raw_plan_minimal() adapter = _build_adapter_plan(plan, _fake_location_profiles()) rip = _build_reference_input_plan(plan, adapter) ref_dry = _ref_contract_dry_run(rip, adapter) report = _build_compatibility_report( bundle=bundle, raw_plan=plan, adapter=adapter, rip=rip, ref_dry=ref_dry, pmap_check={"all_paths_exist": True, "all_symbols_found": True, "missing": []}, production_diff_empty=True, db_write_count=0, image_import_seen=False, ) for inv in ( "production_diff_zero", "db_write_zero", "image_call_zero", "source_verbatim", "all_selected_shots_covered", "bg_id_assigned_by_production_helper", "shot_background_map_n_to_1", "reference_chain_acyclic", "ref_contract_dry_run_pass", ): assert inv in report["invariants"] def test_compatibility_report_fails_when_unmatched_shot(): import hashlib selected = [{"row_id": f"u{i}", "shot_key": f"S99_Shot{i}", "scene_index": 99, "shot_index": i, "is_selected": True} for i in range(1, 6)] canon = json.dumps(selected, sort_keys=True, ensure_ascii=False).encode("utf-8") bundle = {"selected_shot_count": 5, "selected_shots": selected, "source_hash": hashlib.sha256(canon).hexdigest()[:16]} from experiment_background_pipeline_slice import ( _build_compatibility_report, _build_adapter_plan, _build_reference_input_plan, _ref_contract_dry_run, ) plan = _fake_raw_plan_minimal() adapter = _build_adapter_plan(plan, _fake_location_profiles()) rip = _build_reference_input_plan(plan, adapter) ref_dry = _ref_contract_dry_run(rip, adapter) report = _build_compatibility_report( bundle=bundle, raw_plan=plan, adapter=adapter, rip=rip, ref_dry=ref_dry, pmap_check={"all_paths_exist": True, "all_symbols_found": True, "missing": []}, production_diff_empty=True, db_write_count=0, image_import_seen=False, ) assert report["invariants"]["all_selected_shots_covered"]["pass"] is False assert report["all_pass"] is False def test_runner_exits_1_on_any_invariant_fail(tmp_path): out_root = tmp_path / "out" # 강제 fail mode — DB 가 없는 가짜 project_id result = _run([ "--dry-run", "--output-root", str(out_root), "--project-id", "ZZZZZZZZ-0000-0000-0000-000000000000", "--episode-id", "ZZZZZZZZ-0000-0000-0000-000000000000", ]) assert result.returncode == 1 rm = json.loads(next(out_root.glob("*/run_meta.json")).read_text()) assert rm["run_status"] == "validation_failed" assert rm["exit_code"] == 1 assert len(rm["failed_invariants"]) > 0 # ────────────────────────────────────────────────────────────────────── # Phase 8 — HTML diagnostic (2 tests) # ────────────────────────────────────────────────────────────────────── def test_index_html_shows_key_metrics(tmp_path): out_root = tmp_path / "out" _run([ "--dry-run", "--output-root", str(out_root), "--project-id", "6cb862d9-590c-4dce-86e6-d10c2977db19", "--episode-id", "08ad2cd3-3e96-4d84-808f-869ee628473c", ]) html = next(out_root.glob("*/index.html")).read_text() # 첫 화면 metric for needle in ("unit count", "shot coverage", "reference chain", "validation", "payload sample"): assert needle in html.lower(), f"index.html missing '{needle}'" # raw JSON 은 접혀있어야 함 —
사용 assert " 0 # script source 안에 sample-specific literal 0 확인 (Codex 추가 list 반영) src = (_REPO_ROOT / "backend/scripts/experiment_background_pipeline_slice.py").read_text() for banned in ("옥탑방", "L05", "rooftop", "수리영", "민숙", "석창포"): assert banned not in src, f"banned literal '{banned}' in script source" def test_static_guard_no_db_write_method_calls(): """script source 안에 session 의 write method 호출 literal 0건. DB read-only 보장 — IMPORTANT 3 (Codex).""" src = (_REPO_ROOT / "backend/scripts/experiment_background_pipeline_slice.py").read_text() for banned in ("session.commit", "session.add", "session.delete", "session.merge", "session.flush", ".commit(", ".add("): # comment 안에서 written-about 은 허용 — 실제 callable 일 때만 잡으려면 AST 가 더 정확. # 최소 보호 단계로 literal 검사. if banned in src: for line in src.splitlines(): if banned in line and not line.lstrip().startswith("#"): raise AssertionError( f"banned write call '{banned}' in non-comment line: {line.strip()}" ) def test_no_banned_image_modules_imported_at_runtime(tmp_path): """sentinel — 어떤 모드에서도 image 모듈 import 가 일어나면 안 됨.""" out_root = tmp_path / "out" result = _run([ "--dry-run", "--output-root", str(out_root), "--project-id", "6cb862d9-590c-4dce-86e6-d10c2977db19", "--episode-id", "08ad2cd3-3e96-4d84-808f-869ee628473c", "--diag-print-imports", ]) banned = ("gemini_image_client", "fal_angle_helpers", "gemini_i2i_editor") for b in banned: assert b not in result.stderr, f"banned module imported: {b}" def test_production_diff_empty_invariant_recorded(tmp_path): """compatibility report 에 production_diff_zero invariant 가 PASS 로 기록되어야 함 (이 PR 가 production code 를 만지지 않으므로).""" out_root = tmp_path / "out" _run([ "--dry-run", "--output-root", str(out_root), "--project-id", "6cb862d9-590c-4dce-86e6-d10c2977db19", "--episode-id", "08ad2cd3-3e96-4d84-808f-869ee628473c", ]) report = json.loads(next(out_root.glob("*/pipeline_compatibility_report.json")).read_text()) assert report["invariants"]["production_diff_zero"]["pass"] is True # ────────────────────────────────────────────────────────────────────── # W2 narrow patch tests (3) # ────────────────────────────────────────────────────────────────────── def test_w2_locations_for_llm_has_no_db_uuid(tmp_path): """W2 BLOCKING — LLM 에 노출되는 location view 는 UUID 가 없어야 하고 loc_id 는 short_id 그대로여야 함.""" out_root = tmp_path / "out" _run([ "--dry-run", "--output-root", str(out_root), "--project-id", "6cb862d9-590c-4dce-86e6-d10c2977db19", "--episode-id", "08ad2cd3-3e96-4d84-808f-869ee628473c", "--stop-after", "source_bundle", ]) bundle = json.loads(next(out_root.glob("*/source_bundle.json")).read_text()) assert "locations_for_llm" in bundle assert isinstance(bundle["locations_for_llm"], list) assert len(bundle["locations_for_llm"]) > 0 for entry in bundle["locations_for_llm"]: # loc_id 는 short_id (예: 'L01'..'L99') assert entry["loc_id"].startswith("L"), entry # production short_id 패턴 (L## or L###) assert 3 <= len(entry["loc_id"]) <= 5, entry # UUID 같은 키 없어야 함 assert "id" not in entry, f"DB UUID leaked into LLM view: {entry}" # 필수 필드 for key in ("name", "description", "space_profile"): assert key in entry, f"missing {key} in {entry}" # space_profile production shape assert entry["space_profile"]["kind"] in ("single_space", "multi_space"), entry def test_w2_validate_loc_id_membership_detects_db_uuid(): """W2 BLOCKING — LLM 이 loc_id 로 UUID 를 emit 한 경우 errors 반환.""" from experiment_background_pipeline_slice import _validate_loc_id_membership location_profiles = { "L01": {"kind": "single_space", "allowed_space_keys": ["main"]}, "L02": {"kind": "single_space", "allowed_space_keys": ["main"]}, } bad_plan = { "raw_background_intents": [ {"intent_key": "k1", "loc_id": "L01"}, {"intent_key": "k2", "loc_id": "04d6fa34-2a61-4f32-bccd-9e5dd0dd61b6"}, ], "floor_plan_intents": [ {"fp_intent_key": "fp1", "loc_id": "L99"}, ], "building_groups": [ {"group_id": "G01", "anchor_loc_id": "L01", "member_loc_ids": ["L01", "L77"]}, ], } errors = _validate_loc_id_membership(bad_plan, location_profiles) assert len(errors) >= 3 # k2 UUID + fp1 L99 + member L77 joined = "\n".join(errors) assert "04d6fa34" in joined assert "L99" in joined assert "L77" in joined good_plan = { "raw_background_intents": [{"intent_key": "k1", "loc_id": "L01"}], "floor_plan_intents": [{"fp_intent_key": "fp1", "loc_id": "L02"}], "building_groups": [ {"group_id": "G01", "anchor_loc_id": "L01", "member_loc_ids": ["L01"]}, ], } assert _validate_loc_id_membership(good_plan, location_profiles) == [] def test_w3_validate_semantic_key_parents_detects_self_cycle(): """W3 BLOCKING — production semantic_key 가 같은 두 intent 사이 parent 관계 있으면 parent_same_semantic_key fail (production assign_bg_ids 도달 전 잡음).""" from experiment_background_pipeline_slice import _validate_semantic_key_parents location_profiles = {"L01": {"kind": "single_space", "allowed_space_keys": ["main"]}} # 두 intent 가 같은 4-tuple → 같은 semantic_key. parent 관계 시도. bad_plan = { "raw_background_intents": [ {"intent_key": "i_a", "loc_id": "L01", "space_key_hint": "main", "time_phase": "day", "state_class": "normal", "sub_location_label": "sub_a", "state_label_raw": None, "parent_intent_ref": None}, {"intent_key": "i_b", "loc_id": "L01", "space_key_hint": "main", "time_phase": "day", "state_class": "normal", "sub_location_label": "sub_b", "state_label_raw": None, "parent_intent_ref": "i_a"}, # 같은 sem_key 면 self-cycle 후보 ], } errs = _validate_semantic_key_parents(bad_plan, location_profiles) assert any("parent_same_semantic_key" in e for e in errs), errs # 다른 state_class 면 다른 semantic_key — parent OK good_plan = { "raw_background_intents": [ {"intent_key": "i_a", "loc_id": "L01", "space_key_hint": "main", "time_phase": "day", "state_class": "normal", "sub_location_label": None, "state_label_raw": None, "parent_intent_ref": None}, {"intent_key": "i_b", "loc_id": "L01", "space_key_hint": "main", "time_phase": "day", "state_class": "ransacked", "sub_location_label": None, "state_label_raw": None, "parent_intent_ref": "i_a"}, ], } assert _validate_semantic_key_parents(good_plan, location_profiles) == [] def test_w3_validate_space_key_hint_membership(): """W3 — single_space loc 는 space_key_hint 'main' 강제. multi_space 는 allowed_space_keys 안.""" from experiment_background_pipeline_slice import _validate_space_key_hint_membership location_profiles = { "L01": {"kind": "single_space", "allowed_space_keys": ["main"]}, "L02": {"kind": "multi_space", "allowed_space_keys": ["kitchen", "stairs"]}, } # single_space + 'main' OK / 'kitchen' fail bad_plan = { "raw_background_intents": [ {"intent_key": "i1", "loc_id": "L01", "space_key_hint": "kitchen"}, {"intent_key": "i2", "loc_id": "L02", "space_key_hint": "not_allowed_space"}, ], } errs = _validate_space_key_hint_membership(bad_plan, location_profiles) assert len(errs) == 2 assert any("L01" in e for e in errs) assert any("L02" in e and "not_allowed_space" in e for e in errs) good_plan = { "raw_background_intents": [ {"intent_key": "i1", "loc_id": "L01", "space_key_hint": "main"}, {"intent_key": "i2", "loc_id": "L02", "space_key_hint": "kitchen"}, ], } assert _validate_space_key_hint_membership(good_plan, location_profiles) == [] def test_w2_dependent_invariants_skipped_on_adapter_failure(): """W2 IMPORTANT 1 — adapter build failed (empty catalog) 시 dependent invariant 들이 PASS 로 위장되지 않고 skipped_due_to_adapter_failure 로 정직히 표시.""" from experiment_background_pipeline_slice import _build_compatibility_report import hashlib as _h failed_adapter = { "background_catalog": {}, "shot_background_map": {}, "gen_order": [], "intent_key_to_bg_id": {}, "error": "simulated adapter build failure", } sel = [{"row_id": "u1", "shot_key": "S01_Shot1", "scene_index": 1, "shot_index": 1, "is_selected": True}] canon = json.dumps(sel, sort_keys=True, ensure_ascii=False).encode("utf-8") bundle = {"selected_shot_count": 1, "selected_shots": sel, "source_hash": _h.sha256(canon).hexdigest()[:16]} rip = {"entries": []} ref_dry = {"all_pass": False, "failures": []} report = _build_compatibility_report( bundle=bundle, raw_plan={}, adapter=failed_adapter, rip=rip, ref_dry=ref_dry, pmap_check={"all_paths_exist": True, "all_symbols_found": True, "missing": []}, production_diff_empty=True, db_write_count=0, image_import_seen=False, ) for dep in ("shot_background_map_n_to_1", "reference_chain_acyclic", "ref_contract_dry_run_pass"): assert report["invariants"][dep]["pass"] is False, dep assert report["invariants"][dep]["detail"] == "skipped_due_to_adapter_failure", dep # ────────────────────────────────────────────────────────────────────── # W4 narrow tests — background_render_plan + reference_chain_plan # ────────────────────────────────────────────────────────────────────── def _fake_w3_adapter_for_w4(): """Synthetic adapter mimicking a W3 success run for W4 derivation tests.""" # 4 bg across 2 loc groups. Tests master_base selection, state/time/special variants. return { "schema_version": 1, "background_catalog": { "L01B01": { "bg_id": "L01B01", "loc_id": "L01", "space_key": "main", "time_phase": "day", "state_class": "normal", "depends_on_fp": [], "depends_on_bg": [], "applies_to_shots": ["S01_Shot1", "S01_Shot2"], "semantic_key": "L01|main|day|normal", }, "L01B02": { "bg_id": "L01B02", "loc_id": "L01", "space_key": "main", "time_phase": "day", "state_class": "ransacked", "depends_on_fp": [], "depends_on_bg": ["L01B01"], "applies_to_shots": ["S02_Shot1"], "semantic_key": "L01|main|day|ransacked", }, "L01B03": { "bg_id": "L01B03", "loc_id": "L01", "space_key": "main", "time_phase": "night", "state_class": "normal", "depends_on_fp": [], "depends_on_bg": [], "applies_to_shots": ["S03_Shot1"], "semantic_key": "L01|main|night|normal", }, "L02B01": { "bg_id": "L02B01", "loc_id": "L02", "space_key": "main", "time_phase": "day", "state_class": "dream_or_vision_state", "depends_on_fp": [], "depends_on_bg": [], "applies_to_shots": ["S04_Shot1"], "semantic_key": "L02|main|day|dream_or_vision_state", }, }, "shot_background_map": { "S01_Shot1": "L01B01", "S01_Shot2": "L01B01", "S02_Shot1": "L01B02", "S03_Shot1": "L01B03", "S04_Shot1": "L02B01", }, "gen_order": ["L01B01", "L01B03", "L02B01", "L01B02"], "bg_catalog_hash": "fake", "shot_binding_hash": "fake", } def _fake_w3_raw_plan_for_w4(): return { "prompt_payload_intents": [ {"shot_key": "S01_Shot1", "bg_intent_text": "L01 day room"}, {"shot_key": "S02_Shot1", "bg_intent_text": "L01 after disturbance"}, {"shot_key": "S03_Shot1", "bg_intent_text": "L01 night room"}, {"shot_key": "S04_Shot1", "bg_intent_text": "L02 vision scene"}, ], } def _fake_w3_rip_for_w4(): return {"entries": [ {"shot_key": "S01_Shot1"}, {"shot_key": "S01_Shot2"}, {"shot_key": "S02_Shot1"}, {"shot_key": "S03_Shot1"}, {"shot_key": "S04_Shot1"}, ]} def test_w4_classify_render_roles_master_and_variants(): """master_base = state_class==normal w/ minimal deps. variants get correct roles.""" from experiment_background_pipeline_slice import _classify_render_roles roles = _classify_render_roles(_fake_w3_adapter_for_w4()) # L01 group: L01B01 (normal, no deps) → master; L01B02 (ransacked) → state_variant; # L01B03 (normal, different time) → time_variant. assert roles["L01B01"]["render_role"] == "master_base" assert roles["L01B01"]["base_bg_id"] is None assert roles["L01B02"]["render_role"] == "state_variant" assert roles["L01B02"]["base_bg_id"] == "L01B01" # via existing depends_on_bg assert "ransacked" in roles["L01B02"]["variant_delta"] assert roles["L01B03"]["render_role"] == "time_variant" assert roles["L01B03"]["base_bg_id"] == "L01B01" # fallback to group master # L02 group: only one entry, special state → master_base (sole member is master). assert roles["L02B01"]["render_role"] == "master_base" def test_w4_build_render_plan_covers_all_bg_and_orders_by_gen_order(): from experiment_background_pipeline_slice import _build_render_plan adapter = _fake_w3_adapter_for_w4() plan = _build_render_plan(adapter, _fake_w3_raw_plan_for_w4(), {}, _fake_w3_rip_for_w4()) assert plan["schema_version"] == 1 assert plan["stage"] == "w4_render_plan" assert set(plan["render_units"].keys()) == set(adapter["background_catalog"].keys()) # render_order = adapter.gen_order minus fp_ entries assert plan["render_order"] == ["L01B01", "L01B03", "L02B01", "L01B02"] # render_order_index monotonic per render_order for i, bg in enumerate(plan["render_order"]): assert plan["render_units"][bg]["render_order_index"] == i # master_base + variant counts: # group L01: L01B01 (master), L01B02 (state_variant), L01B03 (time_variant) → 1 master # group L02: L02B01 (only entry, special state but sole member) → master_base assert plan["master_base_count"] == 2 assert plan["variant_count"] == 2 def test_w4_build_render_payload_dry_run_has_chain_bg_placeholder(): from experiment_background_pipeline_slice import ( _build_render_plan, _build_render_payload_dry_run_w4, ) adapter = _fake_w3_adapter_for_w4() plan = _build_render_plan(adapter, _fake_w3_raw_plan_for_w4(), {}, _fake_w3_rip_for_w4()) payload = _build_render_payload_dry_run_w4(plan, adapter) assert len(payload["per_bg"]) == 4 for entry in payload["per_bg"]: for k in ("bg_id", "loc_id", "render_role", "base_bg_id", "reference_inputs", "prompt_payload_summary", "applies_to_shots", "expected_chain_bg_asset"): assert k in entry, k assert entry["expected_chain_bg_asset"]["asset_type"] == "chain_bg" assert entry["expected_chain_bg_asset"]["variant_type"] == entry["bg_id"] assert isinstance(payload["unmapped_fields"], list) assert "image_bytes" in payload["unmapped_fields"] def test_w4_reference_chain_plan_is_topo_and_acyclic(): from experiment_background_pipeline_slice import ( _build_render_plan, _build_reference_chain_plan, ) adapter = _fake_w3_adapter_for_w4() plan = _build_render_plan(adapter, _fake_w3_raw_plan_for_w4(), {}, _fake_w3_rip_for_w4()) chain = _build_reference_chain_plan(plan, adapter) bg_ids = [c["bg_id"] for c in chain["chain"]] assert len(set(bg_ids)) == len(bg_ids) # acyclic — unique # L01B02 has base_bg_id=L01B01 → L01B01 must precede in chain pos = {b: i for i, b in enumerate(bg_ids)} assert pos["L01B01"] < pos["L01B02"] assert chain["errors"] == [] def test_w4_render_plan_compatibility_report_all_pass(): from experiment_background_pipeline_slice import ( _build_render_plan, _build_render_payload_dry_run_w4, _build_reference_chain_plan, _build_render_plan_compatibility_report, ) adapter = _fake_w3_adapter_for_w4() plan = _build_render_plan(adapter, _fake_w3_raw_plan_for_w4(), {}, _fake_w3_rip_for_w4()) payload = _build_render_payload_dry_run_w4(plan, adapter) chain = _build_reference_chain_plan(plan, adapter) report = _build_render_plan_compatibility_report( render_plan=plan, render_payload=payload, chain_plan=chain, rip=_fake_w3_rip_for_w4(), adapter=adapter, production_diff_empty=True, db_write_count=0, image_import_seen=False, prev_run_id="fake_w3_run", missing_inputs=[], ) for k in ("w3_inputs_present", "all_bg_covered_in_render_plan", "master_base_exists_per_loc_group", "variants_reference_earlier_bg", "render_chain_acyclic", "production_diff_zero", "db_write_zero", "image_call_zero", "no_human_decision_field", "prompt_payload_coverage_carried"): assert k in report["invariants"], k assert report["invariants"][k]["pass"] is True, (k, report["invariants"][k]) # W4b — coverage detail must equal expected cov = report["invariants"]["prompt_payload_coverage_carried"]["detail"] assert cov["expected_count"] == len(adapter["shot_background_map"]) assert cov["expected_count"] == cov["actual_count"] assert cov["missing_in_rip"] == [] and cov["extra_in_rip"] == [] assert report["all_pass"] is True def test_w4b_coverage_carried_fails_when_rip_mismatches_adapter(): """W4b BLOCKING 1 — expected (adapter shot_background_map) vs actual (rip) set mismatch 시 prompt_payload_coverage_carried FAIL.""" from experiment_background_pipeline_slice import ( _build_render_plan, _build_render_payload_dry_run_w4, _build_reference_chain_plan, _build_render_plan_compatibility_report, ) adapter = _fake_w3_adapter_for_w4() plan = _build_render_plan(adapter, _fake_w3_raw_plan_for_w4(), {}, _fake_w3_rip_for_w4()) payload = _build_render_payload_dry_run_w4(plan, adapter) chain = _build_reference_chain_plan(plan, adapter) # Case A: rip missing one shot short_rip = {"entries": [{"shot_key": k} for k in ["S01_Shot1", "S01_Shot2", "S02_Shot1", "S03_Shot1"]]} report_a = _build_render_plan_compatibility_report( render_plan=plan, render_payload=payload, chain_plan=chain, rip=short_rip, adapter=adapter, production_diff_empty=True, db_write_count=0, image_import_seen=False, prev_run_id="fake", missing_inputs=[], ) assert report_a["invariants"]["prompt_payload_coverage_carried"]["pass"] is False cov_a = report_a["invariants"]["prompt_payload_coverage_carried"]["detail"] assert "S04_Shot1" in cov_a["missing_in_rip"] # Case B: rip contains shot not in adapter extra_rip = {"entries": list(_fake_w3_rip_for_w4()["entries"]) + [{"shot_key": "S99_ShotZ"}]} report_b = _build_render_plan_compatibility_report( render_plan=plan, render_payload=payload, chain_plan=chain, rip=extra_rip, adapter=adapter, production_diff_empty=True, db_write_count=0, image_import_seen=False, prev_run_id="fake", missing_inputs=[], ) assert report_b["invariants"]["prompt_payload_coverage_carried"]["pass"] is False cov_b = report_b["invariants"]["prompt_payload_coverage_carried"]["detail"] assert "S99_ShotZ" in cov_b["extra_in_rip"] def test_w4_cli_derive_from_w3_run(tmp_path): """CLI integration — feed a fake W3 run dir and verify W4 outputs.""" # Build a minimal W3-shaped run dir prev = tmp_path / "fake_w3" prev.mkdir() (prev / "production_adapter_plan.json").write_text( json.dumps(_fake_w3_adapter_for_w4(), ensure_ascii=False) ) (prev / "llm_raw_plan.json").write_text( json.dumps({"plan": _fake_w3_raw_plan_for_w4()}, ensure_ascii=False) ) (prev / "generation_payload_dry_run.json").write_text( json.dumps({"per_shot": [], "unmapped_fields": []}, ensure_ascii=False) ) (prev / "reference_input_plan.json").write_text( json.dumps(_fake_w3_rip_for_w4(), ensure_ascii=False) ) out_root = tmp_path / "w4_out" result = _run([ "--derive-render-plan-from", str(prev), "--output-root", str(out_root), ]) # W4b — fake fixture 가 정확히 매칭하므로 exit 0 강제 (false-pass 회피). assert result.returncode == 0, result.stderr run_dirs = sorted(out_root.glob("*/")) assert len(run_dirs) == 1 rd = run_dirs[0] for fname in ("background_render_plan.json", "background_render_payload_dry_run.json", "reference_chain_plan.json", "render_plan_compatibility_report.json", "index.html", "run_meta.json"): assert (rd / fname).exists(), fname rm = json.loads((rd / "run_meta.json").read_text()) assert rm["stage"] == "w4_render_plan" assert rm["derived_from"] == "fake_w3" assert rm["run_status"] == "succeeded" assert rm["exit_code"] == 0 assert rm["failed_invariants"] == [] # ────────────────────────────────────────────────────────────────────── # W5 narrow tests — background_generation_map + render_batches # ────────────────────────────────────────────────────────────────────── def _fake_w4_render_plan_for_w5(): """4 bg across 2 loc groups, matches W4 _classify_render_roles fixture output.""" return { "schema_version": 1, "stage": "w4_render_plan", "render_units": { "L01B01": { # master_base of L01 "bg_id": "L01B01", "loc_id": "L01", "space_key": "main", "time_phase": "day", "state_class": "normal", "render_role": "master_base", "base_bg_id": None, "render_order_index": 0, "reference_inputs": [], "applies_to_shots": ["S01_Shot1", "S01_Shot2"], "prompt_payload_summary": "L01 day room", "variant_delta": "base", }, "L01B02": { # state_variant of L01B01 "bg_id": "L01B02", "loc_id": "L01", "space_key": "main", "time_phase": "day", "state_class": "ransacked", "render_role": "state_variant", "base_bg_id": "L01B01", "render_order_index": 3, "reference_inputs": [{"kind": "background", "id": "L01B01"}], "applies_to_shots": ["S02_Shot1"], "prompt_payload_summary": "L01 ransacked", "variant_delta": "state:normal->ransacked", }, "L01B03": { # time_variant of L01B01 "bg_id": "L01B03", "loc_id": "L01", "space_key": "main", "time_phase": "night", "state_class": "normal", "render_role": "time_variant", "base_bg_id": "L01B01", "render_order_index": 1, "reference_inputs": [{"kind": "background", "id": "L01B01"}], "applies_to_shots": ["S03_Shot1"], "prompt_payload_summary": "L01 night", "variant_delta": "time:day->night", }, "L02B01": { # vision/special, sole member of L02 group → master_base "bg_id": "L02B01", "loc_id": "L02", "space_key": "main", "time_phase": "day", "state_class": "dream_or_vision_state", "render_role": "master_base", "base_bg_id": None, "render_order_index": 2, "reference_inputs": [], "applies_to_shots": ["S04_Shot1"], "prompt_payload_summary": "L02 vision", "variant_delta": "base", }, }, "render_order": ["L01B01", "L01B03", "L02B01", "L01B02"], "master_base_count": 2, "variant_count": 2, "unmapped_to_production_fields": [], } def _fake_w4_chain_plan_for_w5(): return { "chain": [ {"bg_id": "L01B01", "render_role": "master_base", "base_bg_id": None, "reference_inputs": []}, {"bg_id": "L01B03", "render_role": "time_variant", "base_bg_id": "L01B01", "reference_inputs": [{"kind": "background", "id": "L01B01"}]}, {"bg_id": "L02B01", "render_role": "master_base", "base_bg_id": None, "reference_inputs": []}, {"bg_id": "L01B02", "render_role": "state_variant", "base_bg_id": "L01B01", "reference_inputs": [{"kind": "background", "id": "L01B01"}]}, ], "errors": [], } def test_w5_build_generation_map_groups_structure(): from experiment_background_pipeline_slice import _build_generation_map plan = _fake_w4_render_plan_for_w5() chain = _fake_w4_chain_plan_for_w5() gen_map = _build_generation_map(plan, chain) assert gen_map["stage"] == "w5_generation_map" # 2 groups: L01|main and L02|main groups = {g["group_id"]: g for g in gen_map["groups"]} assert set(groups.keys()) == {"L01|main", "L02|main"} l01 = groups["L01|main"] assert l01["base_bg_id"] == "L01B01" assert set(l01["member_bg_ids"]) == {"L01B01", "L01B02", "L01B03"} assert set(l01["variant_bg_ids"]) == {"L01B02", "L01B03"} assert set(l01["shot_keys"]) == {"S01_Shot1", "S01_Shot2", "S02_Shot1", "S03_Shot1"} # generation_order: base first assert l01["generation_order"][0] == "L01B01" # L02 group: sole master_base l02 = groups["L02|main"] assert l02["base_bg_id"] == "L02B01" assert l02["variant_bg_ids"] == [] def test_w5_render_mode_policy(): from experiment_background_pipeline_slice import _build_generation_map plan = _fake_w4_render_plan_for_w5() gen_map = _build_generation_map(plan, _fake_w4_chain_plan_for_w5()) imgs = gen_map["background_images"] # master_base → independent assert imgs["L01B01"]["render_mode"] == "independent_text_to_image" assert imgs["L01B01"]["primary_reference_bg_id"] is None # state_variant w/ base → derive_from_base_reference assert imgs["L01B02"]["render_mode"] == "derive_from_base_reference" assert imgs["L01B02"]["primary_reference_bg_id"] == "L01B01" # time_variant w/ base → derive_from_base_reference assert imgs["L01B03"]["render_mode"] == "derive_from_base_reference" assert imgs["L01B03"]["primary_reference_bg_id"] == "L01B01" # L02B01 is master_base (sole member) → independent (not weak_reference because it's master) assert imgs["L02B01"]["render_mode"] == "independent_text_to_image" def test_w5_render_batches_topological_independent_first(): from experiment_background_pipeline_slice import _build_generation_map plan = _fake_w4_render_plan_for_w5() gen_map = _build_generation_map(plan, _fake_w4_chain_plan_for_w5()) batches = gen_map["render_batches"] # batch 0 = independent bg (L01B01, L02B01) assert set(batches[0]) == {"L01B01", "L02B01"} # variants in later batch bg_to_batch = {bg: i for i, batch in enumerate(batches) for bg in batch} assert bg_to_batch["L01B02"] > bg_to_batch["L01B01"] assert bg_to_batch["L01B03"] > bg_to_batch["L01B01"] def test_w5_references_in_earlier_batch_invariant(): from experiment_background_pipeline_slice import ( _build_generation_map, _build_generation_map_compatibility_report, ) plan = _fake_w4_render_plan_for_w5() gen_map = _build_generation_map(plan, _fake_w4_chain_plan_for_w5()) report = _build_generation_map_compatibility_report( gen_map=gen_map, render_plan=plan, w4_report={}, production_diff_empty=True, db_write_count=0, image_import_seen=False, prev_run_id="fake_w4", missing_inputs=[], ) inv = report["invariants"] assert inv["references_point_to_earlier_batch"]["pass"] is True assert inv["render_batches_cover_all_bg_once"]["pass"] is True assert inv["all_bg_images_covered"]["pass"] is True assert inv["all_groups_have_base"]["pass"] is True assert inv["no_human_decision_field"]["pass"] is True def test_w5_shot_to_background_image_coverage_exact(): from experiment_background_pipeline_slice import ( _build_generation_map, _build_generation_map_compatibility_report, ) plan = _fake_w4_render_plan_for_w5() gen_map = _build_generation_map(plan, _fake_w4_chain_plan_for_w5()) # shot_to_background_image counts all 4 shots from fixture assert set(gen_map["shot_to_background_image"].keys()) == { "S01_Shot1", "S01_Shot2", "S02_Shot1", "S03_Shot1", "S04_Shot1" } report = _build_generation_map_compatibility_report( gen_map=gen_map, render_plan=plan, w4_report={}, production_diff_empty=True, db_write_count=0, image_import_seen=False, prev_run_id="fake_w4", missing_inputs=[], ) cov = report["invariants"]["shot_to_background_image_coverage"] assert cov["pass"] is True assert cov["detail"]["expected_count"] == cov["detail"]["actual_count"] == 5 assert cov["detail"]["missing"] == [] and cov["detail"]["extra"] == [] def test_w5_image_backend_is_gpt_image_2_everywhere(): """User policy — gen_map top-level + every bg_image declares gpt-image-2 backend. compatibility invariant detects mismatch.""" from experiment_background_pipeline_slice import ( _build_generation_map, _build_generation_map_compatibility_report, ) plan = _fake_w4_render_plan_for_w5() gen_map = _build_generation_map(plan, _fake_w4_chain_plan_for_w5()) assert gen_map["image_generation_backend"] == "gpt-image-2" for bg_id, img in gen_map["background_images"].items(): assert img["expected_image_model"] == "gpt-image-2", bg_id report = _build_generation_map_compatibility_report( gen_map=gen_map, render_plan=plan, w4_report={}, production_diff_empty=True, db_write_count=0, image_import_seen=False, prev_run_id="fake_w4", missing_inputs=[], ) inv = report["invariants"]["image_model_is_gpt_image_2"] assert inv["pass"] is True assert inv["detail"]["gen_map_backend"] == "gpt-image-2" assert inv["detail"]["violating_bg_count"] == 0 # Tamper case — mutating any bg's model triggers FAIL. gen_map["background_images"]["L01B01"]["expected_image_model"] = "other-model" report2 = _build_generation_map_compatibility_report( gen_map=gen_map, render_plan=plan, w4_report={}, production_diff_empty=True, db_write_count=0, image_import_seen=False, prev_run_id="fake_w4", missing_inputs=[], ) assert report2["invariants"]["image_model_is_gpt_image_2"]["pass"] is False def test_w5_no_human_decision_fields_in_gen_map(): from experiment_background_pipeline_slice import _build_generation_map plan = _fake_w4_render_plan_for_w5() gen_map = _build_generation_map(plan, _fake_w4_chain_plan_for_w5()) serialized = json.dumps(gen_map) for banned in ("needs_user_decision", "manual_review_required", "rollup", "pending_human_decision", "human_review"): assert banned not in serialized, banned def test_w5_cli_derive_from_w4_run_exits_0(tmp_path): """CLI integration — feed a fake W4 run dir and verify W5 outputs + exit 0.""" prev = tmp_path / "fake_w4" prev.mkdir() (prev / "background_render_plan.json").write_text( json.dumps(_fake_w4_render_plan_for_w5(), ensure_ascii=False) ) (prev / "background_render_payload_dry_run.json").write_text( json.dumps({"per_bg": [], "unmapped_fields": []}, ensure_ascii=False) ) (prev / "reference_chain_plan.json").write_text( json.dumps(_fake_w4_chain_plan_for_w5(), ensure_ascii=False) ) (prev / "render_plan_compatibility_report.json").write_text( json.dumps({"all_pass": True, "invariants": {}}, ensure_ascii=False) ) (prev / "run_meta.json").write_text( json.dumps({"stage": "w4_render_plan", "run_status": "succeeded"}, ensure_ascii=False) ) out_root = tmp_path / "w5_out" result = _run([ "--derive-generation-map-from", str(prev), "--output-root", str(out_root), ]) assert result.returncode == 0, result.stderr rd = next(out_root.glob("*/")) for fname in ("background_generation_map.json", "generation_map_compatibility_report.json", "index.html", "run_meta.json"): assert (rd / fname).exists(), fname rm = json.loads((rd / "run_meta.json").read_text()) assert rm["stage"] == "w5_generation_map" assert rm["derived_from"] == "fake_w4" assert rm["run_status"] == "succeeded" assert rm["failed_invariants"] == [] # ────────────────────────────────────────────────────────────────────── # W6 narrow tests — background_image_payload_plan (gpt-image-2 mirror) # ────────────────────────────────────────────────────────────────────── def _fake_w5_gen_map_for_w6(prompt_brief_long=True): """Mirror W5 _build_generation_map shape. Covers independent + derive + weak. Uses prompt_brief long enough (>= 180 chars) so anchor report does not flag base_prompt_brief_length_lt_180 (avoid coupling to risk score unless a test explicitly inspects it). """ long_brief = ( "Medium upper-body framing inside a worn interior. Window light pools across the floor; " "muted color palette anchors a single-room layout. Use it as a stable spatial anchor for " "downstream variant generations across time-of-day and state transitions in this group." ) if prompt_brief_long else "short brief" return { "schema_version": 1, "stage": "w5_generation_map", "plan_version": "bps_w1", "image_generation_backend": "gpt-image-2", "groups": [ { "group_id": "L01|main", "loc_id": "L01", "space_key": "main", "base_bg_id": "L01B01", "member_bg_ids": ["L01B01", "L01B02", "L01B03"], "variant_bg_ids": ["L01B02", "L01B03"], "shot_keys": ["S01_Shot1", "S02_Shot1", "S03_Shot1"], "generation_order": ["L01B01", "L01B02", "L01B03"], }, { "group_id": "L02|main", "loc_id": "L02", "space_key": "main", "base_bg_id": "L02B01", "member_bg_ids": ["L02B01", "L02B02"], "variant_bg_ids": ["L02B02"], "shot_keys": ["S04_Shot1", "S05_Shot1"], "generation_order": ["L02B01", "L02B02"], }, ], "background_images": { "L01B01": { "bg_id": "L01B01", "image_title": "L01|main|day|normal", "loc_id": "L01", "space_key": "main", "render_role": "master_base", "state_class": "normal", "time_phase": "day", "render_mode": "independent_text_to_image", "primary_reference_bg_id": None, "secondary_reference_bg_ids": [], "reference_policy": "no_reference", "prompt_brief": long_brief, "applies_to_shots": ["S01_Shot1"], "expected_asset": {"asset_type": "chain_bg", "variant_type": "L01B01"}, "expected_image_model": "gpt-image-2", }, "L01B02": { "bg_id": "L01B02", "image_title": "L01|main|day|ransacked", "loc_id": "L01", "space_key": "main", "render_role": "state_variant", "state_class": "ransacked", "time_phase": "day", "render_mode": "derive_from_base_reference", "primary_reference_bg_id": "L01B01", "secondary_reference_bg_ids": [], "reference_policy": "strict_reference_to_base", "prompt_brief": "L01 ransacked: cabinets thrown open, debris scattered diagonally.", "applies_to_shots": ["S02_Shot1"], "expected_asset": {"asset_type": "chain_bg", "variant_type": "L01B02"}, "expected_image_model": "gpt-image-2", }, "L01B03": { "bg_id": "L01B03", "image_title": "L01|main|day|dream_or_vision_state", "loc_id": "L01", "space_key": "main", "render_role": "vision_or_special_variant", "state_class": "dream_or_vision_state", "time_phase": "day", "render_mode": "weak_reference_variant", "primary_reference_bg_id": "L01B01", "secondary_reference_bg_ids": [], "reference_policy": "weak_reference_continuity_only", "prompt_brief": "L01 dream: warped perspective, washed colors, dust suspended in air.", "applies_to_shots": ["S03_Shot1"], "expected_asset": {"asset_type": "chain_bg", "variant_type": "L01B03"}, "expected_image_model": "gpt-image-2", }, "L02B01": { "bg_id": "L02B01", "image_title": "L02|main|day|normal", "loc_id": "L02", "space_key": "main", "render_role": "master_base", "state_class": "normal", "time_phase": "day", "render_mode": "independent_text_to_image", "primary_reference_bg_id": None, "secondary_reference_bg_ids": [], "reference_policy": "no_reference", "prompt_brief": long_brief, "applies_to_shots": ["S04_Shot1"], "expected_asset": {"asset_type": "chain_bg", "variant_type": "L02B01"}, "expected_image_model": "gpt-image-2", }, "L02B02": { "bg_id": "L02B02", "image_title": "L02|main|night|normal", "loc_id": "L02", "space_key": "main", "render_role": "time_variant", "state_class": "normal", "time_phase": "night", "render_mode": "derive_from_base_reference", "primary_reference_bg_id": "L02B01", "secondary_reference_bg_ids": [], "reference_policy": "strict_reference_to_base", "prompt_brief": "L02 night: cooling cyan ambient, single warm window light.", "applies_to_shots": ["S05_Shot1"], "expected_asset": {"asset_type": "chain_bg", "variant_type": "L02B02"}, "expected_image_model": "gpt-image-2", }, }, "render_batches": [ ["L01B01", "L02B01"], ["L01B02", "L01B03", "L02B02"], ], "shot_to_background_image": { "S01_Shot1": "L01B01", "S02_Shot1": "L01B02", "S03_Shot1": "L01B03", "S04_Shot1": "L02B01", "S05_Shot1": "L02B02", }, "counts": { "background_images": 5, "groups": 2, "independent": 2, "derive_from_base_reference": 2, "weak_reference_variant": 1, "batches": 2, "shots_mapped": 5, }, "unmapped_to_production_fields": [], } def test_w6_payload_plan_covers_all_bg_with_required_fields(): from experiment_background_pipeline_slice import _build_image_payload_plan gen_map = _fake_w5_gen_map_for_w6() plan = _build_image_payload_plan(gen_map) assert plan["stage"] == "w6_image_payload_plan" assert plan["image_generation_backend"] == "gpt-image-2" # 1:1 set equality between W5 background_images and W6 payloads assert set(plan["payloads"].keys()) == set(gen_map["background_images"].keys()) # required per-payload fields required_fields = { "bg_id", "image_title", "image_model", "render_mode", "reference_policy", "prompt_text", "reference_images", "expected_asset", "applies_to_shots", "batch_index", "can_generate_after", "api_call_shape", } for bg_id, p in plan["payloads"].items(): missing = required_fields - set(p.keys()) assert not missing, f"{bg_id} missing fields {missing}" # carries assert plan["groups"] == gen_map["groups"] assert plan["render_batches"] == gen_map["render_batches"] assert plan["shot_to_background_image"] == gen_map["shot_to_background_image"] def test_w6_payload_model_is_gpt_image_2_everywhere(): """User policy + Codex consult — top-level backend + per-payload image_model + api_call_shape.model all gpt-image-2. Tamper mutates trigger compat FAIL.""" from experiment_background_pipeline_slice import ( _build_image_payload_plan, _build_image_payload_compatibility_report, ) gen_map = _fake_w5_gen_map_for_w6() plan = _build_image_payload_plan(gen_map) assert plan["image_generation_backend"] == "gpt-image-2" for bg, p in plan["payloads"].items(): assert p["image_model"] == "gpt-image-2", bg assert p["api_call_shape"]["model"] == "gpt-image-2", bg report = _build_image_payload_compatibility_report( plan=plan, gen_map=gen_map, production_diff_empty=True, db_write_count=0, image_import_seen=False, prev_run_id="fake_w5", missing_inputs=[], ) assert report["invariants"]["payload_model_is_gpt_image_2"]["pass"] is True # tamper — mutate one payload's api_call_shape.model plan["payloads"]["L01B01"]["api_call_shape"]["model"] = "other-model" report2 = _build_image_payload_compatibility_report( plan=plan, gen_map=gen_map, production_diff_empty=True, db_write_count=0, image_import_seen=False, prev_run_id="fake_w5", missing_inputs=[], ) assert report2["invariants"]["payload_model_is_gpt_image_2"]["pass"] is False def test_w6_independent_mode_uses_images_generate_with_no_refs(): from experiment_background_pipeline_slice import ( _build_image_payload_plan, _build_image_payload_compatibility_report, ) gen_map = _fake_w5_gen_map_for_w6() plan = _build_image_payload_plan(gen_map) independents = [bg for bg, p in plan["payloads"].items() if p["render_mode"] == "independent_text_to_image"] assert independents # fixture has 2 master_base for bg in independents: p = plan["payloads"][bg] assert p["reference_images"] == [], bg assert p["api_call_shape"]["client_method"] == "images.generate", bg assert p["api_call_shape"]["reference_input_kind"] == "none", bg assert p["can_generate_after"] == [], bg report = _build_image_payload_compatibility_report( plan=plan, gen_map=gen_map, production_diff_empty=True, db_write_count=0, image_import_seen=False, prev_run_id="fake_w5", missing_inputs=[], ) assert report["invariants"]["independent_payloads_have_no_refs"]["pass"] is True assert report["invariants"]["api_call_shape_matches_render_mode"]["pass"] is True def test_w6_derive_and_weak_modes_use_images_edit_with_single_ref(): from experiment_background_pipeline_slice import ( _build_image_payload_plan, _build_image_payload_compatibility_report, ) gen_map = _fake_w5_gen_map_for_w6() plan = _build_image_payload_plan(gen_map) referenced = [ bg for bg, p in plan["payloads"].items() if p["render_mode"] in ("derive_from_base_reference", "weak_reference_variant") ] assert referenced # fixture has 2 derive + 1 weak for bg in referenced: p = plan["payloads"][bg] refs = p["reference_images"] assert len(refs) == 1, bg assert refs[0]["role"] == "primary", bg assert refs[0]["bg_id"] == p["can_generate_after"][0], bg assert p["api_call_shape"]["client_method"] == "images.edit", bg assert p["api_call_shape"]["reference_input_kind"] == "single_background_image", bg # reference points to a payload that exists in this plan assert refs[0]["bg_id"] in plan["payloads"], bg report = _build_image_payload_compatibility_report( plan=plan, gen_map=gen_map, production_diff_empty=True, db_write_count=0, image_import_seen=False, prev_run_id="fake_w5", missing_inputs=[], ) inv = report["invariants"] assert inv["referenced_payloads_have_refs"]["pass"] is True assert inv["reference_payloads_point_to_existing_payloads"]["pass"] is True assert inv["api_call_shape_matches_render_mode"]["pass"] is True def test_w6_references_point_to_earlier_batch_invariant(): from experiment_background_pipeline_slice import ( _build_image_payload_plan, _build_image_payload_compatibility_report, ) gen_map = _fake_w5_gen_map_for_w6() plan = _build_image_payload_plan(gen_map) # batch_index honest carry from W5 for bg, p in plan["payloads"].items(): in_batch = next(i for i, batch in enumerate(plan["render_batches"]) if bg in batch) assert p["batch_index"] == in_batch, bg # references resolve to earlier batch bg_to_batch = {bg: i for i, batch in enumerate(plan["render_batches"]) for bg in batch} for bg, p in plan["payloads"].items(): for ref in p["reference_images"]: assert bg_to_batch[ref["bg_id"]] < bg_to_batch[bg], f"{bg} refs {ref['bg_id']}" report = _build_image_payload_compatibility_report( plan=plan, gen_map=gen_map, production_diff_empty=True, db_write_count=0, image_import_seen=False, prev_run_id="fake_w5", missing_inputs=[], ) assert report["invariants"]["references_point_to_earlier_batch"]["pass"] is True def test_w6_shot_coverage_carried_exact_and_prompt_text_invariants(): from experiment_background_pipeline_slice import ( _build_image_payload_plan, _build_image_payload_compatibility_report, ) gen_map = _fake_w5_gen_map_for_w6() plan = _build_image_payload_plan(gen_map) # shot map carry exact assert plan["shot_to_background_image"] == gen_map["shot_to_background_image"] # prompt_text contains W5 prompt_brief verbatim — no padding, no truncation for bg, p in plan["payloads"].items(): brief = gen_map["background_images"][bg]["prompt_brief"] assert brief in p["prompt_text"], bg # prompt_text must be brief + augmentation prefix (>= brief length) assert len(p["prompt_text"]) >= len(brief), bg # under hard max assert len(p["prompt_text"]) <= 6000, bg # prompt not empty assert p["prompt_text"].strip(), bg report = _build_image_payload_compatibility_report( plan=plan, gen_map=gen_map, production_diff_empty=True, db_write_count=0, image_import_seen=False, prev_run_id="fake_w5", missing_inputs=[], ) inv = report["invariants"] assert inv["shot_to_background_image_coverage_carried_exact"]["pass"] is True assert inv["payload_prompt_not_empty"]["pass"] is True assert inv["payload_prompt_length_not_over_6000"]["pass"] is True # prompt_length_profile advisory present prof = plan["prompt_length_profile"] assert prof["hard_max_threshold"] == 6000 assert prof["advisory_min_threshold"] == 500 assert prof["min"] >= 1 and prof["max"] <= 6000 def test_w6_variant_without_primary_ref_falls_back_to_independent(): """W5 reference_policy='fallback_independent_no_base' must map to images.generate with no refs, and the policy label is preserved. No human-decision fields.""" from experiment_background_pipeline_slice import ( _build_image_payload_plan, _build_image_payload_compatibility_report, ) gen_map = _fake_w5_gen_map_for_w6() # Mutate fixture: a variant w/o primary_ref (W5 would emit fallback policy) gen_map["background_images"]["L01B02"] = { **gen_map["background_images"]["L01B02"], "primary_reference_bg_id": None, "render_mode": "independent_text_to_image", "reference_policy": "fallback_independent_no_base", } plan = _build_image_payload_plan(gen_map) p = plan["payloads"]["L01B02"] assert p["render_mode"] == "independent_text_to_image" assert p["reference_policy"] == "fallback_independent_no_base" assert p["reference_images"] == [] assert p["can_generate_after"] == [] assert p["api_call_shape"]["client_method"] == "images.generate" assert p["api_call_shape"]["reference_input_kind"] == "none" # plan JSON serialization free of banned human-decision keys serialized = json.dumps(plan) for banned in ("needs_user_decision", "manual_review_required", "rollup", "pending_human_decision", "human_review", "needs_approval"): assert banned not in serialized, banned report = _build_image_payload_compatibility_report( plan=plan, gen_map=gen_map, production_diff_empty=True, db_write_count=0, image_import_seen=False, prev_run_id="fake_w5", missing_inputs=[], ) assert report["invariants"]["no_human_decision_field"]["pass"] is True assert report["invariants"]["referenced_payloads_have_refs"]["pass"] is True # also: fallback-no-base count > 0 reported in plan counts assert plan["counts"]["fallback_independent_no_base"] >= 1 def test_w6_base_prompt_anchor_report_structural_score_no_human_decision(): """Anchor report is per-group, structural-only, score-based, no human-decision keys.""" from experiment_background_pipeline_slice import _build_image_payload_plan gen_map = _fake_w5_gen_map_for_w6() plan = _build_image_payload_plan(gen_map) anchor = plan["base_prompt_anchor_report"] assert len(anchor) == len(gen_map["groups"]) group_ids = {e["group_id"] for e in anchor} assert group_ids == {g["group_id"] for g in gen_map["groups"]} # required fields present required = { "group_id", "base_bg_id", "member_count", "variant_count", "max_reference_depth", "risk_score", "risk_level", "risk_reasons", "recommended_auto_fallback", } for e in anchor: missing = required - set(e.keys()) assert not missing, f"{e.get('group_id')} missing {missing}" assert e["risk_level"] in ("low", "medium", "high"), e["risk_level"] assert isinstance(e["risk_score"], int) assert isinstance(e["risk_reasons"], list) # banned keys must not appear for banned in ("needs_user_decision", "manual_review_required", "rollup", "pending_human_decision", "human_review", "needs_approval"): assert banned not in e, banned # missing-base case → risk_level high + reason no_base_bg_id + fallback label gen_map_no_base = _fake_w5_gen_map_for_w6() gen_map_no_base["groups"][1]["base_bg_id"] = None plan2 = _build_image_payload_plan(gen_map_no_base) l02_entry = next(e for e in plan2["base_prompt_anchor_report"] if e["group_id"] == "L02|main") assert "no_base_bg_id" in l02_entry["risk_reasons"] assert l02_entry["recommended_auto_fallback"] == "fallback_independent_no_base" def test_w6_cli_derive_from_w5_run_exits_0(tmp_path): """CLI integration — feed a fake W5 run dir and verify W6 outputs + exit 0.""" prev = tmp_path / "fake_w5" prev.mkdir() (prev / "background_generation_map.json").write_text( json.dumps(_fake_w5_gen_map_for_w6(), ensure_ascii=False) ) (prev / "generation_map_compatibility_report.json").write_text( json.dumps({"all_pass": True, "invariants": {}}, ensure_ascii=False) ) (prev / "run_meta.json").write_text( json.dumps({"stage": "w5_generation_map", "run_status": "succeeded"}, ensure_ascii=False) ) out_root = tmp_path / "w6_out" result = _run([ "--derive-image-payloads-from", str(prev), "--output-root", str(out_root), ]) assert result.returncode == 0, result.stderr rd = next(out_root.glob("*/")) for fname in ("background_image_payload_plan.json", "image_payload_compatibility_report.json", "index.html", "run_meta.json"): assert (rd / fname).exists(), fname rm = json.loads((rd / "run_meta.json").read_text()) assert rm["stage"] == "w6_image_payload_plan" assert rm["derived_from"] == "fake_w5" assert rm["image_generation_backend"] == "gpt-image-2" assert rm["run_status"] == "succeeded" assert rm["failed_invariants"] == [] rep = json.loads((rd / "image_payload_compatibility_report.json").read_text()) assert rep["all_pass"] is True plan = json.loads((rd / "background_image_payload_plan.json").read_text()) assert plan["stage"] == "w6_image_payload_plan" assert plan["image_generation_backend"] == "gpt-image-2" # ────────────────────────────────────────────────────────────────────── # W7 narrow tests — background_plate_prompt_plan (LLM rewrite layer) # Experimental slice — keep tests light. # ────────────────────────────────────────────────────────────────────── def _fake_w6_plan_for_w7(): """Tiny W6 plan: 2 groups, 4 payloads (1 base + 1 derive + 1 weak + 1 fallback). Mirrors W6 _build_image_payload_plan output shape.""" return { "schema_version": 1, "stage": "w6_image_payload_plan", "plan_version": "bps_w1", "image_generation_backend": "gpt-image-2", "groups": [ {"group_id": "L01|main", "loc_id": "L01", "space_key": "main", "base_bg_id": "L01B01", "member_bg_ids": ["L01B01", "L01B02"], "variant_bg_ids": ["L01B02"], "shot_keys": ["S01_Shot1", "S02_Shot1"], "generation_order": ["L01B01", "L01B02"]}, {"group_id": "L02|main", "loc_id": "L02", "space_key": "main", "base_bg_id": "L02B01", "member_bg_ids": ["L02B01", "L02B02"], "variant_bg_ids": ["L02B02"], "shot_keys": ["S03_Shot1", "S04_Shot1"], "generation_order": ["L02B01", "L02B02"]}, ], "payloads": { "L01B01": { "bg_id": "L01B01", "image_title": "L01|main|day|normal", "image_model": "gpt-image-2", "render_mode": "independent_text_to_image", "reference_policy": "no_reference", "prompt_text": "Create a standalone chain background image. " + ("space "*60), "reference_images": [], "expected_asset": {"asset_type": "chain_bg", "variant_type": "L01B01"}, "applies_to_shots": ["S01_Shot1"], "batch_index": 0, "can_generate_after": [], "api_call_shape": {"client_method": "images.generate", "model": "gpt-image-2", "size": "1024x1024", "quality": "high", "n": 1, "reference_input_kind": "none"}, }, "L01B02": { "bg_id": "L01B02", "image_title": "L01|main|day|ransacked", "image_model": "gpt-image-2", "render_mode": "derive_from_base_reference", "reference_policy": "strict_reference_to_base", "prompt_text": "Use the referenced base background as a strict spatial continuity reference. " + ("debris "*40), "reference_images": [ {"bg_id": "L01B01", "role": "primary", "expected_asset": {"asset_type": "chain_bg", "variant_type": "L01B01"}} ], "expected_asset": {"asset_type": "chain_bg", "variant_type": "L01B02"}, "applies_to_shots": ["S02_Shot1"], "batch_index": 1, "can_generate_after": ["L01B01"], "api_call_shape": {"client_method": "images.edit", "model": "gpt-image-2", "size": "1024x1024", "quality": "high", "n": 1, "reference_input_kind": "single_background_image"}, }, "L02B01": { "bg_id": "L02B01", "image_title": "L02|main|day|normal", "image_model": "gpt-image-2", "render_mode": "weak_reference_variant", "reference_policy": "fallback_independent_no_base", "prompt_text": "Create a standalone chain background image (variant without an available base reference). " + ("tone "*40), "reference_images": [], "expected_asset": {"asset_type": "chain_bg", "variant_type": "L02B01"}, "applies_to_shots": ["S03_Shot1"], "batch_index": 0, "can_generate_after": [], "api_call_shape": {"client_method": "images.generate", "model": "gpt-image-2", "size": "1024x1024", "quality": "high", "n": 1, "reference_input_kind": "none"}, }, "L02B02": { "bg_id": "L02B02", "image_title": "L02|main|night|normal", "image_model": "gpt-image-2", "render_mode": "weak_reference_variant", "reference_policy": "weak_reference_continuity_only", "prompt_text": "Use the referenced base background only for loose spatial tone and continuity. " + ("cyan "*40), "reference_images": [ {"bg_id": "L02B01", "role": "primary", "expected_asset": {"asset_type": "chain_bg", "variant_type": "L02B01"}} ], "expected_asset": {"asset_type": "chain_bg", "variant_type": "L02B02"}, "applies_to_shots": ["S04_Shot1"], "batch_index": 1, "can_generate_after": ["L02B01"], "api_call_shape": {"client_method": "images.edit", "model": "gpt-image-2", "size": "1024x1024", "quality": "high", "n": 1, "reference_input_kind": "single_background_image"}, }, }, "render_batches": [["L01B01", "L02B01"], ["L01B02", "L02B02"]], "shot_to_background_image": { "S01_Shot1": "L01B01", "S02_Shot1": "L01B02", "S03_Shot1": "L02B01", "S04_Shot1": "L02B02", }, "base_prompt_anchor_report": [ {"group_id": "L01|main", "base_bg_id": "L01B01", "member_count": 2, "variant_count": 1, "max_reference_depth": 1, "risk_score": 0, "risk_level": "low", "risk_reasons": [], "recommended_auto_fallback": "generate_as_planned_then_fallback_to_independent_if_visual_validation_fails"}, {"group_id": "L02|main", "base_bg_id": "L02B01", "member_count": 2, "variant_count": 1, "max_reference_depth": 1, "risk_score": 1, "risk_level": "medium", "risk_reasons": ["group_includes_weak_reference_variant"], "recommended_auto_fallback": "use_weak_reference_continuity_for_variants"}, ], "counts": {"payloads": 4, "groups": 2, "render_batches": 2, "shots_mapped": 4, "images_generate_method": 2, "images_edit_method": 2, "fallback_independent_no_base": 1}, } def test_w7_dry_run_placeholder_covers_all_w6_payloads(): """Placeholder plate plan covers every W6 bg, status flagged, compat all_pass.""" from experiment_background_pipeline_slice import ( _build_placeholder_plate_plan, _build_plate_prompt_compatibility_report, ) w6 = _fake_w6_plan_for_w7() plan = _build_placeholder_plate_plan(w6) assert plan["stage"] == "w7_plate_prompt_plan" assert set(plan["plate_prompts"].keys()) == set(w6["payloads"].keys()) for bg, entry in plan["plate_prompts"].items(): assert entry["prompt_status"] == "placeholder_dry_run" assert entry["plate_prompt_text"] == w6["payloads"][bg]["prompt_text"] # verbatim carry assert entry["prompt_role"] in ( "base_plate", "strict_variant_delta", "weak_variant_delta", "fallback_independent_plate", ) # group guides exist for every group assert set(plan["base_plate_group_guides"].keys()) == { g["group_id"] for g in w6["groups"] } report = _build_plate_prompt_compatibility_report( plan=plan, w6_plan=w6, production_diff_empty=True, db_write_count=0, image_import_seen=False, prev_run_id="fake_w6", missing_inputs=[], ) assert report["all_pass"] is True, report["invariants"] def test_w7_gpt_image_2_and_api_call_shape_carried_exact(): """gpt-image-2 backend + api_call_shape + reference_images carry exact from W6.""" from experiment_background_pipeline_slice import ( _build_placeholder_plate_plan, _build_plate_prompt_compatibility_report, ) w6 = _fake_w6_plan_for_w7() plan = _build_placeholder_plate_plan(w6) assert plan["image_generation_backend"] == "gpt-image-2" for bg, p in plan["plate_prompts"].items(): assert p["image_model"] == "gpt-image-2", bg assert p["api_call_shape"] == w6["payloads"][bg]["api_call_shape"], bg assert p["reference_images"] == w6["payloads"][bg]["reference_images"], bg assert plan["render_batches"] == w6["render_batches"] assert plan["shot_to_background_image"] == w6["shot_to_background_image"] report = _build_plate_prompt_compatibility_report( plan=plan, w6_plan=w6, production_diff_empty=True, db_write_count=0, image_import_seen=False, prev_run_id="fake_w6", missing_inputs=[], ) inv = report["invariants"] assert inv["plate_model_is_gpt_image_2"]["pass"] is True assert inv["api_call_shape_carried_exact"]["pass"] is True assert inv["reference_images_carried_exact"]["pass"] is True assert inv["render_batches_carried_exact"]["pass"] is True assert inv["shot_to_background_image_carried_exact"]["pass"] is True def test_w7_over_6000_fails_under_2000_advisory_only(): """Prompt > 6000 → hard fail. <2000 or even >2000 → profile records but no exit blocker.""" from experiment_background_pipeline_slice import ( _build_placeholder_plate_plan, _build_plate_prompt_compatibility_report, ) w6 = _fake_w6_plan_for_w7() plan = _build_placeholder_plate_plan(w6) # baseline: every prompt is under 2000 + under 6000 → invariant pass rep_ok = _build_plate_prompt_compatibility_report( plan=plan, w6_plan=w6, production_diff_empty=True, db_write_count=0, image_import_seen=False, prev_run_id="fake_w6", missing_inputs=[], ) assert rep_ok["invariants"]["plate_prompt_length_not_over_6000"]["pass"] is True # under_2000 / over_2000 advisory recorded in profile but not in invariants prof = plan["prompt_length_profile"] assert "over_2000_count" in prof and "over_6000_count" in prof assert prof["hard_max_threshold"] == 6000 assert prof["advisory_over_threshold"] == 2000 # Inflate one prompt to 6001 → hard fail plan["plate_prompts"]["L01B01"]["plate_prompt_text"] = "x" * 6001 rep_bad = _build_plate_prompt_compatibility_report( plan=plan, w6_plan=w6, production_diff_empty=True, db_write_count=0, image_import_seen=False, prev_run_id="fake_w6", missing_inputs=[], ) assert rep_bad["invariants"]["plate_prompt_length_not_over_6000"]["pass"] is False def test_w7_cli_dry_run_derive_from_w6_exits_0(tmp_path): """CLI integration — dry-run derive from a fake W6 run dir, exit 0.""" prev = tmp_path / "fake_w6" prev.mkdir() (prev / "background_image_payload_plan.json").write_text( json.dumps(_fake_w6_plan_for_w7(), ensure_ascii=False) ) (prev / "image_payload_compatibility_report.json").write_text( json.dumps({"all_pass": True, "invariants": {}}, ensure_ascii=False) ) (prev / "run_meta.json").write_text( json.dumps({"stage": "w6_image_payload_plan", "run_status": "succeeded"}, ensure_ascii=False) ) out_root = tmp_path / "w7_out" result = _run([ "--derive-plate-prompts-from", str(prev), "--output-root", str(out_root), ]) assert result.returncode == 0, result.stderr rd = next(out_root.glob("*/")) for fname in ("background_plate_prompt_plan.json", "plate_prompt_compatibility_report.json", "index.html", "run_meta.json"): assert (rd / fname).exists(), fname rm = json.loads((rd / "run_meta.json").read_text()) assert rm["stage"] == "w7_plate_prompt_plan" assert rm["derived_from"] == "fake_w6" assert rm["run_status"] == "succeeded" assert rm["failed_invariants"] == [] plan = json.loads((rd / "background_plate_prompt_plan.json").read_text()) assert plan["generation_status"] == "placeholder_dry_run" # every plate prompt status is placeholder for p in plan["plate_prompts"].values(): assert p["prompt_status"] == "placeholder_dry_run" def test_w7b_high_risk_group_variants_override_to_weak_reference(): """W7b — high-risk group's derive variants downgrade to reference_strength=weak; group guide declares base_prompt_strategy=weak_reference_variants. Invariant passes.""" from experiment_background_pipeline_slice import ( _build_placeholder_plate_plan, _build_plate_prompt_compatibility_report, ) w6 = _fake_w6_plan_for_w7() # Promote L01|main group to high-risk (variant L01B02 is strict_variant_delta). for e in w6["base_prompt_anchor_report"]: if e["group_id"] == "L01|main": e["risk_level"] = "high" e["risk_reasons"] = ["variant_count_ge_3", "max_reference_depth_ge_2"] e["risk_score"] = 3 plan = _build_placeholder_plate_plan(w6) # L01B01 (base in high-risk group) → reference_strength=none assert plan["plate_prompts"]["L01B01"]["reference_strength"] == "none" # L01B02 (strict_variant in high-risk group) → reference_strength=weak (overridden) assert plan["plate_prompts"]["L01B02"]["reference_strength"] == "weak" assert plan["plate_prompts"]["L01B02"]["group_risk_level"] == "high" # Group guide declares weak_reference_variants strategy assert plan["base_plate_group_guides"]["L01|main"]["base_prompt_strategy"] == "weak_reference_variants" # api_call_shape still images.edit (only prompt/ref interpretation changes) assert plan["plate_prompts"]["L01B02"]["api_call_shape"]["client_method"] == "images.edit" # L02|main (medium-risk) variant stays strict assert plan["plate_prompts"]["L02B02"]["reference_strength"] == "weak" # weak_reference_variant render_mode → weak default # invariant report = _build_plate_prompt_compatibility_report( plan=plan, w6_plan=w6, production_diff_empty=True, db_write_count=0, image_import_seen=False, prev_run_id="fake_w6", missing_inputs=[], ) assert report["invariants"]["high_risk_groups_use_weak_reference_strategy"]["pass"] is True assert report["all_pass"] is True # Tamper: force a high-risk variant to strict → invariant FAIL plan["plate_prompts"]["L01B02"]["reference_strength"] = "strict" rep_bad = _build_plate_prompt_compatibility_report( plan=plan, w6_plan=w6, production_diff_empty=True, db_write_count=0, image_import_seen=False, prev_run_id="fake_w6", missing_inputs=[], ) assert rep_bad["invariants"]["high_risk_groups_use_weak_reference_strategy"]["pass"] is False def test_w7c_derived_from_chain_resolves_to_source_bundle_and_builds_group_context(tmp_path): """W7c chain resolver finds source_bundle.json by walking derived_from from the start run dir; per-group context uses exact loc_id/shot_keys match. No semantic extraction; generic key carry only. No sample-specific literals.""" from experiment_background_pipeline_slice import ( _resolve_source_bundle_via_derived_from_chain, _build_per_group_source_context, ) # Synthetic 3-hop chain: W6 -> W5 -> W4 (root, holds source_bundle.json) parent = tmp_path / "chain" parent.mkdir() (parent / "rootW4").mkdir() (parent / "midW5").mkdir() (parent / "topW6").mkdir() (parent / "rootW4" / "run_meta.json").write_text( json.dumps({"stage": "w4", "derived_from": None}, ensure_ascii=False) ) (parent / "rootW4" / "source_bundle.json").write_text( json.dumps({ "selected_shots": [ {"shot_key": "ShotA", "shot_description": "alpha desc", "screenplay_scene_heading": "head A", "beat_title": "beat A", "scene_summary": "summary A", "scene_type": "normal", "t2i_variations_json": json.dumps([ {"variant_label": "v1", "source_facts": ["fact-a"], "visual_inferences": ["inf-a"], "owned_object_usage": ["own-a"]} ]), "visible_entities_json": json.dumps([{"short_id": "E01"}])}, {"shot_key": "ShotB", "shot_description": "beta desc", "screenplay_scene_heading": "head B", "beat_title": "beat B", "scene_summary": "summary B", "scene_type": "normal", "t2i_variations_json": None, "visible_entities_json": None}, ], "locations_for_llm": [ {"loc_id": "Lx", "name": "Loc X", "description": "generic location", "space_profile": {"kind": "single_space"}}, ], }, ensure_ascii=False) ) (parent / "midW5" / "run_meta.json").write_text( json.dumps({"stage": "w5", "derived_from": "rootW4"}, ensure_ascii=False) ) (parent / "topW6" / "run_meta.json").write_text( json.dumps({"stage": "w6", "derived_from": "midW5"}, ensure_ascii=False) ) # Resolve from W6 dir → should walk to rootW4 and return its source_bundle resolved = _resolve_source_bundle_via_derived_from_chain(parent / "topW6") assert resolved["root_run_id"] == "rootW4" assert resolved["chain"] == ["topW6", "midW5", "rootW4"] bundle = resolved["source_bundle"] # Per-group context: a group with loc_id=Lx and shot_keys=[ShotA] picks # exactly that shot from selected_shots, parses t2i + visible_entities. w6_plan_min = { "groups": [ {"group_id": "Lx|main", "loc_id": "Lx", "shot_keys": ["ShotA"], "base_bg_id": "Lx-base"}, ], } ctx = _build_per_group_source_context(w6_plan_min, bundle) g = ctx["Lx|main"] assert g["loc_id"] == "Lx" assert g["location"]["description"] == "generic location" assert len(g["shots"]) == 1 s = g["shots"][0] assert s["shot_key"] == "ShotA" assert s["beat_title"] == "beat A" assert s["t2i_variant_anchors"][0]["source_facts"] == ["fact-a"] assert s["visible_entity_short_ids"] == ["E01"] # Missing chain step is fail-closed. (parent / "topW6" / "run_meta.json").write_text( json.dumps({"stage": "w6", "derived_from": "ghost"}, ensure_ascii=False) ) import pytest with pytest.raises(FileNotFoundError): _resolve_source_bundle_via_derived_from_chain(parent / "topW6") def test_w7c_llm_input_carries_source_context_and_group_guide_fields_round_trip(monkeypatch): """W7c — generic carry smoke. _generate_llm_plate_plan must put each group's source_context into the LLM input, and the four new source-derived group guide fields must round-trip from a fake LLM output into the assembled plan. No L05/L04 or sample-specific literals are asserted here.""" import experiment_background_pipeline_slice as mod # noqa: F401 import sys as _sys w6 = _fake_w6_plan_for_w7() # Synthetic generic source context — exact-id keys for the W6 fixture groups. fake_ctx = { "L01|main": { "group_id": "L01|main", "loc_id": "L01", "shot_keys": ["S01_Shot1", "S02_Shot1"], "location": {"loc_id": "L01", "name": "n", "description": "generic1"}, "shots": [{"shot_key": "S01_Shot1", "shot_description": "ggA"}], }, "L02|main": { "group_id": "L02|main", "loc_id": "L02", "shot_keys": ["S03_Shot1", "S04_Shot1"], "location": {"loc_id": "L02", "name": "m", "description": "generic2"}, "shots": [{"shot_key": "S03_Shot1", "shot_description": "ggB"}], }, } captured = {} def _fake_completion(**kwargs): # Inspect the user prompt for source_context presence by group_id. msgs = kwargs.get("messages", []) user_payload = json.loads(msgs[1]["content"]) if len(msgs) >= 2 else {} captured["user_payload"] = user_payload # Build a fully-shaped fake LLM output covering both groups. llm_out = { "plate_prompts": { bg: {"plate_prompt_text": f"plate {bg}", "consistency_intent": "ci", "variant_delta": "vd"} for bg in w6["payloads"] }, "base_plate_group_guides": { gid: { "shared_plate_intent": "spi", "shared_continuity_constraints": "scc", "variant_delta_policy": "vdp", "auto_strategy": "as", "source_context_summary": f"summary {gid}", "scale_and_default_avoidance": f"scale {gid}", "must_preserve_spatial_anchors": f"anchors {gid}", "state_transition_anchors": f"transitions {gid}", } for gid in ("L01|main", "L02|main") }, } class _R: def __init__(self, content): self.choices = [type("C", (), {"message": type("M", (), {"content": content})()})] return _R(json.dumps(llm_out)) fake_module = type("FL", (), {"completion": staticmethod(_fake_completion)}) monkeypatch.setitem(_sys.modules, "litellm", fake_module) monkeypatch.setenv("GEMINI_API_KEY", "fake") from experiment_background_pipeline_slice import _generate_llm_plate_plan plan = _generate_llm_plate_plan(w6, model="gemini-3.5-flash", source_context=fake_ctx) # LLM input received source_context per-group. payload = captured["user_payload"] groups_in = {g["group_id"]: g for g in payload["groups"]} assert groups_in["L01|main"]["source_context"]["loc_id"] == "L01" assert groups_in["L02|main"]["source_context"]["loc_id"] == "L02" # Each guide carries the four W7c fields verbatim from the fake LLM output. for gid in ("L01|main", "L02|main"): g = plan["base_plate_group_guides"][gid] for fld in ("source_context_summary", "scale_and_default_avoidance", "must_preserve_spatial_anchors", "state_transition_anchors"): assert g[fld] and gid in g[fld], (gid, fld, g[fld]) assert plan["generation_status"] == "generated" def _fake_w7_plan_for_w8(): """Minimal W7 plan with 3 bg: 1 base + 1 strict_variant + 1 weak_variant. Mirrors W7 _build_image_payload_plan / _assemble_plate_plan shape.""" return { "schema_version": 1, "stage": "w7_plate_prompt_plan", "plan_version": "bps_w1", "image_generation_backend": "gpt-image-2", "plate_prompts": { "Xa": { "bg_id": "Xa", "image_title": "Xa-title", "image_model": "gpt-image-2", "render_mode": "independent_text_to_image", "api_call_shape": {"client_method": "images.generate", "model": "gpt-image-2", "size": "1024x1024", "quality": "high", "n": 1, "reference_input_kind": "none"}, "reference_images": [], "expected_asset": {"asset_type": "chain_bg", "variant_type": "Xa"}, "applies_to_shots": ["Sx_Shot1"], "original_prompt_text": "orig Xa", "plate_prompt_text": "Plate text for Xa", "prompt_role": "base_plate", "reference_strength": "none", "group_risk_level": "low", "consistency_intent": "ci", "variant_delta": "", "prompt_status": "generated", }, "Xb": { "bg_id": "Xb", "image_title": "Xb-title", "image_model": "gpt-image-2", "render_mode": "derive_from_base_reference", "api_call_shape": {"client_method": "images.edit", "model": "gpt-image-2", "size": "1024x1024", "quality": "high", "n": 1, "reference_input_kind": "single_background_image"}, "reference_images": [{"bg_id": "Xa", "role": "primary", "expected_asset": {"asset_type": "chain_bg", "variant_type": "Xa"}}], "expected_asset": {"asset_type": "chain_bg", "variant_type": "Xb"}, "applies_to_shots": ["Sx_Shot2"], "original_prompt_text": "orig Xb", "plate_prompt_text": "Plate text for Xb", "prompt_role": "strict_variant_delta", "reference_strength": "strict", "group_risk_level": "low", "consistency_intent": "ci", "variant_delta": "vd", "prompt_status": "generated", }, "Xc": { "bg_id": "Xc", "image_title": "Xc-title", "image_model": "gpt-image-2", "render_mode": "weak_reference_variant", "api_call_shape": {"client_method": "images.edit", "model": "gpt-image-2", "size": "1024x1024", "quality": "high", "n": 1, "reference_input_kind": "single_background_image"}, "reference_images": [{"bg_id": "Xa", "role": "primary", "expected_asset": {"asset_type": "chain_bg", "variant_type": "Xa"}}], "expected_asset": {"asset_type": "chain_bg", "variant_type": "Xc"}, "applies_to_shots": ["Sx_Shot3"], "original_prompt_text": "orig Xc", "plate_prompt_text": "Plate text for Xc", "prompt_role": "weak_variant_delta", "reference_strength": "weak", "group_risk_level": "low", "consistency_intent": "ci", "variant_delta": "vd", "prompt_status": "generated", }, }, "base_plate_group_guides": {}, "groups": [ {"group_id": "X|main", "base_bg_id": "Xa", "member_bg_ids": ["Xa", "Xb", "Xc"], "variant_bg_ids": ["Xb", "Xc"], "shot_keys": ["Sx_Shot1", "Sx_Shot2", "Sx_Shot3"], "generation_order": ["Xa", "Xb", "Xc"]}, ], "render_batches": [["Xa"], ["Xb", "Xc"]], "shot_to_background_image": {"Sx_Shot1": "Xa", "Sx_Shot2": "Xb", "Sx_Shot3": "Xc"}, "counts": {}, "prompt_length_profile": {}, } def test_w8_subset_selection_follows_render_batch_order(): """W8 selected subset ordered by W7 render_batches (base before variants).""" from experiment_background_pipeline_slice import _select_and_order_bg_ids plan = _fake_w7_plan_for_w8() # Reversed subset → still must come out base-first then variants ordered = _select_and_order_bg_ids(plan, ["Xc", "Xb", "Xa"]) assert ordered == ["Xa", "Xb", "Xc"] # Subset that omits base → variants still ordered by their own batch ordered2 = _select_and_order_bg_ids(plan, ["Xc", "Xb"]) assert ordered2 == ["Xb", "Xc"] def test_w8_dry_run_default_no_openai_import_no_png(tmp_path): """W8 CLI default is dry-run: no openai import, no PNG, but plan + report emitted.""" prev = tmp_path / "fake_w7" prev.mkdir() (prev / "background_plate_prompt_plan.json").write_text( json.dumps(_fake_w7_plan_for_w8(), ensure_ascii=False) ) (prev / "plate_prompt_compatibility_report.json").write_text( json.dumps({"all_pass": True, "invariants": {}}, ensure_ascii=False) ) (prev / "run_meta.json").write_text( json.dumps({"stage": "w7_plate_prompt_plan", "run_status": "succeeded"}, ensure_ascii=False) ) out_root = tmp_path / "w8_out" result = _run([ "--derive-image-generation-from", str(prev), "--output-root", str(out_root), ]) assert result.returncode == 0, result.stderr rd = next(out_root.glob("*/")) for fname in ("background_image_generation_result.json", "image_generation_compatibility_report.json", "index.html", "run_meta.json"): assert (rd / fname).exists(), fname # No PNG written in dry-run pngs = list((rd / "images").glob("*.png")) assert pngs == [] data = json.loads((rd / "background_image_generation_result.json").read_text()) assert data["dry_run"] is True for r in data["results"]: assert r["status"] == "dry_run_skipped" assert r["model_requested"] == "gpt-image-2" rep = json.loads((rd / "image_generation_compatibility_report.json").read_text()) # references_resolved_when_needed is advisory in dry-run assert rep["all_pass"] is True def test_w8_missing_reference_blocks_variant_when_generating(tmp_path, monkeypatch): """When --generate-images is requested but a variant's reference PNG is absent, that variant is flagged blocked_missing_reference and the run fails-closed (exit 1). No fake API call is made by the runtime.""" from experiment_background_pipeline_slice import ( _generate_image_for_bg, _build_image_generation_compatibility_report, ) plan = _fake_w7_plan_for_w8() images_dir = tmp_path / "images" images_dir.mkdir() # variant Xb references Xa, but no Xa PNG exists → blocked_missing_reference r = _generate_image_for_bg( "Xb", plan["plate_prompts"]["Xb"], openai_client=None, # not used on blocked path images_dir=images_dir, dry_run=False, ) assert r["status"] == "blocked_missing_reference" assert "Xa.png" in (r["error"] or "") # report flags missing reference under generate path (not dry-run) rep = _build_image_generation_compatibility_report( results=[r], plan=plan, selected_bg_ids=["Xb"], production_diff_empty=True, db_write_count=0, image_call_made=False, dry_run_requested=False, missing_inputs=[], prev_run_id="fake_w7", ) assert rep["invariants"]["references_resolved_when_needed"]["pass"] is False def test_w8_compatibility_enforces_gpt_image_2_and_no_human_decision(): """W8 compatibility report enforces gpt-image-2 declaration + rejects banned human-decision keys. Tamper a non-gpt-image-2 model → invariant FAIL.""" from experiment_background_pipeline_slice import ( _generate_image_for_bg, _build_image_generation_compatibility_report, ) plan = _fake_w7_plan_for_w8() # Tamper with declared model on Xa plan_tampered = json.loads(json.dumps(plan)) plan_tampered["plate_prompts"]["Xa"]["api_call_shape"]["model"] = "other-model" r = _generate_image_for_bg( "Xa", plan_tampered["plate_prompts"]["Xa"], openai_client=None, images_dir=Path("/tmp"), dry_run=True, ) assert r["status"] == "validation_failed" rep = _build_image_generation_compatibility_report( results=[r], plan=plan_tampered, selected_bg_ids=["Xa"], production_diff_empty=True, db_write_count=0, image_call_made=False, dry_run_requested=True, missing_inputs=[], prev_run_id="fake_w7", ) assert rep["invariants"]["model_is_gpt_image_2_for_all_attempts"]["pass"] is False def test_w8b_weak_reference_prompt_only_mode_drops_refs_for_weak_payloads(tmp_path): """W8b — when weak_reference_mode='prompt_only', payloads with reference_strength='weak' must call images.generate with NO ref PNG, while strict variants keep using images.edit and the declared refs. Generic flag/enum check; no scenario-specific literals.""" from experiment_background_pipeline_slice import ( _generate_image_for_bg, _build_image_generation_compatibility_report, ) plan = _fake_w7_plan_for_w8() images_dir = tmp_path / "images" images_dir.mkdir() # Pre-create base PNG so strict variant Xb is not blocked when generating. (images_dir / "Xa.png").write_bytes(b"fake-png-bytes-for-test") # Edge: do NOT actually call OpenAI; use dry_run=True to short-circuit the # API call but still exercise the effective_reference_mode/ref-skip logic. # The decision is computed before the dry_run early-return, so result # fields (client_method_effective / effective_reference_mode / # reference_strength / weak_reference_mode) reflect the active policy. # Strict variant (Xb): edit_with_ref regardless of mode → declared refs preserved. r_strict = _generate_image_for_bg( "Xb", plan["plate_prompts"]["Xb"], openai_client=None, images_dir=images_dir, dry_run=True, weak_reference_mode="prompt_only", ) assert r_strict["effective_reference_mode"] == "edit_with_ref" assert r_strict["client_method_effective"] == "images.edit" assert r_strict["declared_reference_bg_ids"] == ["Xa"] # Weak variant (Xc) under prompt_only → ref dropped, generate method. r_weak_drop = _generate_image_for_bg( "Xc", plan["plate_prompts"]["Xc"], openai_client=None, images_dir=images_dir, dry_run=True, weak_reference_mode="prompt_only", ) assert r_weak_drop["effective_reference_mode"] == "prompt_only_weak" assert r_weak_drop["client_method_effective"] == "images.generate" # declared refs still preserved for audit; effective ref PNG list is empty. assert r_weak_drop["declared_reference_bg_ids"] == ["Xa"] assert r_weak_drop["reference_pngs_used"] == [] # Same weak variant under default mode → edit_with_ref preserved. r_weak_keep = _generate_image_for_bg( "Xc", plan["plate_prompts"]["Xc"], openai_client=None, images_dir=images_dir, dry_run=True, weak_reference_mode="edit_with_ref", ) assert r_weak_keep["effective_reference_mode"] == "edit_with_ref" assert r_weak_keep["client_method_effective"] == "images.edit" # Dependency invariant under prompt_only mode: weak payload has no # effective dependency, so a base-less weak run can pass. rep = _build_image_generation_compatibility_report( results=[r_weak_drop], plan=plan, selected_bg_ids=["Xc"], production_diff_empty=True, db_write_count=0, image_call_made=False, dry_run_requested=True, missing_inputs=[], prev_run_id="fake_w7", ) assert rep["invariants"]["generation_order_respects_dependency"]["pass"] is True def test_w10_anchor_objects_extracted_from_owned_object_usage_by_usage_kind(): """W10 — _extract_t2i_anchors filters owned_object_usage by enum value usage_kind=='anchor' and surfaces an anchor_objects list per variant. Pure enum filter, no semantic word matching.""" from experiment_background_pipeline_slice import _extract_t2i_anchors t2i = [ { "variant_label": "v1", "owned_object_usage": [ {"owned_token": "TKN_A", "usage_kind": "anchor", "source_phrase": "anchor phrase A"}, {"owned_token": "TKN_B", "usage_kind": "absent", "source_phrase": ""}, {"owned_token": "TKN_C", "usage_kind": "anchor", "source_phrase": "anchor phrase C"}, ], "source_facts": ["sf"], "visual_inferences": ["vi"], }, { "variant_label": "v2", "owned_object_usage": "not a list", # malformed; gracefully empty }, ] out = _extract_t2i_anchors(t2i) assert len(out) == 2 # v1 has two anchors filtered by usage_kind enum a1 = out[0]["anchor_objects"] assert [e["owned_token"] for e in a1] == ["TKN_A", "TKN_C"] assert a1[0]["source_phrase"] == "anchor phrase A" assert a1[1]["source_phrase"] == "anchor phrase C" # v2 with non-list owned_object_usage → empty anchor_objects assert out[1]["anchor_objects"] == [] def test_w8h_continuity_partition_checker_pass_and_fail(): """W8h compatibility checker validates set-partition + acyclic chain shape.""" from experiment_background_pipeline_slice import ( _build_continuity_compatibility_report, ) w7_plan = _fake_w7_plan_for_w8() # bg set: {Xa, Xb, Xc} good = { "image_continuity_subgroups": [ {"subgroup_id": "g1", "member_bg_ids": ["Xa"], "base_bg_id": "Xa", "generation_chain": ["Xa"], "reason_brief": "base only", "confidence": "high"}, {"subgroup_id": "g2", "member_bg_ids": ["Xb", "Xc"], "base_bg_id": "Xb", "generation_chain": ["Xb", "Xc"], "reason_brief": "state transition", "confidence": "high"}, ], } rep_ok = _build_continuity_compatibility_report( continuity=good, w7_plan=w7_plan, production_diff_empty=True, db_write_count=0, image_import_seen=False, prev_run_id="fake_w7", missing_inputs=[], ) assert rep_ok["all_pass"] is True, rep_ok["invariants"] # Tamper: duplicate bg across subgroups → invariant FAIL bad = json.loads(json.dumps(good)) bad["image_continuity_subgroups"][0]["member_bg_ids"].append("Xb") rep_bad = _build_continuity_compatibility_report( continuity=bad, w7_plan=w7_plan, production_diff_empty=True, db_write_count=0, image_import_seen=False, prev_run_id="fake_w7", missing_inputs=[], ) assert rep_bad["invariants"]["bg_set_partitioned_by_subgroups"]["pass"] is False def test_w9_same_subgroup_direct_parent_overrides_weak_to_edit(tmp_path): """W9 — when subgroup_info names a direct parent in the SAME subgroup that is also a declared ref, weak reference_strength is upgraded to images.edit (subgroup_state_transition) regardless of weak_reference_mode=prompt_only.""" from experiment_background_pipeline_slice import _generate_image_for_bg plan = _fake_w7_plan_for_w8() images_dir = tmp_path / "images" images_dir.mkdir() (images_dir / "Xa.png").write_bytes(b"fake-png-bytes") # Xc declares ref=Xa, reference_strength=weak (from fixture). # Subgroup says Xa->Xc are same-room with Xa as base. subgroup_info = { "subgroup_id": "g_same", "base_bg_id": "Xa", "generation_chain": ["Xa", "Xc"], "position_in_chain": 1, } r = _generate_image_for_bg( "Xc", plan["plate_prompts"]["Xc"], openai_client=None, images_dir=images_dir, dry_run=False, weak_reference_mode="prompt_only", subgroup_info=subgroup_info, ) assert r["effective_reference_mode"] == "subgroup_state_transition" assert r["client_method_effective"] == "images.edit" assert r["same_subgroup_parent_bg"] == "Xa" # The reference PNG list is the same-subgroup parent only, not arbitrary declared refs assert r["reference_pngs_used"] == [str(images_dir / "Xa.png")] # status should be blocked-or-actual-call depending on PNG bytes validity; # here we only assert the effective policy decision; the real edit call # would proceed (caller responsibility). We simulate by checking that the # blocked path was NOT taken since the parent PNG exists. assert r["status"] != "blocked_missing_reference" def test_w9_cross_subgroup_weak_stays_prompt_only(tmp_path): """W9 — when subgroup_info is missing or position 0 (no in-subgroup parent), weak payloads still follow weak_reference_mode=prompt_only (cross-subgroup parent like a kitchen base for a bedroom variant is NOT used).""" from experiment_background_pipeline_slice import _generate_image_for_bg plan = _fake_w7_plan_for_w8() images_dir = tmp_path / "images" images_dir.mkdir() # base of a different subgroup (images_dir / "Xa.png").write_bytes(b"fake-png-bytes") # Xc weak variant, subgroup info says Xc has NO same-subgroup parent # (Xc is the base of its own subgroup; declared ref Xa is in another subgroup). subgroup_info = { "subgroup_id": "g_other", "base_bg_id": "Xc", "generation_chain": ["Xc"], "position_in_chain": 0, } r = _generate_image_for_bg( "Xc", plan["plate_prompts"]["Xc"], openai_client=None, images_dir=images_dir, dry_run=True, weak_reference_mode="prompt_only", subgroup_info=subgroup_info, ) assert r["effective_reference_mode"] == "prompt_only_weak" assert r["client_method_effective"] == "images.generate" assert r["same_subgroup_parent_bg"] is None def test_w8f_assembly_prepends_film_set_framing_and_appends_generic_avoidance(tmp_path): """W8f — _assemble_w8c_prompt prepends a generic film-set framing prefix in front of every non-empty per-bg prompt, and (when a group_guide is present) appends the W8e generic default-avoidance suffix. NO group-guide field value is carried verbatim. Audit flags reflect both insertions.""" from experiment_background_pipeline_slice import ( _generate_image_for_bg, _assemble_w8c_prompt, W8E_DEFAULT_AVOIDANCE_SUFFIX, W8F_FILM_SET_FRAMING_PREFIX, ) plan = _fake_w7_plan_for_w8() fake_guide = { "scale_and_default_avoidance": "GUIDE_SCALE_ABCDEF", "must_preserve_spatial_anchors": "GUIDE_ANCHORS_XYZ123", } out = _assemble_w8c_prompt("BASE_PROMPT_TEXT", fake_guide) assert out["prompt_text_used"].startswith(W8F_FILM_SET_FRAMING_PREFIX) assert "BASE_PROMPT_TEXT" in out["prompt_text_used"] assert W8E_DEFAULT_AVOIDANCE_SUFFIX in out["prompt_text_used"] # No group-guide value tokens leak through. assert "GUIDE_SCALE_ABCDEF" not in out["prompt_text_used"] assert "GUIDE_ANCHORS_XYZ123" not in out["prompt_text_used"] assert out["carried_fields"] == [] assert out["applied_default_avoidance_policy"] is True assert out["applied_film_set_framing_prefix"] is True images_dir = tmp_path / "images" images_dir.mkdir() r = _generate_image_for_bg( "Xa", plan["plate_prompts"]["Xa"], openai_client=None, images_dir=images_dir, dry_run=True, weak_reference_mode="prompt_only", group_guide=fake_guide, ) assert r["plate_prompt_text_original"] == plan["plate_prompts"]["Xa"]["plate_prompt_text"] assert r["prompt_text_used"].startswith(W8F_FILM_SET_FRAMING_PREFIX) assert r["plate_prompt_text_original"] in r["prompt_text_used"] assert W8E_DEFAULT_AVOIDANCE_SUFFIX in r["prompt_text_used"] assert r["applied_film_set_framing_prefix"] is True assert r["applied_default_avoidance_policy"] is True # No group_guide → film-set prefix still present, but no avoidance suffix. r_no_guide = _generate_image_for_bg( "Xa", plan["plate_prompts"]["Xa"], openai_client=None, images_dir=images_dir, dry_run=True, weak_reference_mode="prompt_only", group_guide=None, ) assert r_no_guide["prompt_text_used"].startswith(W8F_FILM_SET_FRAMING_PREFIX) assert W8E_DEFAULT_AVOIDANCE_SUFFIX not in r_no_guide["prompt_text_used"] assert r_no_guide["applied_film_set_framing_prefix"] is True assert r_no_guide["applied_default_avoidance_policy"] is False def test_w8e_assembly_appends_generic_default_avoidance_only_not_group_value(tmp_path): """W8e — _generate_image_for_bg + _assemble_w8c_prompt no longer push any group_guide field value verbatim into the final prompt. When a group_guide is present, only a generic default-avoidance policy suffix is appended. The original plate prompt is preserved. Generic policy carry only; no fixture object tokens leak through. (W8f film-set prefix is still present at the head; this test focuses on the suffix policy.)""" from experiment_background_pipeline_slice import ( _generate_image_for_bg, _assemble_w8c_prompt, W8E_DEFAULT_AVOIDANCE_SUFFIX, ) plan = _fake_w7_plan_for_w8() fake_guide = { "group_id": "X|main", "scale_and_default_avoidance": "GUIDE_SCALE_ABCDEF", "must_preserve_spatial_anchors": "GUIDE_ANCHORS_XYZ123", "shared_plate_intent": "ignored", } # Direct assembler — generic suffix appended; NO group value tokens carry. out = _assemble_w8c_prompt("BASE_PROMPT_TEXT", fake_guide) assert "BASE_PROMPT_TEXT" in out["prompt_text_used"] assert "GUIDE_SCALE_ABCDEF" not in out["prompt_text_used"] assert "GUIDE_ANCHORS_XYZ123" not in out["prompt_text_used"] assert W8E_DEFAULT_AVOIDANCE_SUFFIX in out["prompt_text_used"] assert out["carried_fields"] == [] assert out["applied_default_avoidance_policy"] is True # Generator dry-run path: original preserved, only generic policy appended. images_dir = tmp_path / "images" images_dir.mkdir() r = _generate_image_for_bg( "Xa", plan["plate_prompts"]["Xa"], openai_client=None, images_dir=images_dir, dry_run=True, weak_reference_mode="prompt_only", group_guide=fake_guide, ) assert r["plate_prompt_text_original"] == plan["plate_prompts"]["Xa"]["plate_prompt_text"] assert r["plate_prompt_text_original"] in r["prompt_text_used"] assert "GUIDE_SCALE_ABCDEF" not in r["prompt_text_used"] assert "GUIDE_ANCHORS_XYZ123" not in r["prompt_text_used"] assert W8E_DEFAULT_AVOIDANCE_SUFFIX in r["prompt_text_used"] assert r["carried_group_guide_fields"] == [] assert r["applied_default_avoidance_policy"] is True # No group_guide → policy NOT applied, prompt_text_used contains the # original plate text (the W8f film-set framing prefix is still present; # see test_w8f_* for that contract). r_no_guide = _generate_image_for_bg( "Xa", plan["plate_prompts"]["Xa"], openai_client=None, images_dir=images_dir, dry_run=True, weak_reference_mode="prompt_only", group_guide=None, ) assert r_no_guide["carried_group_guide_fields"] == [] assert r_no_guide["applied_default_avoidance_policy"] is False assert r_no_guide["plate_prompt_text_original"] in r_no_guide["prompt_text_used"] assert W8E_DEFAULT_AVOIDANCE_SUFFIX not in r_no_guide["prompt_text_used"] def test_w7d_llm_input_payload_includes_reference_strength_and_group_risk(monkeypatch): """W7d — compact_payloads exposes reference_strength/group_risk_level to the LLM so per-bg wording can match the chosen reference policy. Generic smoke; no sample-specific literals asserted.""" import experiment_background_pipeline_slice as mod # noqa: F401 import sys as _sys w6 = _fake_w6_plan_for_w7() # Promote L01|main to high-risk so derive variant must show reference_strength=weak. for e in w6["base_prompt_anchor_report"]: if e["group_id"] == "L01|main": e["risk_level"] = "high" e["risk_score"] = 3 captured = {} def _fake_completion(**kwargs): msgs = kwargs.get("messages", []) captured["user_payload"] = json.loads(msgs[1]["content"]) if len(msgs) >= 2 else {} llm_out = { "plate_prompts": { bg: {"plate_prompt_text": f"plate {bg}", "consistency_intent": "ci", "variant_delta": "vd"} for bg in w6["payloads"] }, "base_plate_group_guides": { gid: { "shared_plate_intent": "spi", "shared_continuity_constraints": "scc", "variant_delta_policy": "vdp", "auto_strategy": "as", "source_context_summary": "scs", "scale_and_default_avoidance": "sav", "must_preserve_spatial_anchors": "anchors", "state_transition_anchors": "transitions", } for gid in ("L01|main", "L02|main") }, } class _R: def __init__(self, c): self.choices = [type("C", (), {"message": type("M", (), {"content": c})()})] return _R(json.dumps(llm_out)) monkeypatch.setitem(_sys.modules, "litellm", type("FL", (), {"completion": staticmethod(_fake_completion)})) monkeypatch.setenv("GEMINI_API_KEY", "fake") from experiment_background_pipeline_slice import _generate_llm_plate_plan _generate_llm_plate_plan(w6, model="gemini-3.5-flash", source_context=None) payloads_in = captured["user_payload"]["payloads"] # base in high-risk → group_risk_level=high, reference_strength=none assert payloads_in["L01B01"]["group_risk_level"] == "high" assert payloads_in["L01B01"]["reference_strength"] == "none" # derive_from_base_reference variant in high-risk group → reference_strength=weak (override) assert payloads_in["L01B02"]["group_risk_level"] == "high" assert payloads_in["L01B02"]["reference_strength"] == "weak" def test_w7_llm_generate_smoke_with_fake_response(monkeypatch): """Smoke — feed a hand-built LLM response through _generate_llm_plate_plan by monkeypatching litellm.completion. No real network call.""" import experiment_background_pipeline_slice as mod # noqa: F401 import sys as _sys w6 = _fake_w6_plan_for_w7() # Build a valid LLM JSON output covering all bg + groups. llm_out = { "plate_prompts": { bg: { "plate_prompt_text": f"Stable background plate for {bg}, shared spatial anchor.", "consistency_intent": "shared room layout and lighting baseline", "variant_delta": "" if p["render_mode"] == "independent_text_to_image" else "state/time delta only", } for bg, p in w6["payloads"].items() }, "base_plate_group_guides": { g["group_id"]: { "shared_plate_intent": "stable layout shared by all variants", "shared_continuity_constraints": "preserve scale, material, lighting baseline", "variant_delta_policy": "only state/time deltas vary", "auto_strategy": "generate_base_then_variants_with_reference", "source_context_summary": "source summary stub", "scale_and_default_avoidance": "scale/locality/material stub", "must_preserve_spatial_anchors": "spatial anchors stub", "state_transition_anchors": "state transition stub", } for g in w6["groups"] }, } class _FakeResp: def __init__(self, content): self.choices = [type("C", (), {"message": type("M", (), {"content": content})()})] fake_module = type("FL", (), { "completion": staticmethod(lambda **kwargs: _FakeResp(json.dumps(llm_out))), }) monkeypatch.setitem(_sys.modules, "litellm", fake_module) monkeypatch.setenv("GEMINI_API_KEY", "fake") from experiment_background_pipeline_slice import ( _generate_llm_plate_plan, _build_plate_prompt_compatibility_report, ) plan = _generate_llm_plate_plan(w6, model="gemini-3.5-flash") assert plan["generation_status"] == "generated" assert plan["model_used"] == "gemini-3.5-flash" assert set(plan["plate_prompts"].keys()) == set(w6["payloads"].keys()) for bg, entry in plan["plate_prompts"].items(): assert entry["prompt_status"] == "generated", bg assert entry["plate_prompt_text"].startswith("Stable background plate for"), bg # W6 carry preserved assert entry["api_call_shape"] == w6["payloads"][bg]["api_call_shape"] assert entry["reference_images"] == w6["payloads"][bg]["reference_images"] report = _build_plate_prompt_compatibility_report( plan=plan, w6_plan=w6, production_diff_empty=True, db_write_count=0, image_import_seen=False, prev_run_id="fake_w6", missing_inputs=[], ) assert report["all_pass"] is True, report["invariants"] # ───────────────────────────────────────────────────────────────────────────── # W11 — director_set_brief structural tests. # # Use only generic synthetic fixtures. No fixture-specific words (room, prop, # state, color, narrative). Production checkpoint loader test uses the wave's # default project/episode IDs (already used by other wave tests). # ───────────────────────────────────────────────────────────────────────────── def _w11_synthetic_w7_plan() -> dict: return { "plate_prompts": { "L01B01": { "bg_id": "L01B01", "image_title": "t01", "image_model": "gpt-image-2", "render_mode": "independent_text_to_image", "api_call_shape": {}, "reference_images": [], "expected_asset": {}, "applies_to_shots": ["S1_Shot1"], "plate_prompt_text": "synthetic base plate", "prompt_role": "base_plate", "reference_strength": "none", "group_risk_level": "low", "consistency_intent": "", "variant_delta": "", "prompt_status": "generated", }, "L01B02": { "bg_id": "L01B02", "image_title": "t02", "image_model": "gpt-image-2", "render_mode": "derive_from_base_reference", "api_call_shape": {}, "reference_images": [], "expected_asset": {}, "applies_to_shots": ["S1_Shot2"], "plate_prompt_text": "synthetic variant plate", "prompt_role": "strict_variant_delta", "reference_strength": "strict", "group_risk_level": "low", "consistency_intent": "", "variant_delta": "", "prompt_status": "generated", }, }, "groups": [ {"group_id": "G1", "base_bg_id": "L01B01", "member_bg_ids": ["L01B01", "L01B02"], "loc_id": "L01", "space_key": "S01", "shot_keys": ["S1_Shot1", "S1_Shot2"]}, ], } def _w11_synthetic_adapter() -> dict: return { "background_catalog": { "L01B01": {"depends_on_fp": ["FP1"], "depends_on_bg": [], "sub_location_label": "primary_zone", "state_label_raw": "base_state", "loc_id": "L01", "space_key": "S01", "time_phase": "T1", "state_class": "neutral"}, "L01B02": {"depends_on_fp": ["FP1"], "depends_on_bg": ["L01B01"], "sub_location_label": "primary_zone", "state_label_raw": "alt_state", "loc_id": "L01", "space_key": "S01", "time_phase": "T1", "state_class": "altered"}, } } def _w11_synthetic_fp_context() -> dict: return { "fp_prompt_path": "/fake/fp_prompt", "fp_render_path": "/fake/fp_render", "fp_prompt_status": "ok", "fp_render_status": "ok", "floor_plans": { "FP1": { "fp_id": "FP1", "group_id": "G1", "depends_on_fp": [], "key_elements": ["k1"], "numbered_elements": [ {"number": 1, "category": "area", "label": "zone1", "position_hint": "north"}, {"number": 2, "category": "opening", "label": "zone2", "position_hint": "south"}, {"number": 3, "category": "furniture", "label": "zone3", "position_hint": "center"}, ], "camera_recommendations": [ {"bg_id": "L01B01", "camera_position": "near #1", "camera_height": "eye-level", "lens_hint": "35mm", "framing_notes": ""}, ], "t2i_prompt": "synthetic fp_prompt", "applied_shots": ["S1_Shot1", "S1_Shot2"], "png_path": "/fake/fp.png", "png_exists": False, }, }, } def test_w11_production_floor_plan_context_loads_real_manifest(): """integration-light: read the real production checkpoint for the fixture project/episode that the wave already uses. Structural carry only — no semantic assertions about specific rooms / items / scenarios.""" from experiment_background_pipeline_slice import ( DEFAULT_PROJECT_ID, DEFAULT_EPISODE_ID, _load_production_floor_plan_context, ) ctx = _load_production_floor_plan_context(DEFAULT_PROJECT_ID, DEFAULT_EPISODE_ID) assert ctx["fp_prompt_status"] == "ok", ctx assert ctx["fp_render_status"] == "ok", ctx assert ctx["floor_plans"], "expected at least one fp" sample = next(iter(ctx["floor_plans"].values())) assert isinstance(sample["numbered_elements"], list) assert isinstance(sample["camera_recommendations"], list) assert isinstance(sample["t2i_prompt"], str) assert isinstance(sample["png_exists"], bool) def test_w11_compatibility_report_flags_subset_violation(): """relevant_numbered_elements ⊆ fp.numbered_elements.number — synthetic fixture with one valid bg and one violating bg.""" from experiment_background_pipeline_slice import _build_w11_compatibility_report w7 = _w11_synthetic_w7_plan() adapter = _w11_synthetic_adapter() fp_ctx = _w11_synthetic_fp_context() brief_ok = { "group_set_briefs": { "G1": { "group_id": "G1", "fp_blocks": [{"fp_id": "FP1", "png_available": False, "production_camera_bg_ids": ["L01B01"]}], "spatial_zones": [], "openings_and_transitions": [], "persistent_fixtures": [], "state_transition_zones": [], "camera_axes": [], "default_avoidance": "x", "conflicts_and_assumptions": [], "director_set_summary": "", }, }, "per_bg_set_selections": { "L01B01": {"bg_id": "L01B01", "group_id": "G1", "fp_id": "FP1", "applies_to_shots": ["S1_Shot1"], "relevant_zone_ids": [], "relevant_numbered_elements": [1, 2], "matched_production_camera": None, "camera_axis_used": "ax1", "continuity_intent": "", "state_delta": "", "reconciliation_notes": ""}, "L01B02": {"bg_id": "L01B02", "group_id": "G1", "fp_id": "FP1", "applies_to_shots": ["S1_Shot2"], "relevant_zone_ids": [], "relevant_numbered_elements": [2, 3], "matched_production_camera": None, "camera_axis_used": "ax1", "continuity_intent": "", "state_delta": "", "reconciliation_notes": ""}, }, "final_plate_prompt_candidates": { "L01B01": {"bg_id": "L01B01", "final_plate_prompt_text": "x", "source_grounded_anchors_used": [], "set_brief_grounded": True}, "L01B02": {"bg_id": "L01B02", "final_plate_prompt_text": "x", "source_grounded_anchors_used": [], "set_brief_grounded": True}, }, } rep_ok = _build_w11_compatibility_report( brief=brief_ok, w7_plan=w7, adapter_plan=adapter, fp_context=fp_ctx, production_diff_empty=True, db_write_count=0, image_import_seen=False, prev_run_id="fakeW7", missing_inputs=[], set_brief_status="generated", ) assert rep_ok["all_pass"] is True, rep_ok["invariants"] brief_bad = json.loads(json.dumps(brief_ok)) brief_bad["per_bg_set_selections"]["L01B02"]["relevant_numbered_elements"] = [2, 99] rep_bad = _build_w11_compatibility_report( brief=brief_bad, w7_plan=w7, adapter_plan=adapter, fp_context=fp_ctx, production_diff_empty=True, db_write_count=0, image_import_seen=False, prev_run_id="fakeW7", missing_inputs=[], set_brief_status="generated", ) assert rep_bad["invariants"]["per_bg_relevant_numbered_elements_subset_of_fp"]["pass"] is False assert rep_bad["all_pass"] is False def test_w11_placeholder_dry_run_emits_full_shells_and_passes_invariants(): """Dry-run path: every plate bg has a per_bg + candidate shell. grounded invariant is exempt (status=placeholder_dry_run). subset invariant passes because relevant_numbered_elements stays [].""" from experiment_background_pipeline_slice import ( _build_placeholder_director_set_brief, _build_w11_compatibility_report, ) w7 = _w11_synthetic_w7_plan() adapter = _w11_synthetic_adapter() fp_ctx = _w11_synthetic_fp_context() brief = _build_placeholder_director_set_brief(w7, adapter, fp_ctx) plate_keys = set(w7["plate_prompts"].keys()) assert set(brief["per_bg_set_selections"].keys()) == plate_keys assert set(brief["final_plate_prompt_candidates"].keys()) == plate_keys assert all(not c["set_brief_grounded"] for c in brief["final_plate_prompt_candidates"].values()) for bg_id, sel in brief["per_bg_set_selections"].items(): assert sel["fp_id"] == "FP1", bg_id rep = _build_w11_compatibility_report( brief=brief, w7_plan=w7, adapter_plan=adapter, fp_context=fp_ctx, production_diff_empty=True, db_write_count=0, image_import_seen=False, prev_run_id="fakeW7", missing_inputs=[], set_brief_status="placeholder_dry_run", ) assert rep["all_pass"] is True, rep["invariants"] assert rep["invariants"]["final_plate_prompt_candidates_set_brief_grounded"]["pass"] is True def test_w11_html_renders_required_sections_in_order(tmp_path): """HTML section order: trust hierarchy → group brief → per-bg → candidates → fp context. No image section.""" from experiment_background_pipeline_slice import ( _build_placeholder_director_set_brief, _build_w11_compatibility_report, _render_w11_html, ) w7 = _w11_synthetic_w7_plan() adapter = _w11_synthetic_adapter() fp_ctx = _w11_synthetic_fp_context() brief = _build_placeholder_director_set_brief(w7, adapter, fp_ctx) rep = _build_w11_compatibility_report( brief=brief, w7_plan=w7, adapter_plan=adapter, fp_context=fp_ctx, production_diff_empty=True, db_write_count=0, image_import_seen=False, prev_run_id="fakeW7", missing_inputs=[], set_brief_status="placeholder_dry_run", ) run_meta = { "run_id": "RID1", "stage": "w11_director_set_brief", "run_status": "succeeded", "exit_code": 0, "derived_from": "fakeW7", "model_used": None, "image_generation_count": 0, "image_generation_backend": "gpt-image-2", } _render_w11_html(run_meta, brief, rep, fp_ctx, tmp_path) html = (tmp_path / "index.html").read_text() sections_in_order = [ "Trust hierarchy", "Invariants", "Director group briefs", "Per-bg set selections", "Final plate prompt candidates", "Floor-plan context resolution", ] positions = [html.find(s) for s in sections_in_order] assert all(p > 0 for p in positions), positions assert positions == sorted(positions), positions assert " 0 and strong_pos > 0 and weak_pos > 0 assert hard_pos < strong_pos < weak_pos