"""ASV benchmarks for context strategy registry. Measures registry lookup, registration, and can_handle performance against spec targets (can_handle < 20ms per strategy, specification.md:43245-43261). """ from __future__ import annotations import importlib import sys from pathlib import Path # Ensure src is on sys.path for ASV _SRC = Path(__file__).resolve().parents[1] / "src" if str(_SRC) not in sys.path: sys.path.insert(0, str(_SRC)) importlib.invalidate_caches() import cleveragents importlib.reload(cleveragents) from cleveragents.application.services.strategy_registry import ( # noqa: E402 StrategyRegistry, ) from cleveragents.domain.models.acms.crp import ContextRequest # noqa: E402 from cleveragents.domain.models.acms.strategy import ( # noqa: E402 BackendSet, StrategyConfig, ) from cleveragents.domain.models.acms.strategy_stubs import ( # noqa: E402 BUILTIN_STRATEGY_CLASSES, DEFAULT_ENABLED_STRATEGIES, SimpleKeywordStrategy, ) from cleveragents.domain.models.acms.stubs import ( # noqa: E402 InMemoryGraphBackend, InMemoryTextBackend, InMemoryVectorBackend, ) class TimeRegistration: """Benchmark strategy registration performance.""" timeout = 10 repeat = 3 number = 100 def time_register_single(self) -> None: """Register a single strategy.""" registry = StrategyRegistry() registry.register(SimpleKeywordStrategy(), is_builtin=True) def time_register_all_builtins(self) -> None: """Register all 6 built-in strategies.""" registry = StrategyRegistry() for cls in BUILTIN_STRATEGY_CLASSES: inst = cls() registry.register(inst, is_builtin=True) class TimeLookup: """Benchmark registry lookup performance.""" timeout = 10 repeat = 3 number = 1000 def setup(self) -> None: self.registry = StrategyRegistry() for cls in BUILTIN_STRATEGY_CLASSES: inst = cls() enabled = inst.name in DEFAULT_ENABLED_STRATEGIES self.registry.register( inst, config=StrategyConfig(enabled=enabled), is_builtin=True, ) def time_get_by_name(self) -> None: """Look up a strategy by name.""" self.registry.get("simple-keyword") def time_list_enabled(self) -> None: """List enabled strategies.""" self.registry.list_enabled() def time_list_all(self) -> None: """List all strategies.""" self.registry.list_all() class TimeCanHandle: """Benchmark can_handle performance (spec target: < 20ms). Spec: specification.md:43245-43261. """ timeout = 10 repeat = 3 number = 1000 def setup(self) -> None: self.request = ContextRequest(query="test query") self.backends_all = BackendSet( text=InMemoryTextBackend(), vector=InMemoryVectorBackend(), graph=InMemoryGraphBackend(), ) self.strategies = [cls() for cls in BUILTIN_STRATEGY_CLASSES] def time_can_handle_all_strategies(self) -> None: """Poll can_handle on all 6 built-in strategies.""" for strategy in self.strategies: strategy.can_handle(self.request, self.backends_all)