cleveragents-core/benchmarks/agent_skills_loader_bench.py

"""ASV benchmarks for Agent Skills Standard loader parsing throughput.

Measures the performance of:
- SKILL.md string parsing (frontmatter + body)
- Agent skills folder loading (from disk)
- Progressive disclosure lifecycle (discover → activate → deactivate)
- Tool descriptor conversion
- Resource listing from multi-file folders
"""

from __future__ import annotations

import importlib
import sys
import tempfile
from pathlib import Path

_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 cleveragents.skills.agent_skills_loader import (  # noqa: E402
    AgentSkillLoader,
    AgentSkillSpec,
)

_MINIMAL_SKILL_MD = """\
---
name: local/bench-minimal
description: Minimal benchmark skill.
---
Step 1: Do this.
"""

_FULL_SKILL_MD = """\
---
name: local/bench-full
description: Full benchmark skill with all optional fields.
version: 2.1.0
steps:
  - Verify the environment is ready
  - Check that all required tools are available
  - Read the configuration from the reference files
  - Execute the main workflow script
  - Monitor progress and capture output
  - Handle any errors gracefully and log them
  - Release any acquired resources
  - Write the result summary to the output
  - Notify downstream systems if required
compatibility:
  min_agent_version: "3.0.0"
metadata:
  author: benchmark
  tags: [bench, performance]
  priority: high
allowed-tools:
  - local/read-file
  - local/shell
  - local/git-status
  - local/search-files
---
# Full Benchmark Skill

This skill is used to benchmark the Agent Skills loader performance.

## Phase 1 — Preparation

1. Verify the environment is ready.
2. Check that all required tools are available.
3. Read the configuration from the reference files.

## Phase 2 — Execution

4. Execute the main workflow script.
5. Monitor progress and capture output.
6. Handle any errors gracefully and log them.

## Phase 3 — Cleanup

7. Release any acquired resources.
8. Write the result summary to the output.
9. Notify downstream systems if required.
"""


class AgentSkillSpecParsingSuite:
    """Benchmark AgentSkillSpec string parsing throughput."""

    def time_parse_minimal(self) -> None:
        """Benchmark parsing a minimal SKILL.md string."""
        AgentSkillSpec.from_string(_MINIMAL_SKILL_MD)

    def time_parse_full(self) -> None:
        """Benchmark parsing a fully-populated SKILL.md string."""
        AgentSkillSpec.from_string(_FULL_SKILL_MD)

    def time_parse_structured_steps(self) -> None:
        """Benchmark parsing structured steps from frontmatter."""
        spec = AgentSkillSpec.from_string(_FULL_SKILL_MD)
        _ = spec.steps


class AgentSkillFolderLoadSuite:
    """Benchmark AgentSkillLoader.from_folder() throughput."""

    def setup(self) -> None:
        """Create a temporary skill folder for folder-load benchmarks."""
        self._tmp = Path(tempfile.mkdtemp(prefix="asv_agent_skill_"))

        # Write SKILL.md
        (self._tmp / "SKILL.md").write_text(_FULL_SKILL_MD)

        # Write scripts/
        scripts = self._tmp / "scripts"
        scripts.mkdir()
        for i in range(5):
            (scripts / f"script_{i}.py").write_text(f"# Script {i}\nprint({i})\n")

        # Write references/
        refs = self._tmp / "references"
        refs.mkdir()
        for i in range(3):
            (refs / f"ref_{i}.md").write_text(f"# Reference {i}\nContent {i}.\n")

        # Write assets/
        assets = self._tmp / "assets"
        assets.mkdir()
        (assets / "config.json").write_text('{"key": "value"}\n')

    def teardown(self) -> None:
        """Remove the temporary skill folder."""
        import shutil

        shutil.rmtree(self._tmp, ignore_errors=True)

    def time_load_from_folder(self) -> None:
        """Benchmark loading a fully-populated agent skills folder."""
        AgentSkillLoader.from_folder(self._tmp)


class AgentSkillLifecycleSuite:
    """Benchmark the full progressive disclosure lifecycle."""

    def setup(self) -> None:
        """Create a temporary skill folder for lifecycle benchmarks."""
        self._tmp = Path(tempfile.mkdtemp(prefix="asv_agent_skill_lc_"))
        (self._tmp / "SKILL.md").write_text(_FULL_SKILL_MD)
        scripts = self._tmp / "scripts"
        scripts.mkdir()
        (scripts / "run.py").write_text("print('run')\n")

    def teardown(self) -> None:
        """Remove the temporary skill folder."""
        import shutil

        shutil.rmtree(self._tmp, ignore_errors=True)

    def time_discover(self) -> None:
        """Benchmark discover() — Tier 1 metadata-only."""
        loader = AgentSkillLoader.from_folder(self._tmp)
        loader.discover()

    def time_activate(self) -> None:
        """Benchmark discover() + activate() — Tier 1 + Tier 2."""
        loader = AgentSkillLoader.from_folder(self._tmp)
        loader.discover()
        loader.activate()

    def time_full_lifecycle(self) -> None:
        """Benchmark discover → activate → list_resources → deactivate."""
        loader = AgentSkillLoader.from_folder(self._tmp)
        loader.discover()
        loader.activate()
        loader.list_resources()
        loader.deactivate()

    def time_to_tool_descriptor(self) -> None:
        """Benchmark conversion to AgentSkillToolDescriptor."""
        loader = AgentSkillLoader.from_folder(self._tmp)
        loader.to_tool_descriptor()


class AgentSkillResourceListSuite:
    """Benchmark list_resources() with varying resource counts."""

    def setup(self) -> None:
        """Create a temporary folder with many resource files."""
        self._tmp = Path(tempfile.mkdtemp(prefix="asv_agent_skill_res_"))
        (self._tmp / "SKILL.md").write_text(_MINIMAL_SKILL_MD)

        scripts = self._tmp / "scripts"
        scripts.mkdir()
        for i in range(20):
            (scripts / f"script_{i:02d}.py").write_text(f"# {i}\n")

        refs = self._tmp / "references"
        refs.mkdir()
        for i in range(10):
            (refs / f"ref_{i:02d}.md").write_text(f"# {i}\n")

        assets = self._tmp / "assets"
        assets.mkdir()
        for i in range(5):
            (assets / f"asset_{i:02d}.json").write_text(f'{{"id": {i}}}\n')

    def teardown(self) -> None:
        """Remove the temporary folder."""
        import shutil

        shutil.rmtree(self._tmp, ignore_errors=True)

    def time_list_resources_35_files(self) -> None:
        """Benchmark list_resources() with 35 resource files."""
        loader = AgentSkillLoader.from_folder(self._tmp)
        loader.list_resources()