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