cleveragents-core/benchmarks/skill_refresh_bench.py

"""ASV benchmarks for Skill Refresh operations (feat #168).

Measures the performance of:
- ``SkillRefreshResult`` construction and merge
- ``SkillRegistry.refresh()`` — single-skill refresh with and without ToolRegistry
- ``SkillRegistry.refresh_all()`` — batch refresh across 1, 10, and 50 skills
- ``MCPRefreshHook`` construction and notification dispatch overhead
"""

from __future__ import annotations

import sys
from pathlib import Path
from typing import Any

_SRC = str(Path(__file__).resolve().parents[1] / "src")
_ROOT = str(Path(__file__).resolve().parents[1])
for _p in (_SRC, _ROOT):
    if _p not in sys.path:
        sys.path.insert(0, _p)

from cleveragents.domain.models.core.skill import Skill  # noqa: E402
from cleveragents.mcp.adapter import (  # noqa: E402
    MCPServerConfig,
    MCPToolAdapter,
    MCPTransport,
)
from cleveragents.mcp.refresh_hook import MCPRefreshHook  # noqa: E402
from cleveragents.skills.protocol import SkillDefinition, SkillMetadata  # noqa: E402
from cleveragents.skills.refresh import SkillRefreshResult  # noqa: E402
from cleveragents.skills.registry import SkillRegistry  # noqa: E402
from cleveragents.tool.registry import ToolRegistry  # noqa: E402
from cleveragents.tool.runtime import ToolSpec  # noqa: E402

# ---------------------------------------------------------------------------
# Shared helpers
# ---------------------------------------------------------------------------


def _make_skill_def(name: str, tool_refs: list[str] | None = None) -> SkillDefinition:
    skill = Skill(
        name=name,
        description=f"Bench skill {name}",
        tool_refs=tool_refs or [],
    )
    return SkillDefinition(skill=skill, metadata=SkillMetadata.from_skill(skill))


def _make_tool_spec(name: str) -> ToolSpec:
    return ToolSpec(
        name=name, description=f"Bench tool {name}", handler=lambda **_: None
    )


def _make_skill_registry(
    count: int,
    *,
    with_tool_registry: bool = False,
    tool_refs_per_skill: int = 0,
) -> SkillRegistry:
    """Build a SkillRegistry with *count* registered skills."""
    tool_reg: ToolRegistry | None = None
    if with_tool_registry:
        tool_reg = ToolRegistry()
        for i in range(tool_refs_per_skill):
            tool_reg.register(_make_tool_spec(f"bench/tool-{i}"))

    skill_reg = SkillRegistry(tool_registry=tool_reg)
    for i in range(count):
        refs = [f"bench/tool-{j}" for j in range(tool_refs_per_skill)]
        skill_reg.register(_make_skill_def(f"bench/skill-{i:04d}", tool_refs=refs))
    return skill_reg


class _NullTransport(MCPTransport):
    """Minimal no-op transport used by hook benchmarks."""

    def connect(self, config: MCPServerConfig) -> dict[str, Any]:
        return {"capabilities": {"tools": True}}

    def call(self, method: str, params: dict[str, Any]) -> dict[str, Any]:
        return {}

    def close(self) -> None:
        pass


# ---------------------------------------------------------------------------
# SkillRefreshResult benchmarks
# ---------------------------------------------------------------------------


class TimeRefreshResultConstruction:
    """Benchmark SkillRefreshResult construction and properties."""

    def time_construct_empty(self) -> None:
        SkillRefreshResult()

    def time_construct_with_data(self) -> None:
        SkillRefreshResult(
            refreshed=["a", "b", "c"],
            failed={"d": "err"},
            skipped=["e"],
        )

    def time_to_summary(self) -> None:
        result = SkillRefreshResult(
            refreshed=["a", "b", "c"],
            failed={"d": "err"},
            skipped=[],
        )
        result.to_summary()

    def time_merge_two_results(self) -> None:
        r1 = SkillRefreshResult(refreshed=["a", "b"], failed={}, skipped=["c"])
        r2 = SkillRefreshResult(refreshed=["d"], failed={"e": "err"}, skipped=[])
        r1.merge(r2)


# ---------------------------------------------------------------------------
# SkillRegistry.refresh() benchmarks
# ---------------------------------------------------------------------------


class TimeSkillRefreshSingle:
    """Benchmark single-skill refresh with varying configurations."""

    def setup(self) -> None:
        # No tool registry — validation skipped
        self._reg_no_tr = _make_skill_registry(1)
        # With tool registry, no tool refs
        self._reg_with_tr = _make_skill_registry(1, with_tool_registry=True)
        # With tool registry, 5 tool refs
        self._reg_with_refs = _make_skill_registry(
            1, with_tool_registry=True, tool_refs_per_skill=5
        )

    def time_refresh_no_tool_registry(self) -> None:
        self._reg_no_tr.refresh("bench/skill-0000")

    def time_refresh_with_tool_registry_no_refs(self) -> None:
        self._reg_with_tr.refresh("bench/skill-0000")

    def time_refresh_with_tool_registry_5_refs(self) -> None:
        self._reg_with_refs.refresh("bench/skill-0000")


# ---------------------------------------------------------------------------
# SkillRegistry.refresh_all() benchmarks
# ---------------------------------------------------------------------------


class TimeSkillRefreshAll:
    """Benchmark refresh_all() with varying skill counts."""

    timeout = 120.0

    def setup(self) -> None:
        self._reg_1 = _make_skill_registry(
            1, with_tool_registry=True, tool_refs_per_skill=2
        )
        self._reg_10 = _make_skill_registry(
            10, with_tool_registry=True, tool_refs_per_skill=2
        )
        self._reg_50 = _make_skill_registry(
            50, with_tool_registry=True, tool_refs_per_skill=2
        )

    def time_refresh_all_1_skill(self) -> None:
        self._reg_1.refresh_all()

    def time_refresh_all_10_skills(self) -> None:
        self._reg_10.refresh_all()

    def time_refresh_all_50_skills(self) -> None:
        self._reg_50.refresh_all()


# ---------------------------------------------------------------------------
# MCPRefreshHook benchmarks
# ---------------------------------------------------------------------------


class TimeMCPRefreshHookConstruction:
    """Benchmark hook creation overhead."""

    def setup(self) -> None:
        self._config = MCPServerConfig(name="bench", transport="stdio", command="echo")
        self._skill_reg = _make_skill_registry(1)

    def time_hook_construction(self) -> None:
        adapter = MCPToolAdapter(config=self._config, transport=_NullTransport())
        hook = MCPRefreshHook(adapter=adapter, skill_registry=self._skill_reg)
        hook.cancel()


class TimeMCPRefreshHookNotificationDispatch:
    """Benchmark notification dispatch overhead (not including the actual refresh)."""

    timeout = 60.0

    def setup(self) -> None:
        config = MCPServerConfig(name="bench", transport="stdio", command="echo")
        self._adapter = MCPToolAdapter(config=config, transport=_NullTransport())
        self._skill_reg = _make_skill_registry(1)
        # Use a long debounce so the timer never fires during the benchmark
        self._hook = MCPRefreshHook(
            adapter=self._adapter,
            skill_registry=self._skill_reg,
            debounce_seconds=60.0,
        )

    def teardown(self) -> None:
        self._hook.cancel()

    def time_dispatch_ignored_notification(self) -> None:
        """Dispatching an irrelevant notification should be near-zero overhead."""
        self._adapter.dispatch_notification("notifications/other")

    def time_dispatch_tools_list_changed(self) -> None:
        """Dispatching list_changed reschedules the timer each call."""
        self._adapter.dispatch_notification("notifications/tools/list_changed")
        # Cancel so the timer doesn't accumulate across iterations
        self._hook.cancel()
        # Re-register listener (cancel doesn't remove it, just stops the timer)