|
# retoor <retoor@molodetz.nl>
|
|
|
|
from __future__ import annotations
|
|
|
|
import re
|
|
from typing import Any, Literal
|
|
|
|
from pydantic import BaseModel, Field, field_validator, model_validator
|
|
|
|
from .capabilities import DEFAULT_LIMITS
|
|
|
|
WIDGET_TYPES = (
|
|
"confirm",
|
|
"choice",
|
|
"choice_multi",
|
|
"text",
|
|
"number",
|
|
"date",
|
|
"select",
|
|
"//",
|
|
"group",
|
|
)
|
|
|
|
VALUE_RE = re.compile(r"^[a-z0-9_][a-z0-9_-]{0,63}$")
|
|
NAME_RE = re.compile(r"^[a-zA-Z_][a-zA-Z0-9_]{0,63}$")
|
|
ID_RE = re.compile(r"^[a-z][a-z0-9-]{0,63}$")
|
|
CONTROL_CHARS = re.compile(r"[\x00-\x08\x0b\x0c\x0e-\x1f\x7f]")
|
|
|
|
|
|
def sanitize_text(value: Any, max_len: int) -> str:
|
|
text = CONTROL_CHARS.sub("", str(value if value is not None else ""))
|
|
text = " ".join(text.split())
|
|
if len(text) > max_len:
|
|
text = text[:max_len]
|
|
return text
|
|
|
|
|
|
class OptionModel(BaseModel):
|
|
value: str
|
|
label: str
|
|
description: str = ""
|
|
disabled: bool = False
|
|
|
|
@field_validator("value")
|
|
@classmethod
|
|
def _value(cls, v: str) -> str:
|
|
text = sanitize_text(v, 64).lower().replace(" ", "_")
|
|
if not VALUE_RE.match(text):
|
|
raise ValueError(
|
|
"option value must match ^[a-z0-9_][a-z0-9_-]{0,63}$"
|
|
)
|
|
return text
|
|
|
|
@field_validator("label")
|
|
@classmethod
|
|
def _label(cls, v: str) -> str:
|
|
text = sanitize_text(v, 80)
|
|
if not text:
|
|
raise ValueError("option label is required")
|
|
return text
|
|
|
|
@field_validator("description")
|
|
@classmethod
|
|
def _description(cls, v: str) -> str:
|
|
return sanitize_text(v, 120)
|
|
|
|
|
|
class WidgetModel(BaseModel):
|
|
type: Literal[
|
|
"confirm",
|
|
"choice",
|
|
"choice_multi",
|
|
"text",
|
|
"number",
|
|
"date",
|
|
"select",
|
|
"//",
|
|
"group",
|
|
]
|
|
name: str = ""
|
|
label: str = ""
|
|
help: str = ""
|
|
text: str = ""
|
|
required: bool = True
|
|
default: Any = None
|
|
display: Literal["auto", "radio", "select", "buttons"] = "auto"
|
|
options: list[OptionModel] = Field(default_factory=list)
|
|
widgets: list[WidgetModel] = Field(default_factory=list)
|
|
max_length: int | None = None
|
|
placeholder: str = ""
|
|
pattern: str = ""
|
|
min: float | None = None
|
|
max: float | None = None
|
|
step: float | None = None
|
|
min_selected: int | None = None
|
|
max_selected: int | None = None
|
|
|
|
@field_validator("name")
|
|
@classmethod
|
|
def _name(cls, v: str) -> str:
|
|
text = sanitize_text(v, 64)
|
|
return text
|
|
|
|
@field_validator("label", "help", "text", "placeholder", "pattern")
|
|
@classmethod
|
|
def _strings(cls, v: str) -> str:
|
|
return sanitize_text(v, 500)
|
|
|
|
@model_validator(mode="after")
|
|
def _shape(self) -> WidgetModel:
|
|
if self.type == "//":
|
|
if not self.text and not self.label:
|
|
raise ValueError("help widget (//) requires text")
|
|
if not self.text:
|
|
self.text = self.label
|
|
return self
|
|
if self.type == "group":
|
|
if not self.label:
|
|
raise ValueError("group requires label")
|
|
if not self.widgets:
|
|
raise ValueError("group requires nested widgets")
|
|
if len(self.widgets) > 24:
|
|
raise ValueError("group may contain at most 24 widgets")
|
|
return self
|
|
if not self.name or not NAME_RE.match(self.name):
|
|
raise ValueError(
|
|
f"widget name required and must match ^[a-zA-Z_][a-zA-Z0-9_]{{0,63}}$ (got {self.name!r})"
|
|
)
|
|
if not self.label:
|
|
raise ValueError(f"widget '{self.name}' requires label")
|
|
self.label = sanitize_text(self.label, 200)
|
|
self.help = sanitize_text(self.help, 500)
|
|
if self.type in ("choice", "choice_multi", "select"):
|
|
if not self.options:
|
|
raise ValueError(f"widget '{self.name}' requires options")
|
|
if len(self.options) > DEFAULT_LIMITS["max_options"]:
|
|
raise ValueError(
|
|
f"widget '{self.name}' has too many options (max {DEFAULT_LIMITS['max_options']})"
|
|
)
|
|
if self.type == "select":
|
|
self.display = "select"
|
|
return self
|
|
|
|
|
|
class InteractionRequest(BaseModel):
|
|
title: str
|
|
description: str = ""
|
|
widgets: list[WidgetModel]
|
|
submit_label: str = "Confirm"
|
|
cancel_label: str = "Cancel"
|
|
cancelable: bool = True
|
|
timeout_sec: int = 0
|
|
id: str = ""
|
|
|
|
@field_validator("title")
|
|
@classmethod
|
|
def _title(cls, v: str) -> str:
|
|
text = sanitize_text(v, 120)
|
|
if not text:
|
|
raise ValueError("title is required")
|
|
return text
|
|
|
|
@field_validator("description")
|
|
@classmethod
|
|
def _description(cls, v: str) -> str:
|
|
return sanitize_text(v, 500)
|
|
|
|
@field_validator("submit_label", "cancel_label")
|
|
@classmethod
|
|
def _labels(cls, v: str) -> str:
|
|
return sanitize_text(v, 40) or "Confirm"
|
|
|
|
@field_validator("timeout_sec")
|
|
@classmethod
|
|
def _timeout(cls, v: int) -> int:
|
|
n = int(v or 0)
|
|
if n < 0 or n > 86400:
|
|
raise ValueError("timeout_sec must be between 0 and 86400")
|
|
return n
|
|
|
|
@field_validator("id")
|
|
@classmethod
|
|
def _id(cls, v: str) -> str:
|
|
text = sanitize_text(v, 64).lower()
|
|
if text and not ID_RE.match(text):
|
|
raise ValueError("id must match ^[a-z][a-z0-9-]{0,63}$")
|
|
return text
|
|
|
|
@model_validator(mode="after")
|
|
def _widgets(self) -> InteractionRequest:
|
|
if not self.widgets:
|
|
raise ValueError("widgets must contain at least one item")
|
|
if len(self.widgets) > DEFAULT_LIMITS["max_fields"]:
|
|
raise ValueError(
|
|
f"at most {DEFAULT_LIMITS['max_fields']} top-level widgets"
|
|
)
|
|
depth = _max_depth(self.widgets)
|
|
if depth > 3:
|
|
raise ValueError("widget nesting depth must be ≤ 3")
|
|
return self
|
|
|
|
|
|
class InteractionResult(BaseModel):
|
|
interaction_id: str
|
|
status: Literal[
|
|
"submitted", "cancelled", "timeout", "superseded", "error", "channel_changed"
|
|
]
|
|
channel_id: str
|
|
values: dict[str, Any] = Field(default_factory=dict)
|
|
meta: dict[str, Any] = Field(default_factory=dict)
|
|
error: str | None = None
|
|
|
|
def to_dict(self) -> dict[str, Any]:
|
|
return {
|
|
"interaction_id": self.interaction_id,
|
|
"status": self.status,
|
|
"channel_id": self.channel_id,
|
|
"values": {} if self.status != "submitted" else dict(self.values),
|
|
"meta": dict(self.meta),
|
|
"error": self.error,
|
|
}
|
|
|
|
|
|
def _max_depth(widgets: list[WidgetModel], depth: int = 1) -> int:
|
|
deepest = depth
|
|
for widget in widgets:
|
|
if widget.type == "group" and widget.widgets:
|
|
deepest = max(deepest, _max_depth(widget.widgets, depth + 1))
|
|
return deepest
|
|
|
|
|
|
def validate_prompt_args(arguments: dict[str, Any]) -> InteractionRequest:
|
|
try:
|
|
return InteractionRequest.model_validate(arguments or {})
|
|
except Exception as exc:
|
|
from devplacepy.services.devii.errors import ToolInputError
|
|
|
|
raise ToolInputError(f"Invalid ui_prompt arguments: {exc}") from exc
|
|
|
|
|
|
def count_input_fields(widgets: list[WidgetModel]) -> int:
|
|
total = 0
|
|
for widget in widgets:
|
|
if widget.type == "//":
|
|
continue
|
|
if widget.type == "group":
|
|
total += count_input_fields(widget.widgets)
|
|
else:
|
|
total += 1
|
|
return total
|
|
|
|
|
|
def flatten_widgets(widgets: list[WidgetModel]) -> list[WidgetModel]:
|
|
out: list[WidgetModel] = []
|
|
for widget in widgets:
|
|
if widget.type == "group":
|
|
out.extend(flatten_widgets(widget.widgets))
|
|
else:
|
|
out.append(widget)
|
|
return out
|