diff --git a/backends/advanced/init.py b/backends/advanced/init.py index 25a614aa..f0ff8c2e 100644 --- a/backends/advanced/init.py +++ b/backends/advanced/init.py @@ -28,16 +28,27 @@ def __init__(self, args=None): self.console = Console() self.config: Dict[str, Any] = {} self.args = args or argparse.Namespace() - self.config_yml_path = Path("../../config.yml") # Repo root config.yml - self.config_yml_data = None # Check if we're in the right directory if not Path("pyproject.toml").exists() or not Path("src").exists(): self.console.print("[red][ERROR][/red] Please run this script from the backends/advanced directory") sys.exit(1) - # Load config.yml if it exists - self.load_config_yml() + # Initialize ConfigManager + repo_root = Path.cwd().parent.parent # backends/advanced -> repo root + if str(repo_root) not in sys.path: + sys.path.insert(0, str(repo_root)) + + from config_manager import ConfigManager + + self.config_manager = ConfigManager(service_path="backends/advanced") + self.console.print(f"[blue][INFO][/blue] Using config.yml at: {self.config_manager.config_yml_path}") + + # Load existing config or create default structure + self.config_yml_data = self.config_manager.get_full_config() + if not self.config_yml_data: + self.console.print("[yellow][WARNING][/yellow] config.yml not found, will create default structure") + self.config_yml_data = self._get_default_config_structure() def print_header(self, title: str): """Print a colorful header""" @@ -126,21 +137,6 @@ def mask_api_key(self, key: str, show_chars: int = 5) -> str: return f"{key_clean[:show_chars]}{'*' * min(15, len(key_clean) - show_chars * 2)}{key_clean[-show_chars:]}" - def load_config_yml(self): - """Load config.yml from repository root""" - if not self.config_yml_path.exists(): - self.console.print(f"[yellow][WARNING][/yellow] config.yml not found at {self.config_yml_path}") - self.console.print("[yellow]Will create a new config.yml during setup[/yellow]") - self.config_yml_data = self._get_default_config_structure() - return - - try: - with open(self.config_yml_path, 'r') as f: - self.config_yml_data = yaml.safe_load(f) - self.console.print(f"[blue][INFO][/blue] Loaded existing config.yml") - except Exception as e: - self.console.print(f"[red][ERROR][/red] Failed to load config.yml: {e}") - self.config_yml_data = self._get_default_config_structure() def _get_default_config_structure(self) -> Dict[str, Any]: """Return default config.yml structure if file doesn't exist""" @@ -163,36 +159,6 @@ def _get_default_config_structure(self) -> Dict[str, Any]: } } - def save_config_yml(self): - """Save config.yml back to repository root""" - try: - # Backup existing config.yml if it exists - if self.config_yml_path.exists(): - timestamp = datetime.now().strftime("%Y%m%d_%H%M%S") - backup_path = self.config_yml_path.parent / f"config.yml.backup.{timestamp}" - shutil.copy2(self.config_yml_path, backup_path) - self.console.print(f"[blue][INFO][/blue] Backed up config.yml to {backup_path.name}") - - # Write updated config - with open(self.config_yml_path, 'w') as f: - yaml.dump(self.config_yml_data, f, default_flow_style=False, sort_keys=False) - - self.console.print("[green][SUCCESS][/green] config.yml updated successfully") - except Exception as e: - self.console.print(f"[red][ERROR][/red] Failed to save config.yml: {e}") - raise - - def update_config_default(self, key: str, value: str): - """Update a default value in config.yml""" - if "defaults" not in self.config_yml_data: - self.config_yml_data["defaults"] = {} - self.config_yml_data["defaults"][key] = value - - def update_memory_config(self, updates: Dict[str, Any]): - """Update memory configuration in config.yml""" - if "memory" not in self.config_yml_data: - self.config_yml_data["memory"] = {} - self.config_yml_data["memory"].update(updates) def setup_authentication(self): """Configure authentication settings""" @@ -306,8 +272,8 @@ def setup_llm(self): if api_key: self.config["OPENAI_API_KEY"] = api_key # Update config.yml to use OpenAI models - self.update_config_default("llm", "openai-llm") - self.update_config_default("embedding", "openai-embed") + self.config_manager.update_config_defaults({"llm": "openai-llm", "embedding": "openai-embed"}) + self.config_yml_data = self.config_manager.get_full_config() # Reload to stay in sync self.console.print("[green][SUCCESS][/green] OpenAI configured in config.yml") self.console.print("[blue][INFO][/blue] Set defaults.llm: openai-llm") self.console.print("[blue][INFO][/blue] Set defaults.embedding: openai-embed") @@ -317,8 +283,8 @@ def setup_llm(self): elif choice == "2": self.console.print("[blue][INFO][/blue] Ollama selected") # Update config.yml to use Ollama models - self.update_config_default("llm", "local-llm") - self.update_config_default("embedding", "local-embed") + self.config_manager.update_config_defaults({"llm": "local-llm", "embedding": "local-embed"}) + self.config_yml_data = self.config_manager.get_full_config() # Reload to stay in sync self.console.print("[green][SUCCESS][/green] Ollama configured in config.yml") self.console.print("[blue][INFO][/blue] Set defaults.llm: local-llm") self.console.print("[blue][INFO][/blue] Set defaults.embedding: local-embed") @@ -327,7 +293,8 @@ def setup_llm(self): elif choice == "3": self.console.print("[blue][INFO][/blue] Skipping LLM setup - memory extraction disabled") # Disable memory extraction in config.yml - self.update_memory_config({"extraction": {"enabled": False}}) + self.config_manager.update_memory_config({"extraction": {"enabled": False}}) + self.config_yml_data = self.config_manager.get_full_config() # Reload to stay in sync def setup_memory(self): """Configure memory provider - updates config.yml""" @@ -347,9 +314,10 @@ def setup_memory(self): qdrant_url = self.prompt_value("Qdrant URL", "qdrant") self.config["QDRANT_BASE_URL"] = qdrant_url - # Update config.yml - self.update_memory_config({"provider": "chronicle"}) - self.console.print("[green][SUCCESS][/green] Chronicle memory provider configured in config.yml") + # Update config.yml (also updates .env automatically) + self.config_manager.update_memory_config({"provider": "chronicle"}) + self.config_yml_data = self.config_manager.get_full_config() # Reload to stay in sync + self.console.print("[green][SUCCESS][/green] Chronicle memory provider configured in config.yml and .env") elif choice == "2": self.console.print("[blue][INFO][/blue] OpenMemory MCP selected") @@ -359,8 +327,8 @@ def setup_memory(self): user_id = self.prompt_value("OpenMemory user ID", "openmemory") timeout = self.prompt_value("OpenMemory timeout (seconds)", "30") - # Update config.yml with OpenMemory MCP settings - self.update_memory_config({ + # Update config.yml with OpenMemory MCP settings (also updates .env automatically) + self.config_manager.update_memory_config({ "provider": "openmemory_mcp", "openmemory_mcp": { "server_url": mcp_url, @@ -369,7 +337,8 @@ def setup_memory(self): "timeout": int(timeout) } }) - self.console.print("[green][SUCCESS][/green] OpenMemory MCP configured in config.yml") + self.config_yml_data = self.config_manager.get_full_config() # Reload to stay in sync + self.console.print("[green][SUCCESS][/green] OpenMemory MCP configured in config.yml and .env") self.console.print("[yellow][WARNING][/yellow] Remember to start OpenMemory: cd ../../extras/openmemory-mcp && docker compose up -d") elif choice == "3": @@ -378,15 +347,16 @@ def setup_memory(self): mycelia_url = self.prompt_value("Mycelia API URL", "http://localhost:5173") timeout = self.prompt_value("Mycelia timeout (seconds)", "30") - # Update config.yml with Mycelia settings - self.update_memory_config({ + # Update config.yml with Mycelia settings (also updates .env automatically) + self.config_manager.update_memory_config({ "provider": "mycelia", "mycelia": { "api_url": mycelia_url, "timeout": int(timeout) } }) - self.console.print("[green][SUCCESS][/green] Mycelia memory provider configured in config.yml") + self.config_yml_data = self.config_manager.get_full_config() # Reload to stay in sync + self.console.print("[green][SUCCESS][/green] Mycelia memory provider configured in config.yml and .env") self.console.print("[yellow][WARNING][/yellow] Make sure Mycelia is running at the configured URL") def setup_optional_services(self): @@ -559,10 +529,8 @@ def generate_env_file(self): self.console.print("[green][SUCCESS][/green] .env file configured successfully with secure permissions") - # Save config.yml with all updates - self.console.print() - self.console.print("[blue][INFO][/blue] Saving configuration to config.yml...") - self.save_config_yml() + # Note: config.yml is automatically saved by ConfigManager when updates are made + self.console.print("[blue][INFO][/blue] Configuration saved to config.yml and .env (via ConfigManager)") def copy_config_templates(self): """Copy other configuration files""" diff --git a/config_manager.py b/config_manager.py new file mode 100644 index 00000000..2f64b082 --- /dev/null +++ b/config_manager.py @@ -0,0 +1,348 @@ +""" +Shared configuration manager for Chronicle. + +This module provides a unified interface for reading and writing configuration +across both config.yml (source of truth) and .env (backward compatibility). + +Key principles: +- config.yml is the source of truth for memory provider and model settings +- .env files are kept in sync for backward compatibility with legacy code +- All config updates should use this module to maintain consistency + +Usage: + # From any service in the project + from config_manager import ConfigManager + + # For backend service + config = ConfigManager(service_path="backends/advanced") + provider = config.get_memory_provider() + config.set_memory_provider("openmemory_mcp") + + # Auto-detects paths from cwd + config = ConfigManager() +""" + +import logging +import os +import shutil +from datetime import datetime +from pathlib import Path +from typing import Any, Dict, Optional + +import yaml + +logger = logging.getLogger(__name__) + + +class ConfigManager: + """Manages Chronicle configuration across config.yml and .env files.""" + + def __init__(self, service_path: Optional[str] = None, repo_root: Optional[Path] = None): + """ + Initialize ConfigManager. + + Args: + service_path: Path to service directory (e.g., "backends/advanced", "extras/speaker-recognition"). + If None, auto-detects from current working directory. + repo_root: Path to repository root. If None, auto-detects by finding config.yml. + """ + # Find repo root + if repo_root is None: + repo_root = self._find_repo_root() + self.repo_root = Path(repo_root) + + # Find service directory + if service_path is None: + service_path = self._detect_service_path() + self.service_path = self.repo_root / service_path if service_path else None + + # Paths + self.config_yml_path = self.repo_root / "config.yml" + self.env_path = self.service_path / ".env" if self.service_path else None + + logger.debug(f"ConfigManager initialized: repo_root={self.repo_root}, " + f"service_path={self.service_path}, config_yml={self.config_yml_path}") + + def _find_repo_root(self) -> Path: + """Find repository root by searching for config.yml.""" + current = Path.cwd() + + # Walk up until we find config.yml + while current != current.parent: + if (current / "config.yml").exists(): + return current + current = current.parent + + # Fallback to cwd if not found + logger.warning("Could not find config.yml, using current directory as repo root") + return Path.cwd() + + def _detect_service_path(self) -> Optional[str]: + """Auto-detect service path from current working directory.""" + cwd = Path.cwd() + + # Check if we're in a known service directory + known_services = [ + "backends/advanced", + "extras/speaker-recognition", + "extras/openmemory-mcp", + "extras/asr-services", + ] + + for service in known_services: + service_full_path = self.repo_root / service + if cwd == service_full_path or str(cwd).startswith(str(service_full_path)): + return service + + logger.debug("Could not auto-detect service path from cwd") + return None + + def _load_config_yml(self) -> Dict[str, Any]: + """Load config.yml file.""" + if not self.config_yml_path.exists(): + logger.warning(f"config.yml not found at {self.config_yml_path}") + return {} + + try: + with open(self.config_yml_path, 'r') as f: + return yaml.safe_load(f) or {} + except Exception as e: + logger.error(f"Failed to load config.yml: {e}") + return {} + + def _save_config_yml(self, config: Dict[str, Any]): + """Save config.yml file with backup.""" + try: + # Create backup + if self.config_yml_path.exists(): + timestamp = datetime.now().strftime("%Y%m%d_%H%M%S") + backup_path = self.config_yml_path.parent / f"config.yml.backup.{timestamp}" + shutil.copy2(self.config_yml_path, backup_path) + logger.info(f"Backed up config.yml to {backup_path.name}") + + # Write updated config + with open(self.config_yml_path, 'w') as f: + yaml.dump(config, f, default_flow_style=False, sort_keys=False) + + logger.info(f"Saved config.yml to {self.config_yml_path}") + + except Exception as e: + logger.error(f"Failed to save config.yml: {e}") + raise + + def _update_env_file(self, key: str, value: str): + """Update a single key in .env file.""" + if self.env_path is None: + logger.debug("No service path set, skipping .env update") + return + + if not self.env_path.exists(): + logger.warning(f".env file not found at {self.env_path}") + return + + try: + # Read current .env + with open(self.env_path, 'r') as f: + lines = f.readlines() + + # Update or add line + key_found = False + updated_lines = [] + + for line in lines: + if line.strip().startswith(f"{key}="): + updated_lines.append(f"{key}={value}\n") + key_found = True + else: + updated_lines.append(line) + + # If key wasn't found, add it + if not key_found: + updated_lines.append(f"\n# Auto-updated by ConfigManager\n{key}={value}\n") + + # Create backup + backup_path = f"{self.env_path}.bak" + shutil.copy2(self.env_path, backup_path) + logger.debug(f"Backed up .env to {backup_path}") + + # Write updated file + with open(self.env_path, 'w') as f: + f.writelines(updated_lines) + + # Update environment variable for current process + os.environ[key] = value + + logger.info(f"Updated {key}={value} in .env file") + + except Exception as e: + logger.error(f"Failed to update .env file: {e}") + raise + + def get_memory_provider(self) -> str: + """ + Get current memory provider from config.yml. + + Returns: + Memory provider name (chronicle, openmemory_mcp, or mycelia) + """ + config = self._load_config_yml() + provider = config.get("memory", {}).get("provider", "chronicle").lower() + + # Map legacy names + if provider in ("friend-lite", "friend_lite"): + provider = "chronicle" + + return provider + + def set_memory_provider(self, provider: str) -> Dict[str, Any]: + """ + Set memory provider in both config.yml and .env. + + This updates: + 1. config.yml: memory.provider field (source of truth) + 2. .env: MEMORY_PROVIDER variable (backward compatibility, if service_path set) + + Args: + provider: Memory provider name (chronicle, openmemory_mcp, or mycelia) + + Returns: + Dict with status and details of the update + + Raises: + ValueError: If provider is invalid + """ + # Validate provider + provider = provider.lower().strip() + valid_providers = ["chronicle", "openmemory_mcp", "mycelia"] + + if provider not in valid_providers: + raise ValueError( + f"Invalid provider '{provider}'. " + f"Valid providers: {', '.join(valid_providers)}" + ) + + # Update config.yml + config = self._load_config_yml() + + if "memory" not in config: + config["memory"] = {} + + config["memory"]["provider"] = provider + self._save_config_yml(config) + + # Update .env for backward compatibility (if we have a service path) + if self.env_path and self.env_path.exists(): + self._update_env_file("MEMORY_PROVIDER", provider) + + return { + "message": ( + f"Memory provider updated to '{provider}' in config.yml" + f"{' and .env' if self.env_path else ''}. " + "Please restart services for changes to take effect." + ), + "provider": provider, + "config_yml_path": str(self.config_yml_path), + "env_path": str(self.env_path) if self.env_path else None, + "requires_restart": True, + "status": "success" + } + + def get_memory_config(self) -> Dict[str, Any]: + """ + Get complete memory configuration from config.yml. + + Returns: + Full memory configuration dict + """ + config = self._load_config_yml() + return config.get("memory", {}) + + def update_memory_config(self, updates: Dict[str, Any]): + """ + Update memory configuration in config.yml. + + Args: + updates: Dict of updates to merge into memory config + """ + config = self._load_config_yml() + + if "memory" not in config: + config["memory"] = {} + + # Deep merge updates + config["memory"].update(updates) + + self._save_config_yml(config) + + # If provider was updated, also update .env + if "provider" in updates and self.env_path: + self._update_env_file("MEMORY_PROVIDER", updates["provider"]) + + def get_config_defaults(self) -> Dict[str, Any]: + """ + Get defaults configuration from config.yml. + + Returns: + Defaults configuration dict (llm, embedding, stt, tts, vector_store) + """ + config = self._load_config_yml() + return config.get("defaults", {}) + + def update_config_defaults(self, updates: Dict[str, str]): + """ + Update defaults configuration in config.yml. + + Args: + updates: Dict of updates to merge into defaults config + (e.g., {"llm": "openai-llm", "embedding": "openai-embed"}) + """ + config = self._load_config_yml() + + if "defaults" not in config: + config["defaults"] = {} + + # Update defaults + config["defaults"].update(updates) + + self._save_config_yml(config) + + def get_full_config(self) -> Dict[str, Any]: + """ + Get complete config.yml as dictionary. + + Returns: + Full configuration dict + """ + return self._load_config_yml() + + def save_full_config(self, config: Dict[str, Any]): + """ + Save complete config.yml from dictionary. + + Args: + config: Full configuration dict to save + """ + self._save_config_yml(config) + + +# Global singleton instance +_config_manager: Optional[ConfigManager] = None + + +def get_config_manager(service_path: Optional[str] = None) -> ConfigManager: + """ + Get global ConfigManager singleton instance. + + Args: + service_path: Optional service path for .env updates. + If None, uses cached instance or creates new one. + + Returns: + ConfigManager instance + """ + global _config_manager + + if _config_manager is None or service_path is not None: + _config_manager = ConfigManager(service_path=service_path) + + return _config_manager