Centralize configuration through ConfigManager

.gitignore

@@ -79,6 +79,7 @@ test-results/
 # MinIO Data (persistent storage)
 data/minio/
 minio-data/
+backend/minio-data/
 
 # Legacy S3 Mock Storage (deprecated)
 mocks/s3-mock/s3-mock-storage/

backend/core/auth.py

@@ -15,7 +15,7 @@ def is_user_in_group(user_id: str, group_id: str) -> bool:
         True if user is authorized for the group
     """
     # Check if this is debug mode and test user should have admin access
-    from modules.config.manager import config_manager
+    from modules.config.config_manager import config_manager
     app_settings = config_manager.app_settings
 
     if (app_settings.debug_mode and 

backend/core/capabilities.py

@@ -32,14 +32,9 @@ def _get_secret() -> bytes:
     """Get the capability token secret as bytes.
 
     Order of precedence:
-    - ENV: CAPABILITY_TOKEN_SECRET
-    - App settings (config)
+    - App settings (config manager)
     - Fallback development secret (unsafe for production)
     """
-    env_secret = os.getenv("CAPABILITY_TOKEN_SECRET")
-    if env_secret:
-        return env_secret.encode("utf-8")
-
     try:
         settings = config_manager.app_settings
         if getattr(settings, "capability_token_secret", None):

backend/core/otel_config.py

@@ -74,13 +74,8 @@
         self.service_version = service_version
         self.is_development = self._is_development()
         self.log_level = self._get_log_level()
-        # Resolve logs directory robustly: explicit env override else project_root/logs
-        if os.getenv("APP_LOG_DIR"):
-            self.logs_dir = Path(os.getenv("APP_LOG_DIR"))
-        else:
-            # This file: backend/core/otel_config.py -> project root is 2 levels up
-            project_root = Path(__file__).resolve().parents[2]
-            self.logs_dir = project_root / "logs"
+        # Resolve logs directory robustly: use config manager
+        self.logs_dir = self._get_logs_dir()
         self.log_file = self.logs_dir / "app.jsonl"
         self.logs_dir.mkdir(parents=True, exist_ok=True)
         self._setup_telemetry()
@@ -89,18 +84,39 @@
     # ------------------------------------------------------------------
     # Internals
     # ------------------------------------------------------------------
+    def _get_logs_dir(self) -> Path:
+        """Get logs directory from config manager or default to project_root/logs."""
+        try:
+            from modules.config import config_manager
+            if config_manager.app_settings.app_log_dir:
+                return Path(config_manager.app_settings.app_log_dir)
+        except Exception:            
+            pass
+        # Fallback: project_root/logs
+        project_root = Path(__file__).resolve().parents[2]
+        return project_root / "logs"
+
     def _is_development(self) -> bool:
-        return (
-            os.getenv("DEBUG_MODE", "false").lower() == "true"
-            or os.getenv("ENVIRONMENT", "production").lower() in {"dev", "development"}
-        )
+        try:
+            from modules.config import config_manager
+            settings = config_manager.app_settings
+            return (
+                settings.debug_mode
+                or settings.environment.lower() in {"dev", "development"}
+            )
+        except Exception:
+            # Fallback to environment variables if config not available
+            return (
+                os.getenv("DEBUG_MODE", "false").lower() == "true"
+                or os.getenv("ENVIRONMENT", "production").lower() in {"dev", "development"}
+            )
 
     def _get_log_level(self) -> int:
         try:
-            from config import config_manager  # type: ignore  # local import to avoid circular
-
-            level_name = getattr(config_manager.app_settings, "log_level", "INFO").upper()
-        except Exception:  # noqa: BLE001
+            from modules.config import config_manager
+            level_name = config_manager.app_settings.log_level.upper()
+        except Exception:
+            # Fallback to environment variable if config not available
             level_name = os.getenv("LOG_LEVEL", "INFO").upper()
         level = getattr(logging, level_name, None)
         return level if isinstance(level, int) else logging.INFO

backend/core/prompt_risk.py

@@ -10,7 +10,6 @@
 import json
 import logging
 import math
-import os
 import re
 from collections import Counter
 from datetime import datetime
@@ -20,12 +19,23 @@
 logger = logging.getLogger(__name__)
 
 
-# Default thresholds; can be overridden by env vars
-THRESHOLDS = {
-    "low": int(os.getenv("PI_THRESHOLD_LOW", "30")),
-    "medium": int(os.getenv("PI_THRESHOLD_MEDIUM", "50")),
-    "high": int(os.getenv("PI_THRESHOLD_HIGH", "80")),
-}
+def _get_thresholds() -> Dict[str, int]:
+    """Get prompt injection risk thresholds from config manager."""
+    try:
+        from modules.config import config_manager
+        settings = config_manager.app_settings
+        return {
+            "low": settings.pi_threshold_low,
+            "medium": settings.pi_threshold_medium,
+            "high": settings.pi_threshold_high,
+        }
+    except Exception:
+        # Fallback to defaults if config not available
+        return {
+            "low": 30,
+            "medium": 50,
+            "high": 80,
+        }
 
 
 def calculate_prompt_injection_risk(message: str, *, mode: str = "general") -> Dict[str, object]:
@@ -113,12 +123,13 @@ def calculate_prompt_injection_risk(message: str, *, mode: str = "general") -> D
             score -= 10
         score = max(0, score)
 
-    # Risk buckets
-    if score >= THRESHOLDS["high"]:
+    # Risk buckets - get thresholds from config
+    thresholds = _get_thresholds()
+    if score >= thresholds["high"]:
         level = "high"
-    elif score >= THRESHOLDS["medium"]:
+    elif score >= thresholds["medium"]:
         level = "medium"
-    elif score >= THRESHOLDS["low"]:
+    elif score >= thresholds["low"]:
         level = "low"
     else:
         level = "minimal"

backend/modules/config/__init__.py

@@ -7,7 +7,7 @@
 - CLI tools for validation and inspection
 """
 
-from .manager import (
+from .config_manager import (
     ConfigManager,
     AppSettings,
     LLMConfig,

backend/modules/config/cli.py

@@ -12,7 +12,7 @@
 import logging
 import sys
 
-from .manager import ConfigManager
+from .config_manager import ConfigManager
 
 # Set up logging for CLI
 logging.basicConfig(level=logging.INFO, format='%(levelname)s: %(message)s')

backend/modules/config/manager.py → backend/modules/config/config_manager.py

@@ -191,6 +191,24 @@ def agent_mode_available(self) -> bool:
     help_config_file: str = Field(default="help-config.json", validation_alias="HELP_CONFIG_FILE")
     messages_config_file: str = Field(default="messages.txt", validation_alias="MESSAGES_CONFIG_FILE")
 
+    # Config directory paths
+    app_config_overrides: str = Field(default="config/overrides", validation_alias="APP_CONFIG_OVERRIDES")
+    app_config_defaults: str = Field(default="config/defaults", validation_alias="APP_CONFIG_DEFAULTS")
+
+    # Logging directory
+    app_log_dir: Optional[str] = Field(default=None, validation_alias="APP_LOG_DIR")
+
+    # Environment mode
+    environment: str = Field(default="production", validation_alias="ENVIRONMENT")
+
+    # Prompt injection risk thresholds
+    pi_threshold_low: int = Field(default=30, validation_alias="PI_THRESHOLD_LOW")
+    pi_threshold_medium: int = Field(default=50, validation_alias="PI_THRESHOLD_MEDIUM")
+    pi_threshold_high: int = Field(default=80, validation_alias="PI_THRESHOLD_HIGH")
+
+    # Runtime directories
+    runtime_feedback_dir: str = Field(default="runtime/feedback", validation_alias="RUNTIME_FEEDBACK_DIR")
+
     model_config = {
         "env_file": "../.env", 
         "env_file_encoding": "utf-8", 
@@ -216,15 +234,16 @@ def _search_paths(self, file_name: str) -> List[Path]:
         The backend process often runs with CWD=backend/, so relative paths like
         "config/overrides" incorrectly resolve to backend/config/overrides (which doesn't exist).
 
-        Environment variables can override these directories:
-            APP_CONFIG_OVERRIDES, APP_CONFIG_DEFAULTS (can be absolute or relative to project root)
+        Configuration settings can override these directories:
+            app_config_overrides, app_config_defaults (can be absolute or relative to project root)
 
         Legacy fallbacks (backend/configfilesadmin, backend/configfiles) are preserved.
         """
         project_root = self._backend_root.parent  # /workspaces/atlas-ui-3-11
 
-        overrides_env = os.getenv("APP_CONFIG_OVERRIDES", "config/overrides")
-        defaults_env = os.getenv("APP_CONFIG_DEFAULTS", "config/defaults")
+        # Use app_settings for config paths
+        overrides_env = self.app_settings.app_config_overrides
+        defaults_env = self.app_settings.app_config_defaults
 
         overrides_root = Path(overrides_env)
         defaults_root = Path(defaults_env)

backend/modules/mcp_tools/client.py

@@ -23,8 +23,18 @@ class MCPToolManager:
 
     def __init__(self, config_path: Optional[str] = None):
         if config_path is None:
-            overrides_root = os.getenv("APP_CONFIG_OVERRIDES", "config/overrides")
-            candidate = Path(overrides_root) / "mcp.json"
+            # Use config manager to get config path
+            app_settings = config_manager.app_settings
+            overrides_root = Path(app_settings.app_config_overrides)
+
+            # If relative, resolve from project root
+            if not overrides_root.is_absolute():
+                # This file is in backend/modules/mcp_tools/client.py
+                backend_root = Path(__file__).parent.parent.parent
+                project_root = backend_root.parent
+                overrides_root = project_root / overrides_root
+
+            candidate = overrides_root / "mcp.json"
             if not candidate.exists():
                 # Legacy fallback
                 candidate = Path("backend/configfilesadmin/mcp.json")

backend/routes/admin_routes.py

@@ -53,8 +53,18 @@ def require_admin(current_user: str = Depends(get_current_user)) -> str:
 
 def setup_config_overrides() -> None:
     """Ensure editable overrides directory exists; seed from defaults / legacy if empty."""
-    overrides_root = Path(os.getenv("APP_CONFIG_OVERRIDES", "config/overrides"))
-    defaults_root = Path(os.getenv("APP_CONFIG_DEFAULTS", "config/defaults"))
+    app_settings = config_manager.app_settings
+    overrides_root = Path(app_settings.app_config_overrides)
+    defaults_root = Path(app_settings.app_config_defaults)
+
+    # If relative paths, resolve from project root
+    if not overrides_root.is_absolute():
+        project_root = Path(__file__).parent.parent.parent
+        overrides_root = project_root / overrides_root
+    if not defaults_root.is_absolute():
+        project_root = Path(__file__).parent.parent.parent
+        defaults_root = project_root / defaults_root
+
     overrides_root.mkdir(parents=True, exist_ok=True)
     defaults_root.mkdir(parents=True, exist_ok=True)
 
@@ -99,8 +109,7 @@ def get_admin_config_path(filename: str) -> Path:
         custom_filename = filename
 
     # Use same logic as config manager to resolve relative paths from project root
-    overrides_env = os.getenv("APP_CONFIG_OVERRIDES", "config/overrides")
-    base = Path(overrides_env)
+    base = Path(app_settings.app_config_overrides)
 
     # If relative path, resolve from project root (parent of backend directory)
     if not base.is_absolute():
@@ -154,9 +163,9 @@ def _project_root() -> Path:
 
 
 def _log_base_dir() -> Path:
-    env_path = os.getenv("APP_LOG_DIR")
-    if env_path:
-        return Path(env_path)
+    app_settings = config_manager.app_settings
+    if app_settings.app_log_dir:
+        return Path(app_settings.app_log_dir)
     return _project_root() / "logs"
 
 
@@ -608,7 +617,11 @@ async def get_system_status(admin_user: str = Depends(require_admin)):
     """
     try:
         # Configuration status: overrides directory and file count
-        overrides_root = Path(os.getenv("APP_CONFIG_OVERRIDES", "config/overrides"))
+        app_settings = config_manager.app_settings
+        overrides_root = Path(app_settings.app_config_overrides)
+        if not overrides_root.is_absolute():
+            project_root = _project_root()
+            overrides_root = project_root / overrides_root
         overrides_root.mkdir(parents=True, exist_ok=True)
         config_files = list(overrides_root.glob("*"))
         config_status = "healthy" if config_files else "warning"

backend/routes/config_routes.py

@@ -36,11 +36,9 @@ async def get_banners(current_user: str = Depends(get_current_user)):
     # Read messages from messages.txt file
     try:
         from pathlib import Path
-        import os
 
-        # Use same logic as admin routes to find messages file
-        overrides_env = os.getenv("APP_CONFIG_OVERRIDES", "config/overrides")
-        base = Path(overrides_env)
+        # Use app settings for config path
+        base = Path(app_settings.app_config_overrides)
 
         # If relative path, resolve from project root
         if not base.is_absolute():

backend/routes/feedback_routes.py

@@ -46,7 +46,8 @@ class FeedbackResponse(BaseModel):
 
 def get_feedback_directory() -> Path:
     """Get the feedback storage directory."""
-    base = Path(os.getenv("RUNTIME_FEEDBACK_DIR", "runtime/feedback"))
+    from modules.config import config_manager
+    base = Path(config_manager.app_settings.runtime_feedback_dir)
     base.mkdir(parents=True, exist_ok=True)
     return base
 

backend/tests/test_agent_roa.py

@@ -11,7 +11,7 @@
 from application.chat.service import ChatService  # type: ignore
 from interfaces.llm import LLMProtocol  # type: ignore
 from interfaces.transport import ChatConnectionProtocol  # type: ignore
-from modules.config.manager import ConfigManager  # type: ignore
+from modules.config.config_manager import ConfigManager  # type: ignore
 
 
 class FakeLLM(LLMProtocol):

backend/tests/test_compliance_level.py

@@ -1,6 +1,6 @@
 """Test compliance level functionality for MCP servers and data sources."""
 
-from modules.config.manager import MCPServerConfig, MCPConfig
+from backend.modules.config.config_manager import MCPServerConfig, MCPConfig
 
 
 def test_mcp_server_config_with_compliance_level():