feat(context): implement ContextStrategy protocol and plugin registration system
Implement strategy registry integration as spec §47561 by adding: - load_strategies_from_config() in context_strategies.py for TOML-driven strategy bootstrapping (register builtins, discover custom plugins via module:ClassName, set enabled list) - Re-exports of StrategyRegistry, StrategyConfig, StrategyRegistryEntry, StrategyNotFoundError, StrategyRegistrationError and ContextStrategy from strategy_registry and acms_service modules - __all__ export list and DEFAULT_ENABLED_STRATEGIES constant Fix StrategyRegistry.validate_registry() to skip resource_types check for v1 pipeline strategies (context_strategies.py, acms_service.py) whose StrategyCapabilities dataclass lacks the domain-model resource_types field, preventing false-positive warnings. Adds _is_v1_pipeline_caps() helper. Auto-load strategy configuration from Settings in ACMSPipeline.__init__() when a Settings object is provided, reading context.strategies config and registering/ enabling strategies via the plugin loader.
This commit is contained in:
OpenCode AI
Forgejo
@@ -821,6 +821,14 @@ class ACMSPipeline:
|
||||
# can wire configuration (e.g. default budget, strategy allow-lists)
|
||||
# and persistence without changing the constructor signature.
|
||||
self._settings = settings
|
||||
|
||||
# Auto-load strategy configuration from Settings if provided.
|
||||
# Per spec §42947-42980: when a Settings object is passed to the
|
||||
# pipeline, its TOML-derived ``context`` configuration is used to
|
||||
# populate and enable the StrategyRegistry. This eliminates the need
|
||||
# for callers to manually register every strategy.
|
||||
if settings is not None:
|
||||
self._load_strategies_from_settings(settings)
|
||||
self._unit_of_work = unit_of_work
|
||||
|
||||
self._strategies: dict[str, ContextStrategy] = {
|
||||
@@ -874,6 +882,57 @@ class ACMSPipeline:
|
||||
self._plugin_manager = plugin_manager
|
||||
self._discover_context_extensions()
|
||||
|
||||
def _load_strategies_from_settings(self, settings: Settings) -> None:
|
||||
"""Auto-load strategy configuration from a :class:`~cleveragents.config.settings.Settings` object.
|
||||
|
||||
Attempts to read ``context.strategies`` configuration from the TOML-based
|
||||
Settings (typically loaded from a project config file via pydantic-settings).
|
||||
If present, built-in strategies are registered as per default, custom plugins
|
||||
from ``module:ClassName`` strings are discovered, and the enabled list is set.
|
||||
Any errors are logged but do not prevent the pipeline from operating with
|
||||
its core built-in strategies.
|
||||
"""
|
||||
try:
|
||||
context_cfg = getattr(settings, "context", None)
|
||||
if context_cfg is None:
|
||||
return
|
||||
# Extract the strategies sub-section as a dict.
|
||||
strategy_cfg: dict[str, Any] = {}
|
||||
if isinstance(context_cfg, dict):
|
||||
strat_dict = context_cfg.get("strategies", {}) or {}
|
||||
if isinstance(strat_dict, dict):
|
||||
strategy_cfg = dict(strat_dict)
|
||||
else:
|
||||
return
|
||||
|
||||
# Import the lazy-load helper; avoid circular import at module level.
|
||||
from cleveragents.application.services.context_strategies import (
|
||||
load_strategies_from_config,
|
||||
)
|
||||
|
||||
registry = load_strategies_from_config(strategy_cfg)
|
||||
for name in registry.list_all():
|
||||
if name not in self._strategies:
|
||||
strat_inst = registry.get(name)
|
||||
self.register_strategy(name=name, strategy=strat_inst)
|
||||
# Sync the enabled list.
|
||||
enabled = registry.list_enabled()
|
||||
if enabled:
|
||||
self._enforce_enabled_strategies(enabled)
|
||||
|
||||
except Exception as exc: # noqa: BLE001
|
||||
self._logger.warning(
|
||||
"strategy_config_load_failed",
|
||||
error=str(exc),
|
||||
)
|
||||
|
||||
def _enforce_enabled_strategies(self, enabled_names: list[str]) -> None:
|
||||
"""Internal helper to filter ``_strategies`` to the enabled subset."""
|
||||
enabled_set = set(enabled_names)
|
||||
self._strategies = {
|
||||
name: strat for name, strat in self._strategies.items() if name in enabled_set
|
||||
}
|
||||
|
||||
# ------------------------------------------------------------------
|
||||
# Plugin extension point discovery
|
||||
# ------------------------------------------------------------------
|
||||
|
||||
@@ -20,7 +20,12 @@ All strategies implement the v1 ``ContextStrategy`` Protocol defined in
|
||||
``acms_service.py`` and can be registered with ``ACMSPipeline`` via
|
||||
``register_strategy()`` or added to ``BUILTIN_STRATEGIES``.
|
||||
|
||||
Based on ``docs/specification.md`` ¶2520-2521.
|
||||
**Registry integration** (spec §47561): This module is the canonical home for
|
||||
:py:class:`~cleveragents.application.services.strategy_registry.StrategyRegistry`.
|
||||
The strategies here are registered by default when a registry is created, and
|
||||
custom strategies can be loaded via :func:`load_strategies_from_config`.
|
||||
|
||||
Based on ``docs/specification.md`` §25207-25216.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
@@ -39,6 +44,41 @@ from cleveragents.domain.models.core.context_fragment import (
|
||||
ContextFragment,
|
||||
)
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Re-exports for registry integration (spec §47561)
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
from cleveragents.application.services.strategy_registry import (
|
||||
StrategyConfig, # noqa: F401
|
||||
StrategyNotFoundError, # noqa: F401
|
||||
StrategyRegistrationError, # noqa: F401
|
||||
StrategyRegistry, # noqa: F401
|
||||
StrategyRegistryEntry, # noqa: F401
|
||||
)
|
||||
|
||||
from cleveragents.domain.models.acms.strategy import (
|
||||
ContextStrategy, # noqa: F401
|
||||
StrategyCapabilities as _DomainStrategyCapabilities, # noqa: F401
|
||||
)
|
||||
|
||||
__all__: list[str] = [
|
||||
"SimpleKeywordStrategy",
|
||||
"SemanticEmbeddingStrategy",
|
||||
"BreadthDepthNavigatorStrategy",
|
||||
"StrategyRegistry",
|
||||
"StrategyConfig",
|
||||
"StrategyRegistryEntry",
|
||||
"StrategyNotFoundError",
|
||||
"StrategyRegistrationError",
|
||||
]
|
||||
|
||||
# Built-in strategy registry constants (used by load_strategies_from_config)
|
||||
DEFAULT_ENABLED_STRATEGIES: tuple[str, ...] = (
|
||||
"simple-keyword",
|
||||
"semantic-embedding",
|
||||
"breadth-depth-navigator",
|
||||
)
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
_WORD_RE = re.compile(r"\w+", re.UNICODE)
|
||||
@@ -532,3 +572,59 @@ def _max_proximity(node_uri: str, focus_nodes: list[str], max_hops: int) -> floa
|
||||
best = max(best, proximity)
|
||||
|
||||
return best
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Helper: load_strategies_from_config (spec §42947-42980)
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def load_strategies_from_config(config: dict[str, Any]) -> StrategyRegistry:
|
||||
"""Populate and return a :class:`StrategyRegistry` from a TOML config dict.
|
||||
|
||||
Accepts a dict representing the parsed ``[context.strategies]`` section of
|
||||
a TOML configuration file. The expected structure is::
|
||||
|
||||
{
|
||||
"enabled": ["simple-keyword", "semantic-embedding"],
|
||||
"custom": {
|
||||
"my-strategy": "my_package.strategies:MyStrategy",
|
||||
},
|
||||
}
|
||||
|
||||
:param config: Dict with an optional ``"enabled"`` key and an optional
|
||||
``"custom"`` dict mapping strategy names to ``"module:ClassName"``
|
||||
strings.
|
||||
:returns: A fully-populated :class:`StrategyRegistry` ready for use by
|
||||
an ``ACMSPipeline`` or downstream component.
|
||||
|
||||
The built-in strategies from :const:`DEFAULT_ENABLED_STRATEGIES` are
|
||||
always registered first (as built-ins). Custom strategies are discovered
|
||||
and loaded via :meth:`StrategyRegistry.register_from_module`. Finally,
|
||||
the enabled list is set so that only configured strategies are active.
|
||||
"""
|
||||
registry = StrategyRegistry()
|
||||
|
||||
# Register all built-in strategies from DEFAULT_ENABLED_STRATEGIES
|
||||
for strategy_class_name in (
|
||||
"SimpleKeywordStrategy",
|
||||
"SemanticEmbeddingStrategy",
|
||||
"BreadthDepthNavigatorStrategy",
|
||||
):
|
||||
cls = globals()[strategy_class_name]
|
||||
instance = cls()
|
||||
registry.register(
|
||||
instance,
|
||||
config=StrategyConfig(),
|
||||
module_path="cleveragents.application.services.context_strategies",
|
||||
is_builtin=True,
|
||||
)
|
||||
|
||||
# Register custom strategies from module:ClassName strings
|
||||
for name, module_path in (config.get("custom") or {}).items():
|
||||
registry.register_from_module(name=name, module_path=module_path)
|
||||
|
||||
# Set the enabled strategy list
|
||||
enabled = config.get("enabled", DEFAULT_ENABLED_STRATEGIES)
|
||||
registry.set_enabled(list(enabled))
|
||||
|
||||
return registry
|
||||
|
||||
@@ -23,6 +23,9 @@ from typing import Any
|
||||
|
||||
import structlog
|
||||
|
||||
from cleveragents.application.services.acms_service import (
|
||||
StrategyCapabilities as _V1StrategyCapabilities,
|
||||
)
|
||||
from cleveragents.domain.models.acms.strategy import (
|
||||
ContextStrategy,
|
||||
StrategyConfig,
|
||||
@@ -32,6 +35,24 @@ from cleveragents.domain.models.acms.strategy import (
|
||||
logger = structlog.get_logger(__name__)
|
||||
|
||||
|
||||
def _is_v1_pipeline_caps(caps: object) -> bool:
|
||||
"""Return True if *caps* is a v1-pipeline StrategyCapabilities dataclass.
|
||||
|
||||
V1 pipeline strategies (those defined in ``context_strategies.py`` and
|
||||
``acms_service.py``) use a dataclass-style capabilities object that has
|
||||
attributes like :attr:`~StrategyCapabilities.supports_semantic_search`,
|
||||
rather than the domain-model ``StrategyCapabilities`` (from
|
||||
``strategy.py``) which has :attr:`~StrategyCapabilities.uses_text`,
|
||||
:attr:`~StrategyCapabilities.resource_types`, etc.
|
||||
|
||||
We detect v1-capabilities by checking whether the object is an instance
|
||||
of the acms_service dataclass. This allows :meth:`validate_registry` to
|
||||
skip the resource_types check for legacy strategies that do not declare
|
||||
domain-model backend fields.
|
||||
"""
|
||||
return isinstance(caps, _V1StrategyCapabilities)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Exceptions
|
||||
# ---------------------------------------------------------------------------
|
||||
@@ -489,7 +510,12 @@ class StrategyRegistry:
|
||||
f"Strategy '{name}' declares no backend capabilities"
|
||||
)
|
||||
|
||||
if not caps.resource_types:
|
||||
# Skip the resource_types check for v1 pipeline strategies.
|
||||
# Those use a dataclass-style StrategyCapabilities that lacks
|
||||
# the domain-model ``resource_types`` field entirely, so all
|
||||
# warnings about undeclared resource types would be false
|
||||
# positives. See _is_v1_pipeline_caps().
|
||||
if not _is_v1_pipeline_caps(caps) and not caps.resource_types:
|
||||
warnings.append(
|
||||
f"Strategy '{name}' does not declare supported "
|
||||
f"resource types (capabilities.resource_types is empty)"
|
||||
|
||||
@@ -0,0 +1,786 @@
|
||||
"""Tests for the ContextStrategyRegistry and plugin loading system.
|
||||
|
||||
Covers:
|
||||
- Strategy registration and retrieval
|
||||
- Protocol conformance validation
|
||||
- Plugin loading from ``module:ClassName`` strings
|
||||
- TOML configuration loading via :func:`load_strategies_from_config`
|
||||
- Thread safety of registry operations
|
||||
- Fallback degradation chain verification
|
||||
- Config-driven strategy enable/disable
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import threading
|
||||
import time
|
||||
from collections.abc import Sequence
|
||||
from typing import Any
|
||||
|
||||
import pytest
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Test fixtures and helpers
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
class FakeStrategy:
|
||||
"""Minimal strategy that satisfies the v1 ``ContextStrategy`` protocol."""
|
||||
|
||||
def __init__(self, name: str = "fake") -> None:
|
||||
self._name = name
|
||||
|
||||
@property
|
||||
def name(self) -> str: # type: ignore[override]
|
||||
return self._name
|
||||
|
||||
@property
|
||||
def capabilities(self):
|
||||
from cleveragents.application.services.acms_service import (
|
||||
StrategyCapabilities as _Caps,
|
||||
)
|
||||
return _Caps(supports_semantic_search=False)
|
||||
|
||||
def can_handle(self, request: dict[str, Any]) -> float: # type: ignore[override]
|
||||
return 0.5
|
||||
|
||||
def assemble(
|
||||
self,
|
||||
fragments: Sequence[Any],
|
||||
budget: Any,
|
||||
) -> Sequence[Any]:
|
||||
return list(fragments)
|
||||
|
||||
def explain(self) -> str:
|
||||
return "Fake strategy for testing."
|
||||
|
||||
|
||||
class FakeDomainStrategy:
|
||||
"""Strategy that implements the domain-model ``ContextStrategy`` Protocol."""
|
||||
|
||||
def __init__(self, name: str = "domain-fake") -> None:
|
||||
self._name = name
|
||||
|
||||
@property
|
||||
def name(self) -> str: # type: ignore[override]
|
||||
return self._name
|
||||
|
||||
@property
|
||||
def capabilities(self):
|
||||
from cleveragents.domain.models.acms.strategy import StrategyCapabilities
|
||||
return StrategyCapabilities(uses_text=True, resource_types=("text",))
|
||||
|
||||
def can_handle(self, request: dict[str, Any], backends: Any) -> float: # type: ignore[override]
|
||||
return 0.8
|
||||
|
||||
def assemble(
|
||||
self, request: Any, backends: Any, budget: int, plan_context: Any
|
||||
) -> list: # type: ignore[override]
|
||||
return []
|
||||
|
||||
def explain(self) -> str:
|
||||
return "Fake domain strategy for testing."
|
||||
|
||||
|
||||
class FakeStrategyWithNoCaps:
|
||||
"""Strategy with empty capabilities (used to trigger validate_registry warnings)."""
|
||||
|
||||
@property
|
||||
def name(self) -> str:
|
||||
return "noop-cap-fake"
|
||||
|
||||
@property
|
||||
def capabilities(self):
|
||||
from cleveragents.domain.models.acms.strategy import StrategyCapabilities
|
||||
return StrategyCapabilities(uses_text=False, resource_types=()) # empty caps
|
||||
|
||||
def can_handle(self, request: dict[str, Any], backends: Any) -> float:
|
||||
return 0.1
|
||||
|
||||
def assemble(self, request: Any, backends: Any, budget: int, plan_context: Any) -> list:
|
||||
return []
|
||||
|
||||
def explain(self) -> str:
|
||||
return "Strategy with empty capabilities"
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Tests: Strategy registration and retrieval
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
class TestRegistrationAndRetrieval:
|
||||
"""Validate basic register/get/list operations."""
|
||||
|
||||
def test_register_and_get(self):
|
||||
from cleveragents.application.services.strategy_registry import (
|
||||
StrategyRegistry,
|
||||
)
|
||||
|
||||
registry = StrategyRegistry()
|
||||
strategy = FakeStrategy("my-strategy")
|
||||
registry.register(strategy)
|
||||
assert registry.get("my-strategy") is strategy
|
||||
assert "my-strategy" in registry
|
||||
|
||||
def test_register_with_custom_name(self):
|
||||
from cleveragents.application.services.strategy_registry import (
|
||||
StrategyRegistry,
|
||||
)
|
||||
|
||||
registry = StrategyRegistry()
|
||||
strategy = FakeStrategy("original-name")
|
||||
registry.register(strategy, name="override-name")
|
||||
assert "override-name" in registry
|
||||
assert "original-name" not in registry
|
||||
|
||||
def test_register_with_config(self):
|
||||
from cleveragents.application.services.strategy_registry import (
|
||||
StrategyConfig,
|
||||
StrategyRegistry,
|
||||
)
|
||||
|
||||
registry = StrategyRegistry()
|
||||
strategy = FakeStrategy("config-strategy")
|
||||
cfg = StrategyConfig(enabled=False)
|
||||
registry.register(strategy, config=cfg)
|
||||
|
||||
# With enabled=False the strategy should be registered but not in _enabled_order
|
||||
entry = registry.get_entry("config-strategy")
|
||||
assert entry.config.enabled is False
|
||||
|
||||
def test_register_builtin_flag(self):
|
||||
from cleveragents.application.services.strategy_registry import (
|
||||
StrategyRegistry,
|
||||
)
|
||||
|
||||
registry = StrategyRegistry()
|
||||
strategy = FakeStrategy("built-in-1")
|
||||
registry.register(strategy, is_builtin=True)
|
||||
|
||||
assert "built-in-1" in registry.list_builtin()
|
||||
entry = registry.get_entry("built-in-1")
|
||||
assert entry.is_builtin is True
|
||||
|
||||
def test_duplicate_registration_raises(self):
|
||||
from cleveragents.application.services.strategy_registry import (
|
||||
StrategyRegistrationError,
|
||||
StrategyRegistry,
|
||||
)
|
||||
|
||||
registry = StrategyRegistry()
|
||||
s = FakeStrategy("dup")
|
||||
registry.register(s)
|
||||
with pytest.raises(StrategyRegistrationError):
|
||||
registry.register(FakeStrategy("dup"))
|
||||
|
||||
def test_empty_name_raises(self):
|
||||
from cleveragents.application.services.strategy_registry import (
|
||||
StrategyRegistrationError,
|
||||
StrategyRegistry,
|
||||
)
|
||||
|
||||
registry = StrategyRegistry()
|
||||
|
||||
class Nameless:
|
||||
@property
|
||||
def name(self) -> str: # type: ignore[override]
|
||||
return ""
|
||||
|
||||
@property
|
||||
def capabilities(self):
|
||||
from cleveragents.application.services.acms_service import (
|
||||
StrategyCapabilities as _Caps,
|
||||
)
|
||||
return _Caps()
|
||||
|
||||
def can_handle(self, request: dict[str, Any]) -> float: # type: ignore[override]
|
||||
return 0.3
|
||||
|
||||
def assemble(
|
||||
self, fragments: Sequence[Any], budget: Any
|
||||
) -> Sequence[Any]:
|
||||
return []
|
||||
|
||||
def explain(self) -> str:
|
||||
return ""
|
||||
|
||||
with pytest.raises(StrategyRegistrationError, match="empty"):
|
||||
registry.register(Nameless())
|
||||
|
||||
def test_protocol_violation_raises(self):
|
||||
from cleveragents.application.services.strategy_registry import (
|
||||
StrategyRegistrationError,
|
||||
StrategyRegistry,
|
||||
)
|
||||
|
||||
registry = StrategyRegistry()
|
||||
|
||||
class NotStrategy: # No name/abilities etc. at all
|
||||
pass
|
||||
|
||||
with pytest.raises(StrategyRegistrationError):
|
||||
registry.register(NotStrategy()) # type: ignore[arg-type]
|
||||
|
||||
def test_list_all(self):
|
||||
from cleveragents.application.services.strategy_registry import (
|
||||
StrategyRegistry,
|
||||
)
|
||||
|
||||
registry = StrategyRegistry()
|
||||
for i in range(5):
|
||||
entry_name = f"strat-{i}"
|
||||
registry.register(FakeStrategy(entry_name))
|
||||
all_names = registry.list_all()
|
||||
assert sorted(all_names) == ["strat-0", "strat-1", "strat-2", "strat-3", "strat-4"]
|
||||
|
||||
def test_list_enabled_empty(self):
|
||||
from cleveragents.application.services.strategy_registry import (
|
||||
StrategyRegistry,
|
||||
)
|
||||
|
||||
registry = StrategyRegistry()
|
||||
assert registry.list_enabled() == []
|
||||
|
||||
def test_get_entry_and_config(self):
|
||||
from cleveragents.application.services.strategy_registry import (
|
||||
StrategyConfig,
|
||||
StrategyNotFoundError,
|
||||
StrategyRegistry,
|
||||
)
|
||||
|
||||
registry = StrategyRegistry()
|
||||
cfg = StrategyConfig(timeout_seconds=60)
|
||||
registry.register(FakeStrategy("cfg-strat"), config=cfg)
|
||||
|
||||
entry = registry.get_entry("cfg-strat")
|
||||
assert entry.name == "cfg-strat"
|
||||
assert entry.config.timeout_seconds == 60
|
||||
|
||||
loaded_cfg = registry.get_config("cfg-strat")
|
||||
assert loaded_cfg.timeout_seconds == 60
|
||||
|
||||
def test_get_raises_on_missing(self):
|
||||
from cleveragents.application.services.strategy_registry import (
|
||||
StrategyNotFoundError,
|
||||
StrategyRegistry,
|
||||
)
|
||||
|
||||
registry = StrategyRegistry()
|
||||
with pytest.raises(StrategyNotFoundError, match="not found"):
|
||||
registry.get("nonexistent")
|
||||
|
||||
def test_len_and_contains(self):
|
||||
from cleveragents.application.services.strategy_registry import (
|
||||
StrategyRegistry,
|
||||
)
|
||||
|
||||
registry = StrategyRegistry()
|
||||
assert len(registry) == 0
|
||||
assert "empty" not in registry
|
||||
|
||||
registry.register(FakeStrategy("one"))
|
||||
assert len(registry) == 1
|
||||
assert "one" in registry
|
||||
|
||||
def test_unregister(self):
|
||||
from cleveragents.application.services.strategy_registry import (
|
||||
StrategyNotFoundError,
|
||||
StrategyRegistry,
|
||||
)
|
||||
|
||||
registry = StrategyRegistry()
|
||||
registry.register(FakeStrategy("to-unregister"))
|
||||
assert "to-unregister" in registry
|
||||
|
||||
registry.unregister("to-unregister")
|
||||
assert "to-unregister" not in registry
|
||||
|
||||
def test_unregister_unknown_raises(self):
|
||||
from cleveragents.application.services.strategy_registry import (
|
||||
StrategyNotFoundError,
|
||||
StrategyRegistry,
|
||||
)
|
||||
|
||||
registry = StrategyRegistry()
|
||||
with pytest.raises(StrategyNotFoundError):
|
||||
registry.unregister("ghost")
|
||||
|
||||
def test_clear(self):
|
||||
from cleveragents.application.services.strategy_registry import (
|
||||
StrategyRegistry,
|
||||
)
|
||||
|
||||
registry = StrategyRegistry()
|
||||
for i in range(3):
|
||||
registry.register(FakeStrategy(f"clear-{i}"))
|
||||
assert len(registry) == 3
|
||||
|
||||
registry.clear()
|
||||
assert len(registry) == 0
|
||||
assert registry.list_all() == []
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Tests: Protocol conformance validation
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
class TestProtocolValidation:
|
||||
"""Validate :meth:`StrategyRegistry.validate_registry` behavior."""
|
||||
|
||||
def test_validate_returns_no_warnings_empty(self):
|
||||
from cleveragents.application.services.strategy_registry import (
|
||||
StrategyRegistry,
|
||||
)
|
||||
|
||||
registry = StrategyRegistry()
|
||||
warnings = registry.validate_registry()
|
||||
assert warnings == []
|
||||
|
||||
def test_validate_enabled_but_not_registered_warns(self):
|
||||
from cleveragents.application.services.strategy_registry import (
|
||||
StrategyRegistry,
|
||||
)
|
||||
|
||||
registry = StrategyRegistry()
|
||||
registry.inject_stale_enabled_entry("ghost-strategy")
|
||||
warnings = registry.validate_registry()
|
||||
assert any("ghost-strategy" in w for w in warnings)
|
||||
|
||||
def test_validate_domain_strategy_no_resource_types_warns(self):
|
||||
"""Domain-model strategies must declare resource_types."""
|
||||
from cleveragents.application.services.strategy_registry import (
|
||||
StrategyRegistry,
|
||||
)
|
||||
|
||||
registry = StrategyRegistry()
|
||||
strat = FakeStrategyWithNoCaps() # has domain caps with empty resource_types
|
||||
registry.register(strat)
|
||||
warnings = registry.validate_registry()
|
||||
assert any("resource types" in w for w in warnings)
|
||||
|
||||
def test_validate_domain_strategy_with_resource_types_ok(self):
|
||||
"""Domain strategies that declare resource_types pass validation."""
|
||||
from cleveragents.application.services.strategy_registry import (
|
||||
StrategyRegistry,
|
||||
)
|
||||
|
||||
registry = StrategyRegistry()
|
||||
strat = FakeDomainStrategy("with-caps") # has resource_types=("text",)
|
||||
registry.register(strat)
|
||||
|
||||
warnings = registry.validate_registry()
|
||||
assert not any("resource types" in w for w in warnings)
|
||||
|
||||
def test_validate_v1_pipeline_strategies_skip_resource_types(self):
|
||||
"""V1 pipeline strategies (with acms_service StrategyCapabilities) should NOT
|
||||
get resource-types warnings because their capabilities object has no
|
||||
such field. This was the fix introduced by #10590."""
|
||||
from cleveragents.application.services.strategy_registry import (
|
||||
StrategyRegistry,
|
||||
)
|
||||
|
||||
registry = StrategyRegistry()
|
||||
|
||||
# SimpleKeywordStrategy uses v1 pipeline StrategyCapabilities (acms_service)
|
||||
# which lacks the domain-model resource_types field.
|
||||
from cleveragents.application.services.context_strategies import (
|
||||
SimpleKeywordStrategy,
|
||||
)
|
||||
|
||||
registry.register(SimpleKeywordStrategy(), is_builtin=True)
|
||||
warnings = registry.validate_registry()
|
||||
|
||||
# Should NOT have a "resource types" warning for v1 pipeline strategies
|
||||
resource_warnings = [w for w in warnings if "resource types" in w]
|
||||
assert len(resource_warnings) == 0
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Tests: Plugin loading from module:ClassName strings
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
class TestPluginLoading:
|
||||
"""Test :meth:`StrategyRegistry.register_from_module`."""
|
||||
|
||||
def test_register_from_module_success(self):
|
||||
from cleveragents.application.services.context_strategies import (
|
||||
StrategyConfig,
|
||||
StrategyRegistry,
|
||||
)
|
||||
|
||||
registry = StrategyRegistry()
|
||||
registry.register_from_module(
|
||||
name="simple-keyword",
|
||||
module_path="cleveragents.application.services.context_strategies:SimpleKeywordStrategy",
|
||||
config=StrategyConfig(enabled=True),
|
||||
)
|
||||
assert "simple-keyword" in registry
|
||||
strategy = registry.get("simple-keyword")
|
||||
assert strategy.name == "simple-keyword"
|
||||
|
||||
def test_register_from_module_invalid_format_raises(self):
|
||||
from cleveragents.application.services.strategy_registry import (
|
||||
StrategyRegistrationError,
|
||||
StrategyRegistry,
|
||||
)
|
||||
|
||||
registry = StrategyRegistry()
|
||||
with pytest.raises(StrategyRegistrationError, match="module:ClassName"):
|
||||
registry.register_from_module(name="bad", module_path="invalid_format") # type: ignore[arg-type]
|
||||
|
||||
def test_register_from_module_unauthorized_prefix_raises(self):
|
||||
from cleveragents.application.services.strategy_registry import (
|
||||
StrategyRegistrationError,
|
||||
StrategyRegistry,
|
||||
)
|
||||
|
||||
registry = StrategyRegistry()
|
||||
with pytest.raises(StrategyRegistrationError):
|
||||
registry.register_from_module(
|
||||
name="evil",
|
||||
module_path="os.path:join", # type: ignore[arg-type]
|
||||
)
|
||||
|
||||
def test_register_from_module_nonexistent_class_raises(self):
|
||||
from cleveragents.application.services.strategy_registry import (
|
||||
StrategyRegistrationError,
|
||||
StrategyRegistry,
|
||||
)
|
||||
|
||||
registry = StrategyRegistry()
|
||||
with pytest.raises(StrategyRegistrationError, match="not found"):
|
||||
registry.register_from_module(
|
||||
name="bogus",
|
||||
module_path="cleveragents.application.services.context_strategies:"
|
||||
"NoSuchClassDoesNotExist", # type: ignore[arg-type]
|
||||
)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Tests: TOML configuration loading
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
class TestTOMLConfigurationLoading:
|
||||
"""Test :func:`load_strategies_from_config`.
|
||||
|
||||
This exercises the entire pipeline: register built-ins, optionally load
|
||||
custom plugins from module:ClassName strings, and set the enabled list.
|
||||
"""
|
||||
|
||||
def test_load_empty_config_registers_defaults(self):
|
||||
from cleveragents.application.services.context_strategies import (
|
||||
load_strategies_from_config,
|
||||
)
|
||||
|
||||
registry = load_strategies_from_config({}) # {} = minimal config
|
||||
names = registry.list_all()
|
||||
assert "simple-keyword" in names
|
||||
assert "semantic-embedding" in names
|
||||
assert "breadth-depth-navigator" in names
|
||||
|
||||
# Only defaults should be enabled
|
||||
enabled = registry.list_enabled()
|
||||
assert set(enabled) == {
|
||||
"simple-keyword",
|
||||
"semantic-embedding",
|
||||
"breadth-depth-navigator",
|
||||
}
|
||||
|
||||
def test_load_config_with_custom_plugin(self):
|
||||
from cleveragents.application.services.context_strategies import (
|
||||
load_strategies_from_config,
|
||||
)
|
||||
|
||||
registry = load_strategies_from_config(
|
||||
{
|
||||
"custom": {
|
||||
"my-strategy": "cleveragents.application.services.context_strategies:"
|
||||
"SemanticEmbeddingStrategy",
|
||||
},
|
||||
}
|
||||
)
|
||||
assert "my-strategy" in registry
|
||||
assert "simple-keyword" in registry
|
||||
|
||||
def test_load_config_with_custom_and_custom_enabled(self):
|
||||
from cleveragents.application.services.context_strategies import (
|
||||
load_strategies_from_config,
|
||||
)
|
||||
|
||||
registry = load_strategies_from_config(
|
||||
{
|
||||
"enabled": ["simple-keyword", "my-strategy"],
|
||||
"custom": {
|
||||
"my-strategy": "cleveragents.application.services.context_strategies:"
|
||||
"SemanticEmbeddingStrategy", # type: ignore[dict-item]
|
||||
},
|
||||
}
|
||||
)
|
||||
enabled = registry.list_enabled()
|
||||
assert set(enabled) == {"simple-keyword", "my-strategy"}
|
||||
|
||||
def test_load_config_all_builtins_registered_as_builtin(self):
|
||||
from cleveragents.application.services.context_strategies import (
|
||||
load_strategies_from_config,
|
||||
)
|
||||
|
||||
registry = load_strategies_from_config({})
|
||||
for name in ["simple-keyword", "semantic-embedding", "breadth-depth-navigator"]:
|
||||
entry = registry.get_entry(name)
|
||||
assert entry.is_builtin is True
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Tests: Thread safety of registry operations
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
class TestThreadSafety:
|
||||
"""Verify that concurrent register/get/set_enabled calls don't corrupt state."""
|
||||
|
||||
def test_concurrent_register(self):
|
||||
from cleveragents.application.services.strategy_registry import (
|
||||
StrategyRegistry,
|
||||
)
|
||||
|
||||
registry = StrategyRegistry()
|
||||
errors: list[Exception] = [] # type: ignore[var-annotated]
|
||||
|
||||
def register_worker(worker_id: int) -> None:
|
||||
try:
|
||||
for _ in range(50):
|
||||
name = f"concurrent-strat-{worker_id}"
|
||||
registry.register(FakeStrategy(name))
|
||||
except Exception as exc: # noqa: BLE001
|
||||
errors.append(exc)
|
||||
|
||||
threads = [threading.Thread(target=register_worker, args=(i,)) for i in range(10)]
|
||||
for t in threads:
|
||||
t.start()
|
||||
for t in threads:
|
||||
t.join()
|
||||
|
||||
assert not errors, f"Errors during concurrent register: {errors}"
|
||||
# Each worker registers once (deduplicated within its own run)
|
||||
total = len(registry)
|
||||
assert total > 0
|
||||
|
||||
def test_concurrent_get_and_register(self):
|
||||
from cleveragents.application.services.strategy_registry import (
|
||||
StrategyRegistry,
|
||||
)
|
||||
|
||||
registry = StrategyRegistry()
|
||||
registry.register(FakeStrategy("shared")) # Pre-register one strategy
|
||||
errors: list[Exception] = [] # type: ignore[var-annotated]
|
||||
reads: list[str] = [] # type: ignore[var-annotated]
|
||||
|
||||
def reader(count: int) -> None:
|
||||
for _ in range(20):
|
||||
try:
|
||||
s = registry.get("shared")
|
||||
assert s is not None
|
||||
reads.append(s.name)
|
||||
except Exception as exc: # noqa: BLE001
|
||||
errors.append(exc)
|
||||
|
||||
def writer(worker_id: int) -> None:
|
||||
for _ in range(20):
|
||||
name = f"writer-{worker_id}"
|
||||
registry.register(FakeStrategy(name))
|
||||
|
||||
threads: list[threading.Thread] = []
|
||||
threads.append(threading.Thread(target=reader, args=(1,)))
|
||||
for i in range(5):
|
||||
threads.append(threading.Thread(target=writer, args=(i,)))
|
||||
|
||||
for t in threads:
|
||||
t.start()
|
||||
for t in threads:
|
||||
t.join()
|
||||
|
||||
assert not errors
|
||||
assert len(reads) == 20
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Tests: Fallback degradation chain verification
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
class TestFallbackDegradation:
|
||||
"""Verify that SimpleKeywordStrategy is always available as a fallback."""
|
||||
|
||||
def test_simple_keyword_always_produces_results(self):
|
||||
from cleveragents.application.services.context_strategies import (
|
||||
SimpleKeywordStrategy,
|
||||
)
|
||||
from cleveragents.domain.models.core.context_fragment import ContextBudget
|
||||
|
||||
strat = SimpleKeywordStrategy()
|
||||
budget = ContextBudget(max_tokens=1000)
|
||||
|
||||
# Even with empty fragments, should return an empty list (no crash)
|
||||
result = strat.assemble([], budget)
|
||||
assert result == []
|
||||
|
||||
# Create a fake fragment-like object
|
||||
class FakeFrag: # pylint: disable=C0115
|
||||
relevance_score: float = 0.5
|
||||
token_count: int = 10
|
||||
uko_node: str = "project://test/node"
|
||||
content: str = "hello world test fragment"
|
||||
|
||||
fragments = [FakeFrag()] # type: ignore[list-item]
|
||||
result = strat.assemble(fragments, budget)
|
||||
assert len(result) == 1
|
||||
|
||||
def test_simple_keyword_confidence_is_universal(self):
|
||||
from cleveragents.application.services.context_strategies import (
|
||||
SimpleKeywordStrategy,
|
||||
)
|
||||
|
||||
strat = SimpleKeywordStrategy()
|
||||
# Without query in request
|
||||
assert strat.can_handle({}) == 0.3
|
||||
# Request is always handled with at least 0.3 confidence
|
||||
assert strat.can_handle({"query": ""}) == 0.3
|
||||
assert strat.can_handle({"anything": "irrelevant"}) == 0.3
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Tests: Config-driven strategy enable/disable
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
class TestEnableDisable:
|
||||
"""Test :meth:`StrategyRegistry.set_enabled` and config updates."""
|
||||
|
||||
def test_set_enabled_updates_list(self):
|
||||
from cleveragents.application.services.strategy_registry import (
|
||||
StrategyConfig,
|
||||
StrategyRegistry,
|
||||
)
|
||||
|
||||
registry = StrategyRegistry()
|
||||
for name in ["a", "b", "c"]:
|
||||
registry.register(FakeStrategy(name), config=StrategyConfig(enabled=True))
|
||||
|
||||
# Only enable 'a' and 'c'
|
||||
registry.set_enabled(["a", "c"])
|
||||
enabled = registry.list_enabled()
|
||||
assert enabled == ["a", "c"]
|
||||
|
||||
def test_set_unknown_strategy_raises(self):
|
||||
from cleveragents.application.services.strategy_registry import (
|
||||
StrategyNotFoundError,
|
||||
StrategyRegistry,
|
||||
)
|
||||
|
||||
registry = StrategyRegistry()
|
||||
registry.register(FakeStrategy("exists"))
|
||||
with pytest.raises(StrategyNotFoundError, match="unknown"):
|
||||
registry.set_enabled(["nonexistent"])
|
||||
|
||||
def test_set_enabled_updates_config_flags(self):
|
||||
from cleveragents.application.services.strategy_registry import (
|
||||
StrategyConfig,
|
||||
StrategyRegistry,
|
||||
)
|
||||
|
||||
registry = StrategyRegistry()
|
||||
for name in ["x", "y"]:
|
||||
registry.register(FakeStrategy(name))
|
||||
|
||||
# Initially enabled by default (enable_by_default=True)
|
||||
assert registry.get_config("x").enabled is True
|
||||
|
||||
registry.set_enabled(["x"]) # disable y only
|
||||
|
||||
# Check config entries: x stays enabled, y should be disabled
|
||||
assert registry.get_config("x").enabled is True
|
||||
assert registry.get_config("y").enabled is False
|
||||
|
||||
def test_update_config_individual_fields(self):
|
||||
from cleveragents.application.services.strategy_registry import (
|
||||
StrategyConfig,
|
||||
StrategyRegistry,
|
||||
)
|
||||
|
||||
registry = StrategyRegistry()
|
||||
registry.register(FakeStrategy("updatable"))
|
||||
|
||||
registry.update_config(
|
||||
"updatable",
|
||||
enabled=False,
|
||||
timeout_seconds=120,
|
||||
max_fragments=50,
|
||||
)
|
||||
|
||||
cfg = registry.get_config("updatable")
|
||||
assert cfg.enabled is False
|
||||
assert cfg.timeout_seconds == 120
|
||||
assert cfg.max_fragments == 50
|
||||
|
||||
def test_update_config_enable_re_adds_to_enabled_order(self):
|
||||
from cleveragents.application.services.strategy_registry import (
|
||||
StrategyRegistry,
|
||||
)
|
||||
|
||||
registry = StrategyRegistry()
|
||||
registry.register(FakeStrategy("re-enable"))
|
||||
registry.update_config("re-enable", enabled=True)
|
||||
|
||||
assert "re-enable" in registry._enabled_order # pylint: disable=protected-access
|
||||
|
||||
def test_update_config_disable_removes_from_enabled_order(self):
|
||||
from cleveragents.application.services.strategy_registry import (
|
||||
StrategyRegistry,
|
||||
)
|
||||
|
||||
registry = StrategyRegistry()
|
||||
registry.register(FakeStrategy("disable-me"))
|
||||
assert "disable-me" in registry._enabled_order # pylint: disable=protected-access
|
||||
|
||||
registry.update_config("disable-me", enabled=False)
|
||||
assert "disable-me" not in registry._enabled_order # pylint: disable=protected-access
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Tests: Integration with acms_service auto-loading
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
class TestACMSPipelineAutoLoading:
|
||||
"""Verify that ACMSPipeline.__init__ loads strategies from Settings."""
|
||||
|
||||
def test_pipeline_auto_loads_strategies_from_settings(self):
|
||||
from unittest.mock import MagicMock, PropertyMock
|
||||
|
||||
# Create a mock Settings object with context.strategies config.
|
||||
settings = MagicMock()
|
||||
settings.context = { # type: ignore[attr-defined]
|
||||
"strategies": {
|
||||
"enabled": ["simple-keyword", "semantic-embedding"],
|
||||
}
|
||||
}
|
||||
|
||||
from cleveragents.application.services.acms_service import ACMSPipeline
|
||||
|
||||
pipeline = ACMSPipeline(settings=settings)
|
||||
strategies_in_pipeline = list(pipeline._strategies.keys()) # pylint: disable=protected-access
|
||||
|
||||
# Built-in relevance strategy is always registered
|
||||
assert "relevance" in strategies_in_pipeline
|
||||
# Simple-keyword should be auto-loaded
|
||||
assert "simple-keyword" in strategies_in_pipeline
|
||||
|
||||
def test_pipeline_works_without_settings(self):
|
||||
from cleveragents.application.services.acms_service import ACMSPipeline
|
||||
|
||||
pipeline = ACMSPipeline()
|
||||
assert "relevance" in pipeline._strategies # pylint: disable=protected-access
|
||||
Reference in New Issue
Block a user