940 lines
40 KiB
Python
940 lines
40 KiB
Python
"""Administrator model configuration and runtime capabilities.
|
|
|
|
The public ``/api/models`` endpoint intentionally stays small and never
|
|
returns secrets. This router is the complementary administrator surface: it
|
|
edits only the top-level ``models:`` section in the active DeerFlow config,
|
|
keeps a disk backup before every write, and swaps the gateway's runtime config
|
|
after validation so new requests can use the change without a restart.
|
|
|
|
``supports_reasoning_effort`` is deliberately different: it is an
|
|
administrator-managed runtime capability persisted in the database, never in
|
|
``config.yaml``. The field remains accepted on the general create/update
|
|
payload for compatibility with older frontends, while a dedicated endpoint is
|
|
the canonical mutation surface.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import copy
|
|
import errno
|
|
import logging
|
|
import os
|
|
import re
|
|
import shutil
|
|
import sys
|
|
import tempfile
|
|
from datetime import UTC, datetime
|
|
from pathlib import Path
|
|
from time import perf_counter, sleep
|
|
from typing import Any, Literal
|
|
|
|
import yaml
|
|
from fastapi import APIRouter, HTTPException, Request
|
|
from langchain_core.messages import HumanMessage
|
|
from pydantic import BaseModel, ConfigDict, Field, field_validator
|
|
|
|
from app.gateway.deps import get_optional_user_from_request
|
|
from deerflow.config.app_config import AppConfig, reload_app_config
|
|
from deerflow.config.model_config import ModelConfig, WriteFileTransport
|
|
from deerflow.models.factory import create_chat_model
|
|
from deerflow.persistence.model_capabilities import ModelCapabilityStore
|
|
|
|
logger = logging.getLogger(__name__)
|
|
router = APIRouter(prefix="/api/admin/models", tags=["admin-models"])
|
|
|
|
_MODEL_BLOCK_RE = re.compile(r"(?m)^models:[^\r\n]*(?:\r?\n|$)")
|
|
_TOP_LEVEL_KEY_RE = re.compile(r"(?m)^[A-Za-z_][A-Za-z0-9_-]*:[^\r\n]*(?:\r?\n|$)")
|
|
_BACKUP_DIR_NAME = ".model-config-backups"
|
|
_MUTATION_LOCK = asyncio.Lock()
|
|
_ATOMIC_REPLACE_RETRIES = 5
|
|
_ATOMIC_REPLACE_RETRY_ERRNOS = {
|
|
errno.EBUSY,
|
|
errno.EAGAIN,
|
|
getattr(errno, "ETXTBSY", 26),
|
|
}
|
|
|
|
|
|
def _is_file_mountpoint(path: Path) -> bool:
|
|
"""Return whether ``path`` itself is mounted (common with Docker file binds)."""
|
|
if not sys.platform.startswith("linux"):
|
|
return False
|
|
try:
|
|
target = str(path.resolve())
|
|
with Path("/proc/self/mountinfo").open(encoding="utf-8") as mountinfo:
|
|
for line in mountinfo:
|
|
fields = line.split(" - ", maxsplit=1)[0].split()
|
|
if len(fields) < 5:
|
|
continue
|
|
mount_path = re.sub(
|
|
r"\\([0-7]{3})",
|
|
lambda match: chr(int(match.group(1), 8)),
|
|
fields[4],
|
|
)
|
|
if mount_path == target:
|
|
return True
|
|
except OSError:
|
|
logger.debug("Unable to inspect Linux mount table for model config path", exc_info=True)
|
|
return False
|
|
|
|
|
|
def _write_mounted_config_in_place(path: Path, raw_before: str, raw_after: str) -> None:
|
|
"""Update a single-file bind mount, where Linux refuses atomic rename (EBUSY)."""
|
|
try:
|
|
with path.open("w", encoding="utf-8", newline="") as handle:
|
|
handle.write(raw_after)
|
|
handle.flush()
|
|
os.fsync(handle.fileno())
|
|
except Exception:
|
|
# The backup was persisted before reaching this point. Restore the
|
|
# previous contents best-effort so a failed write does not leave YAML
|
|
# truncated or partially replaced.
|
|
try:
|
|
with path.open("w", encoding="utf-8", newline="") as handle:
|
|
handle.write(raw_before)
|
|
handle.flush()
|
|
os.fsync(handle.fileno())
|
|
except OSError:
|
|
logger.exception("Failed to restore model config after an in-place bind-mount write failure")
|
|
raise
|
|
|
|
_EDITOR_FIELDS = frozenset(
|
|
{
|
|
"name",
|
|
"display_name",
|
|
"description",
|
|
"use",
|
|
"model",
|
|
"api_key",
|
|
"base_url",
|
|
"request_timeout",
|
|
"max_retries",
|
|
"max_input_tokens",
|
|
"max_tokens",
|
|
"temperature",
|
|
"stream_usage",
|
|
"supports_thinking",
|
|
"supports_reasoning_effort",
|
|
"supports_vision",
|
|
"when_thinking_enabled",
|
|
"when_thinking_disabled",
|
|
"thinking",
|
|
"use_responses_api",
|
|
"output_version",
|
|
"write_file_transport",
|
|
"strict_stream_integrity",
|
|
}
|
|
)
|
|
_PRESERVED_FIELDS = frozenset(ModelConfig.model_fields) | _EDITOR_FIELDS
|
|
|
|
|
|
class ModelAdminItem(BaseModel):
|
|
"""A model configuration safe to return to an administrator browser."""
|
|
|
|
name: str
|
|
display_name: str | None = None
|
|
description: str | None = None
|
|
use: str
|
|
model: str
|
|
base_url: str | None = None
|
|
request_timeout: float | None = None
|
|
max_retries: int | None = None
|
|
max_input_tokens: int | None = None
|
|
max_tokens: int | None = None
|
|
temperature: float | None = None
|
|
stream_usage: bool | None = None
|
|
supports_thinking: bool = False
|
|
supports_reasoning_effort: bool = False
|
|
default_reasoning_effort: Literal["low", "medium", "xhigh"] = "xhigh"
|
|
supports_vision: bool = False
|
|
when_thinking_enabled: dict[str, Any] | None = None
|
|
when_thinking_disabled: dict[str, Any] | None = None
|
|
thinking: dict[str, Any] | None = None
|
|
use_responses_api: bool | None = None
|
|
output_version: str | None = None
|
|
write_file_transport: WriteFileTransport = WriteFileTransport.NATIVE
|
|
# ``None`` means the runtime default: auto-enabled only for the historically
|
|
# affected qwen3.8-flash-next model id (see deerflow.models.chat_stream_state).
|
|
strict_stream_integrity: bool | None = None
|
|
extra_settings: dict[str, Any] = Field(default_factory=dict)
|
|
api_key_configured: bool = False
|
|
api_key_source: str | None = None
|
|
source: str = "config.yaml"
|
|
|
|
|
|
class ModelAdminListResponse(BaseModel):
|
|
models: list[ModelAdminItem]
|
|
config_path: str
|
|
updated_at: str | None = None
|
|
|
|
|
|
class ModelUpsertRequest(BaseModel):
|
|
"""Editable subset of the ``ModelConfig`` schema plus safe extra fields."""
|
|
|
|
model_config = ConfigDict(extra="forbid")
|
|
|
|
name: str = Field(min_length=1, max_length=160)
|
|
display_name: str | None = Field(default=None, max_length=160)
|
|
description: str | None = Field(default=None, max_length=1000)
|
|
use: str = Field(min_length=1, max_length=255)
|
|
model: str = Field(min_length=1, max_length=255)
|
|
# ``None`` means "keep the currently configured secret" on update.
|
|
# The value is write-only in practice: it is never part of a response.
|
|
api_key: str | None = Field(default=None, max_length=4096)
|
|
clear_api_key: bool = False
|
|
base_url: str | None = Field(default=None, max_length=2048)
|
|
request_timeout: float | None = Field(default=None, ge=1, le=600)
|
|
max_retries: int | None = Field(default=None, ge=0, le=20)
|
|
max_input_tokens: int | None = Field(default=None, ge=1, le=10_000_000)
|
|
max_tokens: int | None = Field(default=None, ge=1, le=1_000_000)
|
|
temperature: float | None = Field(default=None, ge=0, le=2)
|
|
stream_usage: bool | None = None
|
|
supports_thinking: bool = False
|
|
# DB-backed capability. Accepted here for stale clients, but never written
|
|
# into config.yaml; the dedicated capability endpoint is canonical.
|
|
supports_reasoning_effort: bool | None = None
|
|
supports_vision: bool = False
|
|
when_thinking_enabled: dict[str, Any] | None = None
|
|
when_thinking_disabled: dict[str, Any] | None = None
|
|
thinking: dict[str, Any] | None = None
|
|
use_responses_api: bool | None = None
|
|
output_version: str | None = Field(default=None, max_length=128)
|
|
# Optional on the wire so older administrator UIs preserve an existing
|
|
# bridge setting during edits. New models still resolve to ModelConfig's
|
|
# native default when the field is omitted.
|
|
write_file_transport: WriteFileTransport | None = None
|
|
# Optional on the wire so edits preserve an existing toggle; the dedicated
|
|
# strict-stream-integrity endpoint is the canonical switch surface.
|
|
strict_stream_integrity: bool | None = None
|
|
extra_settings: dict[str, Any] = Field(default_factory=dict)
|
|
|
|
@field_validator("name", "use", "model")
|
|
@classmethod
|
|
def _trim_required(cls, value: str) -> str:
|
|
normalized = value.strip()
|
|
if not normalized:
|
|
raise ValueError("value cannot be empty")
|
|
return normalized
|
|
|
|
@field_validator("display_name", "description", "base_url", "api_key", "output_version")
|
|
@classmethod
|
|
def _trim_optional(cls, value: str | None) -> str | None:
|
|
return value.strip() if isinstance(value, str) else value
|
|
|
|
|
|
class ModelTestRequest(BaseModel):
|
|
model: ModelUpsertRequest
|
|
|
|
|
|
class ReasoningEffortCapabilityRequest(BaseModel):
|
|
supports_reasoning_effort: bool
|
|
default_reasoning_effort: Literal["low", "medium", "xhigh"] | None = None
|
|
|
|
|
|
class WriteFileTransportToggleRequest(BaseModel):
|
|
"""Per-model 文件流式兼容 switch: off keeps native, on selects buffered_gateway."""
|
|
|
|
write_file_transport: Literal["native", "buffered_gateway"]
|
|
|
|
|
|
class StrictStreamIntegrityToggleRequest(BaseModel):
|
|
"""Per-model 流完整性检测 switch: detect gateway streams that end without
|
|
a terminal frame and retry instead of committing a half answer."""
|
|
|
|
strict_stream_integrity: bool
|
|
|
|
|
|
class ModelOrderRequest(BaseModel):
|
|
"""The complete ordered set of model stable names."""
|
|
|
|
names: list[str] = Field(min_length=1, max_length=500)
|
|
|
|
|
|
class ModelTestResponse(BaseModel):
|
|
success: bool
|
|
latency_ms: int
|
|
message: str
|
|
preview: str | None = None
|
|
|
|
|
|
def _mask_env_source(value: Any) -> str | None:
|
|
if isinstance(value, str) and value.startswith("$") and len(value) > 1:
|
|
return value[1:]
|
|
return None
|
|
|
|
|
|
def _to_admin_item(
|
|
raw: dict[str, Any],
|
|
*,
|
|
supports_reasoning_effort: bool = False,
|
|
default_reasoning_effort: str = "xhigh",
|
|
) -> ModelAdminItem:
|
|
api_key = raw.get("api_key")
|
|
extra = {key: copy.deepcopy(value) for key, value in raw.items() if key not in _PRESERVED_FIELDS}
|
|
return ModelAdminItem(
|
|
name=str(raw.get("name") or ""),
|
|
display_name=raw.get("display_name"),
|
|
description=raw.get("description"),
|
|
use=str(raw.get("use") or ""),
|
|
model=str(raw.get("model") or ""),
|
|
base_url=raw.get("base_url"),
|
|
request_timeout=raw.get("request_timeout"),
|
|
max_retries=raw.get("max_retries"),
|
|
max_input_tokens=raw.get("max_input_tokens"),
|
|
max_tokens=raw.get("max_tokens"),
|
|
temperature=raw.get("temperature"),
|
|
stream_usage=raw.get("stream_usage"),
|
|
supports_thinking=bool(raw.get("supports_thinking", False)),
|
|
supports_reasoning_effort=bool(supports_reasoning_effort),
|
|
default_reasoning_effort=default_reasoning_effort if default_reasoning_effort in {"low", "medium", "xhigh"} else "xhigh",
|
|
supports_vision=bool(raw.get("supports_vision", False)),
|
|
when_thinking_enabled=copy.deepcopy(raw.get("when_thinking_enabled")),
|
|
when_thinking_disabled=copy.deepcopy(raw.get("when_thinking_disabled")),
|
|
thinking=copy.deepcopy(raw.get("thinking")),
|
|
use_responses_api=raw.get("use_responses_api"),
|
|
output_version=raw.get("output_version"),
|
|
write_file_transport=raw.get("write_file_transport", WriteFileTransport.NATIVE),
|
|
strict_stream_integrity=raw.get("strict_stream_integrity") if isinstance(raw.get("strict_stream_integrity"), bool) else None,
|
|
extra_settings=extra,
|
|
api_key_configured=bool(api_key),
|
|
api_key_source=_mask_env_source(api_key),
|
|
)
|
|
|
|
|
|
def _resolve_config_path() -> Path:
|
|
return AppConfig.resolve_config_path()
|
|
|
|
|
|
def _get_capability_store(request: Request, *, required: bool = True) -> ModelCapabilityStore | None:
|
|
app = getattr(request, "app", None)
|
|
store = getattr(getattr(app, "state", None), "model_capability_store", None)
|
|
if store is None and required:
|
|
raise HTTPException(status_code=503, detail="Model capability store not available")
|
|
return store
|
|
|
|
|
|
async def _capabilities_by_name(request: Request) -> dict[str, dict[str, Any]]:
|
|
store = _get_capability_store(request, required=False)
|
|
if store is None:
|
|
return {}
|
|
rows = await store.list_capabilities()
|
|
return {str(row.get("model_name") or ""): row for row in rows if row.get("model_name")}
|
|
|
|
|
|
def _admin_user_id(user: Any) -> str | None:
|
|
value = getattr(user, "id", None)
|
|
return str(value) if value is not None else None
|
|
|
|
|
|
def _validate_reasoning_capability(model: dict[str, Any], enabled: bool) -> None:
|
|
if enabled and not bool(model.get("supports_thinking", False)):
|
|
raise HTTPException(
|
|
status_code=422,
|
|
detail="Reasoning-effort selection requires a model that supports thinking",
|
|
)
|
|
|
|
|
|
# ``write_file_transport=buffered_gateway`` and ``strict_stream_integrity`` are
|
|
# implemented only by the patched OpenAI-compatible chat-completions classes
|
|
# (see deerflow.models.factory); every other driver is rejected there at
|
|
# construction time, so mirror the same constraint here with a clearer message.
|
|
_PATCHED_OPENAI_COMPATIBLE_USES = frozenset(
|
|
{
|
|
"langchain_openai:ChatOpenAI",
|
|
"deerflow.models.patched_openai:PatchedChatOpenAI",
|
|
}
|
|
)
|
|
|
|
|
|
def _validate_patched_openai_use(model: dict[str, Any], feature: str) -> None:
|
|
if str(model.get("use") or "") not in _PATCHED_OPENAI_COMPATIBLE_USES:
|
|
raise HTTPException(
|
|
status_code=422,
|
|
detail=(
|
|
f"{feature} requires an OpenAI-compatible "
|
|
"langchain_openai:ChatOpenAI model"
|
|
),
|
|
)
|
|
if bool(model.get("use_responses_api")):
|
|
raise HTTPException(
|
|
status_code=422,
|
|
detail=(
|
|
f"{feature} requires the chat-completions API, "
|
|
"not use_responses_api"
|
|
),
|
|
)
|
|
|
|
|
|
def _validate_write_file_transport(model: dict[str, Any], transport: str) -> None:
|
|
if transport != WriteFileTransport.BUFFERED_GATEWAY.value:
|
|
return
|
|
_validate_patched_openai_use(model, "write_file compatibility")
|
|
|
|
|
|
async def _restore_capability(
|
|
store: ModelCapabilityStore,
|
|
model_name: str,
|
|
previous: dict[str, Any] | None,
|
|
request: Request,
|
|
) -> None:
|
|
"""Best-effort semantic rollback for a failed cross-store model update."""
|
|
|
|
try:
|
|
if previous is None:
|
|
await store.delete_capability(model_name)
|
|
return
|
|
await store.set_reasoning_effort_support(
|
|
model_name,
|
|
supports_reasoning_effort=bool(previous.get("supports_reasoning_effort", False)),
|
|
default_reasoning_effort=str(previous.get("default_reasoning_effort") or "xhigh"),
|
|
updated_by=previous.get("updated_by"),
|
|
)
|
|
except Exception:
|
|
logger.exception("Failed to roll back model capability for %s", model_name)
|
|
finally:
|
|
from app.gateway.catalog_cache import invalidate_catalog
|
|
|
|
await invalidate_catalog(request, "models")
|
|
|
|
|
|
def _read_config_data(path: Path) -> tuple[str, dict[str, Any], list[dict[str, Any]]]:
|
|
raw_text = path.read_text(encoding="utf-8")
|
|
try:
|
|
data = yaml.safe_load(raw_text) or {}
|
|
except yaml.YAMLError as exc:
|
|
raise HTTPException(status_code=500, detail=f"Unable to parse active config.yaml: {exc}") from exc
|
|
if not isinstance(data, dict):
|
|
raise HTTPException(status_code=500, detail="The active config.yaml root must be a mapping")
|
|
raw_models = data.get("models") or []
|
|
if not isinstance(raw_models, list) or not all(isinstance(item, dict) for item in raw_models):
|
|
raise HTTPException(status_code=500, detail="config.yaml models must be a list of mappings")
|
|
return raw_text, data, [copy.deepcopy(item) for item in raw_models]
|
|
|
|
|
|
def _render_models_block(models: list[dict[str, Any]], newline: str) -> str:
|
|
rendered = yaml.safe_dump(
|
|
{"models": models},
|
|
allow_unicode=True,
|
|
sort_keys=False,
|
|
default_flow_style=False,
|
|
width=120,
|
|
)
|
|
if newline != "\n":
|
|
rendered = rendered.replace("\n", newline)
|
|
return rendered if rendered.endswith(newline) else f"{rendered}{newline}"
|
|
|
|
|
|
def _replace_models_block(raw_text: str, models: list[dict[str, Any]]) -> str:
|
|
"""Replace the active model entries without touching other config sections.
|
|
|
|
PyYAML does not retain comments. Keep every comment found in the old
|
|
model section as documentation after the newly rendered model list rather
|
|
than silently deleting an operator's notes or disabled examples.
|
|
"""
|
|
|
|
block_match = _MODEL_BLOCK_RE.search(raw_text)
|
|
if block_match is None:
|
|
raise HTTPException(status_code=500, detail="The active config.yaml has no top-level models section")
|
|
newline = "\r\n" if "\r\n" in raw_text else "\n"
|
|
next_key = _TOP_LEVEL_KEY_RE.search(raw_text, block_match.end())
|
|
end = next_key.start() if next_key else len(raw_text)
|
|
previous_section = raw_text[block_match.end() : end]
|
|
preserved_comments = "".join(line for line in previous_section.splitlines(keepends=True) if line.lstrip().startswith("#"))
|
|
rendered = _render_models_block(models, newline)
|
|
if preserved_comments:
|
|
if not preserved_comments.endswith(("\n", "\r")):
|
|
preserved_comments = f"{preserved_comments}{newline}"
|
|
rendered = f"{rendered}{preserved_comments}"
|
|
return f"{raw_text[: block_match.start()]}{rendered}{raw_text[end:]}"
|
|
|
|
|
|
def _replace_config_file_with_retry(
|
|
temp_name: str,
|
|
path: Path,
|
|
raw_before: str,
|
|
raw_after: str,
|
|
) -> None:
|
|
"""Retry short-lived busy errors from shared/container-mounted filesystems.
|
|
|
|
Keep the write atomic: never fall back to truncating the live config. The
|
|
temporary file is unique per mutation, so a bounded retry is safe when
|
|
overlay/NFS scanners briefly hold a rename operation.
|
|
"""
|
|
if _is_file_mountpoint(path):
|
|
logger.info("Model config is a single-file mount; updating mounted file in place")
|
|
_write_mounted_config_in_place(path, raw_before, raw_after)
|
|
os.unlink(temp_name)
|
|
return
|
|
|
|
for attempt in range(_ATOMIC_REPLACE_RETRIES + 1):
|
|
try:
|
|
os.replace(temp_name, path)
|
|
return
|
|
except OSError as exc:
|
|
if exc.errno not in _ATOMIC_REPLACE_RETRY_ERRNOS or attempt >= _ATOMIC_REPLACE_RETRIES:
|
|
raise
|
|
delay = min(0.4, 0.025 * (2**attempt))
|
|
logger.warning(
|
|
"Model config atomic replace is temporarily busy (errno=%s); retry %d/%d after %.3fs",
|
|
exc.errno,
|
|
attempt + 1,
|
|
_ATOMIC_REPLACE_RETRIES,
|
|
delay,
|
|
)
|
|
sleep(delay)
|
|
|
|
|
|
def _backup_and_write_config(path: Path, raw_before: str, raw_after: str) -> Path:
|
|
backup_dir = path.parent / _BACKUP_DIR_NAME
|
|
backup_dir.mkdir(parents=True, exist_ok=True)
|
|
stamp = datetime.now(UTC).strftime("%Y%m%dT%H%M%S%fZ")
|
|
backup_path = backup_dir / f"{path.stem}.{stamp}.yaml"
|
|
# Persist the exact text that was read for this mutation. This makes the
|
|
# rollback deterministic even if another operator changes the file while
|
|
# the gateway is processing a request.
|
|
with backup_path.open("w", encoding="utf-8", newline="") as handle:
|
|
handle.write(raw_before)
|
|
|
|
fd, temp_name = tempfile.mkstemp(prefix=f".{path.name}.", suffix=".tmp", dir=path.parent)
|
|
try:
|
|
with os.fdopen(fd, "w", encoding="utf-8", newline="") as handle:
|
|
handle.write(raw_after)
|
|
handle.flush()
|
|
os.fsync(handle.fileno())
|
|
_replace_config_file_with_retry(temp_name, path, raw_before, raw_after)
|
|
except Exception:
|
|
try:
|
|
os.unlink(temp_name)
|
|
except OSError:
|
|
pass
|
|
raise
|
|
return backup_path
|
|
|
|
|
|
def _restore_backup(path: Path, backup_path: Path) -> None:
|
|
try:
|
|
shutil.copy2(backup_path, path)
|
|
except Exception:
|
|
logger.exception("Failed to restore model config backup at %s", backup_path)
|
|
|
|
|
|
def _request_to_raw_model(payload: ModelUpsertRequest, existing: dict[str, Any] | None = None) -> dict[str, Any]:
|
|
merged = copy.deepcopy(existing or {})
|
|
supplied = payload.model_dump(
|
|
exclude={"extra_settings", "clear_api_key", "supports_reasoning_effort"},
|
|
exclude_none=True,
|
|
)
|
|
transport = supplied.get("write_file_transport")
|
|
if isinstance(transport, WriteFileTransport):
|
|
supplied["write_file_transport"] = transport.value
|
|
merged.update(supplied)
|
|
for key, value in payload.extra_settings.items():
|
|
normalized = str(key).strip()
|
|
if not normalized:
|
|
raise HTTPException(status_code=422, detail="Extra setting names cannot be blank")
|
|
if normalized in _EDITOR_FIELDS or normalized in {"api_key", "clear_api_key", "extra_settings"}:
|
|
raise HTTPException(status_code=422, detail=f"Extra setting '{normalized}' duplicates a managed field")
|
|
merged[normalized] = copy.deepcopy(value)
|
|
if payload.clear_api_key:
|
|
merged.pop("api_key", None)
|
|
elif payload.api_key is None and existing is not None and "api_key" in existing:
|
|
merged["api_key"] = existing["api_key"]
|
|
|
|
# This capability is database-owned. Strip legacy YAML values whenever a
|
|
# model is rewritten so a later reload cannot silently regain authority.
|
|
merged.pop("supports_reasoning_effort", None)
|
|
|
|
# Validate the raw schema without resolving `$ENV_VAR` values. Environment
|
|
# resolution happens during the post-write runtime reload and is rolled back
|
|
# if unavailable.
|
|
try:
|
|
ModelConfig.model_validate(merged)
|
|
except Exception as exc:
|
|
raise HTTPException(status_code=422, detail=f"Invalid model configuration: {exc}") from exc
|
|
return merged
|
|
|
|
|
|
def _redact_text(value: str, secrets: list[str]) -> str:
|
|
result = value
|
|
for secret in secrets:
|
|
if secret:
|
|
result = result.replace(secret, "***")
|
|
return result[:800]
|
|
|
|
|
|
async def _require_admin(request: Request):
|
|
user = getattr(request.state, "user", None)
|
|
if user is None:
|
|
user = await get_optional_user_from_request(request)
|
|
if user is None:
|
|
raise HTTPException(status_code=401, detail="Authentication is required")
|
|
if getattr(user, "system_role", None) != "admin":
|
|
raise HTTPException(status_code=403, detail="Model configuration requires an administrator")
|
|
return user
|
|
|
|
|
|
def _strip_database_owned_fields(models: list[dict[str, Any]]) -> list[dict[str, Any]]:
|
|
"""Remove legacy DB-owned capability keys from every rendered YAML row."""
|
|
|
|
sanitized: list[dict[str, Any]] = []
|
|
for item in models:
|
|
clean = copy.deepcopy(item)
|
|
clean.pop("supports_reasoning_effort", None)
|
|
sanitized.append(clean)
|
|
return sanitized
|
|
|
|
|
|
async def _save_models(request: Request, raw_before: str, models: list[dict[str, Any]]) -> list[dict[str, Any]]:
|
|
from app.gateway.catalog_cache import invalidate_catalog
|
|
|
|
path = _resolve_config_path()
|
|
models = _strip_database_owned_fields(models)
|
|
raw_after = _replace_models_block(raw_before, models)
|
|
backup_path = await asyncio.to_thread(_backup_and_write_config, path, raw_before, raw_after)
|
|
try:
|
|
config = reload_app_config(str(path))
|
|
request.app.state.config = config
|
|
except Exception as exc:
|
|
logger.warning("Updated model configuration could not be activated: %s", type(exc).__name__)
|
|
await asyncio.to_thread(_restore_backup, path, backup_path)
|
|
try:
|
|
request.app.state.config = reload_app_config(str(path))
|
|
except Exception:
|
|
logger.exception("Failed to reload restored model config")
|
|
raise HTTPException(
|
|
status_code=422,
|
|
detail="The model configuration could not be activated; the previous config was restored.",
|
|
) from exc
|
|
await invalidate_catalog(request, "models")
|
|
return models
|
|
|
|
|
|
@router.get("", response_model=ModelAdminListResponse)
|
|
async def list_model_configurations(request: Request) -> ModelAdminListResponse:
|
|
await _require_admin(request)
|
|
path = _resolve_config_path()
|
|
(_, _, models), capabilities = await asyncio.gather(
|
|
asyncio.to_thread(_read_config_data, path),
|
|
_capabilities_by_name(request),
|
|
)
|
|
updated_at = datetime.fromtimestamp(path.stat().st_mtime, tz=UTC).isoformat()
|
|
return ModelAdminListResponse(
|
|
models=[
|
|
_to_admin_item(
|
|
item,
|
|
supports_reasoning_effort=bool(capabilities.get(str(item.get("name") or ""), {}).get("supports_reasoning_effort", False)),
|
|
default_reasoning_effort=str(capabilities.get(str(item.get("name") or ""), {}).get("default_reasoning_effort") or "xhigh"),
|
|
)
|
|
for item in models
|
|
],
|
|
config_path=str(path),
|
|
updated_at=updated_at,
|
|
)
|
|
|
|
|
|
@router.post("", response_model=ModelAdminItem, status_code=201)
|
|
async def create_model_configuration(body: ModelUpsertRequest, request: Request) -> ModelAdminItem:
|
|
admin = await _require_admin(request)
|
|
async with _MUTATION_LOCK:
|
|
path = _resolve_config_path()
|
|
raw_before, _, models = await asyncio.to_thread(_read_config_data, path)
|
|
if any(str(item.get("name")) == body.name for item in models):
|
|
raise HTTPException(status_code=409, detail=f"A model named '{body.name}' already exists")
|
|
created = _request_to_raw_model(body)
|
|
capability_changed = "supports_reasoning_effort" in body.model_fields_set and body.supports_reasoning_effort is not None
|
|
capability_store = _get_capability_store(request, required=capability_changed)
|
|
previous_capability = None
|
|
if capability_changed and capability_store is not None:
|
|
_validate_reasoning_capability(created, bool(body.supports_reasoning_effort))
|
|
previous_capability = await capability_store.get_capability(body.name)
|
|
await capability_store.set_reasoning_effort_support(
|
|
body.name,
|
|
supports_reasoning_effort=bool(body.supports_reasoning_effort),
|
|
default_reasoning_effort="xhigh",
|
|
updated_by=_admin_user_id(admin),
|
|
)
|
|
try:
|
|
await _save_models(request, raw_before, [*models, created])
|
|
except Exception:
|
|
if capability_changed and capability_store is not None:
|
|
await _restore_capability(capability_store, body.name, previous_capability, request)
|
|
raise
|
|
return _to_admin_item(
|
|
created,
|
|
supports_reasoning_effort=bool(body.supports_reasoning_effort) if capability_changed else False,
|
|
default_reasoning_effort="xhigh",
|
|
)
|
|
|
|
|
|
@router.put("/order", response_model=list[ModelAdminItem])
|
|
async def reorder_model_configurations(body: ModelOrderRequest, request: Request) -> list[ModelAdminItem]:
|
|
"""Persist the configured model order.
|
|
|
|
The first item is intentionally the deployment default: public model
|
|
selectors preserve the ``config.yaml`` order and use their first option
|
|
when no model was explicitly chosen. Requiring a complete permutation
|
|
prevents a stale browser from accidentally dropping a newly added model.
|
|
"""
|
|
|
|
await _require_admin(request)
|
|
if len(body.names) != len(set(body.names)):
|
|
raise HTTPException(status_code=422, detail="Each model may appear only once in the order")
|
|
|
|
async with _MUTATION_LOCK:
|
|
path = _resolve_config_path()
|
|
raw_before, _, models = await asyncio.to_thread(_read_config_data, path)
|
|
configured_names = [str(item.get("name") or "") for item in models]
|
|
if len(body.names) != len(configured_names) or set(body.names) != set(configured_names):
|
|
raise HTTPException(
|
|
status_code=409,
|
|
detail="The model list changed. Refresh the page and try again.",
|
|
)
|
|
models_by_name = {str(item.get("name")): item for item in models}
|
|
ordered_models = [models_by_name[name] for name in body.names]
|
|
await _save_models(request, raw_before, ordered_models)
|
|
|
|
capabilities = await _capabilities_by_name(request)
|
|
return [
|
|
_to_admin_item(
|
|
item,
|
|
supports_reasoning_effort=bool(capabilities.get(str(item.get("name") or ""), {}).get("supports_reasoning_effort", False)),
|
|
default_reasoning_effort=str(capabilities.get(str(item.get("name") or ""), {}).get("default_reasoning_effort") or "xhigh"),
|
|
)
|
|
for item in ordered_models
|
|
]
|
|
|
|
|
|
@router.put(
|
|
"/capabilities/reasoning-effort/{model_name:path}",
|
|
response_model=ModelAdminItem,
|
|
)
|
|
async def update_reasoning_effort_capability(
|
|
model_name: str,
|
|
body: ReasoningEffortCapabilityRequest,
|
|
request: Request,
|
|
) -> ModelAdminItem:
|
|
"""Persist the per-model reasoning-effort selector capability in SQL."""
|
|
|
|
from app.gateway.catalog_cache import invalidate_catalog
|
|
|
|
admin = await _require_admin(request)
|
|
async with _MUTATION_LOCK:
|
|
path = _resolve_config_path()
|
|
_, _, models = await asyncio.to_thread(_read_config_data, path)
|
|
model = next((item for item in models if str(item.get("name")) == model_name), None)
|
|
if model is None:
|
|
raise HTTPException(status_code=404, detail=f"Model '{model_name}' was not found")
|
|
_validate_reasoning_capability(model, body.supports_reasoning_effort)
|
|
store = _get_capability_store(request)
|
|
assert store is not None
|
|
await store.set_reasoning_effort_support(
|
|
model_name,
|
|
supports_reasoning_effort=body.supports_reasoning_effort,
|
|
default_reasoning_effort=body.default_reasoning_effort,
|
|
updated_by=_admin_user_id(admin),
|
|
)
|
|
await invalidate_catalog(request, "models")
|
|
return _to_admin_item(
|
|
model,
|
|
supports_reasoning_effort=body.supports_reasoning_effort,
|
|
default_reasoning_effort=body.default_reasoning_effort or "xhigh",
|
|
)
|
|
|
|
|
|
@router.put(
|
|
"/write-file-transport/{model_name:path}",
|
|
response_model=ModelAdminItem,
|
|
)
|
|
async def update_write_file_transport(
|
|
model_name: str,
|
|
body: WriteFileTransportToggleRequest,
|
|
request: Request,
|
|
) -> ModelAdminItem:
|
|
"""Toggle one model's buffered-gateway write_file compatibility in config.yaml.
|
|
|
|
The global system-settings switch keeps overriding every compatible model
|
|
while it is enabled; this per-model value becomes effective as soon as the
|
|
global switch is turned off.
|
|
"""
|
|
|
|
await _require_admin(request)
|
|
async with _MUTATION_LOCK:
|
|
path = _resolve_config_path()
|
|
raw_before, _, models = await asyncio.to_thread(_read_config_data, path)
|
|
index = next((i for i, item in enumerate(models) if str(item.get("name")) == model_name), None)
|
|
if index is None:
|
|
raise HTTPException(status_code=404, detail=f"Model '{model_name}' was not found")
|
|
updated = copy.deepcopy(models[index])
|
|
_validate_write_file_transport(updated, body.write_file_transport)
|
|
updated["write_file_transport"] = body.write_file_transport
|
|
try:
|
|
ModelConfig.model_validate(updated)
|
|
except Exception as exc:
|
|
raise HTTPException(status_code=422, detail=f"Invalid model configuration: {exc}") from exc
|
|
next_models = [*models]
|
|
next_models[index] = updated
|
|
await _save_models(request, raw_before, next_models)
|
|
capabilities = await _capabilities_by_name(request)
|
|
capability = capabilities.get(model_name, {})
|
|
return _to_admin_item(
|
|
updated,
|
|
supports_reasoning_effort=bool(capability.get("supports_reasoning_effort", False)),
|
|
default_reasoning_effort=str(capability.get("default_reasoning_effort") or "xhigh"),
|
|
)
|
|
|
|
|
|
@router.put(
|
|
"/strict-stream-integrity/{model_name:path}",
|
|
response_model=ModelAdminItem,
|
|
)
|
|
async def update_strict_stream_integrity(
|
|
model_name: str,
|
|
body: StrictStreamIntegrityToggleRequest,
|
|
request: Request,
|
|
) -> ModelAdminItem:
|
|
"""Toggle one model's strict stream integrity check in config.yaml.
|
|
|
|
When enabled, a provider stream that ends without a terminal frame raises
|
|
a retriable ``IncompleteChatStreamError`` instead of committing a half
|
|
answer — the LLM error middleware then retries the call. Explicit ``false``
|
|
also overrides the auto-enable default for the affected qwen3.8-flash-next
|
|
model id; omitting the key entirely keeps that default.
|
|
"""
|
|
|
|
await _require_admin(request)
|
|
async with _MUTATION_LOCK:
|
|
path = _resolve_config_path()
|
|
raw_before, _, models = await asyncio.to_thread(_read_config_data, path)
|
|
index = next((i for i, item in enumerate(models) if str(item.get("name")) == model_name), None)
|
|
if index is None:
|
|
raise HTTPException(status_code=404, detail=f"Model '{model_name}' was not found")
|
|
updated = copy.deepcopy(models[index])
|
|
_validate_patched_openai_use(updated, "strict stream integrity")
|
|
updated["strict_stream_integrity"] = body.strict_stream_integrity
|
|
try:
|
|
ModelConfig.model_validate(updated)
|
|
except Exception as exc:
|
|
raise HTTPException(status_code=422, detail=f"Invalid model configuration: {exc}") from exc
|
|
next_models = [*models]
|
|
next_models[index] = updated
|
|
await _save_models(request, raw_before, next_models)
|
|
capabilities = await _capabilities_by_name(request)
|
|
capability = capabilities.get(model_name, {})
|
|
return _to_admin_item(
|
|
updated,
|
|
supports_reasoning_effort=bool(capability.get("supports_reasoning_effort", False)),
|
|
default_reasoning_effort=str(capability.get("default_reasoning_effort") or "xhigh"),
|
|
)
|
|
|
|
|
|
@router.put("/{model_name:path}", response_model=ModelAdminItem)
|
|
async def update_model_configuration(model_name: str, body: ModelUpsertRequest, request: Request) -> ModelAdminItem:
|
|
admin = await _require_admin(request)
|
|
if body.name != model_name:
|
|
raise HTTPException(status_code=422, detail="A model's stable name cannot be changed; edit its display name instead")
|
|
async with _MUTATION_LOCK:
|
|
path = _resolve_config_path()
|
|
raw_before, _, models = await asyncio.to_thread(_read_config_data, path)
|
|
index = next((i for i, item in enumerate(models) if str(item.get("name")) == model_name), None)
|
|
if index is None:
|
|
raise HTTPException(status_code=404, detail=f"Model '{model_name}' was not found")
|
|
updated = _request_to_raw_model(body, models[index])
|
|
capability_explicit = "supports_reasoning_effort" in body.model_fields_set and body.supports_reasoning_effort is not None
|
|
capability_store = _get_capability_store(request, required=capability_explicit)
|
|
previous_capability = await capability_store.get_capability(model_name) if capability_store is not None else None
|
|
auto_disable_capability = bool(capability_store is not None and not bool(updated.get("supports_thinking", False)) and previous_capability and previous_capability.get("supports_reasoning_effort", False))
|
|
capability_changed = capability_explicit or auto_disable_capability
|
|
desired_capability = bool(body.supports_reasoning_effort) if capability_explicit else False
|
|
if capability_changed and capability_store is not None:
|
|
_validate_reasoning_capability(updated, desired_capability)
|
|
await capability_store.set_reasoning_effort_support(
|
|
model_name,
|
|
supports_reasoning_effort=desired_capability,
|
|
default_reasoning_effort=str((previous_capability or {}).get("default_reasoning_effort") or "xhigh"),
|
|
updated_by=_admin_user_id(admin),
|
|
)
|
|
next_models = [*models]
|
|
next_models[index] = updated
|
|
try:
|
|
await _save_models(request, raw_before, next_models)
|
|
except Exception:
|
|
if capability_changed and capability_store is not None:
|
|
await _restore_capability(capability_store, model_name, previous_capability, request)
|
|
raise
|
|
if capability_changed:
|
|
supports_reasoning_effort = desired_capability
|
|
elif capability_store is not None:
|
|
current_capability = await capability_store.get_capability(model_name)
|
|
supports_reasoning_effort = bool(current_capability and current_capability.get("supports_reasoning_effort", False))
|
|
else:
|
|
supports_reasoning_effort = False
|
|
return _to_admin_item(
|
|
updated,
|
|
supports_reasoning_effort=supports_reasoning_effort,
|
|
default_reasoning_effort=str((previous_capability or {}).get("default_reasoning_effort") or "xhigh"),
|
|
)
|
|
|
|
|
|
@router.delete("/{model_name:path}", status_code=204)
|
|
async def delete_model_configuration(model_name: str, request: Request) -> None:
|
|
"""Delete one model configuration and activate the remaining model list."""
|
|
|
|
await _require_admin(request)
|
|
async with _MUTATION_LOCK:
|
|
path = _resolve_config_path()
|
|
raw_before, _, models = await asyncio.to_thread(_read_config_data, path)
|
|
index = next((i for i, item in enumerate(models) if str(item.get("name")) == model_name), None)
|
|
if index is None:
|
|
raise HTTPException(status_code=404, detail=f"Model '{model_name}' was not found")
|
|
capability_store = _get_capability_store(request, required=False)
|
|
previous_capability = await capability_store.get_capability(model_name) if capability_store is not None else None
|
|
if capability_store is not None:
|
|
await capability_store.delete_capability(model_name)
|
|
try:
|
|
await _save_models(request, raw_before, [*models[:index], *models[index + 1 :]])
|
|
except Exception:
|
|
if capability_store is not None:
|
|
await _restore_capability(capability_store, model_name, previous_capability, request)
|
|
raise
|
|
|
|
|
|
@router.post("/test", response_model=ModelTestResponse)
|
|
async def test_model_connection(body: ModelTestRequest, request: Request) -> ModelTestResponse:
|
|
"""Perform one bounded minimal completion without persisting the draft."""
|
|
|
|
await _require_admin(request)
|
|
path = _resolve_config_path()
|
|
_, _, models = await asyncio.to_thread(_read_config_data, path)
|
|
existing = next((item for item in models if str(item.get("name")) == body.model.name), None)
|
|
candidate = _request_to_raw_model(body.model, existing)
|
|
candidate_secret = str(candidate.get("api_key") or "")
|
|
secrets = [candidate_secret]
|
|
try:
|
|
runtime_candidate = AppConfig.resolve_env_variables(copy.deepcopy(candidate))
|
|
# An environment-variable API key is resolved only in memory for this
|
|
# request. Redact both forms because provider exceptions can include
|
|
# either the original `$NAME` reference or its resolved value.
|
|
resolved_secret = str(runtime_candidate.get("api_key") or "")
|
|
secrets = [candidate_secret, resolved_secret]
|
|
model_config = ModelConfig.model_validate(runtime_candidate)
|
|
base_config = request.app.state.config
|
|
test_config = base_config.model_copy(update={"models": [model_config]})
|
|
chat_model = create_chat_model(model_config.name, app_config=test_config, force_disable_thinking=True)
|
|
started = perf_counter()
|
|
response = await asyncio.wait_for(
|
|
chat_model.ainvoke([HumanMessage(content="Reply with exactly: OK")]),
|
|
timeout=min(float(candidate.get("request_timeout") or 30), 45),
|
|
)
|
|
latency_ms = round((perf_counter() - started) * 1000)
|
|
preview = _redact_text(str(getattr(response, "content", "OK")), secrets)
|
|
return ModelTestResponse(success=True, latency_ms=latency_ms, message="Connection successful", preview=preview)
|
|
except Exception as exc:
|
|
# Provider errors sometimes echo a credential. Never log the full
|
|
# exception or return it without redacting the submitted key.
|
|
logger.warning("Model connection test failed for '%s': %s", candidate.get("name"), type(exc).__name__)
|
|
return ModelTestResponse(
|
|
success=False,
|
|
latency_ms=0,
|
|
message=_redact_text(str(exc) or type(exc).__name__, secrets),
|
|
)
|