hkuds--deeptutor/deeptutor/services/subagent/models.py

"""Backend option discovery — the synced model + reasoning-effort lists.

The /settings "Partners & Agents" page lets the user pick the model and
reasoning effort DeepTutor drives each backend with. Those lists change over
time (vendors add/retire models), so this is read live, not hard-coded, and the
page's "sync" button just re-reads it:

* **Codex** publishes an authoritative, server-synced cache at
  ``$CODEX_HOME/models_cache.json`` (slugs + per-model reasoning levels) and the
  user's current default in ``config.toml`` — we read both.
* **Claude Code** has no model-list CLI; its ``--model`` takes stable aliases
  (opus / sonnet / haiku) plus any full name, and ``--effort`` a fixed set — so
  we offer those as suggestions and the UI also allows a free-text model.
"""

from __future__ import annotations

from dataclasses import dataclass, field
import json
import logging
import os
from pathlib import Path
import re
from typing import Any

from deeptutor.services.subagent.process import probe_version
from deeptutor.services.subagent.registry import get_backend

logger = logging.getLogger(__name__)

# Claude Code: curated fallback used until a live ``/model`` sync populates the
# cache (see ``claude_models``). The aliases + ``[1m]`` variants mirror the
# ``/model`` picker; the UI also accepts a free-text model name.
_CLAUDE_MODELS = (
    ("opus", "Opus 4.8 · 1M context"),
    ("sonnet", "Sonnet 4.6"),
    ("sonnet[1m]", "Sonnet 4.6 · 1M context"),
    ("haiku", "Haiku 4.5"),
)
_CLAUDE_EFFORTS = ("low", "medium", "high", "xhigh", "max")


@dataclass(slots=True)
class ModelOption:
    slug: str
    display_name: str
    default_effort: str = ""
    efforts: list[str] = field(default_factory=list)

    def to_dict(self) -> dict[str, Any]:
        return {
            "slug": self.slug,
            "display_name": self.display_name,
            "default_effort": self.default_effort,
            "efforts": list(self.efforts),
        }


@dataclass(slots=True)
class BackendOptions:
    kind: str
    display_name: str
    available: bool
    version: str = ""
    default_model: str = ""
    models: list[ModelOption] = field(default_factory=list)
    # Effort scale when it isn't model-specific (Claude Code). For Codex the
    # per-model ``efforts`` are authoritative; this is a fallback/union.
    efforts: list[str] = field(default_factory=list)
    # Whether the UI should allow a free-text model (true when we can't fully
    # enumerate, e.g. Claude Code aliases + full names).
    allow_custom_model: bool = False
    # Freshness of a synced source, if any (Codex cache fetch time).
    synced_at: str = ""
    detail: str = ""

    def to_dict(self) -> dict[str, Any]:
        return {
            "kind": self.kind,
            "display_name": self.display_name,
            "available": self.available,
            "version": self.version,
            "default_model": self.default_model,
            "models": [m.to_dict() for m in self.models],
            "efforts": list(self.efforts),
            "allow_custom_model": self.allow_custom_model,
            "synced_at": self.synced_at,
            "detail": self.detail,
        }


def _codex_home() -> Path:
    raw = os.environ.get("CODEX_HOME", "").strip()
    return Path(raw).expanduser() if raw else Path.home() / ".codex"


def _codex_default_model() -> str:
    """Read ``model = "..."`` from the user's Codex config.toml (best-effort)."""
    config = _codex_home() / "config.toml"
    try:
        text = config.read_text(encoding="utf-8")
    except Exception:
        return ""
    # The default model is the top-level ``model = "..."`` (ignore nested
    # ``[profiles.*] model`` by matching only a line-start assignment).
    match = re.search(r'(?m)^\s*model\s*=\s*"([^"]+)"', text)
    return match.group(1) if match else ""


async def _claude_options() -> BackendOptions:
    from deeptutor.services.subagent.claude_models import load_cached_claude_models

    backend = get_backend("claude_code")
    ok, text = await probe_version([backend.cli_command, "--version"]) if backend else (False, "")
    # Prefer a live-synced catalog (scraped from ``/model``); fall back to the
    # curated aliases until the user syncs.
    cached, synced_at = load_cached_claude_models()
    pairs = [(m["slug"], m["display_name"]) for m in cached] or list(_CLAUDE_MODELS)
    return BackendOptions(
        kind="claude_code",
        display_name="Claude Code",
        available=ok,
        version=text if ok else "",
        default_model="",
        models=[
            ModelOption(slug=slug, display_name=name, efforts=list(_CLAUDE_EFFORTS))
            for slug, name in pairs
        ],
        efforts=list(_CLAUDE_EFFORTS),
        allow_custom_model=True,
        synced_at=synced_at,
        detail="" if ok else (text or "claude CLI not found on PATH"),
    )


async def _codex_options() -> BackendOptions:
    backend = get_backend("codex")
    ok, version = (
        await probe_version([backend.cli_command, "--version"]) if backend else (False, "")
    )
    models: list[ModelOption] = []
    synced_at = ""
    cache = _codex_home() / "models_cache.json"
    try:
        data = json.loads(cache.read_text(encoding="utf-8"))
        synced_at = str(data.get("fetched_at") or "")
        for entry in data.get("models", []):
            if not isinstance(entry, dict) or not entry.get("slug"):
                continue
            efforts = [
                str(level.get("effort"))
                for level in entry.get("supported_reasoning_levels", [])
                if isinstance(level, dict) and level.get("effort")
            ]
            models.append(
                ModelOption(
                    slug=str(entry["slug"]),
                    display_name=str(entry.get("display_name") or entry["slug"]),
                    default_effort=str(entry.get("default_reasoning_level") or ""),
                    efforts=efforts,
                )
            )
    except FileNotFoundError:
        pass
    except Exception:
        logger.warning("failed to read codex models cache", exc_info=True)
    return BackendOptions(
        kind="codex",
        display_name="Codex",
        available=ok,
        version=version if ok else "",
        default_model=_codex_default_model(),
        models=models,
        efforts=["none", "minimal", "low", "medium", "high", "xhigh"],
        allow_custom_model=True,
        synced_at=synced_at,
        detail="" if ok else (version or "codex CLI not found on PATH"),
    )


async def list_backend_options() -> list[BackendOptions]:
    """Synced model/effort options for every backend (the /settings sync source)."""
    return [await _claude_options(), await _codex_options()]


async def sync_backend_options(kind: str) -> BackendOptions:
    """Refresh and return one backend's options (the /settings "sync" action).

    Claude Code has no machine-readable catalog, so we actively scrape its
    ``/model`` TUI and cache the result. Codex's cache is maintained by its own
    CLI, so syncing is just a fresh read.
    """
    if kind == "claude_code":
        from deeptutor.services.subagent.claude_models import sync_claude_models

        await sync_claude_models()  # writes the cache that _claude_options reads
        return await _claude_options()
    return await _codex_options()


__all__ = [
    "BackendOptions",
    "ModelOption",
    "list_backend_options",
    "sync_backend_options",
]