cleveragents-core/benchmarks/skill_cli_bench.py

"""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"])