"""W20A2.5: real VLM readback provider preflight.

Production-adjacent helper that, **when explicitly invoked by a future
wave**, performs a vision-LLM readback of a base floor-plan PNG and
returns a ``GeometryReadback`` dict with ``status='ok'``.

This module is deliberately separated from
``floor_plan_geometry_readback`` (the W20A2 core geometry module) so
that the core stays pure (no LLM/image/VLM imports). The step wrapper
chooses between the synthetic-fixture path and this real-provider path
via an opt-in setting + dependency injection; the core module only
sees a ``Callable`` slot.

**This wave does not invoke any real API call.** The helper
``litellm_vlm_provider`` exists and is import-safe — its only external
dependency (``litellm``) is imported lazily inside the function body,
so simply importing this module touches no external SDKs. The default
production path keeps ``settings.floor_plan_vlm_readback_real_provider_enabled = False``,
which means the step never resolves to this helper at all.

JSON-shape contract (validated by ``validate_provider_output``):

    {
      "status": "ok",
      "fp_id": <exact dossier fp_id>,
      "grid_size": [10, 10],
      "observed_markers": [
        {"number": int, "row": int, "col": int,
         "kind": one of BASE_KINDS}
      ],
      "missing_markers": [int],
      "extra_markers":   [int],
      "confidence": float | None,
      "diagnostics": [str]
    }

Hard rules:
  - ``status`` MUST be ``"ok"`` (real-provider path; synthetic /
    failed statuses are handled by the synthetic fixture or by raise).
  - ``fp_id`` MUST equal the dossier's ``fp_id`` exact-string.
  - Every ``observed_marker.number`` MUST belong to the dossier's
    ``base_marker_inventory`` (exact integer). Unknown numbers fail.
  - ``observed_marker.kind`` MUST be one of the W20A2 base kinds
    (``base_structural_unit`` / ``base_opening`` /
    ``base_persistent_fixture`` / ``base_persistent_furniture``).
    A ``state_overlay_*`` kind is **explicitly forbidden** — state /
    transient cues do not belong on the structural FP readback.
  - The marker's ``kind`` MUST equal the dossier inventory's
    ``base_layer_decision`` for that number — the VLM must not
    re-classify a marker.
  - ``(row, col)`` MUST be in-grid integers. Two distinct marker
    numbers sharing one cell is **not** a hard failure (the 10x10
    coarse grid genuinely permits two physical things to occupy the
    same cell); the validator surfaces a per-pair diagnostic instead.
    Duplicate marker NUMBERS remain a hard failure.
  - No duplicate marker numbers.
  - ``missing_markers`` and ``extra_markers`` lists must be present
    and consistent with the observed set vs the dossier inventory.
"""
from __future__ import annotations

import base64
import json
import logging
import os
from pathlib import Path
from typing import Any, Dict, FrozenSet, List, Optional, Tuple


logger = logging.getLogger(__name__)


# Mirror the W20A2 base kind enum without importing the geometry module
# (keeps this provider module's import graph independent of the core).
BASE_KINDS: FrozenSet[str] = frozenset(
    {
        "base_structural_unit",
        "base_opening",
        "base_persistent_fixture",
        "base_persistent_furniture",
    }
)
STATE_OVERLAY_KINDS: FrozenSet[str] = frozenset(
    {
        "state_overlay_plot_cue",
        "state_overlay_transient_object",
    }
)


# Helper identifier surfaced on records so audit / telemetry can
# distinguish real-provider readbacks from synthetic-fixture readbacks.
# NOTE: 'gpt55' 는 역사적 안정 식별자(provenance 매칭/lineage 호환) — 모델 교체와
# 무관하게 유지. 실제 물리 모델은 별도 model 인자/PROVIDER_MODEL_DEFAULT 에 기록되며
# 이 식별자에서 추론하지 않는다.
WIRED_PROVIDER_NAME: str = "litellm_gpt55_vision_v1"


# ─────────────────────────────────────────────────────────────────────
# W20A2.6 wire-up constants
# ─────────────────────────────────────────────────────────────────────


# Litellm-style model id. The "openai/" prefix routes through litellm's
# OpenAI provider so ``custom_llm_provider="openai"`` capability queries
# resolve correctly. Kept module-level so tests can assert the lock.
PROVIDER_MODEL_DEFAULT: str = "openai/gpt-5.6-sol"

# Token cap for the JSON readback. The schema is small (≤ 100 markers,
# ≤ 100 cells of (r,c) ints + 4 enum kinds), so 2000 leaves headroom
# for diagnostics strings without inviting model rambling. ``temperature``
# is intentionally omitted from the litellm call — the gpt-5 family
# does not expose a tunable temperature, and passing one trips strict
# response_schema validation on certain routes.
MAX_COMPLETION_TOKENS_DEFAULT: int = 16000  # W20F7-A (Codex 2026-05-28): 2000→16000. gpt-5.5 reasoning+output 합산 budget. 큰 도면 이미지 input 의 reasoning 폭주로 finish_reason=stop + content="" 빈 응답이 fp_wheelhouse 같은 fp 에서 재현됨. retry 도입 없이 budget 확보로 가설 검증.


# Strict OpenAI-compatible JSON schema. All object properties are
# ``required``; every object closes via ``additionalProperties: false``.
# No ``anyOf`` / ``oneOf`` / ``allOf`` / ``minItems`` / ``maxItems`` /
# ``uniqueItems`` / pattern / format / numeric range keywords are used —
# OpenAI strict mode rejects them. Nullable fields use the array-of-types
# form (``{"type": ["number", "null"]}``), not ``anyOf``.
READBACK_SCHEMA: Dict[str, Any] = {
    "type": "object",
    "additionalProperties": False,
    "required": [
        "status",
        "fp_id",
        "grid_size",
        "observed_markers",
        "missing_markers",
        "extra_markers",
        "confidence",
        "diagnostics",
    ],
    "properties": {
        "status": {"type": "string", "enum": ["ok"]},
        "fp_id": {"type": "string"},
        "grid_size": {
            "type": "array",
            "items": {"type": "integer"},
        },
        "observed_markers": {
            "type": "array",
            "items": {
                "type": "object",
                "additionalProperties": False,
                "required": ["number", "row", "col", "kind"],
                "properties": {
                    "number": {"type": "integer"},
                    "row": {"type": "integer"},
                    "col": {"type": "integer"},
                    "kind": {
                        "type": "string",
                        "enum": [
                            "base_structural_unit",
                            "base_opening",
                            "base_persistent_fixture",
                            "base_persistent_furniture",
                        ],
                    },
                },
            },
        },
        "missing_markers": {
            "type": "array",
            "items": {"type": "integer"},
        },
        "extra_markers": {
            "type": "array",
            "items": {"type": "integer"},
        },
        "confidence": {"type": ["number", "null"]},
        "diagnostics": {
            "type": "array",
            "items": {"type": "string"},
        },
    },
}


# Keys that OpenAI's strict structured-outputs mode does not support.
# Code-side guard: tests walk ``READBACK_SCHEMA`` recursively and assert
# none of these appear, so a future schema edit cannot silently slip in
# an unsupported keyword.
_OPENAI_UNSUPPORTED_SCHEMA_KEYS: FrozenSet[str] = frozenset(
    {
        "anyOf",
        "oneOf",
        "allOf",
        "not",
        "minItems",
        "maxItems",
        "uniqueItems",
        "minLength",
        "maxLength",
        "pattern",
        "format",
        "minimum",
        "maximum",
        "exclusiveMinimum",
        "exclusiveMaximum",
        "multipleOf",
        "patternProperties",
        "contains",
        "minContains",
        "maxContains",
    }
)


# Generic observer prompt. Carries no scenario-specific tokens — every
# fact the VLM needs about the current dwelling is injected as a
# structured JSON inventory in the user prompt. The system prompt
# locks the role (observer, not classifier), the coordinate system
# (top-left zero-indexed grid), and the output contract (JSON matching
# the response schema, no commentary).
_SYSTEM_PROMPT_GENERIC: str = (
    "You are a vision-only observer of a base floor-plan rendering. "
    "Examine the rendered PNG and report which numbered base markers "
    "from the dossier inventory are visible. "
    "Coordinate system contract (apply verbatim): grid_size is "
    "[rows, cols]; the grid is zero-indexed with a top-left origin; "
    "row increases downward, col increases rightward; every reported "
    "cell MUST satisfy 0 <= row < rows and 0 <= col < cols. Do not use "
    "1-indexed coordinates, do not use a bottom-left origin, do not "
    "swap row and col. "
    "You do not classify or rename markers — every marker's base kind "
    "is fixed by the dossier inventory and must be carried through "
    "verbatim. "
    "Respond with one JSON document matching the provided response "
    "schema; emit no prose, no labels, no scenario-specific guesses, "
    "no markdown."
)


class VlmProviderError(Exception):
    """Fail-closed signal for the real-VLM provider path."""


# ─────────────────────────────────────────────────────────────────────
# Output validator (pure — no I/O)
# ─────────────────────────────────────────────────────────────────────


def _is_int(value: Any) -> bool:
    return isinstance(value, int) and not isinstance(value, bool)


def _dossier_inventory_map(
    dossier: Dict[str, Any],
) -> Dict[int, str]:
    """Build ``{marker_number: base_layer_decision}`` from a dossier.

    Only ``base_*`` markers are surfaced; overlay markers are
    intentionally excluded from the VLM readback's authority set.
    """
    out: Dict[int, str] = {}
    for entry in (dossier or {}).get("base_marker_inventory") or []:
        if not isinstance(entry, dict):
            continue
        try:
            n = int(entry["number"])
        except (KeyError, TypeError, ValueError):
            continue
        decision = entry.get("base_layer_decision")
        if isinstance(decision, str) and decision in BASE_KINDS:
            out[n] = decision
    return out


def validate_provider_output(
    *,
    output: Any,
    dossier: Dict[str, Any],
    grid_size: Tuple[int, int],
    fp_id: str,
) -> Dict[str, Any]:
    """Pure JSON-shape + content validator for VLM provider output.

    Returns ``{"ok": bool, "blockers": [str], "readback": dict | None}``.

    Code performs only structural / exact-ID checks. No semantic /
    lexical inspection of any field is performed (matches the W20
    boundary: code is the validator, LLM/VLM is the observer).
    """
    blockers: List[str] = []
    if not isinstance(output, dict):
        return {
            "ok": False,
            "blockers": [
                f"provider output is not a dict ({type(output).__name__})"
            ],
            "readback": None,
        }

    if output.get("status") != "ok":
        blockers.append(
            f"status must be 'ok' (got {output.get('status')!r}); the "
            f"real-provider path only emits 'ok' — synthetic / failed "
            f"are handled elsewhere"
        )

    if output.get("fp_id") != fp_id:
        blockers.append(
            f"fp_id mismatch: provider={output.get('fp_id')!r} vs "
            f"dossier={fp_id!r}"
        )

    raw_grid = output.get("grid_size")
    if (
        not isinstance(raw_grid, list)
        or len(raw_grid) != 2
        or not all(_is_int(v) and v > 0 for v in raw_grid)
    ):
        blockers.append(
            f"grid_size must be a 2-list of positive ints "
            f"(got {raw_grid!r})"
        )
    elif tuple(raw_grid) != tuple(grid_size):
        blockers.append(
            f"grid_size mismatch: provider={tuple(raw_grid)} vs "
            f"expected={tuple(grid_size)}"
        )

    inventory = _dossier_inventory_map(dossier)
    if not inventory:
        blockers.append(
            "dossier base_marker_inventory has zero base_* markers; "
            "the VLM has nothing to attest to"
        )

    obs = output.get("observed_markers")
    if not isinstance(obs, list):
        blockers.append("observed_markers must be a list")
        obs = []

    seen_numbers: List[int] = []
    seen_cells: List[Tuple[int, int]] = []
    cell_owner: Dict[Tuple[int, int], int] = {}
    cell_collision_diagnostics: List[str] = []
    rows, cols = (
        (int(raw_grid[0]), int(raw_grid[1]))
        if isinstance(raw_grid, list) and len(raw_grid) == 2
        and all(_is_int(v) for v in raw_grid)
        else grid_size
    )
    for idx, entry in enumerate(obs):
        prefix = f"observed_markers[{idx}]"
        if not isinstance(entry, dict):
            blockers.append(f"{prefix} not a dict")
            continue
        n = entry.get("number")
        r = entry.get("row")
        c = entry.get("col")
        k = entry.get("kind")
        if not _is_int(n):
            blockers.append(f"{prefix}.number must be int (got {n!r})")
            continue
        if n in seen_numbers:
            blockers.append(f"{prefix}.number={n} duplicated")
        seen_numbers.append(n)
        if not _is_int(r) or not _is_int(c):
            blockers.append(
                f"{prefix} row/col must be int (got row={r!r}, col={c!r})"
            )
            continue
        if not (0 <= r < rows and 0 <= c < cols):
            blockers.append(
                f"{prefix} cell ({r},{c}) out of grid ({rows}x{cols})"
            )
        cell = (r, c)
        if cell in seen_cells:
            prior_owner = cell_owner.get(cell)
            cell_collision_diagnostics.append(
                f"{prefix} cell ({r},{c}) shares a 10x10 cell with "
                f"marker #{prior_owner} (W20E6-A: surfaced as diagnostic, "
                f"not a hard blocker -- distinct marker numbers may "
                f"genuinely occupy the same coarse cell)"
            )
        else:
            cell_owner[cell] = int(n)
        seen_cells.append(cell)
        if k in STATE_OVERLAY_KINDS:
            blockers.append(
                f"{prefix}.kind={k!r} is a state-overlay kind; the FP "
                f"VLM readback must not include overlay markers (W20 §2.4)"
            )
            continue
        if k not in BASE_KINDS:
            blockers.append(
                f"{prefix}.kind={k!r} is not a known base kind"
            )
            continue
        if n in inventory:
            expected = inventory[n]
            if k != expected:
                blockers.append(
                    f"{prefix}.kind={k!r} disagrees with dossier "
                    f"base_layer_decision={expected!r} for marker #{n}"
                )
        else:
            blockers.append(
                f"{prefix}.number={n} not in dossier base_marker_inventory"
            )

    inventory_numbers = set(inventory)
    observed_numbers = set(seen_numbers)

    # missing/extra must be lists whose content exactly equals the
    # dossier-vs-observed set difference. Order-insensitive but
    # duplicate-sensitive: a provider that drops or fabricates entries
    # fails the validator (instead of being silently corrected by the
    # normalized readback further below). This is the W20A2.5 narrow
    # patch — the validator must not let a provider self-report
    # inconsistent missing/extra and pass.
    for field, expected_set in (
        ("missing_markers", inventory_numbers - observed_numbers),
        ("extra_markers",   observed_numbers - inventory_numbers),
    ):
        raw = output.get(field)
        if not isinstance(raw, list):
            blockers.append(f"{field} must be a list (got {raw!r})")
            continue
        type_ok = True
        for v in raw:
            if not _is_int(v):
                blockers.append(
                    f"{field} contains non-int value {v!r}"
                )
                type_ok = False
        if not type_ok:
            continue
        if len(set(raw)) != len(raw):
            blockers.append(
                f"{field} contains duplicate values: {sorted(raw)!r}"
            )
            continue
        if sorted(raw) != sorted(expected_set):
            blockers.append(
                f"{field} mismatch: provider={sorted(raw)!r} vs "
                f"expected={sorted(expected_set)!r} (must equal the "
                f"exact-ID set difference between dossier inventory "
                f"and observed markers)"
            )

    diags = output.get("diagnostics")
    if not isinstance(diags, list):
        blockers.append("diagnostics must be a list")
        diags = []

    confidence = output.get("confidence", None)
    if confidence is not None and not isinstance(
        confidence, (int, float)
    ) or isinstance(confidence, bool):
        blockers.append(
            f"confidence must be float|int|None (got {confidence!r})"
        )

    if blockers:
        return {"ok": False, "blockers": blockers, "readback": None}

    # Reassemble a normalized readback (sorts missing/extra
    # deterministically, drops unknown keys, stamps the provider name).
    # ``cell_collision_diagnostics`` carries the W20E6-A soft-degrade
    # surface -- distinct marker numbers that share a 10x10 cell. It is
    # additive: the existing ``diagnostics`` list still receives the
    # same per-pair messages so downstream consumers that only walk
    # ``diagnostics`` continue to see the evidence.
    merged_diagnostics: List[str] = list(diags) + list(
        cell_collision_diagnostics
    )
    readback: Dict[str, Any] = {
        "status": "ok",
        "fp_id": fp_id,
        "grid_size": [int(rows), int(cols)],
        "observed_markers": [
            {
                "number": int(e["number"]),
                "row": int(e["row"]),
                "col": int(e["col"]),
                "kind": str(e["kind"]),
            }
            for e in obs
        ],
        "missing_markers": sorted(inventory_numbers - observed_numbers),
        "extra_markers": sorted(observed_numbers - inventory_numbers),
        "confidence": (
            float(confidence) if isinstance(confidence, (int, float)) else None
        ),
        "diagnostics": merged_diagnostics,
        "cell_collision_diagnostics": list(cell_collision_diagnostics),
        "provider_name": WIRED_PROVIDER_NAME,
    }
    return {"ok": True, "blockers": [], "readback": readback}


# ─────────────────────────────────────────────────────────────────────
# Request assembly helpers (pure — no network)
# ─────────────────────────────────────────────────────────────────────


def _normalize_grid_size(grid_size: Any) -> Tuple[int, int]:
    """Coerce ``grid_size`` to a ``(rows, cols)`` tuple of positive ints.

    Fails closed with ``VlmProviderError`` for any malformed shape:
    non-list/tuple, length other than 2, non-int element (including
    ``bool``, which is an ``int`` subclass in Python but should never
    be treated as a grid dimension), or non-positive value. The guard
    runs before any env / litellm / filesystem step so a bad caller
    cannot inflate the completion call counter.
    """
    if not isinstance(grid_size, (list, tuple)):
        raise VlmProviderError(
            f"grid_size must be a 2-element list or tuple (got "
            f"{type(grid_size).__name__})"
        )
    if len(grid_size) != 2:
        raise VlmProviderError(
            f"grid_size must have exactly 2 elements (got {len(grid_size)})"
        )
    coerced: List[int] = []
    for value in grid_size:
        if isinstance(value, bool) or not isinstance(value, int):
            raise VlmProviderError(
                f"grid_size elements must be non-bool integers "
                f"(got {value!r} of type {type(value).__name__})"
            )
        if value <= 0:
            raise VlmProviderError(
                f"grid_size elements must be positive integers (got {value!r})"
            )
        coerced.append(int(value))
    return coerced[0], coerced[1]


def _build_dossier_facts(
    *,
    dossier: Dict[str, Any],
    grid_size: Tuple[int, int],
) -> Dict[str, Any]:
    """Project the dossier into the JSON inventory the VLM sees.

    Generic shape: ``fp_id``, ``grid_size``, ``base_marker_inventory``.
    The marker entries carry their dossier-stamped ``base_layer_decision``
    so the VLM cannot re-classify; it can only observe placement.
    """
    inventory_out: List[Dict[str, Any]] = []
    for entry in (dossier or {}).get("base_marker_inventory") or []:
        if not isinstance(entry, dict):
            continue
        inventory_out.append(
            {
                "number": entry.get("number"),
                "label": entry.get("label"),
                "category": entry.get("category"),
                "base_layer_decision": entry.get("base_layer_decision"),
            }
        )
    return {
        "fp_id": dossier.get("fp_id"),
        "grid_size": [int(grid_size[0]), int(grid_size[1])],
        "base_marker_inventory": inventory_out,
    }


def _build_messages(
    *,
    dossier_facts: Dict[str, Any],
    image_data_url: str,
) -> List[Dict[str, Any]]:
    """Build the chat messages list — system + user multimodal content."""
    user_intro = (
        "Dossier facts (JSON, generic schema — fp_id, grid_size, "
        "base_marker_inventory). Use these to constrain your output. "
        "Every marker's kind in your response MUST equal its dossier "
        "base_layer_decision; do not re-classify markers."
    )
    user_outro = (
        "Now respond with the JSON readback per the response schema. "
        "status MUST be 'ok'. fp_id MUST equal the dossier fp_id. "
        "grid_size MUST equal the dossier grid_size. observed_markers "
        "may list any subset of the dossier inventory you can see; "
        "missing_markers and extra_markers MUST equal the exact set "
        "difference between the dossier inventory and your observed "
        "set. No commentary outside the JSON document."
    )
    return [
        {"role": "system", "content": _SYSTEM_PROMPT_GENERIC},
        {
            "role": "user",
            "content": [
                {"type": "text", "text": user_intro},
                {
                    "type": "text",
                    "text": json.dumps(dossier_facts, sort_keys=True),
                },
                {
                    "type": "image_url",
                    "image_url": {
                        "url": image_data_url,
                        "detail": "original",
                    },
                },
                {"type": "text", "text": user_outro},
            ],
        },
    ]


def _build_response_format() -> Dict[str, Any]:
    return {
        "type": "json_schema",
        "json_schema": {
            "name": "floor_plan_vlm_readback",
            "strict": True,
            "schema": READBACK_SCHEMA,
        },
    }


# ─────────────────────────────────────────────────────────────────────
# Production-adjacent helper — wired in W20A2.6
# ─────────────────────────────────────────────────────────────────────


def litellm_vlm_provider(
    *,
    dossier: Dict[str, Any],
    fp_image_path: Optional[str],
    grid_size: Tuple[int, int] = (10, 10),
    timeout_seconds: int = 120,
    model: str = PROVIDER_MODEL_DEFAULT,
    max_completion_tokens: int = MAX_COMPLETION_TOKENS_DEFAULT,
) -> Dict[str, Any]:
    """Real-VLM readback via a single litellm vision completion call.

    The step wrapper resolves to this helper only when
    ``settings.floor_plan_vlm_readback_real_provider_enabled`` is True
    AND the step is opt-in enabled. The default-OFF selector keeps
    this function out of every default code path; W20A2.6 wires the
    function body but does **not** flip any selector.

    Flow (fail-closed at every step):

      1. Arg-shape guards: ``fp_image_path`` non-None,
         ``dossier.fp_id`` present.
      2. Env preflight: ``OPENAI_API_KEY`` non-empty.
      3. Lazy ``import litellm``.
      4. Litellm capability preflight:
         ``supports_response_schema(model=model)`` True and
         ``response_format`` listed in
         ``get_supported_openai_params(model=model,
         custom_llm_provider='openai')``.
      5. Filesystem preflight: ``fp_image_path`` exists and is readable.
      6. Read bytes, base64-encode to a ``data:image/png;base64,...``
         URL.
      7. Assemble messages (generic system + structured-JSON user
         content + image_url with detail ``original``).
      8. Single ``litellm.completion`` call (strict ``json_schema``
         response_format, ``num_retries=0``, no temperature).
      9. Response guards: non-empty ``choices``, ``message`` present,
         no truthy ``refusal``, non-empty ``content``,
         ``finish_reason == 'stop'`` exactly (``length`` /
         ``content_filter`` fail closed).
     10. Parse JSON content; validate via
         ``validate_provider_output``; on validator failure raise
         ``VlmProviderError('validator_failed: ...')``.
     11. Return the normalized readback dict.

    Any failure path raises ``VlmProviderError`` BEFORE issuing the
    network call when the failure precedes step 8. The completion call
    count is therefore 0 whenever the function fail-closes during
    preflight.
    """
    # 1. Arg-shape guards (pre-env, pre-network).
    if fp_image_path is None:
        raise VlmProviderError(
            "litellm_vlm_provider requires fp_image_path; got None"
        )
    fp_id = dossier.get("fp_id") if isinstance(dossier, dict) else None
    if not isinstance(fp_id, str) or not fp_id:
        raise VlmProviderError("dossier.fp_id missing or non-string")
    # grid_size shape guard runs here so a malformed dim cannot reach
    # the env/litellm/filesystem/completion stages. completion call
    # counter therefore stays 0 on bad input.
    rows, cols = _normalize_grid_size(grid_size)

    # 2. Env preflight (pre-network).
    from app.core.openai_keys import has_openai_key
    if not has_openai_key():
        raise VlmProviderError(
            "OPENAI_API_KEY missing or empty; refusing to call litellm"
        )

    # 3. Lazy import litellm (keeps the dependency out of module-import
    # graph for callers that never flip the real-provider selector).
    try:
        import litellm  # type: ignore
        from app.core.openai_keys import (  # type: ignore
            llm_completion as _llm_completion,
        )
    except Exception as exc:  # pragma: no cover — env-dependent
        raise VlmProviderError(
            f"litellm import failed: {type(exc).__name__}: {exc}"
        ) from exc

    # 4. Litellm capability preflight (pre-network).
    try:
        supports_ok = bool(litellm.supports_response_schema(model=model))
    except Exception as exc:
        raise VlmProviderError(
            f"litellm.supports_response_schema raised: "
            f"{type(exc).__name__}: {exc}"
        ) from exc
    if not supports_ok:
        raise VlmProviderError(
            f"model {model!r} does not advertise response_schema "
            f"support in litellm; refusing to call"
        )
    try:
        supported_params = (
            litellm.get_supported_openai_params(
                model=model, custom_llm_provider="openai"
            )
            or []
        )
    except Exception as exc:
        raise VlmProviderError(
            f"litellm.get_supported_openai_params raised: "
            f"{type(exc).__name__}: {exc}"
        ) from exc
    if "response_format" not in supported_params:
        raise VlmProviderError(
            f"model {model!r} does not list response_format in supported "
            f"openai params; refusing to call"
        )

    # 5. Filesystem preflight (pre-network).
    path_obj = Path(fp_image_path)
    if not path_obj.exists() or not path_obj.is_file():
        raise VlmProviderError(
            f"fp_image_path does not exist on disk: {fp_image_path!r}"
        )
    try:
        image_bytes = path_obj.read_bytes()
    except Exception as exc:
        raise VlmProviderError(
            f"fp_image_path read failed: {type(exc).__name__}: {exc}"
        ) from exc
    if not image_bytes:
        raise VlmProviderError(
            f"fp_image_path is empty: {fp_image_path!r}"
        )

    # 6. Base64 data URL.
    image_b64 = base64.b64encode(image_bytes).decode("ascii")
    image_data_url = f"data:image/png;base64,{image_b64}"

    # 7. Assemble request payload. rows/cols were already normalized
    # at step 1 by ``_normalize_grid_size``.
    dossier_facts = _build_dossier_facts(
        dossier=dossier, grid_size=(rows, cols)
    )
    messages = _build_messages(
        dossier_facts=dossier_facts, image_data_url=image_data_url
    )
    response_format = _build_response_format()

    # 8. Single completion call. ``num_retries=0`` so a transient
    # failure does not silently inflate counters; ``temperature``
    # omitted (gpt-5 family does not expose tunable temperature and
    # strict response_schema rejects unknown params on some routes).
    try:
        response = _llm_completion(
            model=model,
            messages=messages,
            response_format=response_format,
            timeout=timeout_seconds,
            max_completion_tokens=max_completion_tokens,
            num_retries=0,
        )
    except Exception as exc:
        raise VlmProviderError(
            f"litellm.completion raised: {type(exc).__name__}: {exc}"
        ) from exc

    # 9. Response guards (fail-closed at every joint).
    choices = getattr(response, "choices", None) or []
    if not choices:
        raise VlmProviderError("response.choices is empty")
    choice = choices[0]
    msg = getattr(choice, "message", None)
    if msg is None:
        raise VlmProviderError("response.choices[0].message is missing")
    refusal = getattr(msg, "refusal", None)
    if refusal:
        raise VlmProviderError(
            f"VLM emitted a refusal: {str(refusal)[:200]!r}"
        )
    content = getattr(msg, "content", None)
    if not content:
        raise VlmProviderError(
            "response.choices[0].message.content is empty"
        )
    finish_reason = getattr(choice, "finish_reason", None)
    if finish_reason != "stop":
        raise VlmProviderError(
            f"finish_reason must be 'stop' exactly (got "
            f"{finish_reason!r}); length/content_filter fail closed"
        )

    # 10. Parse JSON content.
    try:
        parsed = json.loads(content)
    except (TypeError, ValueError) as exc:
        raise VlmProviderError(
            f"response content is not valid JSON: {type(exc).__name__}: "
            f"{exc}"
        ) from exc

    # 11. Validate against dossier + grid contract.
    val = validate_provider_output(
        output=parsed,
        dossier=dossier,
        grid_size=(rows, cols),
        fp_id=fp_id,
    )
    if not val["ok"]:
        joined = "; ".join(val["blockers"])
        raise VlmProviderError(f"validator_failed: {joined[:400]}")

    return val["readback"]