deerflow-code/offline-backend-20260512/backend/packages/harness/deerflow/config/system_settings.py

"""System-wide runtime settings editable from the admin UI.

Backed by a single JSON file under the DeerFlow runtime home (default
``.deer-flow/system_settings.json``). Unlike :mod:`extensions_config` this
file is **not** part of the project source tree — it is per-deployment
state, and it stays writable at runtime.

The only feature stored here at the moment is the **prompt prefix toggle**
shown in the chat input: an administrator can configure a label and a
prefix string, and any user can flip the toggle on/off; when enabled, the
prefix is prepended to the first human message of a new thread.
"""

from __future__ import annotations

import json
import logging
import re
import threading
from pathlib import Path
from typing import Literal
from uuid import uuid4

from pydantic import BaseModel, ConfigDict, Field, field_validator

from deerflow.config.llmwiki_config import LlmWikiRuntimeOverride
from deerflow.config.runtime_paths import runtime_home

logger = logging.getLogger(__name__)

_SETTINGS_FILE = "system_settings.json"
_lock = threading.Lock()

# Built-ins remain source-controlled, while administrator-created themes use a
# stable prefixed id and are persisted in the runtime settings JSON.
BUILTIN_THEME_IDS = frozenset({"light", "dark", "lightblue", "lightgreen", "iceblue", "lightyellow"})
CUSTOM_THEME_ID_PREFIX = "custom-theme-"
_CUSTOM_THEME_ID_RE = re.compile(r"^custom-theme-[a-z0-9-]{8,64}$")
_HEX_COLOR_RE = re.compile(r"^#[0-9a-fA-F]{6}$")


def _normalize_optional_hex_color(value: str | None, field_name: str) -> str | None:
    """Accept an optional, safe six-digit colour without allowing raw CSS."""

    if value is None or not str(value).strip():
        return None
    normalized = str(value).strip().upper()
    if not _HEX_COLOR_RE.fullmatch(normalized):
        raise ValueError(f"{field_name} must be a six-digit hexadecimal value such as #0062D9")
    return normalized


class CustomThemeColorOverrides(BaseModel):
    """Optional advanced adjustments to an otherwise generated colour scale.

    The short list intentionally covers interaction surfaces only. It avoids
    turning the theme editor into a fragile form containing every CSS token.
    """

    button_hover: str | None = Field(default=None, description="Primary button hover colour")
    navigation_gradient_start: str | None = Field(
        default=None,
        description="Left colour for the configurable top-navigation gradient",
    )
    navigation_gradient_end: str | None = Field(
        default=None,
        description="Right colour for the configurable top-navigation gradient",
    )
    selected_background: str | None = Field(default=None, description="Selected/active item background")
    border: str | None = Field(default=None, description="Card and input border colour")
    muted: str | None = Field(default=None, description="Muted/secondary surface colour")
    muted_text: str | None = Field(default=None, description="Secondary text colour")
    destructive: str | None = Field(default=None, description="Delete and danger action colour")
    brand_title: str | None = Field(default=None, description="Branded section-title text colour")
    primary_foreground: str | None = Field(
        default=None,
        description="Text colour on primary surfaces such as the logo and status badges",
    )
    model_config = ConfigDict(extra="ignore")

    @field_validator(
        "button_hover",
        "navigation_gradient_start",
        "navigation_gradient_end",
        "selected_background",
        "border",
        "muted",
        "muted_text",
        "destructive",
        "brand_title",
        "primary_foreground",
    )
    @classmethod
    def _validate_color(cls, value: str | None, info) -> str | None:
        return _normalize_optional_hex_color(value, f"Theme override {info.field_name}")


class CustomThemePreset(BaseModel):
    """One administrator-managed theme generated from a single brand color.

    The frontend derives all surface, text, focus and hover tokens from these
    safe inputs at runtime. Storing only the brand color and base mode keeps
    arbitrary CSS out of runtime settings and lets contrast rules remain
    centrally enforced.
    """

    id: str = Field(default_factory=lambda: f"{CUSTOM_THEME_ID_PREFIX}{uuid4().hex}")
    name: str = Field(default="自定义主题", min_length=1, max_length=32)
    primary: str = Field(default="#0062D9", description="Six-digit CSS hexadecimal brand color")
    text_color: str | None = Field(
        default=None,
        description="Optional primary-button text color; omitted = automatic button foreground",
    )
    mode: Literal["light", "dark"] = Field(default="light", description="Surface family used to derive the palette")
    overrides: CustomThemeColorOverrides = Field(default_factory=CustomThemeColorOverrides)
    enabled: bool = Field(default=True)
    sort_order: int = Field(default=0, ge=0, le=999)
    model_config = ConfigDict(extra="ignore")

    @field_validator("id")
    @classmethod
    def _validate_id(cls, value: str) -> str:
        normalized = str(value or "").strip().lower()
        if not _CUSTOM_THEME_ID_RE.fullmatch(normalized):
            raise ValueError("Custom theme id must use the custom-theme- prefix")
        return normalized

    @field_validator("name")
    @classmethod
    def _normalize_name(cls, value: str) -> str:
        normalized = str(value or "").strip()
        if not normalized:
            raise ValueError("Theme name cannot be empty")
        return normalized

    @field_validator("primary")
    @classmethod
    def _validate_primary(cls, value: str) -> str:
        normalized = str(value or "").strip().upper()
        if not _HEX_COLOR_RE.fullmatch(normalized):
            raise ValueError("Theme primary color must be a six-digit hexadecimal value such as #0062D9")
        return normalized

    @field_validator("text_color")
    @classmethod
    def _validate_text_color(cls, value: str | None) -> str | None:
        return _normalize_optional_hex_color(value, "Theme text color")


class PromptPrefixSettings(BaseModel):
    """Admin-configurable first-message prompt prefix."""

    enabled: bool = Field(default=False, description="Whether the prefix toggle is offered to users at all")
    switch_label: str = Field(default="加入立场前缀", description="Toggle label shown in the chat input")
    prompt_prefix: str = Field(default="", description="Text prepended to the first user message when enabled")
    ordinary_qa_markdown_format_enabled: bool = Field(
        default=False,
        description="Whether the formal Markdown heading rules are injected into ordinary Q&A system prompts.",
    )
    model_config = ConfigDict(extra="ignore")


class SkillKnowledgeAutoSyncSettings(BaseModel):
    """Watcher/scheduler policy for bindings that explicitly opt in."""

    watch_enabled: bool = True
    schedule_enabled: bool = True
    schedule_cron: str = Field(default="0 */6 * * *", min_length=5, max_length=128)
    debounce_seconds: int = Field(default=120, ge=10, le=86_400)
    write_remote_enabled: bool = True
    review_mode: Literal["required", "auto_high_confidence", "off"] = "auto_high_confidence"
    model_config = ConfigDict(extra="ignore")


class AppearanceRouteParam(BaseModel):
    """Runtime query-param descriptor for top-bar shortcut buttons.

    The frontend resolves login params from ``userInfo`` and theme params from
    the current UI theme, matching the light-app center's link builder.
    """

    key: str = Field(default="", description="Query parameter name appended to the shortcut URL; path parameters do not use a key")
    paramType: Literal["login", "theme", "path"] = Field(default="login", description="Whether the value is a login query parameter, a theme query parameter, or a path segment")
    source: str | None = Field(default=None, description="login param source field in userInfo, or 'other'")
    customValue: str | None = Field(default=None, description="Literal value when source is 'other'")
    themeValue: str | None = Field(default=None, description="Theme-value mapping string, e.g. red=red;dark-blue=dark")
    model_config = ConfigDict(extra="ignore")


def _default_shortcut_route_params() -> list[AppearanceRouteParam]:
    return [
        AppearanceRouteParam(
            key="access_token",
            paramType="login",
            source="access_token",
            customValue="",
            themeValue="red=red;green=green;blue=blue;dark-blue=dark-blue",
        )
    ]


class CompactShortcutButton(BaseModel):
    """Configurable shortcut shown in top bars before the theme switcher."""

    id: str = Field(default_factory=lambda: uuid4().hex, description="Stable row id for editing")
    label: str = Field(default="快捷入口", description="Button text shown next to the icon")
    url: str = Field(default="", description="Target URL opened in a new tab")
    route_params: list[AppearanceRouteParam] = Field(
        default_factory=_default_shortcut_route_params,
        description="Login/theme query parameters or keyless path segments resolved at launch time",
    )
    enabled: bool = Field(default=True, description="Whether this shortcut is rendered")
    sort_order: int = Field(default=0, description="Display order in top bars")
    model_config = ConfigDict(extra="ignore")


def _default_compact_shortcut_buttons() -> list[CompactShortcutButton]:
    return [
        CompactShortcutButton(
            id="default-compact-shortcut",
            label="快捷入口",
            url="",
            enabled=True,
            sort_order=0,
        )
    ]


class AppearanceSettings(BaseModel):
    """System-wide appearance defaults that any *new* user inherits.

    Once a user explicitly toggles their theme in the UI, that choice is
    persisted in the browser and supersedes this default — so changing
    ``default_theme`` here only affects users who have never touched the
    toggle, and any future first-time loads.
    """

    default_theme: str = Field(
        default="light",
        description="Theme applied for users who haven't picked their own (a built-in id or custom-theme-* id)",
    )
    default_skill_view: Literal["card", "list"] = Field(
        default="list",
        description="Default skill list display mode for users who haven't picked their own (card / list)",
    )
    default_agent_view: Literal["card", "list"] = Field(
        default="card",
        description="Default agent list display mode for users who haven't picked their own (card / list)",
    )
    default_layout_mode: Literal["full", "compact"] = Field(
        default="compact",
        description="Default interface layout for users who haven't picked their own — compact (hidden) / full (sidebar + header)",
    )
    compact_shortcut_buttons: list[CompactShortcutButton] = Field(
        default_factory=_default_compact_shortcut_buttons,
        description="Shortcut buttons rendered before the theme switcher in top bars",
    )
    custom_themes: list[CustomThemePreset] = Field(
        default_factory=list,
        description="Administrator-managed runtime themes available to all users",
    )
    model_config = ConfigDict(extra="ignore")

    @field_validator("default_theme")
    @classmethod
    def _validate_default_theme(cls, value: str) -> str:
        normalized = str(value or "").strip().lower()
        if normalized in BUILTIN_THEME_IDS or _CUSTOM_THEME_ID_RE.fullmatch(normalized):
            return normalized
        raise ValueError("Unknown theme id")


class CitationDisplaySettings(BaseModel):
    """System-wide controls for citation mode entry and display cleanup."""

    reference_mode_enabled: bool = Field(
        default=False,
        description="Whether the reference-mode entry points are exposed to users.",
    )
    cleanup_words: list[str] = Field(
        default_factory=list,
        description="Words or phrases used to remove noisy advertisement/disclaimer lines from citation display.",
    )
    model_config = ConfigDict(extra="ignore")


class PrivateSquareSettings(BaseModel):
    """Admin-configurable settings for the agent private-square (私密广场).

    The password gates read access to all agents flagged ``is_private=True``.
    The hash is stored opaquely — the router layer is responsible for hashing
    and verifying so the harness package keeps zero dependencies on
    ``app.gateway.auth``.
    """

    enabled: bool = Field(default=False, description="Whether the private square is offered to users at all")
    password_hash: str = Field(default="", description="Opaque hash of the admin-configured password. Empty = no password set yet.")
    model_config = ConfigDict(extra="ignore")


#: Id of the always-present built-in default square (used when the admin has
#: configured no squares, or when a published item picks none).
DEFAULT_SQUARE_ID = "default"


class PublishSquare(BaseModel):
    """One admin-configured publish square (发布广场).

    A pure organizational category that published agents/skills can be filed
    under — unlike the password-gated :class:`PrivateSquareSettings`, a square
    has **no access control**: every logged-in user can browse every square.
    Each agent/skill belongs to at most one square (a single ``square_id``
    field on the row), so an item can never be "published twice".
    """

    id: str = Field(..., description="Stable square id (slug/uuid); used as agents.square_id / skills.square_id")
    name: str = Field(..., description="Display name shown as the square's tab label")
    description: str = Field(default="", description="Optional description shown in the admin config / tab hint")
    model_config = ConfigDict(extra="ignore")


class PublishSquaresSettings(BaseModel):
    """Admin-configurable list of publish squares.

    ``squares`` is display-ordered. ``default_square_id`` selects the square a
    publisher lands in when they do not pick one explicitly. There is always at
    least one effective square — :func:`list_publish_squares` injects a built-in
    "默认广场" when the admin has configured none, so the publish flow always has
    a valid target.
    """

    squares: list[PublishSquare] = Field(default_factory=list, description="Ordered list of configured squares")
    default_square_id: str = Field(default=DEFAULT_SQUARE_ID, description="Square id used when a publisher picks none")
    model_config = ConfigDict(extra="ignore")


def list_publish_squares(settings: PublishSquaresSettings) -> list[PublishSquare]:
    """Return the effective square list, always non-empty.

    Falls back to a single built-in default square when the admin has not
    configured any, so callers never have to special-case the empty state.
    """
    if settings.squares:
        return list(settings.squares)
    return [PublishSquare(id=DEFAULT_SQUARE_ID, name="默认广场", description="")]


def resolve_default_square_id(settings: PublishSquaresSettings) -> str:
    """Resolve the effective default square id against the configured set."""
    ids = [s.id for s in settings.squares]
    if settings.default_square_id and settings.default_square_id in ids:
        return settings.default_square_id
    return ids[0] if ids else DEFAULT_SQUARE_ID


def normalize_square_id(square_id: str | None, settings: PublishSquaresSettings) -> str:
    """Coerce an arbitrary ``square_id`` to a valid configured square id.

    Empty / unknown / orphaned (square later deleted) ids all collapse to the
    effective default square, so a published item always resolves to a real
    tab.
    """
    default_id = resolve_default_square_id(settings)
    if not square_id:
        return default_id
    valid = {s.id for s in list_publish_squares(settings)}
    return square_id if square_id in valid else default_id


class LeaderboardSettings(BaseModel):
    """Admin-configurable defaults for the user leaderboard analytics."""

    exclude_user_ids: list[str] = Field(default_factory=list, description="User ids excluded from leaderboard aggregation.")
    include_scheduled: bool = Field(default=True, description="Whether scheduled task runs participate in leaderboard statistics.")
    include_admins: bool = Field(default=True, description="Whether admin accounts participate in leaderboard statistics.")
    include_failed: bool = Field(default=True, description="Whether failed runs and provider calls participate in leaderboard statistics.")
    cleanup_system_questions: bool = Field(default=True, description="Hide attachment/system marker questions such as <uploaded_files>.")
    model_config = ConfigDict(extra="ignore")


class SkillCompressionSettings(BaseModel):
    """Admin-configurable skill compression (技能压缩).

    When ``enabled``, only skills explicitly marked 常驻(免压缩) (``always_on``)
    are injected full-text into the lead agent's system prompt. Every other
    enabled skill is "compressed": kept out of the prompt body and surfaced to
    the agent on demand via the ``search_skills`` keyword-retrieval tool (and a
    compact name index when ``keep_index_in_prompt`` is true). When disabled the
    behaviour is unchanged — every enabled skill is injected full-text.
    """

    enabled: bool = Field(default=False, description="Whether skill compression is active (master switch)")
    search_top_n: int = Field(default=5, ge=1, le=50, description="Max skills returned by the search_skills tool")
    keep_index_in_prompt: bool = Field(
        default=True,
        description="Whether compressed skills still appear as a compact name-only index in the prompt",
    )
    model_config = ConfigDict(extra="ignore")


class AgentModuleSettings(BaseModel):
    """Admin-configured「按类型聚合」module → selected agent-tag ids.

    Keyed by module key (e.g. ``"zzjc"`` / ``"zzxd"``). Each value is the list
    of agent **tag ids** whose agents that module's left sidebar shows. A missing
    key / empty list means "not configured" — the frontend falls back to showing
    all visible agents grouped by tag.
    """

    modules: dict[str, list[str]] = Field(
        default_factory=dict,
        description="module key → selected agent tag ids",
    )
    model_config = ConfigDict(extra="ignore")


class PluginDomainAgentMapping(BaseModel):
    """One configured host -> agent mapping for iframe chat auto-selection."""

    id: str = Field(..., description="Stable row id for client-side editing")
    domain: str = Field(..., description="Normalized host matched against pageContext.url")
    agent_id: str = Field(..., description="Agent id selected when the host matches")
    enabled: bool = Field(default=True, description="Whether this mapping is active")
    model_config = ConfigDict(extra="ignore")


class PluginManagementSettings(BaseModel):
    """Admin-configurable iframe plugin mappings."""

    domain_agent_mappings: list[PluginDomainAgentMapping] = Field(
        default_factory=list,
        description="Ordered host -> agent mappings used by the iframe chat entry page.",
    )
    model_config = ConfigDict(extra="ignore")


def get_module_tag_ids(settings: AgentModuleSettings, key: str) -> list[str]:
    """Return the configured agent-tag ids for ``key`` (empty when unset)."""
    return list(settings.modules.get(key, []))


def set_module_tag_ids(settings: AgentModuleSettings, key: str, tag_ids: list[str]) -> AgentModuleSettings:
    """Return a copy with ``key`` set to a de-duplicated, stripped tag-id list."""
    cleaned = list(dict.fromkeys([str(t).strip() for t in tag_ids if str(t).strip()]))
    modules = dict(settings.modules)
    modules[key] = cleaned
    return settings.model_copy(update={"modules": modules})


class LoginWhitelistSettings(BaseModel):
    """登录白名单总开关(管理员可在「白名单管理」页切换,即时生效、免重启)。

    黑名单模式:``enabled`` 关(默认) → 忽略每个用户的 ``approved`` 放行标记,
    人人可登录;开 → 仅被管理员显式「取消放行」(``approved=False``) 的用户被 403,
    新账号默认 ``approved=True`` 故从未登录过的用户照样放行,admin 永远放行。
    空/默认即关,故部署后行为零变化。
    """

    enabled: bool = Field(default=False, description="是否启用登录白名单;关=所有人可登录,开=仅已放行用户(或 admin)可登录")
    model_config = ConfigDict(extra="ignore")


class ConcurrencyMonitorSettings(BaseModel):
    """管理员「实时并发监控」开关(排行榜页的系统压力监控)。

    ``enabled`` 控制**后台并发采样器**是否持续把当前并发调用数写库(供并发量
    折线图按分钟/小时聚合)。**默认关**(opt-in,由管理员在排行榜页手动开启),
    以对线上服务零影响为先;开启后采样器才开始写库,关闭则立即停止(每 tick 都
    重读该开关,免重启),实时快照查询与「停止」始终可用、零常驻成本。设计上即便
    开启也对线上零影响:采样仅一条极小 INSERT、有保留期清理、聚合查询有行数上限、
    全程 best-effort。
    """

    enabled: bool = Field(default=False, description="是否启用后台并发采样(默认关;开→开始写库,关→停止写库、折线图不再新增历史点)")
    model_config = ConfigDict(extra="ignore")


class ModelConcurrencySettings(BaseModel):
    """Admin-configurable per-model concurrent Q&A limits."""

    enabled: bool = Field(
        default=False,
        description="Whether per-model concurrency limiting is enabled.",
    )
    default_limit: int = Field(
        default=10,
        ge=1,
        le=10_000,
        description="Fallback max concurrent runs for models without an explicit override.",
    )
    model_limits: dict[str, int] = Field(
        default_factory=dict,
        description="Optional per-model overrides: model_name -> max concurrent runs.",
    )
    user_default_limit: int = Field(
        default=0,
        ge=0,
        le=10_000,
        description="Fallback max concurrent runs per user and model. 0 disables the per-user fallback.",
    )
    user_model_limits: dict[str, int] = Field(
        default_factory=dict,
        description="Optional per-user per-model overrides: model_name -> max concurrent runs for one user.",
    )
    model_config = ConfigDict(extra="ignore")


class LocalWikiEncoderSettings(BaseModel):
    """Runtime override for the administrator-managed offline Wiki encoder."""

    model_path: str = Field(
        default="",
        max_length=4096,
        description="Absolute or project-relative directory containing the offline BGE-M3 ONNX model.",
    )
    model_config = ConfigDict(extra="ignore")


class SystemSettings(BaseModel):
    """Top-level system settings document."""

    prompt_prefix: PromptPrefixSettings = Field(default_factory=PromptPrefixSettings)
    login_whitelist: LoginWhitelistSettings = Field(default_factory=LoginWhitelistSettings)
    private_square: PrivateSquareSettings = Field(default_factory=PrivateSquareSettings)
    publish_squares: PublishSquaresSettings = Field(default_factory=PublishSquaresSettings)
    appearance: AppearanceSettings = Field(default_factory=AppearanceSettings)
    citation_display: CitationDisplaySettings = Field(default_factory=CitationDisplaySettings)
    leaderboard: LeaderboardSettings = Field(default_factory=LeaderboardSettings)
    skill_compression: SkillCompressionSettings = Field(default_factory=SkillCompressionSettings)
    plugin_management: PluginManagementSettings = Field(default_factory=PluginManagementSettings)
    agent_modules: AgentModuleSettings = Field(default_factory=AgentModuleSettings)
    concurrency_monitor: ConcurrencyMonitorSettings = Field(default_factory=ConcurrencyMonitorSettings)
    model_concurrency: ModelConcurrencySettings = Field(default_factory=ModelConcurrencySettings)
    local_wiki_encoder: LocalWikiEncoderSettings = Field(default_factory=LocalWikiEncoderSettings)
    llmwiki: LlmWikiRuntimeOverride = Field(default_factory=LlmWikiRuntimeOverride)
    skill_knowledge_auto_sync: SkillKnowledgeAutoSyncSettings = Field(default_factory=SkillKnowledgeAutoSyncSettings)
    knowledge_base_product_version: Literal["v1", "v2"] = Field(
        default="v1",
        description="v1 opens the ordinary WeKnora experience; v2 opens the DeerFlow assistant knowledge base.",
    )
    ordinary_knowledge_landing: Literal["wiki", "weknora"] = "wiki"
    model_config = ConfigDict(extra="ignore")


def _settings_path() -> Path:
    home = runtime_home()
    home.mkdir(parents=True, exist_ok=True)
    return home / _SETTINGS_FILE


def load_system_settings() -> SystemSettings:
    """Read the on-disk system settings, returning defaults when missing."""
    path = _settings_path()
    if not path.is_file():
        return SystemSettings()
    try:
        raw = json.loads(path.read_text(encoding="utf-8"))
        if not isinstance(raw, dict):
            logger.warning("System settings file %s is not a JSON object; ignoring", path)
            return SystemSettings()
        return SystemSettings.model_validate(raw)
    except Exception as exc:
        logger.warning("Failed to read system settings from %s: %s", path, exc)
        return SystemSettings()


def get_skill_compression_settings() -> SkillCompressionSettings:
    """Convenience accessor for the skill-compression block (safe defaults on error)."""
    try:
        return load_system_settings().skill_compression
    except Exception:
        logger.debug("Failed to read skill compression settings; using defaults", exc_info=True)
        return SkillCompressionSettings()


def get_concurrency_monitor_settings() -> ConcurrencyMonitorSettings:
    """Convenience accessor for the concurrency-monitor block (safe defaults on error)."""
    try:
        return load_system_settings().concurrency_monitor
    except Exception:
        logger.debug("Failed to read concurrency monitor settings; using defaults", exc_info=True)
        return ConcurrencyMonitorSettings()


def get_model_concurrency_settings() -> ModelConcurrencySettings:
    """Convenience accessor for the model-concurrency block (safe defaults on error)."""
    try:
        return load_system_settings().model_concurrency
    except Exception:
        logger.debug("Failed to read model concurrency settings; using defaults", exc_info=True)
        return ModelConcurrencySettings()


def save_system_settings(settings: SystemSettings) -> SystemSettings:
    """Persist the settings document atomically, returning the saved value."""
    path = _settings_path()
    payload = settings.model_dump(mode="json")
    with _lock:
        tmp = path.with_suffix(path.suffix + ".tmp")
        tmp.write_text(json.dumps(payload, ensure_ascii=False, indent=2), encoding="utf-8")
        tmp.replace(path)
    return settings