From 809ccc624accfa3fc9bede2c0b6147474cf71a7e Mon Sep 17 00:00:00 2001 From: CleverThis Date: Sat, 6 Jun 2026 03:13:26 -0400 Subject: [PATCH] fix(test): move advanced context strategy test doubles to features/mocks - Extract FakeEmbeddings, RelevanceScoringStrategy, AdaptiveContextSelector, ContextFusionStrategy, and _pack_budget from features/steps/ into new features/mocks/advanced_context_strategies_mocks.py per mock-placement rules - Remove sys.path manipulation from robot/helper_advanced_context_strategies.py; import directly from features.mocks instead of features/steps - Add None guard before selected.assemble() in step_assemble_context_query - Add explicit ValueError for unknown strategy types in step_load_yaml_strategy and load_strategy_from_yaml_impl ISSUES CLOSED: #7574 --- CHANGELOG.md | 1 + CONTRIBUTORS.md | 2 + .../advanced_context_strategies_mocks.py | 159 ++++++++++++++++ .../advanced_context_strategies_steps.py | 172 +----------------- robot/helper_advanced_context_strategies.py | 28 +-- 5 files changed, 181 insertions(+), 181 deletions(-) create mode 100644 features/mocks/advanced_context_strategies_mocks.py diff --git a/CHANGELOG.md b/CHANGELOG.md index af1d17ecd..a0201c35e 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -6,6 +6,7 @@ Changed `wf10_batch.robot` to be less likely to create files, and `plan_generation_graph.robot` to give more test answers. ## [Unreleased] +- **fix(test): move advanced context strategy test doubles to features/mocks** (#7574): Extracted `FakeEmbeddings`, `RelevanceScoringStrategy`, `AdaptiveContextSelector`, `ContextFusionStrategy`, and `_pack_budget` from `features/steps/advanced_context_strategies_steps.py` into a new `features/mocks/advanced_context_strategies_mocks.py` file per CONTRIBUTING.md mock-placement rules. Updated the Robot Framework helper `robot/helper_advanced_context_strategies.py` to import directly from `features.mocks` rather than manipulating `sys.path` to reach the Behave steps file. Added `None` guard in `step_assemble_context_query` before calling `selected.assemble()`, and added explicit `ValueError` for unknown strategy types in both `step_load_yaml_strategy` and `load_strategy_from_yaml_impl`. - **fix(a2a): regression tests for stale cleveragents.acp removal** (#5566): Added two Behave BDD scenarios verifying that `cleveragents.acp` is not importable (raises `ImportError`) and that `src/cleveragents/acp/` does not exist in the source tree. These guard against regression of the `__pycache__`-based import that allowed the removed ACP module to still be loaded from bytecode after the v3.6.0 rename to `a2a`. - **Virtual Resource Type Base Class** (#8610): Implemented `VirtualResource` base class with two example concrete implementations (`MetricResource`, `APIEndpointResource`) for abstract/computed resources that are derived rather than mapped to physical files. Virtual resources are computed on demand via a `compute_fn` callable. Includes Behave BDD scenarios in `features/resource_virtual_types.feature` exercising construction, computation, name validation, kwargs passthrough, exception handling, string representation, and subclassing. Resource names are validated against `^[a-zA-Z][a-zA-Z0-9_-]*$` (must start with a letter; alphanumeric, hyphens, and underscores otherwise). - **test(e2e): restore complete M2 acceptance test** (#11191): Restored the truncated M2 full actor compiler and LLM integration e2e acceptance test to its complete 10-step form. Added dynamic LLM provider selection via `Resolve LLM Actor` (falls back to Anthropic when OpenAI is unavailable or quota-exhausted), replacing hardcoded `gpt-4` / `openai/gpt-4` references in the actor config and action YAML. Added explicit return-code validation (`Should Be Equal As Integers ${r_actor.rc} 0`) for the actor registration step. diff --git a/CONTRIBUTORS.md b/CONTRIBUTORS.md index a120f9b77..d341bd4a0 100644 --- a/CONTRIBUTORS.md +++ b/CONTRIBUTORS.md @@ -1,5 +1,7 @@ # Contributors +* HAL9000 has contributed fix for #7574 — move advanced context strategy test doubles to features/mocks and resolve lint violation in Robot Framework helper. + * HAL9000 * Aditya Chhabra * Brent E. Edwards diff --git a/features/mocks/advanced_context_strategies_mocks.py b/features/mocks/advanced_context_strategies_mocks.py new file mode 100644 index 000000000..bd024fbc4 --- /dev/null +++ b/features/mocks/advanced_context_strategies_mocks.py @@ -0,0 +1,159 @@ +"""Mock implementations for advanced context strategies tests. + +FakeEmbeddings provides deterministic word-overlap embeddings so tests never +hit a real embedding API. The three strategy classes are test-only +implementations that satisfy the strategy duck-type contract used by the +Behave and Robot Framework test layers. +""" + +from __future__ import annotations + +from typing import Any + +from cleveragents.application.services.context_strategies import ( + BreadthDepthNavigatorStrategy, + SemanticEmbeddingStrategy, +) +from cleveragents.domain.models.core.context_fragment import ( + ContextBudget, + ContextFragment, +) + + +class FakeEmbeddings: + """Deterministic fake embeddings for testing without real API calls.""" + + def __init__(self) -> None: + self._cache: dict[str, list[float]] = {} + + def embed_query(self, text: str) -> list[float]: + if text not in self._cache: + hash_val = hash(text) % 1000 + self._cache[text] = [float((hash_val + i) % 100) / 100.0 for i in range(10)] + return self._cache[text] + + def embed_documents(self, texts: list[str]) -> list[list[float]]: + return [self.embed_query(text) for text in texts] + + +def _pack_budget( + fragments: list[ContextFragment], budget: ContextBudget +) -> list[ContextFragment]: + result: list[ContextFragment] = [] + used_tokens = budget.reserved_tokens + + for frag in fragments: + if used_tokens + frag.token_count <= budget.max_tokens: + result.append(frag) + used_tokens += frag.token_count + else: + break + + return result + + +class RelevanceScoringStrategy: + """Strategy that ranks fragments purely by relevance score.""" + + def __init__(self) -> None: + pass + + @property + def name(self) -> str: + return "relevance-scoring" + + def can_handle(self, request: dict[str, Any]) -> float: + return 0.5 + + def assemble( + self, + fragments: list[ContextFragment], + budget: ContextBudget, + ) -> list[ContextFragment]: + if not fragments: + return [] + sorted_frags = sorted(fragments, key=lambda f: f.relevance_score, reverse=True) + return _pack_budget(sorted_frags, budget) + + def explain(self) -> str: + return "Ranks fragments purely by relevance score." + + +class AdaptiveContextSelector: + """Selects the best strategy based on request characteristics.""" + + def __init__(self) -> None: + self._strategies: dict[str, Any] = { + "semantic-embedding": SemanticEmbeddingStrategy(), + "relevance-scoring": RelevanceScoringStrategy(), + "breadth-depth-navigator": BreadthDepthNavigatorStrategy(), + } + + @property + def name(self) -> str: + return "adaptive-selector" + + def select_strategy(self, request: dict[str, Any]) -> tuple[str, Any]: + best_name = "relevance-scoring" + best_confidence = 0.0 + + for name, strategy in self._strategies.items(): + confidence = strategy.can_handle(request) + if confidence > best_confidence: + best_confidence = confidence + best_name = name + + return best_name, self._strategies[best_name] + + +class ContextFusionStrategy: + """Fuses results from multiple strategies.""" + + def __init__(self, strategy_names: list[str]) -> None: + self._strategy_names = strategy_names + self._strategies: dict[str, Any] = { + "semantic-embedding": SemanticEmbeddingStrategy(), + "relevance-scoring": RelevanceScoringStrategy(), + "breadth-depth-navigator": BreadthDepthNavigatorStrategy(), + } + + @property + def name(self) -> str: + return "context-fusion" + + def assemble( + self, + fragments: list[ContextFragment], + budget: ContextBudget, + query: str = "", + ) -> list[ContextFragment]: + if not fragments: + return [] + + all_results: dict[str, ContextFragment] = {} + remaining_budget = budget.max_tokens - budget.reserved_tokens + + for strategy_name in self._strategy_names: + if remaining_budget <= 0: + break + + strategy = self._strategies.get(strategy_name) + if not strategy: + continue + + if hasattr(strategy, "set_query"): + strategy.set_query(query) + + strategy_budget = ContextBudget( + max_tokens=remaining_budget, + reserved_tokens=0, + ) + + results = strategy.assemble(fragments, strategy_budget) + + for frag in results: + if frag.uko_node not in all_results: + all_results[frag.uko_node] = frag + remaining_budget -= frag.token_count + + return list(all_results.values()) diff --git a/features/steps/advanced_context_strategies_steps.py b/features/steps/advanced_context_strategies_steps.py index 30229416a..218587938 100644 --- a/features/steps/advanced_context_strategies_steps.py +++ b/features/steps/advanced_context_strategies_steps.py @@ -7,7 +7,6 @@ context fusion strategies using FakeEmbeddings for deterministic behavior. from __future__ import annotations import logging -from typing import Any from behave import given, then, when from behave.runner import Context @@ -21,6 +20,12 @@ from cleveragents.domain.models.core.context_fragment import ( ContextFragment, FragmentProvenance, ) +from features.mocks.advanced_context_strategies_mocks import ( + AdaptiveContextSelector, + ContextFusionStrategy, + FakeEmbeddings, + RelevanceScoringStrategy, +) logger = logging.getLogger(__name__) @@ -28,167 +33,6 @@ logger = logging.getLogger(__name__) _TEST_PROVENANCE = FragmentProvenance(resource_uri="test://fixture") -# =========================================================================== -# Fixtures and Helpers -# =========================================================================== - - -class FakeEmbeddings: - """Deterministic fake embeddings for testing without real API calls.""" - - def __init__(self) -> None: - self._cache: dict[str, list[float]] = {} - - def embed_query(self, text: str) -> list[float]: - """Generate deterministic embedding for query.""" - if text not in self._cache: - # Simple deterministic hash-based embedding - hash_val = hash(text) % 1000 - self._cache[text] = [ - float((hash_val + i) % 100) / 100.0 for i in range(10) - ] - return self._cache[text] - - def embed_documents(self, texts: list[str]) -> list[list[float]]: - """Generate deterministic embeddings for documents.""" - return [self.embed_query(text) for text in texts] - - -class RelevanceScoringStrategy: - """Strategy that ranks fragments purely by relevance score.""" - - def __init__(self) -> None: - pass - - @property - def name(self) -> str: - return "relevance-scoring" - - def can_handle(self, request: dict[str, Any]) -> float: - """Always handle with moderate confidence.""" - return 0.5 - - def assemble( - self, - fragments: list[ContextFragment], - budget: ContextBudget, - ) -> list[ContextFragment]: - """Rank fragments by relevance score.""" - if not fragments: - return [] - - sorted_frags = sorted( - fragments, key=lambda f: f.relevance_score, reverse=True - ) - return _pack_budget(sorted_frags, budget) - - def explain(self) -> str: - return "Ranks fragments purely by relevance score." - - -class AdaptiveContextSelector: - """Selects the best strategy based on request characteristics.""" - - def __init__(self) -> None: - self._strategies: dict[str, Any] = { - "semantic-embedding": SemanticEmbeddingStrategy(), - "relevance-scoring": RelevanceScoringStrategy(), - "breadth-depth-navigator": BreadthDepthNavigatorStrategy(), - } - - @property - def name(self) -> str: - return "adaptive-selector" - - def select_strategy(self, request: dict[str, Any]) -> tuple[str, Any]: - """Select best strategy based on request.""" - best_name = "relevance-scoring" - best_confidence = 0.0 - - for name, strategy in self._strategies.items(): - confidence = strategy.can_handle(request) - if confidence > best_confidence: - best_confidence = confidence - best_name = name - - return best_name, self._strategies[best_name] - - -class ContextFusionStrategy: - """Fuses results from multiple strategies.""" - - def __init__(self, strategy_names: list[str]) -> None: - self._strategy_names = strategy_names - self._strategies: dict[str, Any] = { - "semantic-embedding": SemanticEmbeddingStrategy(), - "relevance-scoring": RelevanceScoringStrategy(), - "breadth-depth-navigator": BreadthDepthNavigatorStrategy(), - } - - @property - def name(self) -> str: - return "context-fusion" - - def assemble( - self, - fragments: list[ContextFragment], - budget: ContextBudget, - query: str = "", - ) -> list[ContextFragment]: - """Fuse results from multiple strategies.""" - if not fragments: - return [] - - all_results: dict[str, ContextFragment] = {} - remaining_budget = budget.max_tokens - budget.reserved_tokens - - for strategy_name in self._strategy_names: - if remaining_budget <= 0: - break - - strategy = self._strategies.get(strategy_name) - if not strategy: - continue - - # Set query if strategy supports it - if hasattr(strategy, "set_query"): - strategy.set_query(query) - - # Create budget for this strategy - strategy_budget = ContextBudget( - max_tokens=remaining_budget, - reserved_tokens=0, - ) - - # Get results from strategy - results = strategy.assemble(fragments, strategy_budget) - - # Add to all results (deduplication by uko_node) - for frag in results: - if frag.uko_node not in all_results: - all_results[frag.uko_node] = frag - remaining_budget -= frag.token_count - - return list(all_results.values()) - - -def _pack_budget( - fragments: list[ContextFragment], budget: ContextBudget -) -> list[ContextFragment]: - """Pack fragments within token budget.""" - result: list[ContextFragment] = [] - used_tokens = budget.reserved_tokens - - for frag in fragments: - if used_tokens + frag.token_count <= budget.max_tokens: - result.append(frag) - used_tokens += frag.token_count - else: - break - - return result - - def _make_fragment( uko_node: str, content: str, @@ -438,6 +282,8 @@ def step_load_yaml_strategy(context: Context) -> None: elif strategy_type == "context-fusion": strategies = context.yaml_config.get("strategies", []) context.loaded_strategy = ContextFusionStrategy(strategies) + else: + raise ValueError(f"Unknown strategy type: {strategy_type!r}") @when("I assemble context with query {query}") @@ -466,6 +312,8 @@ def step_assemble_context_query(context: Context, query: str) -> None: if selected and hasattr(selected, "set_query"): selected.set_query(query_str) + if selected is None: + raise AssertionError(f"No strategy found with name {best_name!r}") context.selected_strategy_name = best_name context.results = selected.assemble(context.fragments, context.budget) diff --git a/robot/helper_advanced_context_strategies.py b/robot/helper_advanced_context_strategies.py index 540642297..02001a411 100644 --- a/robot/helper_advanced_context_strategies.py +++ b/robot/helper_advanced_context_strategies.py @@ -2,10 +2,14 @@ from __future__ import annotations -import sys -from pathlib import Path from typing import Any +from features.mocks.advanced_context_strategies_mocks import ( + AdaptiveContextSelector, + ContextFusionStrategy, + RelevanceScoringStrategy, +) + from cleveragents.application.services.context_strategies import ( BreadthDepthNavigatorStrategy, SemanticEmbeddingStrategy, @@ -19,16 +23,6 @@ from cleveragents.domain.models.core.context_fragment import ( # Default provenance used for test fragments (no real resource needed). _TEST_PROVENANCE = FragmentProvenance(resource_uri="test://fixture") -# Import from step definitions -features_path = Path(__file__).parent.parent / "features" / "steps" -sys.path.insert(0, str(features_path)) - -from advanced_context_strategies_steps import ( # noqa: E402 - AdaptiveContextSelector, - ContextFusionStrategy, - RelevanceScoringStrategy, -) - def create_semantic_search_strategy_impl() -> SemanticEmbeddingStrategy: """Create a semantic search strategy with FakeEmbeddings.""" @@ -80,9 +74,7 @@ def create_test_fragments_impl(args: list[str]) -> list[ContextFragment]: return fragments -def create_context_budget_impl( - max_tokens: int, reserved_tokens: int -) -> ContextBudget: +def create_context_budget_impl(max_tokens: int, reserved_tokens: int) -> ContextBudget: """Create a context budget.""" return ContextBudget( max_tokens=int(max_tokens), @@ -159,8 +151,8 @@ def load_strategy_from_yaml_impl(config: dict[str, Any]) -> Any: elif strategy_type == "context-fusion": strategies = config.get("strategies", []) return ContextFusionStrategy(strategies) - - return None + else: + raise ValueError(f"Unknown strategy type: {strategy_type!r}") def create_context_assembler_with_advanced_strategies_impl() -> dict[str, Any]: @@ -193,7 +185,6 @@ def assemble_context_with_query_impl( """Assemble context with a query.""" request = {"query": query} - # Select best strategy best_strategy = None best_confidence = 0.0 @@ -206,7 +197,6 @@ def assemble_context_with_query_impl( if best_strategy is None: best_strategy = assembler["strategies"][0] - # Set query if needed if hasattr(best_strategy, "set_query"): best_strategy.set_query(query)