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


WordHeadingNumberingStyle = Literal[
    "hybrid_decimal",
    "official_chinese",
    "decimal",
    "chapter_section",
    "none",
]


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")
    image_generation_unavailable_hint_enabled: bool = Field(
        default=False,
        description="Tell users image generation is unavailable instead of attempting to generate images.",
    )
    python_dependency_delete_protection_enabled: bool = Field(
        default=False,
        description="Inject guidance against standalone Python dependency deletion; installation and replacement remain allowed.",
    )
    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.",
    )
    word_heading_numbering_style: WordHeadingNumberingStyle = Field(
        default="hybrid_decimal",
        description="Heading-numbering preset used by every Markdown-to-Word export.",
    )
    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)",
    )
    default_composer_width_full: int = Field(
        default=100,
        description="Default ordinary-chat composer width percent for full layout, used when a user has not chosen their own. 100 keeps the built-in width.",
    )
    default_composer_width_compact: int = Field(
        default=100,
        description="Default ordinary-chat composer width percent for compact layout, used when a user has not chosen their own. 100 keeps the built-in width.",
    )
    default_composer_height_full: int = Field(
        default=100,
        description="Default ordinary-chat composer height percent for full layout, used when a user has not chosen their own. 100 keeps the built-in height.",
    )
    default_composer_height_compact: int = Field(
        default=100,
        description="Default ordinary-chat composer height percent for compact layout, used when a user has not chosen their own. 100 keeps the built-in height.",
    )
    default_agent_composer_width: int = Field(
        default=100,
        description="Default agent-chat composer width percent, used when a user has not chosen their own. 100 keeps the built-in width; the page still caps it at the parent column.",
    )
    default_agent_composer_height: int = Field(
        default=100,
        description="Default agent-chat composer height percent, used when a user has not chosen their own. 100 keeps the built-in height.",
    )
    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")

    @staticmethod
    def _clamp_scale(value: object, low: int, high: int) -> int:
        if value is None or value == "":
            return 100
        try:
            parsed = int(round(float(value)))  # type: ignore[arg-type]
        except (TypeError, ValueError):
            return 100
        return min(high, max(low, parsed))

    @field_validator("default_composer_width_full", "default_composer_width_compact", mode="before")
    @classmethod
    def _clamp_default_composer_width(cls, value: object) -> int:
        return cls._clamp_scale(value, 60, 150)

    @field_validator(
        "default_composer_height_full",
        "default_composer_height_compact",
        "default_agent_composer_height",
        mode="before",
    )
    @classmethod
    def _clamp_default_composer_height(cls, value: object) -> int:
        return cls._clamp_scale(value, 70, 200)

    @field_validator("default_agent_composer_width", mode="before")
    @classmethod
    def _clamp_default_agent_composer_width(cls, value: object) -> int:
        return cls._clamp_scale(value, 60, 400)

    @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."""

    retrieved_files_enabled: bool = True
    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 SkillVersioningSettings(BaseModel):
    """Deployment-wide gate for custom-skill package version history.

    The default is intentionally disabled so existing deployments retain the
    legacy in-place editing behaviour until an administrator opts in.
    """

    enabled: bool = Field(
        default=False,
        description="Whether custom-skill package versioning and switching are enabled",
    )
    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 WxCommonAgentsSettings(BaseModel):
    """The gallery category shown in the wx chat route's common-agent panel.

    ``None`` keeps the original recommendation logic. ``mine`` and
    ``sentiment`` are built-in category values. Publish-square categories are
    persisted as ``square:<square_id>`` so they cannot collide with the
    built-ins.
    """

    agent_type: str | None = Field(
        default=None,
        description="Selected gallery category; null retains the default recommendation logic",
    )
    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 AuthenticationSettings(BaseModel):
    """Public login capabilities; administrator writes take effect immediately."""

    password_login_enabled: bool = False
    registration_enabled: bool = False
    sso_enabled: bool = True
    single_session_enabled: bool = False
    model_config = ConfigDict(extra="forbid")


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 MemoryRuntimeSettings(BaseModel):
    """Deployment-wide switch for generic/personal Q&A memory.

    Memory is deliberately opt-in.  When this setting is false the lead agent
    does not mount memory middleware or memory tools, does not inject stored
    generic memory into prompts, and does not enqueue generic background memory
    extraction.  Project-scoped Q&A memory is an independent subsystem governed
    by each project's own setting and is intentionally not controlled here.
    """

    enabled: bool = Field(
        default=False,
        description="Whether generic/personal Q&A memory recall, injection and persistence are enabled.",
    )
    model_config = ConfigDict(extra="ignore")


class ConversationContextSettings(BaseModel):
    """Deployment-wide short-context policy for chat model requests.

    The recent-turn window is deliberately opt-in.  When enabled, the lead
    agent keeps the persisted conversation intact but sends only the latest
    three real user turns (plus their assistant/tool continuations) to the
    model.  This avoids an extra summarization-model call while preserving the
    complete transcript for the UI and checkpoint recovery.
    """

    max_run_steps: int = Field(default=1000, ge=20, le=10000, strict=True)
    recent_turn_window_enabled: bool = Field(
        default=False,
        description="Whether model requests use the fixed recent-three-turn sliding window.",
    )
    max_conversation_rounds: int = Field(
        default=20,
        ge=1,
        le=50,
        description="Maximum real user turns allowed in one ordinary chat thread.",
    )
    use_question_as_title: bool = Field(
        default=False,
        description=(
            "Whether the first user question is used as the final conversation title, "
            "skipping post-answer LLM title generation."
        ),
    )
    model_config = ConfigDict(extra="ignore")


class WriteFileCompatibilitySettings(BaseModel):
    """Deployment-wide compatibility for gateways buffering tool arguments."""

    enabled: bool = Field(
        default=False,
        description=(
            "Whether every compatible model uses the buffered-gateway "
            "write_file streaming adapter."
        ),
    )
    model_config = ConfigDict(extra="ignore")


class CallMetricsSettings(BaseModel):
    """Administrator-controlled recording for tool and skill call statistics.

    Both rankings use the same ``tool_call_metrics`` audit stream. Keeping a
    single switch prevents a partially enabled configuration where the
    management page can show one ranking but not the other.
    """

    enabled: bool = Field(
        default=True,
        description=(
            "Whether new tool and skill calls are persisted for management "
            "statistics. Existing records are retained when disabled."
        ),
    )
    model_config = ConfigDict(extra="ignore")


class RetryKeepPartialSettings(BaseModel):
    """Keep already-streamed partial answers visible across silent LLM retries.

    The LLM error middleware silently retries failed model calls. The failed
    attempt never reaches the checkpoint, so the display reconciliation drops
    its already-streamed partial answer — users see the text "cleared" when
    the retry starts thinking again. Enabled keeps that partial visible in
    the chat (frontend display concern, read at stream time).
    """

    enabled: bool = Field(
        default=True,
        description=(
            "Keep the already-streamed partial answer visible when a model "
            "call is silently retried, instead of letting the display "
            "reconciliation clear it."
        ),
    )
    model_config = ConfigDict(extra="ignore")


class RoundtableRoomLimitSettings(BaseModel):
    """Global runtime limits for multi-agent roundtable step 2.

    The count is deployment-wide: every user's foreground step-2 session and
    every non-terminal background roundtable job share one pool. One room is
    one session, not one seat and not one model call. Turning the switch off
    keeps recording occupancy but stops rejecting new entries.
    """

    enabled: bool = Field(
        default=True,
        description="Whether a full roundtable room pool blocks entry into step 2.",
    )
    max_rooms: int = Field(
        default=3,
        ge=1,
        le=20,
        description="Maximum simultaneous step-2 roundtable rooms across all users.",
    )
    max_hold_minutes: int = Field(
        default=30,
        ge=1,
        le=1440,
        description=(
            "Hard lifetime of one step-2 room. Heartbeats preserve network liveness "
            "but never extend this deadline."
        ),
    )
    lease_minutes: int = Field(
        default=5,
        ge=1,
        le=60,
        description=(
            "How long a foreground step-2 lease stays occupied without a renewal. "
            "The browser heartbeat renews it; this does not extend max_hold_minutes."
        ),
    )
    max_cycles: int = Field(
        default=8,
        ge=2,
        le=100,
        description=(
            "Coordinator round that switches to tool-free final closeout. "
            "The value is captured when an orchestration starts."
        ),
    )
    write_file_compatibility_enabled: bool = Field(
        default=False,
        description=(
            "When true, roundtable leader/seat/step-3 runs use buffered-gateway "
            "write_file streaming. Ordinary chat keeps the global switch."
        ),
    )
    model_config = ConfigDict(extra="ignore")


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

    mode: Literal["configured", "local"] = Field(
        default="configured",
        description=(
            "Use the embedding provider from config.yaml, or explicitly override it "
            "with the administrator-managed offline BGE-M3 encoder."
        ),
    )
    model_path: str = Field(
        default="",
        max_length=4096,
        description="Absolute or project-relative directory containing the offline BGE-M3 ONNX model.",
    )
    model_revision: str = Field(
        default="",
        max_length=64,
        description="Opaque revision changed whenever an offline model directory is activated.",
    )
    model_config = ConfigDict(extra="ignore")


class HtmlMapSettings(BaseModel):
    """Which map engine newly generated HTML pages should use."""

    engine: Literal["echarts", "leaflet"] = Field(
        default="echarts",
        description="echarts uses local GeoJSON; leaflet is the tile-based map webpage",
    )
    model_config = ConfigDict(extra="ignore")


_GENERATED_PAGE_LOGO_FILE_RE = re.compile(r"^logo-[a-f0-9]{32}\.(?:png|jpg|webp)$")


class GeneratedPageBrandingSettings(BaseModel):
    """Optional generated-page logo settings retained for offline bundles."""

    chrome_enabled: bool = True
    header_layout: Literal["left", "center"] = "center"
    system_name: str | None = Field(default=None, max_length=80)
    footer_text: str | None = Field(default=None, max_length=160)
    logo_filename: str | None = None
    logo_mode: Literal["custom", "default", "none"] = "custom"
    logo_media_type: Literal["image/png", "image/jpeg", "image/webp"] | None = None
    logo_version: str | None = Field(default=None, max_length=64)
    model_config = ConfigDict(extra="ignore")

    @field_validator("logo_filename")
    @classmethod
    def _validate_logo_filename(cls, value: str | None) -> str | None:
        if value is None:
            return None
        normalized = str(value).strip().lower()
        if not _GENERATED_PAGE_LOGO_FILE_RE.fullmatch(normalized):
            raise ValueError("Invalid generated-page logo filename")
        return normalized


_APP_IDENTITY_LOGO_FILE_RE = re.compile(r"^logo-[a-f0-9]{32}\.(?:png|jpg|webp)$")
_APP_IDENTITY_LIGHT_RED_HEADER_FILE_RE = re.compile(
    r"^light-red-header-[a-f0-9]{32}\.(?:png|jpg|webp|svg)$"
)
_APP_IDENTITY_LIGHT_RED_ACTIVE_FILE_RE = re.compile(
    r"^light-red-active-[a-f0-9]{32}\.(?:png|jpg|webp|svg)$"
)


class AppIdentitySettings(BaseModel):
    """Name and logo shown in this application's own chrome.

    Empty ``system_name`` and a missing logo keep the built-in title and the
    static theme logos. This block is independent of generated-page branding.
    """

    system_name: str = Field(default="", max_length=80)
    header_title_font_mode: Literal["adaptive", "fixed"] = "adaptive"
    header_title_font_size: int = Field(default=30, ge=16, le=40)
    logo_filename: str | None = None
    logo_media_type: Literal["image/png", "image/jpeg", "image/webp"] | None = None
    logo_version: str | None = None
    light_red_header_background_filename: str | None = None
    light_red_header_background_media_type: Literal[
        "image/png", "image/jpeg", "image/webp", "image/svg+xml"
    ] | None = None
    light_red_header_background_version: str | None = None
    light_red_nav_active_filename: str | None = None
    light_red_nav_active_media_type: Literal[
        "image/png", "image/jpeg", "image/webp", "image/svg+xml"
    ] | None = None
    light_red_nav_active_version: str | None = None
    model_config = ConfigDict(extra="ignore")

    @field_validator("system_name")
    @classmethod
    def _normalize_app_system_name(cls, value: str | None) -> str:
        normalized = " ".join(str(value or "").split())
        return normalized

    @field_validator("logo_filename")
    @classmethod
    def _validate_app_logo_filename(cls, value: str | None) -> str | None:
        if value is None:
            return None
        normalized = str(value).strip().lower()
        if not _APP_IDENTITY_LOGO_FILE_RE.fullmatch(normalized):
            raise ValueError("Invalid application logo filename")
        return normalized

    @field_validator("light_red_header_background_filename")
    @classmethod
    def _validate_light_red_header_filename(cls, value: str | None) -> str | None:
        if value is None:
            return None
        normalized = str(value).strip().lower()
        if not _APP_IDENTITY_LIGHT_RED_HEADER_FILE_RE.fullmatch(normalized):
            raise ValueError("Invalid light-red header background filename")
        return normalized

    @field_validator("light_red_nav_active_filename")
    @classmethod
    def _validate_light_red_active_filename(cls, value: str | None) -> str | None:
        if value is None:
            return None
        normalized = str(value).strip().lower()
        if not _APP_IDENTITY_LIGHT_RED_ACTIVE_FILE_RE.fullmatch(normalized):
            raise ValueError("Invalid light-red active navigation filename")
        return normalized


def app_identity_logo_path(
    settings: AppIdentitySettings | None = None,
) -> Path | None:
    """Resolve the uploaded application logo without accepting an arbitrary path."""

    current = settings or load_system_settings().app_identity
    if not current.logo_filename:
        return None
    path = runtime_home() / "app-identity" / current.logo_filename
    return path if path.is_file() else None


def app_identity_light_red_header_background_path(
    settings: AppIdentitySettings | None = None,
) -> Path | None:
    """Resolve the uploaded light-red header background inside runtime storage."""

    current = settings or load_system_settings().app_identity
    if not current.light_red_header_background_filename:
        return None
    path = runtime_home() / "app-identity" / current.light_red_header_background_filename
    return path if path.is_file() else None


def app_identity_light_red_nav_active_path(
    settings: AppIdentitySettings | None = None,
) -> Path | None:
    """Resolve the uploaded light-red active-navigation image inside runtime storage."""

    current = settings or load_system_settings().app_identity
    if not current.light_red_nav_active_filename:
        return None
    path = runtime_home() / "app-identity" / current.light_red_nav_active_filename
    return path if path.is_file() else None


def generated_page_logo_path(
    settings: GeneratedPageBrandingSettings | None = None,
) -> Path | None:
    """Resolve the custom generated-page logo without accepting arbitrary paths."""

    current = settings or load_system_settings().generated_page_branding
    if current.logo_mode != "custom" or not current.logo_filename:
        return None
    path = runtime_home() / "generated-page-branding" / current.logo_filename
    return path if path.is_file() else None


class HtmlRuntimeSettings(BaseModel):
    """Addresses written into newly generated HTML pages.

    Empty strings mean the settings dialog has not saved this block. Readers
    then keep an older config.yaml value, and finally the built-in default.
    """

    base_url: str = Field(
        default="",
        max_length=2048,
        description="Browser-reachable Gateway base for ECharts, G6, Leaflet and generated-page images",
    )
    map_tile_url: str = Field(
        default="",
        max_length=2048,
        description="Leaflet tile template containing {z}, {x} and {y}",
    )
    model_config = ConfigDict(extra="ignore")


class StrategyProjectSettings(BaseModel):
    """Global visibility controls for the strategy-project experience."""

    taiwan_strait_demo_enabled: bool = Field(
        default=False,
        description="Whether Taiwan Strait demo entry points are visible in My Projects.",
    )
    model_config = ConfigDict(extra="ignore")


class GeneralQaPageDescription(BaseModel):
    """Administrator-provided copy for the ordinary Q&A landing page."""

    welcome_title: str = Field(default="", max_length=80)
    welcome_description: str = Field(default="", max_length=500)
    input_placeholder: str = Field(default="", max_length=120)
    model_config = ConfigDict(extra="ignore")

    @field_validator("welcome_title", "welcome_description", "input_placeholder")
    @classmethod
    def _normalize_copy(cls, value: str | None) -> str:
        return str(value or "").strip()


class GeneralQaPageDescriptions(BaseModel):
    """Separate ordinary-Q&A copy for the full and compact layouts."""

    full: GeneralQaPageDescription = Field(default_factory=GeneralQaPageDescription)
    compact: GeneralQaPageDescription = Field(default_factory=GeneralQaPageDescription)
    model_config = ConfigDict(extra="ignore")

    @model_validator(mode="before")
    @classmethod
    def _migrate_legacy_single_description(cls, value: object) -> object:
        if isinstance(value, GeneralQaPageDescription):
            return {"full": value, "compact": value}
        if not isinstance(value, dict):
            return value
        if "full" in value or "compact" in value:
            return value
        legacy = {
            key: value.get(key, "")
            for key in ("welcome_title", "welcome_description", "input_placeholder")
        }
        return {"full": legacy, "compact": legacy}


class PageHeaderDescription(BaseModel):
    """Configurable title and explanatory copy for a product landing page."""

    title: str = Field(default="", max_length=80)
    description: str = Field(default="", max_length=500)
    model_config = ConfigDict(extra="ignore")

    @field_validator("title", "description")
    @classmethod
    def _normalize_copy(cls, value: str | None) -> str:
        return str(value or "").strip()


class AIWritingGuideCard(BaseModel):
    """Editable text for one of the fixed AI-writing welcome cards."""

    title: str = Field(default="", max_length=80)
    description: str = Field(default="", max_length=240)
    model_config = ConfigDict(extra="ignore")

    @field_validator("title", "description")
    @classmethod
    def _normalize_copy(cls, value: str | None) -> str:
        return str(value or "").strip()


def _empty_ai_writing_guide_cards() -> list[AIWritingGuideCard]:
    return [AIWritingGuideCard() for _ in range(3)]


class AIWritingPageDescription(PageHeaderDescription):
    """Configurable copy for the classic AI-writing welcome page."""

    guide_cards: list[AIWritingGuideCard] = Field(
        default_factory=_empty_ai_writing_guide_cards,
        max_length=3,
    )

    @model_validator(mode="before")
    @classmethod
    def _keep_three_guide_cards(cls, value: object) -> object:
        if not isinstance(value, dict) or not isinstance(value.get("guide_cards"), list):
            return value
        cards = value["guide_cards"][:3]
        return {**value, "guide_cards": [*cards, *({} for _ in range(3 - len(cards)))]}


class PageDescriptionsSettings(BaseModel):
    """Page-scoped, administrator-managed display copy."""

    general_qa: GeneralQaPageDescriptions = Field(default_factory=GeneralQaPageDescriptions)
    knowledge_base: PageHeaderDescription = Field(default_factory=PageHeaderDescription)
    ai_writing: AIWritingPageDescription = Field(default_factory=AIWritingPageDescription)
    strategy_projects: PageHeaderDescription = Field(default_factory=PageHeaderDescription)
    deep_research: PageHeaderDescription = Field(default_factory=PageHeaderDescription)
    gongwen: PageHeaderDescription = Field(default_factory=PageHeaderDescription)
    light_app_management: PageHeaderDescription = Field(default_factory=PageHeaderDescription)
    light_app_list: PageHeaderDescription = Field(default_factory=PageHeaderDescription)
    agents: PageHeaderDescription = Field(default_factory=PageHeaderDescription)
    business_chains: PageHeaderDescription = Field(default_factory=PageHeaderDescription)
    skills: PageHeaderDescription = Field(default_factory=PageHeaderDescription)
    knowledge_management: PageHeaderDescription = Field(default_factory=PageHeaderDescription)
    model_config = ConfigDict(extra="ignore")


class ModelPromptSuffix(BaseModel):
    """Administrator-owned instructions appended to one configured model's system prompt."""

    enabled: bool = False
    prompt: str = Field(default="", max_length=20000)
    model_config = ConfigDict(extra="forbid")


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

    prompt_prefix: PromptPrefixSettings = Field(default_factory=PromptPrefixSettings)
    model_prompt_suffixes: dict[str, ModelPromptSuffix] = Field(default_factory=dict)
    login_whitelist: LoginWhitelistSettings = Field(default_factory=LoginWhitelistSettings)
    authentication: AuthenticationSettings = Field(default_factory=AuthenticationSettings)
    private_square: PrivateSquareSettings = Field(default_factory=PrivateSquareSettings)
    publish_squares: PublishSquaresSettings = Field(default_factory=PublishSquaresSettings)
    appearance: AppearanceSettings = Field(default_factory=AppearanceSettings)
    generated_page_branding: GeneratedPageBrandingSettings = Field(
        default_factory=GeneratedPageBrandingSettings
    )
    page_descriptions: PageDescriptionsSettings = Field(default_factory=PageDescriptionsSettings)
    citation_display: CitationDisplaySettings = Field(default_factory=CitationDisplaySettings)
    leaderboard: LeaderboardSettings = Field(default_factory=LeaderboardSettings)
    skill_compression: SkillCompressionSettings = Field(default_factory=SkillCompressionSettings)
    skill_versioning: SkillVersioningSettings = Field(default_factory=SkillVersioningSettings)
    plugin_management: PluginManagementSettings = Field(default_factory=PluginManagementSettings)
    agent_modules: AgentModuleSettings = Field(default_factory=AgentModuleSettings)
    wx_common_agents: WxCommonAgentsSettings = Field(default_factory=WxCommonAgentsSettings)
    concurrency_monitor: ConcurrencyMonitorSettings = Field(default_factory=ConcurrencyMonitorSettings)
    model_concurrency: ModelConcurrencySettings = Field(default_factory=ModelConcurrencySettings)
    memory: MemoryRuntimeSettings = Field(default_factory=MemoryRuntimeSettings)
    conversation_context: ConversationContextSettings = Field(default_factory=ConversationContextSettings)
    write_file_compatibility: WriteFileCompatibilitySettings = Field(
        default_factory=WriteFileCompatibilitySettings
    )
    call_metrics: CallMetricsSettings = Field(default_factory=CallMetricsSettings)
    retry_keep_partial: RetryKeepPartialSettings = Field(
        default_factory=RetryKeepPartialSettings
    )
    roundtable_room_limit: RoundtableRoomLimitSettings = Field(default_factory=RoundtableRoomLimitSettings)
    local_wiki_encoder: LocalWikiEncoderSettings = Field(default_factory=LocalWikiEncoderSettings)
    strategy_projects: StrategyProjectSettings = Field(default_factory=StrategyProjectSettings)
    html_map: HtmlMapSettings = Field(default_factory=HtmlMapSettings)
    html_runtime: HtmlRuntimeSettings = Field(default_factory=HtmlRuntimeSettings)
    app_identity: AppIdentitySettings = Field(default_factory=AppIdentitySettings)
    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 keeps assistant-knowledge day-to-day surfaces admin-only; "
            "v2 releases them to ordinary users while ordinary knowledge remains the default landing."
        ),
    )
    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


# ── mtime 缓存 ────────────────────────────────────────────────────────────
# ``load_system_settings`` 处在登录等热路径上（一次登录会读 3 次：登录方式
# 开关、白名单、单会话策略），而每次调用都是「整文件读盘 + JSON 解析 +
# pydantic 校验」。按 (路径, mtime) 缓存解析结果：文件未变直接返回缓存值的
# 深拷贝（调用方可能就地改写返回值，不能共享实例）；任何写入——
# ``save_system_settings`` 的原子替换——都会更新 mtime，下一次调用自动重读，
# 无需显式清缓存。
_settings_cache_key: tuple[str, int] | None = None
_settings_cache_value: SystemSettings | None = None


def load_system_settings() -> SystemSettings:
    """Read the on-disk system settings, returning defaults when missing.

    Parsed results are cached by (path, mtime); unchanged files return a deep
    copy so callers may safely mutate the returned model without poisoning the
    shared cache.
    """
    global _settings_cache_key, _settings_cache_value

    path = _settings_path()
    try:
        mtime = path.stat().st_mtime_ns
    except OSError:
        return SystemSettings()

    key = (str(path), mtime)
    if _settings_cache_value is not None and _settings_cache_key == key:
        return _settings_cache_value.model_copy(deep=True)

    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()
        settings = SystemSettings.model_validate(raw)
    except Exception as exc:
        logger.warning("Failed to read system settings from %s: %s", path, exc)
        return SystemSettings()
    _settings_cache_key = key
    _settings_cache_value = settings
    return settings.model_copy(deep=True)


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 get_memory_runtime_settings() -> MemoryRuntimeSettings:
    """Return the admin-controlled Q&A memory switch, failing closed."""
    try:
        return load_system_settings().memory
    except Exception:
        logger.debug("Failed to read memory runtime setting; keeping memory disabled", exc_info=True)
        return MemoryRuntimeSettings()


def get_conversation_context_settings() -> ConversationContextSettings:
    """Return the admin-controlled recent-turn policy, failing closed."""
    try:
        return load_system_settings().conversation_context
    except Exception:
        logger.debug("Failed to read conversation context setting; keeping the recent-turn window disabled", exc_info=True)
        return ConversationContextSettings()


def get_write_file_compatibility_settings() -> WriteFileCompatibilitySettings:
    """Return the global buffered-gateway switch, failing closed."""

    try:
        return load_system_settings().write_file_compatibility
    except Exception:
        logger.debug(
            "Failed to read write_file compatibility setting; keeping native transport",
            exc_info=True,
        )
        return WriteFileCompatibilitySettings()


def get_call_metrics_settings() -> CallMetricsSettings:
    """Return the live call-metrics switch, defaulting to recording enabled."""

    try:
        return load_system_settings().call_metrics
    except Exception:
        logger.debug(
            "Failed to read call metrics setting; keeping metrics enabled",
            exc_info=True,
        )
        return CallMetricsSettings()


def get_roundtable_room_limit_settings() -> RoundtableRoomLimitSettings:
    """Return the admin-controlled roundtable room cap, defaulting to 3 enabled rooms."""
    try:
        return load_system_settings().roundtable_room_limit
    except Exception:
        logger.debug("Failed to read roundtable room limit; keeping the default cap", exc_info=True)
        return RoundtableRoomLimitSettings()


def save_system_settings(settings: SystemSettings) -> SystemSettings:
    """Persist the settings document atomically, returning the saved value."""
    global _settings_cache_key, _settings_cache_value
    path = _settings_path()
    payload = settings.model_dump(mode="json")
    with _lock:
        # Multiple Gateway workers share this file.  A unique temporary name
        # prevents their atomic replaces from racing over one common `.tmp`.
        tmp = path.with_name(f".{path.name}.{uuid4().hex}.tmp")
        try:
            tmp.write_text(json.dumps(payload, ensure_ascii=False, indent=2), encoding="utf-8")
            tmp.replace(path)
        finally:
            tmp.unlink(missing_ok=True)
    # mtime 变化本就足以让缓存失效；这里显式清一次，兜底个别文件系统
    # mtime 粒度过粗（同一秒内连续保存两次）读回旧值的极端情况。
    _settings_cache_key = None
    _settings_cache_value = None
    return settings