agent-Specialization/modules/personalization_manager.py

"""Utilities for managing per-user personalization settings."""

from __future__ import annotations

import json
from copy import deepcopy
from pathlib import Path
from typing import Any, Dict, Iterable, Optional, Union

try:
    from config.limits import THINKING_FAST_INTERVAL
except ImportError:
    THINKING_FAST_INTERVAL = 10

from core.tool_config import TOOL_CATEGORIES
from config.model_profiles import get_registered_model_keys

ALLOWED_RUN_MODES = {"fast", "thinking", "deep"}
ALLOWED_PERMISSION_MODES = {"readonly", "approval", "unrestricted"}

PERSONALIZATION_FILENAME = "personalization.json"
MAX_SHORT_FIELD_LENGTH = 20
MAX_CONSIDERATION_LENGTH = 50
MAX_CONSIDERATION_ITEMS = 10
TONE_PRESETS = ["健谈", "幽默", "直言不讳", "鼓励性", "诗意", "企业商务", "打破常规", "同理心"]
THINKING_INTERVAL_MIN = 1
THINKING_INTERVAL_MAX = 50
DEFAULT_SHALLOW_COMPRESS_TRIGGER_TOKENS = 80_000
DEFAULT_SHALLOW_COMPRESS_KEEP_RECENT_TOOLS = 15
DEFAULT_SHALLOW_COMPRESS_MAX_REPLACE_PER_ROUND = 10
DEFAULT_SHALLOW_COMPRESS_TRIGGER_TOOL_CALLS_INTERVAL = 10
DEFAULT_DEEP_COMPRESS_TRIGGER_TOKENS = 150_000
MIN_COMPRESSION_TRIGGER_TOKENS = 1_000
MAX_COMPRESSION_TRIGGER_TOKENS = 2_000_000
MIN_SHALLOW_KEEP_RECENT_TOOLS = 0
MAX_SHALLOW_KEEP_RECENT_TOOLS = 500
MIN_SHALLOW_MAX_REPLACE_PER_ROUND = 1
MAX_SHALLOW_MAX_REPLACE_PER_ROUND = 200
MIN_SHALLOW_TRIGGER_TOOL_CALLS_INTERVAL = 1
MAX_SHALLOW_TRIGGER_TOOL_CALLS_INTERVAL = 200
MIN_SHALLOW_KEEP_USER_TURN_TOOLS = 0
MAX_SHALLOW_KEEP_USER_TURN_TOOLS = 50
DEFAULT_SHALLOW_KEEP_USER_TURN_TOOLS = 3

DEFAULT_PERSONALIZATION_CONFIG: Dict[str, Any] = {
    "enabled": False,
    "self_identify": "",
    "user_name": "",
    "use_custom_names": False,
    "profession": "",
    "tone": "",
    "considerations": [],
    "thinking_interval": None,
    "disabled_tool_categories": [],
    "enabled_skills": None,
    "skills_catalog_snapshot": None,
    "default_run_mode": None,
    "default_permission_mode": "unrestricted",
    "auto_generate_title": True,
    "tool_intent_enabled": True,
    "skill_hints_enabled": False,  # Skill 提示系统开关（默认关闭）
    "skill_strict_terminal_enabled": False,  # 强约束：terminal 系列工具需先阅读 terminal-guide
    "skill_strict_sub_agent_enabled": False,  # 强约束：子智能体系列工具需先阅读 sub-agent-guide
    "skill_strict_run_command_foreground_enabled": False,  # 强约束：run_command 前台模式需先阅读 run-command-guide
    "skill_strict_run_command_background_enabled": False,  # 强约束：run_command 后台模式需先阅读 run-command-guide
    "default_model": "kimi-k2.5",
    "image_compression": "original",  # original / 1080p / 720p / 540p
    "auto_shallow_compress_enabled": False,
    "auto_deep_compress_enabled": False,
    "shallow_compress_trigger_tokens": None,
    "shallow_compress_keep_recent_tools": None,
    "shallow_compress_max_replace_per_round": None,
    "shallow_compress_trigger_tool_calls_interval": None,
    "shallow_compress_keep_user_turn_tools": None,  # 保留最近N次用户输入后的工具不压缩（默认3）
    "deep_compress_trigger_tokens": None,
    "silent_tool_disable": False,  # 禁用工具时不向模型插入提示
    "enhanced_tool_display": True,  # 增强工具显示
    "versioning_restore_mode": "overwrite",  # 版本回溯模式固定为 overwrite
    "agents_md_auto_inject": False,  # AGENTS.md 自动注入开关
    "allow_root_file_creation": False,  # 允许在根目录创建文件开关
}

__all__ = [
    "PERSONALIZATION_FILENAME",
    "DEFAULT_PERSONALIZATION_CONFIG",
    "TONE_PRESETS",
    "MAX_CONSIDERATION_ITEMS",
    "load_personalization_config",
    "save_personalization_config",
    "ensure_personalization_config",
    "build_personalization_prompt",
    "sanitize_personalization_payload",
    "resolve_context_compression_settings",
    "validate_context_compression_settings",
    "ALLOWED_PERMISSION_MODES",
]


PathLike = Union[str, Path]


def _ensure_parent(path: Path) -> None:
    path.parent.mkdir(parents=True, exist_ok=True)


def _to_path(base: PathLike) -> Path:
    base_path = Path(base).expanduser()
    if base_path.is_dir():
        return base_path / PERSONALIZATION_FILENAME
    return base_path


def ensure_personalization_config(base_dir: PathLike) -> Dict[str, Any]:
    """Ensure the personalization file exists and return its content."""
    path = _to_path(base_dir)
    _ensure_parent(path)
    if not path.exists():
        with open(path, "w", encoding="utf-8") as f:
            json.dump(DEFAULT_PERSONALIZATION_CONFIG, f, ensure_ascii=False, indent=2)
        return deepcopy(DEFAULT_PERSONALIZATION_CONFIG)
    return load_personalization_config(base_dir)


def load_personalization_config(base_dir: PathLike) -> Dict[str, Any]:
    """Load personalization config; fall back to defaults on errors."""
    path = _to_path(base_dir)
    _ensure_parent(path)
    if not path.exists():
        return ensure_personalization_config(base_dir)
    try:
        with open(path, "r", encoding="utf-8") as f:
            raw = json.load(f)
            sanitized = sanitize_personalization_payload(raw)
            # 若发现缺失字段（如默认模型）或数据被规范化，主动写回文件，避免下一次读取仍为旧格式
            if sanitized != raw:
                with open(path, "w", encoding="utf-8") as wf:
                    json.dump(sanitized, wf, ensure_ascii=False, indent=2)
            return sanitized
    except (json.JSONDecodeError, OSError):
        # 重置为默认配置，避免错误阻塞
        with open(path, "w", encoding="utf-8") as f:
            json.dump(DEFAULT_PERSONALIZATION_CONFIG, f, ensure_ascii=False, indent=2)
        return deepcopy(DEFAULT_PERSONALIZATION_CONFIG)


def sanitize_personalization_payload(
    payload: Optional[Dict[str, Any]],
    fallback: Optional[Dict[str, Any]] = None
) -> Dict[str, Any]:
    """Normalize payload structure and clamp field lengths."""
    base = deepcopy(DEFAULT_PERSONALIZATION_CONFIG)
    if fallback:
        base.update(fallback)
    data = payload or {}
    allowed_tool_categories = set(TOOL_CATEGORIES.keys())
    allowed_models = set(get_registered_model_keys(visible_only=True))
    allowed_image_modes = {"original", "1080p", "720p", "540p"}

    def _resolve_short_field(key: str) -> str:
        if key in data:
            return _sanitize_short_field(data.get(key))
        return _sanitize_short_field(base.get(key))

    base["enabled"] = bool(data.get("enabled", base["enabled"]))
    base["auto_generate_title"] = bool(data.get("auto_generate_title", base["auto_generate_title"]))
    base["self_identify"] = _resolve_short_field("self_identify")
    base["user_name"] = _resolve_short_field("user_name")
    base["profession"] = _resolve_short_field("profession")
    base["tone"] = _resolve_short_field("tone")
    if "considerations" in data:
        base["considerations"] = _sanitize_considerations(data.get("considerations"))
    else:
        base["considerations"] = _sanitize_considerations(base.get("considerations", []))

    if "thinking_interval" in data:
        base["thinking_interval"] = _sanitize_thinking_interval(data.get("thinking_interval"))
    else:
        base["thinking_interval"] = _sanitize_thinking_interval(base.get("thinking_interval"))

    # 工具意图提示开关
    if "tool_intent_enabled" in data:
        base["tool_intent_enabled"] = bool(data.get("tool_intent_enabled"))
    else:
        base["tool_intent_enabled"] = bool(base.get("tool_intent_enabled"))

    # Skill 提示系统开关
    if "skill_hints_enabled" in data:
        base["skill_hints_enabled"] = bool(data.get("skill_hints_enabled"))
    else:
        base["skill_hints_enabled"] = bool(base.get("skill_hints_enabled"))

    # Skill 强约束开关（仅针对已启用的 skill 生效）
    if "skill_strict_terminal_enabled" in data:
        base["skill_strict_terminal_enabled"] = bool(data.get("skill_strict_terminal_enabled"))
    else:
        base["skill_strict_terminal_enabled"] = bool(base.get("skill_strict_terminal_enabled", False))

    if "skill_strict_sub_agent_enabled" in data:
        base["skill_strict_sub_agent_enabled"] = bool(data.get("skill_strict_sub_agent_enabled"))
    else:
        base["skill_strict_sub_agent_enabled"] = bool(base.get("skill_strict_sub_agent_enabled", False))

    if "skill_strict_run_command_foreground_enabled" in data:
        base["skill_strict_run_command_foreground_enabled"] = bool(data.get("skill_strict_run_command_foreground_enabled"))
    else:
        base["skill_strict_run_command_foreground_enabled"] = bool(
            base.get("skill_strict_run_command_foreground_enabled", False)
        )

    if "skill_strict_run_command_background_enabled" in data:
        base["skill_strict_run_command_background_enabled"] = bool(data.get("skill_strict_run_command_background_enabled"))
    else:
        base["skill_strict_run_command_background_enabled"] = bool(
            base.get("skill_strict_run_command_background_enabled", False)
        )

    if "disabled_tool_categories" in data:
        base["disabled_tool_categories"] = _sanitize_tool_categories(data.get("disabled_tool_categories"), allowed_tool_categories)
    else:
        base["disabled_tool_categories"] = _sanitize_tool_categories(base.get("disabled_tool_categories"), allowed_tool_categories)

    if "enabled_skills" in data:
        base["enabled_skills"] = _sanitize_skills(data.get("enabled_skills"))
    else:
        base["enabled_skills"] = _sanitize_skills(base.get("enabled_skills"))

    if "skills_catalog_snapshot" in data:
        base["skills_catalog_snapshot"] = _sanitize_skills(data.get("skills_catalog_snapshot"))
    else:
        base["skills_catalog_snapshot"] = _sanitize_skills(base.get("skills_catalog_snapshot"))

    if "default_run_mode" in data:
        base["default_run_mode"] = _sanitize_run_mode(data.get("default_run_mode"))
    else:
        base["default_run_mode"] = _sanitize_run_mode(base.get("default_run_mode"))

    permission_mode = data.get("default_permission_mode", base.get("default_permission_mode"))
    if isinstance(permission_mode, str) and permission_mode in ALLOWED_PERMISSION_MODES:
        base["default_permission_mode"] = permission_mode
    elif base.get("default_permission_mode") not in ALLOWED_PERMISSION_MODES:
        base["default_permission_mode"] = "unrestricted"

    base["versioning_restore_mode"] = "overwrite"

    # 默认模型
    chosen_model = data.get("default_model", base.get("default_model"))
    if isinstance(chosen_model, str) and chosen_model in allowed_models:
        base["default_model"] = chosen_model
    elif allowed_models and base.get("default_model") not in allowed_models:
        base["default_model"] = "kimi-k2.5"

    # 图片压缩模式
    img_mode = data.get("image_compression", base.get("image_compression"))
    if isinstance(img_mode, str) and img_mode in allowed_image_modes:
        base["image_compression"] = img_mode
    elif base.get("image_compression") not in allowed_image_modes:
        base["image_compression"] = "original"

    if "auto_shallow_compress_enabled" in data:
        base["auto_shallow_compress_enabled"] = bool(data.get("auto_shallow_compress_enabled"))
    else:
        base["auto_shallow_compress_enabled"] = bool(base.get("auto_shallow_compress_enabled", False))

    if "auto_deep_compress_enabled" in data:
        base["auto_deep_compress_enabled"] = bool(data.get("auto_deep_compress_enabled"))
    else:
        base["auto_deep_compress_enabled"] = bool(base.get("auto_deep_compress_enabled", False))

    if "shallow_compress_trigger_tokens" in data:
        base["shallow_compress_trigger_tokens"] = _sanitize_optional_int(
            data.get("shallow_compress_trigger_tokens"),
            min_value=MIN_COMPRESSION_TRIGGER_TOKENS,
            max_value=MAX_COMPRESSION_TRIGGER_TOKENS,
        )
    else:
        base["shallow_compress_trigger_tokens"] = _sanitize_optional_int(
            base.get("shallow_compress_trigger_tokens"),
            min_value=MIN_COMPRESSION_TRIGGER_TOKENS,
            max_value=MAX_COMPRESSION_TRIGGER_TOKENS,
        )

    if "shallow_compress_keep_recent_tools" in data:
        base["shallow_compress_keep_recent_tools"] = _sanitize_optional_int(
            data.get("shallow_compress_keep_recent_tools"),
            min_value=MIN_SHALLOW_KEEP_RECENT_TOOLS,
            max_value=MAX_SHALLOW_KEEP_RECENT_TOOLS,
        )
    else:
        base["shallow_compress_keep_recent_tools"] = _sanitize_optional_int(
            base.get("shallow_compress_keep_recent_tools"),
            min_value=MIN_SHALLOW_KEEP_RECENT_TOOLS,
            max_value=MAX_SHALLOW_KEEP_RECENT_TOOLS,
        )

    if "shallow_compress_keep_user_turn_tools" in data:
        base["shallow_compress_keep_user_turn_tools"] = _sanitize_optional_int(
            data.get("shallow_compress_keep_user_turn_tools"),
            min_value=MIN_SHALLOW_KEEP_USER_TURN_TOOLS,
            max_value=MAX_SHALLOW_KEEP_USER_TURN_TOOLS,
        )
    else:
        base["shallow_compress_keep_user_turn_tools"] = _sanitize_optional_int(
            base.get("shallow_compress_keep_user_turn_tools"),
            min_value=MIN_SHALLOW_KEEP_USER_TURN_TOOLS,
            max_value=MAX_SHALLOW_KEEP_USER_TURN_TOOLS,
        )

    if "shallow_compress_max_replace_per_round" in data:
        base["shallow_compress_max_replace_per_round"] = _sanitize_optional_int(
            data.get("shallow_compress_max_replace_per_round"),
            min_value=MIN_SHALLOW_MAX_REPLACE_PER_ROUND,
            max_value=MAX_SHALLOW_MAX_REPLACE_PER_ROUND,
        )
    else:
        base["shallow_compress_max_replace_per_round"] = _sanitize_optional_int(
            base.get("shallow_compress_max_replace_per_round"),
            min_value=MIN_SHALLOW_MAX_REPLACE_PER_ROUND,
            max_value=MAX_SHALLOW_MAX_REPLACE_PER_ROUND,
        )

    if "deep_compress_trigger_tokens" in data:
        base["deep_compress_trigger_tokens"] = _sanitize_optional_int(
            data.get("deep_compress_trigger_tokens"),
            min_value=MIN_COMPRESSION_TRIGGER_TOKENS,
            max_value=MAX_COMPRESSION_TRIGGER_TOKENS,
        )
    else:
        base["deep_compress_trigger_tokens"] = _sanitize_optional_int(
            base.get("deep_compress_trigger_tokens"),
            min_value=MIN_COMPRESSION_TRIGGER_TOKENS,
            max_value=MAX_COMPRESSION_TRIGGER_TOKENS,
        )

    if "shallow_compress_trigger_tool_calls_interval" in data:
        base["shallow_compress_trigger_tool_calls_interval"] = _sanitize_optional_int(
            data.get("shallow_compress_trigger_tool_calls_interval"),
            min_value=MIN_SHALLOW_TRIGGER_TOOL_CALLS_INTERVAL,
            max_value=MAX_SHALLOW_TRIGGER_TOOL_CALLS_INTERVAL,
        )
    else:
        base["shallow_compress_trigger_tool_calls_interval"] = _sanitize_optional_int(
            base.get("shallow_compress_trigger_tool_calls_interval"),
            min_value=MIN_SHALLOW_TRIGGER_TOOL_CALLS_INTERVAL,
            max_value=MAX_SHALLOW_TRIGGER_TOOL_CALLS_INTERVAL,
        )

    # 静默禁用工具提示
    if "silent_tool_disable" in data:
        base["silent_tool_disable"] = bool(data.get("silent_tool_disable"))
    else:
        base["silent_tool_disable"] = bool(base.get("silent_tool_disable"))

    # 增强工具显示
    if "enhanced_tool_display" in data:
        base["enhanced_tool_display"] = bool(data.get("enhanced_tool_display"))
    else:
        base["enhanced_tool_display"] = bool(base.get("enhanced_tool_display", True))

    # 使用自定义称呼
    if "use_custom_names" in data:
        base["use_custom_names"] = bool(data.get("use_custom_names"))
    else:
        base["use_custom_names"] = bool(base.get("use_custom_names"))

    # AGENTS.md 自动注入开关
    if "agents_md_auto_inject" in data:
        base["agents_md_auto_inject"] = bool(data.get("agents_md_auto_inject"))
    else:
        base["agents_md_auto_inject"] = bool(base.get("agents_md_auto_inject", False))

    # 允许根目录创建文件开关
    if "allow_root_file_creation" in data:
        base["allow_root_file_creation"] = bool(data.get("allow_root_file_creation"))
    else:
        base["allow_root_file_creation"] = bool(base.get("allow_root_file_creation", False))

    return base


def _sanitize_skills(value: Any) -> Optional[list]:
    """Sanitize enabled skills list / 清洗启用技能列表。"""
    if value is None:
        return None
    if not isinstance(value, list):
        return []
    cleaned: list = []
    seen = set()
    for item in value:
        if not isinstance(item, str):
            continue
        skill_id = item.strip()
        if not skill_id or skill_id in seen:
            continue
        cleaned.append(skill_id)
        seen.add(skill_id)
    return cleaned


def save_personalization_config(base_dir: PathLike, payload: Dict[str, Any]) -> Dict[str, Any]:
    """Persist sanitized personalization config and return it."""
    existing = load_personalization_config(base_dir)
    config = sanitize_personalization_payload(payload, fallback=existing)
    validate_context_compression_settings(config)
    path = _to_path(base_dir)
    _ensure_parent(path)
    with open(path, "w", encoding="utf-8") as f:
        json.dump(config, f, ensure_ascii=False, indent=2)
    return config


def _sanitize_optional_int(value: Any, *, min_value: int, max_value: int) -> Optional[int]:
    if value is None or value == "":
        return None
    if isinstance(value, bool):
        return None
    try:
        parsed = int(value)
    except (TypeError, ValueError):
        return None
    if parsed < min_value:
        return min_value
    if parsed > max_value:
        return max_value
    return parsed


def resolve_context_compression_settings(config: Optional[Dict[str, Any]]) -> Dict[str, int]:
    data = config or {}
    shallow_trigger = _sanitize_optional_int(
        data.get("shallow_compress_trigger_tokens"),
        min_value=MIN_COMPRESSION_TRIGGER_TOKENS,
        max_value=MAX_COMPRESSION_TRIGGER_TOKENS,
    ) or DEFAULT_SHALLOW_COMPRESS_TRIGGER_TOKENS
    shallow_keep_recent = _sanitize_optional_int(
        data.get("shallow_compress_keep_recent_tools"),
        min_value=MIN_SHALLOW_KEEP_RECENT_TOOLS,
        max_value=MAX_SHALLOW_KEEP_RECENT_TOOLS,
    )
    if shallow_keep_recent is None:
        shallow_keep_recent = DEFAULT_SHALLOW_COMPRESS_KEEP_RECENT_TOOLS
    shallow_keep_user_turn = _sanitize_optional_int(
        data.get("shallow_compress_keep_user_turn_tools"),
        min_value=MIN_SHALLOW_KEEP_USER_TURN_TOOLS,
        max_value=MAX_SHALLOW_KEEP_USER_TURN_TOOLS,
    )
    if shallow_keep_user_turn is None:
        shallow_keep_user_turn = DEFAULT_SHALLOW_KEEP_USER_TURN_TOOLS
    shallow_max_replace = _sanitize_optional_int(
        data.get("shallow_compress_max_replace_per_round"),
        min_value=MIN_SHALLOW_MAX_REPLACE_PER_ROUND,
        max_value=MAX_SHALLOW_MAX_REPLACE_PER_ROUND,
    ) or DEFAULT_SHALLOW_COMPRESS_MAX_REPLACE_PER_ROUND
    shallow_trigger_interval = _sanitize_optional_int(
        data.get("shallow_compress_trigger_tool_calls_interval"),
        min_value=MIN_SHALLOW_TRIGGER_TOOL_CALLS_INTERVAL,
        max_value=MAX_SHALLOW_TRIGGER_TOOL_CALLS_INTERVAL,
    ) or DEFAULT_SHALLOW_COMPRESS_TRIGGER_TOOL_CALLS_INTERVAL
    deep_trigger = _sanitize_optional_int(
        data.get("deep_compress_trigger_tokens"),
        min_value=MIN_COMPRESSION_TRIGGER_TOKENS,
        max_value=MAX_COMPRESSION_TRIGGER_TOKENS,
    ) or DEFAULT_DEEP_COMPRESS_TRIGGER_TOKENS
    if deep_trigger <= shallow_trigger:
        deep_trigger = shallow_trigger + 1
    return {
        "shallow_trigger_tokens": shallow_trigger,
        "shallow_keep_recent_tools": shallow_keep_recent,
        "shallow_keep_user_turn_tools": shallow_keep_user_turn,
        "shallow_max_replace_per_round": shallow_max_replace,
        "shallow_trigger_tool_calls_interval": shallow_trigger_interval,
        "deep_trigger_tokens": deep_trigger,
    }


def validate_context_compression_settings(config: Optional[Dict[str, Any]]) -> None:
    data = config or {}
    shallow_trigger = _sanitize_optional_int(
        data.get("shallow_compress_trigger_tokens"),
        min_value=MIN_COMPRESSION_TRIGGER_TOKENS,
        max_value=MAX_COMPRESSION_TRIGGER_TOKENS,
    ) or DEFAULT_SHALLOW_COMPRESS_TRIGGER_TOKENS
    deep_trigger = _sanitize_optional_int(
        data.get("deep_compress_trigger_tokens"),
        min_value=MIN_COMPRESSION_TRIGGER_TOKENS,
        max_value=MAX_COMPRESSION_TRIGGER_TOKENS,
    ) or DEFAULT_DEEP_COMPRESS_TRIGGER_TOKENS
    if deep_trigger <= shallow_trigger:
        raise ValueError("深压缩触发上下文必须大于浅压缩触发上下文")


def build_personalization_prompt(
    config: Optional[Dict[str, Any]],
    include_header: bool = True
) -> Optional[str]:
    """Generate the personalization prompt text based on config."""
    if not config or not config.get("enabled"):
        return None

    lines = []
    if include_header:
        lines.append("用户的个性化数据，请回答时务必参照这些信息")

    if config.get("self_identify"):
        lines.append(f"用户希望你自称：{config['self_identify']}")
    if config.get("user_name"):
        lines.append(f"用户希望你称呼为：{config['user_name']}")
    if config.get("profession"):
        lines.append(f"用户的职业是：{config['profession']}")
    if config.get("tone"):
        lines.append(f"用户希望你使用 {config['tone']} 的语气与TA交流")

    considerations: Iterable[str] = config.get("considerations") or []
    considerations = [item for item in considerations if item]
    if considerations:
        lines.append("用户希望你在回答问题时必须考虑的信息是：")
        for idx, item in enumerate(considerations, 1):
            lines.append(f"{idx}. {item}")

    if len(lines) == (1 if include_header else 0):
        # 没有任何有效内容时不注入
        return None
    return "\n".join(lines)


def _sanitize_short_field(value: Optional[str]) -> str:
    if not value:
        return ""
    text = str(value).strip()
    if not text:
        return ""
    return text[:MAX_SHORT_FIELD_LENGTH]


def _sanitize_considerations(value: Any) -> list:
    if not isinstance(value, list):
        return []
    cleaned = []
    for item in value:
        if not isinstance(item, str):
            continue
        text = item.strip()
        if not text:
            continue
        cleaned.append(text[:MAX_CONSIDERATION_LENGTH])
        if len(cleaned) >= MAX_CONSIDERATION_ITEMS:
            break
    return cleaned


def _sanitize_thinking_interval(value: Any) -> Optional[int]:
    if value is None or value == "":
        return None
    try:
        interval = int(value)
    except (TypeError, ValueError):
        return None
    interval = max(THINKING_INTERVAL_MIN, min(THINKING_INTERVAL_MAX, interval))
    if interval == THINKING_FAST_INTERVAL:
        return None
    return interval


def _sanitize_tool_categories(value: Any, allowed: set) -> list:
    if not isinstance(value, list):
        return []
    result = []
    for item in value:
        if not isinstance(item, str):
            continue
        candidate = item.strip()
        if not candidate or candidate not in allowed:
            continue
        if candidate not in result:
            result.append(candidate)
    return result


def _sanitize_run_mode(value: Any) -> Optional[str]:
    if value is None:
        return None
    if isinstance(value, str):
        candidate = value.strip().lower()
        if candidate in ALLOWED_RUN_MODES:
            return candidate
    return None