Artur Mukhamadiev
e65fbf4b67
FastAPI app, upstream Confluence/model adapters, authoritative history, rootless container lifecycle, artifact storage and downloads, fake peers under backend/dev, tests under tests/backend. Root pytest.ini deselects the live marker by default; requirements gain the backend dependencies.
51 lines
2.8 KiB
Python
51 lines
2.8 KiB
Python
"""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")
|