"""W20E1: shot-aware BG E2E readiness preflight (pure validator). Pure structural validator that gates a proposed shot-aware BG run BEFORE any LLM / VLM / image / DB call is made. Consumes a duck-typed settings object plus an explicit operator request and returns a structured result with diagnostics + estimated call/write budget. Normal validation failures are reported in the result; the function does not raise. Design constraints (W20E1): - No network, no DB write, no projects/ checkpoint write. - No import of litellm / openai / image-gen / VLM SDKs. - Exact-ID / enum / boolean structural validation only — no semantic parsing, no scenario-specific literals, no substring policy. - Default-OFF production behaviour preserved: this module is not wired into any step and is intended for operator-side preflight scripts. Failure model (every gate produces a stable ``code`` string; new codes get added here, never reused): - ``background_mode_incompatible`` - ``floor_plan_prompt_version_not_six`` - ``base_location_dossier_disabled`` - ``floor_plan_geometry_readback_disabled`` - ``shot_aware_bg_render_plan_disabled`` - ``w19b3_w20b_conflict`` - ``background_render_reference_mode_not_shot_aware_plan`` - ``background_prompt_version_not_seven_under_shot_aware_plan`` - ``real_vlm_not_approved`` - ``real_shot_aware_planner_llm_not_approved`` - ``image_call_cap_missing`` - ``image_generation_not_approved_for_e2e`` - ``image_generation_unexpected_for_plan_only_scope`` - ``full_fresh_project_e2e_out_of_scope`` - ``unknown_requested_scope`` """ from __future__ import annotations from dataclasses import dataclass, field from typing import Any, Dict, Optional, Tuple # floor_plan_prompt selectors sharing the v6 output schema. This pure validator # is app-namespace isolated (cannot import the SOT module), so the set is # duplicated here and locked to the SOT # (app.core.fp_prompt_compat.V6_COMPATIBLE_FP_PROMPT_VERSIONS) by # test_fp_prompt_compat. Keep in sync when a new v6-compatible pack lands. _V6_COMPATIBLE_FP_VERSIONS = frozenset({"6", "7", "8", "9", "10", "11"}) # ───────────────────────── scope constants ───────────────────────── SCOPE_SHOT_AWARE_BG_PLAN_ONLY = "shot_aware_bg_plan_only" """Run only the W20B shot_aware_bg_render_plan step. No W20C render, no image gen. Real planner LLM is allowed iff explicitly approved. """ SCOPE_SHOT_AWARE_BG_E2E = "shot_aware_bg_e2e" """Run W20B planner → W20C render. Image gen MUST be explicitly approved with a concrete call cap. Real VLM (W20A2.5) and real planner LLM (W20B) are allowed iff explicitly approved. """ SCOPE_FULL_FRESH_PROJECT_E2E = "full_fresh_project_e2e" """Fresh-project / cross-fp E2E. Out of scope for W20E1 unless the operator supplies the explicit override flag — surfaced for future W21 use, blocked by default. """ _KNOWN_SCOPES = frozenset({ SCOPE_SHOT_AWARE_BG_PLAN_ONLY, SCOPE_SHOT_AWARE_BG_E2E, SCOPE_FULL_FRESH_PROJECT_E2E, }) # Scopes that drive an image-generating render path and therefore require # render mode = ``shot_aware_plan`` exact + the W19B-2 v7 background_prompt # pack + explicit image approval/cap. Full fresh-project E2E is BROADER # than shot-aware BG E2E (more upstream side effects, more checkpoints, # more DB rows), so once the W21 override is granted it must satisfy the # same render / image gates rather than bypass them. _SCOPES_REQUIRING_SHOT_AWARE_RENDER_AND_IMAGE = frozenset({ SCOPE_SHOT_AWARE_BG_E2E, SCOPE_FULL_FRESH_PROJECT_E2E, }) _BACKGROUND_MODE_COMPATIBLE = frozenset({"on", "floor_plan_anchored"}) # ───────────────────────── request / result types ───────────────────────── @dataclass(frozen=True) class ShotAwareBgPreflightRequest: """Operator-supplied scope + explicit approval tokens. Required keyword-only fields force callers to be explicit at every callsite (no boolean parameter ambiguity). """ requested_scope: str approve_real_vlm: bool approve_real_shot_aware_planner_llm: bool approve_image_generation: bool image_call_cap: int approve_full_fresh_project_e2e: bool @dataclass(frozen=True) class ShotAwareBgPreflightFailure: """Single failure entry. ``code`` is a stable enum-string callers can pattern-match on; ``message`` is human-readable for logs; ``selector_path`` points at the setting / request field that triggered the failure (``None`` for cross-cutting gates). """ code: str message: str selector_path: Optional[str] = None @dataclass(frozen=True) class ShotAwareBgPreflightResult: """Aggregate result. ``ok`` is ``True`` iff ``failures`` is empty. ``budget_summary`` is a plain dict of estimated call / write figures — operator-facing, not an authoritative cost meter. """ ok: bool failures: Tuple[ShotAwareBgPreflightFailure, ...] budget_summary: Dict[str, Any] = field(default_factory=dict) # ───────────────────────── gate helpers ───────────────────────── def _get(settings: Any, name: str, default: Any) -> Any: return getattr(settings, name, default) def _build_budget_summary( *, settings: Any, request: ShotAwareBgPreflightRequest ) -> Dict[str, Any]: """Pure dict snapshot of the inputs that bound expected side effects. Two layers are reported separately so operators don't confuse the preflight's own (zero) side effects with the side effects the proposed target run would incur if dispatched: ``preflight_*``: this validator. Always 0 — pure structural check. ``target_run_*``: what the operator's proposed run would do if it actually executed. Cap-based ceiling for image counts; boolean presence flags for dispatch-level checkpoint / DB writes (any real run scope writes step_run rows + checkpoints). """ real_vlm_enabled = bool( _get( settings, "floor_plan_vlm_readback_real_provider_enabled", False, ) ) real_planner_llm_enabled = bool( _get( settings, "shot_aware_bg_render_plan_real_provider_enabled", False, ) ) image_approved = bool(request.approve_image_generation) cap = int(request.image_call_cap or 0) estimated_images = cap if image_approved else 0 scope = request.requested_scope is_real_run_scope = scope in _KNOWN_SCOPES return { "scope_kind": scope, "real_vlm_enabled": real_vlm_enabled, "real_shot_aware_planner_llm_enabled": real_planner_llm_enabled, "image_generation_approved": image_approved, "image_call_cap": cap, "estimated_image_api_calls_max": estimated_images, # Preflight side effects (always 0 — this function is pure). "preflight_db_writes_expected": 0, "preflight_image_api_calls_expected": 0, "preflight_checkpoint_writes_expected": 0, # Target-run side effects if the operator dispatches this scope. "target_run_db_writes_expected": is_real_run_scope, "target_run_checkpoint_writes_expected": is_real_run_scope, "target_run_image_asset_writes_expected": estimated_images, } def _gate_common_selectors( settings: Any, ) -> Tuple[ShotAwareBgPreflightFailure, ...]: """Selectors that must hold for ANY shot-aware scope (plan_only or e2e). Each independent gate appends its own failure; nothing is short-circuited so the operator sees the full picture. """ out: list[ShotAwareBgPreflightFailure] = [] mode = _get(settings, "background_mode", "off") if mode not in _BACKGROUND_MODE_COMPATIBLE: out.append( ShotAwareBgPreflightFailure( code="background_mode_incompatible", message=( f"background_mode={mode!r} is not compatible with the " f"shot-aware BG path. Expected one of " f"{sorted(_BACKGROUND_MODE_COMPATIBLE)}." ), selector_path="background_mode", ) ) fp_version = _get(settings, "floor_plan_prompt_version", "5") # v7/v8/v9 share the v6 output schema; accept all. This pure validator is # app-namespace isolated (cannot import the SOT module) — the local set # ``_V6_COMPATIBLE_FP_VERSIONS`` is locked to app.core.fp_prompt_compat by # test_fp_prompt_compat. if fp_version not in _V6_COMPATIBLE_FP_VERSIONS: out.append( ShotAwareBgPreflightFailure( code="floor_plan_prompt_version_not_six", message=( f"floor_plan_prompt_version={fp_version!r}; shot-aware " f"path requires a v6-compatible pack (6/7/8/9)." ), selector_path="floor_plan_prompt_version", ) ) if not bool(_get(settings, "base_location_dossier_enabled", False)): out.append( ShotAwareBgPreflightFailure( code="base_location_dossier_disabled", message=( "base_location_dossier_enabled is False; W20A dossier " "checkpoint is a hard upstream dependency." ), selector_path="base_location_dossier_enabled", ) ) if not bool( _get(settings, "floor_plan_geometry_readback_enabled", False) ): out.append( ShotAwareBgPreflightFailure( code="floor_plan_geometry_readback_disabled", message=( "floor_plan_geometry_readback_enabled is False; W20A2 " "readback is a hard upstream dependency." ), selector_path="floor_plan_geometry_readback_enabled", ) ) if not bool( _get(settings, "shot_aware_bg_render_plan_enabled", False) ): out.append( ShotAwareBgPreflightFailure( code="shot_aware_bg_render_plan_disabled", message=( "shot_aware_bg_render_plan_enabled is False; the W20B " "planner step must be active for the shot-aware path." ), selector_path="shot_aware_bg_render_plan_enabled", ) ) return tuple(out) def _gate_render_mode( settings: Any, scope: str ) -> Tuple[ShotAwareBgPreflightFailure, ...]: """W19B-3 / W20B / W20C render-mode gates. Always: w18j_overlap conflicts with W20B planner (same SOT, two paths) → fail-closed. E2E scope only: render mode MUST be exactly ``shot_aware_plan`` so W20C is the active render path. When render mode is ``shot_aware_plan``: background_prompt_version MUST be exactly ``"7"`` (W19B-2 v7 pack contract). """ out: list[ShotAwareBgPreflightFailure] = [] render_mode = _get( settings, "background_render_reference_mode", "legacy" ) if render_mode == "w18j_overlap": out.append( ShotAwareBgPreflightFailure( code="w19b3_w20b_conflict", message=( "background_render_reference_mode='w18j_overlap' " "(W19B-3) and shot_aware_bg_render_plan_enabled=True " "(W20B) are mutually exclusive — same reference-graph " "SOT, two render paths." ), selector_path="background_render_reference_mode", ) ) if ( scope in _SCOPES_REQUIRING_SHOT_AWARE_RENDER_AND_IMAGE and render_mode != "shot_aware_plan" ): out.append( ShotAwareBgPreflightFailure( code="background_render_reference_mode_not_shot_aware_plan", message=( f"background_render_reference_mode={render_mode!r} for " f"scope={scope!r}; an image-generating shot-aware " f"render scope requires 'shot_aware_plan' exactly so " f"W20C is the active render path." ), selector_path="background_render_reference_mode", ) ) if render_mode == "shot_aware_plan": bg_version = _get(settings, "background_prompt_version", "6") if bg_version != "7": out.append( ShotAwareBgPreflightFailure( code=( "background_prompt_version_not_seven_under_" "shot_aware_plan" ), message=( f"background_prompt_version={bg_version!r} while " f"render mode is 'shot_aware_plan'; the W19B-2 v7 " f"pack ('7') is required." ), selector_path="background_prompt_version", ) ) return tuple(out) def _gate_real_providers( settings: Any, request: ShotAwareBgPreflightRequest ) -> Tuple[ShotAwareBgPreflightFailure, ...]: out: list[ShotAwareBgPreflightFailure] = [] if bool( _get( settings, "floor_plan_vlm_readback_real_provider_enabled", False, ) ) and not bool(request.approve_real_vlm): out.append( ShotAwareBgPreflightFailure( code="real_vlm_not_approved", message=( "floor_plan_vlm_readback_real_provider_enabled is True " "but the preflight request did not set " "approve_real_vlm=True. Real VLM traffic requires " "explicit operator approval." ), selector_path=( "floor_plan_vlm_readback_real_provider_enabled" ), ) ) if bool( _get( settings, "shot_aware_bg_render_plan_real_provider_enabled", False, ) ) and not bool(request.approve_real_shot_aware_planner_llm): out.append( ShotAwareBgPreflightFailure( code="real_shot_aware_planner_llm_not_approved", message=( "shot_aware_bg_render_plan_real_provider_enabled is " "True but the preflight request did not set " "approve_real_shot_aware_planner_llm=True. Real " "planner LLM traffic requires explicit operator " "approval." ), selector_path=( "shot_aware_bg_render_plan_real_provider_enabled" ), ) ) return tuple(out) def _gate_image_generation( request: ShotAwareBgPreflightRequest, ) -> Tuple[ShotAwareBgPreflightFailure, ...]: """Image-gen approval / cap consistency gates. - approve_image_generation=True without a positive cap is fail-closed (no infinite-budget runs). - E2E scope requires explicit image approval; the planner-only scope forbids it. Surfacing both directions stops smuggling the gate. """ out: list[ShotAwareBgPreflightFailure] = [] cap = int(request.image_call_cap or 0) if bool(request.approve_image_generation) and cap <= 0: out.append( ShotAwareBgPreflightFailure( code="image_call_cap_missing", message=( "approve_image_generation=True requires a concrete " "image_call_cap >= 1; preflight refuses uncapped " "image-gen approvals." ), selector_path="image_call_cap", ) ) if ( request.requested_scope in _SCOPES_REQUIRING_SHOT_AWARE_RENDER_AND_IMAGE and not bool(request.approve_image_generation) ): out.append( ShotAwareBgPreflightFailure( code="image_generation_not_approved_for_e2e", message=( f"Scope={request.requested_scope!r} drives W20C image " f"generation; approve_image_generation must be True " f"with image_call_cap >= 1." ), selector_path="approve_image_generation", ) ) if request.requested_scope == SCOPE_SHOT_AWARE_BG_PLAN_ONLY and bool( request.approve_image_generation ): out.append( ShotAwareBgPreflightFailure( code="image_generation_unexpected_for_plan_only_scope", message=( "Scope is shot_aware_bg_plan_only; image generation " "must not be approved on this path — use " "shot_aware_bg_e2e instead." ), selector_path="approve_image_generation", ) ) return tuple(out) def _gate_full_fresh_project_e2e( request: ShotAwareBgPreflightRequest, ) -> Tuple[ShotAwareBgPreflightFailure, ...]: if ( request.requested_scope == SCOPE_FULL_FRESH_PROJECT_E2E and not bool(request.approve_full_fresh_project_e2e) ): return ( ShotAwareBgPreflightFailure( code="full_fresh_project_e2e_out_of_scope", message=( "Scope is full_fresh_project_e2e; this is W21 / " "out-of-scope for W20E1 unless approve_full_fresh_" "project_e2e=True is supplied explicitly." ), selector_path="approve_full_fresh_project_e2e", ), ) return () # ───────────────────────── public API ───────────────────────── def evaluate_shot_aware_bg_preflight( *, settings: Any, request: ShotAwareBgPreflightRequest, ) -> ShotAwareBgPreflightResult: """Run every gate and return aggregated failures + budget summary. Does not raise on normal validation misses; every gate appends to the failures tuple so the operator can see the whole readiness picture in one pass. The only ``RuntimeError`` cases are programmer errors (non-dict-shaped inputs) — but we keep the signature pure to avoid hiding those upstream. """ failures: list[ShotAwareBgPreflightFailure] = [] scope = request.requested_scope if scope not in _KNOWN_SCOPES: failures.append( ShotAwareBgPreflightFailure( code="unknown_requested_scope", message=( f"requested_scope={scope!r} is not one of " f"{sorted(_KNOWN_SCOPES)}." ), selector_path="requested_scope", ) ) failures.extend(_gate_common_selectors(settings)) failures.extend(_gate_render_mode(settings, scope)) failures.extend(_gate_real_providers(settings, request)) failures.extend(_gate_image_generation(request)) failures.extend(_gate_full_fresh_project_e2e(request)) budget = _build_budget_summary(settings=settings, request=request) return ShotAwareBgPreflightResult( ok=not failures, failures=tuple(failures), budget_summary=budget, )