101 lines
3.4 KiB
Python
101 lines
3.4 KiB
Python
|
"""
|
||
|
|
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
|