"""ASV benchmarks for actor loader discovery and registry performance. Measures the performance of: - Directory scanning and YAML file collection - Single-root and multi-root discovery - Cache hit performance (unchanged files) - Re-discovery after file modification - Namespace-filtered listing """ 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.actor.loader import ActorLoader # noqa: E402 _ACTOR_TEMPLATE = """\ name: {namespace}/{name} type: llm description: Benchmark actor {name} version: "1.0" model: gpt-4 """ def _create_actor_dir(count: int, namespace: str = "bench") -> Path: tmp = Path(tempfile.mkdtemp(prefix="actor_bench_")) for i in range(count): content = _ACTOR_TEMPLATE.format(namespace=namespace, name=f"actor_{i}") (tmp / f"actor_{i}.yaml").write_text(content) return tmp class TimeActorDiscoverySingle: """Benchmark discovering actors from a single directory.""" def setup(self) -> None: self.dir_5 = _create_actor_dir(5) self.dir_20 = _create_actor_dir(20) self.dir_50 = _create_actor_dir(50) def teardown(self) -> None: import shutil for d in (self.dir_5, self.dir_20, self.dir_50): shutil.rmtree(d, ignore_errors=True) def time_discover_5_actors(self) -> None: loader = ActorLoader(search_roots=[self.dir_5]) loader.discover() def time_discover_20_actors(self) -> None: loader = ActorLoader(search_roots=[self.dir_20]) loader.discover() def time_discover_50_actors(self) -> None: loader = ActorLoader(search_roots=[self.dir_50]) loader.discover() class TimeActorDiscoveryMultiRoot: """Benchmark discovering actors from multiple search roots.""" def setup(self) -> None: self.roots = [_create_actor_dir(10, f"ns{i}") for i in range(5)] def teardown(self) -> None: import shutil for d in self.roots: shutil.rmtree(d, ignore_errors=True) def time_discover_5_roots_50_actors(self) -> None: loader = ActorLoader(search_roots=self.roots) loader.discover() class TimeActorCacheHit: """Benchmark re-discovery when files are unchanged (cache hit).""" def setup(self) -> None: self.directory = _create_actor_dir(20) self.loader = ActorLoader(search_roots=[self.directory]) self.loader.discover() def teardown(self) -> None: import shutil shutil.rmtree(self.directory, ignore_errors=True) def time_rediscover_cached(self) -> None: self.loader.discover() class TimeActorLookup: """Benchmark namespaced actor lookup and listing.""" def setup(self) -> None: self.directory = _create_actor_dir(50) self.loader = ActorLoader(search_roots=[self.directory]) self.loader.discover() def teardown(self) -> None: import shutil shutil.rmtree(self.directory, ignore_errors=True) def time_get_by_name(self) -> None: self.loader.get("bench/actor_25") def time_list_all(self) -> None: self.loader.list_actors() def time_list_by_namespace(self) -> None: self.loader.list_actors(namespace="bench") class TimeActorExamplesDiscovery: """Benchmark discovering the project example actors.""" def setup(self) -> None: project_root = Path(__file__).resolve().parents[1] self.examples_dir = project_root / "examples" / "actors" def time_discover_examples(self) -> None: loader = ActorLoader(search_roots=[self.examples_dir]) loader.discover() time_single = TimeActorDiscoverySingle() time_multi = TimeActorDiscoveryMultiRoot() time_cache = TimeActorCacheHit() time_lookup = TimeActorLookup() time_examples = TimeActorExamplesDiscovery()