"""Registry settings management. Provides configurable settings for vetting, moderation, and sync. Settings are stored in the database and can be modified via admin API. """ from __future__ import annotations import json import sqlite3 from dataclasses import dataclass from datetime import datetime, timezone from typing import Any, Dict, List, Optional from .db import query_one, query_all @dataclass class Setting: """A single configuration setting.""" key: str value: Any value_type: str # string, int, float, bool, json description: str category: str updated_at: Optional[str] = None updated_by: Optional[str] = None # Default settings with descriptions DEFAULT_SETTINGS: List[Setting] = [ # Vetting thresholds Setting( key="vetting.approve_threshold", value=0.8, value_type="float", description="Quality score threshold for auto-approval (0.0-1.0)", category="vetting", ), Setting( key="vetting.reject_threshold", value=0.3, value_type="float", description="Quality score threshold for auto-rejection (0.0-1.0)", category="vetting", ), # Vetting weights Setting( key="vetting.weight.honesty", value=0.25, value_type="float", description="Weight for honesty check (description matches behavior)", category="vetting", ), Setting( key="vetting.weight.transparency", value=0.20, value_type="float", description="Weight for transparency check (no obfuscation)", category="vetting", ), Setting( key="vetting.weight.scope", value=0.15, value_type="float", description="Weight for scope check (appropriate boundaries)", category="vetting", ), Setting( key="vetting.weight.efficiency", value=0.15, value_type="float", description="Weight for efficiency check (prompt quality)", category="vetting", ), Setting( key="vetting.weight.safety", value=0.25, value_type="float", description="Weight for safety check (no dangerous patterns)", category="vetting", ), # Similarity detection Setting( key="similarity.duplicate_threshold", value=0.9, value_type="float", description="Similarity score to consider tools as duplicates (0.0-1.0)", category="similarity", ), Setting( key="similarity.similar_threshold", value=0.7, value_type="float", description="Similarity score to flag tools as similar (0.0-1.0)", category="similarity", ), Setting( key="similarity.related_threshold", value=0.5, value_type="float", description="Similarity score to flag tools as related (0.0-1.0)", category="similarity", ), # Fabric sync Setting( key="sync.enabled", value=False, value_type="bool", description="Enable scheduled Fabric pattern sync", category="sync", ), Setting( key="sync.interval_hours", value=24, value_type="int", description="Hours between Fabric sync runs", category="sync", ), Setting( key="sync.auto_approve", value=False, value_type="bool", description="Auto-approve synced Fabric patterns (skip moderation)", category="sync", ), Setting( key="sync.default_provider", value="claude", value_type="string", description="Default AI provider for synced patterns", category="sync", ), # Moderation Setting( key="moderation.require_review", value=True, value_type="bool", description="Require manual review for public tools", category="moderation", ), Setting( key="moderation.auto_approve_private", value=True, value_type="bool", description="Auto-approve private/unlisted tools", category="moderation", ), Setting( key="moderation.max_pending_per_user", value=10, value_type="int", description="Maximum pending tools per user before rate limiting", category="moderation", ), # Rate limits (can override defaults) Setting( key="rate_limit.publish.limit", value=20, value_type="int", description="Max publish requests per window", category="rate_limits", ), Setting( key="rate_limit.publish.window", value=3600, value_type="int", description="Publish rate limit window in seconds", category="rate_limits", ), ] def _convert_value(value: str, value_type: str) -> Any: """Convert stored string value to appropriate type.""" if value_type == "int": return int(value) elif value_type == "float": return float(value) elif value_type == "bool": return value.lower() in ("true", "1", "yes") elif value_type == "json": return json.loads(value) return value def _serialize_value(value: Any, value_type: str) -> str: """Serialize value for storage.""" if value_type == "bool": return "true" if value else "false" elif value_type == "json": return json.dumps(value) return str(value) def get_setting(conn: sqlite3.Connection, key: str) -> Optional[Any]: """Get a single setting value. Returns the default if not set in database. """ row = query_one( conn, "SELECT value, value_type FROM registry_settings WHERE key = ?", [key], ) if row: return _convert_value(row["value"], row["value_type"]) # Return default if exists for setting in DEFAULT_SETTINGS: if setting.key == key: return setting.value return None def get_settings_by_category(conn: sqlite3.Connection, category: str) -> Dict[str, Any]: """Get all settings for a category as a dict.""" result = {} # Start with defaults for setting in DEFAULT_SETTINGS: if setting.category == category: result[setting.key] = setting.value # Override with DB values rows = query_all( conn, "SELECT key, value, value_type FROM registry_settings WHERE category = ?", [category], ) for row in rows: result[row["key"]] = _convert_value(row["value"], row["value_type"]) return result def get_all_settings(conn: sqlite3.Connection) -> List[Dict[str, Any]]: """Get all settings with their current values and metadata.""" result = [] # Get all DB settings db_settings = {} rows = query_all(conn, "SELECT * FROM registry_settings") for row in rows: db_settings[row["key"]] = dict(row) # Build combined list for default in DEFAULT_SETTINGS: if default.key in db_settings: row = db_settings[default.key] result.append({ "key": default.key, "value": _convert_value(row["value"], row["value_type"]), "value_type": row["value_type"], "description": default.description, "category": default.category, "updated_at": row["updated_at"], "updated_by": row["updated_by"], "is_default": False, }) else: result.append({ "key": default.key, "value": default.value, "value_type": default.value_type, "description": default.description, "category": default.category, "updated_at": None, "updated_by": None, "is_default": True, }) return result def set_setting( conn: sqlite3.Connection, key: str, value: Any, updated_by: Optional[str] = None, ) -> bool: """Set a setting value. Returns True if successful, False if key is not a valid setting. """ # Find the setting definition setting_def = None for s in DEFAULT_SETTINGS: if s.key == key: setting_def = s break if not setting_def: return False # Validate and convert value try: if setting_def.value_type == "int": value = int(value) elif setting_def.value_type == "float": value = float(value) # Validate range for thresholds if "threshold" in key or "weight" in key: if not 0.0 <= value <= 1.0: return False elif setting_def.value_type == "bool": if isinstance(value, str): value = value.lower() in ("true", "1", "yes") else: value = bool(value) except (ValueError, TypeError): return False serialized = _serialize_value(value, setting_def.value_type) now = datetime.now(timezone.utc).isoformat() conn.execute( """ INSERT INTO registry_settings (key, value, value_type, description, category, updated_at, updated_by) VALUES (?, ?, ?, ?, ?, ?, ?) ON CONFLICT(key) DO UPDATE SET value = excluded.value, updated_at = excluded.updated_at, updated_by = excluded.updated_by """, [key, serialized, setting_def.value_type, setting_def.description, setting_def.category, now, updated_by], ) conn.commit() return True def reset_setting(conn: sqlite3.Connection, key: str) -> bool: """Reset a setting to its default value (remove from DB).""" conn.execute("DELETE FROM registry_settings WHERE key = ?", [key]) conn.commit() return True def reset_all_settings(conn: sqlite3.Connection) -> int: """Reset all settings to defaults. Returns count of reset settings.""" cursor = conn.execute("SELECT COUNT(*) FROM registry_settings") count = cursor.fetchone()[0] conn.execute("DELETE FROM registry_settings") conn.commit() return count def get_vetting_config(conn: sqlite3.Connection) -> Dict[str, Any]: """Get vetting configuration as a structured dict. This is the main entry point for the scrutiny system to get config. """ return { "approve_threshold": get_setting(conn, "vetting.approve_threshold"), "reject_threshold": get_setting(conn, "vetting.reject_threshold"), "weights": { "honesty": get_setting(conn, "vetting.weight.honesty"), "transparency": get_setting(conn, "vetting.weight.transparency"), "scope": get_setting(conn, "vetting.weight.scope"), "efficiency": get_setting(conn, "vetting.weight.efficiency"), "safety": get_setting(conn, "vetting.weight.safety"), }, } def get_similarity_config(conn: sqlite3.Connection) -> Dict[str, float]: """Get similarity detection configuration.""" return { "duplicate_threshold": get_setting(conn, "similarity.duplicate_threshold"), "similar_threshold": get_setting(conn, "similarity.similar_threshold"), "related_threshold": get_setting(conn, "similarity.related_threshold"), } def get_sync_config(conn: sqlite3.Connection) -> Dict[str, Any]: """Get Fabric sync configuration.""" return { "enabled": get_setting(conn, "sync.enabled"), "interval_hours": get_setting(conn, "sync.interval_hours"), "auto_approve": get_setting(conn, "sync.auto_approve"), "default_provider": get_setting(conn, "sync.default_provider"), } def get_moderation_config(conn: sqlite3.Connection) -> Dict[str, Any]: """Get moderation configuration.""" return { "require_review": get_setting(conn, "moderation.require_review"), "auto_approve_private": get_setting(conn, "moderation.auto_approve_private"), "max_pending_per_user": get_setting(conn, "moderation.max_pending_per_user"), }