"""Validate the neutral model boundary before calling provider adapters.""" from backend.errors import InvalidInputError def object_fields(value, required, optional=()): if not isinstance(value, dict) or not set(required) <= set(value) or set(value) - set(required) - set(optional): raise InvalidInputError("Invalid model boundary fields") def validate_content(content, user=False): if not isinstance(content, list): raise InvalidInputError("Model content must be an array") for item in content: if not isinstance(item, dict): raise InvalidInputError("Invalid model content") if item.get("type") == "text": object_fields(item, ("type", "text")) if not isinstance(item["text"], str): raise InvalidInputError("Model text must be a string") elif not user and item.get("type") == "tool_call": object_fields(item, ("type", "id", "name", "arguments")) if not all(isinstance(item[k], str) and item[k] for k in ("id", "name")) or not isinstance(item["arguments"], dict): raise InvalidInputError("Invalid model tool call") else: raise InvalidInputError("Unsupported model content type") def validate_model_request(payload): object_fields(payload, ("messages", "tools")) if not isinstance(payload["messages"], list) or not isinstance(payload["tools"], list): raise InvalidInputError("Model messages and tools must be arrays") for message in payload["messages"]: if not isinstance(message, dict): raise InvalidInputError("Invalid model message") role = message.get("role") if role in ("user", "assistant"): object_fields(message, ("role", "content"), ("provider_state",) if role == "assistant" else ()) validate_content(message["content"], user=role == "user") if "provider_state" in message and (not isinstance(message["provider_state"], str) or not message["provider_state"]): raise InvalidInputError("Invalid provider_state handle") elif role == "tool": object_fields(message, ("role", "tool_call_id", "name", "content", "is_error")) if not all(isinstance(message[k], str) for k in ("tool_call_id", "name", "content")) or type(message["is_error"]) is not bool: raise InvalidInputError("Invalid model tool result") else: raise InvalidInputError("Unsupported model message role") for tool in payload["tools"]: object_fields(tool, ("name", "description", "input_schema")) if not isinstance(tool["name"], str) or not tool["name"] or not isinstance(tool["description"], str) or not isinstance(tool["input_schema"], dict): raise InvalidInputError("Invalid model tool declaration")