diff --git a/src/cleveragents/application/services/strategy_registry.py b/src/cleveragents/application/services/strategy_registry.py index f63c091d1..e69de29bb 100644 --- a/src/cleveragents/application/services/strategy_registry.py +++ b/src/cleveragents/application/services/strategy_registry.py @@ -1,619 +0,0 @@ -"""Context strategy registry for the ACMS. - -Provides ``StrategyRegistry`` — the central registry for context -strategies. Supports: - -- Registration of built-in and custom strategies -- Configuration-driven enable/disable (spec §25218-25233) -- Per-strategy timeout and max-fragment limits (spec §28706-28708) -- Validation that strategies declare supported resource types -- Per-project strategy overrides via enabled list -- Plugin discovery from ``"module:ClassName"`` strings - -Based on ``docs/specification.md`` §§ 25218-25233, 28682-28708, -42947-42980, and issue #191. -""" - -from __future__ import annotations - -import importlib -import threading -from types import MappingProxyType -from typing import Any - -import structlog - -from cleveragents.domain.models.acms.strategy import ( - ContextStrategy, - StrategyConfig, - StrategyRegistryEntry, -) - -logger = structlog.get_logger(__name__) - - -# --------------------------------------------------------------------------- -# Exceptions -# --------------------------------------------------------------------------- - - -class StrategyRegistrationError(Exception): - """Raised when strategy registration fails.""" - - -class StrategyNotFoundError(KeyError): - """Raised when a requested strategy is not in the registry.""" - - -# --------------------------------------------------------------------------- -# StrategyRegistry -# --------------------------------------------------------------------------- - - -class StrategyRegistry: - """Central registry for context strategies. - - Manages built-in and custom strategies with per-strategy - configuration. The registry is the single source of truth for - which strategies are available and how they are configured. - - Strategies are registered via :meth:`register` and discovered - via :meth:`get`, :meth:`list_enabled`, or :meth:`list_all`. - - Configuration-driven registration follows spec §25218-25233: - - - ``[context.strategies] enabled = [...]`` controls the global - enabled list. - - ``[context.strategies.custom] "name" = "module:Class"`` - registers custom strategies from Python modules. - - Per-project overrides replace the global enabled list. - - Lifecycle:: - - registry = StrategyRegistry() - registry.register(my_strategy, config=StrategyConfig(timeout_seconds=10)) - enabled = registry.list_enabled() - result = registry.get("simple-keyword") - """ - - # INVARIANTS (must hold after every public method returns): - # 1. _strategies.keys() == _entries.keys() - # 2. set(_enabled_order) is a subset of _strategies.keys() - # 3. For all n in _enabled_order: _entries[n].config.enabled is True - # 4. For all n not in _enabled_order: _entries[n].config.enabled is False - - # Default module prefix allowlist for register_from_module(). - # Only modules under these prefixes may be dynamically imported. - # Override via the ``allowed_module_prefixes`` constructor parameter. - DEFAULT_ALLOWED_MODULE_PREFIXES: tuple[str, ...] = ("cleveragents.",) - - def __init__( - self, - *, - allowed_module_prefixes: tuple[str, ...] | None = None, - ) -> None: - """Initialize an empty registry. - - Args: - allowed_module_prefixes: Module prefixes permitted for - dynamic import via :meth:`register_from_module`. - Defaults to ``("cleveragents.",)``. Pass an empty - tuple to disable the allowlist (not recommended). - """ - self._lock = threading.RLock() - self._strategies: dict[str, ContextStrategy] = {} - self._entries: dict[str, StrategyRegistryEntry] = {} - self._enabled_order: list[str] = [] - self._allowed_module_prefixes = ( - allowed_module_prefixes - if allowed_module_prefixes is not None - else self.DEFAULT_ALLOWED_MODULE_PREFIXES - ) - - # ------------------------------------------------------------------ - # Registration - # ------------------------------------------------------------------ - - def register( - self, - strategy: ContextStrategy, - *, - name: str | None = None, - config: StrategyConfig | None = None, - module_path: str = "", - is_builtin: bool = False, - ) -> None: - """Register a strategy with the registry. - - Args: - strategy: Strategy instance implementing ``ContextStrategy``. - name: Override name for the registry key. If ``None``, - ``strategy.name`` is used. Used by - :meth:`register_from_module` to honour the TOML key. - config: Per-strategy configuration. Defaults to - ``StrategyConfig()`` if not provided. - module_path: Python module path (for custom strategies). - is_builtin: Whether this is a built-in strategy. - - Raises: - StrategyRegistrationError: If the strategy name is empty, - already registered, or the strategy doesn't satisfy - the ``ContextStrategy`` protocol. - """ - # Validate protocol conformance FIRST — before accessing any - # attributes (e.g., strategy.name) that a non-protocol object - # may not have. - if not isinstance(strategy, ContextStrategy): - label = name or type(strategy).__name__ - raise StrategyRegistrationError( - f"Strategy '{label}' does not satisfy the ContextStrategy protocol" - ) - - name = name or strategy.name - if not name: - raise StrategyRegistrationError("Strategy name must not be empty") - - with self._lock: - if name in self._strategies: - raise StrategyRegistrationError( - f"Strategy '{name}' is already registered" - ) - - resolved_config = config or StrategyConfig() - entry = StrategyRegistryEntry( - name=name, - config=resolved_config, - module_path=module_path, - is_builtin=is_builtin, - ) - - self._strategies[name] = strategy - self._entries[name] = entry - - if resolved_config.enabled: - self._enabled_order.append(name) - - logger.debug( - "strategy.registered", - name=name, - builtin=is_builtin, - enabled=resolved_config.enabled, - ) - - def register_from_module( - self, - name: str, - module_path: str, - *, - config: StrategyConfig | None = None, - ) -> None: - """Register a custom strategy from a ``"module:ClassName"`` string. - - Performs plugin discovery per spec §25225-25226: - ``[context.strategies.custom] "name" = "module:Class"``. - - Note: - Spec §25226 uses dot notation - (``my_package.strategies.DomainStrategy``) while §25232 - uses colon notation - (``my_extensions.scorers:DomainAwareScorer``). - We use **colon notation** (``module:ClassName``) because - it unambiguously separates the importable module from the - class name, matching Python entry-point conventions. - The spec is internally inconsistent here. - - Security: - Module paths should come from trusted configuration only - (e.g., project TOML files). Importing a module executes its - module-level code, and instantiating the class runs its - ``__init__``. Never pass untrusted or user-supplied strings - as ``module_path``. - - Args: - name: Unique name for the strategy. - module_path: Python path in ``"module:ClassName"`` or - ``"module.path:ClassName"`` format. - config: Per-strategy configuration. - - Raises: - StrategyRegistrationError: If the module prefix is not in - the allowlist, import fails, the class doesn't exist, - or the instance doesn't satisfy the ``ContextStrategy`` - protocol. - """ - if ":" not in module_path: - raise StrategyRegistrationError( - f"Invalid module_path '{module_path}': " - f"expected 'module:ClassName' format" - ) - - module_name, class_name = module_path.rsplit(":", 1) - - # CWE-706: Restrict dynamic imports to an allowlist of trusted - # module prefixes to prevent arbitrary code execution. - if self._allowed_module_prefixes and not any( - module_name.startswith(prefix) for prefix in self._allowed_module_prefixes - ): - raise StrategyRegistrationError( - f"Module '{module_name}' is not in the allowed prefix list: " - f"{self._allowed_module_prefixes}. Only modules under these " - f"prefixes may be dynamically imported." - ) - - try: - module = importlib.import_module(module_name) - except ImportError as exc: - raise StrategyRegistrationError( - f"Cannot import module '{module_name}' for strategy '{name}': {exc}" - ) from exc - - cls = getattr(module, class_name, None) - if cls is None: - raise StrategyRegistrationError( - f"Class '{class_name}' not found in module '{module_name}'" - ) - - try: - instance = cls() - except Exception as exc: - raise StrategyRegistrationError( - f"Cannot instantiate '{class_name}' for strategy '{name}': {exc}" - ) from exc - - self.register( - instance, - name=name, - config=config, - module_path=module_path, - is_builtin=False, - ) - - # ------------------------------------------------------------------ - # Query - # ------------------------------------------------------------------ - - def get(self, name: str) -> ContextStrategy: - """Return the strategy with the given name. - - Args: - name: Strategy name. - - Returns: - The registered ``ContextStrategy`` instance. - - Raises: - StrategyNotFoundError: If the name is not registered. - """ - with self._lock: - if name not in self._strategies: - raise StrategyNotFoundError( - f"Strategy '{name}' not found in registry. " - f"Available: {sorted(self._strategies)}" - ) - return self._strategies[name] - - def get_entry(self, name: str) -> StrategyRegistryEntry: - """Return the registry entry (metadata + config) for a strategy. - - Raises: - StrategyNotFoundError: If the name is not registered. - """ - with self._lock: - if name not in self._entries: - raise StrategyNotFoundError(f"Strategy '{name}' not found in registry") - return self._entries[name] - - def get_config(self, name: str) -> StrategyConfig: - """Return the configuration for a registered strategy. - - Raises: - StrategyNotFoundError: If the name is not registered. - """ - return self.get_entry(name).config - - def list_all(self) -> list[str]: - """Return names of all registered strategies.""" - with self._lock: - return sorted(self._strategies) - - def list_enabled(self) -> list[str]: - """Return names of enabled strategies in registration order. - - The order matches the ``[context.strategies] enabled`` list, - which controls priority during selection (spec §28682). - """ - with self._lock: - return [ - n - for n in self._enabled_order - if n in self._entries and self._entries[n].config.enabled - ] - - def list_builtin(self) -> list[str]: - """Return names of built-in strategies.""" - with self._lock: - return sorted(n for n, e in self._entries.items() if e.is_builtin) - - def is_registered(self, name: str) -> bool: - """Check whether a strategy is registered.""" - with self._lock: - return name in self._strategies - - def __len__(self) -> int: - """Return the number of registered strategies.""" - with self._lock: - return len(self._strategies) - - def __contains__(self, name: object) -> bool: - """Check whether a strategy name is registered.""" - with self._lock: - return name in self._strategies - - # ------------------------------------------------------------------ - # Configuration - # ------------------------------------------------------------------ - - def set_enabled(self, names: list[str]) -> None: - """Replace the enabled strategy list. - - Supports per-project overrides: the ``--strategy`` CLI flag or - project YAML replaces the global ``context.strategies.enabled`` - list (spec §3782). - - Also updates each strategy's ``config.enabled`` flag to match: - strategies in *names* are enabled, all others are disabled. - - Args: - names: Ordered list of strategy names to enable. - - Raises: - StrategyNotFoundError: If any name is not registered. - """ - with self._lock: - for name in names: - if name not in self._strategies: - raise StrategyNotFoundError( - f"Cannot enable unknown strategy '{name}'" - ) - - # Update config.enabled for all strategies - enabled_set = set(names) - for sname, entry in self._entries.items(): - should_enable = sname in enabled_set - if entry.config.enabled != should_enable: - new_config = entry.config.model_copy( - update={"enabled": should_enable}, - ) - self._entries[sname] = entry.model_copy( - update={"config": new_config}, - ) - - self._enabled_order = list(dict.fromkeys(names)) - - def update_config( - self, - name: str, - *, - enabled: bool | None = None, - timeout_seconds: int | None = None, - max_fragments: int | None = None, - max_workers: int | None = None, - circuit_breaker_threshold: int | None = None, - resource_types: tuple[str, ...] | None = None, - extra: dict[str, Any] | None = None, - ) -> None: - """Update configuration for a registered strategy. - - Only provided (non-``None``) fields are updated; the rest - retain their current values. - - Args: - name: Strategy name. - enabled: Whether the strategy is enabled. - timeout_seconds: Assembly timeout in seconds (>=1). - max_fragments: Max fragments per call (>=1). - max_workers: Max parallel workers (>=1). - circuit_breaker_threshold: Failures before circuit opens (>=1). - resource_types: Resource types this strategy is limited to. - extra: Strategy-specific extra configuration. - - Raises: - StrategyNotFoundError: If the name is not registered. - pydantic.ValidationError: If the updated values violate - ``StrategyConfig`` constraints (e.g., ``timeout_seconds < 1``). - """ - with self._lock: - entry = self.get_entry(name) - updates: dict[str, Any] = {} - if enabled is not None: - updates["enabled"] = enabled - if timeout_seconds is not None: - updates["timeout_seconds"] = timeout_seconds - if max_fragments is not None: - updates["max_fragments"] = max_fragments - if max_workers is not None: - updates["max_workers"] = max_workers - if circuit_breaker_threshold is not None: - updates["circuit_breaker_threshold"] = circuit_breaker_threshold - if resource_types is not None: - updates["resource_types"] = resource_types - if extra is not None: - updates["extra"] = MappingProxyType(extra) - merged = entry.config.model_dump() - merged.update(updates) - new_config = StrategyConfig.model_validate(merged) - new_entry = entry.model_copy(update={"config": new_config}) - self._entries[name] = new_entry - - # Keep _enabled_order in sync with the enabled flag - if enabled is True and name not in self._enabled_order: - self._enabled_order.append(name) - elif enabled is False and name in self._enabled_order: - self._enabled_order = [n for n in self._enabled_order if n != name] - - # ------------------------------------------------------------------ - # Validation - # ------------------------------------------------------------------ - - def validate_registry(self) -> list[str]: - """Validate the registry and return a list of warnings. - - Checks: - - Every enabled strategy is actually registered. - - Every strategy declares at least one capability. - - Every strategy declares supported resource types - (per issue #191 subtask). - - Returns: - List of warning messages (empty = valid). - """ - with self._lock: - warnings: list[str] = [] - - for name in self._enabled_order: - if name not in self._strategies: - warnings.append(f"Enabled strategy '{name}' is not registered") - - for name, strategy in self._strategies.items(): - caps = strategy.capabilities - - has_any_backend = ( - caps.uses_text - or caps.uses_vector - or caps.uses_graph - or caps.uses_temporal - ) - if not has_any_backend: - warnings.append( - f"Strategy '{name}' declares no backend capabilities" - ) - - if not caps.resource_types: - warnings.append( - f"Strategy '{name}' does not declare supported " - f"resource types (capabilities.resource_types is empty)" - ) - - return warnings - - # ------------------------------------------------------------------ - # Entry-point discovery - # ------------------------------------------------------------------ - - def discover_from_entry_points( - self, *, group: str = "cleveragents.context_strategies" - ) -> int: - """Discover and register strategies from Python entry points. - - Scans the given entry-point group for strategy registrations - (see ``pyproject.toml`` ``[project.entry-points]``). Each matching - entry point is resolved to a class, instantiated, and registered - with the registry as a built-in strategy. - - Security: - Only modules under :attr:`DEFAULT_ALLOWED_MODULE_PREFIXES` may be - dynamically imported from external packages. Internal built-ins - (under ``cleveragents.``) are always permitted. - - Args: - group: The entry-point group to scan. Defaults to - ``"cleveragents.context_strategies"``. - - Returns: - Number of strategies discovered and registered. - - Example:: - - registry = StrategyRegistry() - count = registry.discover_from_entry_points() - # count == 6 for the six built-in strategies - """ - import importlib.metadata as _metadata - - discovered = 0 - - try: - eps = _metadata.entry_points(group=group) - except (ValueError, TypeError): # group doesn't exist - logger.debug( - "strategy.discovering_no_group", - group=group, - ) - return 0 - - for ep in sorted(eps, key=lambda e: e.name): - name = ep.name - try: - cls_or_module = ep.load() - except Exception as exc: - logger.warning( - "strategy.discovering_failed", - name=name, - error=str(exc), - ) - continue - - # The entry point value is ``"module.path:ClassName"``. - # `.load()` returns the class, so we pass it directly to register. - instance = cls_or_module() # type: ignore[possibly-undefined] - - self.register( - instance, - name=name, - config=StrategyConfig(enabled=True), - is_builtin=True, - ) - discovered += 1 - - if discovered > 0: - logger.info( - "strategy.discovered_entry_points", - count=discovered, - group=group, - ) - return discovered - - # ------------------------------------------------------------------ - # Removal (for testing / reconfiguration) - # ------------------------------------------------------------------ - - def unregister(self, name: str) -> None: - """Remove a strategy from the registry. - - Args: - name: Strategy name. - - Raises: - StrategyNotFoundError: If the name is not registered. - """ - with self._lock: - if name not in self._strategies: - raise StrategyNotFoundError( - f"Cannot unregister unknown strategy '{name}'" - ) - - del self._strategies[name] - del self._entries[name] - self._enabled_order = [n for n in self._enabled_order if n != name] - - def clear(self) -> None: - """Remove all registered strategies.""" - with self._lock: - self._strategies.clear() - self._entries.clear() - self._enabled_order.clear() - - def inject_stale_enabled_entry(self, name: str) -> None: - """Inject a stale name into the enabled list for testing. - - This is a test helper that creates a deliberately inconsistent - state where the enabled list references a name that is not - registered. Used to verify that :meth:`validate_registry` and - :meth:`list_enabled` handle stale entries correctly. - - Args: - name: A strategy name that is NOT currently registered. - """ - with self._lock: - self._enabled_order.append(name)