File size: 2,196 Bytes
24cdea4
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
"""
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