cleveragents-core/benchmarks/skill_context_bench.py

"""ASV benchmarks for Skill context and registry operations.

Measures the performance of:
- SkillContext creation and metadata access
- SkillContext resource resolution
- SkillContext change tracking
- SkillContext write guard enforcement
- SkillRegistry register/get/list/unregister
- SkillRegistry resolve_tools
"""

from __future__ import annotations

import sys
from pathlib import Path

try:
    from cleveragents.domain.models.core.skill import (
        Skill,
        SkillResolver,
    )
    from cleveragents.skills.context import SkillContext
    from cleveragents.skills.protocol import (
        SkillDefinition,
        SkillMetadata,
    )
    from cleveragents.skills.registry import SkillRegistry
except ModuleNotFoundError:
    sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "src"))
    from cleveragents.domain.models.core.skill import (
        Skill,
        SkillResolver,
    )
    from cleveragents.skills.context import SkillContext
    from cleveragents.skills.protocol import (
        SkillDefinition,
        SkillMetadata,
    )
    from cleveragents.skills.registry import SkillRegistry


# ---------------------------------------------------------------------------
# Fixtures
# ---------------------------------------------------------------------------

_SANDBOX = Path("/tmp/bench-sandbox")

_BASIC_SKILL = Skill(
    name="local/bench-ctx-skill",
    description="Benchmark context skill",
    tool_refs=["local/tool-a", "local/tool-b"],
)

_RESOLVER = SkillResolver()
_RESOLVED = _RESOLVER.resolve_tools(_BASIC_SKILL, {})
_METADATA = SkillMetadata.from_skill(_BASIC_SKILL, resolved=_RESOLVED)
_DEFINITION = SkillDefinition(
    skill=_BASIC_SKILL,
    resolved_tools=_RESOLVED,
    metadata=_METADATA,
)


# ---------------------------------------------------------------------------
# Context Benchmarks
# ---------------------------------------------------------------------------


class TimeSkillContextCreation:
    """Benchmark SkillContext creation throughput."""

    timeout = 60

    def time_create_context(self) -> None:
        """Time basic SkillContext creation."""
        SkillContext(
            plan_id="plan-bench",
            project_id="proj-bench",
            sandbox_path=_SANDBOX,
        )

    def time_create_context_with_bindings(self) -> None:
        """Time SkillContext creation with resource bindings."""
        SkillContext(
            plan_id="plan-bench",
            project_id="proj-bench",
            sandbox_path=_SANDBOX,
            resource_bindings={
                "git-checkout": "/repo/main",
                "artifact-store": "/artifacts",
            },
        )


class TimeSkillContextOperations:
    """Benchmark SkillContext operations."""

    timeout = 60

    def setup(self) -> None:
        """Create a context for benchmarking."""
        self.ctx = SkillContext(
            plan_id="plan-ops",
            project_id="proj-ops",
            sandbox_path=_SANDBOX,
            resource_bindings={"git-checkout": "/repo"},
        )

    def time_resolve_resource(self) -> None:
        """Time resource resolution."""
        self.ctx.resolve_resource("git-checkout")

    def time_get_plan_metadata(self) -> None:
        """Time plan metadata retrieval."""
        self.ctx.get_plan_metadata()

    def time_register_invocation(self) -> None:
        """Time tool invocation registration."""
        self.ctx.register_tool_invocation("local/bench-tool", {"a": 1}, {"b": 2}, 5.0)

    def time_enforce_write_guard_pass(self) -> None:
        """Time write guard check (non-read-only context)."""
        self.ctx.enforce_write_guard("local/bench-tool")


# ---------------------------------------------------------------------------
# Registry Benchmarks
# ---------------------------------------------------------------------------


class TimeSkillRegistryOperations:
    """Benchmark SkillRegistry operations."""

    timeout = 60

    def setup(self) -> None:
        """Create a populated registry for benchmarking."""
        self.registry = SkillRegistry()
        for i in range(50):
            skill = Skill(
                name=f"local/bench-skill-{i:03d}",
                description=f"Benchmark skill {i}",
                tool_refs=["local/tool-a", "local/tool-b"],
            )
            resolver = SkillResolver()
            resolved = resolver.resolve_tools(skill, {})
            meta = SkillMetadata.from_skill(skill, resolved=resolved)
            defn = SkillDefinition(skill=skill, resolved_tools=resolved, metadata=meta)
            self.registry.register(defn)

    def time_get_skill(self) -> None:
        """Time skill lookup."""
        self.registry.get("local/bench-skill-025")

    def time_list_all(self) -> None:
        """Time listing all skills."""
        self.registry.list_all()

    def time_resolve_tools(self) -> None:
        """Time tool resolution for a skill."""
        self.registry.resolve_tools("local/bench-skill-010")

    def time_register_and_unregister(self) -> None:
        """Time register + unregister cycle."""
        name = "local/bench-temp"
        skill = Skill(name=name, description="Temp skill")
        resolver = SkillResolver()
        resolved = resolver.resolve_tools(skill, {})
        meta = SkillMetadata.from_skill(skill, resolved=resolved)
        defn = SkillDefinition(skill=skill, resolved_tools=resolved, metadata=meta)
        self.registry.register(defn)
        self.registry.unregister(name)