"""Validated, local presentation blocks. Component props come from the frontend catalog.""" from __future__ import annotations import json import math import re from datetime import datetime from pathlib import Path from typing import Annotated, Any, Literal from jsonschema import Draft7Validator from pydantic import BaseModel, ConfigDict, Field, field_validator, model_serializer, model_validator CHART_SAMPLE_TIME_PATTERN = ( r"^(?:[1-9][0-9]{3}|0[1-9][0-9]{2}|00[1-9][0-9]|000[1-9])-[0-9]{2}-[0-9]{2}T(?:[01][0-9]|2[0-3]):[0-5][0-9]:[0-5][0-9]" r"(?:\.[0-9]+)?(?:Z|[+-](?:[01][0-9]|2[0-3]):[0-5][0-9])$" ) def validate_chart_sample_time(value: str) -> None: """Match the bounded frontend RFC3339 spelling and an actual calendar date.""" if len(value) > 64 or re.fullmatch(CHART_SAMPLE_TIME_PATTERN, value) is None: raise ValueError("Chart sample times require RFC3339 seconds and a timezone") if datetime.fromisoformat(value).tzinfo is None: raise ValueError("Chart sample times require a timezone") _catalog = json.loads( (Path(__file__).resolve().parents[3] / "contracts/answer-ui-catalog.json").read_text() ) _validators = {name: Draft7Validator(schema) for name, schema in _catalog["props"].items()} Key = Annotated[str, Field(pattern=r"^[a-zA-Z0-9_-]{1,80}$")] class UiModel(BaseModel): model_config = ConfigDict(extra="forbid") class AnswerUiElement(UiModel): type: str props: dict[str, Any] children: list[Key] = Field(max_length=48) class AnswerUiSpec(UiModel): root: Key elements: dict[Key, AnswerUiElement] class AnswerUiBlock(UiModel): version: Literal["answer-ui@1"] block_id: Key phase: Literal["process", "final"] | None = None fallback_text: Annotated[str, Field(min_length=1, max_length=4000)] spec: AnswerUiSpec @field_validator("phase", mode="before") @classmethod def validate_explicit_phase(cls, value): if value is None: raise ValueError("UI phase must be process or final when supplied") return value @model_serializer(mode="wrap") def serialize_optional_phase(self, handler): data = handler(self) if self.phase is None: data.pop("phase", None) return data @model_validator(mode="after") def validate_components(self): spec = self.spec if not 1 <= len(spec.elements) <= 48: raise ValueError("UI requires 1 to 48 elements") # Match the existing JavaScript serialized character limit, including UTF-16. serialized = json.dumps(self.model_dump(), ensure_ascii=False, separators=(",", ":")) if len(serialized.encode("utf-16-le")) // 2 > 100000: raise ValueError("UI block is too large") for element in spec.elements.values(): validator = _validators.get(element.type) if validator is None or not validator.is_valid(element.props): raise ValueError("Unknown UI component or invalid properties") if element.children and element.type not in {"Stack", "Section"}: raise ValueError("Leaf components cannot contain children") props = element.props if element.type == "Chart": if any(len(series["data"]) != len(props["x_data"]) for series in props["series"]): raise ValueError("Chart series must match the axis") if any(value is not None and not math.isfinite(value) for series in props["series"] for value in series["data"]): raise ValueError("Chart values must be finite") for series in props["series"]: times = series.get("sample_times") if times is not None: if len(times) != len(series["data"]): raise ValueError("Chart sample times must match the axis") for value, at in zip(series["data"], times, strict=True): if at is not None: if value is None: raise ValueError("Chart sample times require an observed value") validate_chart_sample_time(at) if element.type == "Table" and any(len(row) != len(props["columns"]) for row in props["rows"]): raise ValueError("Table row width must match its columns") if element.type == "FollowUpForm": names = [field["name"] for field in props["fields"]] if len(names) != len(set(names)): raise ValueError("Form field names must be unique") visited: set[str] = set() def visit(key: str, depth: int): if depth > 12 or key in visited or key not in spec.elements: raise ValueError("UI must be a bounded tree without duplicate or missing references") visited.add(key) for child in spec.elements[key].children: visit(child, depth + 1) visit(spec.root, 0) if len(visited) != len(spec.elements): raise ValueError("UI contains unreachable elements") return self class ReportUiBlock(UiModel): block: AnswerUiBlock evidence_refs: list[Annotated[str, Field(min_length=1, max_length=128)]] = Field(min_length=1, max_length=100) @model_validator(mode="after") def validate_report_phase(self): if self.block.phase == "process": raise ValueError("Report UI blocks must contain final results") return self def validate_report_ui_blocks(blocks: list[ReportUiBlock], evidence_refs: list[str]) -> None: ids = [item.block.block_id for item in blocks] if len(ids) != len(set(ids)): raise ValueError("Report UI block ids must be unique") serialized = json.dumps([item.model_dump() for item in blocks], ensure_ascii=False, separators=(",", ":")) if len(serialized.encode("utf-8")) > 400000: raise ValueError("Report UI exceeds 400 KB") for item in blocks: if not set(item.evidence_refs) <= set(evidence_refs): raise ValueError("UI evidence must be included in report evidence") for element in item.block.spec.elements.values(): if element.type == "Evidence" and element.props.get("evidence_id") not in item.evidence_refs: raise ValueError("UI evidence component must reference its block evidence")