213 lines
9.2 KiB
Python
213 lines
9.2 KiB
Python
"""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"},
|
|
"llm.text_replacements": {
|
|
"value": [],
|
|
"description": "List of {from, to} pairs. Each 'from' substring in LLM scene_text output is replaced with 'to' (can be empty string to remove).",
|
|
},
|
|
}
|
|
|
|
# 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
|
|
|
|
|
|
async def apply_text_replacements(session: AsyncSession, text: str) -> str:
|
|
"""Apply llm.text_replacements to a text string.
|
|
|
|
Each replacement is a dict {from: str, to: str}. The 'from' substring is
|
|
replaced with 'to' (which can be empty to remove the substring).
|
|
"""
|
|
if not text:
|
|
return text
|
|
replacements = await get_setting(session, "llm.text_replacements")
|
|
if not replacements or not isinstance(replacements, list):
|
|
return text
|
|
for r in replacements:
|
|
if not isinstance(r, dict):
|
|
continue
|
|
frm = r.get("from")
|
|
to = r.get("to", "")
|
|
if frm and isinstance(frm, str):
|
|
text = text.replace(frm, to)
|
|
return text
|