"""ASV benchmarks for Skill CLI command throughput. Measures the performance of: - Config-only add (YAML load + validate + Skill registration) - List command rendering - Show command rendering - Tools command (resolver flattening) """ from __future__ import annotations import importlib import os import sys import tempfile from datetime import datetime from pathlib import Path # Ensure the local *source* tree is importable even when ASV has an # older build of the package installed. _SRC = str(Path(__file__).resolve().parents[1] / "src") if _SRC not in sys.path: sys.path.insert(0, _SRC) import cleveragents # noqa: E402 importlib.reload(cleveragents) from typer.testing import CliRunner # noqa: E402 from cleveragents.application.services.skill_service import SkillService # noqa: E402 from cleveragents.cli.commands.skill import ( # noqa: E402 _reset_skill_service, ) from cleveragents.cli.commands.skill import ( # noqa: E402 app as skill_app, ) from cleveragents.domain.models.core.skill import Skill # noqa: E402 _VALID_YAML = """\ name: local/bench-skill description: "Benchmark skill" tools: - name: builtin/read_file - name: builtin/write_file - name: builtin/list_directory """ _runner = CliRunner() def _fresh_service() -> SkillService: """Create and install a fresh service for benchmarking.""" svc = SkillService() _reset_skill_service(svc) return svc class SkillCLIAddSuite: """Benchmark skill add --config throughput.""" def setup(self) -> None: """Write a temporary YAML file and set up fresh service.""" fd, self._path = tempfile.mkstemp(suffix=".yaml") with os.fdopen(fd, "w") as fh: fh.write(_VALID_YAML) def teardown(self) -> None: """Clean up.""" Path(self._path).unlink(missing_ok=True) def time_add_from_config(self) -> None: """Benchmark add --config end-to-end.""" _fresh_service() _runner.invoke(skill_app, ["add", "--config", self._path]) class SkillCLIListSuite: """Benchmark skill list throughput.""" def setup(self) -> None: """Set up service with skills.""" svc = _fresh_service() for i in range(50): name = f"local/skill-{i}" skill = Skill( name=name, description=f"Benchmark skill {i}", tool_refs=["builtin/read_file", "builtin/write_file"], ) svc._skills[name] = skill svc._created_at[name] = datetime.now() svc._updated_at[name] = datetime.now() def teardown(self) -> None: _fresh_service() def time_list_all(self) -> None: """Benchmark listing all skills.""" _runner.invoke(skill_app, ["list"]) def time_list_with_namespace(self) -> None: """Benchmark listing with namespace filter.""" _runner.invoke(skill_app, ["list", "--namespace", "local"]) class SkillCLIShowSuite: """Benchmark skill show throughput.""" def setup(self) -> None: svc = _fresh_service() skill = Skill( name="local/bench-show", description="Benchmark skill for show", tool_refs=[ "builtin/read_file", "builtin/write_file", "builtin/list_directory", ], ) svc._skills["local/bench-show"] = skill svc._created_at["local/bench-show"] = datetime.now() svc._updated_at["local/bench-show"] = datetime.now() def teardown(self) -> None: _fresh_service() def time_show(self) -> None: """Benchmark showing a skill.""" _runner.invoke(skill_app, ["show", "local/bench-show"]) class SkillCLIToolsSuite: """Benchmark skill tools (resolver) throughput.""" def setup(self) -> None: svc = _fresh_service() skill = Skill( name="local/bench-tools", description="Benchmark skill for tools resolution", tool_refs=[ "builtin/read_file", "builtin/write_file", "builtin/list_directory", "builtin/search_files", "builtin/delete_file", ], ) svc._skills["local/bench-tools"] = skill svc._created_at["local/bench-tools"] = datetime.now() svc._updated_at["local/bench-tools"] = datetime.now() def teardown(self) -> None: _fresh_service() def time_tools(self) -> None: """Benchmark resolving tools for a skill.""" _runner.invoke(skill_app, ["tools", "local/bench-tools"]) class SkillCLIRefreshSuite: """Benchmark skill refresh command throughput.""" def setup(self) -> None: """Set up service with multiple skills.""" svc = _fresh_service() for i in range(20): name = f"local/refresh-skill-{i}" skill = Skill( name=name, description=f"Refresh benchmark skill {i}", tool_refs=[ "builtin/read_file", "builtin/write_file", "builtin/list_directory", ], ) svc._skills[name] = skill svc._created_at[name] = datetime.now() svc._updated_at[name] = datetime.now() def teardown(self) -> None: _fresh_service() def time_refresh_single(self) -> None: """Benchmark refreshing a single skill.""" _runner.invoke(skill_app, ["refresh", "local/refresh-skill-0"]) def time_refresh_all(self) -> None: """Benchmark refreshing all skills.""" _runner.invoke(skill_app, ["refresh", "--all"])