""" Shared Pydantic models for API communication between voicebot and server components. This module contains all the shared data models used for: - HTTP API requests and responses - WebSocket message payloads - Data persistence structures Test comment for shared reload detection - updated again """ from __future__ import annotations from typing import List, Dict, Optional, Literal, Any from pydantic import BaseModel # ============================================================================= # Core Data Models (for persistence and API communication) # ============================================================================= class NamePasswordRecord(BaseModel): """Password hash record for reserved names""" salt: str hash: str class LobbyModel(BaseModel): """Core lobby model used across components""" id: str name: str private: bool = False class SessionModel(BaseModel): """Core session model used across components""" id: str name: str = "" lobbies: List[LobbyModel] = [] class ParticipantModel(BaseModel): """Represents a participant in a lobby/session""" name: str session_id: str live: bool protected: bool has_media: bool = True # Whether this participant provides audio/video streams bot_run_id: Optional[str] = None bot_provider_id: Optional[str] = None bot_instance_id: Optional[str] = None muted: bool = False video_on: bool = True # ============================================================================= # HTTP API Request/Response Models # ============================================================================= class AdminNamesResponse(BaseModel): """Response for admin names endpoint""" name_passwords: Dict[str, NamePasswordRecord] class AdminActionResponse(BaseModel): """Response for admin actions""" status: Literal["ok", "not_found", "error"] name: str class AdminValidationResponse(BaseModel): """Response for admin session validation""" status: Literal["ok", "error"] issues: List[str] = [] issue_count: int = 0 error: Optional[str] = None class AdminMetricsConfig(BaseModel): """Config data for metrics response""" anonymous_timeout: int displaced_timeout: int cleanup_interval: int max_cleanup_per_cycle: int class AdminMetricsResponse(BaseModel): """Response for admin session metrics""" total_sessions: int active_sessions: int named_sessions: int displaced_sessions: int old_anonymous_sessions: int old_displaced_sessions: int total_lobbies: int cleanup_candidates: int config: AdminMetricsConfig class AdminSetPassword(BaseModel): """Request model for setting admin password""" name: str password: str class AdminClearPassword(BaseModel): """Request model for clearing admin password""" name: str class HealthResponse(BaseModel): """Health check response""" status: str class ClientStatusResponse(BaseModel): """Client status response""" client_running: bool session_name: str lobby: str server_url: str class LobbyListItem(BaseModel): """Lobby item for list responses""" id: str name: str class LobbiesResponse(BaseModel): """Response containing list of lobbies""" lobbies: List[LobbyListItem] class SessionResponse(BaseModel): """Session response model""" id: str name: str lobbies: List[LobbyModel] protected: bool = False has_media: bool = False bot_run_id: Optional[str] = None bot_provider_id: Optional[str] = None bot_instance_id: Optional[str] = None class LobbyCreateData(BaseModel): """Data for lobby creation""" name: str private: bool = False class LobbyCreateRequest(BaseModel): """Request for creating a lobby""" type: Literal["lobby_create"] data: LobbyCreateData class LobbyCreateResponse(BaseModel): """Response for lobby creation""" type: Literal["lobby_created"] data: LobbyModel # ============================================================================= # WebSocket Message Models # ============================================================================= class JoinStatusModel(BaseModel): """WebSocket message for join status updates""" status: str message: str = "" class UserJoinedModel(BaseModel): """WebSocket message for user joined events""" name: str session_id: str class LobbyStateModel(BaseModel): """WebSocket message for lobby state updates""" participants: List[ParticipantModel] = [] class UpdateNameModel(BaseModel): name: str protected: Optional[bool] = False class WebSocketErrorModel(BaseModel): """WebSocket error message""" error: str class WebSocketMessageModel(BaseModel): """Base model for all WebSocket messages""" type: str data: Optional[ JoinStatusModel | UserJoinedModel | LobbyStateModel | UpdateNameModel | ICECandidateDictModel | AddPeerModel | RemovePeerModel | SessionDescriptionModel | IceCandidateModel | LobbyCreateResponse | ChatMessageModel | ChatMessagesListModel | WebSocketErrorModel | Dict[str, str] ] = None # Generic dict for simple messages, optional for status messages # ============================================================================= # WebRTC Signaling Models (specific to voicebot) # ============================================================================= class ICECandidateDictModel(BaseModel): """ICE candidate dictionary structure""" candidate: Optional[str] = None sdpMid: Optional[str] = None sdpMLineIndex: Optional[int] = None class AddPeerModel(BaseModel): """WebRTC add peer message""" peer_id: str peer_name: str has_media: bool = True should_create_offer: bool = False class RemovePeerModel(BaseModel): """WebRTC remove peer message""" peer_id: str peer_name: str class SessionDescriptionTypedModel(BaseModel): """WebRTC session description""" type: str sdp: str class SessionDescriptionModel(BaseModel): """WebRTC session description message""" peer_id: str peer_name: str session_description: SessionDescriptionTypedModel class IceCandidateModel(BaseModel): """WebRTC ICE candidate message""" peer_id: str peer_name: str candidate: ICECandidateDictModel # ============================================================================= # Chat Message Models # ============================================================================= class ChatMessageModel(BaseModel): """Chat message model""" id: str message: str sender_name: str sender_session_id: str timestamp: float lobby_id: str class ChatMessagesSendModel(BaseModel): """WebSocket message for sending a chat message""" message: str class ChatMessagesListModel(BaseModel): """WebSocket message containing list of chat messages""" messages: List[ChatMessageModel] class ChatMessagesRequest(BaseModel): """Request for chat messages""" lobby_id: str limit: Optional[int] = 50 class ChatMessagesResponse(BaseModel): """Response containing chat messages""" messages: List[ChatMessageModel] # ============================================================================= # Data Persistence Models # ============================================================================= class LobbySaved(BaseModel): """Lobby model for persistence""" id: str name: str private: bool = False class SessionSaved(BaseModel): """Session model for persistence""" id: str name: str = "" lobbies: List[LobbySaved] = [] created_at: float = 0.0 last_used: float = 0.0 displaced_at: Optional[float] = None # When name was taken over has_media: bool = True # Whether this session provides audio/video streams bot_run_id: Optional[str] = None # Bot run ID for tracking bot_provider_id: Optional[str] = None # Bot provider ID bot_instance_id: Optional[str] = None # Bot instance ID for tracking class SessionsPayload(BaseModel): """Complete sessions data for persistence""" sessions: List[SessionSaved] = [] name_passwords: Dict[str, NamePasswordRecord] = {} # ============================================================================= # Bot Provider Models # ============================================================================= class BotInfoModel(BaseModel): """Information about a specific bot""" name: str description: str has_media: bool = True # Whether this bot provides audio/video streams configurable: bool = False # Whether this bot supports per-lobby configuration config_schema: Optional[Dict[str, Any]] = ( None # JSON schema for configuration parameters ) class BotConfigParameter(BaseModel): """Definition of a bot configuration parameter""" name: str type: Literal["string", "number", "boolean", "select", "range"] label: str description: str default_value: Optional[Any] = None required: bool = False # For select type options: Optional[List[Dict[str, str]]] = ( None # [{"value": "val", "label": "Label"}] ) # For range/number type min_value: Optional[float] = None max_value: Optional[float] = None step: Optional[float] = None # For string type max_length: Optional[int] = None pattern: Optional[str] = None # Regex pattern class BotConfigSchema(BaseModel): """Schema defining all configurable parameters for a bot""" bot_name: str version: str = "1.0" parameters: List[BotConfigParameter] categories: Optional[List[Dict[str, List[str]]]] = ( None # Group parameters by category ) class BotLobbyConfig(BaseModel): """Bot configuration for a specific lobby""" bot_name: str lobby_id: str provider_id: str config_values: Dict[str, Any] # Parameter name -> value mapping created_at: float updated_at: float created_by: str # Session ID of who created the config class BotConfigUpdateRequest(BaseModel): """Request to update bot configuration""" bot_instance_id: str lobby_id: str config_values: Dict[str, Any] class BotConfigUpdateResponse(BaseModel): """Response to bot configuration update""" success: bool message: str updated_config: Optional[BotLobbyConfig] = None class BotConfigListResponse(BaseModel): """Response listing bot configurations for a lobby""" lobby_id: str configs: List[BotLobbyConfig] class BotProviderBotsResponse(BaseModel): """Response from bot provider's /bots endpoint""" bots: List[BotInfoModel] class BotProviderModel(BaseModel): """Bot provider registration information""" provider_id: str base_url: str name: str description: str = "" provider_key: str registered_at: float last_seen: float class BotProviderRegisterRequest(BaseModel): """Request to register a bot provider""" base_url: str name: str description: str = "" provider_key: str provider_id: str # Required static provider ID class BotProviderRegisterResponse(BaseModel): """Response after registering a bot provider""" provider_id: str status: str = "registered" class BotProviderPublicModel(BaseModel): """Public bot provider information (safe for frontend)""" provider_id: str name: str description: str = "" registered_at: float last_seen: float class BotProviderListResponse(BaseModel): """Response listing all registered bot providers""" providers: List[BotProviderPublicModel] class BotListResponse(BaseModel): """Response listing all available bots from all providers""" bots: List[BotInfoModel] # List of available bots providers: Dict[str, str] # bot_name -> provider_id class BotJoinLobbyRequest(BaseModel): """Request to make a bot join a lobby""" lobby_id: str nick: str = "" provider_id: Optional[str] = None # Optional: specify which provider to use class BotJoinPayload(BaseModel): """Payload sent to bot provider to make a bot join a lobby""" lobby_id: str session_id: str nick: str server_url: str insecure: bool = False config_values: Optional[Dict[str, Any]] = ( None # Existing configuration for the bot in this lobby ) class BotJoinLobbyResponse(BaseModel): """Response after requesting a bot to join a lobby""" status: str bot_instance_id: str # Unique ID for this bot instance bot_name: str # Bot type name run_id: str provider_id: str session_id: str # Session ID in the lobby class BotProviderJoinResponse(BaseModel): """Response from bot provider's /bots/{bot_name}/join endpoint""" status: str bot: str run_id: str class BotLeaveLobbyRequest(BaseModel): """Request to make a bot leave a lobby""" bot_instance_id: str # The unique bot instance ID (not session_id) class BotLeaveLobbyResponse(BaseModel): """Response after requesting a bot to leave a lobby""" status: str bot_instance_id: str session_id: str run_id: Optional[str] = None class BotInstanceModel(BaseModel): """Model representing a bot instance in a lobby""" bot_instance_id: str bot_name: str # Bot type name nick: str lobby_id: str session_id: str provider_id: str run_id: str has_media: bool created_at: float # timestamp