free-claude-code/smoke/lib/config.py

"""Smoke-suite configuration loaded from the real developer environment."""

from __future__ import annotations

import os
from collections.abc import Mapping
from dataclasses import dataclass
from pathlib import Path

from config.provider_catalog import PROVIDER_CATALOG, SUPPORTED_PROVIDER_IDS
from config.settings import Settings, get_settings

DEFAULT_TARGETS = frozenset(
    {
        "api",
        "auth",
        "cli",
        "clients",
        "config",
        "extensibility",
        "llamacpp",
        "lmstudio",
        "messaging",
        "ollama",
        "providers",
        "rate_limit",
        "tools",
    }
)
SIDE_EFFECT_TARGETS = frozenset({"discord", "telegram", "voice"})
ALL_TARGETS = DEFAULT_TARGETS | SIDE_EFFECT_TARGETS
TARGET_ALIASES = {
    "contract": "api",
    "optimizations": "api",
    "thinking": "providers",
    "vscode": "clients",
}
SECRET_KEY_PARTS = ("KEY", "TOKEN", "SECRET", "WEBHOOK", "AUTH")

PROVIDER_SMOKE_DEFAULT_MODELS: dict[str, str] = {
    "nvidia_nim": "nvidia_nim/z-ai/glm4.7",
    "open_router": "open_router/stepfun/step-3.5-flash:free",
    "deepseek": "deepseek/deepseek-v4-pro",
    "lmstudio": "lmstudio/local-model",
    "llamacpp": "llamacpp/local-model",
    "ollama": "ollama/llama3.1",
    "wafer": "wafer/DeepSeek-V4-Pro",
}


TARGET_REQUIRED_ENV: dict[str, tuple[str, ...]] = {
    "api": (),
    "auth": (),
    "cli": ("FCC_SMOKE_CLAUDE_BIN", "configured provider for Claude CLI prompt"),
    "clients": (),
    "config": (),
    "extensibility": (),
    "messaging": (),
    "providers": ("configured provider credentials/endpoints or FCC_SMOKE_MODEL_*",),
    "rate_limit": ("configured provider model",),
    "tools": ("configured tool-capable provider model",),
    "lmstudio": ("LM_STUDIO_BASE_URL with a running LM Studio server",),
    "llamacpp": ("LLAMACPP_BASE_URL with a running llama-server",),
    "ollama": ("OLLAMA_BASE_URL with a running Ollama server",),
    "telegram": (
        "TELEGRAM_BOT_TOKEN",
        "ALLOWED_TELEGRAM_USER_ID or FCC_SMOKE_TELEGRAM_CHAT_ID",
    ),
    "discord": (
        "DISCORD_BOT_TOKEN",
        "ALLOWED_DISCORD_CHANNELS or FCC_SMOKE_DISCORD_CHANNEL_ID",
    ),
    "voice": ("VOICE_NOTE_ENABLED=true", "FCC_SMOKE_RUN_VOICE=1"),
}


@dataclass(frozen=True, slots=True)
class ProviderModel:
    provider: str
    full_model: str
    source: str

    @property
    def model_name(self) -> str:
        return Settings.parse_model_name(self.full_model)


@dataclass(frozen=True, slots=True)
class SmokeConfig:
    root: Path
    results_dir: Path
    live: bool
    interactive: bool
    targets: frozenset[str]
    provider_matrix: frozenset[str]
    timeout_s: float
    prompt: str
    claude_bin: str
    worker_id: str
    settings: Settings

    @classmethod
    def load(cls) -> SmokeConfig:
        root = Path(__file__).resolve().parents[2]
        get_settings.cache_clear()
        settings = get_settings()
        return cls(
            root=root,
            results_dir=root / ".smoke-results",
            live=os.getenv("FCC_LIVE_SMOKE") == "1",
            interactive=os.getenv("FCC_SMOKE_INTERACTIVE") == "1",
            targets=_parse_targets(os.getenv("FCC_SMOKE_TARGETS")),
            provider_matrix=_parse_csv(os.getenv("FCC_SMOKE_PROVIDER_MATRIX")),
            timeout_s=float(os.getenv("FCC_SMOKE_TIMEOUT_S", "45")),
            prompt=os.getenv("FCC_SMOKE_PROMPT", "Reply with exactly: FCC_SMOKE_PONG"),
            claude_bin=os.getenv("FCC_SMOKE_CLAUDE_BIN", "claude"),
            worker_id=os.getenv("PYTEST_XDIST_WORKER", "main"),
            settings=settings,
        )

    def target_enabled(self, *names: str) -> bool:
        return any(name in self.targets for name in names)

    def provider_models(self) -> list[ProviderModel]:
        candidates = (
            ("MODEL", self.settings.model),
            ("MODEL_OPUS", self.settings.model_opus),
            ("MODEL_SONNET", self.settings.model_sonnet),
            ("MODEL_HAIKU", self.settings.model_haiku),
        )
        seen: set[str] = set()
        models: list[ProviderModel] = []
        for source, model in candidates:
            if not model or model in seen:
                continue
            provider = Settings.parse_provider_type(model)
            if self.provider_matrix and provider not in self.provider_matrix:
                continue
            if not self.has_provider_configuration(provider):
                continue
            seen.add(model)
            models.append(
                ProviderModel(provider=provider, full_model=model, source=source)
            )
        return models

    def provider_smoke_models(self) -> list[ProviderModel]:
        """Return one smoke model per configured provider, independent of MODEL_*."""
        models: list[ProviderModel] = []
        mapped_providers = {model.provider for model in self.provider_models()}
        for provider in SUPPORTED_PROVIDER_IDS:
            if self.provider_matrix and provider not in self.provider_matrix:
                continue
            if not self.has_provider_configuration(provider):
                continue
            if not self._include_provider_in_smoke(provider, mapped_providers):
                continue
            full_model, source = _provider_smoke_model(provider)
            models.append(
                ProviderModel(provider=provider, full_model=full_model, source=source)
            )
        return models

    def _include_provider_in_smoke(
        self, provider: str, mapped_providers: set[str]
    ) -> bool:
        descriptor = PROVIDER_CATALOG[provider]
        if "local" not in descriptor.capabilities:
            return True
        if provider in mapped_providers:
            return True
        if self.provider_matrix and provider in self.provider_matrix:
            return True
        return bool(os.getenv(f"FCC_SMOKE_MODEL_{provider.upper()}"))

    def has_provider_configuration(self, provider: str) -> bool:
        if provider == "nvidia_nim":
            return bool(self.settings.nvidia_nim_api_key.strip())
        if provider == "open_router":
            return bool(self.settings.open_router_api_key.strip())
        if provider == "deepseek":
            return bool(self.settings.deepseek_api_key.strip())
        if provider == "lmstudio":
            return bool(self.settings.lm_studio_base_url.strip())
        if provider == "llamacpp":
            return bool(self.settings.llamacpp_base_url.strip())
        if provider == "ollama":
            return bool(self.settings.ollama_base_url.strip())
        if provider == "wafer":
            return bool(self.settings.wafer_api_key.strip())
        return False


def _parse_csv(raw: str | None) -> frozenset[str]:
    if not raw:
        return frozenset()
    return frozenset(part.strip() for part in raw.split(",") if part.strip())


def _parse_targets(raw: str | None) -> frozenset[str]:
    if not raw:
        return DEFAULT_TARGETS
    parsed = _parse_csv(raw)
    if "all" in parsed:
        return ALL_TARGETS
    return frozenset(TARGET_ALIASES.get(target, target) for target in parsed)


def _provider_smoke_model(provider: str) -> tuple[str, str]:
    override_env = f"FCC_SMOKE_MODEL_{provider.upper()}"
    if override := os.getenv(override_env):
        return _normalize_provider_model(provider, override), override_env

    default = PROVIDER_SMOKE_DEFAULT_MODELS.get(provider)
    if default is None:
        descriptor = PROVIDER_CATALOG[provider]
        default = f"{descriptor.provider_id}/smoke-default"
    return default, "provider_default"


def _normalize_provider_model(provider: str, raw_model: str) -> str:
    model = raw_model.strip()
    if not model:
        msg = f"FCC_SMOKE_MODEL_{provider.upper()} must not be empty"
        raise ValueError(msg)
    if "/" not in model:
        return f"{provider}/{model}"
    prefix = Settings.parse_provider_type(model)
    if prefix == provider:
        return model
    if prefix in SUPPORTED_PROVIDER_IDS:
        msg = (
            f"FCC_SMOKE_MODEL_{provider.upper()} must use provider prefix "
            f"{provider!r}, got {model!r}"
        )
        raise ValueError(msg)
    return f"{provider}/{model}"


def auth_headers(token: str | None = None) -> dict[str, str]:
    settings = get_settings()
    resolved = token if token is not None else settings.anthropic_auth_token
    headers = {
        "anthropic-version": "2023-06-01",
        "content-type": "application/json",
    }
    if resolved:
        headers["x-api-key"] = resolved
    return headers


def redacted(value: str, env: Mapping[str, str] | None = None) -> str:
    """Redact known secrets from a string before writing smoke artifacts."""
    if not value:
        return value

    source = env if env is not None else os.environ
    result = value
    for key, secret in source.items():
        if not secret or len(secret) < 4:
            continue
        if any(part in key.upper() for part in SECRET_KEY_PARTS):
            result = result.replace(secret, f"<redacted:{key}>")
    return result