"""Settings service — runtime overrides from the `settings` table. Layered: 1. App config (env vars / .env) — `app.config.get_settings()` 2. DB overrides — `settings` table 3. `get_setting(key)` merges them with DB taking precedence. """ from __future__ import annotations import secrets from typing import Any from sqlalchemy import select from sqlalchemy.ext.asyncio import AsyncSession from app.config import get_settings from app.models import Setting # Default settings written to DB on first run. # These match the seed list in `docs/AI-RPG_TZ_TDD.md` §5.2.2. DEFAULT_SETTINGS: dict[str, dict[str, Any]] = { "llm.api_url": {"value": None, "description": "OpenAI-compatible endpoint URL"}, "llm.api_key": {"value": "", "description": "API key for LLM (stored as string)"}, "llm.model": {"value": "qwen2.5-7b-instruct", "description": "Chat model name"}, "llm.temperature_orchestrator": {"value": 0.7, "description": "Phase 1 temperature"}, "llm.temperature_writer": {"value": 0.85, "description": "Phase 2 temperature"}, "llm.max_tokens": {"value": 2048, "description": "Max completion tokens"}, "llm.timeout_seconds": {"value": 60, "description": "LLM call timeout"}, "embeddings.provider": { "value": "offline_hash", "description": "offline_hash | openai", }, "embeddings.api_url": {"value": "", "description": "OpenAI-compatible embeddings URL"}, "embeddings.api_key": {"value": "", "description": "API key for embeddings"}, "embeddings.model": {"value": "text-embedding-3-small", "description": "Embedding model"}, "embeddings.dimension": {"value": 256, "description": "Embedding dimension"}, "embeddings.timeout_seconds": {"value": 30, "description": "Embeddings API timeout"}, "embeddings.batch_size": {"value": 32, "description": "Batch size for embeddings API"}, "embeddings.cache_ttl_seconds": {"value": 300, "description": "LRU cache TTL"}, "embeddings.max_text_chars": {"value": 4000, "description": "Text truncation before embedding"}, "context.guaranteed_messages": {"value": 10, "description": "Always-in-context messages"}, "context.compression_threshold_messages": {"value": 20, "description": "Compression threshold"}, "context.compression_threshold_tokens": {"value": 6000, "description": "Token-based threshold"}, "context.scene_text_truncate_tokens": {"value": 500, "description": "scene_text truncation"}, "context.auto_rag_on_entity_mention": {"value": False, "description": "Auto RAG on entity mention"}, "context.safety_margin_tokens": {"value": 500, "description": "Safety margin from edge"}, "qdrant.url": {"value": "http://qdrant:6333", "description": "Qdrant URL"}, "qdrant.api_key": {"value": "", "description": "Qdrant API key"}, "qdrant.collection_prefix": {"value": "", "description": "Collection prefix"}, "game.deferred_triggers_enabled": {"value": True, "description": "Enable deferred triggers"}, "game.max_substeps_per_iteration": {"value": 8, "description": "Max Phase 1 substeps"}, "game.max_suggested_actions": {"value": 3, "description": "Max suggested actions"}, "ui.page_title": {"value": "AI-RPG", "description": "Browser tab title"}, "ui.favicon_url": {"value": "/icon.png", "description": "Favicon URL"}, "ui.logo_url": {"value": "/icon.png", "description": "Logo URL"}, "ui.og_image_url": {"value": "", "description": "OpenGraph image URL"}, "admin.setup_token": {"value": "", "description": "Admin setup token"}, } # Keys whose values should never be returned to the client in plaintext. SECRET_KEYS = {"llm.api_key", "embeddings.api_key", "qdrant.api_key", "admin.setup_token"} # Map: setting key -> (env-var attribute on Settings, default value) ENV_OVERRIDE_MAP = { "llm.api_url": ("llm_api_url", None), "llm.api_key": ("llm_api_key", None), "llm.model": ("llm_model", None), "llm.timeout_seconds": ("llm_timeout_seconds", None), "embeddings.provider": ("embeddings_provider", None), "embeddings.api_url": ("embeddings_api_url", None), "embeddings.api_key": ("embeddings_api_key", None), "embeddings.model": ("embeddings_model", None), "embeddings.dimension": ("embeddings_dimension", None), "qdrant.url": ("qdrant_url", None), "qdrant.api_key": ("qdrant_api_key", None), "qdrant.collection_prefix": ("qdrant_collection_prefix", None), "ui.page_title": ("ui_page_title", None), "ui.favicon_url": ("ui_favicon_url", None), "ui.logo_url": ("ui_logo_url", None), } async def seed_default_settings(session: AsyncSession) -> None: """Upsert all DEFAULT_SETTINGS rows. Called on application startup.""" existing = ( await session.execute(select(Setting).where(Setting.key.in_(DEFAULT_SETTINGS.keys()))) ).scalars().all() existing_keys = {row.key for row in existing} cfg = get_settings() for key, spec in DEFAULT_SETTINGS.items(): if key in existing_keys: continue value = spec["value"] # Apply env-var override on first seed (so docker-compose env wins). env_attr = ENV_OVERRIDE_MAP.get(key) if env_attr is not None and env_attr[1] is None: env_val = getattr(cfg, env_attr[0], None) if env_val not in (None, ""): value = env_val # Special: admin.setup_token — generate random if env not set if key == "admin.setup_token" and not value: env_token = cfg.admin_setup_token value = env_token if env_token else secrets.token_urlsafe(16) session.add( Setting(key=key, value=value, description=spec["description"]) ) await session.commit() async def get_all_settings(session: AsyncSession) -> dict[str, Any]: """Return all settings as a dict (with env overrides applied for missing keys).""" rows = (await session.execute(select(Setting))).scalars().all() cfg = get_settings() out: dict[str, Any] = {} for key, spec in DEFAULT_SETTINGS.items(): row = next((r for r in rows if r.key == key), None) if row is not None: out[key] = row.value else: # Fall back to env-var if present, otherwise spec default env_attr = ENV_OVERRIDE_MAP.get(key) env_val = ( getattr(cfg, env_attr[0], None) if env_attr and env_attr[1] is None else None ) out[key] = env_val if env_val not in (None, "") else spec["value"] return out async def get_setting(session: AsyncSession, key: str) -> Any: """Get a single setting by key, with env override fallback.""" row = ( await session.execute(select(Setting).where(Setting.key == key)) ).scalar_one_or_none() if row is not None: return row.value # Env-var fallback env_attr = ENV_OVERRIDE_MAP.get(key) if env_attr and env_attr[1] is None: env_val = getattr(get_settings(), env_attr[0], None) if env_val not in (None, ""): return env_val return DEFAULT_SETTINGS.get(key, {}).get("value") async def set_setting(session: AsyncSession, key: str, value: Any) -> Any: """Upsert a setting value. Returns the new value.""" if key not in DEFAULT_SETTINGS: # Allow ad-hoc keys but warn in logs import logging logging.getLogger(__name__).warning("creating_unregistered_setting", extra={"key": key}) row = ( await session.execute(select(Setting).where(Setting.key == key)) ).scalar_one_or_none() if row is None: row = Setting( key=key, value=value, description=DEFAULT_SETTINGS.get(key, {}).get("description"), ) session.add(row) else: row.value = value await session.commit() return value def mask_secret(key: str, value: Any) -> Any: """Mask secret values for safe display in admin UI.""" if key in SECRET_KEYS and isinstance(value, str) and value: if len(value) <= 4: return "****" return value[:2] + "…" + "*" * (min(len(value) - 4, 8)) + value[-2:] return value async def get_admin_setup_token(session: AsyncSession) -> str: """Return the current admin setup token (generating one if absent).""" token = await get_setting(session, "admin.setup_token") if not token: token = secrets.token_urlsafe(16) await set_setting(session, "admin.setup_token", token) return token