""" Feature Flags Service for Mid Platform. [AC-IDMP-17] Session-level grayscale and rollback support. """ import logging from dataclasses import dataclass from typing import Any from app.models.mid.schemas import FeatureFlags logger = logging.getLogger(__name__) @dataclass class FeatureFlagConfig: """Feature flag configuration.""" agent_enabled: bool = True rollback_to_legacy: bool = False react_max_iterations: int = 5 enable_tool_registry: bool = True enable_trace_logging: bool = True class FeatureFlagService: """ [AC-IDMP-17] Feature flag service for session-level control. Supports: - Session-level Agent enable/disable - Force rollback to legacy pipeline - Dynamic configuration per session """ def __init__(self): self._session_flags: dict[str, FeatureFlagConfig] = {} self._global_config = FeatureFlagConfig() def get_flags(self, session_id: str) -> FeatureFlags: """ [AC-IDMP-17] Get feature flags for a session. Args: session_id: Session ID Returns: FeatureFlags for the session """ config = self._session_flags.get(session_id, self._global_config) return FeatureFlags( agent_enabled=config.agent_enabled, rollback_to_legacy=config.rollback_to_legacy, ) def set_flags(self, session_id: str, flags: FeatureFlags) -> None: """ [AC-IDMP-17] Set feature flags for a session. Args: session_id: Session ID flags: Feature flags to set """ config = FeatureFlagConfig( agent_enabled=flags.agent_enabled if flags.agent_enabled is not None else self._global_config.agent_enabled, rollback_to_legacy=flags.rollback_to_legacy if flags.rollback_to_legacy is not None else self._global_config.rollback_to_legacy, ) self._session_flags[session_id] = config logger.info( f"[AC-IDMP-17] Feature flags set for session {session_id}: " f"agent_enabled={config.agent_enabled}, rollback_to_legacy={config.rollback_to_legacy}" ) def clear_flags(self, session_id: str) -> None: """Clear feature flags for a session.""" if session_id in self._session_flags: del self._session_flags[session_id] logger.info(f"[AC-IDMP-17] Feature flags cleared for session {session_id}") def is_agent_enabled(self, session_id: str) -> bool: """Check if Agent mode is enabled for a session.""" config = self._session_flags.get(session_id, self._global_config) return config.agent_enabled def should_rollback(self, session_id: str) -> bool: """Check if should rollback to legacy for a session.""" config = self._session_flags.get(session_id, self._global_config) return config.rollback_to_legacy def set_global_config(self, config: FeatureFlagConfig) -> None: """Set global default configuration.""" self._global_config = config logger.info(f"[AC-IDMP-17] Global config updated: {config}") def get_react_max_iterations(self, session_id: str) -> int: """Get ReAct max iterations for a session.""" config = self._session_flags.get(session_id, self._global_config) return config.react_max_iterations