"""Base classes for UI components served by navi_ui.""" from __future__ import annotations from typing import Any from pydantic import BaseModel, ValidationError def _resolve_ref(ref: str, defs: dict[str, Any]) -> dict[str, Any]: """Resolve a JSON Schema $ref against the $defs table.""" key = ref.split("/")[-1] return defs.get(key, {}) def _type_label(schema: dict[str, Any], defs: dict[str, Any]) -> str: """Return a compact type label for a property schema.""" if "$ref" in schema: resolved = _resolve_ref(schema["$ref"], defs) return resolved.get("title") or schema["$ref"].split("/")[-1] if "anyOf" in schema: parts = [] for part in schema["anyOf"]: if part.get("type") == "null": parts.append("null") else: parts.append(_type_label(part, defs)) return " | ".join(sorted(set(parts), key=lambda x: (x != "null", x))) if "allOf" in schema and len(schema["allOf"]) == 1: return _type_label(schema["allOf"][0], defs) typ = schema.get("type") if typ == "array": item_type = _type_label(schema.get("items", {}), defs) return f"array of {item_type}" if typ == "object": return "object" if typ: return str(typ) return "any" def _describe_property( name: str, schema: dict[str, Any], defs: dict[str, Any], required: bool, indent: int = 0, ) -> list[str]: """Return markdown lines describing one property.""" lines: list[str] = [] prefix = " " * indent if "$ref" in schema: schema = _resolve_ref(schema["$ref"], defs) if "anyOf" in schema: # Filter out the null branch so we can describe the real type. real_branches = [b for b in schema["anyOf"] if b.get("type") != "null"] if real_branches: schema = real_branches[0] else: schema = {"type": "null"} if "allOf" in schema and len(schema["allOf"]) == 1: schema = schema["allOf"][0] typ = _type_label(schema, defs) marker = "required" if required else "optional" description = schema.get("description", "") constraints: list[str] = [] if "minLength" in schema: constraints.append(f"min {schema['minLength']} chars") if "maxLength" in schema: constraints.append(f"max {schema['maxLength']} chars") if "minItems" in schema: constraints.append(f"min {schema['minItems']} items") if "maxItems" in schema: constraints.append(f"max {schema['maxItems']} items") if "minimum" in schema: constraints.append(f"≥ {schema['minimum']}") if "maximum" in schema: constraints.append(f"≤ {schema['maximum']}") if "pattern" in schema: constraints.append(f"matches `{schema['pattern']}`") if "enum" in schema: constraints.append(f"one of {schema['enum']}") meta = ", ".join(constraints) head = f"{prefix}- `{name}`: `{typ}` ({marker})" if description: head += f" — {description}" if meta: head += f" *[{meta}]*" lines.append(head) # Recurse into inline object properties. if schema.get("type") == "object" and "properties" in schema: sub_required = set(schema.get("required", [])) for sub_name, sub_schema in schema["properties"].items(): lines.extend( _describe_property(sub_name, sub_schema, defs, sub_name in sub_required, indent + 1) ) # Recurse into array item object properties. if schema.get("type") == "array" and "properties" in schema.get("items", {}): items = schema["items"] sub_required = set(items.get("required", [])) for sub_name, sub_schema in items["properties"].items(): lines.extend( _describe_property(sub_name, sub_schema, defs, sub_name in sub_required, indent + 1) ) return lines def _compact_schema_markdown(schema: dict[str, Any]) -> str: """Render a JSON Schema as compact LLM-friendly markdown.""" defs = schema.get("$defs", {}) required = set(schema.get("required", [])) properties = schema.get("properties", {}) if not properties: return "Payload accepts any JSON object." lines: list[str] = [] for name, prop_schema in properties.items(): lines.extend(_describe_property(name, prop_schema, defs, name in required)) # Append inline definitions only when they are actually referenced. referenced_defs: set[str] = set() def _collect_refs(s: dict[str, Any]) -> None: if "$ref" in s: referenced_defs.add(s["$ref"].split("/")[-1]) for v in s.values(): if isinstance(v, dict): _collect_refs(v) elif isinstance(v, list): for item in v: if isinstance(item, dict): _collect_refs(item) _collect_refs(schema) for def_name in sorted(referenced_defs): if def_name not in defs: continue def_schema = defs[def_name] if "properties" not in def_schema: continue lines.append(f"\n**{def_name}**:") sub_required = set(def_schema.get("required", [])) for sub_name, sub_schema in def_schema["properties"].items(): lines.extend( _describe_property(sub_name, sub_schema, defs, sub_name in sub_required) ) return "\n".join(lines) class UIComponent: """Base class for a renderable UI component. Subclasses must define: - ``name`` (str): snake_case identifier used in ``render_component``. - ``description`` (str): Short human-readable description. - ``schema`` (type[BaseModel]): Pydantic model describing the payload. """ name: str = "" description: str = "" schema: type[BaseModel] | None = None @classmethod def validate( cls, payload: dict[str, Any] ) -> tuple[bool, str, dict[str, Any] | None]: """Validate *payload* against the component schema. Returns ``(ok, error_message, validated_payload)``. If validation succeeds, *validated_payload* contains the normalized dict. """ if cls.schema is None: return True, "", payload try: validated = cls.schema.model_validate(payload) except ValidationError as exc: errors = [] for err in exc.errors(): loc = ".".join(str(part) for part in err["loc"]) errors.append(f" - {loc}: {err['msg']} (got {err['input']!r})") return ( False, f"Invalid payload for component {cls.name!r}.\n" + "\n".join(errors), None, ) return True, "", validated.model_dump(mode="json") @classmethod def instructions(cls) -> str: """Return a markdown schema description for the LLM.""" if cls.schema is not None: schema_body = _compact_schema_markdown(cls.schema.model_json_schema()) else: schema_body = "Payload accepts any JSON object." return ( f"### {cls.name}\n\n{cls.description}\n\n" f"Payload schema:\n\n{schema_body}\n" )