diff --git a/features/steps/context_strategy_batch2_steps.py b/features/steps/context_strategy_batch2_steps.py index 1d3a8f4fa..bc58ec750 100644 --- a/features/steps/context_strategy_batch2_steps.py +++ b/features/steps/context_strategy_batch2_steps.py @@ -82,7 +82,7 @@ def step_when_register_all_6(context: Context) -> None: @when("I discover all strategies from the entry point group in the registry") def step_when_discover_via_entry_points(context: Context) -> None: """Trigger actual entry-point discovery via importlib.metadata.""" - discovered = context.registry.discover_from_entry_points() # type: ignore[arg-type] + discovered = context.registry.discover_from_entry_points() context.entry_points_discovered = discovered @@ -91,7 +91,7 @@ def step_when_discover_via_entry_points(context: Context) -> None: ) def step_when_discover_nonexistent_group(context: Context) -> None: """Try to discover from a fake entry-point group — should find nothing.""" - discovered = context.registry.discover_from_entry_points( # type: ignore[arg-type] + discovered = context.registry.discover_from_entry_points( group="cleveragents.nonexistent.group" ) context.entry_points_discovered = discovered @@ -113,3 +113,130 @@ def step_then_arce_via_entry_points(context: Context) -> None: @then("no new strategies should be registered") def step_then_no_new_strategies(context: Context) -> None: """Verify no side effects occurred during discovery.""" + assert context.entry_points_discovered == 0, ( + f"Expected 0 discovered strategies, got {context.entry_points_discovered}" + ) + + +# --------------------------------------------------------------------------- +# Additional Then steps — batch 2 scenarios +# The following step implementations cover scenarios from: +# features/context_strategies_batch2.feature +# features/entry_point_discovery.feature +# --------------------------------------------------------------------------- + + +@then('the registry should contain "{name}"') +def step_then_registry_contains(context: Context, name: str) -> None: + """Verify the registry contains a strategy by name.""" + assert context.registry.is_registered(name), ( + f"'{name}' not in registry. Registered: {context.registry.list_all()}" + ) + + +@then('the entry for "{name}" should be marked as builtin') +def step_then_entry_for_marked_builtin(context: Context, name: str) -> None: + """Verify a named strategy entry is marked as builtin.""" + entry = context.registry.get_entry(name) + assert entry.is_builtin, ( + f"Entry '{name}' is not marked as builtin. " + f"Is registered: {context.registry.is_registered(name)}" + ) + + +@then("the builtin list should include the following strategies") +def step_then_builtin_list_include(context: Context, table: behave.table.Table) -> None: + """Verify a list of strategy names are marked as builtin.""" + builtins = context.registry.list_builtin() + for row in table.dict: + # Iterate over each column (all should map to the same name field) + for key, value in row.items(): + assert value in builtins, ( + f"Expected '{value}' in builtin list. Got: {builtins}" + ) + + +############################################################################## +# --- Named strategy explain steps (batch 2) --- +############################################################################### + + +@then("the arce explain should contain \"{text}\"") +def step_then_arce_explain_contains(context: Context, text: str) -> None: + """Verify ARCE strategy explain mentions the given keyword.""" + strategy = context.registry.get("arce") + explanation = strategy.explain() + assert text in explanation, ( + f"Expected '{text}' in ARCE explain.\nGot:\n{explanation}" + ) + + +@then("the arce name should be \"{name}\"") +def step_then_arce_name(context: Context, name: str) -> None: + """Verify the ARCE strategy has the expected name.""" + strategy = context.registry.get("arce") + assert strategy.name == name, ( + f"Expected ARCE name '{name}', got '{strategy.name}'" + ) + + +############################################################################## +# --- Named strategy explain steps — temporal archaeology --- +############################################################################### + + +@then("the temporal archaeology explain should contain \"{text}\"") +def step_then_temporal_explain_contains(context: Context, text: str) -> None: + """Verify TemporalArchaeology strategy explain mentions the given keyword.""" + strategy = context.registry.get("temporal-archaeology") + explanation = strategy.explain() + assert text in explanation, ( + f"Expected '{text}' in temporal archaeology explain.\nGot:\n{explanation}" + ) + + +############################################################################## +# --- Named strategy explain steps — plan decision context --- +############################################################################### + + +@then("the plan decision context explain should contain \"{text}\"") +def step_then_plan_dec_explain_contains(context: Context, text: str) -> None: + """Verify PlanDecisionContextStrategy explain mentions the given keyword.""" + strategy = context.registry.get("plan-decision-context") + explanation = strategy.explain() + assert text in explanation, ( + f"Expected '{text}' in plan decision context explain.\nGot:\n{explanation}" + ) + + +############################################################################## +# --- Named strategy name steps with quoted single quotes --- +################################################################################ + + +@then("the 'arce' name should be \"{name}\"") +def step_then_arce_quoted_name(context: Context, name: str) -> None: + """Verify ARCE (quoted key) has the expected name. + + This step handles the Gherkin pattern: the 'arce' name should be "arce" + where single quotes delimit the strategy identifier. + """ + strategy = context.registry.get("arce") + assert strategy.name == name, ( + f"Expected 'arce' name '{name}', got '{strategy.name}'" + ) + + +############################################################################## +# --- Builtins should include individual names --- +############################################################################### + + +@then("the builtins should include \"{name}\"") +def step_then_builtin_include(context: Context, name: str) -> None: + """Verify a named strategy exists in the builtin list.""" + builtins = context.registry.list_builtin() + assert name in builtins, ( + f"Expected '{name}' in builtin list. Got: {builtins}" + ) diff --git a/src/cleveragents/application/services/strategy_registry.py b/src/cleveragents/application/services/strategy_registry.py index e69de29bb..1bf03ec10 100644 --- a/src/cleveragents/application/services/strategy_registry.py +++ b/src/cleveragents/application/services/strategy_registry.py @@ -0,0 +1,619 @@ +"""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() + + 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)