"""W21B (2026-06-08): DwellingZoneMapStep — FP-image + VLM zone grouping.

opt-in (default OFF) producer of a per-dwelling **zone map** that bypasses the
edge-judge (``bg_space_partition``), which cannot group same-dwelling bgs framed
from different camera angles. For each interior dwelling (bgs grouped by their
floor-plan fp_id) the step:

  1. assembles a per-bg structural description (priority: shot_projection_card
     plate prose → freshness-verified background_prompt t2i → master-plan
     catalog labels; dossier structural facts attached as enrichment when
     present);
  2. (real provider) distils the base-set rooms (gpt-5.5), writes a clean B&W
     2D floor-plan prompt (gpt-5.5), renders it (gpt-image-2 text-to-image — a
     NEW clean plan, the distorted ``floor_plan_render`` PNG is NOT reused),
     then a gpt-5.5 VISION call maps each bg to its plan zone + grid focus;
  3. deterministically joins that into the zone-map contract (zones + bg/shot
     zone assignments), draws a SEPARATE annotated FP (shot-number overlay,
     ``must_not_be_used_for_render``) for mapping verification only, and
     records a diagnostic comparison against the edge-judge grouping.

The CLEAN render ref and the ANNOTATED (numbered) ref are kept strictly apart
so no downstream i2i anchor can grab the numbered image (number-leak guard).
This step does NOT yet feed any renderer — it only produces + validates the
contract (Phase 1); the background_render zone-1-plate wiring is Phase 2.

Gates (all true, else ``not_applicable`` with byte-stable empty data):
  - ``settings.background_mode`` ∈ {"on", "floor_plan_anchored"}.
  - ``settings.dwelling_zone_map_enabled`` = True.
  - ``settings.floor_plan_prompt_version`` ∈ {"6", "7"}.

When enabled but ``dwelling_zone_map_real_provider_enabled`` is False, every
dwelling gets the non-authoritative synthetic fixture (no LLM/VLM/image call) —
the plumbing path. A per-dwelling provider failure falls back to synthetic for
that dwelling only (never a step failure).
"""
from __future__ import annotations

import hashlib
import json
import logging
import re
from pathlib import Path
from typing import Any, Callable, Dict, List, Optional

from app.core.step_runner import StepRunner
from app.services.image_capture.sink import capture_artifact
from app.modules.pipeline.dwelling_zone_map import (
    assemble_zone_map,
    build_analysis_from_numbered_elements,
    build_analysis_from_space_model,
    compare_to_edge_judge,
    compute_synthetic_zone_map,
    select_dwelling_targets,
    validate_vlm_zone_output,
)

logger = logging.getLogger(__name__)

SCHEMA_VERSION = 1
PROMPT_VERSION = "1"

from app.core.fp_prompt_compat import (  # noqa: E402
    V6_COMPATIBLE_FP_PROMPT_VERSIONS as _SUPPORTED_FP_PROMPT_VERSIONS,
)

# fp_id whitelist (path-traversal guard — mirrors floor_plan_light_sidecar).
_SAFE_FP_RE = re.compile(r"^[a-z0-9][a-z0-9_]*$")

# annotated-overlay zone palette (RGB).
_ZONE_PALETTE = [
    (220, 40, 40), (40, 120, 220), (40, 170, 80), (200, 140, 30),
    (150, 60, 200), (40, 180, 180),
]


class DwellingZoneMapStep(StepRunner):
    # Test-only injection slots (production resolves real clients / fixtures).
    _analyze_override: Optional[Callable[..., Dict[str, Any]]] = None
    _fp_prompt_override: Optional[Callable[..., str]] = None
    _render_override: Optional[Callable[..., Dict[str, Any]]] = None
    _vlm_override: Optional[Callable[..., List[Dict[str, Any]]]] = None
    _annotate_override: Optional[Callable[..., Optional[str]]] = None

    _client: Any = None

    def set_overrides_for_testing(
        self, *, analyze=None, fp_prompt=None, render=None, vlm=None, annotate=None
    ) -> None:
        """Mocking helpers for the experimental LLM/VLM/image path. Production
        must NOT touch these."""
        self._analyze_override = analyze
        self._fp_prompt_override = fp_prompt
        self._render_override = render
        self._vlm_override = vlm
        self._annotate_override = annotate

    # ──────────────────────── checkpoint / config plumbing ────────────────────────
    def _load_prev_checkpoint(self, step_id: str) -> Optional[Dict[str, Any]]:
        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("dwelling_zone_map: %s parse failed: %s", step_id, exc)
        return None

    def _config_hash(self) -> str:
        from app.core.config import settings
        from app.modules.pipeline.dwelling_zone_map_provider import (
            PROMPT_VERSION as PROVIDER_PROMPT_VERSION,
            PROVIDER_VERSION,
        )
        payload = {
            "background_mode": settings.background_mode,
            "floor_plan_prompt_version": settings.floor_plan_prompt_version,
            "enabled": bool(getattr(settings, "dwelling_zone_map_enabled", False)),
            "real_provider": bool(
                getattr(settings, "dwelling_zone_map_real_provider_enabled", False)),
            "text_model": getattr(settings, "dwelling_zone_map_text_model", ""),
            "vision_model": getattr(settings, "dwelling_zone_map_vision_model", ""),
            "render_model": getattr(settings, "dwelling_zone_map_render_model", ""),
            "schema_version": SCHEMA_VERSION,
            "prompt_version": PROMPT_VERSION,
            "provider_version": PROVIDER_VERSION,
            "provider_prompt_version": PROVIDER_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": {},
        }

    # ──────────────────────── per-bg description assembly ────────────────────────
    @staticmethod
    def _fresh_t2i_by_bg(
        bg_prompt_cp: Optional[Dict[str, Any]], master_bg_catalog_hash: Optional[str]
    ) -> Dict[str, str]:
        """background_prompt t2i prompts, but ONLY when the checkpoint is FRESH
        for the current catalog (``consumed_bg_catalog_hash`` == master_plan
        ``bg_catalog_hash``). background_prompt is order 21.60 — AFTER this step
        — so on a fresh run it is absent, and on a re-run a STALE checkpoint may
        linger; either way it must never be a blind primary source (Codex W21B
        review #2). When the hash cannot be confirmed fresh, returns ``{}``."""
        if not isinstance(bg_prompt_cp, dict):
            return {}
        data = bg_prompt_cp.get("data") or {}
        consumed = data.get("consumed_bg_catalog_hash")
        if not master_bg_catalog_hash or consumed != master_bg_catalog_hash:
            return {}  # absent / stale → not usable
        out: Dict[str, str] = {}
        for bid, spec in (data.get("backgrounds") or {}).items():
            prompt = (spec or {}).get("t2i_prompt")
            if isinstance(prompt, str) and prompt.strip():
                out[bid] = prompt.strip()
        return out

    @staticmethod
    def _structural_facts(facts: Any) -> Optional[Dict[str, Any]]:
        """Pure structural (numeric/bool) subset of a dossier ``per_bg_render
        _facts`` entry — marker-number lists, counts, flags. Type-filtered only
        (no field-name hardcoding, no prose, no semantic inference) so it serves
        as generic structural context the VLM can use to disambiguate a plate."""
        if not isinstance(facts, dict):
            return None
        out: Dict[str, Any] = {}
        for k, v in facts.items():
            if isinstance(v, bool) or (isinstance(v, int) and not isinstance(v, bool)):
                out[k] = v
            elif isinstance(v, list) and v and all(
                isinstance(x, int) and not isinstance(x, bool) for x in v
            ):
                out[k] = v
        return out or None

    def _assemble_bg_descriptions(
        self,
        *,
        bg_ids: List[str],
        catalog: Dict[str, Any],
        projection_cp: Optional[Dict[str, Any]],
        fresh_t2i_by_bg: Dict[str, str],
        dossier_facts: Optional[Dict[str, Any]] = None,
    ) -> List[Dict[str, Any]]:
        """Per-bg structural text for the VLM bg→zone mapping. Primary =
        shot_projection_card plate prose (a real upstream dep, freshness-safe at
        this order); secondary = FRESHNESS-VERIFIED background_prompt t2i;
        fallback = master-plan catalog labels. When base_location_dossier ran,
        each bg's pure structural facts are attached as ``structure_facts`` (the
        provider serializes them into the VLM prompt) — real enrichment, never
        the primary signal. The VLM is told to ignore transient events, so even
        the event-bearing catalog text is usable as a degraded last resort.
        (Codex W21B review #2: background_prompt is never a blind primary.)"""
        plate_by_bg: Dict[str, List[str]] = {}
        if isinstance(projection_cp, dict):
            cards = ((projection_cp.get("data") or {}).get("cards") or {})
            for _cid, card in cards.items():
                if not isinstance(card, dict):
                    continue
                bid = card.get("bg_id")
                vlm = card.get("vlm_output") or {}
                desc = vlm.get("bg_plate_visible_description")
                if isinstance(bid, str) and isinstance(desc, str) and desc.strip():
                    plate_by_bg.setdefault(bid, []).append(desc.strip())

        blocks: List[Dict[str, Any]] = []
        for bg_id in bg_ids:
            if bg_id in plate_by_bg:
                desc, src = " / ".join(plate_by_bg[bg_id]), "projection_card"
            elif bg_id in fresh_t2i_by_bg:
                desc, src = fresh_t2i_by_bg[bg_id], "background_prompt_fresh"
            else:
                e = catalog.get(bg_id) or {}
                desc = " ".join(
                    str(e.get(k) or "") for k in ("sub_location_label", "state_label_raw")
                ).strip()
                src = "catalog_label"
            block: Dict[str, Any] = {"bg_id": bg_id, "description": desc, "source": src}
            struct = self._structural_facts((dossier_facts or {}).get(bg_id))
            if struct:
                block["structure_facts"] = struct
                block["source"] = src + "+dossier"
            blocks.append(block)
        return blocks

    # ──────────────────────── image render + annotation ────────────────────────
    def _resolve_openai_client(self):
        from app.core.openai_keys import openai_client
        from app.core.config import settings
        return openai_client(
            timeout=float(settings.llm_timeout_image_gen),
        )

    def _render_clean_fp(self, *, prompt: str, out_path: Path, fp_id: str) -> Dict[str, Any]:
        if self._render_override is not None:
            return self._render_override(prompt=prompt, out_path=out_path, fp_id=fp_id)
        from app.core.config import settings
        from app.modules.pipeline.floor_plan_render import render_one_floor_plan
        if self._client is None:
            self._client = self._resolve_openai_client()
        res = render_one_floor_plan(
            openai_client=self._client,
            image_model=getattr(settings, "dwelling_zone_map_render_model", "gpt-image-2.5-sunburst"),
            prompt=prompt,
            out_path=out_path,
            ref_paths=[],  # text-to-image — a fresh clean plan, no distorted ref
            fp_id=fp_id,
            size="1536x1024",
        )
        return {"status": res.status, "png_path": res.png_path, "error": res.error}

    def _annotate_fp(
        self, *, clean_png: Path, out_path: Path, zone_map: Dict[str, Any]
    ) -> Optional[str]:
        """Draw a SEPARATE numbered overlay (one circle per bg at its grid
        focus, coloured by zone) — verification only, never a render anchor."""
        if self._annotate_override is not None:
            return self._annotate_override(
                clean_png=clean_png, out_path=out_path, zone_map=zone_map)
        try:
            from PIL import Image, ImageDraw
        except Exception as exc:  # pragma: no cover — env-dependent
            logger.warning("dwelling_zone_map: PIL unavailable, skip annotation: %s", exc)
            return None
        try:
            fp = Image.open(clean_png).convert("RGB")
        except Exception as exc:
            logger.warning("dwelling_zone_map: annotate open failed: %s", exc)
            return None
        w, h = fp.size
        draw = ImageDraw.Draw(fp)
        zone_color: Dict[str, Any] = {}
        size = (zone_map.get("grid") or {}).get("size", 100) or 100
        for bg_id, a in (zone_map.get("bg_zone_assignments") or {}).items():
            g = a.get("grid_focus") or {}
            if not isinstance(g.get("x"), (int, float)):
                continue
            zid = a.get("zone_id") or "?"
            if zid not in zone_color:
                zone_color[zid] = _ZONE_PALETTE[len(zone_color) % len(_ZONE_PALETTE)]
            c = zone_color[zid]
            x = g["x"] / size * w
            y = g["y"] / size * h
            rad = 16
            draw.ellipse([x - rad, y - rad, x + rad, y + rad],
                         fill=c, outline=(255, 255, 255), width=3)
            draw.text((x - 11, y - 6), bg_id[-3:], fill=(255, 255, 255))
        out_path.parent.mkdir(parents=True, exist_ok=True)
        fp.save(out_path)
        # Phase B: zone-map 주석 FP capture(verification-only 비모델 아티팩트,
        # scope 미배선이면 no-op). 방금 저장한 파일을 읽어 byte-identical.
        capture_artifact(
            out_path.read_bytes(),
            role="dwelling_zone_annotated_fp",
            disposition="diagnostic",
            pipeline_metadata={"verification_only": True},
        )
        return str(out_path)

    # ──────────────────────── per-dwelling real provider ────────────────────────
    def _run_real_dwelling(
        self,
        *,
        fp_id: str,
        bg_ids: List[str],
        bg_shot_map: Dict[str, List[str]],
        bg_blocks: List[Dict[str, Any]],
        space_model: Optional[Dict[str, Any]],
        numbered_elements: Optional[List[Dict[str, Any]]],
        partition_plan: Optional[Dict[str, Any]],
        out_dir: Path,
    ) -> Dict[str, Any]:
        """Returns ``{"zone_map", "provenance"}`` or raises on any provider miss
        (caller falls back to synthetic for this dwelling). ``bg_blocks`` already
        carry any dossier ``structure_facts`` (attached in
        ``_assemble_bg_descriptions``); the provider serializes them into the VLM
        prompt."""
        from app.core.config import settings
        from app.core.file_paths import to_relative_image_path
        from app.modules.pipeline import dwelling_zone_map_provider as prov

        text_model = getattr(settings, "dwelling_zone_map_text_model", prov.TEXT_MODEL_DEFAULT)
        vision_model = getattr(settings, "dwelling_zone_map_vision_model", prov.VISION_MODEL_DEFAULT)

        # 1) dwelling structure analysis. PRIMARY = floor_plan_prompt.space_model
        # (the canonical full dwelling — every zone, so the FP shows ALL rooms,
        # not just the spaces that have a bg plate). The bg-derived analysis is
        # the FALLBACK only when no space_model is available.
        used_space_model = isinstance(space_model, dict) and bool(
            (space_model.get("zones") if isinstance(space_model, dict) else None))
        # numbered_elements is the structure SOT the fp prompt packs actually
        # emit (no pack emits space_model). Used when space_model is absent,
        # BEFORE the bg-derived fallback — so the FP draws every structural zone.
        numbered_analysis = (
            build_analysis_from_numbered_elements(numbered_elements)
            if not used_space_model else {"rooms": []})
        used_numbered = bool(numbered_analysis.get("rooms"))
        if self._analyze_override is not None:
            analysis = self._analyze_override(bg_blocks=bg_blocks, space_model=space_model)
        elif used_space_model:
            analysis = build_analysis_from_space_model(space_model)
        elif used_numbered:
            analysis = numbered_analysis
        else:
            analysis = prov.analyze_dwelling_space(bg_blocks=bg_blocks, model=text_model)
        room_labels = [
            r.get("label") or r.get("name")
            for r in (analysis.get("rooms") or [])
            if isinstance(r, dict)
        ]

        # 2) clean FP prompt — feed the full structure SOT so every zone is drawn.
        if self._fp_prompt_override is not None:
            fp_prompt = self._fp_prompt_override(space_model=space_model, space_analysis=analysis)
        else:
            fp_prompt = prov.build_clean_fp_prompt(
                space_model=space_model if used_space_model else None,
                space_analysis=analysis, model=text_model)

        # 3) render the clean FP (gpt-image-2 text-to-image).
        clean_path = out_dir / f"{fp_id}_clean.png"
        render = self._render_clean_fp(prompt=fp_prompt, out_path=clean_path, fp_id=fp_id)
        if render.get("status") != "ok" or not (render.get("png_path") and Path(render["png_path"]).exists()):
            raise RuntimeError(f"clean FP render failed: {render.get('error')!r}")
        clean_png = Path(render["png_path"])
        clean_ref = to_relative_image_path(str(clean_png))

        # 4) VLM vision zone mapping (dossier facts enrich the per-bg context).
        if self._vlm_override is not None:
            raw = self._vlm_override(fp_image_path=str(clean_png), bg_blocks=bg_blocks)
        else:
            raw = prov.map_bgs_to_zones(
                fp_image_path=str(clean_png), bg_blocks=bg_blocks,
                room_labels=[r for r in room_labels if isinstance(r, str)],
                model=vision_model)
        val = validate_vlm_zone_output(raw=raw, expected_bg_ids=bg_ids)
        if not val["ok"]:
            raise RuntimeError(f"VLM zone output invalid: {val['blockers'][:5]}")

        # 5) deterministic assembly (annotated ref filled after we draw it).
        assembled = assemble_zone_map(
            fp_id=fp_id, vlm_normalized=val["normalized"], bg_shot_map=bg_shot_map,
            clean_fp_ref=clean_ref, annotated_fp_ref=None, space_analysis=analysis)
        zone_map = assembled["zone_map"]

        # 6) annotated overlay (verification only) → fill annotated ref.
        # persist-all Wave2 (B6): zone-map 주석 FP 는 verification-only diagnostic
        # 중간물 — 최종 등록 경로 없음. ★scope 는 _annotate_fp(=capture_artifact) 한 줄만
        # 좁게 감싼다(이 메서드의 다른 산출물은 최종물 아님이지만, "최종 등록 output 이 나오지
        # 않는 scope" 가드 — Codex 합의). flush 는 scope 종료 시 비치명 영속화.
        annot_path = out_dir / f"{fp_id}_annotated.png"
        from app.services.image_capture.context import generation_context
        with generation_context(
            self.project_id, self.episode_id, stage="dwelling_zone",
        ):
            annot_abs = self._annotate_fp(clean_png=clean_png, out_path=annot_path, zone_map=zone_map)
        annot_ref = to_relative_image_path(annot_abs) if annot_abs else None
        zone_map["annotated_fp_ref"]["ref"] = annot_ref

        # 7) edge-judge diagnostic comparison.
        zone_map["diagnostics"]["edge_judge_comparison"] = compare_to_edge_judge(
            zone_map=zone_map, partition_plan=partition_plan)
        # surface zones whose structure_cues never matched (provider room-label
        # drift diagnostic — Codex W21B review #5: not a silent empty array).
        cues_empty = sorted(
            zid for zid, z in (zone_map.get("zones") or {}).items()
            if not (z.get("structure_cues") or []))
        zone_map["diagnostics"]["zones_without_structure_cues"] = cues_empty

        provenance = {
            "structure_source": (
                "space_model" if used_space_model
                else "numbered_elements" if used_numbered
                else "bg_blocks_fallback"),
            "space_room_count": len(room_labels),
            "space_rooms": room_labels,
            "fp_prompt": fp_prompt,
            "bg_description_sources": {
                b.get("bg_id"): b.get("source") for b in bg_blocks},
            "dossier_enriched_bgs": [
                b.get("bg_id") for b in bg_blocks if b.get("structure_facts")],
            "text_model": text_model,
            "vision_model": vision_model,
            "vlm_raw_count": len(raw) if isinstance(raw, list) else 0,
        }
        return {"zone_map": zone_map, "provenance": provenance}

    # ──────────────────────── execute ────────────────────────
    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, "dwelling_zone_map_enabled", False)):
            return self._not_applicable()
        if settings.floor_plan_prompt_version not in _SUPPORTED_FP_PROMPT_VERSIONS:
            return self._not_applicable()

        mp_cp = self._load_prev_checkpoint("background_master_plan")
        mp_data = (mp_cp or {}).get("data") or {}
        catalog = mp_data.get("background_catalog") or {}
        if not catalog:
            return self._not_applicable()
        master_bg_catalog_hash = mp_data.get("bg_catalog_hash")

        targets = select_dwelling_targets(background_catalog=catalog)
        applicable = {fp: t for fp, t in targets.items() if t["applicable"]}
        if not applicable:
            return self._not_applicable()

        real = bool(getattr(settings, "dwelling_zone_map_real_provider_enabled", False))
        # Structure SOT (always read when real): floor_plan_prompt.space_model
        # per fp_id is the canonical full dwelling; base_location_dossier carries
        # per-bg structural facts (enrichment).
        fp_prompt_cp = self._load_prev_checkpoint("floor_plan_prompt") if real else None
        space_models = self._space_models_by_fp(fp_prompt_cp)
        numbered_by_fp = self._numbered_elements_by_fp(fp_prompt_cp)
        dossier_facts = self._dossier_facts_by_bg(
            self._load_prev_checkpoint("base_location_dossier") if real else None)
        projection_cp = self._load_prev_checkpoint("shot_projection_card") if real else None
        fresh_t2i = self._fresh_t2i_by_bg(
            self._load_prev_checkpoint("background_prompt") if real else None,
            master_bg_catalog_hash)
        partition_cp = self._load_prev_checkpoint("bg_space_partition")
        partition_plans = self._partition_plans_by_fp(partition_cp, catalog)

        out_dir = (
            Path(settings.projects_dir) / self.project_id / "checkpoints"
            / "episodes" / self.episode_id / "dwelling_zone_map" / "fp_renders"
        )
        if real:
            # render_one_floor_plan writes out_path with no parent mkdir, so the
            # clean-FP render dir must exist BEFORE the per-dwelling render —
            # otherwise every real render raises FileNotFoundError and every
            # dwelling silently degrades to synthetic.
            out_dir.mkdir(parents=True, exist_ok=True)

        per_dwelling: Dict[str, Any] = {}
        provenance: Dict[str, Any] = {}
        real_count = 0
        synthetic_count = 0
        fallback_count = 0

        for fp_id, target in sorted(applicable.items()):
            bg_ids = target["bg_ids"]
            bg_shot_map = {
                b: list((catalog.get(b) or {}).get("applies_to_shots") or [])
                for b in bg_ids
            }
            if not real:
                per_dwelling[fp_id] = compute_synthetic_zone_map(
                    fp_id=fp_id, bg_ids=bg_ids, bg_shot_map=bg_shot_map)
                synthetic_count += 1
                continue
            if not _SAFE_FP_RE.match(fp_id):
                logger.error("dwelling_zone_map: unsafe fp_id %r — synthetic", fp_id)
                per_dwelling[fp_id] = compute_synthetic_zone_map(
                    fp_id=fp_id, bg_ids=bg_ids, bg_shot_map=bg_shot_map)
                fallback_count += 1
                continue
            bg_blocks = self._assemble_bg_descriptions(
                bg_ids=bg_ids, catalog=catalog, projection_cp=projection_cp,
                fresh_t2i_by_bg=fresh_t2i,
                dossier_facts={b: dossier_facts[b] for b in bg_ids if b in dossier_facts})
            try:
                result = self._run_real_dwelling(
                    fp_id=fp_id, bg_ids=bg_ids, bg_shot_map=bg_shot_map,
                    bg_blocks=bg_blocks, space_model=space_models.get(fp_id),
                    numbered_elements=numbered_by_fp.get(fp_id),
                    partition_plan=partition_plans.get(fp_id), out_dir=out_dir)
                per_dwelling[fp_id] = result["zone_map"]
                provenance[fp_id] = result["provenance"]
                real_count += 1
            except Exception as exc:
                logger.warning("dwelling_zone_map: real path failed for %s: %s", fp_id, exc)
                zm = compute_synthetic_zone_map(
                    fp_id=fp_id, bg_ids=bg_ids, bg_shot_map=bg_shot_map)
                zm["diagnostics"]["fallback_reason"] = f"{type(exc).__name__}: {exc}"[:200]
                per_dwelling[fp_id] = zm
                fallback_count += 1

        return {
            # fan_out=False, but count the real dwellings processed (Codex W21B
            # review minor: len(per_dwelling) is more honest than a 0/1 flag).
            "applicable_count": len(per_dwelling),
            "completed_count": len(per_dwelling),
            "failed_count": 0,
            "schema_version": SCHEMA_VERSION,
            "config_hash": self._config_hash(),
            "data": {
                "per_dwelling": per_dwelling,
                "provenance": provenance,
                "real_provider": real,
                "real_count": real_count,
                "synthetic_count": synthetic_count,
                "fallback_count": fallback_count,
                "not_applicable_dwellings": {
                    fp: t["not_applicable_reason"]
                    for fp, t in targets.items() if not t["applicable"]
                },
            },
        }

    @staticmethod
    def _space_models_by_fp(
        fp_prompt_cp: Optional[Dict[str, Any]]
    ) -> Dict[str, Dict[str, Any]]:
        """Per-fp ``space_model`` from the floor_plan_prompt checkpoint — the
        canonical dwelling structure SOT (every zone of the dwelling)."""
        out: Dict[str, Dict[str, Any]] = {}
        if not isinstance(fp_prompt_cp, dict):
            return out
        floor_plans = ((fp_prompt_cp.get("data") or {}).get("floor_plans") or {})
        for fp_id, spec in floor_plans.items():
            sm = (spec or {}).get("space_model")
            if isinstance(sm, dict) and isinstance(sm.get("zones"), list):
                out[fp_id] = sm
        return out

    @staticmethod
    def _numbered_elements_by_fp(
        fp_prompt_cp: Optional[Dict[str, Any]]
    ) -> Dict[str, List[Dict[str, Any]]]:
        """Per-fp ``numbered_elements`` from the floor_plan_prompt checkpoint.

        The fp prompt packs emit no ``space_model``; ``numbered_elements`` is
        the structure SOT they DO emit, so this is the production structure
        source. ``_space_models_by_fp`` stays the higher-priority path for any
        future pack that adds space_model."""
        out: Dict[str, List[Dict[str, Any]]] = {}
        if not isinstance(fp_prompt_cp, dict):
            return out
        floor_plans = ((fp_prompt_cp.get("data") or {}).get("floor_plans") or {})
        for fp_id, spec in floor_plans.items():
            ne = (spec or {}).get("numbered_elements")
            if isinstance(ne, list) and ne:
                out[fp_id] = ne
        return out

    @staticmethod
    def _dossier_facts_by_bg(
        dossier_cp: Optional[Dict[str, Any]]
    ) -> Dict[str, Any]:
        """Per-bg structural render facts from base_location_dossier (enrichment
        for the VLM context). Empty when the dossier step did not run."""
        out: Dict[str, Any] = {}
        if not isinstance(dossier_cp, dict):
            return out
        dossiers = ((dossier_cp.get("data") or {}).get("dossiers") or {})
        if isinstance(dossiers, dict):
            for _fp, dossier in dossiers.items():
                facts = (dossier or {}).get("per_bg_render_facts_by_bg_id") or {}
                if isinstance(facts, dict):
                    out.update(facts)
        return out

    @staticmethod
    def _partition_plans_by_fp(
        partition_cp: Optional[Dict[str, Any]], catalog: Dict[str, Any]
    ) -> Dict[str, Dict[str, Any]]:
        """Map each fp_id → its ``space_partition_plan`` (per-fp) from the
        bg_space_partition checkpoint, for the diagnostic comparison only."""
        if not isinstance(partition_cp, dict):
            return {}
        data = partition_cp.get("data") or {}
        plans = data.get("plans_by_fp") or data.get("space_partition_plans") or {}
        if isinstance(plans, dict) and plans:
            return {fp: p for fp, p in plans.items() if isinstance(p, dict)}
        return {}