settings.py

"""
Runtime settings pro CodeAgent v5.

Nastavení lze měnit ZA BĚHU (bez restartu Space):
- přes Gradio tab „Nastavení",
- přes API: GET/POST /admin/settings.

Změny HF Space Variables/Secrets vždy restartují Space (chování HF),
proto runtime konfigurace žije zde: env proměnné slouží jen jako
VÝCHOZÍ hodnoty při prvním startu, poté má přednost persistovaný
JSON (/data/settings.json, pokud je připojen storage bucket, jinak
.agent/settings.json na ephemeral disku).

Dvě kategorie polí:
- INSTANT  — projeví se okamžitě (limity agenta, runner, cache, ...)
- ENGINE   — vyžadují reload vLLM enginu (model, TP, kvantizace, ...);
             reload proběhne na pozadí, aplikace běží dál.
"""

from __future__ import annotations

import copy
import json
import logging
import os
import tempfile
from dataclasses import asdict, dataclass, field, fields
from pathlib import Path
from threading import RLock

logger = logging.getLogger("codeagent.settings")


def _env(name: str, default: str = "") -> str:
    return os.environ.get(name, default)


def _env_int(name: str, default: int) -> int:
    try:
        return int(os.environ.get(name, default))
    except (TypeError, ValueError):
        return default


def _env_float(name: str, default: float) -> float:
    try:
        return float(os.environ.get(name, default))
    except (TypeError, ValueError):
        return default


def _env_bool(name: str, default: bool) -> bool:
    raw = os.environ.get(name)
    if raw is None:
        return default
    return raw.lower() in ("1", "true", "yes", "on")


# Pole vyžadující reload vLLM enginu.
ENGINE_FIELDS = {
    "model",
    "model_revision",
    "download_dir",
    "tensor_parallel_size",
    "gpu_memory_utilization",
    "max_model_len",
    "quantization",
    "dtype",
    "kv_cache_dtype",
    "enforce_eager",
    "tool_call_parser",
    "reasoning_parser",
    "enable_prefix_caching",
    "max_num_seqs",
    "engine_extra_args",
}

# Pole, která se v GET /admin/settings maskují (secrets).
MASKED_FIELDS = {"runner_token"}

_MASK = "********"


@dataclass
class Settings:
    """Kompletní runtime konfigurace. Env proměnné = výchozí hodnoty."""

    # ---- vLLM engine (změna => reload enginu, ne Space) ----
    model: str = field(default_factory=lambda: _env(
        "MODEL_PRIMARY", "Qwen/Qwen3-Coder-Next-FP8"))
    # Adresář pro stahované váhy modelů (vLLM --download-dir).
    # Prázdné = auto: /data/models pokud je namountován RW storage bucket
    # (přežije restart Space), jinak /app/cache/models (ephemeral).
    download_dir: str = field(default_factory=lambda: _env("MODEL_DOWNLOAD_DIR", ""))
    # Pin na konkrétní git revizi HF repa (commit hash / tag). Prázdné =
    # nejnovější. Doporučeno pro produkci u modelů s trust_remote_code.
    model_revision: str = field(default_factory=lambda: _env("MODEL_REVISION", ""))
    tensor_parallel_size: int = field(default_factory=lambda: _env_int(
        "TENSOR_PARALLEL_SIZE", 4))
    gpu_memory_utilization: float = field(default_factory=lambda: _env_float(
        "GPU_MEMORY_UTILIZATION", 0.92))
    max_model_len: int = field(default_factory=lambda: _env_int(
        "MAX_MODEL_LEN", 131072))
    # "auto" = nech vLLM detekovat z checkpointu (FP8/AWQ/GPTQ...);
    # "none" = bez kvantizace; jinak explicitní hodnota pro vLLM.
    quantization: str = field(default_factory=lambda: _env("QUANTIZATION", "auto"))
    dtype: str = field(default_factory=lambda: _env("DTYPE", "auto"))
    kv_cache_dtype: str = field(default_factory=lambda: _env("KV_CACHE_DTYPE", "auto"))
    enforce_eager: bool = field(default_factory=lambda: _env_bool("ENFORCE_EAGER", False))
    # "auto" = odvodit z presetu/rodiny modelu; "" = vypnuto.
    tool_call_parser: str = field(default_factory=lambda: _env("TOOL_CALL_PARSER", "auto"))
    reasoning_parser: str = field(default_factory=lambda: _env("REASONING_PARSER", "auto"))
    enable_prefix_caching: bool = field(default_factory=lambda: _env_bool(
        "ENABLE_PREFIX_CACHING", True))
    max_num_seqs: int = field(default_factory=lambda: _env_int("MAX_NUM_SEQS", 0))  # 0 = default vLLM
    # Únikový ventil: libovolné další CLI argumenty pro `vllm serve`.
    engine_extra_args: str = field(default_factory=lambda: _env("ENGINE_EXTRA_ARGS", ""))

    # ---- agent / inference chování (instant) ----
    agent_mode: str = field(default_factory=lambda: _env("AGENT_MODE", "single").lower())
    # Routing v hybrid módu: "ai" = lokální model klasifikuje složitost úlohy
    # (fallback na keywords při chybě), "keywords" = jen statická pravidla.
    router_mode: str = field(default_factory=lambda: _env("ROUTER_MODE", "ai").lower())
    kimi_model: str = field(default_factory=lambda: _env(
        "KIMI_MODEL", "moonshotai/Kimi-K2.7-Code"))
    local_context_limit: int = field(default_factory=lambda: _env_int(
        "LOCAL_CONTEXT_LIMIT", 24000))
    temperature: float = field(default_factory=lambda: _env_float("TEMPERATURE", 0.2))
    max_output_tokens: int = field(default_factory=lambda: _env_int(
        "MAX_OUTPUT_TOKENS", 8192))
    max_steps: int = field(default_factory=lambda: _env_int("MAX_STEPS", 30))
    max_sub_steps: int = field(default_factory=lambda: _env_int("MAX_SUB_STEPS", 15))
    # Úroveň výpisů v chatu: full = nástroje + argumenty + výsledky,
    # compact = jen jména nástrojů, final = pouze finální odpověď.
    chat_verbosity: str = field(default_factory=lambda: _env("CHAT_VERBOSITY", "full").lower())

    # ---- sub-agenti (instant; prompt = "" znamená výchozí) ----
    subagent_explorer_enabled: bool = field(default_factory=lambda: _env_bool(
        "SUBAGENT_EXPLORER_ENABLED", True))
    subagent_coder_enabled: bool = field(default_factory=lambda: _env_bool(
        "SUBAGENT_CODER_ENABLED", True))
    subagent_reviewer_enabled: bool = field(default_factory=lambda: _env_bool(
        "SUBAGENT_REVIEWER_ENABLED", True))
    max_explorer_steps: int = field(default_factory=lambda: _env_int("MAX_EXPLORER_STEPS", 20))
    max_coder_steps: int = field(default_factory=lambda: _env_int("MAX_CODER_STEPS", 25))
    max_reviewer_steps: int = field(default_factory=lambda: _env_int("MAX_REVIEWER_STEPS", 10))
    subagent_explorer_prompt: str = field(default_factory=lambda: _env(
        "SUBAGENT_EXPLORER_PROMPT", ""))
    subagent_coder_prompt: str = field(default_factory=lambda: _env(
        "SUBAGENT_CODER_PROMPT", ""))
    subagent_reviewer_prompt: str = field(default_factory=lambda: _env(
        "SUBAGENT_REVIEWER_PROMPT", ""))

    # ---- správa kontextu a tokenů (instant) ----
    # trim = deterministická kompakce (stárnutí tool výsledků + vypouštění
    # nejstarších bloků), off = vypnuto.
    context_compaction: str = field(default_factory=lambda: _env(
        "CONTEXT_COMPACTION", "trim").lower())
    # 0 = auto: max_model_len - max_output_tokens - 2048 rezerva.
    context_budget_tokens: int = field(default_factory=lambda: _env_int(
        "CONTEXT_BUDGET_TOKENS", 0))
    # Posledních N bloků konverzace se nikdy nekompaktuje.
    context_keep_last_steps: int = field(default_factory=lambda: _env_int(
        "CONTEXT_KEEP_LAST_STEPS", 6))
    # Tvrdý strop délky výsledku nástroje při vložení do kontextu.
    tool_result_max_chars: int = field(default_factory=lambda: _env_int(
        "TOOL_RESULT_MAX_CHARS", 24000))
    # Na kolik znaků se zkrátí staré tool výsledky při kompakci.
    tool_result_aged_chars: int = field(default_factory=lambda: _env_int(
        "TOOL_RESULT_AGED_CHARS", 2000))

    # ---- runner (instant) ----
    runner_url: str = field(default_factory=lambda: _env("LOCAL_RUNNER_URL", "").rstrip("/"))
    runner_token: str = field(default_factory=lambda: _env("LOCAL_RUNNER_TOKEN", ""))
    runner_timeout: int = field(default_factory=lambda: _env_int("RUNNER_TIMEOUT", 180))

    # ---- ostatní (instant) ----
    cache_ttl_seconds: int = field(default_factory=lambda: _env_int("CACHE_TTL_SECONDS", 300))
    log_level: str = field(default_factory=lambda: _env("LOG_LEVEL", "INFO").upper())


VALID_AGENT_MODES = {"single", "hybrid"}
VALID_ROUTER_MODES = {"ai", "keywords"}
VALID_VERBOSITY = {"full", "compact", "final"}
VALID_COMPACTION = {"trim", "off"}
VALID_DTYPES = {"auto", "half", "float16", "bfloat16", "float32"}
VALID_KV_CACHE_DTYPES = {"auto", "fp8", "fp8_e5m2", "fp8_e4m3"}


def validate_field(name: str, value):
    """Zvaliduje a znormalizuje jednu hodnotu. Vrací (ok, normalized|error_msg)."""
    spec = {f.name: f for f in fields(Settings)}.get(name)
    if spec is None:
        return False, f"Neznámé pole '{name}'"
    typ = spec.type if isinstance(spec.type, type) else None

    try:
        if name in ("tensor_parallel_size", "max_model_len", "max_num_seqs",
                    "local_context_limit", "max_output_tokens", "max_steps",
                    "max_sub_steps", "max_explorer_steps", "max_coder_steps",
                    "max_reviewer_steps", "runner_timeout", "cache_ttl_seconds",
                    "context_budget_tokens", "context_keep_last_steps",
                    "tool_result_max_chars", "tool_result_aged_chars"):
            value = int(value)
            if value < 0:
                return False, f"{name} musí být >= 0"
            if name == "tensor_parallel_size" and not (1 <= value <= 8):
                return False, "tensor_parallel_size musí být 1–8"
            if name == "max_model_len" and value < 1024:
                return False, "max_model_len musí být >= 1024"
            if name == "context_keep_last_steps" and not (1 <= value <= 50):
                return False, "context_keep_last_steps musí být 1–50"
            if name == "tool_result_max_chars" and value < 1000:
                return False, "tool_result_max_chars musí být >= 1000"
            if name == "tool_result_aged_chars" and value < 200:
                return False, "tool_result_aged_chars musí být >= 200"
        elif name in ("gpu_memory_utilization", "temperature"):
            value = float(value)
            if name == "gpu_memory_utilization" and not (0.1 <= value <= 0.99):
                return False, "gpu_memory_utilization musí být 0.10–0.99"
            if name == "temperature" and not (0.0 <= value <= 2.0):
                return False, "temperature musí být 0.0–2.0"
        elif name in ("enforce_eager", "enable_prefix_caching",
                      "subagent_explorer_enabled", "subagent_coder_enabled",
                      "subagent_reviewer_enabled"):
            if isinstance(value, str):
                value = value.lower() in ("1", "true", "yes", "on")
            else:
                value = bool(value)
        elif name == "agent_mode":
            value = str(value).lower().strip()
            if value not in VALID_AGENT_MODES:
                return False, f"agent_mode musí být jedno z {sorted(VALID_AGENT_MODES)}"
        elif name == "router_mode":
            value = str(value).lower().strip()
            if value not in VALID_ROUTER_MODES:
                return False, f"router_mode musí být jedno z {sorted(VALID_ROUTER_MODES)}"
        elif name == "chat_verbosity":
            value = str(value).lower().strip()
            if value not in VALID_VERBOSITY:
                return False, f"chat_verbosity musí být jedno z {sorted(VALID_VERBOSITY)}"
        elif name == "context_compaction":
            value = str(value).lower().strip()
            if value not in VALID_COMPACTION:
                return False, f"context_compaction musí být jedno z {sorted(VALID_COMPACTION)}"
        elif name == "dtype":
            value = str(value).lower().strip()
            if value not in VALID_DTYPES:
                return False, f"dtype musí být jedno z {sorted(VALID_DTYPES)}"
        elif name == "kv_cache_dtype":
            value = str(value).lower().strip()
            if value not in VALID_KV_CACHE_DTYPES:
                return False, f"kv_cache_dtype musí být jedno z {sorted(VALID_KV_CACHE_DTYPES)}"
        elif name == "log_level":
            value = str(value).upper().strip()
            if value not in ("DEBUG", "INFO", "WARNING", "ERROR"):
                return False, "log_level musí být DEBUG/INFO/WARNING/ERROR"
        elif name == "runner_url":
            value = str(value).strip().rstrip("/")
            if value and not value.startswith(("http://", "https://")):
                return False, "runner_url musí začínat http(s)://"
        elif name == "model":
            value = str(value).strip()
            if not value:
                return False, "model nesmí být prázdný"
        elif name == "download_dir":
            value = str(value).strip().rstrip("/")
            if value and not value.startswith("/"):
                return False, ("download_dir musí být absolutní cesta "
                               "(např. /data/models), nebo prázdné = auto")
        else:
            value = str(value).strip() if isinstance(value, str) else value
        _ = typ  # typ zde není potřeba, validace je explicitní
        return True, value
    except (TypeError, ValueError) as e:
        return False, f"Neplatná hodnota pro {name}: {e}"


def _persist_dir() -> Path:
    """/data (storage bucket, přežije restart) > .agent (ephemeral)."""
    data = Path(os.environ.get("SETTINGS_DIR", "/data"))
    try:
        if data.is_dir() and os.access(data, os.W_OK):
            return data
    except OSError:
        pass
    fallback = Path(".agent")
    fallback.mkdir(parents=True, exist_ok=True)
    return fallback


class SettingsManager:
    """Thread-safe držák runtime nastavení s JSON persistencí."""

    def __init__(self, path: Path | None = None):
        self._lock = RLock()
        self.path = path or (_persist_dir() / "settings.json")
        self._settings = Settings()
        self.revision = 0
        self._load_persisted()

    # -------------------------------------------------------------- čtení

    def get(self) -> Settings:
        """Vrátí kopii aktuálních nastavení (bezpečné ke čtení bez zámku)."""
        with self._lock:
            return copy.deepcopy(self._settings)

    def as_dict(self, mask_secrets: bool = True) -> dict:
        with self._lock:
            data = asdict(self._settings)
        if mask_secrets:
            for key in MASKED_FIELDS:
                if data.get(key):
                    data[key] = _MASK
        return data

    # -------------------------------------------------------------- zápis

    def update(self, changes: dict) -> tuple[dict, bool, dict]:
        """Aplikuje změny. Vrací (applied, engine_reload_needed, errors)."""
        applied: dict = {}
        errors: dict = {}
        engine_reload = False
        with self._lock:
            for name, raw in changes.items():
                # maskovaná hodnota z UI => beze změny
                if name in MASKED_FIELDS and raw == _MASK:
                    continue
                ok, result = validate_field(name, raw)
                if not ok:
                    errors[name] = result
                    continue
                current = getattr(self._settings, name)
                if current == result:
                    continue
                setattr(self._settings, name, result)
                applied[name] = result
                if name in ENGINE_FIELDS:
                    engine_reload = True
            if applied:
                self.revision += 1
                self._persist()
                if "log_level" in applied:
                    logging.getLogger("codeagent").setLevel(applied["log_level"])
        if applied:
            safe = {k: (_MASK if k in MASKED_FIELDS else v) for k, v in applied.items()}
            logger.info("Nastavení změněno (rev %s): %s", self.revision, safe)
        return applied, engine_reload, errors

    # -------------------------------------------------------------- persistence

    def _persist(self):
        try:
            self.path.parent.mkdir(parents=True, exist_ok=True)
            payload = json.dumps(asdict(self._settings), ensure_ascii=False, indent=2)
            fd, tmp = tempfile.mkstemp(dir=str(self.path.parent), suffix=".tmp")
            with os.fdopen(fd, "w", encoding="utf-8") as f:
                f.write(payload)
            os.replace(tmp, self.path)
        except OSError as e:
            logger.warning("Persistence nastavení selhala (%s): %s", self.path, e)

    def _load_persisted(self):
        if not self.path.exists():
            return
        try:
            data = json.loads(self.path.read_text(encoding="utf-8"))
        except (OSError, json.JSONDecodeError) as e:
            logger.warning("Načtení %s selhalo: %s", self.path, e)
            return
        valid_names = {f.name for f in fields(Settings)}
        for name, value in data.items():
            if name not in valid_names:
                continue
            ok, result = validate_field(name, value)
            if ok:
                setattr(self._settings, name, result)
        logger.info("Nastavení načteno z %s", self.path)