""" Session utilities for Pipecat server. Provides shared session management logic used across server and runner. """ import os from typing import Any, Dict, Optional from loguru import logger DEFAULT_MAX_SESSION_DURATION = 3540 # 59 minutes def get_max_session_duration( bot_data: Optional[Dict[str, Any]] = None, env_var: str = "MAX_SESSION_DURATION_SECONDS", ) -> int: """ Get max session duration with priority: API > ENV > Default. Priority: 1. API: bot_data.get("max_session_duration") or bot_data.metadata.max_duration 2. ENV: MAX_SESSION_DURATION_SECONDS environment variable 3. Default: DEFAULT_MAX_SESSION_DURATION (3540 seconds = 59 minutes) Args: bot_data: Bot data dictionary or None env_var: Name of environment variable to check Returns: Max session duration in seconds """ # Priority 1: Check if provided in API request (dict format) if bot_data: # Check direct max_session_duration field if "max_session_duration" in bot_data: duration = bot_data["max_session_duration"] if duration and duration > 0: logger.debug(f"Using max session duration from API: {duration}s") return duration # Check nested metadata.max_duration field metadata = bot_data.get("metadata", {}) if metadata and "max_duration" in metadata: duration = metadata["max_duration"] if duration and duration > 0: logger.debug(f"Using max session duration from metadata: {duration}s") return duration # Priority 2: Check environment variable env_duration = os.getenv(env_var) if env_duration: try: duration = int(env_duration) if duration > 0: logger.debug(f"Using max session duration from ENV: {duration}s") return duration except ValueError: logger.warning(f"Invalid {env_var} in ENV: {env_duration}, using default") # Priority 3: Use default logger.debug(f"Using default max session duration: {DEFAULT_MAX_SESSION_DURATION}s") return DEFAULT_MAX_SESSION_DURATION