Unify environment handling in Python

src/python/role_play/chat/chat_logger.py

@@ -535,8 +535,10 @@ async def log_pcm_audio(
         """
         # Security: Defensive check to ensure no PCM logging in production
         # even if environment checks are bypassed elsewhere
-        env = os.environ.get("ENV", "dev").lower()
-        if env == "prod" or env == "production":
+        from role_play.common.environment import resolve_environment
+        from role_play.common.models import Environment
+        env = resolve_environment()
+        if env == Environment.PROD:
             logger.debug("PCM audio logging disabled in production environment")
             return
 

src/python/role_play/common/environment.py

@@ -0,0 +1,96 @@
+"""Unified environment parsing and resolution utilities.
+
+This module centralizes how the application determines its deployment
+environment to avoid duplication and ambiguity across the codebase.
+
+Canonical environments: dev | beta | prod
+
+Accepted synonyms (case-insensitive):
+- development -> dev
+- production -> prod
+
+Priority for resolution (highest to lowest):
+1) CONFIG_FILE (extracts env from filename like config/prod.yaml)
+2) ENV
+3) ENVIRONMENT
+4) default: dev
+"""
+
+from __future__ import annotations
+
+import os
+import re
+from typing import Optional
+
+from .models import Environment, EnvironmentInfo
+
+
+def parse_environment_str(value: Optional[str]) -> Environment:
+    """Parse a string into the canonical Environment enum.
+
+    Accepts canonical values (dev|beta|prod) and common synonyms
+    like development|production.
+    """
+    if not value:
+        return Environment.DEV
+
+    s = value.strip().lower()
+    # Map synonyms
+    if s == "development":
+        s = "dev"
+    elif s == "production":
+        s = "prod"
+
+    try:
+        return Environment(s)
+    except ValueError:
+        # Default safely to DEV and let callers decide if they want to warn
+        return Environment.DEV
+
+
+def resolve_environment(config_file: Optional[str] = None) -> Environment:
+    """Resolve the current environment using a consistent precedence order.
+
+    Precedence:
+    1) CONFIG_FILE (env extracted from its basename, e.g., beta.yaml)
+    2) ENV
+    3) ENVIRONMENT
+    4) default to dev
+    """
+    # 1) CONFIG_FILE (argument overrides env var for testability)
+    config_file = config_file or os.getenv("CONFIG_FILE")
+    if config_file:
+        # Try to extract the environment from the filename
+        # e.g., /app/config/beta.yaml -> beta
+        m = re.search(r"/(\\w+)\\.yaml$", config_file)
+        if m:
+            return parse_environment_str(m.group(1))
+
+    # 2) ENV
+    env = os.getenv("ENV")
+    if env:
+        return parse_environment_str(env)
+
+    # 3) ENVIRONMENT (supports development/production synonyms)
+    env2 = os.getenv("ENVIRONMENT")
+    if env2:
+        return parse_environment_str(env2)
+
+    # 4) fallback
+    return Environment.DEV
+
+
+def get_environment_info() -> EnvironmentInfo:
+    """Return an EnvironmentInfo snapshot using the unified resolver."""
+    env = resolve_environment()
+    return EnvironmentInfo(
+        name=env,
+        is_production=(env == Environment.PROD),
+        is_development=(env == Environment.DEV),
+    )
+
+
+def environment_name() -> str:
+    """Convenience accessor for the current environment name as a string."""
+    return resolve_environment().value
+

src/python/role_play/common/logging_config.py

@@ -33,7 +33,11 @@ def format(self, record: logging.LogRecord) -> str:
             log_entry.update(record.extra_fields)
 
         # Add environment info
-        log_entry["environment"] = os.getenv("ENV", "unknown")
+        try:
+            from .environment import environment_name
+            log_entry["environment"] = environment_name()
+        except Exception:
+            log_entry["environment"] = os.getenv("ENV", "unknown")
         log_entry["service"] = os.getenv("SERVICE_NAME", "rps")
         log_entry["version"] = os.getenv("GIT_VERSION", "unknown")
 
@@ -63,7 +67,13 @@ def setup_logging(log_level: str = "INFO", use_structured: bool = True) -> None:
     console_handler = logging.StreamHandler(sys.stdout)
     console_handler.setLevel(numeric_level)
 
-    if use_structured and os.getenv("ENV", "dev") != "dev":
+    try:
+        from .environment import resolve_environment
+        is_dev = (resolve_environment().value == "dev")
+    except Exception:
+        is_dev = (os.getenv("ENV", "dev") == "dev")
+
+    if use_structured and not is_dev:
         # Use structured JSON logging for non-dev environments
         formatter = StructuredFormatter()
     else:
@@ -83,13 +93,19 @@ def setup_logging(log_level: str = "INFO", use_structured: bool = True) -> None:
 
     # Log the configuration
     logger = logging.getLogger(__name__)
+    try:
+        from .environment import environment_name as _env_name
+        env_name_value = _env_name()
+    except Exception:
+        env_name_value = os.getenv("ENV", "dev")
+
     logger.info(
         "Logging configured",
         extra={
             "extra_fields": {
                 "log_level": log_level,
                 "use_structured": use_structured,
-                "environment": os.getenv("ENV", "dev")
+                "environment": env_name_value,
             }
         }
     )
@@ -105,4 +121,4 @@ def get_logger(name: str) -> logging.Logger:
     Returns:
         Logger instance
     """
-    return logging.getLogger(name)
+    return logging.getLogger(name)

src/python/role_play/common/storage_factory.py

@@ -9,6 +9,7 @@
     FileStorageConfig, GCSStorageConfig, S3StorageConfig
 )
 from .exceptions import StorageError
+from .environment import resolve_environment
 
 
 def create_storage_backend(
@@ -111,11 +112,7 @@ def create_storage_from_env(environment: Union[Environment, str] = None) -> Stor
     """
     # Auto-detect environment if not provided
     if environment is None:
-        env_str = os.getenv("ENV", "dev").lower()
-        try:
-            environment = Environment(env_str)
-        except ValueError:
-            raise StorageError(f"Invalid ENV environment variable: {env_str}")
+        environment = resolve_environment()
 
     # Get storage type
     storage_type = os.getenv("STORAGE_TYPE")
@@ -279,4 +276,4 @@ def validate_storage_config(config: StorageConfigUnion) -> None:
     retry_attempts: 3
     retry_delay_seconds: 1.0
 """
-}
+}

src/python/role_play/server/base_server.py

@@ -71,9 +71,10 @@ async def lifespan(app: FastAPI):
         @self.app.get("/health", tags=["Health"], include_in_schema=False)
         async def health_check():
             """Health check endpoint for monitoring and Cloud Run."""
+            from role_play.common.environment import environment_name
             return {
                 "status": "healthy",
-                "environment": os.getenv("ENV", "unknown"),
+                "environment": environment_name(),
                 "version": os.getenv("GIT_VERSION", "unknown"),
                 "service": os.getenv("SERVICE_NAME", "rps")
             }
@@ -156,7 +157,9 @@ async def serve_spa_index(full_path: str):
                 return FileResponse(index_html_path)
             else:
                 # In development, the frontend might not be built yet
-                if os.getenv("ENV", "dev") == "dev":
+                from role_play.common.environment import resolve_environment
+                from role_play.common.models import Environment
+                if resolve_environment() == Environment.DEV:
                     return {
                         "message": "Frontend not found. In development, use 'npm run dev' for the frontend.",
                         "static_dir": static_files_dir
@@ -167,4 +170,4 @@ async def serve_spa_index(full_path: str):
 
     def get_app(self) -> FastAPI:
         """Return the FastAPI application instance."""
-        return self.app
+        return self.app

src/python/role_play/server/config.py

@@ -109,18 +109,12 @@ def __init__(self, **data):
 
 def get_config(environment: Optional[str] = None) -> ServerConfig:
     """
-    Get configuration based on environment.
+    Deprecated: Use role_play.server.config_loader.get_config instead.
 
-    Args:
-        environment: Environment name (development, production) or None for auto-detection
-
-    Returns:
-        ServerConfig: Configuration instance
+    This wrapper delegates to the unified config loader to avoid confusion
+    between different environment names and selection logic.
+    Accepts canonical envs (dev|beta|prod) and common synonyms.
     """
-    if environment is None:
-        environment = os.getenv("ENVIRONMENT", "development")
-
-    if environment == "production":
-        return ProductionConfig()
-    else:
-        return DevelopmentConfig()
+    # Import here to avoid circular import at module load time
+    from .config_loader import get_config as _get_config
+    return _get_config(environment)

src/python/role_play/server/config_loader.py

@@ -10,19 +10,13 @@
 
 import os
 import yaml
-from enum import Enum
 from pathlib import Path
 from typing import Dict, Any, Optional
 from dotenv import load_dotenv
 
 from .config import ServerConfig, DevelopmentConfig, ProductionConfig
-
-
-class Environment(Enum):
-    """Supported environments."""
-    DEV = "dev"
-    BETA = "beta"
-    PROD = "prod"
+from ..common.models import Environment
+from ..common.environment import resolve_environment
 
 
 class ConfigLoader:
@@ -190,26 +184,12 @@ def get_config(self, environment: Optional[str] = None, force_reload: bool = Fal
         # Load environment variables first
         self.load_environment_variables()
 
-        # Determine and validate environment
+        # Determine and validate environment using unified resolver
         if environment is None:
-            # Check CONFIG_FILE first, then ENV, then ENVIRONMENT, then default to dev
-            config_file = os.getenv("CONFIG_FILE")
-            if config_file:
-                # Extract environment from config file path (e.g., /app/config/beta.yaml -> beta)
-                import re
-                match = re.search(r'/(\w+)\.yaml$', config_file)
-                if match:
-                    environment = match.group(1)
-                else:
-                    environment = "dev"
-            else:
-                environment = os.getenv("ENV", os.getenv("ENVIRONMENT", "dev"))
-
-        try:
+            env_enum = resolve_environment()
+        else:
             env_enum = Environment(environment)
-        except ValueError:
-            supported_envs = [e.value for e in Environment]
-            raise ValueError(f"Unsupported environment '{environment}'. Supported environments: {supported_envs}")
+
 
         # Load YAML config
         yaml_config = self.load_yaml_config(env_enum)
@@ -253,4 +233,4 @@ def get_config(environment: Optional[str] = None) -> ServerConfig:
 def reset_config():
     """Reset the global config loader (for testing)."""
     global _config_loader
-    _config_loader = None
+    _config_loader = None

src/python/role_play/server/dependencies.py

@@ -12,6 +12,7 @@
 from ..common.storage import StorageBackend, FileStorage, FileStorageConfig, LockConfig
 from ..common.storage_factory import create_storage_backend
 from ..common.models import User, UserRole, Environment, EnvironmentInfo
+from ..common.environment import resolve_environment, get_environment_info as resolved_env_info
 from ..common.exceptions import AuthenticationError, TokenExpiredError
 from .config_loader import get_config, ServerConfig
 from ..chat.chat_logger import ChatLogger
@@ -24,21 +25,7 @@
 @lru_cache(maxsize=None)
 def get_environment_info() -> EnvironmentInfo:
     """Provides detailed information about the current deployment environment."""
-    env_str = os.getenv("ENV", "dev")
-    try:
-        env_enum = Environment(env_str)
-    except ValueError:
-        logger.warning(f"Unknown environment '{env_str}', defaulting to DEV")
-        env_enum = Environment.DEV
-
-    is_prod = (env_enum == Environment.PROD)
-    is_dev = (env_enum == Environment.DEV)
-
-    return EnvironmentInfo(
-        name=env_enum,
-        is_production=is_prod,
-        is_development=is_dev
-    )
+    return resolved_env_info()
 
 
 @lru_cache(maxsize=None)
@@ -55,19 +42,13 @@ def get_storage_backend() -> StorageBackend:
     """
     config = get_server_config()
 
-    # Determine environment
-    environment = os.getenv("ENVIRONMENT", "dev")
-    try:
-        env_enum = Environment(environment)
-    except ValueError:
-        # Default to dev for unknown environments
-        env_enum = Environment.DEV
-        logger.warning(f"Unknown environment '{environment}', defaulting to DEV")
+    # Determine environment via unified resolver
+    env_enum = resolve_environment()
 
     # Use storage configuration
     if config.storage:
         backend = create_storage_backend(config.storage, env_enum)
-        logger.info(f"Storage backend: {type(backend).__name__} for {environment}")
+        logger.info(f"Storage backend: {type(backend).__name__} for {env_enum.value}")
         return backend
     else:
         raise ValueError("Storage configuration is required")