cleveragents-core/benchmarks/skill_resolution_bench.py

"""ASV benchmarks for Skill domain model and resolution.

Measures the performance of:
- Skill model construction (Pydantic validation)
- Skill.as_cli_dict() serialization
- Skill.from_config() factory
- SkillResolver.resolve_tools() flattening
- SkillResolver.compute_capability_summary() aggregation
"""

from __future__ import annotations

import sys
from pathlib import Path

try:
    from cleveragents.domain.models.core.skill import (
        Skill,
        SkillAgentSource,
        SkillInclude,
        SkillInlineTool,
        SkillMcpSource,
        SkillResolver,
    )
    from cleveragents.domain.models.core.tool import (
        ToolCapability,
        ToolSource,
    )
except ModuleNotFoundError:
    sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "src"))
    from cleveragents.domain.models.core.skill import (
        Skill,
        SkillAgentSource,
        SkillInclude,
        SkillInlineTool,
        SkillMcpSource,
        SkillResolver,
    )
    from cleveragents.domain.models.core.tool import (
        ToolCapability,
        ToolSource,
    )


def _make_skill() -> Skill:
    """Create a fully-populated Skill for benchmarking."""
    return Skill(
        name="bench/full-skill",
        description="Benchmark skill",
        tool_refs=["bench/tool-a", "bench/tool-b", "bench/tool-c"],
        includes=[SkillInclude(name="bench/base-skill")],
        anonymous_tools=[
            SkillInlineTool(
                description="Inline bench tool",
                source=ToolSource.CUSTOM,
                code="return True",
                capability=ToolCapability(writes=True, side_effects=["fs"]),
            ),
        ],
        mcp_servers=[
            SkillMcpSource(
                server="npx @mcp/bench-server",
                tools=["read", "write"],
                env={"KEY": "val"},
            ),
        ],
        agent_skills=[SkillAgentSource(path="./skills/bench")],
        overrides={"bench/tool-a": {"timeout": 600}},
    )


def _make_registry() -> dict[str, Skill]:
    """Create a skill registry for resolution benchmarks."""
    base = Skill(
        name="bench/base-skill",
        description="Base skill",
        tool_refs=["bench/base-tool-1", "bench/base-tool-2"],
    )
    return {"bench/base-skill": base}


class SkillConstructionSuite:
    """Benchmark Skill model construction (Pydantic validation)."""

    def time_skill_construction(self) -> None:
        """Benchmark creating a fully-populated Skill object."""
        _make_skill()

    def time_skill_minimal_construction(self) -> None:
        """Benchmark creating a minimal Skill object."""
        Skill(
            name="bench/minimal",
            description="Minimal skill",
        )

    def time_inline_tool_construction(self) -> None:
        """Benchmark SkillInlineTool construction."""
        SkillInlineTool(
            description="Bench inline",
            source=ToolSource.CUSTOM,
            code="return True",
        )

    def time_mcp_source_construction(self) -> None:
        """Benchmark SkillMcpSource construction."""
        SkillMcpSource(
            server="npx @mcp/bench",
            tools=["read", "write"],
        )


class SkillSerializationSuite:
    """Benchmark Skill serialization methods."""

    def setup(self) -> None:
        """Create objects for serialization benchmarks."""
        self.skill = _make_skill()

    def time_skill_as_cli_dict(self) -> None:
        """Benchmark Skill.as_cli_dict() ordered dict generation."""
        self.skill.as_cli_dict()

    def time_skill_model_dump(self) -> None:
        """Benchmark Pydantic model_dump() serialization."""
        self.skill.model_dump()

    def time_skill_model_dump_json(self) -> None:
        """Benchmark Pydantic model_dump_json() JSON serialization."""
        self.skill.model_dump_json()


class SkillFromConfigSuite:
    """Benchmark Skill.from_config() factory."""

    def setup(self) -> None:
        """Prepare config dict for benchmarks."""
        self.config = {
            "name": "bench/from-config",
            "description": "Config benchmark",
            "tools": ["bench/tool-a", "bench/tool-b"],
            "includes": [{"name": "bench/base-skill"}],
            "anonymous_tools": [
                {
                    "description": "Inline",
                    "source": "custom",
                    "code": "return True",
                },
            ],
            "mcp_servers": [
                {"server": "npx @mcp/bench", "tools": ["read"]},
            ],
            "agent_skills": [{"path": "./skills/bench"}],
        }

    def time_skill_from_config(self) -> None:
        """Benchmark Skill.from_config()."""
        Skill.from_config(self.config)


class SkillResolutionSuite:
    """Benchmark SkillResolver.resolve_tools()."""

    def setup(self) -> None:
        """Create skill and registry for benchmarks."""
        self.skill = _make_skill()
        self.registry = _make_registry()
        self.registry[self.skill.name] = self.skill
        self.resolver = SkillResolver()

    def time_resolve_tools(self) -> None:
        """Benchmark resolve_tools() with one include level."""
        self.resolver.resolve_tools(self.skill, self.registry)

    def time_compute_capability_summary(self) -> None:
        """Benchmark compute_capability_summary()."""
        resolved = self.resolver.resolve_tools(self.skill, self.registry)
        self.resolver.compute_capability_summary(self.skill, resolved)