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 threading
from pathlib import Path
from typing import Literal
from uuid import uuid4

from pydantic import BaseModel, ConfigDict, Field

from deerflow.config.runtime_paths import runtime_home

logger = logging.getLogger(__name__)

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


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")
    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")
    paramType: Literal["login", "theme"] = Field(default="login", description="Where the parameter value comes from")
    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;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="Query parameters resolved at launch time and appended to url",
    )
    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: Literal["light", "dark", "lightblue"] = Field(
        default="light",
        description="Theme applied for users who haven't picked their own (light / dark / lightblue)",
    )
    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="full",
        description="Default interface layout for users who haven't picked their own — full (sidebar + header) / compact (hidden)",
    )
    compact_shortcut_buttons: list[CompactShortcutButton] = Field(
        default_factory=_default_compact_shortcut_buttons,
        description="Shortcut buttons rendered before the theme switcher in top bars",
    )
    model_config = ConfigDict(extra="ignore")


class CitationDisplaySettings(BaseModel):
    """System-wide cleanup rules for citation cards and detail dialogs."""

    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.",
    )
    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)
    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