"""W21B-wave-4: FloorPlanSemanticReadbackStep. opt-in (default OFF) producer of the marker-SEMANTIC fidelity gate. Per fp_id present in the base-location dossier checkpoint, the step: - Computes a marker-SEMANTIC readback — does the image content drawn at each base marker match the expected object class / label? With the real-provider selector OFF (default) it uses the synthetic fixture (deterministic placeholder verdicts, NO real VLM call). - Computes a fail-closed gate (pass / needs_fix / needs_review / synthetic_unverified). - Records per-fp readback + gate in the manifest. Gates (all must be true): - ``settings.background_mode`` ∈ {"on", "floor_plan_anchored"}. - ``settings.floor_plan_semantic_readback_enabled`` = True. - ``settings.base_location_dossier_enabled`` = True. - ``settings.floor_plan_prompt_version`` ∈ {"6", "7"} (v7 is the fidelity pack; v7 is schema-compatible with v6 — Rule 11 wording only — so both are accepted). Anything else → ``not_applicable`` with a byte-stable empty payload. This is a SEPARATE gate from the W20A2 geometry readback (marker number/cell/base-kind). LLM / image / VLM API call is 0 unless the real-provider selector is explicitly flipped (or a mock provider is injected in tests). DB / ImageAsset write 0. """ from __future__ import annotations import hashlib import json import logging from pathlib import Path from typing import Any, Callable, Dict, Optional from app.core.step_runner import StepRunner from app.modules.pipeline.floor_plan_semantic_readback import ( SemanticReadbackError, compute_semantic_gate, compute_semantic_readback, ) logger = logging.getLogger(__name__) SCHEMA_VERSION = 1 PROMPT_VERSION = "1" # v7 is schema-compatible with v6 (Rule 11 self-fidelity wording only), # so the semantic gate accepts both. Kept as a module constant so the # acceptance set is auditable in one place. from app.core.fp_prompt_compat import ( # noqa: E402 V6_COMPATIBLE_FP_PROMPT_VERSIONS as _COMPATIBLE_FP_PROMPT_VERSIONS, ) class FloorPlanSemanticReadbackStep(StepRunner): # Test-only injection slot for a mock semantic VLM provider. Default # None — production resolves the provider via the settings selector. _vlm_provider_override: Optional[Callable[..., Dict[str, Any]]] = None def set_vlm_provider_for_testing( self, provider: Optional[Callable[..., Dict[str, Any]]] ) -> None: """Mocking helper. Production callers must NOT touch this.""" self._vlm_provider_override = provider def _resolve_vlm_provider(self) -> Optional[Callable[..., Dict[str, Any]]]: """Resolve the semantic VLM provider for this run. Resolution order (only one path active per run): 1. ``set_vlm_provider_for_testing`` injection — bypasses settings (mock-provider tests). 2. ``settings.floor_plan_semantic_readback_real_provider_enabled`` True → the production-adjacent ``litellm_semantic_vlm_provider``. 3. Default → ``None``; the core emits the synthetic fixture. Zero external API calls. """ from app.core.config import settings if self._vlm_provider_override is not None: return self._vlm_provider_override if not bool( getattr( settings, "floor_plan_semantic_readback_real_provider_enabled", False, ) ): return None from app.modules.pipeline.floor_plan_semantic_vlm_provider import ( litellm_semantic_vlm_provider, ) return litellm_semantic_vlm_provider def _load_prev_checkpoint( self, step_id: str ) -> Optional[Dict[str, Any]]: """Read a prior step's ``manifest.json`` from the checkpoint dir. Mirrors the W20A2 geometry step helper — StepRunner does not provide this, so the step must define it (production callers hit ``self._load_prev_checkpoint(...)``; tests may inject a mock). """ from app.core.config import settings cp = ( Path(settings.projects_dir) / self.project_id / "checkpoints" / "episodes" / self.episode_id / step_id / "manifest.json" ) if cp.exists(): try: return json.loads(cp.read_text(encoding="utf-8")) except Exception as exc: logger.warning( "floor_plan_semantic_readback: %s parse failed: %s", step_id, exc, ) return None def _config_hash(self) -> str: from app.core.config import settings payload = { "background_mode": settings.background_mode, "floor_plan_prompt_version": settings.floor_plan_prompt_version, "base_location_dossier_enabled": bool( settings.base_location_dossier_enabled ), "floor_plan_semantic_readback_enabled": bool( getattr( settings, "floor_plan_semantic_readback_enabled", False ) ), "floor_plan_semantic_readback_real_provider_enabled": bool( getattr( settings, "floor_plan_semantic_readback_real_provider_enabled", False, ) ), "schema_version": SCHEMA_VERSION, "prompt_version": PROMPT_VERSION, } return hashlib.sha256( json.dumps(payload, sort_keys=True).encode("utf-8") ).hexdigest()[:16] def _not_applicable(self) -> Dict[str, Any]: return { "applicable_count": 0, "completed_count": 0, "failed_count": 0, "schema_version": SCHEMA_VERSION, "config_hash": self._config_hash(), "data": {}, } def _execute(self, mode: str = "resume") -> Dict[str, Any]: from app.core.config import settings if settings.background_mode not in {"on", "floor_plan_anchored"}: return self._not_applicable() if not bool( getattr(settings, "floor_plan_semantic_readback_enabled", False) ): return self._not_applicable() if not bool(settings.base_location_dossier_enabled): return self._not_applicable() if settings.floor_plan_prompt_version not in _COMPATIBLE_FP_PROMPT_VERSIONS: return self._not_applicable() dossier_cp = self._load_prev_checkpoint("base_location_dossier") dossiers = (dossier_cp or {}).get("data", {}).get("dossiers") or {} if not dossiers: return self._not_applicable() per_fp: Dict[str, Dict[str, Any]] = {} completed = 0 failed = 0 vlm_call_total = 0 # Resolve once per run — either every fp uses the synthetic path # (provider=None) or every fp uses the resolved provider; mixing # per-fp would make the call-count audit ambiguous. base_provider = self._resolve_vlm_provider() for fp_id, dossier in dossiers.items(): counted_provider: Optional[Callable[..., Dict[str, Any]]] = None _counter = [0] if base_provider is not None: def counted(**kw): # noqa: E306 — defined per-iter _counter[0] += 1 return base_provider(**kw) counted_provider = counted try: readback = compute_semantic_readback( dossier=dossier, fp_image_path=dossier.get("fp_image_path"), vlm_provider=counted_provider, ) gate = compute_semantic_gate( readback=readback, dossier=dossier ) per_fp_vlm = ( _counter[0] if counted_provider is not None else 0 ) vlm_call_total += per_fp_vlm per_fp[fp_id] = { "fp_id": fp_id, "readback_status": readback["status"], "readback": readback, "gate": gate, "gate_state": gate["gate_state"], "real_vlm_call": bool(per_fp_vlm), "real_vlm_call_count": per_fp_vlm, } completed += 1 except SemanticReadbackError as exc: logger.error( "floor_plan_semantic_readback fp_id=%s: %s", fp_id, exc ) per_fp_vlm = ( _counter[0] if counted_provider is not None else 0 ) vlm_call_total += per_fp_vlm per_fp[fp_id] = { "fp_id": fp_id, "error": str(exc)[:300], "real_vlm_call": bool(per_fp_vlm), "real_vlm_call_count": per_fp_vlm, } failed += 1 except Exception as exc: # pragma: no cover — defensive logger.exception( "floor_plan_semantic_readback fp_id=%s unexpected: %s", fp_id, exc, ) per_fp_vlm = ( _counter[0] if counted_provider is not None else 0 ) vlm_call_total += per_fp_vlm per_fp[fp_id] = { "fp_id": fp_id, "error": f"unexpected: {type(exc).__name__}: {exc}"[:300], "real_vlm_call": bool(per_fp_vlm), "real_vlm_call_count": per_fp_vlm, } failed += 1 return { "applicable_count": 1 if per_fp else 0, "completed_count": completed, "failed_count": failed, "schema_version": SCHEMA_VERSION, "config_hash": self._config_hash(), "data": { "per_fp": per_fp, "real_vlm_call_count": vlm_call_total, "image_api_call_count": 0, "llm_call_count": 0, }, }