"""W17D readback synthesis tests — minimal coverage only. Codex specified: keep tests minimal, experiment stage. Three checks: synthesis shape baseline, mismatch surfacing, methodology grep. """ from __future__ import annotations import json import re import sys from pathlib import Path import pytest _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 _seed_w17c_chain(tmp_path: Path) -> Path: """Build a minimal chain: W15e candidate → W17A legend → W17B3 PNG run → W17C VLM readback. W17D consumes the W17C run dir.""" w15e = tmp_path / "w15e_fake" w15e.mkdir() (w15e / "floor_plan_prompt_candidate.json").write_text(json.dumps({ "candidate_floor_plans": { "fp_l05_01": { "fp_id": "fp_l05_01", "group_id_pointer": "Gx", "candidate_diagram_t2i_prompt": "schematic.", "candidate_key_elements": [], "candidate_numbered_elements": [ {"number": 1, "label": "u1 area", "category": "area", "position_hint": "central core", "unit_id_pointer": "U1"}, {"number": 2, "label": "u2 area", "category": "area", "position_hint": "nw nook", "unit_id_pointer": "U2"}, {"number": 3, "label": "u3 area", "category": "area", "position_hint": "east room", "unit_id_pointer": "U3"}, ], "candidate_camera_recommendations": [], "reconciliation_notes_vs_production": "", }, }, })) w17a = tmp_path / "w17a_fake" w17a.mkdir() (w17a / "floor_plan_image_prompt.json").write_text(json.dumps({ "floor_plan_image_prompt_by_fp": { "fp_l05_01": { "t2i_prompt_text": "Plan with #1 #2 #3.", "numbered_marker_legend": [ {"marker_number": n, "source_candidate_number": n, "element_kind": "spatial_unit", "visual_encoding": "filled_area", "label": f"u{n} area", "priority": "must_show", "source_refs": [], "must_be_legible": True} for n in (1, 2, 3) ], }, }, })) (w17a / "run_meta.json").write_text(json.dumps({ "run_id": "w17a_fake", "stage": "w17a_floor_plan_image_prompt_slice", "run_status": "succeeded", "exit_code": 0, "stage_status": "generated", "model_used": "gpt-5.5", "args": {"derive_image_prompt_from": str(w15e)}, "derived_from": w15e.name, })) w17b3 = tmp_path / "w17b3_fake" (w17b3 / "png").mkdir(parents=True) (w17b3 / "png" / "fp_l05_01.png").write_bytes(b"PNG_PLACEHOLDER") (w17b3 / "run_meta.json").write_text(json.dumps({ "run_id": "w17b3_fake", "stage": "w17b_floor_plan_image_smoke_slice", "run_status": "succeeded", "exit_code": 0, "stage_status": "generated", "model": "gpt-image-2", "image_api_call_count": 1, "image_generation_count": 1, "args": {"derive_image_smoke_from": str(w17a)}, "derived_from": w17a.name, "png_relative_path": "png/fp_l05_01.png", })) w17c = tmp_path / "w17c_fake" w17c.mkdir() (w17c / "vlm_readback.json").write_text(json.dumps({ "read_markers": [ {"marker_number": 1, "visible": True, "confidence": "high", "approximate_region_or_cell": "central core", "observed_unit_label_or_area": "u1 area", "notes": ""}, {"marker_number": 2, "visible": True, "confidence": "medium", "approximate_region_or_cell": "northwest nook", "observed_unit_label_or_area": "u2 area", "notes": ""}, ], "missing_or_ambiguous_markers": [ {"marker_number": 3, "reason": "marker hard to see"}, ], "unit_boundary_summary": "three rooms; central + nw + east", "readback_conflicts": [], })) (w17c / "run_meta.json").write_text(json.dumps({ "run_id": "w17c_fake", "stage": "w17c_floor_plan_vlm_readback_slice", "run_status": "succeeded", "exit_code": 0, "stage_status": "generated", "model_used": "gpt-5.5", "vlm_api_call_count": 1, "image_api_call_count": 0, "args": {"derive_readback_from": str(w17b3)}, "derived_from": w17b3.name, "target_fp_ids": ["fp_l05_01"], })) return w17c def test_w17d_synthesis_shape_and_marker_partition(tmp_path): """W17D synthesis must build one row per legend marker and partition them into matched / unit_mismatch / missing buckets without LLM call. image_api_call_count must remain 0; vlm_api_call_count must remain 0.""" import experiment_floor_plan_readback_synthesis_slice as mod w17c = _seed_w17c_chain(tmp_path) out_root = tmp_path / "out" exit_code = mod.main([ "--derive-synthesis-from", str(w17c), "--target-fp-ids", "fp_l05_01", "--output-root", str(out_root), ]) assert exit_code == 0 run_dir = sorted(out_root.iterdir())[0] meta = json.loads((run_dir / "run_meta.json").read_text()) assert meta["run_status"] == "succeeded" assert meta.get("vlm_api_call_count", 0) == 0 assert meta.get("image_api_call_count", 0) == 0 synth = json.loads((run_dir / "coarse_layout_synthesis.json").read_text()) assert "synthesis_by_fp" in synth by_fp = synth["synthesis_by_fp"]["fp_l05_01"] markers = by_fp["markers"] assert len(markers) == 3, markers nums = sorted(m["marker_number"] for m in markers) assert nums == [1, 2, 3] by_n = {m["marker_number"]: m for m in markers} # #1 and #2 are visible with matching observed_unit → bucket=matched. assert by_n[1]["bucket"] == "matched" assert by_n[2]["bucket"] == "matched" # #3 is missing in VLM output → bucket=missing. assert by_n[3]["bucket"] == "missing" # Heuristic substring match diagnostic (NOT semantic proof). assert by_n[1]["heuristic_unit_match"] in (True, False) # Expected fields populated from the W15e candidate join. assert by_n[1]["expected_unit"] == "U1" assert by_n[2]["expected_unit"] == "U2" assert by_n[3]["expected_unit"] == "U3" def test_w17d_mismatch_report_surfaces_unit_disagreement(tmp_path): """When a VLM observed_unit disagrees with the W15e expected_unit, the mismatch report must list that marker under `unit_mismatch_markers[]`. VLM-reported readback_conflicts are surfaced verbatim (no semantic filter).""" import experiment_floor_plan_readback_synthesis_slice as mod w17c = _seed_w17c_chain(tmp_path) # Mutate the W17C readback so #2 observed_unit disagrees with U2. raw = json.loads((w17c / "vlm_readback.json").read_text()) for entry in raw["read_markers"]: if entry["marker_number"] == 2: entry["observed_unit_label_or_area"] = "unrelated area" raw["readback_conflicts"] = [ {"marker_number": 2, "expected_unit": "U2", "observed_unit_label_or_area": "unrelated area", "note": "vlm-reported conflict"} ] (w17c / "vlm_readback.json").write_text(json.dumps(raw)) out_root = tmp_path / "out_mismatch" exit_code = mod.main([ "--derive-synthesis-from", str(w17c), "--target-fp-ids", "fp_l05_01", "--output-root", str(out_root), ]) assert exit_code == 0 run_dir = sorted(out_root.iterdir())[0] rep = json.loads((run_dir / "mismatch_report.json").read_text()) by_fp = rep["mismatch_by_fp"]["fp_l05_01"] unit_mm_numbers = {m["marker_number"] for m in by_fp["unit_mismatch_markers"]} assert 2 in unit_mm_numbers vlm_conflicts = by_fp["vlm_reported_conflicts"] assert any(c.get("marker_number") == 2 for c in vlm_conflicts) # Missing markers still reported separately. assert by_fp["missing_marker_count"] == 1 def test_w17d_methodology_grep_no_scenario_specific_static_tokens(): """Static script body must not embed scenario-specific tokens. Per-char assembly so this assertion source does not match itself.""" script_path = _SCRIPTS_DIR / "experiment_floor_plan_readback_synthesis_slice.py" assert script_path.exists(), f"script missing: {script_path}" forbidden_tokens = [ "b" + "edroom", "ki" + "tchen", "blood" + "stain", "cur" + "tain", "coo" + "ktop", "tele" + "vision", "cri" + "me", "vi" + "lla", "roo" + "ftop", "foot" + "print", "pol" + "ice", "de" + "ck", "wheel" + "house", "ba" + "throom", "su" + "ri-young", ] pat = re.compile(r"(?i)\b(" + "|".join(forbidden_tokens) + r")\b") m = pat.search(script_path.read_text()) assert m is None, ( f"{script_path.name}: scenario-specific token leaked → " f"{m.group(0) if m else ''}" )