# -*- coding: utf-8 -*-
from __future__ import annotations

import importlib
import json
import logging
import re
import threading
from pathlib import Path
from typing import Optional, Union, Dict, List, Literal, Any, Set

from pydantic import (
    BaseModel,
    Field,
    ConfigDict,
    field_validator,
    model_validator,
)
import shortuuid
from qwenpaw.exceptions import (
    ConfigurationException,
)

from .timezone import detect_system_timezone
from ..constant import (
    HEARTBEAT_DEFAULT_EVERY,
    HEARTBEAT_DEFAULT_TARGET,
    HEARTBEAT_DEFAULT_TIMEOUT_SECONDS,
    HEARTBEAT_MAX_TIMEOUT_SECONDS,
    LLM_ACQUIRE_TIMEOUT,
    LLM_BACKOFF_BASE,
    LLM_BACKOFF_CAP,
    LLM_MAX_CONCURRENT,
    LLM_MAX_RETRIES,
    LLM_MAX_QPM,
    LLM_RATE_LIMIT_JITTER,
    LLM_RATE_LIMIT_PAUSE,
    WORKING_DIR,
)
from ..utils.io_utils import write_json_atomic
from ..utils.logging import sanitize_log_value

logger = logging.getLogger(__name__)

# A legacy field can be present in the root config and in several agent
# profiles, all of which may be validated repeatedly during one process
# lifetime.  The migration reminder is useful once, but repeating it for
# every request obscures real warnings.
_legacy_scroll_tool_cap_warned = False


# ============================================================================
# Core config models (moved here to avoid circular imports)
# ============================================================================


class ModelSlotConfig(BaseModel):
    """Model slot configuration for LLM routing."""

    provider_id: str = Field(default="")
    model: str = Field(default="")


class ActiveModelsInfo(BaseModel):
    """Active models information for provider manager."""

    active_llm: ModelSlotConfig | None
    effective_max_input_length: int | None = None


class ACPAgentConfig(BaseModel):
    """Configuration for one ACP agent."""

    enabled: bool = False
    command: str = ""
    args: list[str] = Field(default_factory=list)
    env: Dict[str, str] = Field(default_factory=dict)
    trusted: bool = True
    tool_parse_mode: str = "call_title"
    stdio_buffer_limit_bytes: int = Field(
        default=50 * 1024 * 1024,
        gt=0,
    )


def _get_default_acp_agents() -> Dict[str, ACPAgentConfig]:
    """Get default ACP agents configuration."""
    return {
        "opencode": ACPAgentConfig(
            enabled=True,
            command="opencode",
            args=["acp"],
            trusted=True,
            tool_parse_mode="update_detail",
        ),
        "qwen_code": ACPAgentConfig(
            enabled=True,
            command="qwen",
            args=["--acp"],
            trusted=True,
            tool_parse_mode="call_detail",
        ),
        "claude_code": ACPAgentConfig(
            enabled=True,
            command="npx",
            args=["-y", "@zed-industries/claude-agent-acp"],
            trusted=True,
            tool_parse_mode="update_detail",
        ),
        "codex": ACPAgentConfig(
            enabled=True,
            command="npx",
            args=["-y", "@zed-industries/codex-acp"],
            trusted=True,
            tool_parse_mode="call_detail",
        ),
    }


class ACPConfig(BaseModel):
    """ACP (Agent Communication Protocol) configuration."""

    node_path: str = ""
    agents: Dict[str, ACPAgentConfig] = Field(
        default_factory=_get_default_acp_agents,
    )

    @model_validator(mode="after")
    def _merge_default_agents(self):
        """Merge default agents with user-configured agents."""
        for name, agent_cfg in _get_default_acp_agents().items():
            if name not in self.agents:
                self.agents[name] = agent_cfg
        return self


# Agent ID validation: alphanumeric, hyphens, underscores.
_AGENT_ID_PATTERN = re.compile(
    r"^[a-zA-Z0-9][a-zA-Z0-9_-]*[a-zA-Z0-9]$",
)
_AGENT_ID_MIN_LENGTH = 2
_AGENT_ID_MAX_LENGTH = 64
_RESERVED_AGENT_IDS = frozenset({"default"})


def generate_short_agent_id() -> str:
    """Generate a 6-character short UUID for agent identification.

    Returns:
        6-character short UUID string
    """
    return shortuuid.ShortUUID().random(length=6)


def sanitize_agent_id(raw: str) -> str:
    """Normalize raw agent ID input: strip whitespace.

    Args:
        raw: Raw user input for agent ID.

    Returns:
        Sanitized agent ID string.
    """
    return raw.strip()


def validate_agent_id(
    agent_id: str,
    existing_ids: Set[str],
) -> None:
    """Validate a custom agent ID.

    Checks length, character set, reserved words, and uniqueness.

    Args:
        agent_id: The sanitized agent ID to validate.
        existing_ids: Set of already-registered agent IDs.

    Raises:
        ValueError: If the ID is invalid.
    """
    if len(agent_id) < _AGENT_ID_MIN_LENGTH:
        raise ValueError(
            f"Agent ID must be at least {_AGENT_ID_MIN_LENGTH} characters, "
            f"got {len(agent_id)}.",
        )
    if len(agent_id) > _AGENT_ID_MAX_LENGTH:
        raise ValueError(
            f"Agent ID must be at most {_AGENT_ID_MAX_LENGTH} characters, "
            f"got {len(agent_id)}.",
        )
    if not _AGENT_ID_PATTERN.match(agent_id):
        raise ValueError(
            f"Agent ID '{agent_id}' contains invalid characters. "
            "Only letters, digits, hyphens, and underscores "
            "are allowed. Cannot start or end with '-' or '_'.",
        )
    if agent_id in _RESERVED_AGENT_IDS:
        raise ValueError(
            f"Agent ID '{agent_id}' is reserved and cannot be used.",
        )
    if agent_id in existing_ids:
        raise ValueError(
            f"Agent ID '{agent_id}' already exists.",
        )


class BaseChannelConfig(BaseModel):
    """Base for channel config (read from config.json, no env)."""

    enabled: bool = False
    bot_prefix: str = ""
    show_tool_calls: bool = True
    show_tool_results: bool = True
    # A value of 0 means unlimited (do not truncate).
    tool_call_max_length: int = Field(default=200, ge=0)
    tool_result_max_length: int = Field(default=500, ge=0)
    show_thinking: bool = True
    dm_policy: Literal["open", "allowlist"] = "open"
    group_policy: Literal["open", "allowlist"] = "open"
    allow_from: List[str] = Field(default_factory=list)
    deny_message: str = ""
    require_mention: bool = False
    # Buffer media-only messages until a text message arrives, then merge.
    # Disable to process all messages immediately.
    no_text_debounce: bool = True
    access_control_dm: bool = False
    access_control_group: bool = False
    # Channel-level mute: completely disable DM or group messages
    dm_disabled: bool = False
    group_disabled: bool = False


class IMessageChannelConfig(BaseChannelConfig):
    db_path: str = "~/Library/Messages/chat.db"
    poll_sec: float = 1.0
    media_dir: Optional[str] = None
    max_decoded_size: int = (
        10 * 1024 * 1024
    )  # 10MB default limit for Base64 data


class DiscordConfig(BaseChannelConfig):
    bot_token: str = ""
    http_proxy: str = ""
    http_proxy_auth: str = ""
    accept_bot_messages: bool = False
    streaming_enabled: bool = False
    media_dir: Optional[str] = None


class DingTalkConfig(BaseChannelConfig):
    client_id: str = ""
    client_secret: str = ""
    message_type: str = "markdown"
    cron_message_type: str = "markdown"
    card_template_id: str = ""
    card_template_key: str = "content"
    robot_code: str = ""
    media_dir: Optional[str] = None
    card_auto_layout: bool = False
    at_sender_on_reply: bool = False
    streaming_enabled: bool = False
    endpoint: str = ""


class FeishuConfig(BaseChannelConfig):
    """Feishu/Lark channel: app_id, app_secret; optional encrypt_key,
    verification_token for event handler. media_dir for received media.
    domain: 'feishu' for China, 'lark' for international.
    streaming_enabled: enable CardKit streaming card updates for real-time
    typewriter-style text output.
    share_session_in_group: if True, all group members share one session;
    if False (default), each member gets an independent session.
    """

    app_id: str = ""
    app_secret: str = ""
    encrypt_key: str = ""
    verification_token: str = ""
    media_dir: Optional[str] = None
    domain: Literal["feishu", "lark"] = "feishu"
    streaming_enabled: bool = False
    share_session_in_group: bool = False


class QQConfig(BaseChannelConfig):
    app_id: str = ""
    client_secret: str = ""
    markdown_enabled: bool = True
    max_reconnect_attempts: int = 100
    ack_message: str = ""


class OneBotConfig(BaseChannelConfig):
    """OneBot v11 channel: reverse WebSocket for NapCat/go-cqhttp/Lagrange."""

    ws_host: str = "0.0.0.0"
    ws_port: int = 6199
    access_token: str = ""
    share_session_in_group: bool = False


class TelegramConfig(BaseChannelConfig):
    bot_token: str = ""
    base_url: str = ""
    http_proxy: str = ""
    http_proxy_auth: str = ""
    show_typing: Optional[bool] = None
    streaming_enabled: bool = False


class MQTTConfig(BaseChannelConfig):
    host: str = ""
    port: Optional[int] = None
    transport: str = ""
    clean_session: bool = True
    qos: int = 2
    username: Optional[str] = None
    password: Optional[str] = None
    subscribe_topic: str = ""
    publish_topic: str = ""
    tls_enabled: bool = False
    tls_ca_certs: Optional[str] = None
    tls_certfile: Optional[str] = None
    tls_keyfile: Optional[str] = None


class MattermostConfig(BaseChannelConfig):
    """Mattermost channel: WebSocket polling and REST API."""

    url: str = ""
    bot_token: str = ""
    media_dir: Optional[str] = None
    show_typing: Optional[bool] = None
    thread_follow_without_mention: bool = False


class ConsoleConfig(BaseChannelConfig):
    """Console channel: prints agent responses to stdout."""

    enabled: bool = True
    media_dir: Optional[str] = None


class WecomConfig(BaseChannelConfig):
    """WeCom (Enterprise WeChat) AI Bot channel config."""

    bot_id: str = ""
    secret: str = ""
    media_dir: Optional[str] = None
    welcome_text: str = ""
    # If True (default), all group members share one chat; set to
    # False to isolate each member into their own chat.
    share_session_in_group: bool = True
    max_reconnect_attempts: int = -1
    streaming_enabled: bool = False


class MatrixConfig(BaseChannelConfig):
    """Matrix channel configuration."""

    homeserver: str = ""

    @field_validator("homeserver")
    @classmethod
    def strip_trailing_slash(cls, v: str) -> str:
        return v.rstrip("/")

    user_id: str = ""
    access_token: str = ""

    # Extended Matrix channel fields
    group_allow_from: List[str] = Field(default_factory=list)
    groups: Dict[str, Any] = Field(default_factory=dict)
    encryption: bool = False
    # When False, images are surfaced as text placeholders (no vision URL).
    vision_enabled: bool = True
    history_limit: int = 50
    password: str = ""
    device_name: str = "qwenpaw-worker"
    # matrix-nio sync long-poll timeout (ms); typical 30s
    sync_timeout_ms: int = Field(default=30000, ge=5000, le=300000)
    # When True, prepend HTML pill to formatted_body for outbound mentions.
    # Default False: m.mentions is always set for push, but pill is omitted.
    mention_pill_in_body: bool = False
    # When True, apply m.mentions + optional pill on outbound messages.
    outbound_structured_mentions: bool = True
    streaming_enabled: bool = False


class VoiceChannelConfig(BaseChannelConfig):
    """Voice channel: Twilio ConversationRelay + Cloudflare Tunnel."""

    twilio_account_sid: str = ""
    twilio_auth_token: str = ""
    phone_number: str = ""
    phone_number_sid: str = ""
    tts_provider: str = "google"
    tts_voice: str = "en-US-Journey-D"
    stt_provider: str = "deepgram"
    language: str = "en-US"
    welcome_greeting: str = "Hi! This is QwenPaw. How can I help you?"


class SIPChannelConfig(BaseChannelConfig):
    """SIP voice channel: dual-track (pyVoIP dev / LiveKit production)."""

    sip_mode: str = "dev"
    sip_host: str = "0.0.0.0"
    sip_port: int = 5061
    sip_username: str = ""
    sip_password: str = ""
    sip_server: str = ""
    sip_transport: str = "UDP"
    rtp_port_low: int = 10000
    rtp_port_high: int = 20000
    dashscope_api_key: str = ""
    tts_provider: str = "aliyun"
    tts_voice: str = ""
    stt_provider: str = "aliyun"
    language: str = "zh-CN"
    welcome_greeting: str = "你好，我是QwenPaw"
    call_timeout: float = 120.0
    livekit_url: str = ""
    livekit_api_key: str = ""
    livekit_api_secret: str = ""
    livekit_sip_trunk_id: str = ""
    livekit_room_name: str = "sip-inbound"
    livekit_output_sample_rate: int = 24000
    max_concurrent_calls: int = 5


class XiaoYiConfig(BaseChannelConfig):
    """XiaoYi channel: Huawei A2A protocol via WebSocket."""

    ak: str = ""  # Access Key
    sk: str = ""  # Secret Key
    agent_id: str = ""  # Agent ID from XiaoYi platform
    task_timeout_ms: int = 3600000  # 1 hour task timeout


class YuanbaoConfig(BaseChannelConfig):
    """Tencent Yuanbao (元宝) channel config.

    Connects to Yuanbao bot platform via protobuf WebSocket with
    sign-token authentication. Supports C2C and group messaging.
    """

    app_id: str = ""
    app_secret: str = ""
    api_domain: str = "bot.yuanbao.tencent.com"
    media_dir: Optional[str] = None
    accept_bot_messages: bool = False


class WeChatConfig(BaseChannelConfig):
    """WeChat (iLink Bot) personal account channel config.

    bot_token:              Bearer token obtained after QR code login.
    bot_token_file:         Path to persist/load the bot_token
                            (default ~/.qwenpaw/wechat_bot_token).
    base_url:               iLink API base URL (leave empty to use default).
    media_dir:              Local directory for downloaded media files.
    message_merge_enabled:  When True, merge multiple outgoing text messages
                            within a single request to reduce message count
                            (mitigates the 10-message context_token limit).
    message_merge_delay_ms: Controls merge behaviour when merging is enabled.
                            0  → merge ALL text messages and send once at the
                                 end of the request (maximum savings).
                            >0 → buffer messages for this many milliseconds;
                                 if no new message arrives within the window
                                 the buffer is flushed (adjacent-merge mode).
    """

    bot_token: str = ""
    bot_token_file: str = ""
    base_url: str = ""
    media_dir: Optional[str] = None
    message_merge_enabled: bool = False
    message_merge_delay_ms: Optional[int] = 0


class SlackConfig(BaseChannelConfig):
    """Slack channel: Socket Mode connection with edit-in-place streaming.

    Uses slack-bolt AsyncSocketModeHandler (aiohttp WebSocket) to connect
    to a single Slack workspace. Supports incremental message rendering
    via chat.postMessage + chat.update (edit-in-place) when streaming is
    enabled.
    """

    bot_token: str = ""
    app_token: str = ""
    bot_prefix: str = ""
    proxy: Optional[str] = None
    streaming_enabled: bool = False
    require_mention: bool = True
    media_dir: Optional[str] = None
    dm_policy: str = "open"
    group_policy: str = "open"
    allow_from: Optional[list] = None
    deny_message: str = ""
    access_control_dm: bool = False
    access_control_group: bool = False
    dm_disabled: bool = False
    group_disabled: bool = False


class ChannelConfig(BaseModel):
    """Built-in channel configs; extra keys allowed for plugin channels."""

    model_config = ConfigDict(extra="allow")

    imessage: IMessageChannelConfig = IMessageChannelConfig()
    discord: DiscordConfig = DiscordConfig()
    dingtalk: DingTalkConfig = DingTalkConfig()
    feishu: FeishuConfig = FeishuConfig()
    qq: QQConfig = QQConfig()
    telegram: TelegramConfig = TelegramConfig()
    mattermost: MattermostConfig = MattermostConfig()
    mqtt: MQTTConfig = MQTTConfig()
    console: ConsoleConfig = ConsoleConfig()
    matrix: MatrixConfig = MatrixConfig()
    voice: VoiceChannelConfig = VoiceChannelConfig()
    sip: SIPChannelConfig = SIPChannelConfig()
    wecom: WecomConfig = WecomConfig()
    xiaoyi: XiaoYiConfig = XiaoYiConfig()
    yuanbao: YuanbaoConfig = YuanbaoConfig()
    wechat: WeChatConfig = WeChatConfig()
    slack: SlackConfig = SlackConfig()
    onebot: OneBotConfig = OneBotConfig()

    @model_validator(mode="before")
    @classmethod
    def _migrate_legacy_weixin_key(cls, data: Any) -> Any:
        """One-shot migration: legacy ``weixin`` key -> canonical ``wechat``.

        Older config files used ``weixin`` as the WeChat channel key. The
        canonical key is now ``wechat``. When an old config is loaded we
        rename the key in-place so validation succeeds. The on-disk file is
        rewritten by ``load_config`` right after validation (see utils.py).
        """
        if isinstance(data, dict) and "weixin" in data:
            data = dict(data)
            legacy = data.pop("weixin")
            if "wechat" not in data:
                data["wechat"] = legacy
        return data


class LastApiConfig(BaseModel):
    host: Optional[str] = None
    port: Optional[int] = None


class ActiveHoursConfig(BaseModel):
    """Optional active window for heartbeat (e.g. 08:00–22:00)."""

    start: str = "08:00"
    end: str = "22:00"


class HeartbeatConfig(BaseModel):
    """Heartbeat: run agent with HEARTBEAT.md as query at interval."""

    model_config = {"populate_by_name": True}

    enabled: bool = Field(default=False, description="Whether heartbeat is on")
    every: str = Field(default=HEARTBEAT_DEFAULT_EVERY)
    target: str = Field(default=HEARTBEAT_DEFAULT_TARGET)
    timeout_seconds: int = Field(
        default=HEARTBEAT_DEFAULT_TIMEOUT_SECONDS,
        ge=1,
        le=HEARTBEAT_MAX_TIMEOUT_SECONDS,
        alias="timeoutSeconds",
        description="Maximum seconds for one heartbeat execution",
    )
    active_hours: Optional[ActiveHoursConfig] = Field(
        default=None,
        alias="activeHours",
    )


class AgentsDefaultsConfig(BaseModel):
    heartbeat: Optional[HeartbeatConfig] = None


class AutoMemorySearchConfig(BaseModel):
    """Auto memory search configuration."""

    model_config = ConfigDict(extra="ignore")

    enabled: bool = Field(
        default=False,
        description="Whether to auto search memory on every turn",
    )

    max_results: int = Field(
        default=2,
        ge=1,
        description=(
            "Maximum number of results to return when auto memory"
            " search is enabled"
        ),
    )


class EmbeddingModelConfig(BaseModel):
    """Embedding model configuration."""

    model_config = ConfigDict(extra="ignore")

    backend: str = Field(
        default="openai",
        description="Embedding backend (openai, etc.)",
    )
    api_key: str = Field(
        default="",
        description="API key for embedding provider",
    )
    base_url: str = Field(default="", description="Base URL for embedding API")
    model_name: str = Field(default="", description="Embedding model name")
    dimensions: int = Field(default=1024, description="Embedding dimensions")
    enable_cache: bool = Field(
        default=True,
        description="Whether to enable embedding cache",
    )
    use_dimensions: bool = Field(
        default=False,
        description="Whether to use custom dimensions",
    )
    max_cache_size: int = Field(
        default=10000,
        description="Maximum cache size",
    )
    max_input_length: int = Field(
        default=8192,
        description="Maximum input length for embedding",
    )
    max_batch_size: int = Field(
        default=10,
        description="Maximum batch size for embedding",
    )


class ADBPGMemoryConfig(BaseModel):
    """ADBPG (AnalyticDB for PostgreSQL) REST memory configuration."""

    model_config = ConfigDict(extra="ignore")

    rest_base_url: str = ""
    rest_api_key: str = ""

    # Behavior
    memory_isolation: bool = Field(
        default=True,
        description="Per-agent memory isolation (True) or shared (False)",
    )
    search_timeout: float = 10.0
    auto_memory_search_config: AutoMemorySearchConfig = Field(
        default_factory=lambda: AutoMemorySearchConfig(
            enabled=True,
            max_results=3,
        ),
    )


class ReMeLightMemoryConfig(BaseModel):
    """ReMeLight memory manager configuration."""

    model_config = ConfigDict(extra="ignore")

    metadata_dir: str = Field(
        default="mem_metadata",
        description="Subdirectory for ReMe persistent state",
    )
    session_dir: str = Field(
        default="mem_session",
        description=(
            "Subdirectory for ReMe source conversation logs used by "
            "auto-memory"
        ),
    )
    mem_session_dir: str = Field(
        default="mem_agent",
        description="Subdirectory for ReMe internal memory-agent sessions",
    )
    resource_dir: str = Field(
        default="resource",
        description="Subdirectory for external assets",
    )
    daily_dir: str = Field(
        default="memory",
        description="Subdirectory for daily memory",
    )
    digest_dir: str = Field(
        default="digest",
        description="Subdirectory for digest memory",
    )
    summarize_when_compact: bool = Field(
        default=True,
        description="Whether to enable memory summarization during compaction",
    )

    inbox_push_enabled: bool = Field(
        default=True,
        description=(
            "Whether to push ReMe auto-memory, auto-dream, and "
            "auto-resource job results to the inbox"
        ),
    )

    auto_memory_interval: int | None = Field(
        default=5,
        description="Auto memory every N user queries. 1 means auto "
        "memory after every user query, 2 means every 2 queries, etc. "
        "None or <= 0 disables periodic auto memory. WARNING: Setting "
        "too small (e.g., 1-3) may cause high token usage and heavy "
        "background task burden.",
    )

    dream_cron_enabled: bool = Field(
        default=True,
        description=(
            "Whether to enable the dream-based memory optimization job"
        ),
    )

    dream_cron: str = Field(
        default="0 23 * * *",
        description=(
            "Cron expression for dream-based memory optimization job "
            "(use dream_cron_enabled to enable/disable). Scheduled runs "
            "start after a random delay of 0 to 60 seconds."
        ),
    )

    auto_memory_search_config: AutoMemorySearchConfig = Field(
        default_factory=AutoMemorySearchConfig,
    )

    embedding_model_config: EmbeddingModelConfig = Field(
        default_factory=EmbeddingModelConfig,
    )


class ContextCompactConfig(BaseModel):
    """Context compaction configuration."""

    model_config = ConfigDict(extra="ignore")

    enabled: bool = Field(
        default=True,
        description="Whether to enable automatic context compaction",
    )

    compact_threshold_ratio: float = Field(
        default=0.8,
        ge=0.1,
        le=0.9,
        description=(
            "Compaction trigger threshold ratio: compaction is triggered when "
            "the context length reaches this fraction of max_input_length"
        ),
    )

    reserve_threshold_ratio: float = Field(
        default=0.1,
        gt=0,
        le=0.3,
        description=(
            "Context reserve threshold ratio: the most recent fraction of the "
            "context is preserved after compaction to maintain continuity"
        ),
    )


class ToolResultPruningConfig(BaseModel):
    """Tool result pruning configuration."""

    model_config = ConfigDict(extra="ignore")

    enabled: bool = Field(
        default=True,
        description="Whether to enable tool result pruning",
    )

    pruning_recent_n: int = Field(
        default=2,
        ge=1,
        le=10,
        description=(
            "Number of recent tool-result-bearing messages to keep at the "
            "recent preview byte limit. Scroll keeps all live previews at "
            "this limit until pressure-driven pointer folding is required."
        ),
    )

    pruning_old_msg_max_bytes: int = Field(
        default=3000,
        ge=100,
        description=(
            "Older tool-result preview byte limit for non-Scroll context "
            "strategies. Scroll does not use a fixed old-result size "
            "threshold; it folds recoverable results only while the rebuilt "
            "context remains under pressure."
        ),
    )

    pruning_recent_msg_max_bytes: int = Field(
        default=50000,
        ge=1000,
        description=(
            "Byte threshold for tool result previews before they enter the "
            "agent context and while they remain recent."
        ),
    )

    offload_retention_days: int = Field(
        default=30,
        ge=1,
        le=365,
        description=(
            "Number of days to retain complete archived tool result files. "
            "This lifetime is independent of Scroll history retention; after "
            "expiry, history may still contain the bounded preview but not "
            "the complete artifact."
        ),
    )

    tool_results_cache: str = Field(
        default="tool_results",
        description="Directory name for tool result cache files "
        "relative to working_dir",
    )

    exempt_file_extensions: List[str] = Field(
        default_factory=lambda: [".md"],
        description=(
            "File extensions exempt from tool result pruning. "
            "Tool results for read_file operations on these file types "
            "will use recent_max_bytes instead of old_max_bytes."
        ),
    )

    exempt_tool_names: List[str] = Field(
        default_factory=lambda: ["chat_with_agent"],
        description=(
            "Tool names exempt from tool result pruning. "
            "Tool results from these tools will use recent_max_bytes "
            "instead of old_max_bytes."
        ),
    )


class ScrollContextConfig(BaseModel):
    """Scroll (retrieval-driven) context manager configuration.

    Only consulted when ``LightContextConfig.strategy == "scroll"``. The
    durable history lives at ``{working_dir}/{db_filename}``; evicted turns
    fold into an in-context eviction index recallable from the sandboxed
    ``recall_history_python`` REPL.
    """

    model_config = ConfigDict(extra="ignore")

    db_filename: str = Field(
        default="history.db",
        description="SQLite history store filename, relative to working_dir.",
    )

    tool_output_token_cap: int = Field(
        default=3000,
        ge=100,
        exclude=True,
        description=(
            "Deprecated scroll-only tool result cap. Tool output sizing is "
            "handled by tool_result_pruning_config. Excluded when saving so "
            "legacy configurations migrate on their next write."
        ),
    )

    repl_timeout_s: int = Field(
        default=300,
        ge=1,
        description=(
            "Per-call timeout for the recall_history_python REPL tool."
        ),
    )

    history_retention_days: int = Field(
        default=30,
        ge=0,
        description=(
            "Days of durable history to keep; rows older than this are "
            "purged automatically on startup and on agent teardown. Default "
            "30 keeps roughly the last month. Set 0 to keep history forever "
            "(unbounded growth — only the capacity warning fires)."
        ),
    )

    allow_unsandboxed: bool = Field(
        default=False,
        description=(
            "UNSAFE escape hatch. The recall_history_python recall REPL runs "
            "model-authored Python and is only isolated by the sandbox; the "
            "sandbox config is injected by the governance layer. When that "
            "layer is degraded the tool fails closed and refuses to run. Set "
            "this to true to run the REPL with NO isolation (arbitrary host "
            "code as the agent user) — trusted local/dev use only."
        ),
    )

    offload_dialog: bool = Field(
        default=False,
        description=(
            "Also archive evicted turns to legacy ``dialog/{date}.jsonl`` "
            "files. Off by default: under scroll the durable ``history.db`` "
            "is already the full record, so dialog files are a redundant "
            "opt-in for external consumers (analytics, backup). When on, "
            "dialog is written on every eviction AND on /clear, /new, "
            "/compact; when off, scroll never writes dialog anywhere."
        ),
    )

    summarize_unheadlined_evictions: bool = Field(
        default=True,
        description=(
            "When an evicted span carries NO model headline, generate a "
            "one-line summary of it (via the active model) to use as its "
            "eviction-index entry instead of a bare ``(no milestone)`` line. "
            "Keeps the index readable for legacy 1.x conversations (whose "
            "turns predate headlines) and for tool-heavy spans the model "
            "never headlined. The full turns stay recallable either way; "
            "this only affects the descriptive label. Best-effort — a "
            "model/timeout failure falls back to ``(no milestone)`` and never "
            "blocks eviction. Costs one extra model call per such eviction."
        ),
    )

    summarize_eviction_timeout_seconds: int = Field(
        default=20,
        ge=1,
        description=(
            "Per-eviction timeout for the un-headlined-span summary call "
            "above. On timeout the span keeps a ``(no milestone)`` label; "
            "eviction itself is never delayed beyond this."
        ),
    )


class LightContextConfig(BaseModel):
    """Light context manager configuration."""

    model_config = ConfigDict(extra="ignore")

    strategy: Literal["native", "scroll"] = Field(
        default="scroll",
        description=(
            "Context management strategy. 'native' = AgentScope compression; "
            "'scroll' = retrieval-driven history.db + eviction index with a "
            "sandboxed recall_history_python recall REPL (the default)."
        ),
    )

    dialog_path: str = Field(
        default="dialog",
        description="Path for dialog persistence to jsonl files "
        "relative to working_dir.",
    )

    token_count_estimate_divisor: float = Field(
        default=4,
        ge=2,
        le=5,
        description=(
            "Divisor for byte-based token estimation (byte_len / divisor)"
        ),
    )

    context_compact_config: ContextCompactConfig = Field(
        default_factory=ContextCompactConfig,
    )
    tool_result_pruning_config: ToolResultPruningConfig = Field(
        default_factory=ToolResultPruningConfig,
    )
    scroll_config: ScrollContextConfig = Field(
        default_factory=ScrollContextConfig,
    )

    @model_validator(mode="after")
    def warn_deprecated_scroll_tool_cap(self) -> "LightContextConfig":
        """Warn once when the removed scroll-only tool cap is configured."""
        global _legacy_scroll_tool_cap_warned
        configured = (
            "tool_output_token_cap" in self.scroll_config.model_fields_set
        )
        if configured and not _legacy_scroll_tool_cap_warned:
            _legacy_scroll_tool_cap_warned = True
            logger.warning(
                "scroll_config.tool_output_token_cap is deprecated and "
                "ignored; use tool_result_pruning_config."
                "pruning_recent_msg_max_bytes instead (bytes, not tokens)",
            )
        return self


class AutoTitleConfig(BaseModel):
    """Async chat-title generation configuration.

    The console handler creates each new chat with a 10-character
    placeholder name and spawns a background task that asks the active
    LLM for a concise title. Each new chat costs one short extra LLM
    call; flip ``enabled`` to ``False`` to keep the placeholder and
    avoid the spend.
    """

    model_config = ConfigDict(extra="ignore")

    enabled: bool = Field(
        default=True,
        description=(
            "Generate a chat title via the active LLM after the first "
            "user message. Disable to keep the truncated placeholder "
            "and skip the extra per-chat LLM call."
        ),
    )

    timeout_seconds: float = Field(
        default=30.0,
        ge=1.0,
        description=(
            "Hard timeout for the title-generation LLM call. The "
            "background task is swallowed if this fires, leaving the "
            "placeholder name in place."
        ),
    )


class DoomLoopStageConfig(BaseModel):
    """One escalation stage in doom loop detection."""

    after: int = Field(
        ge=1,
        description=("Trigger after N consecutive repetitions"),
    )
    action: str = Field(
        default="modify_prompt",
        description=("Action when triggered: " "'modify_prompt' or 'stop'"),
    )
    prompt: str = Field(
        default="",
        description=("Warning text (modify_prompt) " "or stop reason (stop)"),
    )


class DoomLoopConfig(BaseModel):
    """Doom loop detection configuration."""

    enabled: bool = Field(
        default=True,
        description="Enable doom loop detection",
    )
    window_size: int = Field(
        default=3,
        ge=2,
        description=("Sliding window size for " "repetition detection"),
    )
    similarity_threshold: float = Field(
        default=1.0,
        ge=0.0,
        le=1.0,
        description=(
            "Similarity threshold to consider " "calls as repetitive"
        ),
    )
    stages: List[DoomLoopStageConfig] = Field(
        default_factory=lambda: [
            DoomLoopStageConfig(
                after=3,
                action="modify_prompt",
                prompt=(
                    "[WARNING] Repetitive pattern "
                    "detected. You are repeating "
                    "similar actions without "
                    "progress. Try a completely "
                    "different approach."
                ),
            ),
            DoomLoopStageConfig(
                after=4,
                action="stop",
                prompt=(
                    "Doom loop: agent stuck "
                    "after 4 consecutive "
                    "repetitions"
                ),
            ),
        ],
        description=("Escalation stages (sorted by after)"),
    )
    in_loop_modes: bool = Field(
        default=False,
        description=("Also run during /goal and " "/mission loop modes"),
    )


class IterationGateConfig(BaseModel):
    """Standalone iteration gate configuration."""

    enabled: bool = Field(
        default=True,
        description="Enable iteration limit",
    )
    max_iterations: Optional[int] = Field(
        default=None,
        ge=1,
        le=500,
        description=(
            "Maximum loop turns before stopping. "
            "Falls back to AgentsRunningConfig.max_iters "
            "when not set (legacy compat)."
        ),
    )


class RubricGateConfig(BaseModel):
    """Completion check gate configuration.

    Prevents premature agent stop when the LLM
    outputs text-only responses without tool calls.
    """

    enabled: bool = Field(
        default=False,
        description=(
            "Enable completion check to prevent "
            "early stop on text-only responses"
        ),
    )
    prompt: str = Field(
        default=(
            "You did not call any tool in the "
            "last turn. If the task is truly "
            "complete, confirm it. Otherwise, "
            "continue working with tool calls."
        ),
        description=(
            "Prompt injected when the agent " "produces a text-only response"
        ),
    )
    max_interventions: int = Field(
        default=1,
        ge=1,
        le=10,
        description=(
            "Max times to re-prompt per loop " "turn to avoid infinite retries"
        ),
    )
    in_loop_modes: bool = Field(
        default=False,
        description=("Also run during /goal and " "/mission loop modes"),
    )


class GateInstanceConfig(BaseModel):
    """One built-in gate configured in a custom loop mode."""

    model_config = ConfigDict(extra="forbid")

    id: str = Field(
        min_length=1,
        max_length=64,
        pattern=r"^[a-z0-9][a-z0-9_-]*$",
    )
    type: str = Field(
        min_length=1,
        max_length=64,
        pattern=r"^[a-z0-9][a-z0-9_]*$",
    )
    enabled: bool = True
    params: Dict[str, Any] = Field(default_factory=dict)


class CustomLoopModeConfig(BaseModel):
    """A saved custom loop mode made from built-in gates."""

    model_config = ConfigDict(extra="forbid")

    id: str = Field(
        min_length=1,
        max_length=64,
        pattern=r"^[a-z0-9][a-z0-9_-]*$",
    )
    name: str = Field(min_length=1, max_length=80)
    description: str = Field(default="", max_length=500)
    slash_command: str = Field(
        min_length=1,
        max_length=64,
        pattern=r"^[a-z0-9][a-z0-9_-]*$",
    )
    enabled: bool = False
    gates: List[GateInstanceConfig] = Field(
        default_factory=list,
        max_length=20,
    )

    @field_validator("name", mode="before")
    @classmethod
    def strip_display_name(cls, value: Any) -> Any:
        """Strip display names before length and uniqueness validation."""
        if isinstance(value, str):
            return value.strip()
        return value

    @model_validator(mode="after")
    def validate_pipeline(self) -> "CustomLoopModeConfig":
        """Reject ambiguous or unsafe pipeline shapes."""
        gate_ids = [gate.id for gate in self.gates]
        if len(gate_ids) != len(set(gate_ids)):
            raise ValueError("Gate instance IDs must be unique")

        gate_types = [gate.type for gate in self.gates if gate.enabled]
        if len(gate_types) != len(set(gate_types)):
            raise ValueError("Gate types cannot be repeated")
        from ..loop.catalog import get_gate_catalog

        get_gate_catalog().validate_exclusive_groups(gate_types)
        if self.enabled and not gate_types:
            raise ValueError("Enabled custom modes require an enabled gate")
        return self


def normalize_custom_loop_mode_name(name: str) -> str:
    """Return the canonical value used for custom mode name uniqueness."""
    return name.strip().casefold()


class GoalLoopModeConfig(BaseModel):
    """Editable values for the fixed built-in Goal pipeline."""

    max_iterations: int = Field(default=20, ge=1, le=500)
    max_tokens: int = Field(default=300_000, ge=1)


class MissionLoopModeConfig(BaseModel):
    """Editable values for the fixed built-in Mission pipeline."""

    max_iterations: int = Field(default=20, ge=1, le=100)
    max_retries_per_story: int = Field(default=3, ge=0, le=10)
    default_verification_instructions: str = Field(
        default="",
        max_length=4000,
    )
    default_verify_command: str = Field(default="", max_length=2000)


class LoopConfig(BaseModel):
    """Loop engineering configuration."""

    iteration: IterationGateConfig = Field(
        default_factory=IterationGateConfig,
        description="Iteration limit settings",
    )
    doom_loop: DoomLoopConfig = Field(
        default_factory=DoomLoopConfig,
        description="Repetition protection settings",
    )
    rubric: RubricGateConfig = Field(
        default_factory=RubricGateConfig,
        description="Completion check settings",
    )
    goal: GoalLoopModeConfig = Field(
        default_factory=GoalLoopModeConfig,
        description="Fixed Goal mode gate values",
    )
    mission: MissionLoopModeConfig = Field(
        default_factory=MissionLoopModeConfig,
        description="Fixed Mission mode gate values",
    )
    custom_modes: List[CustomLoopModeConfig] = Field(
        default_factory=list,
        max_length=20,
        description="User-defined loop modes built from built-in gates",
    )

    @model_validator(mode="after")
    def validate_custom_modes(self) -> "LoopConfig":
        """Keep custom mode identity and commands unambiguous."""
        mode_ids = [mode.id for mode in self.custom_modes]
        if len(mode_ids) != len(set(mode_ids)):
            raise ValueError("Custom loop mode IDs must be unique")
        commands = [mode.slash_command for mode in self.custom_modes]
        if len(commands) != len(set(commands)):
            raise ValueError("Custom loop slash commands must be unique")
        names = [
            normalize_custom_loop_mode_name(mode.name)
            for mode in self.custom_modes
        ]
        if len(names) != len(set(names)):
            raise ValueError("Custom loop mode names must be unique")

        from ..loop.catalog import get_gate_catalog

        catalog = get_gate_catalog()
        for mode in self.custom_modes:
            for gate in mode.gates:
                catalog.validate_params(gate.type, gate.params)
        return self


def _sanitize_custom_loop_modes(
    data: Dict[str, Any],
    agent_id: str,
) -> None:
    """Skip invalid saved custom modes before profile validation.

    Custom Loop Modes are optional extensions. One stale or malformed mode
    must not make the entire Agent profile unavailable.
    """
    running = data.get("running")
    if not isinstance(running, dict):
        return
    loop = running.get("loop")
    if not isinstance(loop, dict):
        return
    raw_modes = loop.get("custom_modes")
    if raw_modes is None:
        return
    if not isinstance(raw_modes, list):
        logger.warning(
            "Agent '%s' custom Loop Modes were ignored: expected a list",
            sanitize_log_value(agent_id),
        )
        loop["custom_modes"] = []
        return

    from ..loop.catalog import get_gate_catalog

    catalog = get_gate_catalog()
    valid_modes: List[Dict[str, Any]] = []
    mode_ids: Set[str] = set()
    commands: Set[str] = set()
    names: Set[str] = set()
    for index, raw_mode in enumerate(raw_modes):
        if len(valid_modes) >= 20:
            logger.warning(
                "Agent '%s' custom Loop Mode at index %d was skipped: "
                "the maximum of 20 valid modes was reached",
                sanitize_log_value(agent_id),
                index,
            )
            continue
        try:
            mode = CustomLoopModeConfig.model_validate(raw_mode)
            for gate in mode.gates:
                catalog.validate_params(gate.type, gate.params)
        except (TypeError, ValueError) as exc:
            logger.warning(
                "Agent '%s' custom Loop Mode at index %d was skipped: %s",
                sanitize_log_value(agent_id),
                index,
                sanitize_log_value(exc),
            )
            continue

        normalized_name = normalize_custom_loop_mode_name(mode.name)
        if (
            mode.id in mode_ids
            or mode.slash_command in commands
            or normalized_name in names
        ):
            logger.warning(
                "Agent '%s' duplicate custom Loop Mode '%s' was skipped",
                sanitize_log_value(agent_id),
                sanitize_log_value(mode.id),
            )
            continue
        mode_ids.add(mode.id)
        commands.add(mode.slash_command)
        names.add(normalized_name)
        valid_modes.append(mode.model_dump(exclude_none=True))

    loop["custom_modes"] = valid_modes


def _sanitize_loop_config(
    data: Dict[str, Any],
    agent_id: str,
) -> None:
    """Keep invalid optional Loop data from blocking Agent startup."""
    running = data.get("running")
    if not isinstance(running, dict) or "loop" not in running:
        return
    if not isinstance(running["loop"], dict):
        logger.warning(
            "Agent '%s' Loop configuration was invalid; using defaults",
            sanitize_log_value(agent_id),
        )
        running["loop"] = LoopConfig().model_dump(exclude_none=True)
        return

    _sanitize_custom_loop_modes(data, agent_id)
    try:
        validated = LoopConfig.model_validate(running["loop"])
    except (TypeError, ValueError) as exc:
        logger.warning(
            "Agent '%s' Loop configuration was invalid; using defaults: %s",
            sanitize_log_value(agent_id),
            sanitize_log_value(exc),
        )
        running["loop"] = LoopConfig().model_dump(exclude_none=True)
        return
    running["loop"] = validated.model_dump(exclude_none=True)


class AgentsRunningConfig(BaseModel):
    """Agent runtime behavior configuration."""

    model_config = ConfigDict(extra="ignore")

    max_iters: int = Field(
        default=100,
        ge=1,
        description=(
            "Maximum number of reasoning-acting iterations for ReAct agent"
        ),
    )

    loop: LoopConfig = Field(
        default_factory=LoopConfig,
        description="Loop engineering configuration",
    )

    llm_retry_enabled: bool = Field(
        default=LLM_MAX_RETRIES > 0,
        description="Whether to auto-retry transient LLM API errors",
    )

    llm_max_retries: int = Field(
        default=max(LLM_MAX_RETRIES, 1),
        ge=1,
        description="Maximum retry attempts for transient LLM API errors",
    )

    llm_backoff_base: float = Field(
        default=LLM_BACKOFF_BASE,
        ge=0.1,
        description="Base delay in seconds for exponential LLM retry backoff",
    )

    llm_backoff_cap: float = Field(
        default=LLM_BACKOFF_CAP,
        ge=0.5,
        description=(
            "Maximum delay cap in seconds for LLM retry backoff; "
            "must be greater than or equal to the base delay"
        ),
    )

    llm_max_concurrent: int = Field(
        default=LLM_MAX_CONCURRENT,
        ge=1,
        description=(
            "Maximum number of concurrent in-flight LLM calls. "
            "Shared across all agents; only the first initialization wins."
        ),
    )

    llm_max_qpm: int = Field(
        default=LLM_MAX_QPM,
        ge=0,
        description=(
            "Maximum queries per minute (60-second sliding window). "
            "New requests that would exceed this limit wait before being "
            "dispatched — proactively preventing 429s. 0 = disabled."
        ),
    )

    llm_rate_limit_pause: float = Field(
        default=LLM_RATE_LIMIT_PAUSE,
        ge=1.0,
        description=(
            "Default pause duration (seconds) applied globally when a 429 "
            "rate-limit response is received."
        ),
    )

    llm_rate_limit_jitter: float = Field(
        default=LLM_RATE_LIMIT_JITTER,
        ge=0.0,
        description=(
            "Random jitter range (seconds) added on top of the pause so "
            "concurrent waiters stagger their wake-up."
        ),
    )

    llm_acquire_timeout: float = Field(
        default=LLM_ACQUIRE_TIMEOUT,
        ge=10.0,
        description=(
            "Maximum time (seconds) a caller waits to acquire a rate-limiter "
            "slot before giving up with an error."
        ),
    )

    shell_command_timeout: float = Field(
        default=60.0,
        ge=1.0,
        description=(
            "Default timeout in seconds for execute_shell_command. "
            "The LLM may still override this per-call via the timeout "
            "parameter."
        ),
    )

    shell_command_executable: str = Field(
        default="",
        description=(
            "Path to the shell used by execute_shell_command. "
            "Linux/macOS: e.g. /bin/bash, /bin/zsh. "
            "Windows: supports powershell.exe, pwsh.exe, or POSIX-like "
            "shells such as Git Bash. "
            "When empty, falls back to the $SHELL environment variable, "
            "then to the platform default (/bin/sh on Unix, cmd.exe on "
            "Windows)."
        ),
    )

    @model_validator(mode="after")
    def validate_llm_retry_backoff(self) -> "AgentsRunningConfig":
        """Validate LLM retry backoff relationships."""
        if self.llm_backoff_cap < self.llm_backoff_base:
            raise ConfigurationException(
                config_key="llm_backoff",
                message=(
                    "llm_backoff_cap must be greater than or equal to "
                    "llm_backoff_base"
                ),
            )
        return self

    max_input_length: int = Field(
        default=128 * 1024,  # 128K = 131072 tokens
        ge=1000,
        description=(
            "Maximum input length (tokens) for the model context window"
        ),
    )

    history_max_length: int = Field(
        default=10000,
        ge=1000,
        description="Maximum length for /history command output",
    )

    context_manager_backend: str = Field(default="light")

    light_context_config: LightContextConfig = Field(
        default_factory=LightContextConfig,
    )

    auto_title_config: AutoTitleConfig = Field(
        default_factory=AutoTitleConfig,
        description=(
            "Async chat-title generation toggle and timeout. See "
            "AutoTitleConfig."
        ),
    )

    memory_manager_backend: str = Field(default="remelight")

    adbpg_memory_config: Optional[ADBPGMemoryConfig] = Field(
        default=None,
        description="ADBPG memory configuration (used when "
        "memory_manager_backend='adbpg')",
    )

    reme_light_memory_config: ReMeLightMemoryConfig = Field(
        default_factory=ReMeLightMemoryConfig,
    )

    daily_memory_dir: str = Field(
        default="memory",
        description="Dir name to daily summary file",
    )

    approval_level: Optional[str] = Field(
        default=None,
        description=(
            "Tool execution security level (proxied from agent profile): "
            "STRICT, SMART, AUTO, or OFF.  When set via running-config API, "
            "the value is written back to the agent profile."
        ),
    )


class AgentsLLMRoutingConfig(BaseModel):
    model_config = ConfigDict(extra="ignore")

    enabled: bool = Field(default=False)
    mode: Literal["local_first", "cloud_first"] = Field(
        default="local_first",
        description=(
            "local_first routes to the local slot by default; cloud_first "
            "routes to the cloud slot by default. Smarter switching can be "
            "added later without changing the dual-slot config shape."
        ),
    )
    local: ModelSlotConfig = Field(
        default_factory=ModelSlotConfig,
        description="Local model slot (required when routing is enabled).",
    )
    cloud: Optional[ModelSlotConfig] = Field(
        default=None,
        description=(
            "Optional explicit cloud model slot; when null, uses "
            "providers.json active_llm."
        ),
    )


class AgentProfileRef(BaseModel):
    """Agent Profile reference (stored in root config.json).

    Only contains ID and workspace directory reference.
    Full agent configuration is stored in workspace/agent.json.
    """

    model_config = ConfigDict(extra="ignore")

    id: str = Field(..., description="Unique agent ID")
    workspace_dir: str = Field(
        ...,
        description="Path to agent's workspace directory",
    )
    enabled: bool = Field(
        default=True,
        description="Whether agent is enabled (controls instance loading)",
    )
    pinned: bool = Field(
        default=False,
        description="Whether agent is pinned in agent selectors",
    )


class PlanConfig(BaseModel):
    """Plan mode configuration (stored in agent.json)."""

    enabled: bool = Field(
        default=False,
        description="Whether plan mode is enabled for this agent",
    )


class CodingModeConfig(BaseModel):
    """Configuration for the Coding Mode feature."""

    enabled: bool = Field(
        default=False,
        description="Enable Coding Mode IDE layout and tools",
    )
    project_dir: Optional[str] = Field(
        default=None,
        description=(
            "Active coding project directory (absolute path). "
            "When set, Coding Mode file / git operations use this path "
            "instead of the agent workspace_dir. "
            "None means use the default workspace_dir."
        ),
    )


class AgentProfileConfig(BaseModel):
    """Complete Agent Profile configuration (stored in workspace/agent.json).

    Each agent has its own configuration file with all settings.
    """

    id: str = Field(..., description="Unique agent ID")
    name: str = Field(..., description="Human-readable agent name")
    description: str = Field(default="", description="Agent description")
    workspace_dir: str = Field(
        default="",
        description="Path to agent's workspace (optional, for reference)",
    )
    template_id: Optional[str] = Field(
        default=None,
        description="Builtin template used when this agent was created",
    )

    # Agent-specific configurations
    channels: Optional["ChannelConfig"] = Field(
        default=None,
        description="Channel configurations for this agent",
    )
    mcp: Optional["MCPConfig"] = Field(
        default=None,
        description="MCP clients for this agent",
    )
    heartbeat: Optional[HeartbeatConfig] = Field(
        default=None,
        description="Heartbeat configuration for this agent",
    )
    last_dispatch: Optional["LastDispatchConfig"] = Field(
        default=None,
        description="Last dispatch target for this agent",
    )
    running: AgentsRunningConfig = Field(
        default_factory=AgentsRunningConfig,
        description="Runtime configuration",
    )
    llm_routing: AgentsLLMRoutingConfig = Field(
        default_factory=AgentsLLMRoutingConfig,
        description="LLM routing settings",
    )
    active_model: Optional["ModelSlotConfig"] = Field(
        default=None,
        description="Active model for this agent (provider_id + model)",
    )
    language: str = Field(
        default="zh",
        description="Language setting for this agent",
    )
    approval_level: str = Field(
        default="AUTO",
        description=(
            "Tool execution security level: "
            "STRICT (all tools need approval), "
            "SMART (low-risk auto-allowed), "
            "AUTO (only guarded tools), "
            "OFF (guard disabled)"
        ),
    )
    system_prompt_files: List[str] = Field(
        default_factory=lambda: ["AGENTS.md", "SOUL.md", "PROFILE.md"],
        description="System prompt markdown files",
    )
    tools: Optional["ToolsConfig"] = Field(
        default=None,
        description="Tools configuration for this agent",
    )
    security: Optional["SecurityConfig"] = Field(
        default=None,
        description="Security configuration for this agent",
    )
    acp: Optional[ACPConfig] = Field(
        default=None,
        description="ACP configuration for this agent",
    )
    plan: PlanConfig = Field(
        default_factory=PlanConfig,
        description="Plan mode configuration for this agent",
    )
    coding_mode: CodingModeConfig = Field(
        default_factory=CodingModeConfig,
        description="Coding Mode configuration for this agent",
    )


class AgentsConfig(BaseModel):
    """Agents configuration (root config.json only contains references)."""

    active_agent: str = Field(
        default="default",
        description="Currently active agent ID",
    )
    agent_order: List[str] = Field(
        default_factory=lambda: ["default"],
        description="Persisted UI order for configured agents",
    )
    profiles: Dict[str, AgentProfileRef] = Field(
        default_factory=lambda: {
            "default": AgentProfileRef(
                id="default",
                workspace_dir=f"{WORKING_DIR}/workspaces/default",
            ),
        },
        description="Agent profile references (ID and workspace path only)",
    )

    # Legacy fields for backward compatibility (deprecated)
    # These fields MUST have default values (not None) to support downgrade
    defaults: Optional[AgentsDefaultsConfig] = None
    running: AgentsRunningConfig = Field(
        default_factory=AgentsRunningConfig,
    )
    llm_routing: AgentsLLMRoutingConfig = Field(
        default_factory=AgentsLLMRoutingConfig,
    )
    language: str = Field(default="zh")
    installed_md_files_language: Optional[str] = None
    system_prompt_files: List[str] = Field(
        default_factory=lambda: ["AGENTS.md", "SOUL.md", "PROFILE.md"],
    )
    audio_mode: Literal["auto", "native"] = Field(
        default="auto",
        description=(
            "How to handle incoming audio/voice messages. "
            '"auto": transcribe if a provider is available, otherwise show '
            "file-uploaded placeholder; "
            '"native": send audio blocks directly to the model '
            "(may need ffmpeg)."
        ),
    )

    transcription_provider_type: Literal[
        "disabled",
        "whisper_api",
        "local_whisper",
    ] = Field(
        default="disabled",
        description=(
            "Transcription backend. "
            '"disabled": no transcription; '
            '"whisper_api": remote OpenAI-compatible endpoint; '
            '"local_whisper": locally installed openai-whisper.'
        ),
    )
    transcription_provider_id: str = Field(
        default="",
        description=(
            "Provider ID for Whisper API transcription. "
            "Empty = no provider selected. "
            'Only used when transcription_provider_type is "whisper_api".'
        ),
    )
    transcription_model: str = Field(
        default="whisper-1",
        description=(
            "Model name for Whisper API transcription. "
            'e.g. "whisper-1", "whisper-large-v3".'
        ),
    )


class LastDispatchConfig(BaseModel):
    """Last channel/user/session that received a user-originated reply."""

    channel: str = ""
    user_id: str = ""
    session_id: str = ""


class MCPOAuthConfig(BaseModel):
    """OAuth 2.1 configuration for a remote MCP client.

    Stores OAuth credentials and endpoints discovered via RFC 8414 /
    RFC 9728.  Tokens are masked in API responses; stored plain-text in
    agent.json (file is local to the user's workspace).
    """

    client_id: str = ""
    scope: str = ""
    access_token: str = ""
    refresh_token: str = ""
    expires_at: float = 0.0
    token_endpoint: str = ""
    auth_endpoint: str = ""


class MCPClientConfig(BaseModel):
    """Configuration for a single MCP client."""

    model_config = ConfigDict(populate_by_name=True)

    name: str
    description: str = ""
    enabled: bool = True
    transport: Literal["stdio", "streamable_http", "sse"] = "stdio"
    url: str = ""
    headers: Dict[str, str] = Field(default_factory=dict)
    command: str = ""
    args: List[str] = Field(default_factory=list)
    env: Dict[str, str] = Field(default_factory=dict)
    cwd: str = ""
    tools: Optional[List[str]] = Field(
        default=None,
        description="Tool whitelist. Only listed tools will be loaded. "
        "None means load all tools from the server.",
    )
    oauth: Optional[MCPOAuthConfig] = None

    @model_validator(mode="before")
    @classmethod
    def _normalize_legacy_fields(cls, data):
        """Normalize common MCP field aliases from third-party examples."""
        if not isinstance(data, dict):
            return data

        payload = dict(data)

        if "isActive" in payload and "enabled" not in payload:
            payload["enabled"] = payload["isActive"]

        if "baseUrl" in payload and "url" not in payload:
            payload["url"] = payload["baseUrl"]

        if "type" in payload and "transport" not in payload:
            payload["transport"] = payload["type"]

        if (
            "transport" not in payload
            and (payload.get("url") or payload.get("baseUrl"))
            and not payload.get("command")
        ):
            payload["transport"] = "streamable_http"

        raw_transport = payload.get("transport")
        if isinstance(raw_transport, str):
            normalized = raw_transport.strip().lower()
            transport_alias_map = {
                "streamable_http": "streamable_http",
                "streamable-http": "streamable_http",
                "streamablehttp": "streamable_http",
                "http": "streamable_http",
                "stdio": "stdio",
                "sse": "sse",
            }
            payload["transport"] = transport_alias_map.get(
                normalized,
                normalized,
            )

        return payload

    @model_validator(mode="after")
    def _validate_transport_config(self):
        """Validate required fields for each MCP transport type."""
        if self.transport == "stdio":
            if not self.command.strip():
                raise ConfigurationException(
                    config_key="mcp.command",
                    message="stdio MCP client requires non-empty command",
                )
            return self

        if not self.url.strip():
            raise ConfigurationException(
                config_key="mcp.url",
                message=f"{self.transport} MCP client requires non-empty url",
            )
        return self


class MCPConfig(BaseModel):
    """MCP clients configuration.

    Uses a dict to allow dynamic client definitions.
    Default tavily_search client is created and auto-enabled if API key exists.
    """

    clients: Dict[str, MCPClientConfig] = Field(
        default_factory=lambda: {
            "tavily_search": MCPClientConfig(
                name="tavily_mcp",
                enabled=False,
                command="npx",
                args=["-y", "tavily-mcp@latest"],
                env={"TAVILY_API_KEY": ""},
            ),
        },
    )
    # One-shot migration watermark, persisted in agent.json.  Decoupled from
    # DriverCard existence so that deleting a migrated client no longer lets
    # startup migration resurrect it (#6130).  0 = not migrated; steps are
    # defined by CURRENT_MCP_MIGRATION_VERSION in
    # drivers.adapters.mcp_legacy_config.
    migration_version: int = 0


class BuiltinToolConfig(BaseModel):
    """Configuration for a single built-in tool."""

    name: str = Field(..., description="Tool function name")
    enabled: bool = Field(
        default=True,
        description="Whether the tool is enabled",
    )
    description: str = Field(default="", description="Tool description")
    display_to_user: bool = Field(
        default=True,
        description="Whether tool output is rendered to user channels",
    )
    async_execution: bool = Field(
        default=False,
        description="Whether to execute the tool asynchronously in background",
    )
    icon: str | None = Field(
        default=None,
        description="Emoji icon for the tool",
    )
    config: Dict[str, Any] = Field(
        default_factory=dict,
        description="Tool-specific configuration (e.g., API keys)",
    )


_BUILTIN_TOOLS_CACHE: Dict[str, BuiltinToolConfig] | None = None
_BUILTIN_TOOLS_LOCK = threading.Lock()


def _reset_builtin_tools_cache_for_tests() -> None:
    """Clear cached BuiltinToolConfig map (test helper only)."""
    global _BUILTIN_TOOLS_CACHE
    with _BUILTIN_TOOLS_LOCK:
        _BUILTIN_TOOLS_CACHE = None


def _copy_builtin_tools(
    tools: Dict[str, BuiltinToolConfig],
) -> Dict[str, BuiltinToolConfig]:
    """Return a shallow dict of model copies so callers cannot mutate cache."""
    return {name: cfg.model_copy() for name, cfg in tools.items()}


def _add_plugin_tool_default(
    tools: Dict[str, BuiltinToolConfig],
    tool_name: str,
    *,
    description: str,
    icon: str,
) -> None:
    """Insert a disabled-by-default plugin tool if *tool_name* is absent."""
    if tool_name in tools:
        return
    tools[tool_name] = BuiltinToolConfig(
        name=tool_name,
        enabled=False,
        description=description,
        display_to_user=True,
        async_execution=False,
        icon=icon,
    )


def _merge_plugin_manifest_tools(
    tools: Dict[str, BuiltinToolConfig],
) -> None:
    """Merge current plugin manifests into *tools* (disabled by default).

    Mutates *tools* in place. Manifests are always read live so late-loaded
    or unloaded plugins are reflected without process restart.
    """
    try:
        from ..plugins.registry import PluginRegistry

        registry = PluginRegistry()
        all_manifests = registry.get_all_plugin_manifests()
    except Exception as exc:
        logger.debug("Plugin tool merge skipped: %s", exc)
        return

    for plugin_id, manifest in all_manifests.items():
        meta = manifest.get("meta", {})
        if meta.get("tool_name"):
            _add_plugin_tool_default(
                tools,
                meta["tool_name"],
                description=meta.get(
                    "tool_description",
                    f"Tool from plugin {plugin_id}",
                ),
                icon=meta.get("tool_icon", "🔧"),
            )
        tools_list = meta.get("tools", [])
        if not isinstance(tools_list, list):
            continue
        for tool_info in tools_list:
            if not isinstance(tool_info, dict) or "name" not in tool_info:
                continue
            _add_plugin_tool_default(
                tools,
                tool_info["name"],
                description=tool_info.get(
                    "description",
                    f"Tool from plugin {plugin_id}",
                ),
                icon=tool_info.get("icon", "🔧"),
            )


def _default_builtin_tools() -> Dict[str, BuiltinToolConfig]:
    """Return built-in tool definitions from ``@tool_descriptor`` UI metadata.

    Descriptor-derived configs are process-cached (stable). Plugin tools from
    manifests are merged on every call (disabled by default) so late-loaded
    plugins are not permanently omitted after startup warm-up.
    Descriptor import failure fails closed (raises).
    """
    global _BUILTIN_TOOLS_CACHE
    with _BUILTIN_TOOLS_LOCK:
        if _BUILTIN_TOOLS_CACHE is None:
            tools: Dict[str, BuiltinToolConfig] = {}
            try:
                # Side-effect import via importlib (not `from ..agents import
                # tools`) so mypy --follow-imports=skip does not form a static
                # cycle with agents.tools → delegate_external_agent → config.
                importlib.import_module("qwenpaw.agents.tools")
                from ..runtime.tool_registry import get_builtin_tool_funcs

                for fn in get_builtin_tool_funcs():
                    desc = getattr(fn, "_tool_descriptor", None)
                    if desc is None:
                        continue
                    ui = getattr(desc, "ui", None)
                    tools[desc.name] = BuiltinToolConfig(
                        name=desc.name,
                        enabled=desc.enabled_by_default,
                        description=(
                            (ui.description if ui and ui.description else "")
                            or desc.description
                            or ""
                        ),
                        display_to_user=(
                            ui.display_to_user if ui is not None else True
                        ),
                        async_execution=desc.async_execution,
                        icon=(ui.icon if ui and ui.icon else None),
                    )
            except Exception as exc:
                logger.error(
                    "Failed to build BuiltinToolConfig from tool "
                    "descriptors: %s",
                    exc,
                    exc_info=True,
                )
                raise RuntimeError(
                    "Failed to build built-in tool config from descriptors; "
                    "refusing to persist an empty/incomplete ToolsConfig",
                ) from exc

            _BUILTIN_TOOLS_CACHE = tools

        merged = _copy_builtin_tools(_BUILTIN_TOOLS_CACHE)
        _merge_plugin_manifest_tools(merged)
        return merged


class ToolsConfig(BaseModel):
    """Built-in tools management configuration."""

    builtin_tools: Dict[str, BuiltinToolConfig] = Field(
        default_factory=_default_builtin_tools,
    )

    @model_validator(mode="after")
    def _merge_default_tools(self):
        """Ensure new code-defined tools are present in saved configs.

        Also normalises legacy entries whose ``icon`` is ``None`` so that
        downstream serialisation (e.g. ``ToolInfo``) never receives a null
        icon value.
        """
        defaults = _default_builtin_tools()
        for name, tc in defaults.items():
            if name not in self.builtin_tools:
                self.builtin_tools[name] = tc
            elif self.builtin_tools[name].icon is None:
                self.builtin_tools[name].icon = tc.icon
        # Normalise legacy/stale entries not in the current defaults
        for name, tc in self.builtin_tools.items():
            if name not in defaults and tc.icon is None:
                tc.icon = ""
        return self


def build_qa_agent_tools_config() -> ToolsConfig:
    """Tools preset for builtin ``default_qa_agent`` (first workspace init).

    Only these are enabled: execute_shell_command, read_file, edit_file,
    write_file, view_image. All other built-ins are disabled.
    """
    allow = frozenset(
        {
            "execute_shell_command",
            "read_file",
            "write_file",
            "edit_file",
            "view_image",
        },
    )
    builtin_tools = {
        name: tc.model_copy(update={"enabled": name in allow})
        for name, tc in _default_builtin_tools().items()
    }
    return ToolsConfig(builtin_tools=builtin_tools)


def build_local_agent_tools_config() -> ToolsConfig:
    """Tools preset for local collaborative agents.

    Inter-agent coordination tools are enabled by default, along with
    execute_shell_command and file read/write/edit tools, so a local small
    model can escalate planning work while still handling basic workspace
    actions. All other built-ins are disabled.
    """
    allow = frozenset(
        {
            "list_agents",
            "chat_with_agent",
            "submit_to_agent",
            "check_agent_task",
            "execute_shell_command",
            "read_file",
            "write_file",
            "edit_file",
        },
    )
    builtin_tools = {
        name: tc.model_copy(update={"enabled": name in allow})
        for name, tc in _default_builtin_tools().items()
    }
    return ToolsConfig(builtin_tools=builtin_tools)


class ToolGuardRuleConfig(BaseModel):
    """A single user-defined guard rule (stored in config.json)."""

    id: str
    tools: List[str] = Field(default_factory=list)
    params: List[str] = Field(default_factory=list)
    category: str = "command_injection"
    severity: str = "HIGH"
    patterns: List[str] = Field(default_factory=list)
    exclude_patterns: List[str] = Field(default_factory=list)
    description: str = ""
    remediation: str = ""


def _default_shell_evasion_checks() -> Dict[str, bool]:
    """Return default shell-evasion checks (all disabled at startup)."""
    return {
        "command_substitution": False,
        "obfuscated_flags": False,
        "backslash_escaped_whitespace": False,
        "backslash_escaped_operators": False,
        "newlines": False,
        "comment_quote_desync": False,
        "quoted_newline": False,
    }


class ToolGuardConfig(BaseModel):
    """Tool guard settings under ``security.tool_guard``.

    ``guarded_tools``: ``None`` → use built-in default set; empty list → guard
    nothing; non-empty list → guard only those tools.
    """

    enabled: bool = True
    guarded_tools: Optional[List[str]] = None
    denied_tools: List[str] = Field(default_factory=list)
    auto_denied_rules: List[str] = Field(
        default_factory=lambda: ["SAFETY_CHECKS_DESTRUCTIVE_COMMAND"],
        description=(
            "Rule IDs that unconditionally deny matched tool calls. "
            "Defaults to SAFETY_CHECKS_DESTRUCTIVE_COMMAND (catastrophic "
            "wipes/mkfs/dd only). An empty list is treated as unset and "
            "keeps that default (legacy configs). To disable auto-deny, "
            "set env QWENPAW_TOOL_GUARD_AUTO_DENIED_RULES=none."
        ),
    )
    custom_rules: List[ToolGuardRuleConfig] = Field(default_factory=list)
    disabled_rules: List[str] = Field(default_factory=list)
    shell_evasion_checks: Dict[str, bool] = Field(
        default_factory=_default_shell_evasion_checks,
    )


class FileGuardConfig(BaseModel):
    """File guard settings under ``security.file_guard``."""

    enabled: bool = True
    sensitive_files: List[str] = Field(default_factory=list)
    allow_preview_outside_workspace: bool = True


class SkillScannerWhitelistEntry(BaseModel):
    """A whitelisted skill (identified by name + content hash)."""

    skill_name: str
    content_hash: str = Field(
        default="",
        description="SHA-256 of concatenated file contents at whitelist time. "
        "Empty string means any content is allowed.",
    )
    added_at: str = Field(
        default="",
        description="ISO 8601 timestamp when the entry was added.",
    )


class SkillScannerConfig(BaseModel):
    """Skill scanner settings under ``security.skill_scanner``.

    ``mode`` controls the scanner behavior:
    * ``"block"`` – scan and block unsafe skills.
    * ``"warn"``  – scan but only log warnings, do not block (default).
    * ``"off"``   – disable scanning entirely.
    """

    mode: Literal["block", "warn", "off"] = Field(
        default="warn",
        description="Scanner mode: block, warn, or off.",
    )
    timeout: int = Field(
        default=30,
        ge=5,
        le=300,
        description="Max seconds to wait for a scan to complete.",
    )
    whitelist: List[SkillScannerWhitelistEntry] = Field(
        default_factory=list,
        description="Skills that bypass security scanning.",
    )


class SecurityConfig(BaseModel):
    """Top-level ``security`` section in config.json."""

    tool_guard: ToolGuardConfig = Field(default_factory=ToolGuardConfig)
    file_guard: FileGuardConfig = Field(default_factory=FileGuardConfig)
    skill_scanner: SkillScannerConfig = Field(
        default_factory=SkillScannerConfig,
    )
    sandbox_enabled: bool = Field(
        default=False,
        description=(
            "Global switch for governance sandbox execution. Defaults to "
            "False (sandbox off). When True, shell tools with no matching "
            "rule run inside the sandbox (no user prompt). When False, such "
            "calls run directly without the sandbox (no prompt). Phase 0-2 "
            "protections (secret-file / dangerous-command blocking) are "
            "unaffected either way."
        ),
    )
    allow_no_auth_hosts: List[str] = Field(
        default_factory=lambda: ["127.0.0.1", "::1"],
        description=(
            "List of client IP addresses that can access API endpoints "
            "without authentication. By default, localhost addresses "
            "(127.0.0.1 for IPv4, ::1 for IPv6) are allowed. "
            "WARNING: Only add trusted IP addresses to this list."
        ),
    )
    trusted_proxies: List[str] = Field(
        default_factory=list,
        description=(
            "Reverse proxy IP/CIDR list. X-Forwarded-For / X-Real-IP "
            "headers are only trusted when the direct TCP peer matches "
            "an entry in this list. Empty (default) = never trust proxy "
            "headers. Example: ['127.0.0.1', '172.17.0.0/16']"
        ),
    )

    @field_validator("trusted_proxies")
    @classmethod
    def _validate_trusted_proxies(cls, v: List[str]) -> List[str]:
        import ipaddress as _ipaddress

        _DENY = {"0.0.0.0/0", "::/0", "0.0.0.0", "::"}
        cleaned = []
        for entry in v:
            entry = entry.strip()
            if entry in _DENY:
                raise ValueError(
                    f"trusted_proxies must not contain"
                    f" '{entry}' (equivalent to disabling"
                    f" the security fix)",
                )
            net = _ipaddress.ip_network(entry, strict=False)
            cleaned.append(str(net))
        return cleaned


class Config(BaseModel):
    """Root config (config.json)."""

    channels: ChannelConfig = ChannelConfig()
    mcp: MCPConfig = MCPConfig()
    tools: ToolsConfig = Field(default_factory=ToolsConfig)
    last_api: LastApiConfig = LastApiConfig()
    agents: AgentsConfig = Field(default_factory=AgentsConfig)
    last_dispatch: Optional[LastDispatchConfig] = None
    security: SecurityConfig = Field(default_factory=SecurityConfig)
    acp: ACPConfig = Field(default_factory=ACPConfig)
    show_tool_details: bool = True
    user_timezone: str = Field(
        default_factory=detect_system_timezone,
        description="User IANA timezone (e.g. Asia/Shanghai). "
        "Defaults to the system timezone.",
    )
    plugins: Dict[str, Dict[str, Any]] = Field(
        default_factory=dict,
        description="Plugin configurations. Key is plugin_id, "
        "value is plugin-specific config dict.",
    )
    skill_paths: List[str] = Field(
        default_factory=list,
        description="Additional read-only skill pool roots, scanned after "
        "the primary skill_pool in order. Paths support ~ expansion. "
        "Skills found here are read-only (no edit/create); they can be "
        "listed, downloaded to a workspace, and deleted.",
    )


ChannelConfigUnion = Union[
    IMessageChannelConfig,
    DiscordConfig,
    DingTalkConfig,
    FeishuConfig,
    QQConfig,
    TelegramConfig,
    MattermostConfig,
    MQTTConfig,
    ConsoleConfig,
    MatrixConfig,
    VoiceChannelConfig,
    SIPChannelConfig,
    SlackConfig,
    WecomConfig,
    XiaoYiConfig,
    WeChatConfig,
]


# Agent configuration utility functions


def build_fallback_agent_profile_config(
    agent_id: str,
    config: "Config",
) -> AgentProfileConfig:
    """Build the same profile as when ``agent.json``
    is missing (no disk read/write).

    Used by :func:`load_agent_config` and ``qwenpaw doctor fix``
    so defaults stay in sync.
    """
    if agent_id not in config.agents.profiles:
        raise ValueError(f"Agent '{agent_id}' not found in config")

    agent_ref = config.agents.profiles[agent_id]
    workspace_dir = Path(agent_ref.workspace_dir).expanduser()
    return AgentProfileConfig(
        id=agent_id,
        name=agent_id.title(),
        description=f"{agent_id} agent",
        workspace_dir=str(workspace_dir),
        channels=(
            config.channels
            if hasattr(config, "channels") and config.channels
            else None
        ),
        mcp=config.mcp if hasattr(config, "mcp") and config.mcp else None,
        tools=(
            config.tools if hasattr(config, "tools") and config.tools else None
        ),
        security=(
            config.security
            if hasattr(config, "security") and config.security
            else None
        ),
        running=(
            config.agents.running
            if hasattr(config.agents, "running") and config.agents.running
            else AgentsRunningConfig()
        ),
        llm_routing=(
            config.agents.llm_routing
            if hasattr(config.agents, "llm_routing")
            and config.agents.llm_routing
            else AgentsLLMRoutingConfig()
        ),
        system_prompt_files=(
            config.agents.system_prompt_files
            if hasattr(config.agents, "system_prompt_files")
            and config.agents.system_prompt_files
            else ["AGENTS.md", "SOUL.md", "PROFILE.md"]
        ),
        acp=(config.acp if hasattr(config, "acp") and config.acp else None),
    )


def _migrate_access_control_fields(  # pylint: disable=too-many-branches
    channels: dict,
    workspace_dir: Path,
) -> bool:
    """Migrate legacy dm_policy/group_policy/allow_from to new fields.

    Returns True if any field was migrated (caller should rewrite file).
    """
    migrated = False
    for ch_key, ch_cfg in channels.items():
        if not isinstance(ch_cfg, dict):
            continue
        # dm_policy → access_control_dm or dm_disabled
        dm_policy = ch_cfg.get("dm_policy")
        if dm_policy is not None:
            if dm_policy == "allowlist" and "access_control_dm" not in ch_cfg:
                ch_cfg["access_control_dm"] = True
            elif dm_policy == "disabled" and "dm_disabled" not in ch_cfg:
                ch_cfg["dm_disabled"] = True
            del ch_cfg["dm_policy"]
            migrated = True
        # group_policy → access_control_group or group_disabled
        group_policy = ch_cfg.get("group_policy")
        if group_policy is not None:
            if (
                group_policy == "allowlist"
                and "access_control_group" not in ch_cfg
            ):
                ch_cfg["access_control_group"] = True
            elif group_policy == "disabled" and "group_disabled" not in ch_cfg:
                ch_cfg["group_disabled"] = True
            del ch_cfg["group_policy"]
            migrated = True
        # allow_from → access_control.json whitelist
        allow_from = ch_cfg.get("allow_from")
        if allow_from and isinstance(allow_from, list):
            try:
                from ..app.channels.access_control import (
                    get_access_control_store,
                )

                store = get_access_control_store(workspace_dir)
                store.import_allow_from(ch_key, set(allow_from))
            except Exception:
                pass
            del ch_cfg["allow_from"]
            migrated = True
        # group_allow_from (matrix legacy) → whitelist
        grp_allow = ch_cfg.get("group_allow_from")
        if grp_allow is not None:
            if isinstance(grp_allow, list) and grp_allow:
                try:
                    from ..app.channels.access_control import (
                        get_access_control_store,
                    )

                    store = get_access_control_store(workspace_dir)
                    store.import_allow_from(ch_key, set(grp_allow))
                except Exception:
                    pass
            del ch_cfg["group_allow_from"]
            migrated = True
    return migrated


def migrate_channel_display_fields(channels: object) -> bool:
    """Migrate legacy channel display settings in-place.

    Only translates the legacy boolean flags into their replacements; the
    remaining fields fall back to the model defaults, so channels without
    legacy settings are left untouched (no spurious config rewrite).
    """
    if not isinstance(channels, dict):
        return False
    migrated = False
    for channel_cfg in channels.values():
        if not isinstance(channel_cfg, dict):
            continue
        legacy = channel_cfg.pop("filter_tool_messages", None)
        if legacy is not None:
            channel_cfg.setdefault("show_tool_calls", not bool(legacy))
            channel_cfg.setdefault("show_tool_results", not bool(legacy))
            migrated = True
        legacy_thinking = channel_cfg.pop("filter_thinking", None)
        if legacy_thinking is not None:
            channel_cfg.setdefault("show_thinking", not bool(legacy_thinking))
            migrated = True
    return migrated


def load_agent_config(  # pylint: disable=too-many-branches,too-many-statements
    agent_id: str,
) -> AgentProfileConfig:
    """Load agent's complete configuration from workspace/agent.json with
    mtime-based caching.

    Uses file modification time to avoid unnecessary disk reads.

    Args:
        agent_id: Agent ID to load

    Returns:
        AgentProfileConfig: Complete agent configuration

    Raises:
        ValueError: If agent ID not found in root config
    """
    from .utils import (
        load_config,
        _agent_config_cache,
        _agent_config_lock,
    )

    config = load_config()

    if agent_id not in config.agents.profiles:
        raise ConfigurationException(
            config_key="agent",
            message=f"Agent '{agent_id}' not found in config",
        )

    agent_ref = config.agents.profiles[agent_id]
    workspace_dir = Path(agent_ref.workspace_dir).expanduser()
    agent_config_path = workspace_dir / "agent.json"

    if not agent_config_path.exists():
        fallback_config = build_fallback_agent_profile_config(agent_id, config)
        # Save for future use
        save_agent_config(agent_id, fallback_config)
        return fallback_config

    # Check mtime to see if we can use cached config
    try:
        current_mtime = agent_config_path.stat().st_mtime
    except OSError:
        fallback_config = build_fallback_agent_profile_config(agent_id, config)
        save_agent_config(agent_id, fallback_config)
        return fallback_config

    with _agent_config_lock:
        # Return cached config if mtime hasn't changed
        if agent_id in _agent_config_cache:
            cached_config, cached_mtime = _agent_config_cache[agent_id]
            if cached_mtime == current_mtime:
                return cached_config

        # Need to reload config from disk
        with open(agent_config_path, "r", encoding="utf-8") as f:
            data = json.load(f)

        # Match the existing migration behavior: migrate this workspace only
        # when its agent configuration is loaded.
        channels = data.get("channels")
        weixin_migrated = False
        if isinstance(channels, dict) and "weixin" in channels:
            legacy = channels.pop("weixin")
            channels.setdefault("wechat", legacy)
            weixin_migrated = True

        if isinstance(channels, dict):
            display_migrated = migrate_channel_display_fields(channels)
            access_control_migrated = _migrate_access_control_fields(
                channels,
                workspace_dir,
            )
        else:
            display_migrated = False
            access_control_migrated = False

        if weixin_migrated or display_migrated or access_control_migrated:
            try:
                if weixin_migrated or display_migrated:
                    import uuid as _uuid
                    import shutil as _shutil

                    migration_name = (
                        "channel-display" if display_migrated else "weixin"
                    )
                    backup_path = agent_config_path.with_suffix(
                        f".{_uuid.uuid4().hex[:8]}."
                        f"{migration_name}-migrate.bak",
                    )
                    _shutil.copy2(agent_config_path, backup_path)
                with open(
                    agent_config_path,
                    "w",
                    encoding="utf-8",
                ) as file:
                    json.dump(data, file, ensure_ascii=False, indent=2)
                try:
                    current_mtime = agent_config_path.stat().st_mtime
                except OSError:
                    pass
            except OSError:
                pass

        # Normalize legacy ~/.copaw-bound paths to current WORKING_DIR.
        # This keeps QWENPAW_WORKING_DIR effective even if existing agent.json
        # contains older hard-coded paths like "~/.copaw/media".
        # NOTE: this transform is applied in-memory only; it must not be
        # persisted back to disk.
        try:
            from .utils import _normalize_working_dir_bound_paths

            data = _normalize_working_dir_bound_paths(data)
        except Exception:
            pass

        # Pre-validate MCP clients: skip invalid ones so a
        # single misconfigured MCP client does not prevent the
        # entire agent from loading.
        from .utils import sanitize_mcp_clients

        sanitize_mcp_clients(data, agent_id)
        _sanitize_loop_config(data, agent_id)

        agent_config = AgentProfileConfig(**data)

        # Cache the config with its mtime
        _agent_config_cache[agent_id] = (agent_config, current_mtime)

        return agent_config


def save_agent_config(
    agent_id: str,
    agent_config: AgentProfileConfig,
) -> None:
    """Save agent configuration to workspace/agent.json and invalidate cache.

    Args:
        agent_id: Agent ID
        agent_config: Complete agent configuration to save

    Raises:
        ValueError: If agent ID not found in root config
    """
    from .utils import (
        load_config,
        _agent_config_cache,
        _agent_config_lock,
    )

    config = load_config()

    if agent_id not in config.agents.profiles:
        raise ConfigurationException(
            config_key="agent",
            message=f"Agent '{agent_id}' not found in config",
        )

    agent_ref = config.agents.profiles[agent_id]
    workspace_dir = Path(agent_ref.workspace_dir).expanduser()
    agent_config_path = workspace_dir / "agent.json"
    with _agent_config_lock:
        write_json_atomic(
            agent_config_path,
            agent_config.model_dump(exclude_none=True),
        )
        _agent_config_cache.pop(agent_id, None)


def migrate_legacy_config_to_multi_agent() -> bool:
    """Migrate legacy single-agent config to new multi-agent structure.

    Returns:
        bool: True if migration was performed, False if already migrated
    """
    from .utils import load_config, save_config

    config = load_config()

    # Check if already migrated (new structure has only AgentProfileRef)
    if "default" in config.agents.profiles:
        agent_ref = config.agents.profiles["default"]
        # If it's already a AgentProfileRef, migration done
        if isinstance(agent_ref, AgentProfileRef):
            # Check if default agent config exists
            workspace_dir = Path(agent_ref.workspace_dir).expanduser()
            agent_config_path = workspace_dir / "agent.json"
            if agent_config_path.exists():
                return False  # Already migrated

    # Perform migration
    print("Migrating legacy config to multi-agent structure...")

    # Extract legacy agent configuration
    legacy_agents = config.agents

    # Create default agent workspace
    default_workspace = Path(f"{WORKING_DIR}/workspaces/default").expanduser()
    default_workspace.mkdir(parents=True, exist_ok=True)

    # Inherit the global active model so the new agent.json has a valid
    # active_model pointer from the start (fixes #4937).
    try:
        from ..providers import ProviderManager

        global_active_model = ProviderManager.get_instance().get_active_model()
    except Exception:
        global_active_model = None
        logger.info(
            "Could not resolve global active model during migration; "
            "agent will be created without active_model.",
        )

    # Create default agent configuration from legacy settings
    default_agent_config = AgentProfileConfig(
        id="default",
        name="Default Agent",
        description="Default QwenPaw agent",
        workspace_dir=str(default_workspace),
        channels=config.channels if config.channels else None,
        mcp=config.mcp if config.mcp else None,
        heartbeat=(
            legacy_agents.defaults.heartbeat
            if legacy_agents.defaults
            else None
        ),
        running=(
            legacy_agents.running
            if legacy_agents.running
            else AgentsRunningConfig()
        ),
        llm_routing=(
            legacy_agents.llm_routing
            if legacy_agents.llm_routing
            else AgentsLLMRoutingConfig()
        ),
        system_prompt_files=(
            legacy_agents.system_prompt_files
            if legacy_agents.system_prompt_files
            else ["AGENTS.md", "SOUL.md", "PROFILE.md"]
        ),
        tools=config.tools if config.tools else None,
        security=config.security if config.security else None,
        active_model=global_active_model,
    )

    # Save default agent configuration to workspace
    agent_config_path = default_workspace / "agent.json"
    with open(agent_config_path, "w", encoding="utf-8") as f:
        json.dump(
            default_agent_config.model_dump(exclude_none=True),
            f,
            ensure_ascii=False,
            indent=2,
        )

    # Migrate existing workspace files from legacy default working dir.
    # When QWENPAW_WORKING_DIR is customized, historical data may still exist
    # under "~/.copaw".
    old_workspace = Path("~/.copaw").expanduser().resolve()

    # Move sessions, memory, and other workspace files
    for item_name in ["sessions", "memory", "jobs.json"]:
        old_path = old_workspace / item_name
        if old_path.exists():
            new_path = default_workspace / item_name
            if not new_path.exists():
                import shutil

                if old_path.is_dir():
                    shutil.copytree(old_path, new_path)
                else:
                    shutil.copy2(old_path, new_path)
                print(f"  Migrated {item_name} to default workspace")

    # Copy markdown files (AGENTS.md, SOUL.md, PROFILE.md)
    for md_file in ["AGENTS.md", "SOUL.md", "PROFILE.md"]:
        old_md = old_workspace / md_file
        if old_md.exists():
            new_md = default_workspace / md_file
            if not new_md.exists():
                import shutil

                shutil.copy2(old_md, new_md)
                print(f"  Migrated {md_file} to default workspace")

    # Update root config.json to new structure
    # CRITICAL: Preserve legacy agent fields for downgrade compatibility
    config.agents = AgentsConfig(
        active_agent="default",
        profiles={
            "default": AgentProfileRef(
                id="default",
                workspace_dir=str(default_workspace),
            ),
        },
        # Preserve legacy fields with values from migrated agent config
        running=default_agent_config.running,
        llm_routing=default_agent_config.llm_routing,
        language=(
            default_agent_config.language
            if hasattr(default_agent_config, "language")
            else "zh"
        ),
        system_prompt_files=default_agent_config.system_prompt_files,
    )

    # IMPORTANT: Keep channels, mcp, tools, security in root config for
    # backward compatibility. Do NOT clear these fields.
    # Old versions expect these fields to exist with valid values.

    save_config(config)

    print("Migration completed successfully!")
    print(f"  Default agent workspace: {default_workspace}")
    print(f"  Default agent config: {agent_config_path}")

    return True


def get_model_max_input_length(
    agent_config: "AgentProfileConfig",
) -> int:
    """Return the active model's resolved context window.

    Delegates to ``Provider.get_context_size`` — the SAME resolution the
    compaction trigger uses (explicit ``max_input_length`` > static
    context-window catalog > 128k default) — so /history, usage%%, and
    daemon status can never disagree with when compression actually fires.
    Falls back to 128 * 1024 (131072) if the provider is unavailable.
    Accepts an already-loaded *agent_config* to avoid redundant file I/O
    on hot paths (pre_reasoning, compact_context, summarize, etc.).
    """
    from ..providers import ProviderManager

    model_slot = agent_config.active_model
    # Fallback: if agent.json doesn't have active_model, try ProviderManager
    if not model_slot or not model_slot.provider_id:
        try:
            manager = ProviderManager.get_instance()
            model_slot = manager.get_active_model()
        except Exception:
            pass

    if model_slot and model_slot.provider_id and model_slot.model:
        try:
            manager = ProviderManager.get_instance()
            provider = manager.get_provider(model_slot.provider_id)
            if provider:
                return provider.get_context_size(model_slot.model)
        except Exception:
            pass
    logger.debug(
        "Could not resolve max_input_length for agent '%s' "
        "(active_model=%s), falling back to 128K default.",
        getattr(agent_config, "id", "?"),
        agent_config.active_model,
    )
    return 128 * 1024
