"""Pure, dependency-free contracts for DiffusionGemma visual grounding.

This module deliberately has no ComfyUI, Transformers, Torch, or network imports.
It owns the host-verifiable parts of the Grounding Guard: settings, asset IDs,
strict ledger validation, refusal-clause detection, guard decisions, compact
reports, and JSON-only trace persistence.
"""

from __future__ import annotations

import copy
import json
import math
import re
import uuid
from dataclasses import asdict, dataclass, field
from decimal import Decimal, DecimalException
from datetime import datetime, timezone
from pathlib import Path, PurePath
from typing import Any, Mapping, Sequence


GROUNDING_CONFIG_TYPE = "DG_GROUNDING_GUARD_CONFIG"
GROUNDING_LEDGER_SCHEMA_ID = "dg-grounding-ledger/1"
ASSET_REGISTRY_SCHEMA_ID = "dg-asset-registry/1"
GROUNDING_REPORT_SCHEMA_ID = "dg-grounding-report/1"
EXTERNAL_EVIDENCE_SCHEMA_ID = "dg-external-evidence/1"
GROUNDING_EVIDENCE_SCHEMA_PATH = (
    Path(__file__).resolve().parent / "schemas" / "grounding_evidence.schema.json"
)
EXTERNAL_EVIDENCE_SCHEMA_PATH = (
    Path(__file__).resolve().parent / "schemas" / "external_evidence.schema.json"
)

GROUNDING_MODES = ("off", "audit", "strict")
SAMPLING_PROFILES = ("checkpoint_defaults", "full_48_diagnostic")
EVIDENCE_TOKEN_BUDGETS = ("auto", "768", "1024", "1280")
ANALYSIS_STATUSES = (
    "not_run",
    "not_applicable",
    "grounded",
    "uncertain",
    "refused",
    "transport_error",
)
LEDGER_ANALYSIS_STATUSES = ("grounded", "uncertain", "refused", "transport_error")
GUARD_DECISIONS = ("disabled", "not_applicable", "pass", "warn", "block")
CONFIDENCE_LEVELS = ("high", "medium", "low")
FACT_CATEGORIES = (
    "identity",
    "appearance",
    "object",
    "count",
    "color",
    "text",
    "spatial",
    "environment",
    "lighting",
    "action",
    "motion",
    "camera",
    "composition",
    "temporal",
    "other",
)
_FACT_CATEGORY_ALIASES = {
    # The model sometimes describes visible clothing with this intuitive label.
    # Keep the public ledger taxonomy canonical and normalize only this exact,
    # unambiguous alias at the model-output validation boundary.
    "wardrobe": "appearance",
}
KNOWN_EXTERNAL_PROVIDERS = frozenset(
    {"manual", "ocr", "detector", "pose", "tracking", "motion", "siglip"}
)

DEFAULT_TRACE_SUBFOLDER = "diffusiongemma_grounding"
MAX_COMPACT_REPORT_BYTES = 256 * 1024
MAX_TRACE_BYTES = 8 * 1024 * 1024
UINT64_MAX = (1 << 64) - 1


def _canonicalize_fact_categories(
    categories_value: Any,
) -> tuple[list[str] | None, tuple[tuple[str, str], ...]]:
    """Return canonical model-authored categories without accepting unknown values."""

    if not isinstance(categories_value, list) or not categories_value:
        return None, ()
    if any(not isinstance(category, str) for category in categories_value):
        return None, ()
    # Preserve the existing strict rule for duplicate model output. A collision
    # introduced only by an approved alias (for example appearance + wardrobe)
    # is safely collapsed below.
    if len(set(categories_value)) != len(categories_value):
        return None, ()

    canonical: list[str] = []
    canonicalizations: list[tuple[str, str]] = []
    seen: set[str] = set()
    for category in categories_value:
        normalized = _FACT_CATEGORY_ALIASES.get(category, category)
        if normalized not in FACT_CATEGORIES:
            return None, ()
        if normalized != category:
            canonicalizations.append((category, normalized))
        if normalized not in seen:
            canonical.append(normalized)
            seen.add(normalized)
    return canonical, tuple(canonicalizations)

SAMPLING_PROFILE_SETTINGS: dict[str, dict[str, Any]] = {
    "checkpoint_defaults": {
        "max_denoising_steps": 48,
        "temperature_start": 0.8,
        "temperature_end": 0.4,
        "entropy_bound": 0.1,
        "token_stability_threshold": 1,
        "confidence_threshold": 0.005,
        "adaptive_stopping": True,
    },
    "full_48_diagnostic": {
        "max_denoising_steps": 48,
        "temperature_start": 0.8,
        "temperature_end": 0.4,
        "entropy_bound": 0.1,
        "token_stability_threshold": 1,
        "confidence_threshold": 0.005,
        "adaptive_stopping": False,
    },
}

_FACT_ID_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9_.:-]{0,63}$")
_CLAIM_TYPE_RE = re.compile(r"^[a-z][a-z0-9_.:-]{0,63}$")
_ASSET_ID_RE = re.compile(r"^(?:image|picture|video):[1-9][0-9]*$")
_MANIFEST_LINE_RE = re.compile(
    r"<(?P<kind>Picture|Video)\s+(?P<number>[1-9][0-9]*)>\s*:\s*(?P<role>[^\r\n]+)",
    re.IGNORECASE,
)

_ROLE_SHARED_FACT_CATEGORIES = {
    "object",
    "count",
    "color",
    "text",
    "spatial",
    "other",
}


def _expand_role_category_family(categories: set[str]) -> set[str]:
    """Allow tightly coupled wording without weakening cross-role boundaries."""

    expanded = set(categories)
    for family in (
        {"identity", "appearance"},
        {"environment", "lighting", "composition"},
        {"action", "motion", "temporal"},
        {"camera", "composition", "motion", "temporal"},
    ):
        if categories & family:
            expanded.update(family)
    return expanded


_ROLE_CATEGORY_GROUPS: tuple[tuple[re.Pattern[str], set[str]], ...] = (
    (
        re.compile(
            r"\b(?:identity|appearance|face|body|silhouette|wardrobe|accessor(?:y|ies)|"
            r"distinguishing|subject|character|protagonist|likeness|outfit|costume|attire|"
            r"garment)\b",
            re.IGNORECASE,
        ),
        {"identity", "appearance"},
    ),
    (
        re.compile(
            r"\b(?:environment|setting|background|location|production\s+design|style|palette|"
            r"lighting|scenery|architecture|landscape)\b",
            re.IGNORECASE,
        ),
        {"environment", "lighting", "composition"},
    ),
    (
        re.compile(
            r"\b(?:action|motion|pose|movement|timing|choreograph\w*|performance|gesture|"
            r"blocking|temporal|whole[-\s]?video\s+edit\w*|edit\w*|continuation|"
            r"continue\w*|cut|cuts|rhythm)\b",
            re.IGNORECASE,
        ),
        {"action", "motion", "composition", "temporal"},
    ),
    (
        re.compile(r"\b(?:camera|framing|shot|composition|lens|angle)\b", re.IGNORECASE),
        {"camera", "composition", "motion", "temporal"},
    ),
    (
        re.compile(r"\b(?:object|product|item|prop|logo|packaging)\b", re.IGNORECASE),
        {"object"},
    ),
    (
        re.compile(r"\b(?:count|number|quantity|how\s+many)\b", re.IGNORECASE),
        {"count"},
    ),
    (
        re.compile(r"\b(?:color|colour|hue|palette)\b", re.IGNORECASE),
        {"color"},
    ),
    (
        re.compile(
            r"\b(?:text|title|copy|lettering|logo|typography|label|caption|wording)\b",
            re.IGNORECASE,
        ),
        {"text"},
    ),
    (
        re.compile(
            r"\b(?:spatial|layout|position|placement|arrangement|left[-\s]?right|"
            r"top[-\s]?bottom|geometry)\b",
            re.IGNORECASE,
        ),
        {"spatial"},
    ),
)
_ROLE_FORBIDS_IDENTITY_RE = re.compile(
    r"(?:"
    r"\b(?:exclude|ignore|omit)\b[^.;\r\n]{0,48}\bidentity\b"
    r"|"
    r"\b(?:do\s+not|don't|must\s+not|should\s+not|never)\b"
    r"[^.;\r\n]{0,32}\b(?:copy|use|derive|take|transfer|source)\w*\b"
    r"[^.;\r\n]{0,48}\bidentity\b"
    r"|"
    r"\b(?:without|no)\b[^.;\r\n]{0,32}\bidentity\b"
    r"[^.;\r\n]{0,32}\b(?:transfer|source|copy|override|replacement)\b"
    r"|"
    r"\bidentity\b[^.;\r\n]{0,96}"
    r"(?:\b(?:must\s+not|should\s+not|does\s+not|is\s+not|cannot|can't|won't|"
    r"mustn't|shouldn't)\b[^.;\r\n]{0,32}\b(?:transfer|override|replace|copy|"
    r"source)\w*\b|\bnot\s+a\s+source\b)"
    r")",
    re.IGNORECASE,
)
_EXPLICIT_ROLE_CATEGORIES_RE = re.compile(
    r"\[\s*dg\s*:\s*([a-z0-9_,\s-]+)\]",
    re.IGNORECASE,
)
_CLAIM_CATEGORY_GROUPS: tuple[tuple[re.Pattern[str], set[str]], ...] = (
    (
        re.compile(
            r"\b(?:identity|appearance|face|facial|body\s+type|hair|wardrobe|clothing|"
            r"beard(?:ed)?|distinguishing\s+trait|same\s+person|likeness)\b",
            re.IGNORECASE,
        ),
        {"identity", "appearance"},
    ),
    (
        re.compile(
            r"\b(?:environment|setting|background|location|town[-\s]?square|production\s+design|"
            r"style|palette|lighting|scenery|architecture|landscape|plaza|cobblestone|"
            r"arches?|room|street|forest|building|interior|exterior)\b",
            re.IGNORECASE,
        ),
        {"environment", "lighting", "composition"},
    ),
    (
        re.compile(
            r"\b(?:action|motion|moves?|moving|walk\w*|run\w*|movement|timing|"
            r"choreograph\w*|performance|gesture|blocking|temporal|turn\w*|jump\w*|"
            r"raise\w*|lower\w*|enter\w*|exit\w*|fall\w*|dance\w*|"
            r"reach\w*|grab\w*|edit\w*|continuation|continue\w*|cuts?|rhythm)\b",
            re.IGNORECASE,
        ),
        {"action", "motion", "temporal"},
    ),
    (
        re.compile(
            r"\b(?:camera|framing|shot|lens|pan\w*|zoom\w*|dolly|tilt\w*|orbit\w*|composition)\b",
            re.IGNORECASE,
        ),
        {"camera", "composition", "motion"},
    ),
    (
        re.compile(r"\b(?:object|product|item|prop|logo|packaging)\b", re.IGNORECASE),
        {"object"},
    ),
    (
        re.compile(
            r"\b(?:count|number|quantity|one|two|three|four|five|six|seven|eight|nine|ten)\b|\b\d+\b",
            re.IGNORECASE,
        ),
        {"count"},
    ),
    (
        re.compile(
            r"\b(?:color|colour|hue|red|orange|yellow|green|blue|purple|violet|pink|"
            r"brown|black|white|gr[ae]y|gold|silver|scarlet|crimson|teal|cyan|magenta|ochre)\b",
            re.IGNORECASE,
        ),
        {"color"},
    ),
    (
        re.compile(
            r"\b(?:text|title|word|words|lettering|logo|typography|reads?|says?|spells?|"
            r"label|caption|wording)\b",
            re.IGNORECASE,
        ),
        {"text"},
    ),
    (
        re.compile(
            r"\b(?:spatial|layout|left|right|above|below|behind|front|center(?:ed)?|centred|"
            r"adjacent|next\s+to|position|placement|arrangement)\b",
            re.IGNORECASE,
        ),
        {"spatial"},
    ),
)

_CONCRETE_APPAREL_PATTERN = (
    r"(?:coats?|jackets?|shirts?|dress(?:es)?(?!\s+(?:rehearsal|code)\b)|"
    r"outfits?|costumes?|attire|garments?|"
    r"accessor(?:y|ies))(?!-)"
)
_PERSON_ROLE_PATTERN = (
    r"(?:persons?|subjects?|characters?|protagonists?|actors?|performers?|couriers?|"
    r"models?|individuals?|m(?:a|e)n|wom(?:a|e)n|girls?|boys?|hero(?:es)?|villains?|"
    r"people)"
)
_PROPER_NAME_PATTERN = r"(?-i:[A-Z][a-z]{1,31})"
_PERSON_REFERENCE_PATTERN = (
    rf"(?:{_PERSON_ROLE_PATTERN}|he|she|they|{_PROPER_NAME_PATTERN})"
)
_CONCRETE_APPAREL_NOUN_RE = re.compile(
    rf"\b{_CONCRETE_APPAREL_PATTERN}\b",
    re.IGNORECASE,
)
_APPAREL_NON_MODIFIER_WORDS = (
    r"a|an|the|this|that|these|those|his|her|their|its|in|on|at|by|for|from|of|"
    r"to|with|without|within|into|onto|over|under|around|near|beside|behind|before|"
    r"after|across|through|past|as|while|and|or|but|which|who|whose|is|are|was|"
    r"were|be|been|being|has|have|had|do|does|did|will|would|can|could|should|"
    r"must|may|might|see|sees|show|shows|showing|hang|hangs|hanging|display|"
    r"displays|displaying|provide|provides|provided|supply|supplies|supplied|use|"
    r"uses|used|take|takes|took|define|defines|control|controls|determine|determines|"
    r"stand|stands|standing|turn|turns|walk|walks|flutter|flutters|remain|remains|"
    r"appear|appears|include|includes|contain|contains"
)
_APPAREL_MODIFIER_TOKEN_PATTERN = (
    rf"(?!(?:{_APPAREL_NON_MODIFIER_WORDS})\b)[a-z0-9]+(?:-[a-z0-9]+)*"
)
_APPAREL_MODIFIER_SEQUENCE_PATTERN = (
    rf"(?:{_APPAREL_MODIFIER_TOKEN_PATTERN}(?:\s+|,\s*)){{0,8}}"
)
_APPAREL_PHRASE_PATTERN = (
    rf"(?:(?:the|an?|this|that|these|those|his|her|their|its)\s+)?"
    rf"{_APPAREL_MODIFIER_SEQUENCE_PATTERN}{_CONCRETE_APPAREL_PATTERN}\b"
)
_APPAREL_WEARER_CUE_RE = re.compile(
    rf"(?:\b{_PERSON_REFERENCE_PATTERN}\b\s*,?\s+"
    rf"(?:(?:is|are|was|were|has\s+been|had\s+been|will\s+be)\s+)?"
    rf"(?:[a-z]+ly\s+){{0,2}}"
    rf"(?:wear|wears|wearing|wore|dressed\s+in|clad\s+in|sporting|dons|donning)\s+"
    rf"{_APPAREL_PHRASE_PATTERN}"
    rf"|\b{_PERSON_REFERENCE_PATTERN}\b\s*,?\s+in\s+{_APPAREL_PHRASE_PATTERN}"
    rf"|\b(?:wearing|dressed\s+in|clad\s+in|donning)\s+{_APPAREL_PHRASE_PATTERN}"
    rf"|\b(?:wear|wears|wore|dons)\s+{_APPAREL_PHRASE_PATTERN}"
    rf"|\bsporting\s+(?:the|an?)\s+"
    rf"{_APPAREL_MODIFIER_SEQUENCE_PATTERN}{_CONCRETE_APPAREL_PATTERN}\b"
    rf"|{_APPAREL_PHRASE_PATTERN}\s+"
    rf"(?:(?:is|are|was|were)\s+)?worn\s+by\s+"
    rf"(?:(?:the|an?)\s+)?{_PERSON_REFERENCE_PATTERN}\b"
    rf"|{_APPAREL_PHRASE_PATTERN}\s+on\s+"
    rf"(?:(?:the|an?)\s+)?{_PERSON_REFERENCE_PATTERN}\b)",
    re.IGNORECASE,
)
_PERSON_APPAREL_CUE_RE = re.compile(
    rf"(?:\b(?:his|her|their)\s+{_APPAREL_MODIFIER_SEQUENCE_PATTERN}"
    rf"{_CONCRETE_APPAREL_PATTERN}\b"
    rf"|\b{_PERSON_ROLE_PATTERN}['\u2019](?:s)?\s+"
    rf"{_APPAREL_MODIFIER_SEQUENCE_PATTERN}{_CONCRETE_APPAREL_PATTERN}\b)",
    re.IGNORECASE,
)
_PROPER_NAME_POSSESSIVE_APPAREL_RE = re.compile(
    rf"\b{_PROPER_NAME_PATTERN}['\u2019]s\s+"
    rf"{_APPAREL_MODIFIER_SEQUENCE_PATTERN}{_CONCRETE_APPAREL_PATTERN}\b"
)
_APPAREL_COMPOUND_WEARER_RE = re.compile(
    rf"(?:\b(?:[a-z0-9]+-)+(?:coated|jacketed|shirted|dressed|costumed)\s+"
    rf"{_PERSON_ROLE_PATTERN}\b"
    rf"|\b(?:coat|jacket|shirt|dress|outfit|costume|garment)-wearing\s+"
    rf"{_PERSON_ROLE_PATTERN}\b)",
    re.IGNORECASE,
)
_APPAREL_TRANSFER_CUE_RE = re.compile(
    r"(?:\b(?:cop(?:y|ies|ied|ying)|deriv(?:e|es|ed|ing)|"
    r"transfer(?:s|red|ring)?|sourc(?:e|es|ed|ing)|preserv(?:e|es|ed|ing)|"
    r"suppl(?:y|ies|ied|ying)|provid(?:e|es|ed|ing))\b\s+"
    rf"(?:directly\s+)?{_APPAREL_PHRASE_PATTERN}"
    rf"|\b(?:uses?|takes?)\s+{_APPAREL_PHRASE_PATTERN}"
    rf"|{_APPAREL_PHRASE_PATTERN}\s*(?:,|\u2014)?\s*"
    r"(?:(?:that\s+)?(?:(?:is|are|was|were)|(?:must|should|will|can)\s+be|"
    r"(?:has|have|had)\s+been)\s+)?(?:directly\s+)?"
    r"(?:copied|derived|transferred|sourced|preserved|supplied|provided)\b)",
    re.IGNORECASE,
)

_REFUSAL_PATTERNS: tuple[tuple[str, re.Pattern[str]], ...] = (
    (
        "apology_refusal",
        re.compile(
            r"\bi(?:\s+am|['\u2019]m)\s+sorry\s*,?\s+but\s+i\s+(?:cannot|can['\u2019]t|won['\u2019]t)\b",
            re.IGNORECASE,
        ),
    ),
    (
        "cannot_assist",
        re.compile(
            r"\bi\s+(?:cannot|can['\u2019]t|won['\u2019]t)\s+(?:"
            r"(?:assist|help|comply|provide|analy[sz]e|describe|process)\b|"
            r"(?:review|view)\s+(?:this|that|the)\s+(?:image|video|clip|media|content)\b)",
            re.IGNORECASE,
        ),
    ),
    (
        "unable_to_assist",
        re.compile(
            r"\bi(?:\s+am|['\u2019]m)\s+unable\s+to\s+(?:"
            r"(?:assist|help|comply|provide|analy[sz]e|describe|process)\b|"
            r"(?:review|view)\s+(?:this|that|the)\s+(?:image|video|clip|media|content)\b)",
            re.IGNORECASE,
        ),
    ),
    (
        "policy_refusal",
        re.compile(
            r"\b(?:this|that|your)\s+(?:request|content)\s+(?:violates|is\s+against|conflicts\s+with)\s+(?:my\s+)?(?:policy|policies|guidelines)\b",
            re.IGNORECASE,
        ),
    ),
    (
        "must_decline",
        re.compile(
            r"\bi\s+(?:must|have\s+to)\s+(?:decline|refuse)\s+(?:this|that|the)\s+request\b",
            re.IGNORECASE,
        ),
    ),
    (
        "apology_cannot",
        re.compile(
            r"\bi\s+apologi[sz]e\s*,?\s+but\s+i\s+(?:cannot|can['\u2019]t|won['\u2019]t)\b",
            re.IGNORECASE,
        ),
    ),
)

_FORBIDDEN_TRACE_KEYS = frozenset(
    {
        "logits",
        "raw_logits",
        "logits_tensor",
        "pixel_values",
        "pixels",
        "image_tensor",
        "video_tensor",
        "video_frames",
        "raw_media",
        "image_bytes",
        "video_bytes",
        "media_bytes",
    }
)


def _is_safe_pixel_tensor_descriptor(path: str, key: str, value: Any) -> bool:
    """Allow processor transport facts while continuing to reject pixel payloads."""

    if not path.endswith(".pixel_tensors") or not key.startswith("pixel_values"):
        return False
    if not isinstance(value, Mapping) or set(value) != {"shape", "dtype"}:
        return False
    shape = value.get("shape")
    dtype = value.get("dtype")
    return bool(
        isinstance(shape, (list, tuple))
        and shape
        and len(shape) <= 8
        and all(isinstance(item, int) and not isinstance(item, bool) and item >= 0 for item in shape)
        and all(item <= 2_147_483_647 for item in shape)
        and isinstance(dtype, str)
        and 0 < len(dtype) <= 128
    )


class GroundingGuardError(ValueError):
    """Base error for malformed Grounding Guard data."""


class GroundingLedgerError(GroundingGuardError):
    """Raised when a ledger cannot be extracted as strict JSON."""


@dataclass(frozen=True)
class GroundingGuardConfig:
    mode: str = "audit"
    retry_on_uncertain: bool = True
    sampling_profile: str = "checkpoint_defaults"
    seed: int = 0
    save_detailed_trace: bool = False
    trace_subfolder: str = DEFAULT_TRACE_SUBFOLDER
    external_evidence_json: str = ""
    evidence_token_budget: str = "auto"

    @property
    def sampling_settings(self) -> dict[str, Any]:
        return copy.deepcopy(SAMPLING_PROFILE_SETTINGS[self.sampling_profile])

    def to_dict(self) -> dict[str, Any]:
        result = asdict(self)
        result["sampling_settings"] = self.sampling_settings
        return result


@dataclass(frozen=True)
class GroundingLedgerValidation:
    ledger: dict[str, Any] | None
    schema_valid: bool
    grounded: bool
    analysis_status: str
    model_analysis_status: str = "uncertain"
    errors: tuple[str, ...] = ()
    warnings: tuple[str, ...] = ()
    blocking_reasons: tuple[str, ...] = ()
    covered_asset_ids: tuple[str, ...] = ()
    temporal_regions: dict[str, tuple[str, ...]] = field(default_factory=dict)
    accepted_external_claim_count: int = 0
    external_conflicts: tuple[str, ...] = ()
    refusal_matches: tuple[str, ...] = ()

    @property
    def valid(self) -> bool:
        """Compatibility alias for callers that treat verified grounding as valid."""

        return self.grounded

    def to_dict(self) -> dict[str, Any]:
        return {
            "schema_valid": self.schema_valid,
            "grounded": self.grounded,
            "analysis_status": self.analysis_status,
            "model_analysis_status": self.model_analysis_status,
            "errors": list(self.errors),
            "warnings": list(self.warnings),
            "blocking_reasons": list(self.blocking_reasons),
            "covered_asset_ids": list(self.covered_asset_ids),
            "temporal_regions": {
                key: list(value) for key, value in self.temporal_regions.items()
            },
            "accepted_external_claim_count": self.accepted_external_claim_count,
            "external_conflicts": list(self.external_conflicts),
            "refusal_matches": list(self.refusal_matches),
        }


@dataclass(frozen=True)
class GroundingGuardDecision:
    mode: str
    analysis_status: str
    decision: str
    would_block: bool
    blocked_reasons: tuple[str, ...] = ()
    warnings: tuple[str, ...] = ()

    def to_dict(self) -> dict[str, Any]:
        return {
            "mode": self.mode,
            "analysis_status": self.analysis_status,
            "decision": self.decision,
            "would_block": self.would_block,
            "blocked_reasons": list(self.blocked_reasons),
            "warnings": list(self.warnings),
        }


def _as_mapping(value: Any) -> dict[str, Any]:
    if isinstance(value, GroundingGuardConfig):
        return asdict(value)
    if isinstance(value, Mapping):
        return dict(value)
    if isinstance(value, str) and value.strip():
        try:
            parsed = json.loads(value)
        except json.JSONDecodeError:
            return {}
        return dict(parsed) if isinstance(parsed, Mapping) else {}
    return {}


def _as_bool(value: Any, default: bool) -> bool:
    if isinstance(value, bool):
        return value
    if isinstance(value, (int, float)) and not isinstance(value, bool):
        return bool(value)
    if isinstance(value, str):
        lowered = value.strip().lower()
        if lowered in {"1", "true", "yes", "on"}:
            return True
        if lowered in {"0", "false", "no", "off"}:
            return False
    return default


def _safe_trace_subfolder(value: Any) -> str:
    text = str(value or "").strip().replace("\\", "/")
    if not text:
        return DEFAULT_TRACE_SUBFOLDER
    pure = PurePath(text)
    if pure.is_absolute() or any(part in {"", ".", ".."} for part in pure.parts):
        return DEFAULT_TRACE_SUBFOLDER
    if re.match(r"^[A-Za-z]:", text):
        return DEFAULT_TRACE_SUBFOLDER
    return "/".join(pure.parts)


def normalize_grounding_guard_config(
    value: GroundingGuardConfig | Mapping[str, Any] | str | None = None,
    **overrides: Any,
) -> GroundingGuardConfig:
    """Return a safe, canonical settings value.

    Invalid externally supplied modes fail toward ``audit`` rather than silently
    disabling the guard. Seeds are clamped to ComfyUI's unsigned 64-bit range.
    """

    raw = _as_mapping(value)
    raw.update({key: item for key, item in overrides.items() if item is not None})

    mode = str(raw.get("mode", "audit")).strip().lower()
    if mode not in GROUNDING_MODES:
        mode = "audit"
    profile = str(raw.get("sampling_profile", "checkpoint_defaults")).strip().lower()
    if profile not in SAMPLING_PROFILES:
        profile = "checkpoint_defaults"
    evidence_token_budget = str(raw.get("evidence_token_budget", "auto")).strip().lower()
    if evidence_token_budget not in EVIDENCE_TOKEN_BUDGETS:
        evidence_token_budget = "auto"
    try:
        seed = int(raw.get("seed", 0))
    except (TypeError, ValueError, OverflowError):
        seed = 0
    seed = max(0, min(UINT64_MAX, seed))

    external = raw.get("external_evidence_json", "")
    if isinstance(external, (Mapping, list, tuple)):
        external_text = json.dumps(external, separators=(",", ":"), ensure_ascii=False)
    else:
        external_text = str(external or "").strip()

    return GroundingGuardConfig(
        mode=mode,
        retry_on_uncertain=_as_bool(raw.get("retry_on_uncertain"), True),
        sampling_profile=profile,
        seed=seed,
        save_detailed_trace=_as_bool(raw.get("save_detailed_trace"), False),
        trace_subfolder=_safe_trace_subfolder(raw.get("trace_subfolder")),
        external_evidence_json=external_text,
        evidence_token_budget=evidence_token_budget,
    )


def load_grounding_evidence_schema() -> dict[str, Any]:
    """Load the bundled strict ledger schema used in model instructions."""

    value = json.loads(GROUNDING_EVIDENCE_SCHEMA_PATH.read_text(encoding="utf-8"))
    schema_identity = value.get("properties", {}).get("schema", {}).get("const") if isinstance(value, dict) else None
    if schema_identity != GROUNDING_LEDGER_SCHEMA_ID:
        raise GroundingGuardError("bundled grounding evidence schema has an unexpected identity")
    return value


def load_external_evidence_schema() -> dict[str, Any]:
    """Load the optional typed external-evidence envelope schema."""

    value = json.loads(EXTERNAL_EVIDENCE_SCHEMA_PATH.read_text(encoding="utf-8"))
    schema_identity = (
        value.get("properties", {}).get("schema", {}).get("const")
        if isinstance(value, dict)
        else None
    )
    if schema_identity != EXTERNAL_EVIDENCE_SCHEMA_ID:
        raise GroundingGuardError("bundled external evidence schema has an unexpected identity")
    return value


def _safe_int(value: Any, default: int = 0) -> int:
    try:
        if isinstance(value, bool):
            return default
        return int(value)
    except (TypeError, ValueError, OverflowError):
        return default


def _safe_float(value: Any, default: float = 0.0) -> float:
    try:
        result = float(value)
    except (TypeError, ValueError, OverflowError):
        return default
    return result if math.isfinite(result) else default


def _manifest_roles(media_metadata: Mapping[str, Any]) -> dict[str, str]:
    roles: dict[str, str] = {}
    manifest = str(media_metadata.get("minimax_h3_reference_manifest", "") or "")
    for match in _MANIFEST_LINE_RE.finditer(manifest):
        prefix = "picture" if match.group("kind").lower() == "picture" else "video"
        roles[f"{prefix}:{int(match.group('number'))}"] = match.group("role").strip()
    return roles


def _explicit_h3_role_categories(role: Any) -> set[str] | None:
    text = str(role or "").strip()
    match = _EXPLICIT_ROLE_CATEGORIES_RE.search(text)
    if not match:
        return None
    values = {
        item.strip().lower().replace("-", "_")
        for item in match.group(1).split(",")
        if item.strip()
    }
    if not values or "other" in values or not values.issubset(FACT_CATEGORIES):
        return set()
    return values


def _strip_h3_role_annotation(role: Any) -> str:
    return re.sub(r"\s*\[\s*dg\s*:[^\]]*\]\s*", " ", str(role or "")).strip()


def _h3_role_category_contract(role: Any) -> tuple[set[str], set[str]]:
    """Translate an H3 manifest role into a host-owned category contract.

    ``[dg:identity,appearance]`` annotations are authoritative. Untagged
    manifests retain conservative vocabulary inference for compatibility.
    """

    text = str(role or "").strip()
    if not text:
        return set(), set()
    explicit = _explicit_h3_role_categories(text)
    if explicit is not None:
        return (
            (_ROLE_SHARED_FACT_CATEGORIES | explicit, explicit)
            if explicit
            else (set(), set())
        )
    allowed = set(_ROLE_SHARED_FACT_CATEGORIES)
    role_specific: set[str] = set()
    matched = False
    forbids_identity = bool(_ROLE_FORBIDS_IDENTITY_RE.search(text))
    for index, (pattern, categories) in enumerate(_ROLE_CATEGORY_GROUPS):
        if index == 0 and forbids_identity:
            continue
        if pattern.search(text):
            allowed.update(categories)
            role_specific.update(categories)
            matched = True
    return (allowed, role_specific) if matched else (set(), set())


def _claim_role_categories(claim: Any) -> set[str]:
    categories: set[str] = set()
    text = str(claim or "")
    for pattern, matched_categories in _CLAIM_CATEGORY_GROUPS:
        if pattern.search(text):
            categories.update(matched_categories)
    if _APPAREL_COMPOUND_WEARER_RE.search(text) or (
        _CONCRETE_APPAREL_NOUN_RE.search(text)
        and (
            _APPAREL_WEARER_CUE_RE.search(text)
            or _PERSON_APPAREL_CUE_RE.search(text)
            or _PROPER_NAME_POSSESSIVE_APPAREL_RE.search(text)
            or _APPAREL_TRANSFER_CUE_RE.search(text)
        )
    ):
        categories.update({"identity", "appearance"})
    return categories


def _temporal_region(ordinal: int, sample_count: int) -> str:
    if sample_count <= 1:
        return "opening"
    if ordinal == 1:
        return "opening"
    if ordinal == sample_count:
        return "closing"
    position = (ordinal - 1) / max(1, sample_count - 1)
    if position < 1.0 / 3.0:
        return "opening"
    if position > 2.0 / 3.0:
        return "closing"
    return "middle"


def _video_samples(
    positions: Sequence[int], media_metadata: Mapping[str, Any]
) -> list[dict[str, Any]]:
    indices_value = media_metadata.get("sampled_indices")
    indices = list(indices_value) if isinstance(indices_value, (list, tuple)) else []
    explicit_timecodes_value = media_metadata.get("sampled_timecodes_seconds")
    explicit_timecodes = (
        list(explicit_timecodes_value)
        if isinstance(explicit_timecodes_value, (list, tuple))
        else []
    )
    source_fps = _safe_float(media_metadata.get("source_fps"), 0.0)
    sample_fps = max(0.001, _safe_float(media_metadata.get("sample_fps"), 1.0))
    duration = max(0.0, _safe_float(media_metadata.get("duration_seconds"), 0.0))
    sample_count = len(positions)
    samples: list[dict[str, Any]] = []
    for offset, tensor_position in enumerate(positions):
        source_index = max(0, _safe_int(indices[offset], offset)) if offset < len(indices) else offset
        if offset < len(explicit_timecodes):
            timecode = max(0.0, _safe_float(explicit_timecodes[offset], 0.0))
        elif source_fps > 0:
            timecode = source_index / source_fps
        elif duration > 0 and sample_count > 1:
            timecode = duration * offset / (sample_count - 1)
        else:
            timecode = offset / sample_fps
        ordinal = offset + 1
        samples.append(
            {
                "sample_ordinal": ordinal,
                "tensor_batch_position": int(tensor_position),
                "source_frame_index": source_index,
                "timecode_seconds": round(timecode, 6),
                "temporal_region": _temporal_region(ordinal, sample_count),
            }
        )
    return samples


def _still_asset(
    asset_id: str,
    role: str,
    tensor_position: int,
) -> dict[str, Any]:
    return {
        "asset_id": asset_id,
        "kind": "image",
        "role": role or "visual_reference",
        "required_for_grounding": True,
        "tensor_batch_positions": [int(tensor_position)],
        "samples": [
            {
                "sample_ordinal": 1,
                "tensor_batch_position": int(tensor_position),
                "source_frame_index": 0,
                "timecode_seconds": 0.0,
                "temporal_region": "opening",
            }
        ],
    }


def build_asset_registry(
    media_metadata: Mapping[str, Any] | None,
    image_count: int,
) -> dict[str, Any]:
    """Build stable host-owned IDs for the batched visual inputs.

    ``image_count`` is the number of image tensors that will be handed to the
    processor. A sampled video therefore contributes one image per sampled
    frame. The function never trusts model-produced IDs.
    """

    metadata = dict(media_metadata or {})
    count = max(0, _safe_int(image_count, 0))
    source = str(metadata.get("source", "none") or "none").strip().lower()
    synthesis_mode = str(
        metadata.get("media_synthesis_mode", "video_recreation") or "video_recreation"
    ).strip().lower()
    h3_ref = str(metadata.get("minimax_h3_mode", "") or "").strip().lower() == "ref2va"
    roles = _manifest_roles(metadata)
    warnings: list[str] = []
    assets: list[dict[str, Any]] = []

    if count == 0:
        declared_visual_asset_ids = sorted(roles)
        return {
            "schema": ASSET_REGISTRY_SCHEMA_ID,
            "source": source,
            "media_synthesis_mode": synthesis_mode,
            "minimax_h3_mode": "ref2va" if h3_ref else "",
            "expected_image_count": 0,
            "assigned_image_count": 0,
            "assets": [],
            "declared_visual_asset_ids": declared_visual_asset_ids,
            "unattached_declared_visual_asset_ids": declared_visual_asset_ids,
            "unexpected_attached_visual_asset_ids": [],
            "warnings": (
                ["declared_visual_assets_are_not_attached"]
                if declared_visual_asset_ids
                else []
            ),
        }

    if h3_ref:
        reference_count = _safe_int(
            metadata.get(
                "minimax_h3_reference_image_batch_count",
                metadata.get("reference_image_count", 0),
            ),
            0,
        )
        attachment_flag_present = "reference_image_backend_attached" in metadata
        reference_attached = (
            bool(metadata.get("reference_image_backend_attached"))
            if attachment_flag_present
            else source in {"image", "image+video"}
        )
        if reference_count and not reference_attached:
            warnings.append("unattached_reference_image_excluded_from_asset_registry")
            reference_count = 0
        reference_count = max(0, min(count, reference_count))
        if reference_count == 0 and source == "image":
            reference_count = count
        for index in range(reference_count):
            asset_id = f"picture:{index + 1}"
            assets.append(
                _still_asset(
                    asset_id,
                    roles.get(asset_id, f"H3 reference picture {index + 1}"),
                    index,
                )
            )
        remaining = list(range(reference_count, count))
        if remaining:
            video_asset = {
                "asset_id": "video:1",
                "kind": "video",
                "role": roles.get(
                    "video:1",
                    str(metadata.get("video_role", "H3 reference video")),
                ),
                "required_for_grounding": True,
                "tensor_batch_positions": remaining,
                "samples": _video_samples(remaining, metadata),
            }
            assets.append(video_asset)
        declared_picture_count = sum(key.startswith("picture:") for key in roles)
        if declared_picture_count and declared_picture_count != reference_count:
            warnings.append(
                "declared_picture_count_does_not_match_attached_reference_count"
            )
    elif source in {"video", "image+video"}:
        attachment_flag_present = "reference_image_backend_attached" in metadata
        reference_attached = (
            bool(metadata.get("reference_image_backend_attached"))
            if attachment_flag_present
            else source == "image+video"
        )
        reference_count = (
            max(0, _safe_int(metadata.get("reference_image_count"), 0))
            if source == "image+video" and reference_attached
            else 0
        )
        if source == "image+video" and reference_attached and reference_count == 0:
            reference_count = 1
        if metadata.get("reference_image_count") and not reference_attached:
            warnings.append("unattached_reference_image_excluded_from_asset_registry")
        reference_count = min(count, reference_count)
        for index in range(reference_count):
            role = str(metadata.get("image_role", "visual_reference"))
            assets.append(_still_asset(f"image:{index + 1}", role, index))
        remaining = list(range(reference_count, count))
        if remaining:
            assets.append(
                {
                    "asset_id": "video:1",
                    "kind": "video",
                    "role": str(metadata.get("video_role", "primary_video_source")),
                    "required_for_grounding": True,
                    "tensor_batch_positions": remaining,
                    "samples": _video_samples(remaining, metadata),
                }
            )
        elif source == "video":
            warnings.append("video_source_has_no_sampled_frames")
    else:
        declared_count = max(0, _safe_int(metadata.get("reference_image_count"), count))
        still_count = count if declared_count == 0 else min(count, declared_count)
        if still_count != count:
            warnings.append("unassigned_tensor_positions_treated_as_still_images")
            still_count = count
        per_image_roles = metadata.get("image_roles")
        if not isinstance(per_image_roles, list):
            per_image_roles = []
        for index in range(still_count):
            role = (
                str(per_image_roles[index])
                if index < len(per_image_roles) and str(per_image_roles[index]).strip()
                else str(metadata.get("image_role", "primary_image_source"))
            )
            assets.append(_still_asset(f"image:{index + 1}", role, index))

    for asset in assets:
        raw_role = str(asset.get("role", ""))
        _allowed_categories, role_categories = _h3_role_category_contract(
            raw_role
        )
        explicit_categories = _explicit_h3_role_categories(raw_role)
        asset["role"] = _strip_h3_role_annotation(raw_role)
        asset["grounding_role_categories"] = sorted(role_categories)
        asset["grounding_role_contract_source"] = (
            "explicit"
            if explicit_categories
            else "invalid_explicit"
            if explicit_categories is not None
            else "inferred"
            if role_categories
            else "unrecognized"
        )

    assigned_positions = sorted(
        position
        for asset in assets
        for position in asset.get("tensor_batch_positions", [])
    )
    if assigned_positions != list(range(count)):
        warnings.append("asset_registry_does_not_cover_tensor_batch")

    declared_visual_asset_ids = sorted(roles) if h3_ref else []
    attached_visual_asset_ids = sorted(
        str(asset.get("asset_id"))
        for asset in assets
        if str(asset.get("asset_id", "")).startswith(("picture:", "video:"))
    )
    unattached_declared = (
        sorted(set(declared_visual_asset_ids) - set(attached_visual_asset_ids))
        if h3_ref
        else []
    )
    unexpected_attached = (
        sorted(set(attached_visual_asset_ids) - set(declared_visual_asset_ids))
        if h3_ref
        else []
    )
    if h3_ref and (unattached_declared or unexpected_attached):
        warnings.append("declared_visual_assets_do_not_match_attached_registry")

    return {
        "schema": ASSET_REGISTRY_SCHEMA_ID,
        "source": source,
        "media_synthesis_mode": synthesis_mode,
        "minimax_h3_mode": "ref2va" if h3_ref else "",
        "expected_image_count": count,
        "assigned_image_count": len(assigned_positions),
        "assets": assets,
        "declared_visual_asset_ids": declared_visual_asset_ids,
        "unattached_declared_visual_asset_ids": unattached_declared,
        "unexpected_attached_visual_asset_ids": unexpected_attached,
        "warnings": list(dict.fromkeys(warnings)),
    }


class _DuplicateJSONKeyError(ValueError):
    pass


class _NonFiniteJSONNumberError(ValueError):
    pass


class _InvalidJSONNumberError(ValueError):
    pass


def _parse_bounded_json_decimal(value: str) -> Decimal:
    """Decode a JSON number without Python's huge-int or Decimal edge crashes."""

    if len(value) > 256:
        raise _InvalidJSONNumberError("JSON number exceeds 256 characters")
    try:
        number = Decimal(value)
        if not number.is_finite():
            raise _InvalidJSONNumberError("JSON number must be finite")
        exponent = number.as_tuple().exponent
        if not isinstance(exponent, int) or abs(exponent) > 4096:
            raise _InvalidJSONNumberError("JSON number exponent is out of range")
        return number
    except _InvalidJSONNumberError:
        raise
    except (DecimalException, ValueError) as exc:
        raise _InvalidJSONNumberError("JSON number is invalid or out of range") from exc


def _reject_duplicate_json_keys(pairs: list[tuple[str, Any]]) -> dict[str, Any]:
    result: dict[str, Any] = {}
    for key, value in pairs:
        if key in result:
            raise _DuplicateJSONKeyError(f"duplicate JSON object key: {key}")
        result[key] = value
    return result


def _reject_nonfinite_json_number(value: str) -> None:
    raise _NonFiniteJSONNumberError(f"non-finite JSON number: {value}")


def extract_grounding_ledger(
    value: str | Mapping[str, Any],
    *,
    allow_combined: bool = True,
) -> dict[str, Any]:
    """Extract a ledger from exact JSON or a combined packet.

    Markdown fences, leading commentary, trailing data, and JSON salvage are
    intentionally rejected. Combined Director packets may contain the ledger
    under the exact ``grounding_ledger`` key.
    """

    if isinstance(value, Mapping):
        try:
            parsed: Any = copy.deepcopy(dict(value))
        except RecursionError as exc:
            raise GroundingLedgerError("grounding ledger exceeds the nesting limit") from exc
    elif isinstance(value, str):
        text = value.strip()
        if not text:
            raise GroundingLedgerError("grounding ledger is empty")
        decoder = json.JSONDecoder(
            object_pairs_hook=_reject_duplicate_json_keys,
            parse_float=_parse_bounded_json_decimal,
            parse_int=_parse_bounded_json_decimal,
            parse_constant=_reject_nonfinite_json_number,
        )
        try:
            parsed, end = decoder.raw_decode(text)
        except _DuplicateJSONKeyError as exc:
            raise GroundingLedgerError(f"grounding ledger is not strict JSON: {exc}") from exc
        except _NonFiniteJSONNumberError as exc:
            raise GroundingLedgerError(f"grounding ledger is not strict JSON: {exc}") from exc
        except _InvalidJSONNumberError as exc:
            raise GroundingLedgerError(f"grounding ledger has an invalid JSON number: {exc}") from exc
        except RecursionError as exc:
            raise GroundingLedgerError("grounding ledger exceeds the JSON nesting limit") from exc
        except json.JSONDecodeError as exc:
            raise GroundingLedgerError(f"grounding ledger is not strict JSON: {exc.msg}") from exc
        if text[end:].strip():
            raise GroundingLedgerError("grounding ledger contains trailing non-JSON content")
    else:
        raise GroundingLedgerError("grounding ledger must be a JSON object or JSON string")

    if not isinstance(parsed, Mapping):
        raise GroundingLedgerError("grounding ledger root must be an object")
    if parsed.get("schema") == GROUNDING_LEDGER_SCHEMA_ID:
        try:
            return copy.deepcopy(dict(parsed))
        except RecursionError as exc:
            raise GroundingLedgerError("grounding ledger exceeds the nesting limit") from exc
    nested = parsed.get("grounding_ledger")
    if allow_combined and isinstance(nested, Mapping):
        try:
            return copy.deepcopy(dict(nested))
        except RecursionError as exc:
            raise GroundingLedgerError("grounding ledger exceeds the nesting limit") from exc
    if not allow_combined:
        raise GroundingLedgerError("strict grounding evidence root must be the ledger object")
    raise GroundingLedgerError("grounding ledger object is missing")


def detect_refusal_phrases(text: Any) -> list[str]:
    """Return complete refusal-clause labels, never isolated token matches."""

    haystack = str(text or "")
    candidates: list[tuple[int, int, str]] = []
    for label, pattern in _REFUSAL_PATTERNS:
        candidates.extend((match.start(), match.end(), label) for match in pattern.finditer(haystack))
    selected: list[tuple[int, int, str]] = []
    for candidate in sorted(candidates, key=lambda item: (-(item[1] - item[0]), item[0], item[2])):
        start, end, _label = candidate
        if any(start < other_end and other_start < end for other_start, other_end, _ in selected):
            continue
        selected.append(candidate)
    return [label for _start, _end, label in sorted(selected)]


def _append_error(errors: list[str], blockers: list[str], error: str, blocker: str) -> None:
    errors.append(error)
    blockers.append(blocker)


def _strict_text_list(
    value: Any,
    path: str,
    errors: list[str],
    blockers: list[str],
) -> None:
    if not isinstance(value, list):
        _append_error(errors, blockers, f"{path}:must_be_array", "schema_invalid")
        return
    if len(value) > 256:
        _append_error(errors, blockers, f"{path}:too_many_items", "schema_invalid")
    for index, item in enumerate(value):
        if not isinstance(item, str) or not item.strip() or len(item) > 2048:
            _append_error(
                errors,
                blockers,
                f"{path}[{index}]:must_be_nonempty_string_at_most_2048_chars",
                "schema_invalid",
            )


def _decimal_number(value: Any) -> Decimal | None:
    if isinstance(value, bool) or not isinstance(value, (int, float, Decimal)):
        return None
    if isinstance(value, float) and not math.isfinite(value):
        return None
    try:
        number = value if isinstance(value, Decimal) else Decimal(str(value))
        return number if number.is_finite() else None
    except (DecimalException, ValueError):
        return None


def _json_scalar(value: Any) -> bool:
    return value is None or isinstance(value, (str, bool)) or _decimal_number(value) is not None


def _schema_integer(
    value: Any,
    minimum: int | None = None,
    maximum: int = (1 << 63) - 1,
) -> int | None:
    number = _decimal_number(value)
    try:
        if number is None or number != number.to_integral_value():
            return None
        if number > maximum or (minimum is not None and number < minimum):
            return None
        return int(number)
    except (DecimalException, ValueError, OverflowError):
        return None


def _canonical_decimal(value: Decimal) -> str:
    sign, digits_tuple, exponent = value.as_tuple()
    digits = list(digits_tuple)
    if not digits or all(digit == 0 for digit in digits):
        return "0e0"
    while len(digits) > 1 and digits[-1] == 0:
        digits.pop()
        exponent += 1
    coefficient = "".join(str(digit) for digit in digits)
    return f"{'-' if sign else ''}{coefficient}e{exponent}"


def _json_native_number(value: Any) -> int | float | None:
    number = _decimal_number(value)
    if number is None:
        return None
    try:
        if number == number.to_integral_value():
            if number.adjusted() > 1024:
                return None
            return int(number)
        converted = float(number)
    except (DecimalException, ValueError, OverflowError):
        return None
    if not math.isfinite(converted) or Decimal(str(converted)) != number:
        return None
    return converted


def _typed_claim_key(
    claim: Mapping[str, Any],
    assets: Mapping[str, Mapping[str, Any]] | None = None,
) -> tuple[str, str, int | None]:
    asset_id = str(claim.get("asset_id", ""))
    ordinal = claim.get("sample_ordinal")
    normalized_ordinal = _schema_integer(ordinal, 1)
    asset = assets.get(asset_id) if assets is not None else None
    if asset is not None and len(_sample_index(asset)) == 1:
        normalized_ordinal = None
    return (
        asset_id,
        str(claim.get("claim_type", "")),
        normalized_ordinal,
    )


def _typed_value(value: Any) -> str:
    if isinstance(value, bool):
        value_type = "boolean"
        canonical_value: Any = value
    elif _decimal_number(value) is not None:
        value_type = "number"
        canonical_value = _canonical_decimal(_decimal_number(value))
    elif value is None:
        value_type = "null"
        canonical_value = None
    else:
        value_type = "string"
        canonical_value = str(value)
    return json.dumps(
        {"type": value_type, "value": canonical_value},
        sort_keys=True,
        separators=(",", ":"),
        ensure_ascii=False,
    )


def _decimal_to_json_native(value: Any) -> Any:
    if isinstance(value, Decimal):
        converted = _json_native_number(value)
        if converted is None:
            raise GroundingGuardError(
                "decimal value cannot be represented losslessly in the validated ledger"
            )
        return converted
    if isinstance(value, list):
        return [_decimal_to_json_native(item) for item in value]
    if isinstance(value, tuple):
        return [_decimal_to_json_native(item) for item in value]
    if isinstance(value, Mapping):
        return {
            str(key): _decimal_to_json_native(item)
            for key, item in value.items()
        }
    return value


def _contains_unrepresentable_decimal(value: Any) -> bool:
    if not isinstance(value, bool) and isinstance(value, (int, float, Decimal)):
        return _json_native_number(value) is None
    if isinstance(value, Mapping):
        return any(_contains_unrepresentable_decimal(item) for item in value.values())
    if isinstance(value, (list, tuple)):
        return any(_contains_unrepresentable_decimal(item) for item in value)
    return False


def parse_external_evidence(
    value: str | Mapping[str, Any] | None,
) -> tuple[list[dict[str, Any]], list[str]]:
    """Parse recognized typed claims; unknown/unstructured inputs stay advisory."""

    if value is None or (isinstance(value, str) and not value.strip()):
        return [], []
    if isinstance(value, Mapping):
        parsed: Any = dict(value)
    elif isinstance(value, str):
        try:
            parsed = json.loads(
                value,
                object_pairs_hook=_reject_duplicate_json_keys,
                parse_float=_parse_bounded_json_decimal,
                parse_int=_parse_bounded_json_decimal,
                parse_constant=_reject_nonfinite_json_number,
            )
        except _DuplicateJSONKeyError:
            return [], ["external_evidence_duplicate_json_key"]
        except _NonFiniteJSONNumberError:
            return [], ["external_evidence_nonfinite_json_number"]
        except _InvalidJSONNumberError:
            return [], ["external_evidence_json_number_invalid_or_out_of_range"]
        except RecursionError:
            return [], ["external_evidence_json_nesting_limit"]
        except json.JSONDecodeError:
            return [], ["external_evidence_unstructured_or_invalid_json"]
    else:
        return [], ["external_evidence_unstructured"]
    if not isinstance(parsed, Mapping):
        return [], ["external_evidence_unstructured"]
    required_root = {"schema", "provider", "claims"}
    if set(parsed) != required_root:
        return [], ["external_evidence_envelope_invalid"]
    if parsed.get("schema") != EXTERNAL_EVIDENCE_SCHEMA_ID:
        return [], ["external_evidence_unknown_schema"]
    provider = parsed.get("provider")
    if not isinstance(provider, str) or provider not in KNOWN_EXTERNAL_PROVIDERS:
        return [], ["external_evidence_unknown_provider"]
    claims = parsed.get("claims")
    if not isinstance(claims, list) or len(claims) > 512:
        return [], ["external_evidence_claims_invalid"]

    valid: list[dict[str, Any]] = []
    for index, claim in enumerate(claims):
        if not isinstance(claim, Mapping):
            return [], [f"external_evidence_claim_{index}_invalid"]
        allowed = {
            "asset_id",
            "claim_type",
            "value",
            "sample_ordinal",
            "source_frame_index",
            "timecode_seconds",
            "confidence",
        }
        required = {"asset_id", "claim_type", "value"}
        if set(claim) - allowed:
            return [], [f"external_evidence_claim_{index}_has_unknown_fields"]
        if required - set(claim):
            return [], [f"external_evidence_claim_{index}_missing_required_fields"]
        asset_id = claim.get("asset_id")
        claim_type = claim.get("claim_type")
        if (
            not isinstance(asset_id, str)
            or not isinstance(claim_type, str)
            or not _ASSET_ID_RE.fullmatch(asset_id)
            or not _CLAIM_TYPE_RE.fullmatch(claim_type)
        ):
            return [], [f"external_evidence_claim_{index}_invalid_key"]
        claim_value = claim.get("value")
        if not _json_scalar(claim_value):
            return [], [f"external_evidence_claim_{index}_invalid_value"]
        if _decimal_number(claim_value) is not None and _schema_integer(
            claim_value,
            -(1 << 63),
        ) is None:
            return [], [f"external_evidence_claim_{index}_numeric_value_not_int64"]
        if asset_id.startswith("video:") and "sample_ordinal" not in claim:
            return [], [f"external_evidence_claim_{index}_video_sample_ordinal_required"]
        ordinal = claim.get("sample_ordinal")
        normalized_ordinal = _schema_integer(ordinal, 1)
        if "sample_ordinal" in claim and normalized_ordinal is None:
            return [], [f"external_evidence_claim_{index}_invalid_sample"]
        confidence = claim.get("confidence")
        if "confidence" in claim and confidence not in CONFIDENCE_LEVELS:
            return [], [f"external_evidence_claim_{index}_invalid_confidence"]
        frame = claim.get("source_frame_index")
        if "sample_ordinal" not in claim and (
            "source_frame_index" in claim or "timecode_seconds" in claim
        ):
            return [], [f"external_evidence_claim_{index}_frame_or_time_requires_sample"]
        normalized_frame = _schema_integer(frame, 0)
        if "source_frame_index" in claim and normalized_frame is None:
            return [], [f"external_evidence_claim_{index}_invalid_frame"]
        timecode = claim.get("timecode_seconds")
        if "timecode_seconds" in claim and (
            _decimal_number(timecode) is None
            or _decimal_number(timecode) < 0
        ):
            return [], [f"external_evidence_claim_{index}_invalid_timecode"]
        normalized_claim = copy.deepcopy(dict(claim))
        if "sample_ordinal" in normalized_claim:
            normalized_claim["sample_ordinal"] = normalized_ordinal
        if "source_frame_index" in normalized_claim:
            normalized_claim["source_frame_index"] = normalized_frame
        valid.append(normalized_claim)
    return valid, []


def _asset_index(registry: Mapping[str, Any]) -> dict[str, dict[str, Any]]:
    assets = registry.get("assets")
    if not isinstance(assets, list):
        return {}
    return {
        str(asset.get("asset_id")): dict(asset)
        for asset in assets
        if isinstance(asset, Mapping) and asset.get("asset_id")
    }


def _sample_index(asset: Mapping[str, Any]) -> dict[int, dict[str, Any]]:
    samples = asset.get("samples")
    if not isinstance(samples, list):
        return {}
    return {
        int(sample.get("sample_ordinal")): dict(sample)
        for sample in samples
        if isinstance(sample, Mapping)
        and isinstance(sample.get("sample_ordinal"), int)
        and not isinstance(sample.get("sample_ordinal"), bool)
    }


def _validate_evidence_reference(
    reference: Any,
    path: str,
    assets: Mapping[str, dict[str, Any]],
    errors: list[str],
    blockers: list[str],
) -> tuple[str | None, str | None]:
    if not isinstance(reference, Mapping):
        _append_error(errors, blockers, f"{path}:must_be_object", "schema_invalid")
        return None, None
    allowed = {"asset_id", "sample_ordinal", "source_frame_index", "timecode_seconds"}
    if set(reference) - allowed:
        _append_error(errors, blockers, f"{path}:unknown_fields", "schema_invalid")
    asset_id = reference.get("asset_id")
    if not isinstance(asset_id, str) or not _ASSET_ID_RE.fullmatch(asset_id):
        _append_error(errors, blockers, f"{path}.asset_id:invalid", "schema_invalid")
        return None, None
    video_missing_fields: set[str] = set()
    if asset_id.startswith("video:"):
        required = {"sample_ordinal", "source_frame_index", "timecode_seconds"}
        video_missing_fields = required - set(reference)
        if video_missing_fields:
            _append_error(
                errors,
                blockers,
                f"{path}:video_reference_missing:{','.join(sorted(video_missing_fields))}",
                "schema_invalid",
            )
            blockers.append("invalid_frame_reference")
    asset = assets.get(asset_id)
    if asset is None:
        _append_error(errors, blockers, f"{path}.asset_id:unknown_asset:{asset_id}", "unknown_asset")
        return asset_id, None

    samples = _sample_index(asset)
    ordinal = reference.get("sample_ordinal")
    if video_missing_fields:
        return asset_id, None
    if "sample_ordinal" in reference and ordinal is None:
        _append_error(errors, blockers, f"{path}.sample_ordinal:null_not_allowed", "schema_invalid")
        return asset_id, None
    if ordinal is None:
        if "source_frame_index" in reference or "timecode_seconds" in reference:
            _append_error(
                errors,
                blockers,
                f"{path}:frame_or_time_requires_sample_ordinal",
                "schema_invalid",
            )
            blockers.append("invalid_frame_reference")
        return asset_id, None
    normalized_ordinal = _schema_integer(ordinal, 1)
    if normalized_ordinal is None:
        _append_error(errors, blockers, f"{path}.sample_ordinal:invalid", "schema_invalid")
        return asset_id, None
    expected = samples.get(normalized_ordinal)
    if expected is None:
        _append_error(
            errors,
            blockers,
            f"{path}.sample_ordinal:out_of_bounds:{normalized_ordinal}",
            "invalid_frame_reference",
        )
        return asset_id, None
    if "source_frame_index" in reference:
        frame = reference.get("source_frame_index")
        normalized_frame = _schema_integer(frame, 0)
        if normalized_frame is None:
            _append_error(
                errors,
                blockers,
                f"{path}.source_frame_index:invalid",
                "schema_invalid",
            )
        elif normalized_frame != expected.get("source_frame_index"):
            _append_error(
                errors,
                blockers,
                f"{path}.source_frame_index:does_not_match_registry",
                "invalid_frame_reference",
            )
    if "timecode_seconds" in reference:
        timecode = reference.get("timecode_seconds")
        normalized_timecode = _decimal_number(timecode)
        if (
            normalized_timecode is None
            or normalized_timecode < 0
        ):
            _append_error(errors, blockers, f"{path}.timecode_seconds:invalid", "schema_invalid")
        elif abs(
            normalized_timecode
            - (Decimal(str(expected.get("timecode_seconds", 0.0))))
        ) > Decimal("0.075"):
            _append_error(
                errors,
                blockers,
                f"{path}.timecode_seconds:does_not_match_registry",
                "invalid_frame_reference",
            )
    return asset_id, str(expected.get("temporal_region", "")) or None


def _validate_typed_claim(
    claim: Any,
    path: str,
    assets: Mapping[str, dict[str, Any]],
    errors: list[str],
    blockers: list[str],
) -> dict[str, Any] | None:
    initial_error_count = len(errors)
    if not isinstance(claim, Mapping):
        _append_error(errors, blockers, f"{path}:must_be_object", "schema_invalid")
        return None
    allowed = {"asset_id", "claim_type", "value", "sample_ordinal"}
    if set(claim) - allowed:
        _append_error(errors, blockers, f"{path}:unknown_fields", "schema_invalid")
    if not {"asset_id", "claim_type", "value"}.issubset(claim):
        _append_error(errors, blockers, f"{path}:missing_required_fields", "schema_invalid")
        return None
    asset_id = claim.get("asset_id")
    claim_type = claim.get("claim_type")
    if not isinstance(asset_id, str) or not _ASSET_ID_RE.fullmatch(asset_id):
        _append_error(errors, blockers, f"{path}.asset_id:invalid", "schema_invalid")
    elif asset_id not in assets:
        _append_error(errors, blockers, f"{path}.asset_id:unknown_asset", "unknown_asset")
    if not isinstance(claim_type, str) or not _CLAIM_TYPE_RE.fullmatch(claim_type):
        _append_error(errors, blockers, f"{path}.claim_type:invalid", "schema_invalid")
    claim_value = claim.get("value")
    if not _json_scalar(claim_value):
        _append_error(errors, blockers, f"{path}.value:must_be_json_scalar", "schema_invalid")
    elif _decimal_number(claim_value) is not None and _schema_integer(
        claim_value,
        -(1 << 63),
    ) is None:
        _append_error(
            errors,
            blockers,
            f"{path}.value:numeric_value_must_be_signed_64_bit_integer",
            "schema_invalid",
        )
    ordinal = claim.get("sample_ordinal")
    if isinstance(asset_id, str) and asset_id.startswith("video:") and "sample_ordinal" not in claim:
        _append_error(
            errors,
            blockers,
            f"{path}.sample_ordinal:required_for_video_claim",
            "schema_invalid",
        )
    if "sample_ordinal" in claim and ordinal is None:
        _append_error(errors, blockers, f"{path}.sample_ordinal:null_not_allowed", "schema_invalid")
    elif ordinal is not None and _schema_integer(ordinal, 1) is None:
        _append_error(errors, blockers, f"{path}.sample_ordinal:invalid", "schema_invalid")
    elif ordinal is not None and isinstance(asset_id, str) and asset_id in assets:
        normalized_ordinal = _schema_integer(ordinal, 1)
        if normalized_ordinal not in _sample_index(assets[asset_id]):
            _append_error(
                errors,
                blockers,
                f"{path}.sample_ordinal:out_of_bounds:{normalized_ordinal}",
                "invalid_frame_reference",
            )
    if len(errors) != initial_error_count:
        return None
    return copy.deepcopy(dict(claim))


def validate_grounding_ledger(
    ledger: Mapping[str, Any],
    asset_registry: Mapping[str, Any],
    external_evidence_json: str | Mapping[str, Any] | None = None,
    *,
    final_text: str = "",
) -> GroundingLedgerValidation:
    """Strictly validate a model ledger against host-created assets."""

    candidate = copy.deepcopy(dict(ledger)) if isinstance(ledger, Mapping) else None
    errors: list[str] = []
    warnings: list[str] = []
    blockers: list[str] = []
    covered_assets: set[str] = set()
    temporal_regions: dict[str, set[str]] = {}
    typed_claims: list[dict[str, Any]] = []

    if candidate is None:
        return GroundingLedgerValidation(
            ledger=None,
            schema_valid=False,
            grounded=False,
            analysis_status="uncertain",
            errors=("ledger:must_be_object",),
            blocking_reasons=("schema_invalid",),
        )

    required_top = {
        "schema",
        "analysis_status",
        "observed_facts",
        "inferred_facts",
        "creative_additions",
        "uncertainties",
        "grounding_failure_reasons",
    }
    allowed_top = required_top
    missing_top = required_top - set(candidate)
    unknown_top = set(candidate) - allowed_top
    if missing_top:
        _append_error(
            errors,
            blockers,
            f"ledger:missing_fields:{','.join(sorted(missing_top))}",
            "schema_invalid",
        )
    if unknown_top:
        _append_error(
            errors,
            blockers,
            f"ledger:unknown_fields:{','.join(sorted(unknown_top))}",
            "schema_invalid",
        )
    if candidate.get("schema") != GROUNDING_LEDGER_SCHEMA_ID:
        _append_error(errors, blockers, "ledger.schema:unsupported", "schema_invalid")
    reported_model_status = candidate.get("analysis_status")
    model_status = reported_model_status
    if model_status not in LEDGER_ANALYSIS_STATUSES:
        _append_error(errors, blockers, "ledger.analysis_status:invalid", "schema_invalid")
        model_status = "uncertain"

    for key in (
        "inferred_facts",
        "creative_additions",
        "uncertainties",
        "grounding_failure_reasons",
    ):
        _strict_text_list(candidate.get(key), f"ledger.{key}", errors, blockers)

    assets = _asset_index(asset_registry)
    facts = candidate.get("observed_facts")
    if not isinstance(facts, list):
        _append_error(errors, blockers, "ledger.observed_facts:must_be_array", "schema_invalid")
        facts = []
    elif len(facts) > 256:
        _append_error(errors, blockers, "ledger.observed_facts:too_many_items", "schema_invalid")

    seen_fact_ids: set[str] = set()
    high_medium_by_asset: dict[str, list[tuple[set[str], set[str]]]] = {}
    categories_by_asset: dict[str, set[str]] = {}
    claims_by_asset: dict[str, list[tuple[str, set[str]]]] = {}
    for fact_index, fact in enumerate(facts):
        path = f"ledger.observed_facts[{fact_index}]"
        if not isinstance(fact, Mapping):
            _append_error(errors, blockers, f"{path}:must_be_object", "schema_invalid")
            continue
        # Validation accepts generic Mapping inputs, but canonicalization below
        # must never assume that a caller-owned/custom mapping is writable.
        fact = dict(fact)
        facts[fact_index] = fact
        required_fact = {"fact_id", "claim", "confidence", "categories", "evidence"}
        allowed_fact = required_fact | {"typed_claims"}
        if set(fact) - allowed_fact:
            _append_error(errors, blockers, f"{path}:unknown_fields", "schema_invalid")
        if required_fact - set(fact):
            _append_error(errors, blockers, f"{path}:missing_required_fields", "schema_invalid")
        fact_id = fact.get("fact_id")
        if not isinstance(fact_id, str) or not _FACT_ID_RE.fullmatch(fact_id):
            _append_error(errors, blockers, f"{path}.fact_id:invalid", "schema_invalid")
        elif fact_id in seen_fact_ids:
            _append_error(errors, blockers, f"{path}.fact_id:duplicate", "schema_invalid")
        else:
            seen_fact_ids.add(fact_id)
        claim = fact.get("claim")
        if not isinstance(claim, str) or not claim.strip() or len(claim) > 2048:
            _append_error(errors, blockers, f"{path}.claim:invalid", "schema_invalid")
        confidence = fact.get("confidence")
        if confidence not in CONFIDENCE_LEVELS:
            _append_error(errors, blockers, f"{path}.confidence:invalid", "schema_invalid")
        categories_value = fact.get("categories")
        canonical_categories, category_canonicalizations = _canonicalize_fact_categories(
            categories_value
        )
        if canonical_categories is None:
            _append_error(errors, blockers, f"{path}.categories:invalid", "schema_invalid")
            categories: set[str] = set()
        else:
            categories = set(canonical_categories)
            if category_canonicalizations:
                fact["categories"] = canonical_categories
                for source_category, target_category in category_canonicalizations:
                    warnings.append(
                        f"{path}.categories:canonicalized:"
                        f"{source_category}->{target_category}"
                    )

        evidence_value = fact.get("evidence")
        if not isinstance(evidence_value, list) or not evidence_value or len(evidence_value) > 64:
            _append_error(errors, blockers, f"{path}.evidence:invalid", "schema_invalid")
            evidence_value = []
        fact_assets: set[str] = set()
        fact_regions_by_asset: dict[str, set[str]] = {}
        fact_evidence_ordinals: dict[str, set[int | None]] = {}
        for evidence_index, evidence in enumerate(evidence_value):
            evidence_error_count = len(errors)
            asset_id, region = _validate_evidence_reference(
                evidence,
                f"{path}.evidence[{evidence_index}]",
                assets,
                errors,
                blockers,
            )
            # The host registry is authoritative. Once a sample reference has
            # matched, publish its canonical integer/frame/time values instead
            # of preserving model-authored numeric spellings or precision.
            if (
                len(errors) == evidence_error_count
                and asset_id in assets
                and isinstance(evidence, Mapping)
            ):
                normalized_ordinal = _schema_integer(evidence.get("sample_ordinal"), 1)
                expected_sample = _sample_index(assets[asset_id]).get(normalized_ordinal)
                if expected_sample is not None:
                    normalized_evidence = dict(evidence)
                    normalized_evidence["sample_ordinal"] = normalized_ordinal
                    if "source_frame_index" in normalized_evidence:
                        normalized_evidence["source_frame_index"] = expected_sample.get(
                            "source_frame_index"
                        )
                    if "timecode_seconds" in normalized_evidence:
                        normalized_evidence["timecode_seconds"] = expected_sample.get(
                            "timecode_seconds"
                        )
                    evidence_value[evidence_index] = normalized_evidence
                    evidence = normalized_evidence
            if asset_id in assets:
                fact_assets.add(asset_id)
                if isinstance(evidence, Mapping):
                    ordinal = evidence.get("sample_ordinal")
                    normalized_ordinal = _schema_integer(ordinal, 1)
                    if normalized_ordinal in _sample_index(assets[asset_id]):
                        fact_evidence_ordinals.setdefault(asset_id, set()).add(
                            normalized_ordinal
                        )
                    elif (
                        ordinal is None
                        and assets[asset_id].get("kind") != "video"
                        and "source_frame_index" not in evidence
                        and "timecode_seconds" not in evidence
                    ):
                        fact_evidence_ordinals.setdefault(asset_id, set()).add(None)
                if region:
                    fact_regions_by_asset.setdefault(asset_id, set()).add(region)
        for asset_id in fact_assets:
            categories_by_asset.setdefault(asset_id, set()).update(categories)
            claims_by_asset.setdefault(asset_id, []).append(
                (str(claim or ""), set(categories))
            )
        if isinstance(confidence, str) and confidence in ("high", "medium"):
            covered_assets.update(fact_assets)
            for asset_id in fact_assets:
                regions = fact_regions_by_asset.get(asset_id, set())
                if regions:
                    temporal_regions.setdefault(asset_id, set()).update(regions)
                high_medium_by_asset.setdefault(asset_id, []).append((categories, regions))

        fact_typed = fact.get("typed_claims", [])
        if not isinstance(fact_typed, list) or len(fact_typed) > 64:
            _append_error(errors, blockers, f"{path}.typed_claims:invalid", "schema_invalid")
        else:
            for claim_index, typed_claim in enumerate(fact_typed):
                validated = _validate_typed_claim(
                    typed_claim,
                    f"{path}.typed_claims[{claim_index}]",
                    assets,
                    errors,
                    blockers,
                )
                if validated is not None:
                    typed_asset_id = str(validated.get("asset_id", ""))
                    typed_ordinal = _typed_claim_key(validated, assets)[2]
                    bound_ordinals = fact_evidence_ordinals.get(typed_asset_id, set())
                    if typed_asset_id not in fact_assets or (
                        typed_ordinal is not None and typed_ordinal not in bound_ordinals
                    ):
                        _append_error(
                            errors,
                            blockers,
                            f"{path}.typed_claims[{claim_index}]:not_bound_to_fact_evidence",
                            "schema_invalid",
                        )
                    else:
                        typed_claims.append(validated)

    typed_values_by_key: dict[tuple[str, str, int | None], set[str]] = {}
    for typed_claim in typed_claims:
        typed_values_by_key.setdefault(_typed_claim_key(typed_claim, assets), set()).add(
            _typed_value(typed_claim.get("value"))
        )
    internally_conflicting_typed_keys = sorted(
        (key for key, values in typed_values_by_key.items() if len(values) > 1),
        key=lambda item: (
            item[0],
            item[1],
            -1 if item[2] is None else item[2],
        ),
    )
    if internally_conflicting_typed_keys:
        rendered = [
            f"{asset_id}:{claim_type}:{ordinal if ordinal is not None else 'asset'}"
            for asset_id, claim_type, ordinal in internally_conflicting_typed_keys
        ]
        _append_error(
            errors,
            blockers,
            "ledger.typed_claims:internal_conflict:" + ",".join(rendered),
            "schema_invalid",
        )

    if model_status == "grounded" and not facts:
        _append_error(errors, blockers, "ledger.observed_facts:empty", "insufficient_coverage")

    required_assets = {
        asset_id
        for asset_id, asset in assets.items()
        if asset.get("required_for_grounding", True)
    }
    missing_coverage = sorted(required_assets - covered_assets)
    if model_status == "grounded" and missing_coverage:
        _append_error(
            errors,
            blockers,
            "ledger.observed_facts:missing_high_or_medium_coverage:"
            + ",".join(missing_coverage),
            "insufficient_coverage",
        )

    for asset_id, asset in assets.items():
        if asset.get("kind") != "video" or len(asset.get("samples", [])) <= 1:
            continue
        regions = temporal_regions.get(asset_id, set())
        if model_status == "grounded" and len(regions) < 2:
            _append_error(
                errors,
                blockers,
                f"ledger.observed_facts:insufficient_temporal_regions:{asset_id}",
                "insufficient_coverage",
            )

    if str(asset_registry.get("media_synthesis_mode", "")) == "image_identity_video_control":
        identity_assets = {
            asset_id
            for asset_id, asset in assets.items()
            if asset.get("kind") == "image" and "identity" in str(asset.get("role", "")).lower()
        }
        video_assets = {
            asset_id for asset_id, asset in assets.items() if asset.get("kind") == "video"
        }
        identity_ok = any(
            categories & {"identity", "appearance"}
            for asset_id in identity_assets
            for categories, _regions in high_medium_by_asset.get(asset_id, [])
        )
        action_or_camera_ok = any(
            categories & {"action", "motion", "temporal", "camera", "composition"}
            for asset_id in video_assets
            for categories, _regions in high_medium_by_asset.get(asset_id, [])
        )
        if model_status == "grounded" and not identity_ok:
            _append_error(errors, blockers, "ledger:missing_identity_image_evidence", "insufficient_coverage")
        if model_status == "grounded" and not action_or_camera_ok:
            _append_error(
                errors,
                blockers,
                "ledger:missing_video_action_or_camera_evidence",
                "insufficient_coverage",
            )

        for asset_id in identity_assets:
            if categories_by_asset.get(asset_id, set()) & {
                "action",
                "motion",
                "temporal",
                "camera",
            }:
                _append_error(errors, blockers, f"ledger:role_cross_wired:{asset_id}", "role_cross_wired")
        for asset_id in video_assets:
            if categories_by_asset.get(asset_id, set()) & {"identity", "appearance"}:
                _append_error(errors, blockers, f"ledger:role_cross_wired:{asset_id}", "role_cross_wired")

    # H3 reference manifests assign each Picture/Video an explicit role.  The
    # role comes from the host-created registry, not from model output.  Every
    # role-bound fact must declare a role-specific category, and claim text is
    # independently screened so relabeling a swapped claim as ``other`` cannot
    # bypass the boundary.
    if str(asset_registry.get("minimax_h3_mode", "")) == "ref2va":
        declared_visual_assets = {
            str(asset_id)
            for asset_id in asset_registry.get("declared_visual_asset_ids", [])
            if isinstance(asset_id, str)
        }
        attached_visual_assets = {
            asset_id
            for asset_id in assets
            if asset_id.startswith(("picture:", "video:"))
        }
        if declared_visual_assets != attached_visual_assets:
            _append_error(
                errors,
                blockers,
                "ledger:h3_declared_attached_asset_mismatch:"
                f"missing={','.join(sorted(declared_visual_assets - attached_visual_assets)) or 'none'};"
                f"unexpected={','.join(sorted(attached_visual_assets - declared_visual_assets)) or 'none'}",
                "insufficient_coverage",
            )
        for asset_id, asset in assets.items():
            if not asset_id.startswith(("picture:", "video:")):
                continue
            if asset.get("grounding_role_contract_source") != "explicit":
                _append_error(
                    errors,
                    blockers,
                    f"ledger:h3_role_contract_not_explicit:{asset_id}",
                    "insufficient_coverage",
                )
                continue
            stored_role_categories = asset.get("grounding_role_categories")
            if isinstance(stored_role_categories, list):
                role_specific = {
                    str(category)
                    for category in stored_role_categories
                    if category in FACT_CATEGORIES and category != "other"
                }
                allowed_categories = (
                    _ROLE_SHARED_FACT_CATEGORIES
                    | _expand_role_category_family(role_specific)
                )
            else:
                allowed_categories, role_specific = _h3_role_category_contract(
                    asset.get("role")
                )
            if not role_specific:
                _append_error(
                    errors,
                    blockers,
                    f"ledger:h3_role_contract_unrecognized:{asset_id}",
                    "insufficient_coverage",
                )
                continue
            for claim_text, declared_categories in claims_by_asset.get(asset_id, []):
                semantic_categories = _claim_role_categories(claim_text)
                effective_categories = declared_categories | semantic_categories
                disallowed = effective_categories - allowed_categories
                if disallowed:
                    _append_error(
                        errors,
                        blockers,
                        f"ledger:role_cross_wired:{asset_id}:{','.join(sorted(disallowed))}",
                        "role_cross_wired",
                    )
                if not declared_categories & role_specific:
                    _append_error(
                        errors,
                        blockers,
                        f"ledger:role_specific_category_missing:{asset_id}",
                        "role_cross_wired",
                    )

    external_claims, external_warnings = parse_external_evidence(external_evidence_json)
    warnings.extend(external_warnings)
    accepted_external_claims: list[dict[str, Any]] = []
    external_by_key: dict[tuple[str, str, int | None], set[str]] = {}
    external_registry_valid = True
    for claim_index, claim in enumerate(external_claims):
        asset = assets.get(str(claim.get("asset_id", "")))
        if asset is None:
            warnings.append(f"external_evidence_unknown_asset:{claim.get('asset_id')}")
            external_registry_valid = False
            continue
        ordinal = claim.get("sample_ordinal")
        if str(claim.get("asset_id", "")).startswith("video:") and ordinal is None:
            warnings.append(
                f"external_evidence_claim_{claim_index}_video_sample_ordinal_required"
            )
            external_registry_valid = False
            continue
        if ordinal is not None:
            expected_sample = _sample_index(asset).get(int(ordinal))
            if expected_sample is None:
                warnings.append(f"external_evidence_claim_{claim_index}_sample_out_of_bounds")
                external_registry_valid = False
                continue
            if (
                "source_frame_index" in claim
                and claim.get("source_frame_index") != expected_sample.get("source_frame_index")
            ):
                warnings.append(f"external_evidence_claim_{claim_index}_frame_mismatch")
                external_registry_valid = False
                continue
            if "timecode_seconds" in claim and abs(
                _decimal_number(claim.get("timecode_seconds"))
                - Decimal(str(expected_sample.get("timecode_seconds", 0.0)))
            ) > Decimal("0.075"):
                warnings.append(f"external_evidence_claim_{claim_index}_timecode_mismatch")
                external_registry_valid = False
                continue
        elif "source_frame_index" in claim or "timecode_seconds" in claim:
            warnings.append(f"external_evidence_claim_{claim_index}_missing_sample_ordinal")
            external_registry_valid = False
            continue
        accepted_external_claims.append(claim)
        external_by_key.setdefault(_typed_claim_key(claim, assets), set()).add(
            _typed_value(claim.get("value"))
        )
    if not external_registry_valid:
        accepted_external_claims.clear()
        external_by_key.clear()
    conflicts: list[str] = []
    for key, values in external_by_key.items():
        if len(values) > 1:
            conflicts.append(
                f"{key[0]}:{key[1]}:{key[2] if key[2] is not None else 'asset'}"
            )
    for claim in typed_claims:
        key = _typed_claim_key(claim, assets)
        expected_values = external_by_key.get(key)
        if expected_values and _typed_value(claim.get("value")) not in expected_values:
            rendered_key = f"{key[0]}:{key[1]}:{key[2] if key[2] is not None else 'asset'}"
            conflicts.append(rendered_key)
    if conflicts:
        _append_error(
            errors,
            blockers,
            "ledger.typed_claims:external_conflict:" + ",".join(sorted(set(conflicts))),
            "external_typed_conflict",
        )

    failure_reasons_for_refusal = candidate.get("grounding_failure_reasons", [])
    if not isinstance(failure_reasons_for_refusal, (list, tuple)):
        failure_reasons_for_refusal = ()
    text_for_refusal = "\n".join(
        [
            final_text,
            *(str(fact.get("claim", "")) for fact in facts if isinstance(fact, Mapping)),
            *(
                str(item)
                for item in failure_reasons_for_refusal
                if isinstance(item, str)
            ),
        ]
    )
    refusal_matches = detect_refusal_phrases(text_for_refusal)
    if refusal_matches:
        blockers.append("refusal_detected")

    candidate_has_unrepresentable_decimal = _contains_unrepresentable_decimal(candidate)
    if candidate_has_unrepresentable_decimal:
        _append_error(
            errors,
            blockers,
            "ledger:numeric_value_not_losslessly_representable",
            "schema_invalid",
        )

    unique_errors = tuple(dict.fromkeys(errors))
    unique_warnings = tuple(dict.fromkeys(warnings))
    unique_blockers = tuple(dict.fromkeys(blockers))
    schema_valid = not any(reason == "schema_invalid" for reason in unique_blockers)
    if model_status == "transport_error":
        host_status = "transport_error"
    elif model_status == "refused" or refusal_matches:
        host_status = "refused"
    elif model_status == "grounded" and not unique_blockers:
        host_status = "grounded"
    else:
        host_status = "uncertain"
    candidate["analysis_status"] = host_status
    validated_candidate = (
        None
        if candidate_has_unrepresentable_decimal
        else _decimal_to_json_native(candidate)
    )

    return GroundingLedgerValidation(
        ledger=validated_candidate,
        schema_valid=schema_valid,
        grounded=host_status == "grounded",
        analysis_status=host_status,
        model_analysis_status=str(reported_model_status or "uncertain"),
        errors=unique_errors,
        warnings=unique_warnings,
        blocking_reasons=unique_blockers,
        covered_asset_ids=tuple(sorted(covered_assets)),
        temporal_regions={
            asset_id: tuple(sorted(regions))
            for asset_id, regions in sorted(temporal_regions.items())
        },
        accepted_external_claim_count=len(accepted_external_claims),
        external_conflicts=tuple(sorted(set(conflicts))),
        refusal_matches=tuple(refusal_matches),
    )


def _specific_block_reason(
    validation: GroundingLedgerValidation | None,
    *,
    transport_confirmed: bool,
    telemetry_error: str,
    telemetry_refusal: bool,
) -> tuple[str, str]:
    if telemetry_error:
        return "uncertain", "visual_grounding_telemetry_error"
    if not transport_confirmed:
        return "transport_error", "visual_transport_error"
    if telemetry_refusal or (validation and validation.analysis_status == "refused"):
        return "refused", "visual_grounding_refused"
    if validation is None or not validation.schema_valid:
        return "uncertain", "visual_grounding_schema_invalid"
    if validation.grounded:
        return "grounded", ""
    if "insufficient_coverage" in validation.blocking_reasons:
        return "uncertain", "visual_grounding_insufficient_coverage"
    if validation.analysis_status == "transport_error":
        return "transport_error", "visual_transport_error"
    return "uncertain", "visual_grounding_uncertain"


def decide_grounding_guard(
    config: GroundingGuardConfig | Mapping[str, Any] | str | None,
    validation: GroundingLedgerValidation | None,
    *,
    media_present: bool,
    transport_confirmed: bool,
    telemetry_error: str = "",
    telemetry_refusal: bool = False,
) -> GroundingGuardDecision:
    """Map evidence into the public off/audit/strict decision contract."""

    normalized = normalize_grounding_guard_config(config)
    if normalized.mode == "off":
        return GroundingGuardDecision(
            mode="off",
            analysis_status="not_run",
            decision="disabled",
            would_block=False,
        )
    if not media_present:
        return GroundingGuardDecision(
            mode=normalized.mode,
            analysis_status="not_applicable",
            decision="not_applicable",
            would_block=False,
        )

    status, specific = _specific_block_reason(
        validation,
        transport_confirmed=transport_confirmed,
        telemetry_error=str(telemetry_error or ""),
        telemetry_refusal=bool(telemetry_refusal),
    )
    would_block = bool(specific)
    blocked_reasons = (
        ("visual_grounding_unverified", specific) if specific else ()
    )
    warnings = list(validation.warnings if validation else ())
    if telemetry_error:
        warnings.append(f"grounding telemetry failed: {telemetry_error}")
    if normalized.mode == "audit":
        decision = "warn" if would_block else "pass"
    else:
        decision = "block" if would_block else "pass"
    return GroundingGuardDecision(
        mode=normalized.mode,
        analysis_status=status,
        decision=decision,
        would_block=would_block,
        blocked_reasons=blocked_reasons,
        warnings=tuple(dict.fromkeys(warnings)),
    )


def _jsonable(value: Any, *, path: str = "trace") -> Any:
    if value is None or isinstance(value, (str, bool, int)):
        return value
    if isinstance(value, float):
        if not math.isfinite(value):
            raise GroundingGuardError(f"{path} contains a non-finite float")
        return value
    if isinstance(value, Path):
        return str(value)
    if hasattr(value, "to_dict") and callable(value.to_dict):
        return _jsonable(value.to_dict(), path=path)
    if isinstance(value, Mapping):
        result: dict[str, Any] = {}
        for key, item in value.items():
            text_key = str(key)
            normalized_key = text_key.strip().lower()
            forbidden = (
                normalized_key in _FORBIDDEN_TRACE_KEYS
                or normalized_key.startswith("pixel_values")
            )
            if forbidden and not _is_safe_pixel_tensor_descriptor(
                path,
                normalized_key,
                item,
            ):
                raise GroundingGuardError(f"{path}.{text_key} may contain raw media or logits")
            result[text_key] = _jsonable(item, path=f"{path}.{text_key}")
        return result
    if isinstance(value, (list, tuple)):
        return [_jsonable(item, path=f"{path}[{index}]") for index, item in enumerate(value)]
    raise GroundingGuardError(f"{path} contains unsupported value type {type(value).__name__}")


def _json_size(value: Any) -> int:
    # The graph-facing report is emitted with indentation, sorted keys, and
    # ASCII escaping. Measure that exact representation so the public STRING
    # output cannot exceed the advertised 256 KiB even when it contains
    # non-ASCII text or deeply nested telemetry.
    return len(
        json.dumps(
            value,
            indent=2,
            sort_keys=True,
            ensure_ascii=True,
        ).encode("utf-8")
    )


def _truncate_strings(value: Any, max_chars: int) -> Any:
    if isinstance(value, str):
        if len(value) <= max_chars:
            return value
        marker = "...[truncated]"
        if max_chars <= len(marker):
            return marker[:max_chars]
        return value[: max_chars - len(marker)] + marker
    if isinstance(value, list):
        return [_truncate_strings(item, max_chars) for item in value]
    if isinstance(value, dict):
        return {key: _truncate_strings(item, max_chars) for key, item in value.items()}
    return value


def _bounded_sequence(value: Sequence[Any], max_items: int) -> list[Any]:
    """Keep deterministic endpoints when a report-only sequence must shrink."""

    items = list(value)
    limit = max(0, int(max_items))
    if len(items) <= limit:
        return items
    if limit == 0:
        return []
    if limit == 1:
        return items[:1]
    head_count = (limit + 1) // 2
    tail_count = limit - head_count
    return [*items[:head_count], *items[-tail_count:]]


def _compact_json_tree(value: Any, *, max_chars: int, max_items: int) -> Any:
    """Compact arbitrary JSON report detail without changing scalar types."""

    if isinstance(value, str):
        return _truncate_strings(value, max_chars)
    if isinstance(value, list):
        return [
            _compact_json_tree(item, max_chars=max_chars, max_items=max_items)
            for item in _bounded_sequence(value, max_items)
        ]
    if isinstance(value, dict):
        ordered_items = sorted(value.items(), key=lambda item: str(item[0]))
        selected_items = _bounded_sequence(ordered_items, max_items)
        result: dict[str, Any] = {}
        for key, item in selected_items:
            original_key = str(key)
            compact_key = _truncate_strings(original_key, max_chars)
            if compact_key in result:
                suffix_index = 1
                while True:
                    suffix = f"~{suffix_index}"
                    prefix_chars = max(0, max_chars - len(suffix))
                    collision_key = original_key[:prefix_chars] + suffix
                    if collision_key not in result:
                        compact_key = collision_key
                        break
                    suffix_index += 1
            result[compact_key] = _compact_json_tree(
                item,
                max_chars=max_chars,
                max_items=max_items,
            )
        omitted = len(ordered_items) - len(selected_items)
        if omitted:
            marker = "__dg_omitted_mapping_keys__"
            suffix = 1
            while marker in value or marker in result:
                marker = f"__dg_omitted_mapping_keys_{suffix}__"
                suffix += 1
            result[marker] = omitted
        return result
    return value


def _compact_trajectory_summary(
    trajectory: Any,
    *,
    max_chars: int,
    max_items: int,
) -> tuple[Any, dict[str, int]]:
    """Retain the first/final snapshots and summarize omitted trajectory detail."""

    if not isinstance(trajectory, dict):
        return (
            _compact_json_tree(
                trajectory,
                max_chars=max_chars,
                max_items=max_items,
            ),
            {},
        )

    result = copy.deepcopy(trajectory)
    stats: dict[str, int] = {}
    snapshots = result.get("snapshots")
    if isinstance(snapshots, list):
        selected = _bounded_sequence(snapshots, 2)
        omitted = len(snapshots) - len(selected)
        result["snapshots"] = [
            _compact_json_tree(
                item,
                max_chars=max_chars,
                max_items=max_items,
            )
            for item in selected
        ]
        if omitted:
            result["snapshots_truncated"] = omitted
            result["truncated"] = True
            stats["snapshots_omitted"] = omitted

    calls = result.get("calls")
    if isinstance(calls, list):
        selected_calls = _bounded_sequence(calls, max_items)
        omitted_calls = len(calls) - len(selected_calls)
        result["calls"] = [
            _compact_json_tree(
                item,
                max_chars=max_chars,
                max_items=max_items,
            )
            for item in selected_calls
        ]
        if omitted_calls:
            result["calls_truncated"] = omitted_calls
            result["truncated"] = True
            stats["calls_omitted"] = omitted_calls

    for key, item in list(result.items()):
        if key not in {"snapshots", "calls"}:
            result[key] = _compact_json_tree(
                item,
                max_chars=max_chars,
                max_items=max_items,
            )
    return result, stats


def _compact_validated_ledger(
    ledger: Any,
    *,
    max_chars: int,
    max_facts: int,
    max_text_items: int,
    max_evidence: int,
    max_typed_claims: int,
) -> tuple[Any, dict[str, Any]]:
    """Bound a validated ledger while retaining its typed ledger structure."""

    if not isinstance(ledger, dict):
        return ledger, {}

    result = copy.deepcopy(ledger)
    stats: dict[str, Any] = {}
    original_facts = result.get("observed_facts")
    if isinstance(original_facts, list):
        selected_facts = _bounded_sequence(original_facts, max_facts)
        compact_facts: list[Any] = []
        omitted_evidence = 0
        omitted_typed_claims = 0
        for fact in selected_facts:
            if not isinstance(fact, dict):
                compact_facts.append(
                    _compact_json_tree(
                        fact,
                        max_chars=max_chars,
                        max_items=max_evidence,
                    )
                )
                continue
            compact_fact = {
                key: _compact_json_tree(
                    item,
                    max_chars=max_chars,
                    max_items=max_evidence,
                )
                for key, item in fact.items()
                if key not in {"evidence", "typed_claims"}
            }
            evidence = fact.get("evidence")
            if isinstance(evidence, list):
                selected_evidence = _bounded_sequence(evidence, max_evidence)
                omitted_evidence += len(evidence) - len(selected_evidence)
                compact_fact["evidence"] = [
                    _compact_json_tree(
                        item,
                        max_chars=max_chars,
                        max_items=max_evidence,
                    )
                    for item in selected_evidence
                ]
            elif "evidence" in fact:
                compact_fact["evidence"] = _compact_json_tree(
                    evidence,
                    max_chars=max_chars,
                    max_items=max_evidence,
                )
            typed_claims = fact.get("typed_claims")
            if isinstance(typed_claims, list):
                selected_claims = _bounded_sequence(typed_claims, max_typed_claims)
                omitted_typed_claims += len(typed_claims) - len(selected_claims)
                compact_fact["typed_claims"] = [
                    _compact_json_tree(
                        item,
                        max_chars=max_chars,
                        max_items=max_typed_claims,
                    )
                    for item in selected_claims
                ]
            elif "typed_claims" in fact:
                compact_fact["typed_claims"] = _compact_json_tree(
                    typed_claims,
                    max_chars=max_chars,
                    max_items=max_typed_claims,
                )
            compact_facts.append(compact_fact)
        result["observed_facts"] = compact_facts
        stats["observed_facts_original"] = len(original_facts)
        stats["observed_facts_retained"] = len(compact_facts)
        if omitted_evidence:
            stats["evidence_references_omitted"] = omitted_evidence
        if omitted_typed_claims:
            stats["typed_claims_omitted"] = omitted_typed_claims

    text_list_stats: dict[str, dict[str, int]] = {}
    for key in (
        "inferred_facts",
        "creative_additions",
        "uncertainties",
        "grounding_failure_reasons",
    ):
        values = result.get(key)
        if not isinstance(values, list):
            continue
        selected_values = _bounded_sequence(values, max_text_items)
        result[key] = [
            _compact_json_tree(
                item,
                max_chars=max_chars,
                max_items=max_text_items,
            )
            for item in selected_values
        ]
        text_list_stats[key] = {
            "original": len(values),
            "retained": len(selected_values),
        }
    if text_list_stats:
        stats["text_lists"] = text_list_stats

    for key, item in list(result.items()):
        if key not in {
            "observed_facts",
            "inferred_facts",
            "creative_additions",
            "uncertainties",
            "grounding_failure_reasons",
        }:
            result[key] = _compact_json_tree(
                item,
                max_chars=max_chars,
                max_items=max_text_items,
            )
    return result, stats


_REPORT_COMPACTION_PROFILES: tuple[dict[str, int], ...] = (
    {
        "max_chars": 2048,
        "max_items": 128,
        "max_facts": 256,
        "max_text_items": 128,
        "max_evidence": 32,
        "max_typed_claims": 32,
    },
    {
        "max_chars": 1024,
        "max_items": 64,
        "max_facts": 256,
        "max_text_items": 64,
        "max_evidence": 16,
        "max_typed_claims": 16,
    },
    {
        "max_chars": 512,
        "max_items": 32,
        "max_facts": 192,
        "max_text_items": 32,
        "max_evidence": 8,
        "max_typed_claims": 8,
    },
    {
        "max_chars": 256,
        "max_items": 16,
        "max_facts": 128,
        "max_text_items": 16,
        "max_evidence": 4,
        "max_typed_claims": 4,
    },
    {
        "max_chars": 128,
        "max_items": 8,
        "max_facts": 96,
        "max_text_items": 8,
        "max_evidence": 2,
        "max_typed_claims": 2,
    },
    {
        "max_chars": 64,
        "max_items": 4,
        "max_facts": 48,
        "max_text_items": 4,
        "max_evidence": 1,
        "max_typed_claims": 1,
    },
    {
        "max_chars": 32,
        "max_items": 2,
        "max_facts": 16,
        "max_text_items": 2,
        "max_evidence": 1,
        "max_typed_claims": 1,
    },
)


def _compacted_report_candidate(
    report: dict[str, Any],
    profile: Mapping[str, int],
    *,
    original_bytes: int,
    profile_index: int,
) -> dict[str, Any]:
    max_chars = profile["max_chars"]
    max_items = profile["max_items"]
    candidate: dict[str, Any] = {}
    trajectory_stats: dict[str, int] = {}
    ledger_stats: dict[str, Any] = {}
    for key, value in report.items():
        if key == "trajectory_summary":
            candidate[key], trajectory_stats = _compact_trajectory_summary(
                value,
                max_chars=max_chars,
                max_items=max_items,
            )
        elif key == "validated_ledger":
            candidate[key], ledger_stats = _compact_validated_ledger(
                value,
                max_chars=max_chars,
                max_facts=profile["max_facts"],
                max_text_items=profile["max_text_items"],
                max_evidence=profile["max_evidence"],
                max_typed_claims=profile["max_typed_claims"],
            )
        else:
            candidate[key] = _compact_json_tree(
                value,
                max_chars=max_chars,
                max_items=max_items,
            )
    candidate["report_truncated"] = True
    candidate["compaction"] = {
        "schema": "dg-grounding-report-compaction/1",
        "original_bytes": original_bytes,
        "profile": profile_index,
        "trajectory": trajectory_stats,
        "validated_ledger": ledger_stats,
    }
    return candidate


def _ledger_structure_only(ledger: Any) -> Any:
    """Last-resort ledger shape used only for unusually tiny custom limits."""

    if not isinstance(ledger, dict):
        return ledger
    return {
        "schema": _truncate_strings(
            str(ledger.get("schema", GROUNDING_LEDGER_SCHEMA_ID)), 32
        ),
        "analysis_status": _truncate_strings(
            str(ledger.get("analysis_status", "uncertain")), 16
        ),
        "observed_facts": [],
        "inferred_facts": [],
        "creative_additions": [],
        "uncertainties": [],
        "grounding_failure_reasons": [],
    }


def _structural_compact_report(
    report: dict[str, Any],
    *,
    original_bytes: int,
) -> dict[str, Any]:
    """Retain every public report section even after content is exhausted."""

    result = {
        "schema": report.get("schema", GROUNDING_REPORT_SCHEMA_ID),
        "config": _compact_json_tree(
            report.get("config", {}), max_chars=16, max_items=1
        ),
        "analysis_status": report.get("analysis_status", "uncertain"),
        "decision": report.get("decision", "warn"),
        "would_block": bool(report.get("would_block", True)),
        "grounding_guard_would_block": bool(
            report.get(
                "grounding_guard_would_block",
                report.get("would_block", True),
            )
        ),
        "blocked_reasons": _compact_json_tree(
            report.get("blocked_reasons", []), max_chars=16, max_items=1
        ),
        "asset_registry": _compact_json_tree(
            report.get("asset_registry", {}), max_chars=16, max_items=1
        ),
        "transport": _compact_json_tree(
            report.get("transport", {}), max_chars=16, max_items=1
        ),
        "validated_ledger": _ledger_structure_only(
            report.get("validated_ledger")
        ),
        "validation": _compact_json_tree(
            report.get("validation", {}), max_chars=16, max_items=1
        ),
        "attempt_count": report.get("attempt_count", 0),
        "retry_reasons": _compact_json_tree(
            report.get("retry_reasons", []), max_chars=16, max_items=1
        ),
        "trajectory_summary": {"truncated": True},
        "effective_sampling": _compact_json_tree(
            report.get("effective_sampling", {}), max_chars=16, max_items=1
        ),
        "timings": _compact_json_tree(
            report.get("timings", {}), max_chars=16, max_items=1
        ),
        "forward_call_count": report.get("forward_call_count", 0),
        "warnings": _compact_json_tree(
            report.get("warnings", []), max_chars=16, max_items=1
        ),
        "verification_level": _truncate_strings(
            str(report.get("verification_level", "unknown")), 16
        ),
        "report_truncated": True,
        "compaction": {
            "schema": "dg-grounding-report-compaction/1",
            "original_bytes": original_bytes,
            "profile": "structural",
        },
    }
    if "transport_proof" in report:
        result["transport_proof"] = _compact_json_tree(
            report["transport_proof"], max_chars=16, max_items=1
        )
    return result


def compact_grounding_report(
    report: Mapping[str, Any],
    max_bytes: int = MAX_COMPACT_REPORT_BYTES,
) -> dict[str, Any]:
    """Bound metadata without dropping the public grounding report sections."""

    if max_bytes < 2048:
        raise GroundingGuardError("compact report limit must be at least 2048 bytes")
    compact = _jsonable(report, path="report")
    original_bytes = _json_size(compact)
    if original_bytes <= max_bytes:
        return compact

    # Telemetry is diagnostic detail rather than grounding evidence. Discard
    # intermediate snapshots first, while retaining the first/final views.
    trajectory = compact.get("trajectory_summary")
    if isinstance(trajectory, dict):
        snapshots = trajectory.get("snapshots")
        if isinstance(snapshots, list) and len(snapshots) > 2:
            trajectory["snapshots"] = [snapshots[0], snapshots[-1]]
            trajectory["snapshots_truncated"] = len(snapshots) - 2
            trajectory["truncated"] = True
    compact["report_truncated"] = True
    if _json_size(compact) <= max_bytes:
        return compact

    # Exhaust trajectory-only compaction before touching evidence. This keeps
    # a small validated ledger byte-for-byte intact when telemetry alone is
    # responsible for the overflow.
    trajectory_profiles = (
        (4096, 64),
        (2048, 32),
        (1024, 16),
        (512, 8),
        (256, 4),
        (128, 2),
        (64, 1),
    )
    if "trajectory_summary" in compact:
        original_trajectory = compact.get("trajectory_summary")
        trajectory_probe = dict(compact)
        trajectory_probe["trajectory_summary"] = {}
        trajectory_can_resolve_overflow = _json_size(trajectory_probe) <= max_bytes
        last_trajectory = original_trajectory
        profiles = list(enumerate(trajectory_profiles, start=1))
        if not trajectory_can_resolve_overflow:
            profiles = profiles[-1:]
        for detail_index, (max_chars, max_items) in profiles:
            candidate = dict(compact)
            compacted_trajectory, trajectory_stats = _compact_trajectory_summary(
                original_trajectory,
                max_chars=max_chars,
                max_items=max_items,
            )
            candidate["trajectory_summary"] = compacted_trajectory
            candidate["compaction"] = {
                "schema": "dg-grounding-report-compaction/1",
                "original_bytes": original_bytes,
                "profile": f"trajectory-{detail_index}",
                "trajectory": trajectory_stats,
            }
            if _json_size(candidate) <= max_bytes:
                return candidate
            last_trajectory = compacted_trajectory
        compact["trajectory_summary"] = last_trajectory

    # Validation diagnostics duplicate host conclusions already represented
    # by analysis_status/decision. Bound that detail before reducing the
    # validated evidence ledger itself.
    if "validation" in compact:
        original_validation = compact.get("validation")
        validation_probe = dict(compact)
        validation_probe["validation"] = {}
        validation_can_resolve_overflow = _json_size(validation_probe) <= max_bytes
        last_validation = original_validation
        profiles = list(enumerate(trajectory_profiles, start=1))
        if not validation_can_resolve_overflow:
            profiles = profiles[-1:]
        for detail_index, (max_chars, max_items) in profiles:
            candidate = dict(compact)
            compacted_validation = _compact_json_tree(
                original_validation,
                max_chars=max_chars,
                max_items=max_items,
            )
            candidate["validation"] = compacted_validation
            candidate["compaction"] = {
                "schema": "dg-grounding-report-compaction/1",
                "original_bytes": original_bytes,
                "profile": f"validation-{detail_index}",
            }
            if _json_size(candidate) <= max_bytes:
                return candidate
            last_validation = compacted_validation
        compact["validation"] = last_validation

    # A schema-valid ledger can still be several megabytes (and typed string
    # values are intentionally open-ended). Compact it in deterministic,
    # progressively tighter profiles rather than replacing the report with a
    # small status-only object.
    for profile_index, profile in enumerate(_REPORT_COMPACTION_PROFILES, start=1):
        candidate = _compacted_report_candidate(
            compact,
            profile,
            original_bytes=original_bytes,
            profile_index=profile_index,
        )
        if _json_size(candidate) <= max_bytes:
            return candidate

    structural = _structural_compact_report(
        compact,
        original_bytes=original_bytes,
    )
    if _json_size(structural) > max_bytes:
        raise GroundingGuardError("critical grounding report fields exceed size limit")
    return structural


def build_grounding_report(
    config: GroundingGuardConfig | Mapping[str, Any] | str | None,
    asset_registry: Mapping[str, Any],
    decision: GroundingGuardDecision,
    validation: GroundingLedgerValidation | None = None,
    *,
    transport: Mapping[str, Any] | None = None,
    trajectory_summary: Mapping[str, Any] | None = None,
    attempt_count: int = 0,
    retry_reasons: Sequence[str] = (),
    timings: Mapping[str, Any] | None = None,
    forward_call_count: int = 0,
    warnings: Sequence[str] = (),
    verification_level: str = "transport+structured_self_report",
    effective_sampling: Mapping[str, Any] | None = None,
    max_bytes: int = MAX_COMPACT_REPORT_BYTES,
) -> dict[str, Any]:
    """Build the compact graph-facing report; detailed raw text stays in traces."""

    normalized = normalize_grounding_guard_config(config)
    combined_warnings = [*decision.warnings, *warnings]
    report = {
        "schema": GROUNDING_REPORT_SCHEMA_ID,
        "config": normalized.to_dict(),
        "analysis_status": decision.analysis_status,
        "decision": decision.decision,
        "would_block": decision.would_block,
        "grounding_guard_would_block": decision.would_block,
        "blocked_reasons": list(decision.blocked_reasons),
        "asset_registry": copy.deepcopy(dict(asset_registry)),
        "transport": copy.deepcopy(dict(transport or {})),
        "validated_ledger": copy.deepcopy(validation.ledger) if validation else None,
        "validation": validation.to_dict() if validation else None,
        "attempt_count": max(0, _safe_int(attempt_count, 0)),
        "retry_reasons": [str(reason) for reason in retry_reasons],
        "trajectory_summary": copy.deepcopy(dict(trajectory_summary or {})),
        "effective_sampling": copy.deepcopy(
            dict(effective_sampling)
            if effective_sampling is not None
            else normalized.sampling_settings
        ),
        "timings": copy.deepcopy(dict(timings or {})),
        "forward_call_count": max(0, _safe_int(forward_call_count, 0)),
        "warnings": list(dict.fromkeys(str(item) for item in combined_warnings if str(item))),
        "verification_level": str(verification_level),
    }
    return compact_grounding_report(report, max_bytes=max_bytes)


def grounding_report_json(report: Mapping[str, Any]) -> str:
    return json.dumps(report, indent=2, sort_keys=True, ensure_ascii=True)


def persist_grounding_trace(
    trace: Mapping[str, Any],
    output_root: str | Path,
    trace_subfolder: str = DEFAULT_TRACE_SUBFOLDER,
    *,
    filename: str | None = None,
    max_bytes: int = MAX_TRACE_BYTES,
) -> Path:
    """Persist a JSON-only trace beneath an explicitly supplied Comfy output root."""

    root = Path(output_root).expanduser().resolve()
    safe_subfolder = _safe_trace_subfolder(trace_subfolder)
    if safe_subfolder != str(trace_subfolder or "").strip().replace("\\", "/"):
        if str(trace_subfolder or "").strip() not in {"", DEFAULT_TRACE_SUBFOLDER}:
            raise GroundingGuardError("trace_subfolder must be a relative path without traversal")
    target_dir = (root / safe_subfolder).resolve()
    try:
        target_dir.relative_to(root)
    except ValueError as exc:
        raise GroundingGuardError("trace path escapes the Comfy output root") from exc

    if filename is None:
        timestamp = datetime.now(timezone.utc).strftime("%Y%m%dT%H%M%S.%fZ")
        chosen_name = f"grounding-{timestamp}-{uuid.uuid4().hex[:8]}.json"
    else:
        chosen_name = str(filename).strip()
        if not chosen_name or Path(chosen_name).name != chosen_name:
            raise GroundingGuardError("trace filename must be a plain filename")
        if not chosen_name.lower().endswith(".json"):
            chosen_name += ".json"

    payload = _jsonable(trace, path="trace")
    encoded = json.dumps(payload, indent=2, sort_keys=True, ensure_ascii=True).encode("utf-8")
    if len(encoded) > max_bytes:
        raise GroundingGuardError(
            f"grounding trace is {len(encoded)} bytes, above the {max_bytes}-byte limit"
        )
    target_dir.mkdir(parents=True, exist_ok=True)
    path = target_dir / chosen_name
    with path.open("x", encoding="utf-8", newline="\n") as handle:
        handle.write(encoded.decode("utf-8"))
        handle.write("\n")
    return path


__all__ = [
    "ANALYSIS_STATUSES",
    "ASSET_REGISTRY_SCHEMA_ID",
    "EXTERNAL_EVIDENCE_SCHEMA_ID",
    "EXTERNAL_EVIDENCE_SCHEMA_PATH",
    "EVIDENCE_TOKEN_BUDGETS",
    "FACT_CATEGORIES",
    "GROUNDING_CONFIG_TYPE",
    "GROUNDING_EVIDENCE_SCHEMA_PATH",
    "GROUNDING_LEDGER_SCHEMA_ID",
    "GROUNDING_MODES",
    "GROUNDING_REPORT_SCHEMA_ID",
    "GUARD_DECISIONS",
    "GroundingGuardConfig",
    "GroundingGuardDecision",
    "GroundingGuardError",
    "GroundingLedgerError",
    "GroundingLedgerValidation",
    "SAMPLING_PROFILE_SETTINGS",
    "SAMPLING_PROFILES",
    "build_asset_registry",
    "build_grounding_report",
    "compact_grounding_report",
    "decide_grounding_guard",
    "detect_refusal_phrases",
    "extract_grounding_ledger",
    "grounding_report_json",
    "load_grounding_evidence_schema",
    "load_external_evidence_schema",
    "normalize_grounding_guard_config",
    "parse_external_evidence",
    "persist_grounding_trace",
    "validate_grounding_ledger",
]