"""ASV benchmarks for tool runtime registration, lookup, and execution. Measures the performance of: - ToolSpec construction - ToolRegistry register / get / list_tools - ToolRunner execute (happy path) - ToolRunner execute (error normalisation) """ from __future__ import annotations import sys from pathlib import Path from typing import Any try: from cleveragents.tool.registry import ToolRegistry from cleveragents.tool.runner import ToolRunner from cleveragents.tool.runtime import ToolSpec except ModuleNotFoundError: sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "src")) from cleveragents.tool.registry import ToolRegistry from cleveragents.tool.runner import ToolRunner from cleveragents.tool.runtime import ToolSpec def _echo(inputs: dict[str, Any]) -> dict[str, Any]: return dict(inputs) def _fail(inputs: dict[str, Any]) -> dict[str, Any]: raise RuntimeError("boom") class ToolSpecConstructionSuite: """Benchmark ToolSpec model construction.""" def time_toolspec_construction(self) -> None: ToolSpec( name="bench/spec", description="Bench spec", handler=_echo, ) def time_toolspec_with_schemas(self) -> None: ToolSpec( name="bench/schema", description="Bench schema", input_schema={"type": "object", "properties": {"x": {"type": "integer"}}}, output_schema={"type": "object", "properties": {"y": {"type": "integer"}}}, handler=_echo, ) class ToolRegistrySuite: """Benchmark ToolRegistry operations.""" def setup(self) -> None: self.registry = ToolRegistry() for i in range(100): spec = ToolSpec( name=f"bench/tool-{i}", description=f"Bench tool {i}", handler=_echo, ) self.registry.register(spec) def time_registry_get(self) -> None: self.registry.get("bench/tool-50") def time_registry_list_all(self) -> None: self.registry.list_tools() def time_registry_list_namespace(self) -> None: self.registry.list_tools(namespace="bench") class ToolRunnerSuite: """Benchmark ToolRunner execute.""" def setup(self) -> None: self.registry = ToolRegistry() self.registry.register( ToolSpec(name="bench/echo", description="Echo", handler=_echo) ) self.registry.register( ToolSpec(name="bench/fail", description="Fail", handler=_fail) ) self.runner = ToolRunner(self.registry) def time_execute_success(self) -> None: self.runner.execute("bench/echo", {"key": "value"}) def time_execute_error(self) -> None: self.runner.execute("bench/fail", {})