"""W20A: base location dossier — pure deterministic builder. Consumes existing W19A v6 / W19B-1 outputs and produces a per-fp_id dossier that downstream W20B (shot-aware LLM planner) consumes. LLM / image / VLM / DB / ImageAsset write 0. **The VLM 10x10 readback gate is represented in the artifact but never invoked in W20A — only the synthetic_placeholder path is emitted.** Real VLM wiring is deferred to a future wave behind explicit approval. Dossier shape (per fp_id): { "fp_id": str, "fp_image_path": Optional[str], # from floor_plan_render cp "grid_size": [10, 10], "dwelling_identity": { "structure": { unit_markers, opening_markers, fixed_fixture_markers, anchor_furniture_markers, marker_count_per_layer }, "materials": { "diagnostics": [...] }, # LLM-emit slot "fixed_elements_inventory": [...], # base_persistent_fixture "standard_of_living_band": "modest_residential", "lighting_identity": { "diagnostics": [...] }, # LLM-emit slot }, "fp_geometry_candidates": { "grid_size": [10, 10], "camera_cell_candidates_per_unit": {}, # populated by VLM "look_at_cell_candidates_per_unit": {}, # populated by VLM "view_cone_lens_enum_map": {wide:90, normal:50, telephoto:25}, "visible_units_candidates": {}, # populated by VLM "visible_openings_candidates": {}, # populated by VLM "wall_door_invalidation_diagnostics": [], "diagnostics": [...], }, "fp_geometry_vlm_readback": { "status": "synthetic_placeholder", "grid_size": [10, 10], "observed_markers": None, "missing_markers": None, "extra_markers": None, "confidence": None, "gate_decision": "synthetic_pass", "diagnostics": [...], }, "base_marker_inventory": [...], # base_* "overlay_marker_inventory": [...], # state_overlay_* per (bg_id, number) "per_bg_render_facts_by_bg_id": { # W20A revision (Codex) # exact-ID per-bg facts W20B planner consumes without re-reading # raw W19 / master_plan checkpoints. integer / string equality only. "": { "bg_id", "fp_id", "target_unit_marker_numbers", "dominant_target_unit_marker_number", "use_numbered_elements", "ignore_numbered_elements", "base_marker_numbers_to_reference", "transient_marker_numbers_to_describe", "ignored_state_overlay_marker_numbers", "clean_background_expected", "applies_to_shots", "depends_on_bg", "diagnostics", }, }, "anchor_selection_metadata": { "candidate_bg_ids": [...], "selection_diagnostics": [...], "selected_anchor_bg_id": None, # W20B owns selection }, "diagnostics": [...], } W20 spec boundaries enforced: - exact-ID only (marker numbers / bg_id / fp_id exact equality). - no substring / regex / lexical inference over labels or position hints. - static contract carries no scenario-specific tokens — runtime label passthrough (label / position_hint) is data, not contract. - no BG-id keyed prose dict. - reference / camera selection NOT performed here (W20B). - anchor selection NOT performed here — only the candidate surface. """ from __future__ import annotations from typing import Any, Dict, FrozenSet, List, Optional, Tuple # Aligned with W19B-1 (floor_plan_overlay_payload). Importing the # frozensets would create a step→module reverse dep; re-declaring here is # safer + the partition enum is a locked W19 contract. BASE_LAYER_DECISIONS: FrozenSet[str] = frozenset( { "base_structural_unit", "base_opening", "base_persistent_fixture", "base_persistent_furniture", } ) STATE_OVERLAY_DECISIONS: FrozenSet[str] = frozenset( { "state_overlay_plot_cue", "state_overlay_transient_object", } ) ALL_LAYER_DECISIONS: FrozenSet[str] = BASE_LAYER_DECISIONS | STATE_OVERLAY_DECISIONS DEFAULT_GRID_SIZE: Tuple[int, int] = (10, 10) # Generic lens enum → approximate horizontal FOV degrees. Generic — not # scenario-specific. Keys are intentionally generic enum strings. VIEW_CONE_LENS_ENUM_MAP: Dict[str, int] = { "wide": 90, "normal": 50, "telephoto": 25, } # Aligned with W19D modest residential style contract. Static text uses a # generic band identifier — the actual prose lives in # shot_aware_bg_render_adapter._STYLE_CONTRACT_PREAMBLE (W20C), not here. STANDARD_OF_LIVING_BAND_DEFAULT: str = "modest_residential" class BaseLocationDossierError(Exception): """Fail-closed signal for the dossier builder.""" def _index_floor_plans(fp_prompt_data: Dict[str, Any]) -> Dict[str, Dict[str, Any]]: fps = (fp_prompt_data or {}).get("floor_plans") or {} return { fid: entry for fid, entry in fps.items() if isinstance(entry, dict) and entry.get("status") == "ok" } def _index_floor_plan_render( fp_render_data: Dict[str, Any], ) -> Dict[str, Optional[str]]: """Best-effort fp_id → png_path resolution. The floor_plan_render cp shape carries ``data.floor_plans[fp_id].png_path``. Dossier accepts a None when the path is missing — the gate registers a diagnostic but does not raise, because a dossier may legitimately be built before fp render completes. """ out: Dict[str, Optional[str]] = {} fps = (fp_render_data or {}).get("floor_plans") or {} for fid, entry in fps.items(): if not isinstance(entry, dict): continue out[fid] = entry.get("png_path") return out def _collect_bg_specs_by_fp( master_plan_data: Dict[str, Any], ) -> Dict[str, List[Dict[str, Any]]]: grouped: Dict[str, List[Dict[str, Any]]] = {} plans = (master_plan_data or {}).get("plans") or {} for _gid, entry in plans.items(): if not isinstance(entry, dict) or entry.get("status") != "ok": continue plan = entry.get("plan") or {} for bg in plan.get("backgrounds") or []: if not isinstance(bg, dict) or not bg.get("bg_id"): continue depends = bg.get("depends_on_fp") or [] if not depends: continue grouped.setdefault(depends[0], []).append(bg) return grouped def _collect_overlays_by_fp( overlay_payload_data: Dict[str, Any], ) -> Dict[str, Dict[str, Dict[str, Any]]]: overlays = (overlay_payload_data or {}).get("overlays") or {} grouped: Dict[str, Dict[str, Dict[str, Any]]] = {} for bg_id, payload in overlays.items(): if not isinstance(payload, dict): continue fp_id = payload.get("fp_id") if not isinstance(fp_id, str): continue grouped.setdefault(fp_id, {})[bg_id] = payload return grouped def _validate_base_partition( fp_id: str, markers: Dict[int, Dict[str, Any]] ) -> None: """Re-assert the W19B-1 partition enum on the dossier consumer side. Mirrors floor_plan_overlay_payload's exact-ID partition check. We re-do this here rather than trust the overlay payload alone, because the dossier consumes floor_plan_prompt v6 directly for its own marker inventory and a downstream change in the overlay step should not silently relax the dossier's contract. """ for num, ne in markers.items(): decision = ne.get("base_layer_decision") if not isinstance(decision, str): raise BaseLocationDossierError( f"fp_id={fp_id!r} marker #{num} base_layer_decision missing " f"or non-string (got {decision!r})" ) if decision not in ALL_LAYER_DECISIONS: raise BaseLocationDossierError( f"fp_id={fp_id!r} marker #{num} base_layer_decision=" f"{decision!r} not in allowed enum " f"{sorted(ALL_LAYER_DECISIONS)}" ) def _marker_inventory_entry(num: int, ne: Dict[str, Any]) -> Dict[str, Any]: return { "number": num, "label": ne.get("label", ""), "category": ne.get("category", ""), "position_hint": ne.get("position_hint", ""), "base_layer_decision": ne.get("base_layer_decision", ""), } def _structure_block( markers: Dict[int, Dict[str, Any]], ) -> Dict[str, Any]: units: List[int] = [] openings: List[int] = [] fixtures: List[int] = [] furniture: List[int] = [] for num, ne in markers.items(): d = ne.get("base_layer_decision") if d == "base_structural_unit": units.append(num) elif d == "base_opening": openings.append(num) elif d == "base_persistent_fixture": fixtures.append(num) elif d == "base_persistent_furniture": furniture.append(num) units.sort() openings.sort() fixtures.sort() furniture.sort() return { "unit_markers": units, "opening_markers": openings, "fixed_fixture_markers": fixtures, "anchor_furniture_markers": furniture, "marker_count_per_layer": { "base_structural_unit": len(units), "base_opening": len(openings), "base_persistent_fixture": len(fixtures), "base_persistent_furniture": len(furniture), }, } def _fixed_elements_inventory( markers: Dict[int, Dict[str, Any]], ) -> List[Dict[str, Any]]: """Generic shape-kind inventory tied to base_persistent_fixture markers. Runtime labels (LLM-authored upstream) pass through verbatim — they may carry scenario-specific names because that content is data, not contract. W20 spec §4.1 / F-11 cover this static-vs-runtime split. """ out: List[Dict[str, Any]] = [] for num in sorted(markers): ne = markers[num] if ne.get("base_layer_decision") != "base_persistent_fixture": continue out.append( { "number": num, "label": ne.get("label", ""), "position_hint": ne.get("position_hint", ""), "category": ne.get("category", ""), } ) return out def _fp_geometry_candidates_shape( *, grid_size: Tuple[int, int], structure: Dict[str, Any], vlm_status: str, ) -> Dict[str, Any]: """Return the geometry-candidates shape with VLM-dependent slots initialized to empty containers. W20A intentionally does **not** populate camera_cell_candidates etc. from numbered_elements alone — position_hint is a label string and the user's no-literal-substring rule forbids parsing meaning out of it. The populated values arrive once the VLM 10x10 readback gate is wired up in a follow-up wave. For now the shape is declared so downstream W20B can rely on the keys being present. """ diagnostics: List[str] = [] if vlm_status == "synthetic_placeholder": diagnostics.append( "VLM 10x10 readback is synthetic_placeholder; camera / look-at " "/ visible-units / visible-openings cell enumerations are left " "empty pending a real VLM wiring wave." ) if not structure["unit_markers"]: diagnostics.append( "no base_structural_unit markers — geometry candidates cannot " "be enumerated against zero units." ) return { "grid_size": list(grid_size), "camera_cell_candidates_per_unit": {}, "look_at_cell_candidates_per_unit": {}, "view_cone_lens_enum_map": dict(VIEW_CONE_LENS_ENUM_MAP), "visible_units_candidates": {}, "visible_openings_candidates": {}, "wall_door_invalidation_diagnostics": [], "diagnostics": diagnostics, } def _vlm_readback_synthetic_placeholder( *, grid_size: Tuple[int, int], ) -> Dict[str, Any]: """Synthetic-placeholder readback. W20A never invokes a real VLM. The gate decision is ``synthetic_pass`` — not ``ok`` — so downstream consumers can distinguish "VLM readback passed" from "no VLM readback was attempted". Wiring a real VLM provider would replace this function in a follow-up wave behind explicit approval; the gate decision then becomes ``ok`` / ``failed`` based on the VLM's observation set. """ return { "status": "synthetic_placeholder", "grid_size": list(grid_size), "observed_markers": None, "missing_markers": None, "extra_markers": None, "confidence": None, "gate_decision": "synthetic_pass", "diagnostics": [ "VLM readback not invoked (W20A synthetic-only path).", "A real VLM provider must be wired and approved separately " "before this slot can transition out of synthetic_placeholder.", ], } def _collect_overlay_marker_numbers( overlay: Dict[str, Any], *, field: str ) -> List[int]: """Pull exact integer marker numbers from an overlay payload list field. Skips non-dict entries and any number that isn't already a strict int (no string-int coercion; the upstream W19B-1 validator already enforced integer-only on its way in). """ out: List[int] = [] for entry in overlay.get(field) or []: if not isinstance(entry, dict): continue num = entry.get("number") if isinstance(num, bool) or not isinstance(num, int): continue out.append(num) return sorted(set(out)) def _per_bg_render_facts( *, fp_id: str, overlays_by_bg: Dict[str, Dict[str, Any]], bg_specs_by_bg_id: Dict[str, Dict[str, Any]], ) -> Dict[str, Dict[str, Any]]: """Per-bg exact-ID facts the W20B shot-aware planner consumes. Joins the W19B-1 overlay payload (use/ignore arrays, base / transient marker partitions, target unit numbers, clean expectation) with the W19 background_master_plan bg spec (applies_to_shots, depends_on_bg) on ``bg_id`` exact-string equality. **No semantic inference** — integer / string equality only. Every bg_id seen on either side is surfaced once with a ``diagnostics`` list naming the missing side (if any). When overlay is missing the overlay-derived fields default to empty lists / None / False; when master_plan spec is missing the spec-derived fields default to empty lists. The W20B consumer can read the diagnostics to decide whether the bg has enough fact coverage to act on. """ out: Dict[str, Dict[str, Any]] = {} all_bg_ids = sorted(set(overlays_by_bg) | set(bg_specs_by_bg_id)) for bg_id in all_bg_ids: overlay = overlays_by_bg.get(bg_id) spec = bg_specs_by_bg_id.get(bg_id) diagnostics: List[str] = [] if overlay is None: diagnostics.append( "missing floor_plan_overlay_payload entry for this bg_id" ) if spec is None: diagnostics.append( "missing background_master_plan bg spec for this bg_id" ) if overlay is not None: target_units = [ n for n in (overlay.get("target_unit_marker_numbers") or []) if isinstance(n, int) and not isinstance(n, bool) ] dominant_raw = overlay.get("dominant_target_unit_marker_number") if ( isinstance(dominant_raw, int) and not isinstance(dominant_raw, bool) ): dominant: Optional[int] = dominant_raw else: dominant = None use_list = [ n for n in (overlay.get("use_numbered_elements") or []) if isinstance(n, int) and not isinstance(n, bool) ] ignore_list = [ n for n in (overlay.get("ignore_numbered_elements") or []) if isinstance(n, int) and not isinstance(n, bool) ] base_marker_nums = _collect_overlay_marker_numbers( overlay, field="base_markers_to_reference" ) transient_nums = _collect_overlay_marker_numbers( overlay, field="transient_markers_to_describe" ) ignored_overlay_nums = _collect_overlay_marker_numbers( overlay, field="ignored_state_overlay_markers" ) clean = bool(overlay.get("clean_background_expected", False)) else: target_units = [] dominant = None use_list = [] ignore_list = [] base_marker_nums = [] transient_nums = [] ignored_overlay_nums = [] clean = False if spec is not None: applies_to_shots = [ s for s in (spec.get("applies_to_shots") or []) if isinstance(s, str) ] depends_on_bg = [ s for s in (spec.get("depends_on_bg") or []) if isinstance(s, str) ] else: applies_to_shots = [] depends_on_bg = [] out[bg_id] = { "bg_id": bg_id, "fp_id": fp_id, "target_unit_marker_numbers": sorted(set(target_units)), "dominant_target_unit_marker_number": dominant, "use_numbered_elements": sorted(set(use_list)), "ignore_numbered_elements": sorted(set(ignore_list)), "base_marker_numbers_to_reference": base_marker_nums, "transient_marker_numbers_to_describe": transient_nums, "ignored_state_overlay_marker_numbers": ignored_overlay_nums, "clean_background_expected": clean, "applies_to_shots": applies_to_shots, "depends_on_bg": depends_on_bg, "diagnostics": diagnostics, } return out def _anchor_candidate_surface( overlays_by_bg: Dict[str, Dict[str, Any]], ) -> Dict[str, Any]: """Surface anchor candidates only — never pick. Per W20 spec §4.4 the anchor BG selection is LLM-owned. W20A surfaces the candidate set (every overlay whose ``clean_background_expected`` is True) and leaves selection to W20B. W20F5 fallback (Codex 2026-05-28 narrow wave): if no overlay has ``clean_background_expected=True`` but ≥1 overlay exists for the fp, fall back to the BGs with the fewest transient markers (ramped candidate set). The fallback is explicitly tagged in diagnostics so the LLM and downstream auditors know the candidate set was widened. Empty overlays_by_bg still produces empty candidates (fail-closed at the consumer). """ clean_candidates = sorted( bg_id for bg_id, payload in overlays_by_bg.items() if bool(payload.get("clean_background_expected", False)) ) diagnostics: List[str] = [ "anchor selection deferred to W20B (LLM-owned per W19J pivot)." ] if clean_candidates: return { "candidate_bg_ids": clean_candidates, "selection_diagnostics": diagnostics, "selected_anchor_bg_id": None, } # W20F5 fallback path. if not overlays_by_bg: diagnostics.append( "no overlays exist for this fp — anchor candidate set is " "empty (fail-closed at consumer)." ) return { "candidate_bg_ids": [], "selection_diagnostics": diagnostics, "selected_anchor_bg_id": None, } def _transient_count(payload: Dict[str, Any]) -> int: items = payload.get("transient_markers_to_describe") or payload.get( "transient_marker_numbers_to_describe" ) or [] return len(items) if isinstance(items, (list, tuple)) else 0 min_transient = min( _transient_count(p) for p in overlays_by_bg.values() ) ramped = sorted( bg_id for bg_id, payload in overlays_by_bg.items() if _transient_count(payload) == min_transient ) diagnostics.append( "no clean_background_expected=True overlay for this fp — W20F5 " f"fallback widened candidate set to the {len(ramped)} BG(s) with " f"the fewest transient markers ({min_transient}); LLM may still " "pick any of these but must declare the chosen anchor." ) return { "candidate_bg_ids": ramped, "selection_diagnostics": diagnostics, "selected_anchor_bg_id": None, "ramped_fallback": True, } def _build_one_dossier( *, fp_id: str, fp_entry: Dict[str, Any], fp_png_path: Optional[str], overlays_by_bg: Dict[str, Dict[str, Any]], bg_specs: List[Dict[str, Any]], grid_size: Tuple[int, int], ) -> Dict[str, Any]: raw_markers = fp_entry.get("numbered_elements") or [] markers: Dict[int, Dict[str, Any]] = {} for ne in raw_markers: if not isinstance(ne, dict) or "number" not in ne: continue try: num = int(ne["number"]) except (TypeError, ValueError): continue markers[num] = ne _validate_base_partition(fp_id, markers) structure = _structure_block(markers) if not structure["unit_markers"]: raise BaseLocationDossierError( f"fp_id={fp_id!r} has zero base_structural_unit markers; a " f"dossier cannot be built without at least one unit." ) base_inventory: List[Dict[str, Any]] = [ _marker_inventory_entry(num, markers[num]) for num in sorted(markers) if markers[num].get("base_layer_decision") in BASE_LAYER_DECISIONS ] overlay_inventory: List[Dict[str, Any]] = [] for bg_id in sorted(overlays_by_bg): payload = overlays_by_bg[bg_id] for entry in payload.get("transient_markers_to_describe") or []: if not isinstance(entry, dict): continue try: num = int(entry.get("number")) except (TypeError, ValueError): continue overlay_inventory.append( { "bg_id": bg_id, "number": num, "label": entry.get("label", ""), "base_layer_decision": entry.get("base_layer_decision", ""), } ) diagnostics: List[str] = [] if not bg_specs: diagnostics.append( "no backgrounds reference this fp in background_master_plan; " "downstream BG consumers will produce nothing." ) if fp_png_path is None: diagnostics.append( "floor_plan_render did not yet emit a png_path for this fp; " "downstream consumers must wait for the render checkpoint." ) if not overlays_by_bg: diagnostics.append( "no floor_plan_overlay_payload entries reference this fp; the " "overlay step may be not_applicable upstream (v5 default path)." ) vlm_readback = _vlm_readback_synthetic_placeholder(grid_size=grid_size) geometry = _fp_geometry_candidates_shape( grid_size=grid_size, structure=structure, vlm_status=vlm_readback["status"], ) anchor = _anchor_candidate_surface(overlays_by_bg) bg_specs_by_bg_id: Dict[str, Dict[str, Any]] = { spec["bg_id"]: spec for spec in bg_specs if spec.get("bg_id") } per_bg_facts = _per_bg_render_facts( fp_id=fp_id, overlays_by_bg=overlays_by_bg, bg_specs_by_bg_id=bg_specs_by_bg_id, ) return { "fp_id": fp_id, "fp_image_path": fp_png_path, "grid_size": list(grid_size), "dwelling_identity": { "structure": structure, "materials": { "diagnostics": [ "materials palette is an LLM-emit slot (W20B); " "W20A leaves this empty by contract." ], }, "fixed_elements_inventory": _fixed_elements_inventory(markers), "standard_of_living_band": STANDARD_OF_LIVING_BAND_DEFAULT, "lighting_identity": { "diagnostics": [ "lighting identity is an LLM-emit slot (W20B); " "W20A leaves this empty by contract." ], }, }, "fp_geometry_candidates": geometry, "fp_geometry_vlm_readback": vlm_readback, "base_marker_inventory": base_inventory, "overlay_marker_inventory": overlay_inventory, "per_bg_render_facts_by_bg_id": per_bg_facts, "anchor_selection_metadata": anchor, "diagnostics": diagnostics, } def build_dossiers( *, fp_prompt_data: Dict[str, Any], master_plan_data: Dict[str, Any], overlay_payload_data: Dict[str, Any], fp_render_data: Optional[Dict[str, Any]] = None, grid_size: Tuple[int, int] = DEFAULT_GRID_SIZE, ) -> Dict[str, Dict[str, Any]]: """Build one dossier per fp_id present in the floor_plan_prompt cp. The iteration order matches the floor_plan_prompt cp insertion order. Each fp's dossier is independent — a per-fp failure raises ``BaseLocationDossierError`` and aborts the whole call; the step wrapper turns that into a ``failed_count=1`` checkpoint. Parameters ---------- fp_prompt_data: ``floor_plan_prompt`` checkpoint ``data`` block. master_plan_data: ``background_master_plan`` checkpoint ``data`` block. overlay_payload_data: ``floor_plan_overlay_payload`` checkpoint ``data`` block. Empty dict is acceptable (the per-fp dossier records a diagnostic). fp_render_data: Optional ``floor_plan_render`` checkpoint ``data`` block; used only to surface ``fp_image_path``. Missing render is non-fatal. grid_size: VLM readback grid size. Defaults to (10, 10) per W20 spec §2.4. """ fp_index = _index_floor_plans(fp_prompt_data) if not fp_index: raise BaseLocationDossierError( "floor_plan_prompt cp carries no floor_plans entries with " "status='ok'; cannot build any dossier." ) bg_specs_by_fp = _collect_bg_specs_by_fp(master_plan_data) overlays_by_fp = _collect_overlays_by_fp(overlay_payload_data) fp_png_index = _index_floor_plan_render(fp_render_data or {}) if not isinstance(grid_size, tuple) or len(grid_size) != 2: raise BaseLocationDossierError( f"grid_size must be a 2-tuple; got {grid_size!r}" ) if any(not isinstance(v, int) or v <= 0 for v in grid_size): raise BaseLocationDossierError( f"grid_size dimensions must be positive ints; got {grid_size!r}" ) out: Dict[str, Dict[str, Any]] = {} for fp_id, fp_entry in fp_index.items(): out[fp_id] = _build_one_dossier( fp_id=fp_id, fp_entry=fp_entry, fp_png_path=fp_png_index.get(fp_id), overlays_by_bg=overlays_by_fp.get(fp_id, {}), bg_specs=bg_specs_by_fp.get(fp_id, []), grid_size=grid_size, ) return out