"""W21B-wave-4 Phase A: floor-plan marker SEMANTIC readback sidecar (core). Pure, deterministic core for a readback that answers a question the W20A2 geometry readback does **not**: *does the image content drawn at marker N actually match the expected object class / label from the dossier inventory?* The W20A2 ``floor_plan_geometry_readback`` only observes marker number / coarse cell / base-kind — it trusts that "marker N means the object its label claims" because the metadata says so. The FP comparison gallery (W21B-wave-4 brief) showed that this trust is unsafe: a floor-plan PNG can drift from its own metadata (e.g., render a larger, more furnished dwelling than the inventory describes). This sidecar turns marker-label consistency into an explicit evidence-backed gate. Design boundaries (brief §3.1 — scenario leakage guard): - Code performs ONLY structural / exact-ID checks and reads the LLM-emitted ``semantic_match`` enum. It NEVER lexically inspects the free-text evidence fields (``observed_object_summary`` / ``mismatch_reason`` / ``reasoning_basis`` / ``source_ref``). There is no substring / regex / lexicon matching anywhere in this module — the only "meaning" signal is the structured enum the VLM emits. - ``expected_label`` / ``expected_layer`` are carried from the dossier inventory verbatim and validated by exact-string equality; the VLM must not re-label or re-classify a marker. - The gate fails closed: a synthetic-fixture readback, an unattested base marker, or an ``uncertain`` base marker never auto-passes. Only a real ``ok`` readback whose every base structural / persistent marker reads ``match`` passes. This module performs NO LLM / image / VLM / DB / ImageAsset I/O. A real VLM provider is injected via the ``vlm_provider`` slot of ``compute_semantic_readback`` (mirrors the W20A2 geometry pattern); the provider itself lives in a separate module and is wired behind an explicit opt-in selector in a later step of this wave. With ``vlm_provider=None`` (default) the synthetic-fixture path is used and ``fp_image_path`` is ignored. """ from __future__ import annotations from typing import Any, Callable, Dict, List, Optional # ── Enums (generic — no scenario tokens) ───────────────────────────── # The VLM's per-marker verdict. ``uncertain`` is a first-class value so # the model can decline to assert a match it cannot see — that routes to # manual review rather than a false pass (brief §9 stop condition). SEMANTIC_MATCH_VALUES: frozenset = frozenset({"match", "mismatch", "uncertain"}) # Base layers whose semantic fidelity gates the BG anchor downstream. # Mirrors the W20A2 base-kind enum; re-declared to keep this module's # import graph independent of the geometry core. BASE_STRUCTURAL_LAYERS: frozenset = frozenset( { "base_structural_unit", "base_opening", "base_persistent_fixture", "base_persistent_furniture", } ) # Keys OpenAI's strict structured-outputs mode rejects. A test walks the # schema recursively and asserts none appear, so a future edit cannot # silently slip in an unsupported keyword. Mirrors the geometry provider. _OPENAI_UNSUPPORTED_SCHEMA_KEYS: frozenset = frozenset( { "anyOf", "oneOf", "allOf", "not", "minItems", "maxItems", "uniqueItems", "minLength", "maxLength", "pattern", "format", "minimum", "maximum", "exclusiveMinimum", "exclusiveMaximum", "multipleOf", "patternProperties", "contains", "minContains", "maxContains", } ) # Strict OpenAI-compatible JSON schema. Every object property is # ``required``; every object closes via ``additionalProperties: false``. # Nullable ``confidence`` uses the array-of-types form, not ``anyOf``. SEMANTIC_READBACK_SCHEMA: Dict[str, Any] = { "type": "object", "additionalProperties": False, "required": ["status", "fp_id", "observed_marker_semantics", "diagnostics"], "properties": { "status": {"type": "string", "enum": ["ok"]}, "fp_id": {"type": "string"}, "observed_marker_semantics": { "type": "array", "items": { "type": "object", "additionalProperties": False, "required": [ "number", "expected_label", "expected_layer", "observed_object_summary", "semantic_match", "mismatch_reason", "source_ref", "confidence", "reasoning_basis", ], "properties": { "number": {"type": "integer"}, "expected_label": {"type": "string"}, "expected_layer": { "type": "string", "enum": [ "base_structural_unit", "base_opening", "base_persistent_fixture", "base_persistent_furniture", ], }, "observed_object_summary": {"type": "string"}, "semantic_match": { "type": "string", "enum": ["match", "mismatch", "uncertain"], }, "mismatch_reason": {"type": "string"}, "source_ref": {"type": "string"}, "confidence": {"type": ["number", "null"]}, "reasoning_basis": {"type": "string"}, }, }, }, "diagnostics": { "type": "array", "items": {"type": "string"}, }, }, } class SemanticReadbackError(Exception): """Fail-closed signal for the semantic readback / gate.""" # ── helpers ────────────────────────────────────────────────────────── def _is_int(value: Any) -> bool: return isinstance(value, int) and not isinstance(value, bool) def _base_inventory_map(dossier: Dict[str, Any]) -> Dict[int, Dict[str, str]]: """Build ``{number: {"label": ..., "layer": ...}}`` for base markers. Only ``base_*`` markers are surfaced — overlay / transient markers are out of the FP self-fidelity authority set, exactly as in the geometry readback. """ out: Dict[int, Dict[str, str]] = {} for entry in (dossier or {}).get("base_marker_inventory") or []: if not isinstance(entry, dict): continue try: num = int(entry["number"]) except (KeyError, TypeError, ValueError): continue decision = entry.get("base_layer_decision") if isinstance(decision, str) and decision in BASE_STRUCTURAL_LAYERS: out[num] = { "label": str(entry.get("label", "")), "layer": decision, } return out # ── validator (pure — no I/O) ──────────────────────────────────────── def validate_semantic_output( *, output: Any, dossier: Dict[str, Any], fp_id: str, ) -> Dict[str, Any]: """Pure JSON-shape + exact-ID validator for VLM semantic output. Returns ``{"ok": bool, "blockers": [str], "readback": dict | None}``. Code performs only structural / exact-ID checks plus the ``semantic_match`` enum check. No semantic / lexical inspection of the free-text evidence fields is performed (W21B-wave-4 §3.1). """ blockers: List[str] = [] if not isinstance(output, dict): return { "ok": False, "blockers": [f"output is not a dict ({type(output).__name__})"], "readback": None, } if output.get("status") != "ok": blockers.append( f"status must be 'ok' (got {output.get('status')!r}); the " f"real-provider path only emits 'ok' — synthetic / failed are " f"handled elsewhere" ) if output.get("fp_id") != fp_id: blockers.append( f"fp_id mismatch: provider={output.get('fp_id')!r} vs " f"dossier={fp_id!r}" ) inventory = _base_inventory_map(dossier) if not inventory: blockers.append( "dossier base_marker_inventory has zero base_* markers; the " "VLM has nothing to attest to" ) entries = output.get("observed_marker_semantics") if not isinstance(entries, list): blockers.append("observed_marker_semantics must be a list") entries = [] seen_numbers: List[int] = [] for idx, entry in enumerate(entries): prefix = f"observed_marker_semantics[{idx}]" if not isinstance(entry, dict): blockers.append(f"{prefix} not a dict") continue required = ( "number", "expected_label", "expected_layer", "observed_object_summary", "semantic_match", "mismatch_reason", "source_ref", "confidence", "reasoning_basis", ) missing = [f for f in required if f not in entry] if missing: blockers.append(f"{prefix} missing required fields: {missing}") continue n = entry["number"] if not _is_int(n): blockers.append(f"{prefix}.number must be int (got {n!r})") continue if n in seen_numbers: blockers.append(f"{prefix}.number={n} duplicated") seen_numbers.append(n) match = entry["semantic_match"] # isinstance guard FIRST — a list/dict value would raise # TypeError on frozenset membership; the validator boundary must # fail closed with a blocker, never crash. if not isinstance(match, str) or match not in SEMANTIC_MATCH_VALUES: blockers.append( f"{prefix}.semantic_match={match!r} not in " f"{sorted(SEMANTIC_MATCH_VALUES)}" ) # Evidence fields are validated for TYPE only — never inspected # for meaning. The decision signal is the enum above. for text_field in ( "expected_label", "expected_layer", "observed_object_summary", "mismatch_reason", "source_ref", "reasoning_basis", ): if not isinstance(entry[text_field], str): blockers.append( f"{prefix}.{text_field} must be a string " f"(got {entry[text_field]!r})" ) conf = entry["confidence"] if conf is not None and ( not isinstance(conf, (int, float)) or isinstance(conf, bool) ): blockers.append( f"{prefix}.confidence must be float|int|None (got {conf!r})" ) # Exact-ID join: the marker must exist in the base inventory and # carry the dossier's verbatim label + layer. The VLM observes; # it does not re-label or re-classify. if n not in inventory: blockers.append( f"{prefix}.number={n} not in dossier base_marker_inventory" ) else: exp = inventory[n] if entry["expected_label"] != exp["label"]: blockers.append( f"{prefix}.expected_label={entry['expected_label']!r} " f"disagrees with dossier label={exp['label']!r} for " f"marker #{n}" ) if entry["expected_layer"] != exp["layer"]: blockers.append( f"{prefix}.expected_layer={entry['expected_layer']!r} " f"disagrees with dossier base_layer_decision=" f"{exp['layer']!r} for marker #{n}" ) diags = output.get("diagnostics") if not isinstance(diags, list): blockers.append("diagnostics must be a list") diags = [] if blockers: return {"ok": False, "blockers": blockers, "readback": None} readback: Dict[str, Any] = { "status": "ok", "fp_id": fp_id, "observed_marker_semantics": [ { "number": int(e["number"]), "expected_label": str(e["expected_label"]), "expected_layer": str(e["expected_layer"]), "observed_object_summary": str(e["observed_object_summary"]), "semantic_match": str(e["semantic_match"]), "mismatch_reason": str(e["mismatch_reason"]), "source_ref": str(e["source_ref"]), "confidence": ( float(e["confidence"]) if isinstance(e["confidence"], (int, float)) and not isinstance(e["confidence"], bool) else None ), "reasoning_basis": str(e["reasoning_basis"]), } for e in entries ], "diagnostics": list(diags), } return {"ok": True, "blockers": [], "readback": readback} # ── gate decision (pure) ───────────────────────────────────────────── def compute_semantic_gate( *, readback: Dict[str, Any], dossier: Dict[str, Any], ) -> Dict[str, Any]: """Decide the FP self-fidelity gate state from a semantic readback. Gate states (fail-closed ordering): - ``synthetic_unverified`` — readback came from the synthetic fixture (no real VLM). Never authoritative. - ``needs_fix`` — at least one base marker reads ``mismatch``. FP render should retry (bounded) / be marked needs_fix for BG anchor consumption. - ``needs_review`` — no mismatch, but at least one base marker is ``uncertain`` or was never attested. Routes to manual visual review (brief §9) rather than a false pass. - ``pass`` — every base structural / persistent marker reads ``match``. The decision uses ONLY the structured ``semantic_match`` enum and the exact-ID join with the dossier base inventory — no text inspection. """ status = readback.get("status") if status == "synthetic_fixture": return { "gate_state": "synthetic_unverified", "mismatch_markers": [], "uncertain_markers": [], "unattested_base_markers": [], "diagnostics": [ "semantic gate computed from a synthetic_fixture readback; " "placeholder verdicts, not a production fidelity attestation." ], } if status != "ok": raise SemanticReadbackError( f"semantic gate requires readback.status in " f"{{ok, synthetic_fixture}} (got {status!r})" ) inventory = _base_inventory_map(dossier) verdict_by_number: Dict[int, str] = {} for entry in readback.get("observed_marker_semantics") or []: num = entry.get("number") if _is_int(num): verdict_by_number[num] = entry.get("semantic_match") mismatch_markers: List[int] = [] uncertain_markers: List[int] = [] unattested_base_markers: List[int] = [] for num in sorted(inventory): verdict = verdict_by_number.get(num) if verdict is None: unattested_base_markers.append(num) elif verdict == "mismatch": mismatch_markers.append(num) elif verdict == "uncertain": uncertain_markers.append(num) if mismatch_markers: gate_state = "needs_fix" elif uncertain_markers or unattested_base_markers: gate_state = "needs_review" else: gate_state = "pass" return { "gate_state": gate_state, "mismatch_markers": mismatch_markers, "uncertain_markers": uncertain_markers, "unattested_base_markers": unattested_base_markers, "diagnostics": [], } # ── synthetic fixture + provider dispatch ──────────────────────────── def compute_synthetic_semantic_fixture( *, dossier: Dict[str, Any], ) -> Dict[str, Any]: """Return a synthetic_fixture-status semantic readback. Every base marker is recorded as ``uncertain`` — the fixture asserts nothing about real image content. It exists so downstream code (gate, step wrapper, tests) can be exercised before a real VLM provider is wired, mirroring the W20A2 geometry synthetic fixture. """ fp_id = dossier.get("fp_id") if not isinstance(fp_id, str) or not fp_id: raise SemanticReadbackError( f"dossier.fp_id missing or non-string (got {fp_id!r})" ) inventory = _base_inventory_map(dossier) if not inventory: raise SemanticReadbackError( f"dossier fp_id={fp_id!r} has empty base_marker_inventory" ) entries: List[Dict[str, Any]] = [] for num in sorted(inventory): exp = inventory[num] entries.append( { "number": num, "expected_label": exp["label"], "expected_layer": exp["layer"], "observed_object_summary": "", "semantic_match": "uncertain", "mismatch_reason": "", "source_ref": "", "confidence": None, "reasoning_basis": ( "synthetic_fixture placeholder — no real VLM observation" ), } ) return { "status": "synthetic_fixture", "fp_id": fp_id, "observed_marker_semantics": entries, "diagnostics": [ "synthetic_fixture: semantic verdicts are deterministic " "placeholders, not real VLM observations. The gate must not " "promote this readback to a production fidelity pass." ], } def compute_semantic_readback( *, dossier: Dict[str, Any], fp_image_path: Optional[str] = None, vlm_provider: Optional[Callable[..., Dict[str, Any]]] = None, ) -> Dict[str, Any]: """Return a semantic readback for the dossier's fp_id. With ``vlm_provider=None`` (default) the synthetic_fixture path is used and ``fp_image_path`` is ignored. When a provider callable is passed, it is invoked as ``vlm_provider(dossier=..., fp_image_path=...)`` and its return value is run through the FULL ``validate_semantic_output`` contract (status / fp_id / shape / exact-ID joins / required fields). Any non-conforming return — including an output that merely *looks* ok at the top level but carries an unknown marker number, a re-labelled marker, or a missing field — raises ``SemanticReadbackError``. The dispatcher never returns an unvalidated provider payload. """ if vlm_provider is None: return compute_synthetic_semantic_fixture(dossier=dossier) out = vlm_provider(dossier=dossier, fp_image_path=fp_image_path) if not isinstance(out, dict): raise SemanticReadbackError( f"vlm_provider returned non-dict ({type(out).__name__})" ) res = validate_semantic_output( output=out, dossier=dossier, fp_id=dossier.get("fp_id") ) if not res["ok"]: joined = "; ".join(res["blockers"]) raise SemanticReadbackError( f"vlm_provider output failed validation: {joined[:400]}" ) return res["readback"]