"""W21B-w5 (D2): BgSpacePartitionStep. opt-in (default OFF) producer of the per-fp ``space_partition_plan`` — the LLM SOT for which background plates share a physical space (same zone), which anchors each plate, the render action per bg, and the bounded reference tree. This REPLACES the dwelling-level ``same_physical_space_view`` reference source that incorrectly chained cross-zone plates (an enclosed bathroom inheriting an open living-room's plate within a multi-zone dwelling). Pipeline order **21.594** — after ``shot_projection_card`` (21.593, whose visible_items seed the judge) and before ``shot_aware_bg_render_plan`` (21.595, the consumer that mirrors render_action / ref_tree_parents). Per fp_id the step: 1. Reads the ``base_location_dossier`` checkpoint (the candidate-edge inputs: ``base_marker_inventory`` structural units + ``per_bg_render_facts_by_bg_id`` camera-target floors). Geometry presence is a hard precondition (consistency with the substrate the FP geometry validated). 2. Builds the deterministic candidate edges (``build_candidate_edges`` — IDF ubiquity discount + hub isolation; a diagnostic floor, NEVER the final SOT). 3. Adjudicates each borderline candidate with the pass-2 edge judge — a single text LLM call over the two bgs' projection-card ``visible_items`` (no VLM re-call). With the real-provider selector OFF (default) the provider is None and every edge is recorded ``skipped`` (provider_disabled): no strong edge forms, so each bg keeps its own plate. R3 precondition: an edge is judged only when BOTH bgs have a real, passed, non-synthetic card with visible items; otherwise it is ``skipped`` (never a same-space strong parent). 4. Assembles the ``space_partition_plan`` (``build_space_partition_plan`` — anchor-centered constrained clustering, transitive closure forbidden, hub never bridges, R2 strong-parent gate at conf >= 0.75 with non-empty distinctive features). Cost caps (Codex lock ⑤): a per-fp edge cap and a global LLM-call cap bound the judge spend. When a cap is hit the remaining edges are recorded ``skipped`` with the cap reason — never silently dropped, never auto-merged. Gates (all must be true, else ``not_applicable`` with a byte-stable empty data): - ``settings.background_mode`` ∈ {"on", "floor_plan_anchored"}. - ``settings.bg_space_partition_enabled`` = True. - ``settings.base_location_dossier_enabled`` = True. LLM call count is 0 unless the real-provider selector is flipped (or a mock provider is injected in tests). Image / DB / ImageAsset write 0. """ from __future__ import annotations import hashlib import json import logging from pathlib import Path from typing import Any, Callable, Dict, List, Optional from app.core.step_runner import StepRunner from app.modules.pipeline.bg_space_partition import ( build_candidate_edges, build_edge_judge_prompt, build_space_partition_plan, card_is_judgeable, skipped_edge_judgement, validate_edge_judge_output, ) logger = logging.getLogger(__name__) SCHEMA_VERSION = 1 PROMPT_VERSION = "1" # card_state preference when one bg has several per-shot cards (best first). _CARD_STATE_RANK = {"pass": 0, "needs_review": 1, "blocked": 2} class BgSpacePartitionStep(StepRunner): # Test-only injection slot for a mock edge-judge provider. Default None — # production resolves the provider via the settings selector. _edge_judge_provider_override: Optional[Callable[..., Dict[str, Any]]] = None def set_edge_judge_provider_for_testing( self, provider: Optional[Callable[..., Dict[str, Any]]] ) -> None: """Mocking helper. Production callers must NOT touch this.""" self._edge_judge_provider_override = provider def _resolve_edge_judge_provider( self, ) -> Optional[Callable[..., Dict[str, Any]]]: """Resolve the edge-judge LLM provider for this run. Resolution order (only one path active per run): 1. ``set_edge_judge_provider_for_testing`` injection — bypasses settings (mock-provider tests). 2. ``settings.bg_space_partition_real_provider_enabled`` True → the production-adjacent ``litellm_edge_judge_provider``. 3. Default → ``None``; every edge is recorded ``skipped`` (provider_disabled). Zero external API calls. """ from app.core.config import settings if self._edge_judge_provider_override is not None: return self._edge_judge_provider_override if not bool( getattr(settings, "bg_space_partition_real_provider_enabled", False) ): return None from app.modules.pipeline.bg_space_partition_provider import ( litellm_edge_judge_provider, ) return litellm_edge_judge_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 W21B-wave-4 projection-card / semantic / geometry helpers — StepRunner does not provide this, so the step defines it. """ 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( "bg_space_partition: %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, "base_location_dossier_enabled": bool( settings.base_location_dossier_enabled ), "bg_space_partition_enabled": bool( getattr(settings, "bg_space_partition_enabled", False) ), "bg_space_partition_real_provider_enabled": bool( getattr( settings, "bg_space_partition_real_provider_enabled", False ) ), "bg_space_partition_edge_cap_per_fp": int( getattr(settings, "bg_space_partition_edge_cap_per_fp", 40) ), "bg_space_partition_llm_call_cap": int( getattr(settings, "bg_space_partition_llm_call_cap", 120) ), "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": {}, } @staticmethod def _judge_cards_per_bg( cards_cp: Optional[Dict[str, Any]], fp_id: str ) -> Dict[str, Dict[str, Any]]: """Reduce the per-(bg,shot) projection cards down to one judge card per bg. The edge judge compares two bgs, so a bg with several per-shot cards is collapsed to its best card (``pass`` > ``needs_review`` > ``blocked``, then stable shot_id). The returned dict is the NORMALISED judge-card shape — ``card_state`` + ``visible_items`` (+ ``synthetic`` passthrough) — that ``card_is_judgeable`` / ``build_edge_judge_prompt`` read. NO label / text parsing; visible_items pass through verbatim. """ cards = (cards_cp or {}).get("data", {}).get("cards", {}) or {} best: Dict[str, Dict[str, Any]] = {} best_key: Dict[str, tuple] = {} for _card_key, entry in cards.items(): if not isinstance(entry, dict) or entry.get("fp_id") != fp_id: continue bg_id = entry.get("bg_id") shot_id = entry.get("shot_id") if not isinstance(bg_id, str): continue envelope = entry.get("card") or {} vlm_output = envelope.get("vlm_output") or {} normalised = { "bg_id": bg_id, "shot_id": shot_id, "card_state": entry.get("card_state"), "visible_items": vlm_output.get("visible_items"), "synthetic": bool( envelope.get("synthetic") or vlm_output.get("synthetic") ), } rank = ( _CARD_STATE_RANK.get(normalised["card_state"], 9), str(shot_id), ) if bg_id not in best or rank < best_key[bg_id]: best[bg_id] = normalised best_key[bg_id] = rank return best 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, "bg_space_partition_enabled", False)): return self._not_applicable() if not bool(settings.base_location_dossier_enabled): 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() geometry_cp = self._load_prev_checkpoint("floor_plan_geometry_readback") card_cp = self._load_prev_checkpoint("shot_projection_card") geometry_per_fp = (geometry_cp or {}).get("data", {}).get("per_fp", {}) or {} edge_cap_per_fp = int( getattr(settings, "bg_space_partition_edge_cap_per_fp", 40) ) llm_call_cap = int( getattr(settings, "bg_space_partition_llm_call_cap", 120) ) provider = self._resolve_edge_judge_provider() per_fp: Dict[str, Any] = {} completed = 0 failed = 0 llm_call_total = 0 for fp_id in sorted(dossiers): dossier = dossiers.get(fp_id) or {} if not isinstance(dossier, dict): continue # geometry is a hard precondition — a dwelling whose FP geometry # was never validated must NOT be partitioned (fail closed). if fp_id not in geometry_per_fp: per_fp[fp_id] = { "status": "blocked", "fallback_reason": "geometry_missing", } continue try: candidate = build_candidate_edges( base_marker_inventory=dossier.get("base_marker_inventory"), per_bg_render_facts_by_bg_id=dossier.get( "per_bg_render_facts_by_bg_id" ), ) except Exception as exc: # pragma: no cover — defensive logger.exception( "bg_space_partition candidate %s: %s", fp_id, exc ) per_fp[fp_id] = { "status": "error", "error": f"candidate: {type(exc).__name__}: {exc}"[:300], } failed += 1 continue bg_ids = sorted(candidate.get("signatures", {}).keys()) hub_bg_ids = frozenset(candidate.get("hub_bg_ids", [])) judge_cards = self._judge_cards_per_bg(card_cp, fp_id) edge_judgements: List[Dict[str, Any]] = [] judged = 0 skipped = 0 fp_llm_calls = 0 for edge in candidate.get("candidate_edges", []): bg_a, bg_b = edge.get("bg_a"), edge.get("bg_b") card_a = judge_cards.get(bg_a) card_b = judge_cards.get(bg_b) # R3 precondition: both bgs need a real, passed, non-synthetic # card with visible items, else the edge is skipped (never a # strong parent — the rest of the partition still proceeds). # Checked FIRST so an unjudgeable edge never consumes the cost # cap (Codex Required: the cap counts judge spend, not candidate # ordinal — early unjudgeable edges must not starve later ones). if not (card_is_judgeable(card_a) and card_is_judgeable(card_b)): edge_judgements.append( skipped_edge_judgement(edge, reason="card_not_judgeable") ) skipped += 1 continue if provider is None: edge_judgements.append( skipped_edge_judgement(edge, reason="provider_disabled") ) skipped += 1 continue # cost caps bound the number of edges actually JUDGED — per-fp # judge calls then the global call total — NOT the candidate # position. A capped edge is recorded skipped (never auto-merged). if fp_llm_calls >= edge_cap_per_fp: edge_judgements.append( skipped_edge_judgement(edge, reason="edge_cap_per_fp") ) skipped += 1 continue if llm_call_total >= llm_call_cap: edge_judgements.append( skipped_edge_judgement(edge, reason="llm_call_cap") ) skipped += 1 continue prompt_bundle = build_edge_judge_prompt( edge=edge, card_a=card_a, card_b=card_b ) try: raw = provider(prompt_bundle=prompt_bundle) fp_llm_calls += 1 llm_call_total += 1 judgement = validate_edge_judge_output( edge=edge, raw=raw, both_cards_pass=True ) judged += 1 except Exception as exc: logger.error( "bg_space_partition edge %s~%s: %s", bg_a, bg_b, exc ) fp_llm_calls += 1 llm_call_total += 1 judgement = skipped_edge_judgement( edge, reason=f"judge_error:{type(exc).__name__}" ) skipped += 1 edge_judgements.append(judgement) plan = build_space_partition_plan( bg_ids=bg_ids, edge_judgements=edge_judgements, hub_bg_ids=hub_bg_ids, ) per_fp[fp_id] = { "status": "ok", "bg_ids": bg_ids, "hub_bg_ids": sorted(hub_bg_ids), "candidate": candidate, "edge_judgements": edge_judgements, "space_partition_plan": plan, "edge_judge_call_count": fp_llm_calls, "edges_judged": judged, "edges_skipped": skipped, } completed += 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, "edge_judge_call_count": llm_call_total, "llm_call_count": llm_call_total, "image_api_call_count": 0, }, }