VoiceAgent/model_manager.py

"""Model Manager for listing and dynamically changing LLM models at runtime.

Supports OpenAI models, OpenCode Cloud models, Claude Code models, and local MLX models,
persisting user model preferences to disk in model_settings.json.
"""

import json
import logging
import os
import shutil
import subprocess
from pathlib import Path

try:
    from loguru import logger
except ImportError:
    logger = logging.getLogger("model_manager")

DEFAULT_MODEL = "default"

OPENAI_MODELS = {
    "openai/gpt-5.6-luna": "OpenAI GPT-5.6 Luna",
    "openai/gpt-5.6-luna-fast": "OpenAI GPT-5.6 Luna (Fast)",
    "openai/gpt-5.6-sol": "OpenAI GPT-5.6 Sol",
    "openai/gpt-5.6-terra": "OpenAI GPT-5.6 Terra",
    "openai/gpt-5.5": "OpenAI GPT-5.5",
    "openai/gpt-5.4": "OpenAI GPT-5.4",
    "openai/gpt-5.4-mini": "OpenAI GPT-4.4 Mini",
    "openai/gpt-4o": "OpenAI GPT-4o",
    "openai/gpt-4o-mini": "OpenAI GPT-4o Mini",
}

POPULAR_HERMES_MODELS = {
    "default": "Hermes Configured Default Model",
    "hermes-agent": "Hermes Agent",
}

CLAUDE_MODELS = {
    "claude-sonnet-4-6": "Claude 3.7 / Sonnet (Fast, High Capability)",
    "claude-opus-4-6": "Claude 3 Opus (Deep Reasoning)",
    "claude-haiku-4-6": "Claude 3.5 Haiku (Ultra Fast)",
}

MACOS_MODELS = {
    "mlx-community/Qwen2.5-7B-Instruct-4bit": "MLX Qwen 2.5 7B Instruct 4-bit (On-Device Apple Silicon)",
}

MODEL_ALIASES = {
    "luna": "openai/gpt-5.6-luna",
    "luna-fast": "openai/gpt-5.6-luna-fast",
    "sol": "openai/gpt-5.6-sol",
    "terra": "openai/gpt-5.6-terra",
    "gpt5.5": "openai/gpt-5.5",
    "gpt5.4": "openai/gpt-5.4",
    "gpt4o": "openai/gpt-4o",
    "gpt-4o": "openai/gpt-4o",
    "hermes": "hermes-3",
    "hermes3": "hermes-3",
    "hermes-agent": "hermes-agent",
    "sonnet": "claude-sonnet-4-6",
    "claude": "claude-sonnet-4-6",
    "opus": "claude-opus-4-6",
    "haiku": "claude-haiku-4-6",
    "qwen-local": "mlx-community/Qwen2.5-7B-Instruct-4bit",
}


def find_hermes_binary() -> str | None:
    candidates = [
        shutil.which("hermes"),
        os.path.expanduser("~/.hermes/bin/hermes"),
        os.path.expanduser("~/.local/bin/hermes"),
        "/opt/homebrew/bin/hermes",
        "/usr/local/bin/hermes",
    ]
    for candidate in candidates:
        if candidate and os.path.exists(candidate) and os.access(candidate, os.X_OK):
            return candidate
    return shutil.which("hermes")


class ModelManager:
    """Manages active LLM model configuration and dynamic model switching."""

    def __init__(self, workspace_dir: Path | None = None, llm_processor=None):
        self._workspace_dir = Path(workspace_dir) if workspace_dir else Path(__file__).parent
        self._llm_processor = llm_processor
        self._config_file = self._workspace_dir / "model_settings.json"
        self._active_model = DEFAULT_MODEL
        self.load_saved_model()

    def set_llm_processor(self, llm_processor):
        self._llm_processor = llm_processor
        if self._active_model:
            self.apply_model(self._active_model)

    def load_saved_model(self) -> str:
        if self._config_file.exists():
            try:
                data = json.loads(self._config_file.read_text())
                if "model" in data and isinstance(data["model"], str) and data["model"].strip():
                    self._active_model = data["model"].strip()
                    logger.info(f"Loaded saved model preference: {self._active_model}")
            except Exception as e:
                logger.warning(f"Could not load saved model settings: {e}")
        return self._active_model

    def sync_model(self) -> str:
        """Check if model_settings.json was updated on disk and update LLM processor live."""
        current_disk_model = self.load_saved_model()
        if self._llm_processor and current_disk_model:
            if hasattr(self._llm_processor, "_model"):
                if getattr(self._llm_processor, "_model") != current_disk_model:
                    setattr(self._llm_processor, "_model", current_disk_model)
                    if hasattr(self._llm_processor, "_server_session_id"):
                        self._llm_processor._server_session_id = None
                    logger.info(f"Live LLM model synced to: {current_disk_model}")
            elif hasattr(self._llm_processor, "set_model"):
                self._llm_processor.set_model(current_disk_model)
        return current_disk_model

    def save_model(self, model_name: str):
        try:
            self._workspace_dir.mkdir(parents=True, exist_ok=True)
            self._config_file.write_text(json.dumps({"model": model_name}, indent=2))
        except Exception as e:
            logger.warning(f"Could not save model setting: {e}")

    def fetch_all_models(self) -> list[str]:
        cli = find_hermes_binary()
        if cli:
            try:
                res = subprocess.run([cli, "models"], capture_output=True, text=True, timeout=5.0)
                if res.returncode == 0:
                    models = [line.strip() for line in res.stdout.splitlines() if line.strip()]
                    if models:
                        return models
            except Exception as e:
                logger.debug(f"Could not query hermes models: {e}")
        return list(OPENAI_MODELS.keys()) + list(POPULAR_HERMES_MODELS.keys())

    def get_models_dict(self) -> dict:
        """Return structured model categories for the Web UI."""
        fetched = self.fetch_all_models()
        openai_list = [m for m in fetched if m.startswith("openai/")]
        hermes_list = [m for m in fetched if not m.startswith("openai/")]

        if not openai_list:
            openai_list = list(OPENAI_MODELS.keys())
        if not hermes_list:
            hermes_list = list(POPULAR_HERMES_MODELS.keys())

        return {
            "OpenAI Models": [{"id": m, "name": OPENAI_MODELS.get(m, m)} for m in openai_list],
            "Hermes Models": [{"id": m, "name": POPULAR_HERMES_MODELS.get(m, m)} for m in hermes_list],
            "Claude Code Models": [{"id": m, "name": name} for m, name in CLAUDE_MODELS.items()],
            "macOS On-Device MLX": [{"id": m, "name": name} for m, name in MACOS_MODELS.items()],
        }

    def list_available_models(self) -> str:
        """Fetch models and format as readable CLI catalog."""
        models_dict = self.get_models_dict()
        lines = ["Available Models:\n", f"Active Model: {self._active_model}\n"]

        for category, items in models_dict.items():
            lines.append(f"{category}:")
            for item in items:
                m_id = item["id"]
                name = item["name"]
                active = " (ACTIVE)" if m_id == self._active_model else ""
                lines.append(f"  - {m_id}: {name}{active}")
            lines.append("")

        return "\n".join(lines)

    def apply_model(self, model_name: str) -> tuple[bool, str]:
        model_name = model_name.strip()
        matched_model = None

        clean_name = model_name.lower()
        if clean_name in MODEL_ALIASES:
            matched_model = MODEL_ALIASES[clean_name]

        all_known = {
            **OPENAI_MODELS,
            **POPULAR_HERMES_MODELS,
            **CLAUDE_MODELS,
            **MACOS_MODELS,
        }

        if not matched_model:
            for m in all_known:
                if clean_name == m.lower():
                    matched_model = m
                    break

        if not matched_model:
            for m in all_known:
                if clean_name in m.lower():
                    matched_model = m
                    break

        # Fall back to literal model string if explicit format
        if not matched_model and ("/" in model_name or ":" in model_name or "claude" in clean_name):
            matched_model = model_name

        if not matched_model:
            return False, f"Model '{model_name}' not found. Run list to view available models."

        self._active_model = matched_model
        self.save_model(matched_model)

        if self._llm_processor:
            try:
                if hasattr(self._llm_processor, "_model"):
                    setattr(self._llm_processor, "_model", matched_model)
                    if hasattr(self._llm_processor, "_server_session_id"):
                        self._llm_processor._server_session_id = None
                    logger.info(f"Dynamic model updated to: {matched_model}")
                    return True, f"Model changed to {matched_model}."
                elif hasattr(self._llm_processor, "set_model"):
                    self._llm_processor.set_model(matched_model)
                    logger.info(f"Dynamic model updated to: {matched_model}")
                    return True, f"Model changed to {matched_model}."
            except Exception as e:
                logger.error(f"Failed to apply model to LLM processor: {e}")
                return False, f"Could not change model: {e}"

        return True, f"Model set to {matched_model}."