models.py

import contextvars
from dataclasses import dataclass
import ipaddress
import json
import logging
import os
import pathlib
import threading
from urllib.parse import urlparse
import urllib.error
import urllib.request

try:
    import ollama as _ollama_mod
except ImportError:
    _ollama_mod = None  # type: ignore[assignment]

try:
    from langchain_ollama import ChatOllama
except ImportError:
    ChatOllama = None  # type: ignore[assignment,misc]

logger = logging.getLogger(__name__)

# ── Cloud provider URLs ─────────────────────────────────────────────────────
OPENAI_BASE_URL = "https://api.openai.com/v1"
OPENROUTER_BASE_URL = "https://openrouter.ai/api/v1"
ANTHROPIC_BASE_URL = "https://api.anthropic.com/v1"
GOOGLE_GENAI_BASE_URL = "https://generativelanguage.googleapis.com/v1beta"
XAI_BASE_URL = "https://api.x.ai/v1"
MINIMAX_ANTHROPIC_BASE_URL = "https://api.minimax.io/anthropic"
OLLAMA_CLOUD_BASE_URL = "https://ollama.com"

# ── Context-size heuristics (prefix-match, checked top-to-bottom) ───────────
# Used when the provider API doesn't expose context_length (e.g. OpenAI) and
# the model isn't in the OpenRouter cross-reference cache.  More-specific
# prefixes must come before shorter ones.  Covers OpenAI, Anthropic & Gemini.
_CONTEXT_HEURISTICS: list[tuple[str, int]] = [
    # ── OpenAI ────────────────────────────────────────────────────────
    ("gpt-4.1",       1_048_576),   # gpt-4.1 / mini / nano  — 1M
    ("gpt-4.5",       1_048_576),   # gpt-4.5 family          — 1M
    ("gpt-5",         1_048_576),   # gpt-5 / 5.4 etc         — 1M
    ("gpt-4o",          128_000),   # gpt-4o / 4o-mini        — 128K
    ("gpt-4-turbo",     128_000),   # gpt-4-turbo             — 128K
    ("gpt-4",             8_192),   # base gpt-4 (legacy)     — 8K
    ("gpt-3.5",          16_385),   # gpt-3.5-turbo           — 16K
    ("o1",              200_000),   # o1 / o1-mini / o1-pro   — 200K
    ("o3",              200_000),   # o3 / o3-mini / o3-pro   — 200K
    ("o4",              200_000),   # o4-mini etc              — 200K
    ("chatgpt-",        128_000),   # chatgpt- aliases        — 128K
    # ── Anthropic ─────────────────────────────────────────────────────
    ("claude-opus-4",  1_000_000),  # Opus 4.x                — 1M
    ("claude-sonnet-4",1_000_000),  # Sonnet 4.x              — 1M
    ("claude-haiku-4",   200_000),  # Haiku 4.x               — 200K
    ("claude-3-5",       200_000),  # Claude 3.5 family       — 200K
    ("claude-3",         200_000),  # Claude 3 family         — 200K
    ("claude-2",         100_000),  # Claude 2.x (legacy)     — 100K
    ("claude",           200_000),  # Catch-all Claude        — 200K
    # ── Google Gemini ─────────────────────────────────────────────────
    ("gemini-3",       1_048_576),  # Gemini 3.x              — 1M
    ("gemini-2.5",     1_048_576),  # Gemini 2.5 Flash/Pro    — 1M
    ("gemini-2.0",     1_048_576),  # Gemini 2.0              — 1M
    ("gemini-1.5-pro", 2_097_152),  # Gemini 1.5 Pro          — 2M
    ("gemini-1.5",     1_048_576),  # Gemini 1.5 Flash        — 1M
    ("gemini-1.0",        32_768),  # Legacy Gemini 1.0       — 32K
    ("gemini",         1_048_576),  # Catch-all Gemini        — 1M
    # ── xAI (Grok) ─────────────────────────────────────────────────────
    ("grok-4",         2_000_000),  # Grok 4 / 4.20           — 2M
    ("grok-3",           131_072),  # Grok 3 & 3-mini         — 131K
    ("grok-2",           131_072),  # Grok 2 family           — 131K
    ("grok",             131_072),  # Catch-all Grok          — 131K
    # ── MiniMax ──────────────────────────────────────────────────────
    ("minimax-m2",       204_800),  # MiniMax M2.x Anthropic-compatible models
]

_CLOUD_CONTEXT_FALLBACK = 256_000   # safe default for totally unknown models


def _estimate_context_heuristic(model_name: str) -> int:
    """Guess context size from the model name using prefix heuristics.

    Strips any ``provider/`` prefix (e.g. ``openai/gpt-4o`` → ``gpt-4o``)
    before matching.  Returns ``_CLOUD_CONTEXT_FALLBACK`` if nothing matches.
    """
    bare = _runtime_model_name(model_name).split("/")[-1].lower()  # strip provider/ slug
    for prefix, ctx in _CONTEXT_HEURISTICS:
        if bare.startswith(prefix):
            return ctx
    return _CLOUD_CONTEXT_FALLBACK


def _parse_provider_model_ref(model_name: str | None) -> tuple[str, str] | None:
    try:
        from providers.selection import parse_model_ref

        return parse_model_ref(model_name)
    except Exception:
        return None


def _runtime_model_name(model_name: str | None) -> str:
    raw = str(model_name or "")
    parsed = _parse_provider_model_ref(raw)
    return parsed[1] if parsed else raw


def _provider_qualified_cloud_cache_key(provider_id: str | None, model_id: str | None) -> str:
    provider = str(provider_id or "").strip()
    model = str(model_id or "").strip()
    return f"model:{provider}:{model}" if provider and model else ""


def _cloud_cache_entry_for(model_name: str | None, provider_id: str | None = None) -> dict | None:
    parsed = _parse_provider_model_ref(model_name)
    runtime_model = _runtime_model_name(model_name)
    provider = provider_id or (parsed[0] if parsed else None)
    qualified_key = _provider_qualified_cloud_cache_key(provider, runtime_model)
    if qualified_key:
        info = _cloud_model_cache.get(qualified_key)
        if isinstance(info, dict):
            cached_provider = str(info.get("provider") or "")
            if not cached_provider or cached_provider == provider:
                return info
    info = _cloud_model_cache.get(runtime_model)
    if isinstance(info, dict):
        cached_provider = str(info.get("provider") or "")
        if not provider or not cached_provider or cached_provider == provider:
            return info
    return None

# Prefixes considered chat-capable when filtering OpenAI /v1/models
_OPENAI_CHAT_PREFIXES = ("gpt-", "o1", "o3", "o4", "chatgpt-")
# Substrings that indicate a non-chat model — skip these
_OPENAI_SKIP_SUBSTRINGS = ("dall-e", "whisper", "tts", "embedding", "davinci",
                           "babbage", "moderation", "realtime", "audio",
                           "transcri", "search")

# ── Dynamic cloud model cache ───────────────────────────────────────────────
_cloud_model_cache: dict[str, dict] = {}   # model_id → {label, ctx, provider}
_cloud_cache_lock = threading.Lock()

# ── Context catalog (keyless OpenRouter public data) ────────────────────────
_context_catalog: dict[str, int] = {}      # model_id → context_length
_context_catalog_lock = threading.Lock()

# ── Deprecated Ollama discovery cache (kept as inert compatibility state) ───
_trending_ollama_cache: list[str] = []
_trending_fetched: bool = False

DEFAULT_MODEL = "qwen3:14b"
DEFAULT_CONTEXT_SIZE = 32768

CONTEXT_SIZE_OPTIONS = [16384, 32768, 65536, 131072, 262144]
CONTEXT_SIZE_LABELS = {16384: "16K", 32768: "32K",
                       65536: "64K", 131072: "128K", 262144: "256K"}

# Cloud-model context options (user-selectable cap — reduces cost / rate-limit pressure)
DEFAULT_CLOUD_CONTEXT_SIZE = 131072   # 128K — safe default for most API tiers
CLOUD_CONTEXT_SIZE_OPTIONS = [32768, 65536, 131072, 262144, 524288, 1048576]
CLOUD_CONTEXT_SIZE_LABELS = {
    32768: "32K", 65536: "64K", 131072: "128K",
    262144: "256K", 524288: "512K", 1048576: "1M",
}

# ── Persistent settings file ────────────────────────────────────────────────
def _coerce_context_size(
    value,
    default: int,
    *,
    allowed: list[int] | tuple[int, ...] | None = None,
    minimum: int | None = None,
) -> int:
    """Return a numeric context size from persisted/UI values."""
    fallback = int(default or 0)
    try:
        if isinstance(value, bool):
            raise ValueError
        if isinstance(value, (int, float)):
            parsed = int(value)
        else:
            text = str(value or "").strip().lower().replace(",", "").replace("_", "")
            multiplier = 1
            if text.endswith("k"):
                multiplier = 1_000
                text = text[:-1]
            elif text.endswith("m"):
                multiplier = 1_000_000
                text = text[:-1]
            parsed = int(float(text) * multiplier)
    except (TypeError, ValueError):
        parsed = fallback
    if allowed:
        allowed_ints = sorted(int(item) for item in allowed)
        if parsed in allowed_ints:
            return parsed
        if parsed < allowed_ints[0]:
            return allowed_ints[0]
        return min(allowed_ints, key=lambda item: abs(item - parsed))
    if minimum is not None and parsed < int(minimum):
        return int(minimum)
    return parsed if parsed > 0 else fallback


_DATA_DIR = pathlib.Path(os.environ.get("THOTH_DATA_DIR", pathlib.Path.home() / ".thoth"))
_SETTINGS_PATH = _DATA_DIR / "model_settings.json"
_CLOUD_CACHE_PATH = _DATA_DIR / "cloud_models_cache.json"
_CONTEXT_CATALOG_PATH = _DATA_DIR / "context_catalog_cache.json"


def _load_settings() -> dict:
    """Load persisted model settings, or return defaults."""
    try:
        if _SETTINGS_PATH.exists():
            return json.loads(_SETTINGS_PATH.read_text())
    except Exception:
        logger.warning("Failed to load model settings from %s", _SETTINGS_PATH, exc_info=True)
    return {}


def _save_settings(settings: dict):
    """Persist model settings to disk."""
    _DATA_DIR.mkdir(parents=True, exist_ok=True)
    _SETTINGS_PATH.write_text(json.dumps(settings, indent=2))


def _load_cloud_cache() -> dict:
    """Load persisted cloud model cache from disk."""
    try:
        if _CLOUD_CACHE_PATH.exists():
            data = json.loads(_CLOUD_CACHE_PATH.read_text())
            if isinstance(data, dict):
                return data
    except Exception:
        logger.warning("Failed to load cloud cache from %s", _CLOUD_CACHE_PATH, exc_info=True)
    return {}


def _save_cloud_cache():
    """Persist current cloud model cache to disk."""
    try:
        _DATA_DIR.mkdir(parents=True, exist_ok=True)
        with _cloud_cache_lock:
            _CLOUD_CACHE_PATH.write_text(json.dumps(_cloud_model_cache))
    except Exception:
        logger.warning("Failed to save cloud cache to %s", _CLOUD_CACHE_PATH, exc_info=True)


def _load_context_catalog() -> dict[str, int]:
    """Load persisted context catalog from disk."""
    try:
        if _CONTEXT_CATALOG_PATH.exists():
            data = json.loads(_CONTEXT_CATALOG_PATH.read_text())
            if isinstance(data, dict):
                return data
    except Exception:
        logger.warning("Failed to load context catalog from %s",
                       _CONTEXT_CATALOG_PATH, exc_info=True)
    return {}


def _save_context_catalog():
    """Persist current context catalog to disk."""
    try:
        _DATA_DIR.mkdir(parents=True, exist_ok=True)
        with _context_catalog_lock:
            _CONTEXT_CATALOG_PATH.write_text(json.dumps(_context_catalog))
    except Exception:
        logger.warning("Failed to save context catalog to %s",
                       _CONTEXT_CATALOG_PATH, exc_info=True)


# Initialise from saved settings (fall back to defaults for first run)
_saved = _load_settings()

# Load persisted cloud cache so is_cloud_model() works before refresh
_cloud_model_cache.update(_load_cloud_cache())

# Load persisted context catalog so context resolution works before live fetch
_context_catalog.update(_load_context_catalog())

POPULAR_MODELS = [
    # ── Qwen family ──────────────────────────────────────────────────────
    "qwen3:8b", "qwen3:14b", "qwen3:30b", "qwen3:32b", "qwen3:235b",
    "qwen3.5:9b", "qwen3.5:27b", "qwen3.5:35b", "qwen3.5:122b",
    "qwen3-coder:30b",
    # ── Llama family ─────────────────────────────────────────────────────
    "llama3.1:8b", "llama3.1:70b", "llama3.1:405b",
    "llama3.3:70b",
    "llama3-groq-tool-use:8b", "llama3-groq-tool-use:70b",
    # ── Mistral family ───────────────────────────────────────────────────
    "mistral:7b",
    "mistral-nemo:12b",
    "mistral-small:22b", "mistral-small:24b",
    "mistral-small3.1:24b",
    "mistral-small3.2:24b",
    "mistral-large:123b",
    "mixtral:8x7b", "mixtral:8x22b",
    "magistral:24b",
    "ministral-3:8b", "ministral-3:14b",
    # ── Other tool-capable models ────────────────────────────────────────
    "rnj-1:8b",
    "glm-4.7-flash:30b",
    "nemotron-3-nano:30b",
    "nemotron:70b",
    "devstral-small-2:24b",
    "devstral-2:123b",
    "olmo-3.1:32b",
    "lfm2:24b",
    "gpt-oss:20b", "gpt-oss:120b",
    "firefunction-v2:70b",
]

# Set of all model *family* prefixes known to support Ollama tool calling.
# Used to flag downloaded models NOT in this set with a ⚠️ warning.
_TOOL_COMPATIBLE_FAMILIES: set[str] = {
    m.split(":")[0] for m in POPULAR_MODELS
}

try:
    from providers.selection import model_choice_value as _canonical_model_choice_value

    _current_model = _canonical_model_choice_value(_saved.get("model", DEFAULT_MODEL))
except Exception:
    _current_model = _saved.get("model", DEFAULT_MODEL)
_num_ctx = _coerce_context_size(
    _saved.get("context_size", DEFAULT_CONTEXT_SIZE),
    DEFAULT_CONTEXT_SIZE,
    allowed=CONTEXT_SIZE_OPTIONS,
)
_cloud_num_ctx = _coerce_context_size(
    _saved.get("cloud_context_size", DEFAULT_CLOUD_CONTEXT_SIZE),
    DEFAULT_CLOUD_CONTEXT_SIZE,
    allowed=CLOUD_CONTEXT_SIZE_OPTIONS,
)
_llm_instance = None
_model_max_ctx_cache: dict[str, int | None] = {}  # model_name → max context


@dataclass(frozen=True)
class ContextPolicy:
    model_ref: str
    provider_id: str
    runtime_model: str
    native_max: int | None
    user_cap: int
    effective_context: int
    policy_kind: str
    cap_source: str
    request_application: str


def _normalize_ollama_client_host(host: str) -> str:
    normalized = (host or "127.0.0.1").strip()
    if normalized == "0.0.0.0":
        return "127.0.0.1"
    if normalized == "::":
        return "::1"
    return normalized


def _format_ollama_base_url(host: str, port: int, scheme: str = "http") -> str:
    formatted_host = host
    try:
        if isinstance(ipaddress.ip_address(host), ipaddress.IPv6Address):
            formatted_host = f"[{host}]"
    except ValueError:
        pass
    return f"{scheme or 'http'}://{formatted_host}:{port}"


def _ollama_endpoint_parts() -> tuple[str, int, str]:
    raw = (os.environ.get("OLLAMA_HOST") or "127.0.0.1").strip() or "127.0.0.1"
    parsed = urlparse(raw if "://" in raw else f"//{raw}")
    host = _normalize_ollama_client_host(parsed.hostname or raw)
    try:
        port = parsed.port or 11434
    except ValueError:
        port = 11434
    return host, port, parsed.scheme or "http"


def _ollama_base_url() -> str:
    """Return a client-safe Ollama base URL derived from OLLAMA_HOST."""
    host, port, scheme = _ollama_endpoint_parts()
    return _format_ollama_base_url(host, port, scheme)


def _ollama_client():
    if not _ollama_mod:
        return None
    client_factory = getattr(_ollama_mod, "Client", None)
    if callable(client_factory):
        return client_factory(host=_ollama_base_url())
    return _ollama_mod


def _ollama_http_json(path: str, payload: dict | None = None, *, timeout: float = 4.0) -> dict:
    url = f"{_ollama_base_url().rstrip('/')}/{str(path or '').lstrip('/')}"
    data = None
    headers = {"Accept": "application/json"}
    if payload is not None:
        data = json.dumps(payload).encode("utf-8")
        headers["Content-Type"] = "application/json"
    request = urllib.request.Request(url, data=data, headers=headers, method="POST" if payload is not None else "GET")
    with urllib.request.urlopen(request, timeout=timeout) as response:
        raw = response.read()
    parsed = json.loads(raw.decode("utf-8") or "{}")
    return parsed if isinstance(parsed, dict) else {}


def _chat_ollama(model: str, **kwargs):
    if ChatOllama is None:
        raise RuntimeError("langchain-ollama is not installed")
    if "reasoning" not in kwargs:
        try:
            from providers.ollama import is_ollama_reasoning_model

            if is_ollama_reasoning_model(model):
                kwargs["reasoning"] = True
        except Exception:
            pass
    return ChatOllama(model=model, base_url=_ollama_base_url(), **kwargs)


def get_llm():
    """Return the current LLM instance, creating one if needed.

    If the default model is a cloud model, returns the appropriate
    ``ChatOpenAI`` instance.  Otherwise returns ``ChatOllama``.
    """
    global _llm_instance
    if _llm_instance is None:
        if is_cloud_model(_current_model):
            _llm_instance = _get_cloud_llm(_current_model)
        else:
            runtime_model = _runtime_model_name(_current_model)
            num_ctx = _local_num_ctx_for(_current_model)
            logger.info("Creating LLM instance: model=%s, num_ctx=%s", runtime_model, num_ctx)
            _llm_instance = _chat_ollama(model=runtime_model, num_ctx=num_ctx)
    return _llm_instance


_override_llm_cache: dict[tuple[str, int], object] = {}  # model → ChatOllama or ChatOpenAI


def clear_llm_cache() -> None:
    """Drop cached chat model clients so provider credential changes take effect."""
    global _llm_instance
    _llm_instance = None
    _override_llm_cache.clear()


def _local_num_ctx_for(model_name: str | None) -> int:
    """Return the Ollama ``num_ctx`` after applying native model caps."""
    model_max = get_model_max_context(model_name)
    user_ctx = _coerce_context_size(_num_ctx, DEFAULT_CONTEXT_SIZE, minimum=CONTEXT_SIZE_OPTIONS[0])
    model_ctx = _coerce_context_size(model_max, 0) if model_max else None
    return min(model_ctx, user_ctx) if model_ctx and model_ctx > 0 else user_ctx


def get_llm_for(model_name: str, num_ctx: int | None = None):
    """Return an LLM for a specific model (not the global singleton).

    For local (Ollama) models, returns a ``ChatOllama``.
    For cloud (OpenRouter) models, returns a ``ChatOpenAI`` pointed at
    the OpenRouter API.  Results are cached per (model, ctx) pair.
    """
    if is_cloud_model(model_name):
        return _get_cloud_llm(model_name)

    runtime_model = _runtime_model_name(model_name)
    if num_ctx is None:
        num_ctx = _local_num_ctx_for(model_name)
    key = (runtime_model, num_ctx)
    if key not in _override_llm_cache:
        logger.info("Creating override LLM: model=%s, num_ctx=%s", runtime_model, num_ctx)
        _override_llm_cache[key] = _chat_ollama(model=runtime_model, num_ctx=num_ctx)
    return _override_llm_cache[key]


def get_model_max_context(model_name: str | None = None) -> int | None:
    """Query Ollama for the model's native max context length.

    For cloud models, returns the hardcoded context size from the catalog.
    Returns the context_length from model metadata, or *None* if it
    cannot be determined.  Results are cached per model name.
    """
    raw_name = model_name or _current_model
    resolved = _resolved_context_identity(raw_name)
    if resolved and resolved.provider_id.startswith("custom_openai_"):
        endpoint = resolved.endpoint or {}
        models = endpoint.get("models") if isinstance(endpoint, dict) else []
        if isinstance(models, list):
            for item in models:
                if not isinstance(item, dict):
                    continue
                item_model = str(item.get("model_id") or item.get("id") or "")
                if item_model != resolved.runtime_model:
                    continue
                ctx = _coerce_context_size(item.get("context_window") or item.get("ctx"), 0)
                return ctx if ctx > 0 else None
        probe = endpoint.get("last_probe") if isinstance(endpoint.get("last_probe"), dict) else {}
        ctx = _coerce_context_size(probe.get("context_window"), 0)
        return ctx if ctx > 0 else None
    if is_cloud_model(raw_name):
        return get_cloud_model_context(raw_name)
    name = _runtime_model_name(raw_name)
    if name in _model_max_ctx_cache:
        return _model_max_ctx_cache[name]
    client = _ollama_client()
    metadata: dict | None = None
    try:
        if client:
            info = client.show(name)
            metadata = getattr(info, "modelinfo", None) or getattr(info, "model_info", None) or {}
    except Exception:
        logger.debug("Could not query max context for model %s", name, exc_info=True)
    if metadata is None:
        try:
            info = _ollama_http_json("/api/show", {"model": name})
            metadata = info.get("model_info") or info.get("modelinfo") or {}
        except Exception:
            logger.debug("Could not query max context over Ollama HTTP for model %s", name, exc_info=True)
            metadata = {}
    arch = str(metadata.get("general.architecture") or "")
    ctx = metadata.get(f"{arch}.context_length") if arch else None
    if ctx is None:
        for key, value in metadata.items():
            if str(key).endswith(".context_length"):
                ctx = value
                break
    try:
        _model_max_ctx_cache[name] = int(ctx) if ctx is not None else None
    except (TypeError, ValueError):
        _model_max_ctx_cache[name] = None
    return _model_max_ctx_cache[name]


def set_model(model_name: str):
    """Switch the active model.  Accepts both local Ollama and cloud model IDs.

    For local models: unloads the previous model from Ollama's VRAM.
    For cloud models: just updates the setting (no Ollama interaction).
    """
    global _current_model, _llm_instance
    if model_name != _current_model:
        logger.info("Switching model: %s → %s", _current_model, model_name)
        # Unload previous local model from Ollama memory
        client = _ollama_client()
        if not is_cloud_model(_current_model) and client:
            try:
                client.generate(model=_runtime_model_name(_current_model), prompt="", keep_alive=0)
            except Exception:
                logger.debug("Could not unload previous model %s", _current_model, exc_info=True)
    _current_model = model_name
    if is_cloud_model(model_name):
        _llm_instance = _get_cloud_llm(model_name)
    else:
        _llm_instance = _chat_ollama(
            model=_runtime_model_name(model_name),
            num_ctx=_local_num_ctx_for(model_name),
        )
    _save_settings({"model": _current_model, "context_size": _num_ctx,
                    "cloud_context_size": _cloud_num_ctx})


# Thread-local model override — allows agent.py to propagate the per-thread
# cloud model override so that get_context_size() (and everything downstream:
# get_tool_budget, _keep_browser_snapshots, tool budgets) automatically uses
# the correct context window without every caller needing an explicit argument.
_active_model_override: contextvars.ContextVar[str] = contextvars.ContextVar(
    "active_model_override", default=""
)


def set_active_model_override(name: str) -> None:
    """Set the thread-local model override (called by agent.py before execution)."""
    _active_model_override.set(name)


def _resolved_context_identity(model_name: str):
    try:
        from providers.resolution import resolve_provider_config

        return resolve_provider_config(model_name, allow_legacy_local=True)
    except Exception:
        return None


def get_context_policy(model_name: str | None = None) -> ContextPolicy:
    """Return the resolved context policy for the given (or active) model."""
    name = model_name or _active_model_override.get() or _current_model
    resolved = _resolved_context_identity(name)
    provider_id = resolved.provider_id if resolved else (get_cloud_provider(name) or "ollama")
    runtime_model = resolved.runtime_model if resolved else _runtime_model_name(name)
    model_ref_value = resolved.selection_ref if resolved else name
    remote_policy = bool(resolved and resolved.execution_location != "local") or is_cloud_model(name)
    if resolved and resolved.execution_location == "local" and provider_id.startswith("custom_openai_"):
        remote_policy = False

    user_cap = _coerce_context_size(
        _cloud_num_ctx if remote_policy else _num_ctx,
        DEFAULT_CLOUD_CONTEXT_SIZE if remote_policy else DEFAULT_CONTEXT_SIZE,
        minimum=(CLOUD_CONTEXT_SIZE_OPTIONS[0] if remote_policy else CONTEXT_SIZE_OPTIONS[0]),
    )
    model_max = get_model_max_context(name)
    model_max_int = _coerce_context_size(model_max, 0) if model_max else 0
    native_max = model_max_int if model_max_int > 0 else None
    cap_source = "provider_metadata" if native_max else "unknown"
    if native_max is None and remote_policy:
        native_max = _estimate_context_heuristic(runtime_model)
        cap_source = "heuristic"
    if native_max is None and provider_id.startswith("custom_openai_"):
        endpoint = resolved.endpoint if resolved else {}
        fallback_context = int(endpoint.get("unknown_context_fallback") or 0) if isinstance(endpoint, dict) else 0
        if fallback_context > 0:
            native_max = fallback_context
            cap_source = "profile_default"

    effective = int(min(user_cap, native_max) if native_max else user_cap)
    if provider_id == "ollama" and not remote_policy:
        request_application = "ollama_num_ctx"
    elif provider_id.startswith("custom_openai_"):
        endpoint = resolved.endpoint if resolved else {}
        context_param = str(endpoint.get("context_param_name") or "")
        if endpoint.get("supports_runtime_context_override") and context_param:
            request_application = f"request_param:{context_param}"
        else:
            request_application = "trim_only"
    else:
        request_application = "trim_only"

    return ContextPolicy(
        model_ref=model_ref_value,
        provider_id=provider_id,
        runtime_model=runtime_model,
        native_max=int(native_max) if native_max else None,
        user_cap=int(user_cap),
        effective_context=effective,
        policy_kind="provider" if remote_policy else "local",
        cap_source=cap_source,
        request_application=request_application,
    )


def get_context_size(model_name: str | None = None) -> int:
    """Return the *effective* context size for the given (or current) model.

    - **Cloud models** use ``min(user_cloud_cap, model_native_max)``.
      The user-configurable cap reduces cost and rate-limit pressure.
    - **Local (Ollama) models** use ``min(user_setting, model_native_max)``
      because ``num_ctx`` directly controls VRAM usage.

    Resolution order for the model name:
    1. Explicit *model_name* argument.
    2. Thread-local ``_active_model_override`` (set by agent.py).
    3. Global ``_current_model``.
    """
    return get_context_policy(model_name).effective_context


def get_tool_budget(fraction: float, *,
                    floor: int = 10_000, ceiling: int = 200_000) -> int:
    """Dynamic char budget for a tool result, scaled to the model's context.

    Returns a *character* limit suitable for string slicing.
    Assumes ~3 chars per token for the tokens-to-chars conversion.
    Clamped between *floor* and *ceiling* to prevent extremes.
    """
    ctx = get_context_size()
    return min(ceiling, max(floor, int(ctx * fraction * 3)))


def get_user_context_size() -> int:
    """Return the raw user-selected context size (before model capping)."""
    return _coerce_context_size(_num_ctx, DEFAULT_CONTEXT_SIZE, minimum=CONTEXT_SIZE_OPTIONS[0])


def get_cloud_context_size() -> int:
    """Return the raw user-selected cloud context cap."""
    return _coerce_context_size(
        _cloud_num_ctx,
        DEFAULT_CLOUD_CONTEXT_SIZE,
        minimum=CLOUD_CONTEXT_SIZE_OPTIONS[0],
    )


def set_cloud_context_size(size: int):
    """Change the cloud context cap and recreate the LLM instance."""
    global _cloud_num_ctx, _llm_instance
    coerced = _coerce_context_size(size, DEFAULT_CLOUD_CONTEXT_SIZE, allowed=CLOUD_CONTEXT_SIZE_OPTIONS)
    logger.info("Cloud context size changed: %s → %s", _cloud_num_ctx, coerced)
    _cloud_num_ctx = coerced
    _override_llm_cache.clear()
    if is_cloud_model(_current_model):
        _llm_instance = _get_cloud_llm(_current_model)
    _save_settings({"model": _current_model, "context_size": _num_ctx,
                    "cloud_context_size": _cloud_num_ctx})


def set_context_size(size: int):
    """Change the context window size and recreate the LLM instance."""
    global _num_ctx, _llm_instance
    coerced = _coerce_context_size(size, DEFAULT_CONTEXT_SIZE, allowed=CONTEXT_SIZE_OPTIONS)
    logger.info("Context size changed: %s → %s", _num_ctx, coerced)
    _num_ctx = coerced
    _override_llm_cache.clear()
    if is_cloud_model(_current_model):
        _llm_instance = _get_cloud_llm(_current_model)
    else:
        _llm_instance = _chat_ollama(
            model=_runtime_model_name(_current_model),
            num_ctx=_local_num_ctx_for(_current_model),
        )
    _save_settings({"model": _current_model, "context_size": _num_ctx,
                    "cloud_context_size": _cloud_num_ctx})


def get_current_model() -> str:
    _reset_current_model_if_missing_custom_provider()
    return _current_model


def _ollama_host_port() -> tuple[str, int]:
    """Return the TCP host/port configured for the local Ollama daemon."""
    host, port, _scheme = _ollama_endpoint_parts()
    return host, port


def _ollama_reachable(timeout: float = 1.0) -> bool:
    """Fast TCP probe to check if the Ollama server is listening."""
    import socket
    host, port = _ollama_host_port()
    try:
        with socket.create_connection((host, port), timeout=timeout):
            return True
    except (OSError, TimeoutError):
        return False


def list_local_models() -> list[str]:
    """Return names of models currently exposed by the Ollama daemon."""
    if not _ollama_reachable():
        return []
    names: set[str] = set()
    client = _ollama_client()
    try:
        if client:
            response = client.list()
            for model in getattr(response, "models", []) or []:
                name = getattr(model, "model", None) or getattr(model, "name", None)
                if isinstance(model, dict):
                    name = model.get("model") or model.get("name") or name
                if name:
                    names.add(str(name))
    except Exception:
        logger.debug("Could not list local Ollama models", exc_info=True)
    if not names:
        try:
            response = _ollama_http_json("/api/tags")
            for model in response.get("models", []) or []:
                if not isinstance(model, dict):
                    continue
                name = model.get("model") or model.get("name")
                if name:
                    names.add(str(name))
        except Exception:
            logger.debug("Could not list local Ollama models over HTTP", exc_info=True)
    return sorted(names)


def list_all_models() -> list[str]:
    """Return models currently exposed by the Ollama daemon."""
    return sorted(set(list_local_models()))


def get_trending_models() -> list[str]:
    """Return no public Ollama discovery rows; Thoth manages daemon models only."""
    return []


def fetch_trending_ollama_models() -> list[str]:
    """Deprecated compatibility shim.

    Thoth no longer fetches public Ollama model listings or offers downloads.
    Local model management lives in Ollama; this app only reads daemon-exposed
    tags.
    """
    global _trending_ollama_cache, _trending_fetched
    _trending_ollama_cache = []
    _trending_fetched = True
    return []


def is_model_local(model_name: str) -> bool:
    """Check whether a model is already downloaded."""
    parsed = _parse_provider_model_ref(model_name)
    if parsed and parsed[0] not in {"local", "ollama"}:
        return False
    runtime_model = _runtime_model_name(model_name)
    local = list_local_models()
    return any(
        runtime_model == m
        or f"{runtime_model}:latest" == m
        or runtime_model == m.split(":")[0]
        for m in local
    )


def _looks_like_cloud_model(model_name: str) -> bool:
    """Heuristic: return True if *model_name* looks like a cloud model.

    Only used as a last-resort fallback when the persisted cache is empty
    AND the in-memory cache hasn't been populated yet.
    """
    if "/" in model_name:                       # OpenRouter format: provider/model
        return True
    if any(model_name.startswith(p) for p in _OPENAI_CHAT_PREFIXES):
        # Exclude known Ollama models that happen to share a prefix
        family = model_name.split(":")[0]
        if family in _TOOL_COMPATIBLE_FAMILIES:
            return False
        return True
    if model_name.split("/")[-1].lower().startswith("minimax"):
        return True
    return False


def _infer_cloud_provider(model_name: str) -> str | None:
    """Infer a cloud provider from a model ID when the cache is unavailable."""
    parsed = _parse_provider_model_ref(model_name)
    if parsed:
        provider_id, _model_id = parsed
        if provider_id == "ollama":
            try:
                from providers.ollama import is_ollama_cloud_offload_model
                return "ollama" if is_ollama_cloud_offload_model(_model_id) else None
            except Exception:
                return None
        return None if provider_id == "local" else provider_id
    model_name = _runtime_model_name(model_name)
    try:
        from providers.ollama import is_ollama_cloud_offload_model
        if is_ollama_cloud_offload_model(model_name):
            return "ollama"
    except Exception:
        pass
    _sync_custom_model_cache()
    if model_name in _cloud_model_cache:
        return _cloud_model_cache[model_name]["provider"]
    if "/" in model_name:
        return "openrouter"

    bare_name = model_name.split("/")[-1]
    if any(bare_name.startswith(prefix) for prefix in _OPENAI_CHAT_PREFIXES):
        family = bare_name.split(":")[0]
        if family not in _TOOL_COMPATIBLE_FAMILIES:
            return "openai"
    if bare_name.startswith("claude"):
        return "anthropic"
    if bare_name.startswith("gemini"):
        return "google"
    if bare_name.startswith("grok"):
        return "xai"
    if bare_name.lower().startswith("minimax"):
        return "minimax"
    return None


def is_cloud_model(model_name: str) -> bool:
    """Return True if *model_name* is a known cloud model.

    Uses the persisted cache (loaded from disk at startup), then falls back
    to provider-prefix inference so a saved cloud default survives cache or
    key outages.
    """
    return _infer_cloud_provider(model_name) is not None


def get_cloud_provider(model_name: str) -> str | None:
    """Return the cloud provider id for a model, or ``None`` for local models."""
    return _infer_cloud_provider(model_name)


# ── Provider emoji mapping ───────────────────────────────────────────────────
_PROVIDER_EMOJI: dict[str | None, str] = {
    "openai": "⬡",
    "ollama": "☁️",
    "ollama_cloud": "☁️",
    "codex": "C",
    "opencode_zen": "OZ",
    "opencode_go": "OG",
    "openrouter": "🌐",
    "anthropic": "🔶",
    "google": "💎",
    "xai": "𝕏",
    "minimax": "M",
    None: "☁️",  # fallback for unknown cloud
}


def get_provider_emoji(model_name: str) -> str:
    """Return a provider-specific emoji for a model.

    Local models get 🖥️, cloud models get a provider-specific icon.
    """
    if not is_cloud_model(model_name):
        return "🖥️"
    prov = get_cloud_provider(model_name)
    return _PROVIDER_EMOJI.get(prov, _PROVIDER_EMOJI[None])


def is_cloud_available() -> bool:
    """Return True if any cloud API key is configured."""
    from providers.runtime import list_configured_provider_ids
    return bool(list_configured_provider_ids())


def is_openai_available() -> bool:
    """Return True if an OpenAI API key is configured."""
    from api_keys import get_key
    return bool(get_key("OPENAI_API_KEY"))


def is_openrouter_available() -> bool:
    """Return True if an OpenRouter API key is configured."""
    from api_keys import get_key
    return bool(get_key("OPENROUTER_API_KEY"))


def is_ollama_cloud_available() -> bool:
    """Return True if an Ollama Cloud API key is configured."""
    from api_keys import get_key
    return bool(get_key("OLLAMA_API_KEY"))


def is_anthropic_available() -> bool:
    """Return True if an Anthropic API key is configured."""
    from api_keys import get_key
    return bool(get_key("ANTHROPIC_API_KEY"))


def is_google_available() -> bool:
    """Return True if a Google AI API key is configured."""
    from api_keys import get_key
    return bool(get_key("GOOGLE_API_KEY"))


def is_xai_available() -> bool:
    """Return True if an xAI API key is configured."""
    from api_keys import get_key
    return bool(get_key("XAI_API_KEY"))


def is_minimax_available() -> bool:
    """Return True if a MiniMax API key is configured."""
    from api_keys import get_key
    return bool(get_key("MINIMAX_API_KEY"))


def list_cloud_models(provider: str | None = None) -> list[str]:
    """Return cached cloud model IDs, optionally filtered by provider."""
    _sync_custom_model_cache()
    if provider:
        models = [m for m, info in _cloud_model_cache.items() if info["provider"] == provider]
        if provider == "codex":
            try:
                from providers.codex import list_codex_model_infos
                seen = set(models)
                for model_info in list_codex_model_infos():
                    if model_info.model_id not in seen:
                        models.append(model_info.model_id)
                        seen.add(model_info.model_id)
            except Exception:
                pass
        return models
    return list(_cloud_model_cache.keys())


def _sync_custom_model_cache() -> None:
    try:
        from providers.custom import custom_model_cache_entries
        entries = custom_model_cache_entries()
    except Exception:
        return
    with _cloud_cache_lock:
        for model_id, info in list(_cloud_model_cache.items()):
            if isinstance(info, dict) and str(info.get("provider") or "").startswith("custom_openai_"):
                replacement = entries.get(model_id)
                if not replacement or replacement.get("provider") != info.get("provider"):
                    _cloud_model_cache.pop(model_id, None)
        _cloud_model_cache.update(entries)


def reset_current_model_if_removed(
    provider_id: str,
    *,
    removed_model_ids: set[str] | None = None,
) -> bool:
    """Reset the saved Brain default if it points at a removed provider/model."""
    global _current_model, _llm_instance
    provider_id = str(provider_id or "").strip()
    if not provider_id:
        return False
    try:
        from providers.selection import list_quick_choices, model_choice_value, parse_model_ref
    except Exception:
        return False
    parsed = parse_model_ref(_current_model)
    if not parsed:
        return False
    current_provider, current_model = parsed
    removed = {str(model_id) for model_id in (removed_model_ids or set()) if str(model_id)}
    if current_provider != provider_id:
        return False
    if removed and current_model not in removed:
        return False
    fallback = ""
    try:
        for choice in list_quick_choices("chat"):
            if choice.get("kind") != "model" or choice.get("active") is False:
                continue
            choice_provider = str(choice.get("provider_id") or "")
            choice_model = str(choice.get("model_id") or "")
            if choice_provider == provider_id and (not removed or choice_model in removed):
                continue
            fallback = model_choice_value(choice_model, provider_id=choice_provider)
            if fallback:
                break
    except Exception:
        fallback = ""
    if not fallback:
        fallback = model_choice_value(DEFAULT_MODEL, provider_id="ollama")
    _current_model = fallback
    _llm_instance = None
    _override_llm_cache.clear()
    _save_settings({"model": _current_model, "context_size": _num_ctx,
                    "cloud_context_size": _cloud_num_ctx})
    return True


def _reset_current_model_if_missing_custom_provider() -> None:
    try:
        from providers.custom import get_custom_endpoint, is_custom_openai_provider
        from providers.selection import parse_model_ref
    except Exception:
        return
    parsed = parse_model_ref(_current_model)
    if not parsed:
        return
    provider_id, _model_id = parsed
    if is_custom_openai_provider(provider_id) and not get_custom_endpoint(provider_id):
        reset_current_model_if_removed(provider_id)


def list_starred_cloud_models() -> list[str]:
    """Return cloud models the user has starred (for the thread picker)."""
    from providers.selection import list_quick_model_ids, migrate_legacy_starred_models
    from api_keys import get_cloud_config
    _sync_custom_model_cache()
    migrate_legacy_starred_models(cloud_models=_cloud_model_cache.keys())
    quick_models = [m for m in list_quick_model_ids("chat") if m in _cloud_model_cache]
    if quick_models:
        return quick_models
    starred = set(get_cloud_config().get("starred_models", []))
    return [m for m in _cloud_model_cache if m in starred]


def star_cloud_model(model_id: str) -> None:
    """Add a model to the starred list."""
    from api_keys import get_cloud_config, set_cloud_config
    from providers.selection import add_quick_choice_for_model
    starred = list(get_cloud_config().get("starred_models", []))
    if model_id not in starred:
        starred.append(model_id)
        set_cloud_config("starred_models", starred)
    add_quick_choice_for_model(model_id, source="legacy_starred_cloud")


def unstar_cloud_model(model_id: str) -> None:
    """Remove a model from the starred list."""
    from api_keys import get_cloud_config, set_cloud_config
    from providers.selection import remove_quick_choice_for_model
    starred = list(get_cloud_config().get("starred_models", []))
    if model_id in starred:
        starred.remove(model_id)
        set_cloud_config("starred_models", starred)
    remove_quick_choice_for_model(model_id)


def get_cloud_model_context(model_name: str) -> int:
    """Return the context window size for a cloud model.

    Resolution order:
    1. Cached value (from OpenRouter API or previous fetch).
    2. Context catalog (public OpenRouter data, no key needed).
    3. Prefix-based heuristic covering OpenAI / Anthropic / Gemini.
    4. Safe fallback (256K).
    """
    parsed = _parse_provider_model_ref(model_name)
    provider_id = parsed[0] if parsed else None
    runtime_model = _runtime_model_name(model_name)
    _sync_custom_model_cache()
    info = _cloud_cache_entry_for(model_name, provider_id)
    if info and (not provider_id or info.get("provider") == provider_id):
        cached_ctx = int(info.get("ctx") or 0)
        if cached_ctx > 0:
            return cached_ctx
    if provider_id == "codex":
        try:
            from providers.codex import list_codex_model_infos

            for model_info in list_codex_model_infos():
                if model_info.model_id == runtime_model and model_info.context_window:
                    return int(model_info.context_window)
        except Exception:
            pass
    return _catalog_or_heuristic(runtime_model)


def list_cloud_vision_models() -> list[str]:
    """Return cloud model IDs that support vision / image input."""
    from providers.capabilities import snapshot_supports_surface
    vision_models = [
        m for m, info in _cloud_model_cache.items()
        if info.get("vision")
        or (isinstance(info.get("capabilities_snapshot"), dict) and bool(info.get("capabilities_snapshot"))
            and snapshot_supports_surface(info.get("capabilities_snapshot"), "vision"))
    ]
    try:
        from providers.codex import list_codex_model_infos
        seen = set(vision_models)
        for model_info in list_codex_model_infos():
            if model_info.model_id not in seen and snapshot_supports_surface(model_info.capability_snapshot(), "vision"):
                vision_models.append(model_info.model_id)
                seen.add(model_info.model_id)
    except Exception:
        pass
    return vision_models


def is_cloud_vision_model(model_name: str) -> bool:
    """Return True if *model_name* is a cloud model with vision support."""
    parsed = _parse_provider_model_ref(model_name)
    runtime_model = _runtime_model_name(model_name)
    if parsed and parsed[0] == "codex":
        try:
            from providers.capabilities import snapshot_supports_surface
            from providers.codex import list_codex_model_infos
            return any(
                model_info.model_id == runtime_model
                and snapshot_supports_surface(model_info.capability_snapshot(), "vision")
                for model_info in list_codex_model_infos()
            )
        except Exception:
            return False
    info = _cloud_cache_entry_for(model_name, parsed[0] if parsed else None)
    if not info:
        return False
    from providers.capabilities import snapshot_supports_surface
    snapshot = info.get("capabilities_snapshot")
    return bool(
        info.get("vision")
        or (isinstance(snapshot, dict) and bool(snapshot) and snapshot_supports_surface(snapshot, "vision"))
    )


def _get_cloud_llm(model_name: str):
    """Return a cached LLM for a cloud model.

    OpenAI-direct models use ``ChatOpenAI``.  OpenRouter models use
    ``ChatOpenRouter`` which correctly surfaces ``reasoning_content``
    in ``additional_kwargs`` for reasoning models.
    """
    from providers.runtime import create_chat_model

    provider = get_cloud_provider(model_name)
    runtime_model = _runtime_model_name(model_name)
    ctx = get_cloud_model_context(model_name)
    key = (f"{provider or 'openrouter'}:{runtime_model}", ctx)
    if key in _override_llm_cache:
        return _override_llm_cache[key]

    logger.info("Creating provider LLM: model=%s via %s", runtime_model, provider or "openrouter")

    _override_llm_cache[key] = create_chat_model(runtime_model, provider)
    return _override_llm_cache[key]


# ── Dynamic model fetching ──────────────────────────────────────────────────


def fetch_context_catalog() -> int:
    """Fetch the public OpenRouter model catalog (no API key needed).

    Populates ``_context_catalog`` with ``model_id → context_length``
    for every model OpenRouter lists.  Returns the number of entries.
    Safe to call from background threads.
    """
    import httpx

    try:
        resp = httpx.get(f"{OPENROUTER_BASE_URL}/models", timeout=15)
        resp.raise_for_status()
        data = resp.json().get("data", [])
    except Exception as exc:
        logger.warning("Failed to fetch context catalog: %s", exc)
        return 0

    count = 0
    with _context_catalog_lock:
        _context_catalog.clear()
        for m in data:
            mid = m.get("id", "")
            ctx = m.get("context_length")
            if mid and ctx and isinstance(ctx, int) and ctx > 0:
                _context_catalog[mid] = ctx
                count += 1
    _save_context_catalog()
    logger.info("Context catalog: %d models indexed", count)
    return count


def validate_openrouter_key(api_key: str) -> bool:
    """Validate an OpenRouter API key by hitting the auth endpoint.

    Returns True if the key is accepted, False otherwise.
    """
    import httpx

    try:
        resp = httpx.get(
            f"{OPENROUTER_BASE_URL}/auth/key",
            headers={"Authorization": f"Bearer {api_key}"},
            timeout=10,
        )
        return resp.status_code == 200
    except Exception:
        return False


def validate_ollama_cloud_key(api_key: str) -> bool:
    """Validate an Ollama Cloud API key with a tiny authenticated chat probe."""
    import httpx
    from providers.transports.ollama_cloud import normalize_ollama_cloud_api_key

    clean_key = normalize_ollama_cloud_api_key(api_key)
    if not clean_key:
        return False

    try:
        probe_model = "gpt-oss:20b"
        resp = httpx.post(
            f"{OLLAMA_CLOUD_BASE_URL}/api/chat",
            headers={"Authorization": f"Bearer {clean_key}"},
            json={
                "model": probe_model,
                "messages": [{"role": "user", "content": "ok"}],
                "stream": False,
                "options": {"num_predict": 1},
            },
            timeout=20,
        )
        if resp.status_code == 200:
            return True
        logger.warning("Ollama Cloud key validation: %d - %s", resp.status_code, resp.text[:200])
        return False
    except Exception as exc:
        logger.warning("Ollama Cloud key validation error: %s", exc)
        return False


def validate_anthropic_key(api_key: str) -> bool:
    """Validate an Anthropic API key by listing models.

    Returns True if the key is accepted, False otherwise.
    """
    import httpx

    try:
        resp = httpx.get(
            f"{ANTHROPIC_BASE_URL}/models",
            headers={
                "x-api-key": api_key,
                "anthropic-version": "2023-06-01",
            },
            params={"limit": 1},
            timeout=10,
        )
        return resp.status_code == 200
    except Exception:
        return False


def validate_google_key(api_key: str) -> bool:
    """Validate a Google AI API key by listing models.

    Returns True if the key is accepted, False otherwise.
    """
    import httpx

    try:
        resp = httpx.get(
            f"{GOOGLE_GENAI_BASE_URL}/models",
            params={"key": api_key, "pageSize": 1},
            timeout=10,
        )
        return resp.status_code == 200
    except Exception:
        return False


def validate_xai_key(api_key: str) -> bool:
    """Validate an xAI API key by attempting to list models.

    Returns True if the key is accepted, False otherwise.
    Uses the same ``_fetch_xai_models`` path to validate: if at least one
    model can be fetched, the key is valid.
    """
    import httpx

    try:
        resp = httpx.get(
            f"{XAI_BASE_URL}/language-models",
            headers={"Authorization": f"Bearer {api_key}"},
            timeout=10,
        )
        if resp.status_code == 200:
            return True
        logger.warning("xAI key validation: %d — %s", resp.status_code, resp.text[:200])
        return False
    except Exception as exc:
        logger.warning("xAI key validation error: %s", exc)
        return False


def validate_minimax_key(api_key: str) -> bool:
    """Validate a MiniMax API key with the Anthropic-compatible messages API.

    MiniMax's Anthropic-compatible docs do not expose a model-list endpoint, so
    validation uses the smallest possible message request. MiniMax reports a
    valid key with no available balance as ``insufficient balance (1008)``;
    treat that as accepted credentials so the UI does not mislabel the key as
    invalid.
    """
    import httpx

    try:
        resp = httpx.post(
            f"{MINIMAX_ANTHROPIC_BASE_URL}/v1/messages",
            headers={
                "x-api-key": api_key,
                "anthropic-version": "2023-06-01",
                "content-type": "application/json",
            },
            json={
                "model": "MiniMax-M2.7",
                "max_tokens": 1,
                "messages": [{"role": "user", "content": [{"type": "text", "text": "Hi"}]}],
            },
            timeout=15,
        )
        if resp.status_code == 200:
            return True
        body = resp.text[:500]
        lower_body = body.lower()
        if "insufficient balance" in lower_body or "1008" in lower_body:
            logger.warning(
                "MiniMax key validation accepted credentials but account is not billable: %d — %s",
                resp.status_code,
                resp.text[:200],
            )
            return True
        logger.warning("MiniMax key validation: %d — %s", resp.status_code, resp.text[:200])
        return False
    except Exception as exc:
        logger.warning("MiniMax key validation error: %s", exc)
        return False


def _catalog_or_heuristic(model_id: str) -> int:
    """Resolve context size for any model via catalog → heuristic → fallback.

    Checks ``_context_catalog`` first (public OpenRouter data, covers all
    providers).  Falls back to prefix heuristic, then ``_CLOUD_CONTEXT_FALLBACK``.
    """
    # Try exact match in catalog (e.g. "openai/gpt-4o")
    with _context_catalog_lock:
        cat_val = _context_catalog.get(f"openai/{model_id}")
        if cat_val:
            return cat_val
        # Also try bare ID for OpenRouter-style keys
        cat_val = _context_catalog.get(model_id)
        if cat_val:
            return cat_val
    return _estimate_context_heuristic(model_id)


# Anthropic model ID substrings to skip (non-chat or internal models)
_ANTHROPIC_SKIP_SUBSTRINGS = ("embed", "tokenizer")

# Google model substrings to skip
_GOOGLE_SKIP_SUBSTRINGS = ("embed", "aqa", "tts")


def _fetch_ollama_cloud_models(api_key: str) -> int:
    """Fetch models from Ollama Cloud's native ``/api/tags`` endpoint."""
    import httpx
    from providers.capabilities import model_supports_surface
    from providers.catalog import model_info_from_metadata, model_info_to_cache_entry
    from providers.transports.ollama_cloud import normalize_ollama_cloud_api_key

    clean_key = normalize_ollama_cloud_api_key(api_key)

    try:
        resp = httpx.get(
            f"{OLLAMA_CLOUD_BASE_URL}/api/tags",
            headers={"Authorization": f"Bearer {clean_key}"} if clean_key else {},
            timeout=15,
        )
        resp.raise_for_status()
        models = resp.json().get("models", [])
        if not isinstance(models, list):
            logger.warning("Ollama Cloud model fetch returned invalid models payload")
            return 0
        if len(models) > 500:
            logger.warning("Ollama Cloud model fetch returned %d entries; limiting to first 500", len(models))
            models = models[:500]
    except Exception as exc:
        logger.warning("Failed to fetch ollama_cloud models: %s", exc)
        return 0

    count = 0
    with _cloud_cache_lock:
        for m in models:
            if not isinstance(m, dict):
                continue
            mid = str(m.get("name") or m.get("model") or "").strip()
            if not mid:
                continue
            metadata = dict(m)
            details = m.get("details")
            if isinstance(details, dict):
                metadata.update({f"details_{key}": value for key, value in details.items()})
            ctx = int(m.get("context_length") or m.get("context_window") or 0)
            if ctx <= 0:
                ctx = _catalog_or_heuristic(mid)
            model_info = model_info_from_metadata(
                "ollama_cloud",
                mid,
                metadata,
                display_name=mid,
                context_window=ctx,
                source="ollama_cloud_catalog",
            )
            if not model_supports_surface(model_info, "chat"):
                continue
            _cloud_model_cache[mid] = model_info_to_cache_entry(model_info)
            count += 1
    logger.info("Fetched %d ollama_cloud models", count)
    return count


def fetch_cloud_models(provider: str) -> int:
    """Fetch available models from *provider*.

    Supported providers: ``'openai'``, ``'ollama_cloud'``, ``'openrouter'``, ``'anthropic'``,
    ``'google'``, ``'xai'``, and ``'minimax'``.  Populates ``_cloud_model_cache``.  Returns the number
    of models found.  Safe to call from background threads.
    """
    import httpx
    from api_keys import get_key
    from providers.capabilities import model_supports_surface
    from providers.catalog import model_info_from_metadata, model_info_to_cache_entry

    if provider == "openai":
        api_key = get_key("OPENAI_API_KEY")
        if not api_key:
            return 0
        url = f"{OPENAI_BASE_URL}/models"
        headers = {"Authorization": f"Bearer {api_key}"}
    elif provider == "ollama_cloud":
        api_key = get_key("OLLAMA_API_KEY")
        if not api_key:
            return 0
        return _fetch_ollama_cloud_models(api_key)
    elif provider == "openrouter":
        api_key = get_key("OPENROUTER_API_KEY")
        if not api_key:
            return 0
        url = f"{OPENROUTER_BASE_URL}/models"
        headers = {"Authorization": f"Bearer {api_key}"}
    elif provider == "anthropic":
        api_key = get_key("ANTHROPIC_API_KEY")
        if not api_key:
            return 0
        return _fetch_anthropic_models(api_key)
    elif provider == "google":
        api_key = get_key("GOOGLE_API_KEY")
        if not api_key:
            return 0
        return _fetch_google_models(api_key)
    elif provider == "xai":
        api_key = get_key("XAI_API_KEY")
        if not api_key:
            return 0
        return _fetch_xai_models(api_key)
    elif provider == "minimax":
        api_key = get_key("MINIMAX_API_KEY")
        if not api_key:
            return 0
        return _fetch_minimax_models(api_key)
    elif provider in {"opencode_zen", "opencode_go"}:
        from providers.auth_store import get_provider_secret

        api_key = get_provider_secret(provider)
        if not api_key:
            return 0
        return _fetch_opencode_models(provider)
    else:
        return 0

    try:
        resp = httpx.get(url, headers=headers, timeout=15)
        resp.raise_for_status()
        data = resp.json().get("data", [])
    except Exception as exc:
        logger.warning("Failed to fetch %s models: %s", provider, exc)
        return 0

    count = 0
    with _cloud_cache_lock:
        if provider == "openai":
            for m in data:
                mid = m.get("id", "")
                model_info = model_info_from_metadata(
                    "openai", mid, m,
                    display_name=mid,
                    context_window=_catalog_or_heuristic(mid),
                )
                if not any(model_supports_surface(model_info, surface) for surface in ("chat", "image", "video")):
                    continue
                entry = model_info_to_cache_entry(model_info)
                _cloud_model_cache[mid] = entry
                count += 1
        else:  # openrouter
            for m in data:
                mid = m.get("id", "")
                name = m.get("name", mid)
                ctx = m.get("context_length", 128_000)
                # Only include models with a '/' (provider/model format)
                if "/" not in mid:
                    continue
                model_info = model_info_from_metadata(
                    "openrouter", mid, m,
                    display_name=name,
                    context_window=ctx,
                )
                if not any(model_supports_surface(model_info, surface) for surface in ("chat", "image", "video")):
                    continue
                _cloud_model_cache[mid] = model_info_to_cache_entry(model_info)
                count += 1
    logger.info("Fetched %d %s models", count, provider)
    return count


def _fetch_anthropic_models(api_key: str) -> int:
    """Fetch models from the Anthropic ``/v1/models`` endpoint.

    Handles pagination via ``after_id``.  Uses ``max_input_tokens`` for
    context size when available, falling back to ``_catalog_or_heuristic``.
    """
    import httpx
    from providers.catalog import model_info_from_metadata, model_info_to_cache_entry

    headers = {
        "x-api-key": api_key,
        "anthropic-version": "2023-06-01",
    }
    count = 0
    after_id: str | None = None

    try:
        while True:
            params: dict[str, str | int] = {"limit": 100}
            if after_id:
                params["after_id"] = after_id
            resp = httpx.get(
                f"{ANTHROPIC_BASE_URL}/models",
                headers=headers, params=params, timeout=15,
            )
            resp.raise_for_status()
            body = resp.json()
            models = body.get("data", [])

            with _cloud_cache_lock:
                for m in models:
                    mid = m.get("id", "")
                    if not mid:
                        continue
                    if any(s in mid for s in _ANTHROPIC_SKIP_SUBSTRINGS):
                        continue
                    display = m.get("display_name", mid)
                    # Use API context size if available; fall back to catalog/heuristic
                    api_ctx = m.get("max_input_tokens", 0)
                    ctx = api_ctx if api_ctx and api_ctx > 0 else _catalog_or_heuristic(mid)
                    # Vision from capabilities
                    caps = m.get("capabilities", {})
                    has_vision = bool(
                        caps.get("image_input", {}).get("supported")
                    )
                    metadata = dict(m)
                    metadata["vision"] = has_vision
                    _cloud_model_cache[mid] = model_info_to_cache_entry(model_info_from_metadata(
                        "anthropic", mid, metadata,
                        display_name=display,
                        context_window=ctx,
                    ))
                    count += 1

            if not body.get("has_more"):
                break
            after_id = body.get("last_id")
            if not after_id:
                break
    except Exception as exc:
        logger.warning("Failed to fetch anthropic models: %s", exc)

    logger.info("Fetched %d anthropic models", count)
    return count


def _fetch_google_models(api_key: str) -> int:
    """Fetch models from the Google Generative AI ``models.list`` endpoint.

    Includes chat, image, and video-capable models. Media generation models are
    often exposed through non-chat APIs, so they must stay in the provider cache
    for the image/video tools even when they are hidden from chat pickers.
    Uses ``inputTokenLimit`` for context size.
    """
    import httpx
    from providers.capabilities import model_supports_surface
    from providers.catalog import model_info_from_metadata, model_info_to_cache_entry

    count = 0
    page_token: str | None = None

    try:
        while True:
            params: dict[str, str | int] = {"key": api_key, "pageSize": 100}
            if page_token:
                params["pageToken"] = page_token
            resp = httpx.get(
                f"{GOOGLE_GENAI_BASE_URL}/models",
                params=params, timeout=15,
            )
            resp.raise_for_status()
            body = resp.json()
            models = body.get("models", [])

            with _cloud_cache_lock:
                for m in models:
                    name = m.get("name", "")        # e.g. "models/gemini-2.5-flash"
                    mid = name.removeprefix("models/")  # → "gemini-2.5-flash"
                    if not mid:
                        continue
                    methods = m.get("supportedGenerationMethods", [])
                    if any(s in mid for s in _GOOGLE_SKIP_SUBSTRINGS):
                        continue
                    display = m.get("displayName", mid)
                    ctx = m.get("inputTokenLimit", 0)
                    if not ctx or ctx <= 0:
                        ctx = _catalog_or_heuristic(mid)
                    metadata = dict(m)
                    metadata["supportedGenerationMethods"] = methods
                    metadata["vision"] = "generateContent" in methods
                    model_info = model_info_from_metadata("google", mid, metadata, display_name=display, context_window=ctx)
                    if not any(model_supports_surface(model_info, surface) for surface in ("chat", "image", "video")):
                        continue
                    _cloud_model_cache[mid] = model_info_to_cache_entry(model_info)
                    count += 1

            page_token = body.get("nextPageToken")
            if not page_token:
                break
    except Exception as exc:
        logger.warning("Failed to fetch google models: %s", exc)

    logger.info("Fetched %d google models", count)
    return count


# Substrings that mark non-chat xAI models to drop from the provider cache.
_XAI_SKIP_SUBSTRINGS: tuple[str, ...] = ()

_MINIMAX_SUPPORTED_MODELS: tuple[tuple[str, int], ...] = (
    ("MiniMax-M2.7", 204_800),
    ("MiniMax-M2.7-highspeed", 204_800),
    ("MiniMax-M2.5", 204_800),
    ("MiniMax-M2.5-highspeed", 204_800),
    ("MiniMax-M2.1", 204_800),
    ("MiniMax-M2.1-highspeed", 204_800),
    ("MiniMax-M2", 204_800),
)


def _fetch_xai_models(api_key: str) -> int:
    """Fetch language models from the xAI ``/v1/language-models`` endpoint.

    Uses ``input_modalities`` for vision detection.  Context size is resolved
    via the OpenRouter catalog / prefix heuristic (the xAI API does not expose
    context window sizes).
    """
    import httpx
    from providers.catalog import model_info_from_metadata, model_info_to_cache_entry

    try:
        resp = httpx.get(
            f"{XAI_BASE_URL}/language-models",
            headers={"Authorization": f"Bearer {api_key}"},
            timeout=15,
        )
        resp.raise_for_status()
        body = resp.json()
        models = body.get("models", [])
    except Exception as exc:
        logger.warning("Failed to fetch xai models: %s", exc)
        return 0

    count = 0
    with _cloud_cache_lock:
        for m in models:
            mid = m.get("id", "")
            if not mid:
                continue
            if any(s in mid for s in _XAI_SKIP_SUBSTRINGS):
                continue
            display = mid  # xAI doesn't return a display name
            ctx = _catalog_or_heuristic(mid)
            has_vision = "image" in m.get("input_modalities", [])
            metadata = dict(m)
            metadata["vision"] = has_vision
            _cloud_model_cache[mid] = model_info_to_cache_entry(model_info_from_metadata(
                "xai", mid, metadata,
                display_name=display,
                context_window=ctx,
            ))
            count += 1

    logger.info("Fetched %d xai models", count)
    return count


def _fetch_minimax_models(api_key: str) -> int:
    """Populate MiniMax Anthropic-compatible models from the documented catalog."""
    if not api_key:
        return 0
    from providers.catalog import model_info_from_metadata, model_info_to_cache_entry

    count = 0
    with _cloud_cache_lock:
        for model_id, context_window in _MINIMAX_SUPPORTED_MODELS:
            _cloud_model_cache[model_id] = model_info_to_cache_entry(model_info_from_metadata(
                "minimax",
                model_id,
                {"max_input_tokens": context_window},
                display_name=model_id,
                context_window=context_window,
                source="provider_static_catalog",
            ))
            count += 1
    logger.info("Fetched %d minimax models", count)
    return count


def _fetch_opencode_models(provider_id: str) -> int:
    """Populate OpenCode models from live discovery using provider-qualified keys."""
    import httpx

    from api_keys import get_key
    from providers.catalog import model_info_to_cache_entry
    from providers.opencode import list_opencode_model_infos, opencode_models_url

    def _string_list(value: object) -> list[str] | None:
        if not isinstance(value, list):
            return None
        return [str(item).strip().lower() for item in value if str(item).strip()]

    def _opencode_input_modalities(item: dict) -> list[str] | None:
        modalities = item.get("modalities")
        if isinstance(modalities, dict):
            values = _string_list(modalities.get("input"))
            if values is not None:
                return values
        for key in ("input_modalities", "inputModalities"):
            values = _string_list(item.get(key))
            if values is not None:
                return values
        architecture = item.get("architecture")
        if isinstance(architecture, dict):
            values = _string_list(architecture.get("input_modalities"))
            if values is not None:
                return values
        return None

    try:
        from providers.auth_store import provider_api_key_env
        env_var = provider_api_key_env(provider_id)
    except Exception:
        env_var = ""
    api_key = get_key(env_var) if env_var else ""
    headers = {"Authorization": f"Bearer {api_key}"} if api_key else {}
    model_ids: list[str] | None = None
    image_input_model_ids: set[str] | None = None
    source = "live"
    try:
        resp = httpx.get(opencode_models_url(provider_id), headers=headers, timeout=15)
        resp.raise_for_status()
        data = resp.json().get("data", [])
        model_ids = []
        discovered_image_inputs: set[str] = set()
        saw_input_metadata = False
        for item in data:
            if not isinstance(item, dict):
                continue
            model_id = str(item.get("id") or "").strip()
            if not model_id:
                continue
            model_ids.append(model_id)
            input_modalities = _opencode_input_modalities(item)
            if input_modalities is not None:
                saw_input_metadata = True
                if "image" in input_modalities:
                    discovered_image_inputs.add(model_id)
        if saw_input_metadata:
            image_input_model_ids = discovered_image_inputs
    except Exception as exc:
        logger.warning("Failed to fetch %s models: %s; using static OpenCode fallback", provider_id, exc)
        source = "fallback"

    infos = list_opencode_model_infos(
        provider_id,
        model_ids=model_ids,
        image_input_model_ids=image_input_model_ids,
    )
    count = 0
    with _cloud_cache_lock:
        for model_info in infos:
            entry = model_info_to_cache_entry(model_info)
            entry["source"] = "opencode_live_catalog" if source == "live" else "opencode_static_fallback"
            _cloud_model_cache[model_info.selection_ref] = entry
            count += 1
    logger.info("Fetched %d %s models", count, provider_id)
    return count


def refresh_cloud_models() -> int:
    """Clear cache and re-fetch from all configured providers.

    Also refreshes the context catalog (keyless) for accurate context sizes.
    If the current default model is temporarily absent from the refreshed
    cache, keep the user's saved choice instead of rewriting it to a local
    fallback.
    """
    global _current_model, _llm_instance
    # Remember current model before clearing
    _prev_model = _current_model
    _was_cloud = is_cloud_model(_prev_model)

    # Fetch context catalog first so OpenAI models get accurate sizes
    fetch_context_catalog()
    with _cloud_cache_lock:
        _cloud_model_cache.clear()
    total = 0
    total += fetch_cloud_models("openai")
    total += fetch_cloud_models("ollama_cloud")
    total += fetch_cloud_models("openrouter")
    total += fetch_cloud_models("anthropic")
    total += fetch_cloud_models("google")
    total += fetch_cloud_models("xai")
    total += fetch_cloud_models("minimax")
    total += fetch_cloud_models("opencode_zen")
    total += fetch_cloud_models("opencode_go")
    _save_cloud_cache()

    # Do not rewrite the user's default just because a provider refresh missed
    # it. Keyring migrations, offline starts, and provider list gaps can all be
    # temporary, and clobbering model_settings.json makes new threads surprise
    # the user with the first local Ollama model.
    if _was_cloud and not _cloud_model_available_after_refresh(_prev_model):
        logger.warning(
            "Default cloud model '%s' was not returned by refresh; preserving saved default.",
            _prev_model,
        )
        _llm_instance = None  # lazy-recreate on next get_llm()

    return total


def _cloud_model_available_after_refresh(model_name: str) -> bool:
    """Return whether a saved cloud default is known after provider refresh.

    Direct API providers store bare runtime IDs in ``_cloud_model_cache`` while
    newer selectors may save provider-qualified refs such as
    ``model:codex:gpt-5.5``.  Codex subscription models are maintained by the
    provider catalog rather than ``refresh_cloud_models()``, so check that
    catalog before warning about a missing default.
    """
    runtime_model = _runtime_model_name(model_name)
    if model_name in _cloud_model_cache or runtime_model in _cloud_model_cache:
        return True

    parsed = _parse_provider_model_ref(model_name)
    if not parsed:
        return False
    provider_id, parsed_model = parsed
    if _provider_qualified_cloud_cache_key(provider_id, parsed_model) in _cloud_model_cache:
        return True
    if parsed_model in _cloud_model_cache:
        info = _cloud_model_cache.get(parsed_model, {})
        return not provider_id or info.get("provider") == provider_id
    if provider_id == "codex":
        try:
            from providers.codex import list_codex_model_infos

            return any(
                model_info.model_id == parsed_model
                for model_info in list_codex_model_infos()
            )
        except Exception:
            logger.debug("Could not validate Codex default model after refresh", exc_info=True)
    return False


def is_tool_compatible(model_name: str) -> bool:
    """Check whether a model is in the known tool-compatible set.

    All cloud models (fetched dynamically) support tool calling."""
    if is_cloud_model(model_name):
        info = _cloud_model_cache.get(model_name, {})
        snapshot = info.get("capabilities_snapshot") if isinstance(info, dict) else {}
        if isinstance(snapshot, dict) and snapshot:
            return snapshot.get("tool_calling") is not False
        return True
    try:
        from providers.ollama import is_ollama_tool_capable
        return is_ollama_tool_capable(model_name)
    except Exception:
        pass
    family = model_name.split(":")[0]
    return family in _TOOL_COMPATIBLE_FAMILIES


def check_tool_support(model_name: str) -> bool:
    """Send a minimal tool-call request to verify the model supports tools.

    Returns True if the model accepts tools, False if it rejects them (400).
    """
    client = _ollama_client()
    if not client:
        return False
    try:
        client.chat(
            model=model_name,
            messages=[{"role": "user", "content": "hi"}],
            tools=[{
                "type": "function",
                "function": {
                    "name": "_ping",
                    "description": "test",
                    "parameters": {"type": "object", "properties": {}},
                },
            }],
        )
        return True
    except Exception as exc:
        if "does not support tools" in str(exc) or "400" in str(exc):
            return False
        logger.debug("Tool support check for %s failed: %s", model_name, exc)
        return True  # Network or other error — don't block