script-build-and-framework-tmp/agent/agent/tools/registry.py

"""
Tool Registry - 工具注册表和装饰器

职责：
1. @tool 装饰器：自动注册工具并生成 Schema
2. 管理所有工具的 Schema 和实现
3. 路由工具调用到具体实现
4. 支持域名过滤、敏感数据处理、工具统计

从 Resonote/llm/tools/registry.py 抽取并扩展
"""

import json
import inspect
import logging
import time
from typing import Any, Callable, Dict, List, Optional, Sequence

from agent.failures import (
	FailureDetail,
	FailureDisposition,
	ToolExecutionError,
)
from agent.tools.models import ToolCapability

from agent.tools.url_matcher import filter_by_url

logger = logging.getLogger(__name__)


class ToolStats:
	"""工具使用统计"""

	def __init__(self):
		self.call_count: int = 0
		self.success_count: int = 0
		self.failure_count: int = 0
		self.total_duration: float = 0.0
		self.last_called: Optional[float] = None

	@property
	def average_duration(self) -> float:
		"""平均执行时间（秒）"""
		return self.total_duration / self.call_count if self.call_count > 0 else 0.0

	@property
	def success_rate(self) -> float:
		"""成功率"""
		return self.success_count / self.call_count if self.call_count > 0 else 0.0

	def to_dict(self) -> Dict[str, Any]:
		return {
			"call_count": self.call_count,
			"success_count": self.success_count,
			"failure_count": self.failure_count,
			"average_duration": self.average_duration,
			"success_rate": self.success_rate,
			"last_called": self.last_called
		}


def _record_tool_outcome(stats: ToolStats, start_time: float, *, success: bool) -> None:
	if success:
		stats.success_count += 1
	else:
		stats.failure_count += 1
	stats.total_duration += time.time() - start_time


def _failure_result(failure: FailureDetail) -> Dict[str, Any]:
	payload = {"failure": failure.to_dict()}
	return {
		"text": json.dumps(payload, ensure_ascii=False, indent=2),
		"_control": {"failure": failure.to_dict()},
	}


class ToolRegistry:
	"""工具注册表"""
	_REGISTRATION_FIELDS = (
		"schema",
		"url_patterns",
		"hidden_params",
		"inject_params",
		"groups",
		"capabilities",
		"ui_metadata",
	)

	def __init__(self):
		self._tools: Dict[str, Dict[str, Any]] = {}
		self._stats: Dict[str, ToolStats] = {}

	def register(
		self,
		func: Callable,
		schema: Optional[Dict] = None,
		requires_confirmation: bool = False,
		editable_params: Optional[List[str]] = None,
		display: Optional[Dict[str, Dict[str, Any]]] = None,
		url_patterns: Optional[List[str]] = None,
		hidden_params: Optional[List[str]] = None,
		inject_params: Optional[Dict[str, Any]] = None,
		groups: Optional[List[str]] = None,
		capabilities: Optional[Sequence[ToolCapability | str]] = None,
	):
		"""
		注册工具

		Args:
			func: 工具函数
			schema: 工具 Schema（如果为 None，自动生成）
			requires_confirmation: 是否需要用户确认
			editable_params: 允许用户编辑的参数列表
			display: i18n 展示信息 {"zh": {"name": "xx", "params": {...}}, "en": {...}}
			url_patterns: URL 模式列表（如 ["*.google.com"]，None = 无限制）
			hidden_params: 隐藏参数列表（不生成 schema，LLM 看不到）
			inject_params: 注入参数规则 {param_name: injector_func}
			groups: 工具分组标签（如 ["core"]、["browser"]），用于 RunConfig.tool_groups 过滤
			capabilities: 安全副作用标签；explicit_validation 工具必须显式声明
		"""
		func_name = func.__name__

		# 如果没有提供 Schema，自动生成
		if schema is None:
			try:
				from agent.tools.schema import SchemaGenerator
				schema = SchemaGenerator.generate(func, hidden_params=hidden_params or [])
			except Exception as e:
				logger.error(f"Failed to generate schema for {func_name}: {e}")
				raise

		registration = {
			"func": func,
			"schema": schema,
			"url_patterns": url_patterns,
			"hidden_params": hidden_params or [],
			"inject_params": inject_params or {},
			"groups": groups or [],
			"capabilities": frozenset(ToolCapability(item) for item in capabilities or []),
			"ui_metadata": {
				"requires_confirmation": requires_confirmation,
				"editable_params": editable_params or [],
				"display": display or {}
			}
		}

		existing = self._tools.get(func_name)
		if existing is not None:
			same_func = existing["func"] is func
			same_definition = self._callable_origin(existing["func"]) == self._callable_origin(func)
			same_metadata = all(
				existing.get(field) == registration.get(field)
				for field in self._REGISTRATION_FIELDS
			)
			if same_func and same_metadata:
				# 重复 import 或装配代码可能对同一函数重复注册。这是安全的
				# 幂等操作，且不应重置已累计的调用统计。
				logger.debug("[ToolRegistry] Already registered (idempotent): %s", func_name)
				return
			if same_definition and same_metadata:
				# Host composition can recreate a closure from the same factory
				# definition to bind a fresh adapter/gateway. Replace only the
				# callable and preserve accumulated statistics.
				existing["func"] = func
				logger.debug(
					"[ToolRegistry] Rebound same-definition tool: %s (%s)",
					func_name,
					self._describe_callable(func),
				)
				return
			if self._is_host_layer_override(existing, registration):
				# A Host may deliberately replace a framework builtin with a
				# capability-equivalent adapter. This preserves the historical
				# layering contract while still rejecting builtin/builtin clashes,
				# capability escalation, and competing Host implementations.
				self._tools[func_name] = registration
				logger.warning(
					"[ToolRegistry] Host replaced builtin tool: %s (%s -> %s)",
					func_name,
					self._describe_callable(existing["func"]),
					self._describe_callable(func),
				)
				return

			existing_source = self._describe_callable(existing["func"])
			attempted_source = self._describe_callable(func)
			if same_func:
				changed_fields = [
					field
					for field in self._REGISTRATION_FIELDS
					if existing.get(field) != registration.get(field)
				]
				detail = f"registration metadata differs: {', '.join(changed_fields)}"
			else:
				detail = "a different callable already owns this name"
			raise ValueError(
				f"Duplicate tool registration for '{func_name}': {detail}; "
				f"existing={existing_source}, attempted={attempted_source}"
			)

		self._tools[func_name] = registration

		# 初始化统计
		self._stats[func_name] = ToolStats()

		logger.debug(
			f"[ToolRegistry] Registered: {func_name} "
			f"(requires_confirmation={requires_confirmation}, "
			f"editable_params={editable_params or []}, "
			f"url_patterns={url_patterns or 'none'})"
		)

	@staticmethod
	def _describe_callable(func: Callable) -> str:
		"""返回可用于定位重复注册来源的稳定描述。"""
		module = getattr(func, "__module__", type(func).__module__)
		qualname = getattr(func, "__qualname__", getattr(func, "__name__", type(func).__qualname__))
		code = getattr(func, "__code__", None)
		if code is None:
			return f"{module}.{qualname}"
		return f"{module}.{qualname} ({code.co_filename}:{code.co_firstlineno})"

	@staticmethod
	def _callable_origin(func: Callable) -> tuple[str, str, Optional[str], Optional[int]]:
		"""Identify one source definition while allowing fresh closure instances."""
		module = getattr(func, "__module__", type(func).__module__)
		qualname = getattr(
			func,
			"__qualname__",
			getattr(func, "__name__", type(func).__qualname__),
		)
		code = getattr(func, "__code__", None)
		if code is None:
			return module, qualname, None, None
		return module, qualname, code.co_filename, code.co_firstlineno

	@staticmethod
	def _is_host_layer_override(
		existing: Dict[str, Any],
		registration: Dict[str, Any],
	) -> bool:
		existing_module = getattr(existing["func"], "__module__", "")
		attempted_module = getattr(registration["func"], "__module__", "")
		existing_capabilities = existing.get("capabilities", frozenset())
		return (
			existing_module.startswith(("agent.tools.builtin.", "agent.trace."))
			and not attempted_module.startswith("agent.")
			and bool(existing_capabilities)
			and registration.get("capabilities") == existing_capabilities
		)

	@staticmethod
	def _resolve_key_path(context: Dict[str, Any], key_path: str) -> Any:
		"""
		从 context 中按路径取值。

		支持 "obj.field" 格式：第一段从 context dict 取值，后续段用 getattr。
		例如 "knowledge_config.default_tags" → context["knowledge_config"].default_tags

		Args:
			context: 上下文字典
			key_path: 取值路径

		Returns:
			取到的值，路径无效返回 None
		"""
		parts = key_path.split(".")
		value = context.get(parts[0])
		for part in parts[1:]:
			if value is None:
				return None
			value = getattr(value, part, None)
		return value

	def is_registered(self, tool_name: str) -> bool:
		"""检查工具是否已注册"""
		return tool_name in self._tools

	def get_schemas(self, tool_names: Optional[List[str]] = None) -> List[Dict]:
		"""
		获取工具 Schema

		Args:
			tool_names: 工具名称列表（None = 所有工具）

		Returns:
			OpenAI Tool Schema 列表
		"""
		if tool_names is None:
			tool_names = list(self._tools.keys())

		schemas = []
		for name in tool_names:
			if name in self._tools:
				schemas.append(self._tools[name]["schema"])
			else:
				logger.warning(f"[ToolRegistry] Tool not found: {name}")

		return schemas

	def get_tool_names(self, current_url: Optional[str] = None, groups: Optional[List[str]] = None) -> List[str]:
		"""
		获取工具名称列表（可选 URL 过滤 + group 过滤）

		Args:
			current_url: 当前 URL（None = 不过滤 URL）
			groups: 工具分组白名单（None = 不过滤 group，返回所有工具）

		Returns:
			工具名称列表
		"""
		# 1. group 过滤
		if groups is not None:
			group_set = set(groups)
			candidates = {
				name for name, tool in self._tools.items()
				if group_set & set(tool.get("groups", []))
			}
		else:
			candidates = set(self._tools.keys())

		# 2. URL 过滤
		if current_url is None:
			return list(candidates)

		tool_items = [
			{"name": name, "url_patterns": self._tools[name]["url_patterns"]}
			for name in candidates
		]
		filtered = filter_by_url(tool_items, current_url, url_field="url_patterns")
		return [item["name"] for item in filtered]

	def get_available_groups(self) -> List[str]:
		"""获取所有已注册的工具分组"""
		groups = set()
		for tool in self._tools.values():
			groups.update(tool.get("groups", []))
		return sorted(groups)

	def get_capabilities(self, tool_name: str) -> frozenset[ToolCapability]:
		"""Return declared effects; an empty set means unclassified."""
		tool = self._tools.get(tool_name)
		if not tool:
			return frozenset()
		return frozenset(tool.get("capabilities", ()))

	def get_schemas_for_url(self, current_url: Optional[str] = None) -> List[Dict]:
		"""
		根据当前 URL 获取匹配的工具 Schema

		Args:
			current_url: 当前 URL（None = 返回无 URL 限制的工具）

		Returns:
			过滤后的工具 Schema 列表
		"""
		tool_names = self.get_tool_names(current_url)
		return self.get_schemas(tool_names)

	async def execute(
		self,
		name: str,
		arguments: Dict[str, Any],
		uid: str = "",
		context: Optional[Dict[str, Any]] = None,
		sensitive_data: Optional[Dict[str, Any]] = None,
		inject_values: Optional[Dict[str, Any]] = None
	) -> str:
		"""
		执行工具调用

		Args:
			name: 工具名称
			arguments: 工具参数
			uid: 用户ID（自动注入）
			context: 额外上下文
			sensitive_data: 敏感数据字典（用于替换 <secret> 占位符）
			inject_values: 兼容注入值；仅在调用参数未提供时生效

		Returns:
			JSON 字符串格式的结果
		"""
		if name not in self._tools:
			failure = FailureDetail(
				code="UNKNOWN_TOOL",
				message=f"Unknown tool: {name}",
				disposition=FailureDisposition.ABORT_RUN,
				source_tool=name,
			)
			logger.error("[ToolRegistry] %s", failure.message)
			return _failure_result(failure)

		start_time = time.time()
		stats = self._stats[name]
		stats.call_count += 1
		stats.last_called = start_time

		try:
			func = self._tools[name]["func"]
			tool_info = self._tools[name]

			# 处理敏感数据占位符
			if sensitive_data:
				from agent.tools.sensitive import replace_sensitive_data
				current_url = context.get("page_url") if context else None
				arguments = replace_sensitive_data(arguments, sensitive_data, current_url)

			# 准备参数：只注入函数需要的参数
			sig = inspect.signature(func)
			# 过滤掉函数签名中不存在的参数（如 Claude SDK 发送的 {"_": true} 占位符）
			valid_params = set(sig.parameters.keys())
			kwargs = {k: v for k, v in arguments.items() if k in valid_params}

			# 注入隐藏参数（hidden_params）
			hidden_params = tool_info.get("hidden_params", [])
			if "uid" in hidden_params and "uid" in sig.parameters:
				kwargs["uid"] = uid
			if "context" in hidden_params and "context" in sig.parameters:
				kwargs["context"] = context

			# 旧调用方可以在执行时提供默认注入值。模型明确给出的
			# 参数优先，且仍只允许函数签名中存在的字段。
			for param_name, value in (inject_values or {}).items():
				if param_name in sig.parameters and param_name not in kwargs:
					kwargs[param_name] = value

			# 注入参数（inject_params）
			inject_params = tool_info.get("inject_params", {})
			for param_name, rule in inject_params.items():
				if param_name not in sig.parameters:
					continue

				if not isinstance(rule, dict) or "mode" not in rule:
					# 兼容旧格式：直接值或 callable
					if param_name not in kwargs or kwargs[param_name] is None:
						kwargs[param_name] = rule() if callable(rule) else rule
					continue

				mode = rule["mode"]
				key_path = rule.get("key")

				# 从 context 中按路径取值
				value = self._resolve_key_path(context, key_path) if key_path and context else None

				if value is None:
					continue

				if mode == "default":
					# 默认值模式：LLM 未提供则注入
					if param_name not in kwargs or kwargs[param_name] is None:
						kwargs[param_name] = value
				elif mode == "merge":
					# 合并模式：框架值始终保留，LLM 可追加新内容
					llm_value = kwargs.get(param_name)
					if isinstance(value, dict):
						# dict: LLM 追加新 key，同名 key 以框架值为准
						kwargs[param_name] = {**(llm_value or {}), **value}
					elif isinstance(value, list):
						# list: 合并去重
						kwargs[param_name] = list(set((llm_value or []) + value))
					else:
						kwargs[param_name] = value

			# 执行函数
			if inspect.iscoroutinefunction(func):
				result = await func(**kwargs)
			else:
				result = func(**kwargs)

			# 返回结果：ToolResult 转为可序列化格式
			if isinstance(result, str):
				_record_tool_outcome(stats, start_time, success=True)
				return result

			# 处理 ToolResult 对象
			from agent.tools.models import ToolResult
			if isinstance(result, ToolResult):
				failure = result.failure
				if failure is None and result.error:
					failure = FailureDetail(
						code="TOOL_REPORTED_ERROR",
						message=result.error,
						disposition=FailureDisposition.RETRY_CALL,
						source_tool=name,
					)
					result.failure = failure
				ret = {"text": result.to_llm_message()}
				if failure is not None:
					ret["_control"] = {"failure": failure.to_dict()}

				# Runner 消费的控制字段与业务文本分离，不能靠解析 output 猜测终止。
				if result.terminate_run:
					ret.setdefault("_control", {}).update({
						"terminate_run": True,
						"result_summary": result.result_summary or result.long_term_memory or result.output,
					})

				# 保留images
				if result.images:
					ret["images"] = result.images

				# 保留tool_usage
				if result.tool_usage:
					ret["tool_usage"] = result.tool_usage

				_record_tool_outcome(stats, start_time, success=failure is None)

				# 向后兼容：只有 text 时返回字符串；有控制信息时必须保留 dict。
				if len(ret) == 1:
					return ret["text"]
				return ret

			_record_tool_outcome(stats, start_time, success=True)
			return json.dumps(result, ensure_ascii=False, indent=2)

		except ToolExecutionError as exc:
			_record_tool_outcome(stats, start_time, success=False)
			failure = exc.failure
			if failure.source_tool is None:
				failure = FailureDetail(
					code=failure.code,
					message=failure.message,
					disposition=failure.disposition,
					source_tool=name,
					details=failure.details,
				)
			logger.warning(
				"[ToolRegistry] expected rejection tool=%s code=%s disposition=%s: %s",
				name,
				failure.code,
				failure.disposition.value,
				failure.message,
			)
			return _failure_result(failure)
		except Exception:
			_record_tool_outcome(stats, start_time, success=False)
			logger.exception("[ToolRegistry] unexpected failure executing tool '%s'", name)
			return _failure_result(
				FailureDetail(
					code="UNEXPECTED_TOOL_ERROR",
					message="The tool failed unexpectedly",
					disposition=FailureDisposition.ABORT_RUN,
					source_tool=name,
				)
			)

	def get_stats(self, tool_name: Optional[str] = None) -> Dict[str, Dict[str, Any]]:
		"""
		获取工具统计信息

		Args:
			tool_name: 工具名称（None = 所有工具）

		Returns:
			统计信息字典
		"""
		if tool_name:
			if tool_name in self._stats:
				return {tool_name: self._stats[tool_name].to_dict()}
			return {}

		return {name: stats.to_dict() for name, stats in self._stats.items()}

	def get_top_tools(self, limit: int = 10, by: str = "call_count") -> List[str]:
		"""
		获取排名靠前的工具

		Args:
			limit: 返回数量
			by: 排序依据（call_count, success_rate, average_duration）

		Returns:
			工具名称列表
		"""
		if by == "call_count":
			sorted_tools = sorted(
				self._stats.items(),
				key=lambda x: x[1].call_count,
				reverse=True
			)
		elif by == "success_rate":
			sorted_tools = sorted(
				self._stats.items(),
				key=lambda x: x[1].success_rate,
				reverse=True
			)
		elif by == "average_duration":
			sorted_tools = sorted(
				self._stats.items(),
				key=lambda x: x[1].average_duration,
				reverse=False  # 越快越好
			)
		else:
			raise ValueError(f"Invalid sort by: {by}")

		return [name for name, _ in sorted_tools[:limit]]

	def check_confirmation_required(self, tool_calls: List[Dict]) -> bool:
		"""检查是否有工具需要用户确认"""
		for tc in tool_calls:
			tool_name = tc.get("function", {}).get("name")
			if tool_name and tool_name in self._tools:
				if self._tools[tool_name]["ui_metadata"].get("requires_confirmation", False):
					return True
		return False

	def get_confirmation_flags(self, tool_calls: List[Dict]) -> List[bool]:
		"""返回每个工具是否需要确认"""
		flags = []
		for tc in tool_calls:
			tool_name = tc.get("function", {}).get("name")
			if tool_name and tool_name in self._tools:
				flags.append(self._tools[tool_name]["ui_metadata"].get("requires_confirmation", False))
			else:
				flags.append(False)
		return flags

	def check_any_param_editable(self, tool_calls: List[Dict]) -> bool:
		"""检查是否有任何工具允许参数编辑"""
		for tc in tool_calls:
			tool_name = tc.get("function", {}).get("name")
			if tool_name and tool_name in self._tools:
				editable_params = self._tools[tool_name]["ui_metadata"].get("editable_params", [])
				if editable_params:
					return True
		return False

	def get_editable_params_map(self, tool_calls: List[Dict]) -> Dict[str, List[str]]:
		"""返回每个工具调用的可编辑参数列表"""
		params_map = {}
		for tc in tool_calls:
			tool_call_id = tc.get("id")
			tool_name = tc.get("function", {}).get("name")

			if tool_name and tool_name in self._tools:
				editable_params = self._tools[tool_name]["ui_metadata"].get("editable_params", [])
				params_map[tool_call_id] = editable_params
			else:
				params_map[tool_call_id] = []

		return params_map

	def get_ui_metadata(
		self,
		locale: str = "zh",
		tool_names: Optional[List[str]] = None
	) -> Dict[str, Dict[str, Any]]:
		"""
		获取工具的UI元数据（用于前端展示）

		Returns:
			{
				"tool_name": {
					"display_name": "搜索笔记",
					"param_display_names": {"query": "搜索关键词"},
					"requires_confirmation": false,
					"editable_params": ["query"]
				}
			}
		"""
		if tool_names is None:
			tool_names = list(self._tools.keys())

		metadata = {}
		for name in tool_names:
			if name not in self._tools:
				continue

			ui_meta = self._tools[name]["ui_metadata"]
			display = ui_meta.get("display", {}).get(locale, {})

			metadata[name] = {
				"display_name": display.get("name", name),
				"param_display_names": display.get("params", {}),
				"requires_confirmation": ui_meta.get("requires_confirmation", False),
				"editable_params": ui_meta.get("editable_params", [])
			}

		return metadata


# 全局单例
_global_registry = ToolRegistry()


def tool(
	description: Optional[str] = None,
	param_descriptions: Optional[Dict[str, str]] = None,
	requires_confirmation: bool = False,
	editable_params: Optional[List[str]] = None,
	display: Optional[Dict[str, Dict[str, Any]]] = None,
	url_patterns: Optional[List[str]] = None,
	hidden_params: Optional[List[str]] = None,
	inject_params: Optional[Dict[str, Any]] = None,
	groups: Optional[List[str]] = None,
	capabilities: Optional[Sequence[ToolCapability | str]] = None,
):
	"""
	工具装饰器 - 自动注册工具并生成 Schema

	Args:
		description: 函数描述（可选，从 docstring 提取）
		param_descriptions: 参数描述（可选，从 docstring 提取）
		requires_confirmation: 是否需要用户确认（默认 False）
		editable_params: 允许用户编辑的参数列表
		display: i18n 展示信息
		url_patterns: URL 模式列表（如 ["*.google.com"]，None = 无限制）
		hidden_params: 隐藏参数列表（不生成 schema，LLM 看不到）
		inject_params: 注入参数规则 {param_name: injector_func}
		groups: 工具分组标签（如 ["core"]、["browser"]），用于 RunConfig.tool_groups 过滤
		capabilities: 安全副作用标签（read/write/agent_spawn/external_send 等）

	Example:
		@tool(
			hidden_params=["context", "uid"],
			inject_params={
				"owner": lambda ctx: ctx.config.knowledge.get_owner(),
			},
			editable_params=["query"],
			url_patterns=["*.google.com"],
			display={
				"zh": {"name": "搜索笔记", "params": {"query": "搜索关键词"}},
				"en": {"name": "Search Notes", "params": {"query": "Query"}}
			}
		)
		async def search_blocks(
			query: str,
			limit: int = 10,
			owner: Optional[str] = None,
			context: Optional[ToolContext] = None,
			uid: str = ""
		) -> str:
			'''搜索用户的笔记块'''
			...
	"""
	def decorator(func: Callable) -> Callable:
		schema = None
		if description is not None or param_descriptions:
			from agent.tools.schema import SchemaGenerator

			schema = SchemaGenerator.generate(
				func,
				hidden_params=hidden_params or [],
			)
			function_schema = schema["function"]
			if description is not None:
				function_schema["description"] = description
			properties = function_schema["parameters"].get("properties", {})
			for param_name, param_description in (param_descriptions or {}).items():
				if param_name in properties:
					properties[param_name]["description"] = param_description

		# 注册到全局 registry
		_global_registry.register(
			func,
			schema=schema,
			requires_confirmation=requires_confirmation,
			editable_params=editable_params,
			display=display,
			url_patterns=url_patterns,
			hidden_params=hidden_params,
			inject_params=inject_params,
			groups=groups,
			capabilities=capabilities,
		)
		return func

	return decorator


def get_tool_registry() -> ToolRegistry:
	"""获取全局工具注册表"""
	return _global_registry