"""W21B-w5 — BG space partition (pure module). Turns pass-2 edge judgements (the LLM/VLM space-adjudication of bg-pair relationships) into a deterministic ``space_partition_plan``: which bgs share a plate (same physical space), which anchor 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 zone inheriting an adjacent open zone's plate within a multi-zone dwelling). Design (Codex + Claude consensus, 2026-05-31): * The deterministic floor / edge judge already classified each pair. This module is PURE clustering + assignment — NO LLM/VLM/image/IO, NO text parsing. * Clustering is **anchor-centered constrained**, NOT connected-components (a plain transitive closure would re-merge cross-zone spaces via bridges): - a node joins a group only if it has a DIRECT strong edge to the anchor AND has no ``different_space`` edge with ANY current member, - so an A~B / B~C strong pair with A~C ``different_space`` never collapses into one group, and an establishing/wide hub never bridges two clusters. * Edge eligibility (Codex strong-parent gate v0, conservative — the failure mode we fix is false reuse): ``same_space`` AND ``confidence >= 0.75`` AND ``strong_parent_allowed`` true. ``different_space`` is a hard negative; ``adjacent_related`` / ``style_only`` / ``uncertain`` never form a group and never become a strong spatial parent (they may surface as weak cross-group context only). * Anchor selection prefers an existing production anchor, then highest strong degree, then confidence-sum, then a NON-hub over a hub, then stable bg_id. * Reference tree: within a group, non-anchor members reuse/derive from the anchor (``ref_tree_parents`` capped at ``max_refs``). Across groups there is NO spatial parent (the substrate keeps FP-first; cross-group context, if any, is weak/style/text only — never a parent PNG across a cross-zone edge). NOTE (Codex caveat): the edge judgements consumed here are the SOT for the *partition*, but if the FP / projection-card substrate is later simplified the judgements must be regenerated before a production run — this module's contract is stable across that, only its inputs change. """ from __future__ import annotations from itertools import combinations from typing import Any, Dict, FrozenSet, Iterable, List, Optional, Tuple STRONG_CONF_THRESHOLD: float = 0.75 MAX_REFS: int = 2 EDGE_STRONG = "strong" EDGE_DIFFERENT = "different" EDGE_WEAK = "weak" RENDER_NEW = "render_new_plate" RENDER_REUSE = "reuse_existing_plate" # ── candidate-edge generator tunables (Codex+Claude w5 dry calibration) ── #: A structural unit targeted by >= this fraction of a dwelling's bgs is a #: shared/common/circulation zone (e.g. an open living-kitchen). Sharing only #: such a zone is NOT evidence of same-space, so it is discounted (an IDF-style #: down-weight) when deciding whether a candidate edge is an obvious positive. UBIQUITY_FRACTION: float = 0.8 #: A bg whose camera-target floor spans >= this many structural units is an #: establishing / wide "hub" view; it must never auto-merge or bridge clusters. HUB_NODE_UNIT_COUNT: int = 5 # deterministic candidate edge states (a DIAGNOSTIC floor, never the final SOT) CAND_DIFFERENT = "different_space" # hard negative — not sent to the judge CAND_SAME = "same_space" # obvious positive — still judged, for safety CAND_UNCERTAIN = "uncertain" # borderline — must be judged CAND_HUB_UNCERTAIN = "hub_uncertain" # hub view borderline — must be judged def _int_set(values: Any) -> set: """Filter an iterable down to genuine ints (drop bools / non-ints).""" out: set = set() if not isinstance(values, (list, tuple, set, frozenset)): return out for n in values: if isinstance(n, int) and not isinstance(n, bool): out.add(n) return out def _structural_unit_numbers(base_marker_inventory: Any) -> Dict[int, Any]: """Map ``base_structural_unit`` marker number -> its label. ``base_layer_decision`` is a closed enum produced upstream; we read it and the integer ``number`` only. No label/text parsing (absolute rule) — the label is carried through as opaque passthrough evidence, never matched on. """ out: Dict[int, Any] = {} if not isinstance(base_marker_inventory, list): return out for m in base_marker_inventory: if not isinstance(m, dict): continue if m.get("base_layer_decision") != "base_structural_unit": continue num = m.get("number") if isinstance(num, int) and not isinstance(num, bool): out[num] = m.get("label") return out def _classify_candidate_edge( *, shared: set, distinctive: set, jaccard: float, overlap_min: float, hub: bool ) -> str: """Deterministic borderline classification for one candidate edge. Conservative v0 (the failure mode being fixed is *false* cross-zone reuse): deterministic only decides the hard negative, the hub isolation, and an OBVIOUS positive. Everything else defers to the pass-2 edge judge. * hub view -> hub_uncertain * no shared structural unit -> different_space (negative) * distinctive non-empty AND jaccard==1 AND overlap_min==1 -> same_space (obvious) * else (shares only common/hub zone, partial distinctive, or small-dwelling degeneracy) -> uncertain """ if hub: return CAND_HUB_UNCERTAIN if not shared: return CAND_DIFFERENT if distinctive and jaccard >= 1.0 and overlap_min >= 1.0: return CAND_SAME return CAND_UNCERTAIN def build_candidate_edges( *, base_marker_inventory: Any, per_bg_render_facts_by_bg_id: Any, ubiquity_fraction: float = UBIQUITY_FRACTION, hub_node_unit_count: int = HUB_NODE_UNIT_COUNT, ) -> Dict[str, Any]: """Generate bounded pairwise candidate edges for one floor plan. PURE deterministic floor — NO LLM/VLM/image/IO, NO text/label parsing. Both inputs come from the ``base_location_dossier`` checkpoint (upstream of the plan), so this can run in a step placed *before* shot_aware_bg_render_plan: * ``base_marker_inventory`` — the dossier's flat marker list; we read only ``base_layer_decision == "base_structural_unit"`` integer numbers. * ``per_bg_render_facts_by_bg_id`` — per-bg exact-ID facts; we read only ``target_unit_marker_numbers`` (the camera-target floor) per bg. A bg's *signature* is its target-unit floor intersected with the dwelling's structural units. Pairwise edges whose deterministic state is a hard negative (no shared structural unit) are dropped; every other pair with any shared structural signal becomes a candidate the pass-2 judge must rule on (over-generation is fine — the judge is the SOT; under-generation is not). Returns ``{signatures, structural_units, ubiquitous_units, hub_bg_ids, candidate_edges, diagnostics}``. Each candidate edge carries the deterministic ``state`` plus the shared/distinctive marker sets and the jaccard / overlap_min metrics for the judge's prompt context. """ units = _structural_unit_numbers(base_marker_inventory) unit_set = set(units) facts = per_bg_render_facts_by_bg_id if isinstance(per_bg_render_facts_by_bg_id, dict) else {} diagnostics: List[str] = [] # per-bg signature = target-unit floor ∩ structural units signatures: Dict[str, set] = {} for bg_id, f in facts.items(): if not isinstance(bg_id, str) or not isinstance(f, dict): continue signatures[bg_id] = _int_set(f.get("target_unit_marker_numbers")) & unit_set nb = len(signatures) or 1 unit_freq = {u: sum(1 for s in signatures.values() if u in s) for u in unit_set} ubiquitous = {u for u, c in unit_freq.items() if c / nb >= ubiquity_fraction} hub_bg_ids = { bg for bg, s in signatures.items() if len(s) >= hub_node_unit_count } candidate_edges: List[Dict[str, Any]] = [] for a, b in combinations(sorted(signatures), 2): A, B = signatures[a], signatures[b] if not A or not B: # a bg with no structural-unit target floor gives no spatial signal; # leave it for the judge via card evidence only (surfaced as diag). diagnostics.append( f"{a if not A else b} has empty structural signature — " f"edge {a}~{b} not deterministically generated" ) continue shared = A & B union = A | B jaccard = len(shared) / len(union) if union else 0.0 overlap_min = len(shared) / min(len(A), len(B)) distinctive = shared - ubiquitous hub = a in hub_bg_ids or b in hub_bg_ids state = _classify_candidate_edge( shared=shared, distinctive=distinctive, jaccard=jaccard, overlap_min=overlap_min, hub=hub, ) if state == CAND_DIFFERENT: # hard negative — not worth a judge call, never a same-space parent. continue candidate_edges.append( { "bg_a": a, "bg_b": b, "state": state, "shared_units": sorted(shared), "distinctive_units": sorted(distinctive), "jaccard": round(jaccard, 4), "overlap_min": round(overlap_min, 4), "hub": hub, } ) return { "signatures": {bg: sorted(s) for bg, s in signatures.items()}, "structural_units": sorted(unit_set), "ubiquitous_units": sorted(ubiquitous), "hub_bg_ids": sorted(hub_bg_ids), "candidate_edges": candidate_edges, "diagnostics": diagnostics, } # ── pass-2 edge judge (LLM over projection-card visible_items) ────────── #: closed vocabulary the judge may return for a successfully judged edge. EDGE_JUDGE_STATES = ( "same_space", "adjacent_related", "style_only", "different_space", "uncertain", ) #: judge_status separates a real adjudication from operational non-results so #: a skipped/blocked/invalid edge is never mistaken for an ``uncertain`` verdict #: (Codex lock). Only ``judged`` edges carry a meaningful ``edge_state``. JUDGE_STATUS_JUDGED = "judged" JUDGE_STATUS_SKIPPED = "skipped" # precondition not met (e.g. partial card) JUDGE_STATUS_INVALID = "invalid" # provider returned an unusable payload EDGE_JUDGE_SYSTEM = ( "You are a spatial reasoning judge for film background plates. Two " "background plates (bg_a, bg_b) from the same dwelling are given, each with " "a projection card describing what is VISIBLE in that shot (visible_items). " "Decide their spatial relationship from the visible evidence ONLY.\n\n" "States:\n" "- same_space: the two plates show the SAME physical room/zone (possibly a " "different angle). Requires shared DISTINCTIVE features, not just a common " "corridor or shared open zone.\n" "- adjacent_related: different rooms that are spatially connected (one " "visible through a doorway from the other).\n" "- style_only: only generic style/material similarity, no spatial " "continuity.\n" "- different_space: clearly different rooms with no spatial continuity.\n" "- uncertain: insufficient visible evidence.\n\n" "Rules:\n" "- Ground every judgement in visible_items evidence; cite it.\n" "- Sharing only a common/circulation zone (corridor, shared " "living-kitchen) is NOT enough for same_space.\n" "- A wide/establishing 'hub' view that sees many zones must NOT auto-merge " "with a specific narrower zone.\n" "- Label text is NOT evidence; only what is visibly shown.\n" "- Be conservative: when in doubt use uncertain, never same_space.\n" "- shared_distinctive_features must list ONLY concrete distinctive items " "visible in BOTH plates (empty if none).\n" "Output strict JSON only." ) EDGE_JUDGE_SCHEMA: Dict[str, Any] = { "type": "object", "additionalProperties": False, "properties": { "edge_state": {"type": "string", "enum": list(EDGE_JUDGE_STATES)}, "confidence": {"type": "number"}, "evidence": {"type": "string"}, "shared_distinctive_features": { "type": "array", "items": {"type": "string"}, }, }, "required": [ "edge_state", "confidence", "evidence", "shared_distinctive_features", ], } def card_is_judgeable(card: Any) -> bool: """R3 precondition: a card may seed a judge call only if it is a real, passed projection card with visible items — never a synthetic placeholder, a non-pass card, or a card lacking visible evidence.""" if not isinstance(card, dict): return False if card.get("card_state") != "pass": return False if card.get("synthetic") is True or card.get("is_synthetic") is True: return False vis = card.get("visible_items") return isinstance(vis, list) and len(vis) > 0 def build_edge_judge_prompt( *, edge: Dict[str, Any], card_a: Dict[str, Any], card_b: Dict[str, Any], ) -> Dict[str, Any]: """Assemble the strict prompt bundle (system/user/schema) for one edge. Carries the deterministic candidate ``state`` as a non-binding hint and the two cards' ``visible_items`` as the only evidence. No text/label matching here — the visible_items pass through verbatim for the model to read. """ import json as _json user = { "bg_a": edge.get("bg_a"), "bg_b": edge.get("bg_b"), "deterministic_hint": edge.get("state"), "shared_units": edge.get("shared_units"), "distinctive_units": edge.get("distinctive_units"), "card_a_visible_items": card_a.get("visible_items"), "card_b_visible_items": card_b.get("visible_items"), } return { "system": EDGE_JUDGE_SYSTEM, "user": _json.dumps(user, ensure_ascii=False), "schema": EDGE_JUDGE_SCHEMA, } def skipped_edge_judgement(edge: Dict[str, Any], *, reason: str) -> Dict[str, Any]: """A non-result for an edge that could not be judged (e.g. a partial card). Codex lock: a missing card must NOT fail the whole step — it yields a ``skipped`` judgement that can never become a same-space strong parent, so the rest of the dwelling's partition still proceeds. """ return { "bg_a": edge.get("bg_a"), "bg_b": edge.get("bg_b"), "deterministic_state": edge.get("state"), "judge_status": JUDGE_STATUS_SKIPPED, "skip_reason": reason, "edge_state": "uncertain", "confidence": 0.0, "evidence": "", "shared_distinctive_features": [], "strong_parent_allowed": False, } def validate_edge_judge_output( *, edge: Dict[str, Any], raw: Any, both_cards_pass: bool = True, strong_conf_threshold: float = STRONG_CONF_THRESHOLD, ) -> Dict[str, Any]: """Validate one raw judge response into a canonical edge judgement. Enforces the closed state vocabulary and a numeric confidence in [0,1]. Computes the R2 strong-parent gate WITHOUT any semantic reading of the free text: ``strong_parent_allowed`` is true iff ``edge_state == same_space AND confidence >= threshold AND shared_distinctive_features is non-empty AND both_cards_pass`` (Codex lock — the gate is a validator computation, never trusted from provider output). The free-text ``evidence`` and the feature strings are carried through as opaque passthrough audit only. """ base = { "bg_a": edge.get("bg_a"), "bg_b": edge.get("bg_b"), "deterministic_state": edge.get("state"), } if not isinstance(raw, dict): base.update( judge_status=JUDGE_STATUS_INVALID, edge_state="uncertain", confidence=0.0, evidence="", shared_distinctive_features=[], strong_parent_allowed=False, validation_ok=False, validation_error="non-dict judge output", ) return base state = raw.get("edge_state") if state not in EDGE_JUDGE_STATES: state = "uncertain" conf_raw = raw.get("confidence") if isinstance(conf_raw, (int, float)) and not isinstance(conf_raw, bool): conf = max(0.0, min(1.0, float(conf_raw))) else: conf = 0.0 feats = [ f for f in (raw.get("shared_distinctive_features") or []) if isinstance(f, str) and f.strip() ] evidence = raw.get("evidence") evidence = evidence if isinstance(evidence, str) else "" strong_parent_allowed = ( state == "same_space" and conf >= strong_conf_threshold and len(feats) > 0 and bool(both_cards_pass) ) base.update( judge_status=JUDGE_STATUS_JUDGED, edge_state=state, confidence=conf, evidence=evidence, shared_distinctive_features=feats, strong_parent_allowed=strong_parent_allowed, validation_ok=True, ) return base def classify_edge(j: Dict[str, Any], strong_conf_threshold: float = STRONG_CONF_THRESHOLD) -> str: """Map one edge judgement to ``strong`` / ``different`` / ``weak``. Strong requires ALL of: ``same_space`` AND confidence >= threshold AND ``strong_parent_allowed`` (the bool the judge layer already computed — we do NOT re-parse ``shared_distinctive_features`` content here). """ state = j.get("edge_state") if state == "different_space": return EDGE_DIFFERENT conf = j.get("confidence") if ( state == "same_space" and isinstance(conf, (int, float)) and not isinstance(conf, bool) and conf >= strong_conf_threshold and j.get("strong_parent_allowed") is True ): return EDGE_STRONG return EDGE_WEAK def _pair_key(a: str, b: str) -> Tuple[str, str]: return (a, b) if a <= b else (b, a) def build_space_partition_plan( *, bg_ids: Iterable[str], edge_judgements: List[Dict[str, Any]], hub_bg_ids: FrozenSet[str] = frozenset(), existing_anchor_bg_ids: FrozenSet[str] = frozenset(), strong_conf_threshold: float = STRONG_CONF_THRESHOLD, max_refs: int = MAX_REFS, ) -> Dict[str, Any]: """Build the deterministic ``space_partition_plan`` from edge judgements. Returns a dict with ``edge_judgements`` (echoed, classified), ``plate_groups`` ``[{plate_group_id, anchor_bg_id, member_bg_ids, share_score, evidence_edges}]``, ``node_assignments`` ``{bg: group_id}``, ``render_actions`` ``{bg: {render_action, reuse_target_bg_id}}``, ``ref_tree_parents`` ``{bg: [parent_bg_ids]}``, and ``diagnostics``. """ bgs: List[str] = sorted({str(b) for b in bg_ids}) bg_set = set(bgs) diagnostics: List[str] = [] # ── classify edges; build strong adjacency + different-pair set ──── strong_conf: Dict[Tuple[str, str], float] = {} different_pairs: set = set() weak_pairs: List[Tuple[str, str]] = [] classified: List[Dict[str, Any]] = [] for j in edge_judgements: a, b = j.get("bg_a"), j.get("bg_b") if a not in bg_set or b not in bg_set or a == b: continue kind = classify_edge(j, strong_conf_threshold) classified.append({**j, "edge_kind": kind}) key = _pair_key(a, b) if kind == EDGE_STRONG: strong_conf[key] = float(j.get("confidence") or 0.0) elif kind == EDGE_DIFFERENT: different_pairs.add(key) else: weak_pairs.append(key) def is_different(x: str, y: str) -> bool: return _pair_key(x, y) in different_pairs # strong neighbours per node strong_nbrs: Dict[str, Dict[str, float]] = {bg: {} for bg in bgs} for (a, b), c in strong_conf.items(): strong_nbrs[a][b] = c strong_nbrs[b][a] = c def degree(bg: str) -> int: return len(strong_nbrs[bg]) def conf_sum(bg: str) -> float: return sum(strong_nbrs[bg].values()) # anchor preference: existing anchor first, then degree, conf_sum, # non-hub before hub, then stable bg_id. def anchor_rank(bg: str) -> Tuple: return ( 0 if bg in existing_anchor_bg_ids else 1, -degree(bg), -conf_sum(bg), 1 if bg in hub_bg_ids else 0, bg, ) # ── anchor-centered constrained clustering ──────────────────────── assigned: Dict[str, str] = {} groups: List[Dict[str, Any]] = [] for anchor in sorted(bgs, key=anchor_rank): if anchor in assigned: continue members: List[str] = [] # candidate members = strong neighbours of the anchor (best conf first). for m in sorted(strong_nbrs[anchor], key=lambda x: (-strong_nbrs[anchor][x], x)): if m in assigned: continue # reject if m contradicts the anchor or any current member. if is_different(m, anchor) or any(is_different(m, mm) for mm in members): if is_different(m, anchor): # cannot happen (strong+different exclusive) but guard diagnostics.append(f"{m} both strong and different vs anchor {anchor} — skipped") continue members.append(m) group_id = f"grp::{anchor}" assigned[anchor] = group_id for m in members: assigned[m] = group_id groups.append({"anchor_bg_id": anchor, "member_bg_ids": members, "plate_group_id": group_id}) # ambiguity diagnostic: a node that had a strong edge to an anchor of # another group (resolved by greedy order) — surface it. anchor_of = {g["plate_group_id"]: g["anchor_bg_id"] for g in groups} for bg in bgs: my_anchor = anchor_of[assigned[bg]] cross = [n for n in strong_nbrs[bg] if assigned.get(n) != assigned[bg]] if cross: diagnostics.append( f"{bg} had strong edge(s) to {sorted(cross)} outside its group " f"(anchor {my_anchor}) — resolved by anchor order; review if ambiguous") # ── per-group share_score + evidence_edges ──────────────────────── for g in groups: gset = {g["anchor_bg_id"], *g["member_bg_ids"]} ev = [{"pair": list(k), "confidence": c} for k, c in strong_conf.items() if k[0] in gset and k[1] in gset] g["evidence_edges"] = ev g["share_score"] = round(min((e["confidence"] for e in ev), default=0.0), 4) # ── render actions + ref tree ───────────────────────────────────── render_actions: Dict[str, Dict[str, Any]] = {} ref_tree_parents: Dict[str, List[str]] = {} for g in groups: anchor = g["anchor_bg_id"] render_actions[anchor] = {"render_action": RENDER_NEW, "reuse_target_bg_id": ""} ref_tree_parents[anchor] = [] for m in g["member_bg_ids"]: render_actions[m] = {"render_action": RENDER_REUSE, "reuse_target_bg_id": anchor} ref_tree_parents[m] = [anchor][:max_refs] # weak cross-group adjacency context (never a strong parent) — diagnostic only. cross_ctx = sorted( {_pair_key(a, b) for (a, b) in weak_pairs if assigned.get(a) != assigned.get(b)}) if cross_ctx: diagnostics.append( f"{len(cross_ctx)} weak (adjacent/style/uncertain) cross-group edges " f"available as context only, not strong parents") return { "edge_judgements": classified, "plate_groups": groups, "node_assignments": dict(assigned), "render_actions": render_actions, "ref_tree_parents": ref_tree_parents, "cross_group_context_edges": [list(k) for k in cross_ctx], "diagnostics": diagnostics, }