539 lines
30 KiB
Python
539 lines
30 KiB
Python
import asyncio
|
||
import json
|
||
from datetime import datetime
|
||
from pathlib import Path
|
||
from typing import Any, Dict, List, Optional, Set
|
||
|
||
try:
|
||
from config import (
|
||
OUTPUT_FORMATS, DATA_DIR, PROMPTS_DIR, NEED_CONFIRMATION,
|
||
MAX_TERMINALS, TERMINAL_BUFFER_SIZE, TERMINAL_DISPLAY_SIZE,
|
||
MAX_READ_FILE_CHARS, READ_TOOL_DEFAULT_MAX_CHARS,
|
||
READ_TOOL_DEFAULT_CONTEXT_BEFORE, READ_TOOL_DEFAULT_CONTEXT_AFTER,
|
||
READ_TOOL_MAX_CONTEXT_BEFORE, READ_TOOL_MAX_CONTEXT_AFTER,
|
||
READ_TOOL_DEFAULT_MAX_MATCHES, READ_TOOL_MAX_MATCHES,
|
||
READ_TOOL_MAX_FILE_SIZE,
|
||
TERMINAL_SANDBOX_MOUNT_PATH,
|
||
TERMINAL_SANDBOX_MODE,
|
||
TERMINAL_SANDBOX_CPUS,
|
||
TERMINAL_SANDBOX_MEMORY,
|
||
PROJECT_MAX_STORAGE_MB,
|
||
CUSTOM_TOOLS_ENABLED,
|
||
)
|
||
except ImportError:
|
||
import sys
|
||
project_root = Path(__file__).resolve().parents[2]
|
||
if str(project_root) not in sys.path:
|
||
sys.path.insert(0, str(project_root))
|
||
from config import (
|
||
OUTPUT_FORMATS, DATA_DIR, PROMPTS_DIR, NEED_CONFIRMATION,
|
||
MAX_TERMINALS, TERMINAL_BUFFER_SIZE, TERMINAL_DISPLAY_SIZE,
|
||
MAX_READ_FILE_CHARS, READ_TOOL_DEFAULT_MAX_CHARS,
|
||
READ_TOOL_DEFAULT_CONTEXT_BEFORE, READ_TOOL_DEFAULT_CONTEXT_AFTER,
|
||
READ_TOOL_MAX_CONTEXT_BEFORE, READ_TOOL_MAX_CONTEXT_AFTER,
|
||
READ_TOOL_DEFAULT_MAX_MATCHES, READ_TOOL_MAX_MATCHES,
|
||
READ_TOOL_MAX_FILE_SIZE,
|
||
TERMINAL_SANDBOX_MOUNT_PATH,
|
||
TERMINAL_SANDBOX_MODE,
|
||
TERMINAL_SANDBOX_CPUS,
|
||
TERMINAL_SANDBOX_MEMORY,
|
||
PROJECT_MAX_STORAGE_MB,
|
||
CUSTOM_TOOLS_ENABLED,
|
||
)
|
||
|
||
from modules.file_manager import FileManager
|
||
from modules.search_engine import SearchEngine
|
||
from modules.terminal_ops import TerminalOperator
|
||
from modules.memory_manager import MemoryManager
|
||
from modules.terminal_manager import TerminalManager
|
||
from modules.todo_manager import TodoManager
|
||
from modules.sub_agent import SubAgentManager
|
||
from modules.webpage_extractor import extract_webpage_content, tavily_extract
|
||
from modules.ocr_client import OCRClient
|
||
from modules.easter_egg_manager import EasterEggManager
|
||
from modules.personalization_manager import (
|
||
load_personalization_config,
|
||
build_personalization_prompt,
|
||
)
|
||
from modules.skills_manager import (
|
||
get_skills_catalog,
|
||
build_skills_list,
|
||
merge_enabled_skills,
|
||
build_skills_prompt,
|
||
infer_private_skills_dir,
|
||
)
|
||
from modules.custom_tool_registry import CustomToolRegistry, build_default_tool_category
|
||
from modules.custom_tool_executor import CustomToolExecutor
|
||
from modules.mcp_server_registry import build_default_mcp_category
|
||
|
||
try:
|
||
from config.limits import REASONING_EFFORT_LEVELS
|
||
except ImportError:
|
||
REASONING_EFFORT_LEVELS = ("low", "medium", "high", "xhigh", "max")
|
||
|
||
from modules.container_monitor import collect_stats, inspect_state
|
||
from core.tool_config import TOOL_CATEGORIES
|
||
from utils.api_client import APIClient
|
||
from utils.context_manager import ContextManager
|
||
from utils.tool_result_formatter import format_tool_result_for_context
|
||
from utils.logger import setup_logger
|
||
from config.model_profiles import (
|
||
get_model_profile,
|
||
get_model_prompt_replacements,
|
||
get_model_context_window,
|
||
)
|
||
|
||
logger = setup_logger(__name__)
|
||
DISABLE_LENGTH_CHECK = True
|
||
PERMISSION_MODES = {"readonly", "approval", "auto_approval", "unrestricted"}
|
||
# 运行模式（交互方式档）：计划 / 询问 / 执行。
|
||
# 与权限模式（能力档）正交；plan 档会联动锁死权限为只读。
|
||
WORK_MODES = {"plan", "ask", "execute"}
|
||
WORK_MODE_DEFAULT = "plan"
|
||
|
||
class MainTerminalToolsPolicyMixin:
|
||
def _ensure_runtime_tool_overrides(self) -> Dict[str, bool]:
|
||
overrides = getattr(self, "runtime_tool_category_overrides", None)
|
||
if not isinstance(overrides, dict):
|
||
overrides = {}
|
||
self.runtime_tool_category_overrides = overrides
|
||
return overrides
|
||
|
||
def _prune_runtime_tool_overrides(self) -> None:
|
||
overrides = self._ensure_runtime_tool_overrides()
|
||
valid_keys = set(getattr(self, "tool_categories_map", {}).keys())
|
||
for key in list(overrides.keys()):
|
||
if key not in valid_keys:
|
||
overrides.pop(key, None)
|
||
|
||
def _apply_runtime_tool_overrides(self) -> None:
|
||
overrides = self._ensure_runtime_tool_overrides()
|
||
if not overrides:
|
||
return
|
||
categories = getattr(self, "tool_categories_map", {}) or {}
|
||
for key, value in list(overrides.items()):
|
||
if key not in categories:
|
||
overrides.pop(key, None)
|
||
continue
|
||
forced = getattr(self, "admin_forced_category_states", {}).get(key)
|
||
if isinstance(forced, bool):
|
||
continue
|
||
self.tool_category_states[key] = bool(value)
|
||
|
||
def apply_personalization_preferences(self, config: Optional[Dict[str, Any]] = None, *, apply_default_model: bool = True, apply_default_modes: bool = True):
|
||
"""Apply persisted personalization settings that affect runtime behavior.
|
||
|
||
apply_default_modes=False 时跳过 default_model / default_run_mode /
|
||
default_reasoning_effort 三项默认值应用——对话加载链路必须传 False，
|
||
三者以对话 meta 为权威，默认值仅用于创建空对话。
|
||
"""
|
||
try:
|
||
effective_config = config or load_personalization_config(self.data_dir)
|
||
except Exception:
|
||
effective_config = {}
|
||
|
||
# 工具意图开关
|
||
self.tool_intent_enabled = bool(effective_config.get("tool_intent_enabled"))
|
||
# Skill 强约束开关
|
||
self.skill_strict_terminal_enabled = bool(effective_config.get("skill_strict_terminal_enabled", False))
|
||
self.skill_strict_sub_agent_enabled = bool(effective_config.get("skill_strict_sub_agent_enabled", False))
|
||
self.skill_strict_run_command_foreground_enabled = bool(
|
||
effective_config.get("skill_strict_run_command_foreground_enabled", False)
|
||
)
|
||
self.skill_strict_run_command_background_enabled = bool(
|
||
effective_config.get("skill_strict_run_command_background_enabled", False)
|
||
)
|
||
|
||
# 解析当前启用的 skills（用于强约束“仅对已启用 skill 生效”）
|
||
try:
|
||
skills_catalog = get_skills_catalog(private_dir=infer_private_skills_dir(self.data_dir))
|
||
enabled_skills = merge_enabled_skills(
|
||
effective_config.get("enabled_skills") if isinstance(effective_config, dict) else None,
|
||
skills_catalog,
|
||
effective_config.get("skills_catalog_snapshot") if isinstance(effective_config, dict) else None,
|
||
)
|
||
self.enabled_skill_ids = set(enabled_skills or [])
|
||
except Exception:
|
||
self.enabled_skill_ids = set()
|
||
|
||
disabled_categories = []
|
||
raw_disabled = effective_config.get("disabled_tool_categories")
|
||
if isinstance(raw_disabled, list):
|
||
disabled_categories = [
|
||
key for key in raw_disabled
|
||
if isinstance(key, str) and key in self.tool_categories_map
|
||
]
|
||
self.default_disabled_tool_categories = disabled_categories
|
||
|
||
# 图片压缩模式传递给上下文
|
||
img_mode = effective_config.get("image_compression")
|
||
if isinstance(img_mode, str):
|
||
self.context_manager.image_compression_mode = img_mode
|
||
|
||
# Reset category states to defaults before applying overrides
|
||
for key, category in self.tool_categories_map.items():
|
||
self.tool_category_states[key] = False if key in disabled_categories else category.default_enabled
|
||
self._prune_runtime_tool_overrides()
|
||
self._apply_runtime_tool_overrides()
|
||
self._refresh_disabled_tools()
|
||
|
||
# 默认模型偏好（优先应用，再处理运行模式）
|
||
preferred_model = effective_config.get("default_model")
|
||
if apply_default_modes and apply_default_model and isinstance(preferred_model, str) and preferred_model != self.model_key:
|
||
try:
|
||
self.set_model(preferred_model)
|
||
except Exception as exc:
|
||
logger.warning("忽略无效默认模型: %s (%s)", preferred_model, exc)
|
||
|
||
preferred_mode = effective_config.get("default_run_mode")
|
||
if apply_default_modes and isinstance(preferred_mode, str):
|
||
normalized_mode = preferred_mode.strip().lower()
|
||
if normalized_mode == "deep": # 旧版标识符映射
|
||
normalized_mode = "thinking"
|
||
if normalized_mode in {"fast", "thinking"} and normalized_mode != self.run_mode:
|
||
try:
|
||
self.set_run_mode(normalized_mode)
|
||
except ValueError:
|
||
logger.warning("忽略无效默认运行模式: %s", preferred_mode)
|
||
|
||
# 默认推理强度（None=默认，不传参）
|
||
if apply_default_modes:
|
||
preferred_effort = effective_config.get("default_reasoning_effort")
|
||
if isinstance(preferred_effort, str):
|
||
preferred_effort = preferred_effort.strip().lower() or None
|
||
if preferred_effort not in REASONING_EFFORT_LEVELS:
|
||
preferred_effort = None
|
||
else:
|
||
preferred_effort = None
|
||
try:
|
||
self.set_reasoning_effort(preferred_effort)
|
||
except (ValueError, AttributeError):
|
||
pass
|
||
|
||
# 静默禁用工具提示
|
||
self.silent_tool_disable = bool(effective_config.get("silent_tool_disable"))
|
||
permission_mode = effective_config.get("default_permission_mode")
|
||
if isinstance(permission_mode, str) and permission_mode in PERMISSION_MODES:
|
||
self.default_permission_mode = permission_mode
|
||
else:
|
||
self.default_permission_mode = "unrestricted"
|
||
if not getattr(self, "current_permission_mode", None):
|
||
self.current_permission_mode = self.default_permission_mode
|
||
work_mode = effective_config.get("default_work_mode")
|
||
if isinstance(work_mode, str) and work_mode.strip().lower() in WORK_MODES:
|
||
self.default_work_mode = work_mode.strip().lower()
|
||
else:
|
||
self.default_work_mode = WORK_MODE_DEFAULT
|
||
if not getattr(self, "current_work_mode", None):
|
||
self.current_work_mode = self.default_work_mode
|
||
|
||
def get_permission_mode(self) -> str:
|
||
mode = str(getattr(self, "current_permission_mode", "unrestricted") or "unrestricted")
|
||
if mode not in PERMISSION_MODES:
|
||
return "unrestricted"
|
||
return mode
|
||
|
||
def get_work_mode(self) -> str:
|
||
mode = str(getattr(self, "current_work_mode", None) or WORK_MODE_DEFAULT).strip().lower()
|
||
if mode not in WORK_MODES:
|
||
return WORK_MODE_DEFAULT
|
||
return mode
|
||
|
||
def set_work_mode(self, mode: str, *, persist: bool = True, conversation_id: Optional[str] = None) -> str:
|
||
normalized = str(mode or "").strip().lower()
|
||
if normalized not in WORK_MODES:
|
||
raise ValueError("无效运行模式，仅支持 plan / ask / execute")
|
||
self.current_work_mode = normalized
|
||
if not persist:
|
||
return normalized
|
||
|
||
conv_id = conversation_id or getattr(getattr(self, "context_manager", None), "current_conversation_id", None)
|
||
if conv_id and getattr(self, "context_manager", None):
|
||
try:
|
||
self.context_manager._get_conversation_manager_for_id(conv_id).update_conversation_metadata(conv_id,
|
||
{"work_mode": normalized},
|
||
)
|
||
if self.context_manager.current_conversation_id == conv_id:
|
||
self.context_manager.conversation_metadata["work_mode"] = normalized
|
||
except Exception:
|
||
pass
|
||
return normalized
|
||
|
||
def switch_work_mode(self, mode: str, *, persist: bool = True, conversation_id: Optional[str] = None) -> Dict[str, Any]:
|
||
"""切换运行模式（plan/ask/execute），并处理 plan ⇄ 只读权限的联动。
|
||
|
||
- 进入 plan：先把当前权限模式存为 pre_plan_permission_mode，再强制只读；
|
||
- 离开 plan：恢复 pre_plan_permission_mode（无记录则不动当前权限）；
|
||
- ask / execute 之间切换不动权限。
|
||
返回 {"mode": ..., "permission_mode": ...} 供调用方做通知与状态广播。
|
||
"""
|
||
normalized = str(mode or "").strip().lower()
|
||
if normalized not in WORK_MODES:
|
||
raise ValueError("无效运行模式，仅支持 plan / ask / execute")
|
||
previous = self.get_work_mode()
|
||
conv_id = conversation_id or getattr(getattr(self, "context_manager", None), "current_conversation_id", None)
|
||
|
||
entering_plan = normalized == "plan" and previous != "plan"
|
||
leaving_plan = previous == "plan" and normalized != "plan"
|
||
|
||
if entering_plan:
|
||
current_permission = self.get_permission_mode()
|
||
if current_permission != "readonly":
|
||
# 记录进入前的权限模式，供离开 plan 时恢复（Claude Code prePlanMode 式）
|
||
if persist:
|
||
try:
|
||
self._persist_runtime_mode_metadata({"pre_plan_permission_mode": current_permission})
|
||
except Exception:
|
||
pass
|
||
self.set_permission_mode("readonly", persist=persist, conversation_id=conv_id)
|
||
# 执行环境联动：direct ⇒ sandbox。只读权限在宿主机依赖 OS 沙箱硬限制，
|
||
# direct（完全访问）下无沙箱，只读形同虚设，必须一并锁回沙箱。
|
||
try:
|
||
if hasattr(self, "get_execution_mode") and self.get_execution_mode() == "direct":
|
||
if persist:
|
||
try:
|
||
self._persist_runtime_mode_metadata({"pre_plan_execution_mode": "direct"})
|
||
except Exception:
|
||
pass
|
||
if hasattr(self, "set_execution_mode"):
|
||
self.set_execution_mode("sandbox")
|
||
if persist and hasattr(self, "_persist_runtime_mode_metadata"):
|
||
try:
|
||
self._persist_runtime_mode_metadata({"execution_mode": "sandbox"})
|
||
except Exception:
|
||
pass
|
||
except Exception:
|
||
pass
|
||
# 先落新模式，再恢复权限：set_permission_mode 的 plan 锁以新模式为准
|
||
self.set_work_mode(normalized, persist=persist, conversation_id=conv_id)
|
||
if leaving_plan:
|
||
restore = None
|
||
try:
|
||
meta = getattr(getattr(self, "context_manager", None), "conversation_metadata", None) or {}
|
||
restore = str(meta.get("pre_plan_permission_mode") or "").strip().lower() or None
|
||
except Exception:
|
||
restore = None
|
||
if restore not in PERMISSION_MODES:
|
||
restore = None
|
||
if not restore:
|
||
# 无记录（如新对话直接以 plan 创建）：回落到个性化默认权限模式
|
||
try:
|
||
prefs = load_personalization_config(getattr(self, "data_dir", None))
|
||
candidate = str(prefs.get("default_permission_mode") or "").strip().lower()
|
||
if candidate in PERMISSION_MODES:
|
||
restore = candidate
|
||
except Exception:
|
||
restore = None
|
||
if restore and restore != "readonly":
|
||
self.set_permission_mode(restore, persist=persist, conversation_id=conv_id)
|
||
if persist:
|
||
try:
|
||
self._persist_runtime_mode_metadata({"pre_plan_permission_mode": None})
|
||
except Exception:
|
||
pass
|
||
# 恢复执行环境（仅当有明确进入前记录；无记录保持 sandbox，安全默认）
|
||
try:
|
||
meta = getattr(getattr(self, "context_manager", None), "conversation_metadata", None) or {}
|
||
pre_exec = str(meta.get("pre_plan_execution_mode") or "").strip().lower()
|
||
if pre_exec == "direct" and hasattr(self, "set_execution_mode"):
|
||
self.set_execution_mode("direct")
|
||
if persist and hasattr(self, "_persist_runtime_mode_metadata"):
|
||
try:
|
||
self._persist_runtime_mode_metadata({"execution_mode": "direct", "pre_plan_execution_mode": None})
|
||
except Exception:
|
||
pass
|
||
except Exception:
|
||
pass
|
||
return {"mode": normalized, "previous_mode": previous, "permission_mode": self.get_permission_mode()}
|
||
|
||
def set_permission_mode(self, mode: str, *, persist: bool = True, conversation_id: Optional[str] = None) -> str:
|
||
normalized = str(mode or "").strip().lower()
|
||
if normalized not in PERMISSION_MODES:
|
||
raise ValueError("无效权限模式，仅支持 readonly / approval / auto_approval / unrestricted")
|
||
# 计划模式下权限锁死为只读（后端强制）：仅允许保持/切到 readonly。
|
||
# 解除锁定的唯一路径是 switch_work_mode 先切离 plan 再恢复权限。
|
||
if normalized != "readonly" and getattr(self, "get_work_mode", None):
|
||
try:
|
||
if self.get_work_mode() == "plan":
|
||
raise ValueError("计划模式下权限模式锁定为只读，请先切换运行模式")
|
||
except AttributeError:
|
||
pass
|
||
previous = self.get_permission_mode()
|
||
entering_readonly = normalized == "readonly" and previous != "readonly"
|
||
leaving_readonly = previous == "readonly" and normalized != "readonly"
|
||
self.current_permission_mode = normalized
|
||
if not persist:
|
||
# 只读联动不依赖持久化：运行中 pending 切换路径（apply_pending_runtime_mode_changes）
|
||
# 也要内存级强制沙箱，只是不落 metadata（无记录则切离时保持沙箱，安全默认）。
|
||
self._apply_readonly_execution_mode_link(entering_readonly, leaving_readonly, persist=False)
|
||
return normalized
|
||
|
||
conv_id = conversation_id or getattr(getattr(self, "context_manager", None), "current_conversation_id", None)
|
||
if conv_id and getattr(self, "context_manager", None):
|
||
try:
|
||
self.context_manager._get_conversation_manager_for_id(conv_id).update_conversation_metadata(conv_id,
|
||
{"permission_mode": normalized},
|
||
)
|
||
if self.context_manager.current_conversation_id == conv_id:
|
||
self.context_manager.conversation_metadata["permission_mode"] = normalized
|
||
except Exception:
|
||
pass
|
||
self._apply_readonly_execution_mode_link(entering_readonly, leaving_readonly, persist=True)
|
||
return normalized
|
||
|
||
def _apply_readonly_execution_mode_link(self, entering: bool, leaving: bool, *, persist: bool) -> None:
|
||
"""只读权限 ⇄ 执行环境联动：进入 readonly 强制切沙箱，切离恢复进入前执行环境。
|
||
|
||
与 plan ⇄ readonly+sandbox 双锁逻辑对称：只读权限在宿主机依赖 OS 沙箱硬限制，
|
||
direct（完全访问）下无沙箱，只读形同虚设，必须一并锁回沙箱。
|
||
进入 readonly 时若执行环境为 direct，先存 pre_readonly_execution_mode 供切离时恢复；
|
||
切离 readonly 时仅当有明确进入前记录才恢复（无记录保持 sandbox，安全默认）。
|
||
"""
|
||
if entering:
|
||
try:
|
||
if hasattr(self, "get_execution_mode") and self.get_execution_mode() == "direct":
|
||
if persist and hasattr(self, "_persist_runtime_mode_metadata"):
|
||
try:
|
||
self._persist_runtime_mode_metadata({"pre_readonly_execution_mode": "direct"})
|
||
except Exception:
|
||
pass
|
||
if hasattr(self, "set_execution_mode"):
|
||
self.set_execution_mode("sandbox")
|
||
if persist and hasattr(self, "_persist_runtime_mode_metadata"):
|
||
try:
|
||
self._persist_runtime_mode_metadata({"execution_mode": "sandbox"})
|
||
except Exception:
|
||
pass
|
||
except Exception:
|
||
pass
|
||
return
|
||
if leaving:
|
||
try:
|
||
meta = getattr(getattr(self, "context_manager", None), "conversation_metadata", None) or {}
|
||
pre_exec = str(meta.get("pre_readonly_execution_mode") or "").strip().lower()
|
||
except Exception:
|
||
pre_exec = ""
|
||
if pre_exec == "direct" and hasattr(self, "set_execution_mode"):
|
||
try:
|
||
self.set_execution_mode("direct")
|
||
if persist and hasattr(self, "_persist_runtime_mode_metadata"):
|
||
try:
|
||
self._persist_runtime_mode_metadata({"execution_mode": "direct", "pre_readonly_execution_mode": None})
|
||
except Exception:
|
||
pass
|
||
except Exception:
|
||
pass
|
||
elif persist and hasattr(self, "_persist_runtime_mode_metadata"):
|
||
try:
|
||
self._persist_runtime_mode_metadata({"pre_readonly_execution_mode": None})
|
||
except Exception:
|
||
pass
|
||
|
||
def set_tool_category_enabled(self, category: str, enabled: bool) -> None:
|
||
"""设置工具类别的启用状态 / Toggle tool category enablement."""
|
||
categories = self.tool_categories_map
|
||
if category not in categories:
|
||
raise ValueError(f"未知的工具类别: {category}")
|
||
forced = self.admin_forced_category_states.get(category)
|
||
if isinstance(forced, bool) and forced != enabled:
|
||
raise ValueError("该类别被管理员强制为启用/禁用，无法修改")
|
||
final_enabled = bool(enabled)
|
||
self.tool_category_states[category] = final_enabled
|
||
self._ensure_runtime_tool_overrides()[category] = final_enabled
|
||
self._refresh_disabled_tools()
|
||
|
||
def set_admin_policy(
|
||
self,
|
||
categories: Optional[Dict[str, "ToolCategory"]] = None,
|
||
forced_category_states: Optional[Dict[str, Optional[bool]]] = None,
|
||
disabled_models: Optional[List[str]] = None,
|
||
) -> None:
|
||
"""应用管理员策略（工具分类、强制开关、模型禁用）。"""
|
||
if categories:
|
||
self.tool_categories_map = dict(categories)
|
||
# 保证自定义工具分类存在（仅当功能启用）
|
||
if self.custom_tools_enabled and "custom" not in self.tool_categories_map:
|
||
self.tool_categories_map["custom"] = type(next(iter(TOOL_CATEGORIES.values())))(
|
||
label="自定义工具",
|
||
tools=[],
|
||
default_enabled=True,
|
||
silent_when_disabled=False,
|
||
)
|
||
if getattr(self, "mcp_tools_enabled", False) and "mcp" not in self.tool_categories_map:
|
||
default_mcp_cat = build_default_mcp_category()
|
||
self.tool_categories_map["mcp"] = type(next(iter(TOOL_CATEGORIES.values())))(
|
||
label=default_mcp_cat["label"],
|
||
tools=["list_mcp_servers"],
|
||
default_enabled=True,
|
||
silent_when_disabled=False,
|
||
)
|
||
elif "mcp" in self.tool_categories_map:
|
||
self.tool_categories_map["mcp"].tools = ["list_mcp_servers"]
|
||
# 重新构建启用状态映射，保留已有值
|
||
new_states: Dict[str, bool] = {}
|
||
for key, cat in self.tool_categories_map.items():
|
||
if key in self.tool_category_states:
|
||
new_states[key] = self.tool_category_states[key]
|
||
else:
|
||
new_states[key] = cat.default_enabled
|
||
self.tool_category_states = new_states
|
||
# 清理已被移除的类别
|
||
for removed in list(self.tool_category_states.keys()):
|
||
if removed not in self.tool_categories_map:
|
||
self.tool_category_states.pop(removed, None)
|
||
|
||
self.admin_forced_category_states = forced_category_states or {}
|
||
self.admin_disabled_models = disabled_models or []
|
||
self._prune_runtime_tool_overrides()
|
||
self._apply_runtime_tool_overrides()
|
||
self._refresh_disabled_tools()
|
||
|
||
def get_tool_settings_snapshot(self) -> List[Dict[str, object]]:
|
||
"""获取工具类别状态快照 / Return tool category states snapshot."""
|
||
snapshot: List[Dict[str, object]] = []
|
||
categories = self.tool_categories_map
|
||
for key, category in categories.items():
|
||
forced = self.admin_forced_category_states.get(key)
|
||
enabled = self.tool_category_states.get(key, category.default_enabled)
|
||
if isinstance(forced, bool):
|
||
enabled = forced
|
||
snapshot.append({
|
||
"id": key,
|
||
"label": category.label,
|
||
"enabled": enabled,
|
||
"tools": list(category.tools),
|
||
"locked": isinstance(forced, bool),
|
||
"locked_state": forced if isinstance(forced, bool) else None,
|
||
})
|
||
return snapshot
|
||
|
||
def _refresh_disabled_tools(self) -> None:
|
||
"""刷新禁用工具列表 / Refresh disabled tool set."""
|
||
disabled: Set[str] = set()
|
||
notice: Set[str] = set()
|
||
categories = self.tool_categories_map
|
||
for key, category in categories.items():
|
||
state = self.tool_category_states.get(key, category.default_enabled)
|
||
forced = self.admin_forced_category_states.get(key)
|
||
if isinstance(forced, bool):
|
||
state = forced
|
||
if not state:
|
||
disabled.update(category.tools)
|
||
if not getattr(category, "silent_when_disabled", False):
|
||
notice.update(category.tools)
|
||
self.disabled_tools = disabled
|
||
self.disabled_notice_tools = notice
|
||
|
||
def _format_disabled_tool_notice(self) -> Optional[str]:
|
||
"""生成禁用工具提示信息 / Format disabled tool notice."""
|
||
if getattr(self, "silent_tool_disable", False):
|
||
return None
|
||
if not self.disabled_notice_tools:
|
||
return None
|
||
|
||
lines = ["=== 工具可用性提醒 ==="]
|
||
for tool_name in sorted(self.disabled_notice_tools):
|
||
lines.append(f"{tool_name}：已被用户禁用")
|
||
lines.append("=== 提示结束 ===")
|
||
return "\n".join(lines)
|