"""W19E5b dry-run + W19E6 generate-chain harness + W19E7 wrapper tests + W19E8 opt-in surface tests. W20E7-A NOTE: this whole experimental script (and the matching test module) drives the W19B-3 ``w18j_overlap`` opt-in production path. That path is now fail-closed deprecated in production (see ``BackgroundRenderStep._execute``). The script still imports the orphaned ``background_image_planner`` and exercises its deterministic top-K logic. The orphan module + this experimental script are slated for a later cleanup wave; in the meantime the module-level skip below keeps CI clean without touching the script itself. Original test list (skipped pending cleanup): 1. dry-run smoke (api counters 0, seam counts 0) 2. generate default seams RuntimeError (first seam fail-closed, rest 0) 3. AST/token ban: DB/ImageAsset write, OpenAI client import/creation 4. production_diff_audit surfaces approved scope without failing dry-run 5. generate + dirty production diff: RuntimeError, seam calls 0 6. generate monkeypatch clean fake-seam chain: exact counts (4 LLM + 4 image) 7. generate unmatched chain_bg gate: RuntimeError, seam calls 0 8. (W19E7) generate via production wrappers + injected fake call_structured_fn + fake image client: exact counts + artifact bg_id set. 9. (W19E8) dry-run surfaces real_api_gate disabled_default + api_call_caps. 10. (W19E8) --allow-real-api without env stays real_api_mode=False (disabled_env_only). 11. (W19E8) env without --allow-real-api stays real_api_mode=False (disabled_flag_only). 12. (W19E8) both flags+env set → real resolver fail-closed, all counters 0. 13. (W19E8) hard cap breach aborts before seam invocation. """ from __future__ import annotations import ast import json import sys from pathlib import Path import pytest # W20E7-A: this experimental script targets the deprecated W19B-3 # ``w18j_overlap`` reference planner path. Skipping the whole module # until the orphan-removal follow-up wave; the script and the # ``background_image_planner`` module remain on disk untouched. pytestmark = pytest.mark.skip( reason=( "W20E7-A: w18j_overlap path is fail-closed deprecated in " "production; this experimental script's tests still exercise " "the orphaned W19B-3 planner. Quarantined pending the orphan-" "removal follow-up wave." ) ) _REPO_ROOT = Path(__file__).resolve().parents[3] _SCRIPTS_DIR = _REPO_ROOT / "backend" / "scripts" if str(_SCRIPTS_DIR) not in sys.path: sys.path.insert(0, str(_SCRIPTS_DIR)) def test_dry_run_smoke_default_target_api_counters_zero(tmp_path): import experiment_w19e_production_opt_in_single_fp_smoke as mod out_root = tmp_path / "out" exit_code = mod.main([ "--output-root", str(out_root), ]) assert exit_code == 0, "dry-run should pass with default checkpoints" runs = sorted(out_root.iterdir()) assert len(runs) == 1 run_dir = runs[0] meta = json.loads((run_dir / "run_meta.json").read_text(encoding="utf-8")) assert meta["stage_status"] == "dry_run" assert meta["run_status"] == "succeeded" assert meta["exit_code"] == 0 counters = meta["api_call_counters"] assert counters["llm_api_call_count"] == 0 assert counters["image_api_call_count"] == 0 assert counters["vlm_api_call_count"] == 0 assert counters["db_write_count"] == 0 assert counters["image_asset_write_count"] == 0 # W19E6: seam_attempt_counts 모두 0 in dry-run. seams = meta["seam_attempt_counts"] assert all(v == 0 for v in seams.values()), seams assert meta["generated_artifacts"] == {} assert meta["planned_call_counts"]["llm_total"] == 1 + 3 assert meta["planned_call_counts"]["image_total"] == 1 + 3 gate = meta["background_gate_provenance"] assert gate["kind"] == "chain_bg" assert gate["matches_chain_bg"] is True assert (run_dir / "index.html").exists() assert meta["production_diff_guard_mode"] == "diagnostic_in_dry_run" assert meta["generate_requires_clean_production_diff"] is True def test_generate_default_seam_raises_first_seam_only(tmp_path): """--generate 시 production_diff clean (conftest fake) + matches_chain_bg True 통과 후, 첫 seam (_run_floor_plan_prompt_v6) default RuntimeError. 이후 seam 들은 호출 0. """ import experiment_w19e_production_opt_in_single_fp_smoke as mod out_root = tmp_path / "out" with pytest.raises(RuntimeError, match="_run_floor_plan_prompt_v6"): mod.main([ "--output-root", str(out_root), "--generate", ]) _BANNED_TOKENS = ( "session.add(", "session.commit(", "session.flush(", "session.delete(", "self.db.add(", "self.db.commit(", "ImageAsset(", "INSERT ", "UPDATE ", "DELETE FROM", "openai.OpenAI(", "OpenAI()", "from openai import", "import openai", ) def test_script_has_no_db_or_image_asset_write_surface(): src_path = _SCRIPTS_DIR / "experiment_w19e_production_opt_in_single_fp_smoke.py" body = src_path.read_text(encoding="utf-8") leaks = sorted(t for t in _BANNED_TOKENS if t in body) assert not leaks, ( f"W19E5/E6 script contains DB/ImageAsset/API write surface: {leaks}" ) tree = ast.parse(body) banned_imports = {"openai", "app.models.project", "app.modules.llm.llm_client"} found = set() for node in ast.walk(tree): if isinstance(node, ast.Import): for alias in node.names: if alias.name in banned_imports: found.add(alias.name) elif isinstance(node, ast.ImportFrom): if node.module in banned_imports: found.add(node.module) assert not found, f"W19E5/E6 script imports forbidden symbols: {sorted(found)}" def test_production_diff_false_does_not_fail_dry_run(tmp_path, monkeypatch): import experiment_w19e_production_opt_in_single_fp_smoke as mod monkeypatch.setattr( mod, "_check_clean_or_approved_w19_diff", lambda: { "is_clean_or_approved": False, "violations": ["backend/app/some/unapproved.py"], "alembic_violations": [], "approved_paths_seen": [], "all_diff_paths": ["backend/app/some/unapproved.py"], }, ) out_root = tmp_path / "out" exit_code = mod.main(["--output-root", str(out_root)]) assert exit_code == 0 runs = sorted(out_root.iterdir()) meta = json.loads((runs[0] / "run_meta.json").read_text(encoding="utf-8")) assert meta["production_diff_empty"] is False assert meta["production_diff_audit"]["violations"] == [ "backend/app/some/unapproved.py" ] assert meta["production_diff_guard_mode"] == "diagnostic_in_dry_run" assert meta["generate_requires_clean_production_diff"] is True assert "production_diff_dirty" not in meta["failed_invariants"] def _install_seam_call_counters(monkeypatch, mod, *, raise_on_call: bool = False): """Replace 5 generate-chain seams with counting fakes; raise if asked.""" counts = { "_run_floor_plan_prompt_v6": 0, "_render_floor_plan_png": 0, "_build_overlay_payload": 0, "_run_background_prompt_v7": 0, "_render_background_png": 0, } def _make(name): def _fake(**kwargs): counts[name] += 1 if raise_on_call: raise AssertionError( f"seam {name} called unexpectedly: kwargs={list(kwargs)}" ) # Return minimal synthetic payload to keep the chain going. if name == "_run_floor_plan_prompt_v6": return { "t2i_prompt": "FAKE", "key_elements": [], "numbered_elements": [], "camera_recommendations": [], } if name == "_render_floor_plan_png": return {"png_path": "/tmp/fake/fp.png", "status": "ok", "attempts": 1} if name == "_build_overlay_payload": return { bid: {"bg_id": bid, "fp_id": "fp_police_office_main", "base_markers_to_reference": [], "transient_markers_to_describe": [], "ignored_state_overlay_markers": [], "target_unit_marker_numbers": [], "dominant_target_unit_marker_number": None, "clean_background_expected": True, "use_numbered_elements": [], "ignore_numbered_elements": [], "diagnostics": []} for bid in ("L14B01", "L14B02", "L14B03") } if name == "_run_background_prompt_v7": return {"t2i_prompt": "FAKE-BG", "shot_guides": [], "objects_owned_by_background": ["door"]} if name == "_render_background_png": return {"png_path": f"/tmp/fake/{kwargs.get('bg_id', 'unk')}.png", "status": "ok", "attempts": 1, "reference_decision": {}} return {} return _fake for name in list(counts): monkeypatch.setattr(mod, name, _make(name)) return counts def test_generate_dirty_production_diff_fails_before_seam_calls(tmp_path, monkeypatch): import experiment_w19e_production_opt_in_single_fp_smoke as mod monkeypatch.setattr( mod, "_check_clean_or_approved_w19_diff", lambda: { "is_clean_or_approved": False, "violations": ["backend/app/some/unapproved.py"], "alembic_violations": ["backend/alembic/versions/zzz.py"], "approved_paths_seen": [], "all_diff_paths": [ "backend/alembic/versions/zzz.py", "backend/app/some/unapproved.py", ], }, ) counts = _install_seam_call_counters(monkeypatch, mod, raise_on_call=True) out_root = tmp_path / "out" with pytest.raises(RuntimeError, match="production_diff_dirty"): mod.main(["--output-root", str(out_root), "--generate"]) assert all(v == 0 for v in counts.values()), counts def test_generate_unmatched_chain_bg_fails_before_seam_calls(tmp_path, monkeypatch): import experiment_w19e_production_opt_in_single_fp_smoke as mod original_summarize = mod._summarize_gate def _fake_summarize(*args, **kwargs): gate = original_summarize(*args, **kwargs) # Force the gate to NOT match chain_bg. gate["matches_chain_bg"] = False gate["kind"] = "prev_shot_ref" return gate monkeypatch.setattr(mod, "_summarize_gate", _fake_summarize) counts = _install_seam_call_counters(monkeypatch, mod, raise_on_call=True) out_root = tmp_path / "out" with pytest.raises(RuntimeError, match="chain_bg"): mod.main(["--output-root", str(out_root), "--generate"]) assert all(v == 0 for v in counts.values()), counts def test_generate_clean_fake_seam_chain_exact_counts(tmp_path, monkeypatch): import experiment_w19e_production_opt_in_single_fp_smoke as mod counts = _install_seam_call_counters(monkeypatch, mod, raise_on_call=False) out_root = tmp_path / "out" exit_code = mod.main(["--output-root", str(out_root), "--generate"]) assert exit_code == 0 assert counts == { "_run_floor_plan_prompt_v6": 1, "_render_floor_plan_png": 1, "_build_overlay_payload": 1, "_run_background_prompt_v7": 3, "_render_background_png": 3, } runs = sorted(out_root.iterdir()) meta = json.loads((runs[0] / "run_meta.json").read_text(encoding="utf-8")) assert meta["stage_status"] == "generated" # run_meta seam keys 는 short form (no leading underscore). assert meta["seam_attempt_counts"] == { "floor_plan_prompt_v6": 1, "floor_plan_render_png": 1, "build_overlay_payload": 1, "background_prompt_v7": 3, "background_render_png": 3, } api = meta["api_call_counters"] assert api["llm_api_call_count"] == 4 assert api["image_api_call_count"] == 4 assert api["vlm_api_call_count"] == 0 assert api["db_write_count"] == 0 assert api["image_asset_write_count"] == 0 ga = meta["generated_artifacts"] assert ga["floor_plan_prompt"]["t2i_prompt"] == "FAKE" assert ga["floor_plan_render"]["status"] == "ok" assert len(ga["backgrounds"]) == 3 assert {b["bg_id"] for b in ga["backgrounds"]} == {"L14B01", "L14B02", "L14B03"} # ───── W19E7 production-wrapper-wired fake-generate ──────────────────── def _fake_image_response(): """Build a fake OpenAI images response with a minimal b64 PNG payload.""" import base64 class _Datum: def __init__(self): self.b64_json = base64.b64encode(b"\x89PNG_w19e7_fake").decode("ascii") class _Resp: def __init__(self): self.data = [_Datum()] return _Resp() def _extract_shots_from_bg_user_prompt(user_prompt: str): """Pull bullet list from the v7 ``## Applies to shots`` block.""" import re m = re.search( r"## Applies to shots\n(.*?)\n## ", user_prompt, re.DOTALL, ) if not m: return [] return [ line[2:].strip() for line in m.group(1).splitlines() if line.startswith("- ") ] def _make_fake_call_structured(counter, *, expected_fp_id="fp_police_office_main"): """LLM stub satisfying floor_plan_prompt v6 + background_prompt v7 validators. Generic prose only — no scenario-specific wording. Floor-plan output includes ONE ``base_structural_unit`` marker plus ONE ``base_opening`` marker, both referenced by every camera. This makes each per-bg overlay carry a non-empty ``target_unit_marker_numbers`` + ``base_markers_to_reference``, so once the live catalog grows after the first BG render, later BGs share enough markers to escape ``fp_seeded_anchor`` and become ``reference_derived``. """ def fake(*, step, system_prompt, user_prompt, response_schema, **_): counter["llm"] += 1 if step == "floor_plan_prompt": cam_enum = ( response_schema.get("properties", {}) .get("camera_recommendations", {}) .get("items", {}) .get("properties", {}) .get("bg_id", {}) .get("enum") or [] ) return { "fp_id": expected_fp_id, "t2i_prompt": ( "Top-down architectural floor-plan diagram of the target " "interior, numbered markers visible, ascii-only test fixture." ), "key_elements": ["primary room", "main entry"], "numbered_elements": [ { "number": 1, "label": "primary interior unit", "category": "room", "position_hint": "central", "base_layer_decision": "base_structural_unit", }, { "number": 2, "label": "main entry opening", "category": "opening", "position_hint": "north wall", "base_layer_decision": "base_opening", }, ], "camera_recommendations": [ { "bg_id": bg, "camera_position": "north corner", "camera_height": "eye level", "lens_hint": "35mm", "framing_notes": "wide establishing shot", # Both markers consumed by every cam → each overlay # payload carries shared unit + base marker numbers, # which is what lets the live catalog feed real # reference_derived overlap into later BGs. "use_numbered_elements": [1, 2], "ignore_numbered_elements": [], } for bg in cam_enum ], } if step == "background_prompt": bg_enum = ( response_schema.get("properties", {}) .get("bg_id", {}) .get("enum") or [] ) bg_id = bg_enum[0] if bg_enum else "" shots = _extract_shots_from_bg_user_prompt(user_prompt) return { "bg_id": bg_id, "t2i_prompt": ( "Photoreal architectural still of the target interior, " "ascii-only fixture content for w19e7 smoke verification." ), "shot_guides": [ {"shot_id": s, "guide_text": "ascii guide"} for s in shots ], "objects_owned_by_background": ["door"], } return {} return fake def _install_wrapper_seams( monkeypatch, mod, *, call_structured_fn, openai_image_client, ): """Redirect each default seam to its production-wrapper helper, threading the injected fake ``call_structured_fn`` / image client through.""" counts = { "_run_floor_plan_prompt_v6": 0, "_render_floor_plan_png": 0, "_build_overlay_payload": 0, "_run_background_prompt_v7": 0, "_render_background_png": 0, } def fp_prompt_seam(*, fp_id, fp_spec, applied_bg_specs, run_dir, **_): counts["_run_floor_plan_prompt_v6"] += 1 applied_shots = list(dict.fromkeys( s for bg in applied_bg_specs for s in (bg.get("applies_to_shots") or []) )) return mod._wrap_floor_plan_prompt_v6( fp_id=fp_id, fp_spec=fp_spec, applied_bg_specs=applied_bg_specs, applied_shots=applied_shots, scene_segments=[], visual_world_rules="", call_structured_fn=call_structured_fn, run_dir=run_dir, ) def fp_render_seam(*, fp_id, fp_prompt_result, run_dir, **_): counts["_render_floor_plan_png"] += 1 return mod._wrap_floor_plan_render_png( fp_id=fp_id, fp_prompt_result=fp_prompt_result, openai_image_client=openai_image_client, image_model="gpt-image-2", run_dir=run_dir, ) def overlay_seam(*, fp_id, fp_prompt_result, bg_specs, **_): counts["_build_overlay_payload"] += 1 return mod._wrap_build_overlay_payload( fp_id=fp_id, fp_prompt_result=fp_prompt_result, bg_specs=bg_specs, ) def bg_prompt_seam( *, bg_id, bg_spec, fp_prompt_result, overlay_payload_bg, run_dir, **_ ): counts["_run_background_prompt_v7"] += 1 return mod._wrap_background_prompt_v7( bg_id=bg_id, bg_spec=bg_spec, fp_prompt_result=fp_prompt_result, overlay_payload_bg=overlay_payload_bg, scene_segments=[], visual_world_rules="", source_language="ko", floor_plan_path="", prior_bg_paths=[], call_structured_fn=call_structured_fn, run_dir=run_dir, ) def bg_render_seam( *, bg_id, bg_prompt_result, overlay_payload_bg, catalog, base_fp_png, run_dir, **_, ): counts["_render_background_png"] += 1 return mod._wrap_background_render_png( bg_id=bg_id, bg_prompt_result=bg_prompt_result, overlay_payload_bg=overlay_payload_bg, catalog=catalog, base_fp_png=base_fp_png, openai_image_client=openai_image_client, image_model="gpt-image-2", run_dir=run_dir, ) monkeypatch.setattr(mod, "_run_floor_plan_prompt_v6", fp_prompt_seam) monkeypatch.setattr(mod, "_render_floor_plan_png", fp_render_seam) monkeypatch.setattr(mod, "_build_overlay_payload", overlay_seam) monkeypatch.setattr(mod, "_run_background_prompt_v7", bg_prompt_seam) monkeypatch.setattr(mod, "_render_background_png", bg_render_seam) return counts def test_generate_via_production_wrappers_with_injected_fakes(tmp_path, monkeypatch): """W19E7 — production wrappers fired with fake LLM + fake image client. No real API. Verifies exact seam counts + bg_id artifact set. """ import experiment_w19e_production_opt_in_single_fp_smoke as mod llm_counter = {"llm": 0} fake_call_structured = _make_fake_call_structured(llm_counter) image_counter = {"image": 0} class _FakeImagesAPI: def generate(self, **kw): image_counter["image"] += 1 return _fake_image_response() def edit(self, **kw): image_counter["image"] += 1 return _fake_image_response() class _FakeOpenAIClient: def __init__(self): self.images = _FakeImagesAPI() fake_client = _FakeOpenAIClient() counts = _install_wrapper_seams( monkeypatch, mod, call_structured_fn=fake_call_structured, openai_image_client=fake_client, ) out_root = tmp_path / "out" exit_code = mod.main(["--output-root", str(out_root), "--generate"]) assert exit_code == 0 assert counts == { "_run_floor_plan_prompt_v6": 1, "_render_floor_plan_png": 1, "_build_overlay_payload": 1, "_run_background_prompt_v7": 3, "_render_background_png": 3, } assert llm_counter["llm"] == 4 # 1 fp prompt + 3 bg prompts assert image_counter["image"] == 4 # 1 fp render + 3 bg renders runs = sorted(out_root.iterdir()) meta = json.loads((runs[0] / "run_meta.json").read_text(encoding="utf-8")) assert meta["stage_status"] == "generated" assert meta["seam_attempt_counts"] == { "floor_plan_prompt_v6": 1, "floor_plan_render_png": 1, "build_overlay_payload": 1, "background_prompt_v7": 3, "background_render_png": 3, } api = meta["api_call_counters"] assert api["llm_api_call_count"] == 4 assert api["image_api_call_count"] == 4 assert api["vlm_api_call_count"] == 0 assert api["db_write_count"] == 0 assert api["image_asset_write_count"] == 0 ga = meta["generated_artifacts"] assert ga["floor_plan_render"]["status"] == "ok" assert {b["bg_id"] for b in ga["backgrounds"]} == { "L14B01", "L14B02", "L14B03", } # PNG files were written by render_one_floor_plan / render_one_background # via the fake b64 path. Each artifact carries a png_path string. fp_png = ga["floor_plan_render"]["png_path"] assert fp_png and Path(fp_png).exists() for bg in ga["backgrounds"]: rr = bg["render_result"] assert rr.get("status") == "ok" assert Path(rr["png_path"]).exists() # background_image_planner decision recorded on each render result. assert rr["reference_decision"]["bg_id"] == bg["bg_id"] # W19E7-R1 G1: catalog actually grew via production ``make_catalog_entry`` # — final length matches the rendered BG count and the live catalog # snapshot is surfaced in run_meta at the top level. assert meta["catalog_final_len"] == 3 assert meta["rendered_bg_count"] == 3 assert meta["catalog_final_len"] == meta["rendered_bg_count"] assert ga["catalog_final_len"] == 3 assert ga["rendered_bg_count"] == 3 entries = ga["catalog_entries"] assert len(entries) == 3 # Catalog entry order follows iteration order of bg_specs from the master # plan checkpoint (not lexical bg_id order). The invariants we care about: # (a) every target bg_id appears exactly once; (b) ingestion_order is the # natural enumeration; (c) every entry carries the structural-unit marker # set from the fake floor-plan output. assert {e["bg_id"] for e in entries} == {"L14B01", "L14B02", "L14B03"} assert all(e["fp_id"] == "fp_police_office_main" for e in entries) assert all(e["unit_marker_set"] == [1] for e in entries) assert all(e["base_marker_set"] == [1, 2] for e in entries) assert [e["ingestion_order"] for e in entries] == [0, 1, 2] # Reference graph evidence: the first BG processed hits an empty catalog # so it must fall back to fp_seeded_anchor; subsequent BGs see the # freshly-appended catalog and must decide reference_derived (strong # overlap on unit + base markers). If catalog growth were missing this # set would collapse to {fp_seeded_anchor} only. Order-robust assertions # walk the BGs in the same order the generator processed them. bg_iteration_order = [b["bg_id"] for b in ga["backgrounds"]] decisions = { b["bg_id"]: b["render_result"]["reference_decision"] for b in ga["backgrounds"] } first_bg = bg_iteration_order[0] later_bgs = bg_iteration_order[1:] assert decisions[first_bg]["mode"] == "fp_seeded_anchor" assert decisions[first_bg]["source_bg_ids"] == [] for later in later_bgs: d = decisions[later] assert d["mode"] == "reference_derived", ( f"BG {later} must escape fp_seeded_anchor when the live " f"catalog already contains {bg_iteration_order[: bg_iteration_order.index(later)]}; " f"got mode={d['mode']!r}" ) assert d["source_bg_ids"], ( f"BG {later} reference_derived must list prior catalog refs" ) # source_bg_ids must be a subset of the BGs catalogged before this one. prior = set(bg_iteration_order[: bg_iteration_order.index(later)]) assert set(d["source_bg_ids"]).issubset(prior) # ref_modes_observed surfaces the union of modes across BGs. assert set(meta["ref_modes_observed"]) == { "fp_seeded_anchor", "reference_derived", } # ───── W19E8 opt-in surface tests ───────────────────────────────────── _EXPECTED_W19E8_CAPS = { "llm_api_call_count": 4, "image_api_call_count": 4, "vlm_api_call_count": 0, "db_write_count": 0, "image_asset_write_count": 0, } def _read_meta(out_root): runs = sorted(out_root.iterdir()) assert runs, "no run dir written" return json.loads((runs[0] / "run_meta.json").read_text(encoding="utf-8")) def test_dry_run_surfaces_w19g_preflight_audit_and_readiness(tmp_path, monkeypatch): """W19G preflight: dry-run carries context_checkpoint_audit, target_resolution, and real_smoke_readiness so reviewers can verify real-smoke preconditions without descending into JSON. Default target on this episode has world_guide MISSING (WARNING, not BLOCKER) and floor_plan_overlay_payload pre-existing optional. """ import experiment_w19e_production_opt_in_single_fp_smoke as mod monkeypatch.delenv(mod.W19E8_REAL_API_ENV_VAR, raising=False) out_root = tmp_path / "out" exit_code = mod.main(["--output-root", str(out_root)]) assert exit_code == 0 meta = _read_meta(out_root) ctx = meta["context_checkpoint_audit"] assert ctx["scene_save"]["present"] is True assert ctx["scene_save"]["segments_count"] > 0 assert ctx["visual_world_rules"]["present"] is True assert ctx["visual_world_rules"]["source_language"] # non-empty assert ctx["visual_world_rules"]["rules_count"] >= 0 # world_guide is intentionally absent on this episode — WARNING only. assert ctx["world_guide"]["present"] is False # floor_plan_overlay_payload pre-existence is informational only. assert "present" in ctx["floor_plan_overlay_payload"] tr = meta["target_resolution"] assert tr["fp_id_resolved"] is True assert tr["resolved_bg_count"] == 3 assert tr["missing_bg_ids"] == [] assert tr["resolved_group_id"] # non-empty rs = meta["real_smoke_readiness"] # No BLOCKERS on the default target — diff clean, gate matches, target # resolved, scene_save + visual_world_rules present. assert rs["blockers"] == [] assert "world_guide_missing" in rs["warnings"] assert rs["ready_for_user_approval"] is True assert rs["explicit_user_approval_required_before_real_run"] is True # HTML chrome surfaces the readiness verdict. runs = sorted(out_root.iterdir()) body = (runs[0] / "index.html").read_text(encoding="utf-8") assert "real_smoke_ready=YES" in body def test_dry_run_real_smoke_readiness_blockers_chain_bg_mismatch(tmp_path, monkeypatch): """W19G preflight: when the gate doesn't match chain_bg, readiness must record ``background_gate_mismatch`` and ``ready_for_user_approval`` must be False.""" import experiment_w19e_production_opt_in_single_fp_smoke as mod original_summarize = mod._summarize_gate def _fake_summarize(*args, **kwargs): gate = original_summarize(*args, **kwargs) gate["matches_chain_bg"] = False gate["kind"] = "prev_shot_ref" return gate monkeypatch.setattr(mod, "_summarize_gate", _fake_summarize) out_root = tmp_path / "out" # Dry-run still exits 1 because failed_invariants picks up the gate # mismatch (existing W19E6 contract). We assert on the readiness path. exit_code = mod.main(["--output-root", str(out_root)]) assert exit_code == 1 meta = _read_meta(out_root) rs = meta["real_smoke_readiness"] assert "background_gate_mismatch" in rs["blockers"] assert rs["ready_for_user_approval"] is False runs = sorted(out_root.iterdir()) body = (runs[0] / "index.html").read_text(encoding="utf-8") assert "real_smoke_ready=NO" in body def test_dry_run_surfaces_real_api_gate_disabled_default_and_caps(tmp_path, monkeypatch): """W19E8-D: dry-run run_meta carries real_api_gate state + caps even when --generate is not used.""" import experiment_w19e_production_opt_in_single_fp_smoke as mod monkeypatch.delenv(mod.W19E8_REAL_API_ENV_VAR, raising=False) out_root = tmp_path / "out" exit_code = mod.main(["--output-root", str(out_root)]) assert exit_code == 0 meta = _read_meta(out_root) assert meta["real_api_mode"] is False assert meta["real_api_gate_status"] == "disabled_default" assert meta["real_api_gate"]["cli_flag"] is False assert meta["real_api_gate"]["env_flag"] is False assert meta["api_call_caps"] == _EXPECTED_W19E8_CAPS # No PNGs written in dry-run; artifact audit records each as absent. aa = meta["output_artifacts_audit"] assert aa["floor_plan_png"]["present"] is False assert all(bg["present"] is False for bg in aa["background_pngs"]) # HTML carries real_api_mode label. html_path = (sorted(out_root.iterdir())[0] / "index.html") body = html_path.read_text(encoding="utf-8") assert "real_api_mode=OFF" in body assert "[dry_run]" in body def test_flag_only_keeps_real_api_off_disabled_env_only(tmp_path, monkeypatch): """W19E8-B: --allow-real-api without env stays fake-safe.""" import experiment_w19e_production_opt_in_single_fp_smoke as mod monkeypatch.delenv(mod.W19E8_REAL_API_ENV_VAR, raising=False) counts = _install_seam_call_counters(monkeypatch, mod, raise_on_call=False) out_root = tmp_path / "out" exit_code = mod.main([ "--output-root", str(out_root), "--generate", "--allow-real-api", ]) assert exit_code == 0 meta = _read_meta(out_root) assert meta["real_api_mode"] is False assert meta["real_api_gate_status"] == "disabled_env_only" # Fake-seam path ran normally — exact counts preserved. assert counts == { "_run_floor_plan_prompt_v6": 1, "_render_floor_plan_png": 1, "_build_overlay_payload": 1, "_run_background_prompt_v7": 3, "_render_background_png": 3, } assert meta["api_call_counters"]["llm_api_call_count"] == 4 assert meta["api_call_counters"]["image_api_call_count"] == 4 def test_env_only_keeps_real_api_off_disabled_flag_only(tmp_path, monkeypatch): """W19E8-B: env without --allow-real-api stays fake-safe.""" import experiment_w19e_production_opt_in_single_fp_smoke as mod monkeypatch.setenv( mod.W19E8_REAL_API_ENV_VAR, mod.W19E8_REAL_API_ENV_VALUE ) counts = _install_seam_call_counters(monkeypatch, mod, raise_on_call=False) out_root = tmp_path / "out" exit_code = mod.main([ "--output-root", str(out_root), "--generate", ]) assert exit_code == 0 meta = _read_meta(out_root) assert meta["real_api_mode"] is False assert meta["real_api_gate_status"] == "disabled_flag_only" assert counts["_run_floor_plan_prompt_v6"] == 1 def test_both_gates_real_api_resolver_failure_writes_blocked_artifact( tmp_path, monkeypatch ): """W19E9-B failure recovery path: both flag+env set, but the lazy resolver raises (simulated by monkeypatch). The run still completes by writing ``run_meta`` + ``index`` so the blocked-path evidence survives for offline review. exit_code is 1 (run_status=real_api_blocked). No seam/wrapper is invoked → counters all zero.""" import experiment_w19e_production_opt_in_single_fp_smoke as mod monkeypatch.setenv( mod.W19E8_REAL_API_ENV_VAR, mod.W19E8_REAL_API_ENV_VALUE ) def _boom_llm(): raise RuntimeError( f"{mod._W19E9_RESOLVER_FAILURE_PREFIX}: llm: simulated failure" ) def _boom_image(): raise RuntimeError( f"{mod._W19E9_RESOLVER_FAILURE_PREFIX}: image: simulated failure" ) monkeypatch.setattr(mod, "_resolve_real_call_structured", _boom_llm) monkeypatch.setattr(mod, "_resolve_real_openai_image_client", _boom_image) # raise_on_call=True on seams so any accidental invocation surfaces. counts = _install_seam_call_counters(monkeypatch, mod, raise_on_call=True) out_root = tmp_path / "out" exit_code = mod.main([ "--output-root", str(out_root), "--generate", "--allow-real-api", ]) assert exit_code == 1 meta = _read_meta(out_root) assert meta["stage_status"] == "real_api_resolver_failed" assert meta["run_status"] == "real_api_blocked" assert meta["real_api_mode"] is True assert meta["real_api_gate_status"] == "enabled" assert "real_api_resolver_failed" in meta["failed_invariants"] # Counters / seam attempts must remain at zero — resolver fail-closed # before any seam touched. assert all(v == 0 for v in meta["seam_attempt_counts"].values()) assert all(v == 0 for v in meta["api_call_counters"].values()) assert all(v == 0 for v in counts.values()), counts # Artifact audit shows nothing was written. aa = meta["output_artifacts_audit"] assert aa["floor_plan_png"]["present"] is False assert all(bg["present"] is False for bg in aa["background_pngs"]) # HTML labels the failure path. runs = sorted(out_root.iterdir()) body = (runs[0] / "index.html").read_text(encoding="utf-8") assert "real_api_resolver_failed" in body assert "real_api_mode=ON" in body def test_both_gates_with_fake_resolvers_executes_full_chain(tmp_path, monkeypatch): """W19E9-B success path. Both gates ON + resolvers monkeypatched to return fake call_structured_fn + fake image client → main() drives the production wrappers directly (real-mode chain), no fake seam injection used. Verifies seam attempts 1/1/1/3/3, api counters llm=4/image=4, catalog growth via make_catalog_entry, ref_modes including reference_derived after the first catalog entry lands. Real external API count remains zero — the resolvers return fakes.""" import experiment_w19e_production_opt_in_single_fp_smoke as mod monkeypatch.setenv( mod.W19E8_REAL_API_ENV_VAR, mod.W19E8_REAL_API_ENV_VALUE ) llm_counter = {"llm": 0} fake_call_structured = _make_fake_call_structured(llm_counter) image_counter = {"image": 0} class _FakeImagesAPI: def generate(self, **kw): image_counter["image"] += 1 return _fake_image_response() def edit(self, **kw): image_counter["image"] += 1 return _fake_image_response() class _FakeOpenAIClient: def __init__(self): self.images = _FakeImagesAPI() fake_client = _FakeOpenAIClient() monkeypatch.setattr( mod, "_resolve_real_call_structured", lambda: fake_call_structured ) monkeypatch.setattr( mod, "_resolve_real_openai_image_client", lambda: fake_client ) # W19E9-R1: spy ``build_visual_context_block`` to confirm real-mode uses # the production visual-context helper (not a script-local rules-join # shortcut). Spy delegates to the real helper so the resulting string is # production-equivalent. ``extract_source_language`` is exercised # implicitly via the loaded visual_world_rules cp. import app.services.visual_context_helper as vch real_build = vch.build_visual_context_block real_extract_sl = vch.extract_source_language build_calls = [] extract_calls = [] def spy_build(*, world_guide_cp, visual_world_rules_cp): build_calls.append({ "world_guide_cp_present": world_guide_cp is not None, "vwr_cp_present": visual_world_rules_cp is not None, }) return real_build( world_guide_cp=world_guide_cp, visual_world_rules_cp=visual_world_rules_cp, ) def spy_extract(cp): extract_calls.append(cp is not None) return real_extract_sl(cp) monkeypatch.setattr(vch, "build_visual_context_block", spy_build) monkeypatch.setattr(vch, "extract_source_language", spy_extract) out_root = tmp_path / "out" exit_code = mod.main([ "--output-root", str(out_root), "--generate", "--allow-real-api", ]) assert exit_code == 0 # W19E9-R1: production helpers were the SOT for visual context + source # language. exactly one call each (single-fp smoke). assert len(build_calls) == 1 assert build_calls[0]["vwr_cp_present"] is True # world_guide cp may be present or absent depending on target episode; # invariant is that the script attempted to pass it through. assert isinstance(build_calls[0]["world_guide_cp_present"], bool) assert extract_calls == [True] meta = _read_meta(out_root) assert meta["stage_status"] == "generated" assert meta["run_status"] == "succeeded" assert meta["real_api_mode"] is True assert meta["real_api_gate_status"] == "enabled" assert meta["seam_attempt_counts"] == { "floor_plan_prompt_v6": 1, "floor_plan_render_png": 1, "build_overlay_payload": 1, "background_prompt_v7": 3, "background_render_png": 3, } api = meta["api_call_counters"] assert api["llm_api_call_count"] == 4 assert api["image_api_call_count"] == 4 assert api["vlm_api_call_count"] == 0 assert api["db_write_count"] == 0 assert api["image_asset_write_count"] == 0 assert llm_counter["llm"] == 4 assert image_counter["image"] == 4 ga = meta["generated_artifacts"] assert ga["floor_plan_render"]["status"] == "ok" assert {b["bg_id"] for b in ga["backgrounds"]} == { "L14B01", "L14B02", "L14B03", } assert meta["catalog_final_len"] == 3 assert meta["rendered_bg_count"] == 3 assert set(meta["ref_modes_observed"]) == { "fp_seeded_anchor", "reference_derived", } # Each rendered artifact actually written. aa = meta["output_artifacts_audit"] assert aa["floor_plan_png"]["present"] is True assert all(bg["present"] for bg in aa["background_pngs"]) # HTML labels the real-mode run. runs = sorted(out_root.iterdir()) body = (runs[0] / "index.html").read_text(encoding="utf-8") assert "real_api_mode=ON" in body assert "[generated]" in body def test_hard_cap_breach_aborts_before_seam_overrun(tmp_path, monkeypatch): """W19E8-C: lowering the llm cap to 0 makes the pre-call enforcer abort before the first seam is invoked. Cap check protects even when a helper leaks retries.""" import experiment_w19e_production_opt_in_single_fp_smoke as mod # Reduce the llm cap to zero; image cap stays 4. monkeypatch.setitem(mod.W19E8_API_CALL_CAPS, "llm_api_call_count", 0) counts = _install_seam_call_counters(monkeypatch, mod, raise_on_call=True) out_root = tmp_path / "out" with pytest.raises(RuntimeError, match="api_call_cap_exhausted"): mod.main([ "--output-root", str(out_root), "--generate", ]) # Cap fires before any seam wrapper runs. raise_on_call would surface # an AssertionError if the seam were touched. assert all(v == 0 for v in counts.values()), counts def test_retry_helper_semantics_one_attempt_only(): """W19E8-C semantics check: production helpers configured with ``max_retries=1`` / ``max_attempts=1`` must run **exactly one attempt** (not "one retry after the first"). This protects against silent retry budget leaks that could exceed the W19E8 hard caps. Read both helpers' loops via inspection and the actual ``range`` shape they iterate to confirm the contract. """ import inspect from app.modules.pipeline.floor_plan_prompt import run_floor_plan_prompt from app.modules.pipeline.background_prompt import run_background_prompt from app.modules.pipeline.floor_plan_render import render_one_floor_plan from app.modules.pipeline.background_render import render_one_background src_fp_prompt = inspect.getsource(run_floor_plan_prompt) src_bg_prompt = inspect.getsource(run_background_prompt) # Both prompt helpers loop ``for attempt in range(max_retries)`` → # ``max_retries=1`` yields 1 iteration / 1 attempt total. assert "for attempt in range(max_retries)" in src_fp_prompt assert "for attempt in range(max_retries)" in src_bg_prompt src_fp_render = inspect.getsource(render_one_floor_plan) src_bg_render = inspect.getsource(render_one_background) # Render helpers loop ``for attempt in range(1, max_attempts + 1)`` → # ``max_attempts=1`` yields range(1, 2) = exactly attempt 1. assert "for attempt in range(1, max_attempts + 1)" in src_fp_render assert "for attempt in range(1, max_attempts + 1)" in src_bg_render