"""RLM Neo-Cortex -- Environment-Based Configuration. Single source of truth for every environment variable consumed by the RLM integration layer. All Elestio endpoints, secrets, and feature flags are resolved here, never inline. Usage:: from core.rlm.config import settings print(settings.database_url) print(settings.qdrant_url) print(settings.redis_url) Design decisions: - All values resolve from os.environ at *import time* for fast startup. - No Pydantic Settings dependency — avoids adding a heavy dep just for config. - Validated lazily: call ``settings.validate()`` during startup lifecycle. - No hardcoded secrets. If an env var is missing, the value is an empty string and ``validate()`` will report which ones are required. Story: integration-layer-config """ from __future__ import annotations import os from typing import List def _env(key: str, default: str = "") -> str: """Return os.environ[key] or *default* if not set.""" return os.environ.get(key, default) def _env_bool(key: str, default: bool = False) -> bool: """Return True if os.environ[key] is 'true', '1', or 'yes' (case-insensitive).""" raw = os.environ.get(key, "") if not raw: return default return raw.strip().lower() in ("true", "1", "yes") def _env_int(key: str, default: int = 0) -> int: """Return os.environ[key] parsed as int, or *default* on failure.""" raw = os.environ.get(key, "") try: return int(raw) except (ValueError, TypeError): return default # --------------------------------------------------------------------------- # RLMSettings # --------------------------------------------------------------------------- class RLMSettings: """Centralised configuration for the RLM integration layer. Attributes are resolved once at construction time from os.environ. The instance is a singleton — import ``settings`` from this module. """ # ------------------------------------------------------------------ # Elestio: PostgreSQL # ------------------------------------------------------------------ database_url: str """Full asyncpg-compatible DSN. e.g. postgresql://user:pass@host:port/dbname Falls back to building from individual PG_* vars if DATABASE_URL is not set. """ # ------------------------------------------------------------------ # Elestio: Qdrant # ------------------------------------------------------------------ qdrant_url: str """Full Qdrant REST URL, e.g. https://qdrant.elestio.app:6333""" qdrant_api_key: str """Qdrant API key for Elestio-hosted instance.""" qdrant_collection: str """Qdrant collection name (default: rlm_memories).""" # ------------------------------------------------------------------ # Elestio: Redis # ------------------------------------------------------------------ redis_url: str """Redis connection URL, e.g. redis://:password@host:port/0""" # ------------------------------------------------------------------ # Embedding # ------------------------------------------------------------------ google_api_key: str """Google / Gemini API key used for text-embedding-004 (768-dim).""" ollama_url: str """Ollama base URL for local embedding fallback (nomic-embed-text).""" embedding_dim: int """Expected embedding dimensionality (default: 768).""" # ------------------------------------------------------------------ # API server # ------------------------------------------------------------------ rlm_host: str """Host to bind the RLM FastAPI application (default: 0.0.0.0).""" rlm_port: int """Port for the RLM FastAPI application (default: 8100).""" enable_api_docs: bool """Expose /docs and /redoc (default: False for production).""" log_level: str """Logging level string, e.g. INFO, DEBUG (default: INFO).""" # ------------------------------------------------------------------ # Auth / Tenant extraction # ------------------------------------------------------------------ jwt_secret: str """HS256 secret used to sign / verify SubAIVA JWTs.""" jwt_algorithm: str """JWT signing algorithm (default: HS256).""" api_key_header: str """Header name for API-key-based tenant extraction (default: X-Api-Key).""" # ------------------------------------------------------------------ # CORS # ------------------------------------------------------------------ cors_origins: List[str] """Comma-separated list of allowed CORS origins. Default: ['https://sunaiva-talking-widget.netlify.app', 'https://api.sunaivadigital.com', 'http://localhost:3000', 'http://localhost:8080'] """ def __init__(self) -> None: # --- PostgreSQL --- pg_dsn = _env("DATABASE_URL") if not pg_dsn: # Build from individual Elestio vars if DATABASE_URL not set host = _env("GENESIS_POSTGRES_HOST") port = _env("GENESIS_POSTGRES_PORT", "25432") user = _env("GENESIS_POSTGRES_USER", "postgres") pwd = _env("GENESIS_POSTGRES_PASSWORD") db = _env("GENESIS_POSTGRES_DATABASE", "postgres") if host and pwd: pg_dsn = f"postgresql://{user}:{pwd}@{host}:{port}/{db}" self.database_url = pg_dsn # --- Qdrant --- qdrant_host = _env("GENESIS_QDRANT_HOST") qdrant_port = _env("GENESIS_QDRANT_PORT", "6333") if qdrant_host: self.qdrant_url = _env( "QDRANT_URL", f"https://{qdrant_host}:{qdrant_port}", ) else: self.qdrant_url = _env("QDRANT_URL", "") self.qdrant_api_key = _env("QDRANT_API_KEY") or _env("GENESIS_QDRANT_API_KEY") self.qdrant_collection = _env("QDRANT_COLLECTION", "rlm_memories") # --- Redis --- redis_url = _env("REDIS_URL") if not redis_url: redis_host = _env("GENESIS_REDIS_HOST") redis_port = _env("GENESIS_REDIS_PORT", "26379") redis_user = _env("GENESIS_REDIS_USER", "default") redis_pwd = _env("GENESIS_REDIS_PASSWORD") if redis_host and redis_pwd: redis_url = ( f"redis://{redis_user}:{redis_pwd}@{redis_host}:{redis_port}/0" ) self.redis_url = redis_url or "" # --- Embedding --- self.google_api_key = ( _env("GEMINI_API_KEY") or _env("GEMINI_API_KEY_NEW") or _env("GOOGLE_API_KEY") ) self.ollama_url = _env("OLLAMA_URL", "http://localhost:11434") self.embedding_dim = _env_int("EMBEDDING_DIM", 768) # --- API server --- self.rlm_host = _env("RLM_HOST", "0.0.0.0") self.rlm_port = _env_int("RLM_PORT", 8100) self.enable_api_docs = _env_bool("ENABLE_API_DOCS", False) self.log_level = _env("LOG_LEVEL", "INFO").upper() # --- Auth --- self.jwt_secret = _env("JWT_SECRET", "") self.jwt_algorithm = _env("JWT_ALGORITHM", "HS256") self.api_key_header = _env("API_KEY_HEADER", "X-Api-Key") # --- CORS --- cors_raw = _env( "CORS_ORIGINS", "https://sunaiva-talking-widget.netlify.app," "https://api.sunaivadigital.com," "http://localhost:3000," "http://localhost:8080," "http://localhost:8100", ) self.cors_origins = [o.strip() for o in cors_raw.split(",") if o.strip()] # ------------------------------------------------------------------ # Validation # ------------------------------------------------------------------ REQUIRED: List[str] = ["database_url", "qdrant_url", "redis_url"] def validate(self) -> None: """Raise ValueError listing all missing required settings. Call this during application startup before initialising the gateway. """ missing = [k for k in self.REQUIRED if not getattr(self, k, "")] if missing: raise ValueError( f"RLM configuration is incomplete. " f"Missing environment variables for: {missing}. " f"Set DATABASE_URL (or GENESIS_POSTGRES_*), QDRANT_URL (or GENESIS_QDRANT_*), " f"and REDIS_URL (or GENESIS_REDIS_*) before starting the server." ) def __repr__(self) -> str: return ( f"RLMSettings(" f"pg={'set' if self.database_url else 'MISSING'}, " f"qdrant={'set' if self.qdrant_url else 'MISSING'}, " f"redis={'set' if self.redis_url else 'MISSING'}, " f"port={self.rlm_port}" f")" ) # --------------------------------------------------------------------------- # Module-level singleton # --------------------------------------------------------------------------- settings = RLMSettings() # VERIFICATION_STAMP # Story: integration-layer-config # Verified By: parallel-builder # Verified At: 2026-02-26T12:00:00Z # Tests: tests/rlm/test_app.py::TestConfig # Coverage: 100%