diff --git a/features/adaptive_context_strategy.feature b/features/adaptive_context_strategy.feature new file mode 100644 index 000000000..616e48083 --- /dev/null +++ b/features/adaptive_context_strategy.feature @@ -0,0 +1,211 @@ +Feature: Adaptive Context Strategy Selector and Context Fusion + As a context assembly system + I want to intelligently select context strategies based on plan type + And combine results from multiple strategies with configurable weights + So that I can provide optimal context for different types of plans + + Background: + Given I have an adaptive context selector + And I have a context fusion engine + + Scenario: Register a context strategy + When I register a strategy named "semantic" with a mock implementation + Then the strategy "semantic" should be registered + And the list of registered strategies should contain "semantic" + + Scenario: Register configuration for a plan type + Given I have registered a strategy named "semantic" + When I register configuration for plan type "coding" with primary strategy "semantic" + Then the configuration for plan type "coding" should exist + And the primary strategy for "coding" should be "semantic" + + Scenario: Select strategy for a plan type + Given I have registered a strategy named "semantic" + And I have registered configuration for plan type "coding" with primary strategy "semantic" + When I select a strategy for plan type "coding" + Then the selected strategy should be "semantic" + + Scenario: Select multiple strategies including fallbacks + Given I have registered strategies: "semantic", "syntactic", "lexical" + And I have registered configuration for plan type "analysis" with: + | primary_strategy | semantic | + | fallback_strategies | syntactic, lexical | + When I select all strategies for plan type "analysis" + Then I should get 3 strategies in order: "semantic", "syntactic", "lexical" + + Scenario: Reject duplicate strategy registration + Given I have registered a strategy named "semantic" + When I try to register a strategy named "semantic" again + Then I should get an error about duplicate registration + + Scenario: Reject configuration with unregistered primary strategy + When I try to register configuration with unregistered primary strategy "unknown" + Then I should get an error about unregistered strategy + + Scenario: Reject configuration with unregistered fallback strategy + Given I have registered a strategy named "semantic" + When I try to register configuration with primary "semantic" and fallback "unknown" + Then I should get an error about unregistered fallback strategy + + Scenario: Fuse results from multiple strategies with equal weights + Given I have registered strategies: "semantic", "syntactic" + And I have registered configuration for plan type "coding" with: + | primary_strategy | semantic | + | fallback_strategies | syntactic | + And I have strategy results: + | strategy | files | + | semantic | file1.py:0.8, file2.py:0.6 | + | syntactic | file1.py:0.7, file3.py:0.5 | + When I fuse the results for plan type "coding" with equal weights + Then the fused result should have ranked files: + | file | score | + | file1.py | 1.5 | + | file2.py | 0.6 | + | file3.py | 0.5 | + + Scenario: Fuse results with custom weights + Given I have registered strategies: "semantic", "syntactic" + And I have registered configuration for plan type "coding" with: + | primary_strategy | semantic | + | fallback_strategies | syntactic | + And I have strategy results: + | strategy | files | + | semantic | file1.py:0.8, file2.py:0.6 | + | syntactic | file1.py:0.7, file3.py:0.5 | + When I fuse the results with custom weights: + | semantic | 0.7 | + | syntactic | 0.3 | + Then the fused result should have ranked files: + | file | score | + | file1.py | 0.77 | + | file2.py | 0.42 | + | file3.py | 0.15 | + + Scenario: Get top files from fused result + Given I have a fused result with ranked files: + | file | score | + | file1.py | 1.5 | + | file2.py | 0.6 | + | file3.py | 0.5 | + When I get the top 2 files + Then I should get: "file1.py", "file2.py" + + Scenario: Get file score from fused result + Given I have a fused result with ranked files: + | file | score | + | file1.py | 1.5 | + | file2.py | 0.6 | + When I get the score for "file1.py" + Then the file score should be 1.5 + + Scenario: Get file score for non-existent file + Given I have a fused result with ranked files: + | file | score | + | file1.py | 1.5 | + When I get the score for "nonexistent.py" + Then the file score should be None + + Scenario: Reject fusion with no results + Given I have registered configuration for plan type "coding" + When I try to fuse with empty results + Then I should get an error about no results provided + + Scenario: Reject fusion with invalid weights + Given I have registered strategies: "semantic", "syntactic" + And I have registered configuration for plan type "coding" with primary strategy "semantic" + And I have strategy results with "semantic" and "syntactic" + When I try to fuse with negative weight for "semantic" + Then I should get an error about invalid weight + + Scenario: Normalize weights to sum to 1.0 + Given I have a context fusion engine + When I normalize weights: semantic=2.0, syntactic=1.0 + Then the normalized weights should be: + | semantic | 0.6666666666666666 | + | syntactic | 0.3333333333333333 | + + Scenario: Get fusion metadata + Given I have registered strategies: "semantic", "syntactic" + And I have registered configuration for plan type "coding" with: + | primary_strategy | semantic | + | fallback_strategies | syntactic | + And I have strategy results: + | strategy | files | + | semantic | file1.py:0.8 | + | syntactic | file2.py:0.6 | + When I fuse the results for plan type "coding" + Then the fusion metadata should contain: + | plan_type | coding | + | num_strategies | 2 | + | num_files | 2 | + + Scenario: List configured plan types + Given I have registered configuration for plan types: "coding", "analysis", "testing" + When I list all configured plan types + Then I should get plan types: "coding", "analysis", "testing" + + Scenario: Get configuration for plan type + Given I have registered configuration for plan type "coding" with primary strategy "semantic" + When I get the configuration for plan type "coding" + Then the configuration should have primary strategy "semantic" + + Scenario: Get configuration for non-existent plan type + When I get the configuration for plan type "unknown" + Then the configuration should be None + + Scenario: Fuse with selector configuration weights + Given I have registered strategies: "semantic", "syntactic" + And I have registered configuration for plan type "coding" with: + | primary_strategy | semantic | + | fallback_strategies | syntactic | + | fusion_weights | semantic=0.7, syntactic=0.3 | + And I have strategy results: + | strategy | files | + | semantic | file1.py:0.8 | + | syntactic | file1.py:0.7 | + When I fuse with selector configuration weights + Then the fused result should have file "file1.py" with score 0.77 + + Scenario: Handle strategy results without ranked_files attribute + Given I have registered strategies: "semantic", "syntactic" + And I have registered configuration for plan type "coding" with primary strategy "semantic" + And I have strategy results where "syntactic" has no ranked_files attribute + When I fuse the results for plan type "coding" + Then the fusion should skip the strategy without ranked_files + + Scenario: Validate strategy weight configuration + When I try to create a strategy weight with negative weight + Then I should get an error about invalid weight + + Scenario: Validate adaptive strategy config + When I try to create adaptive strategy config without primary strategy + Then I should get an error about missing primary strategy + + Scenario: Plan type enumeration + Then I should have plan types: "coding", "analysis", "documentation", "refactoring", "testing", "debugging", "unknown" + + Scenario: Select strategy for unconfigured plan type raises error + When I try to select a strategy for unconfigured plan type "coding" + Then I should get an error about no configuration + + Scenario: Select all strategies for unconfigured plan type raises error + When I try to select all strategies for unconfigured plan type "coding" + Then I should get an error about no configuration + + Scenario: Fuse results for unconfigured plan type raises error + Given I have strategy results: + | strategy | files | + | semantic | file1.py:0.8 | + When I try to fuse results for unconfigured plan type "coding" + Then I should get an error about no configuration + + Scenario: Fuse with selector for unconfigured plan type raises error + Given I have strategy results: + | strategy | files | + | semantic | file1.py:0.8 | + When I try to fuse with selector for unconfigured plan type "coding" + Then I should get an error about no configuration + + Scenario: Create strategy weight with valid positive weight + When I create a strategy weight named "semantic" with weight 0.5 + Then the strategy weight should have name "semantic" and weight 0.5 diff --git a/features/steps/adaptive_context_strategy_steps.py b/features/steps/adaptive_context_strategy_steps.py new file mode 100644 index 000000000..d190c88f3 --- /dev/null +++ b/features/steps/adaptive_context_strategy_steps.py @@ -0,0 +1,717 @@ +"""Step definitions for adaptive context strategy selector and fusion tests.""" + +from __future__ import annotations + +from typing import Any + +from behave import given, then, when + +from cleveragents.domain.models.acms.adaptive_selector import ( + AdaptiveContextSelector, + AdaptiveStrategyConfig, + ContextFusion, + FusedResult, + PlanType, + StrategyWeight, +) +from cleveragents.domain.models.acms.crp import ContextFragment +from cleveragents.domain.models.acms.strategy import ( + BackendSet, + ContextRequest, + ContextStrategy, + PlanContext, + StrategyCapabilities, +) + + +def _strip_quoted_csv(value: str) -> list[str]: + """Split a comma-separated quoted-string list and strip outer quotes. + + The Gherkin form ``"a", "b", "c"`` is captured by behave as a single + placeholder containing the inner quotes (``a", "b", "c``); split by + ``", "`` and stripping leftover quotes recovers the original tokens. + """ + return [item.strip().strip('"') for item in value.split(", ")] + + +def _table_pairs(table: Any) -> list[tuple[str, str]]: + """Read a headerless 2-column behave table as ``(key, value)`` pairs. + + Behave promotes the first row of a table to ``headings`` automatically. + For the no-header tables used by these scenarios we recover the lost + pair from ``headings`` and then iterate the data rows. + """ + pairs: list[tuple[str, str]] = [] + if len(table.headings) >= 2: + pairs.append((table.headings[0], table.headings[1])) + for row in table: + pairs.append((row.cells[0], row.cells[1])) + return pairs + + +class MockStrategy: + """Mock strategy for testing.""" + + def __init__(self, strategy_name: str) -> None: + """Initialize mock strategy.""" + self._name = strategy_name + + @property + def name(self) -> str: + """Return strategy name.""" + return self._name + + @property + def capabilities(self) -> StrategyCapabilities: + """Return strategy capabilities.""" + return StrategyCapabilities() + + def can_handle( + self, + request: ContextRequest, + backends: BackendSet, + ) -> float: + """Return confidence for this request.""" + return 1.0 + + def assemble( + self, + request: ContextRequest, + backends: BackendSet, + budget: int, + plan_context: PlanContext, + ) -> list[ContextFragment]: + """Execute mock strategy.""" + return [] + + def explain(self) -> str: + """Return explanation.""" + return f"Mock strategy: {self._name}" + + +# Verify MockStrategy satisfies the ContextStrategy protocol +assert isinstance(MockStrategy("test"), ContextStrategy) + + +class MockStrategyResult: + """Mock strategy result for testing.""" + + def __init__(self, ranked_files: list[tuple[str, float]]) -> None: + """Initialize mock result.""" + self.ranked_files = ranked_files + + +@given("I have an adaptive context selector") +def step_have_selector(context: Any) -> None: + """Initialize adaptive context selector.""" + context.selector = AdaptiveContextSelector() + + +@given("I have a context fusion engine") +def step_have_fusion(context: Any) -> None: + """Initialize context fusion engine.""" + if not hasattr(context, "selector"): + context.selector = AdaptiveContextSelector() + context.fusion = ContextFusion(context.selector) + + +@when('I register a strategy named "{name}" with a mock implementation') +def step_register_strategy(context: Any, name: str) -> None: + """Register a mock strategy.""" + strategy = MockStrategy(name) + context.selector.register_strategy(name, strategy) + + +@then('the strategy "{name}" should be registered') +def step_verify_strategy_registered(context: Any, name: str) -> None: + """Verify strategy is registered.""" + strategies = context.selector.list_registered_strategies() + assert name in strategies, f"Strategy {name} not found in {strategies}" + + +@then('the list of registered strategies should contain "{name}"') +def step_verify_strategy_in_list(context: Any, name: str) -> None: + """Verify strategy is in list.""" + strategies = context.selector.list_registered_strategies() + assert name in strategies + + +@when( + 'I register configuration for plan type "{plan_type}" with primary strategy "{strategy}"' +) +def step_register_config_simple(context: Any, plan_type: str, strategy: str) -> None: + """Register configuration for plan type.""" + plan_type_enum = PlanType(plan_type) + config = AdaptiveStrategyConfig( + plan_type=plan_type_enum, + primary_strategy=strategy, + ) + context.selector.register_config(config) + + +@then('the configuration for plan type "{plan_type}" should exist') +def step_verify_config_exists(context: Any, plan_type: str) -> None: + """Verify configuration exists.""" + plan_type_enum = PlanType(plan_type) + config = context.selector.get_config(plan_type_enum) + assert config is not None, f"No configuration for {plan_type}" + + +@then('the primary strategy for "{plan_type}" should be "{strategy}"') +def step_verify_primary_strategy(context: Any, plan_type: str, strategy: str) -> None: + """Verify primary strategy.""" + plan_type_enum = PlanType(plan_type) + config = context.selector.get_config(plan_type_enum) + assert config is not None + assert config.primary_strategy == strategy + + +@when('I select a strategy for plan type "{plan_type}"') +def step_select_strategy(context: Any, plan_type: str) -> None: + """Select strategy for plan type.""" + plan_type_enum = PlanType(plan_type) + context.selected_strategy = context.selector.select_strategy(plan_type_enum) + + +@then('the selected strategy should be "{name}"') +def step_verify_selected_strategy(context: Any, name: str) -> None: + """Verify selected strategy.""" + assert context.selected_strategy.name == name + + +@given('I have registered strategies: "{strategies}"') +def step_register_multiple_strategies(context: Any, strategies: str) -> None: + """Register multiple strategies.""" + for strategy_name in _strip_quoted_csv(strategies): + strategy = MockStrategy(strategy_name) + context.selector.register_strategy(strategy_name, strategy) + + +@given('I have registered configuration for plan type "{plan_type}" with:') +def step_register_config_with_table(context: Any, plan_type: str) -> None: + """Register configuration with table data (headerless 2-column table).""" + plan_type_enum = PlanType(plan_type) + + config_data: dict[str, Any] = {} + for key, value in _table_pairs(context.table): + if key == "fallback_strategies": + config_data["fallback_strategies"] = [s.strip() for s in value.split(",")] + elif key == "fusion_weights": + weights = {} + for pair in value.split(","): + strategy, weight = pair.strip().split("=") + weights[strategy] = float(weight) + config_data["fusion_weights"] = weights + else: + config_data[key] = value + + config = AdaptiveStrategyConfig( + plan_type=plan_type_enum, + primary_strategy=config_data.get("primary_strategy", ""), + fallback_strategies=config_data.get("fallback_strategies", []), + fusion_weights=config_data.get("fusion_weights", {}), + ) + context.selector.register_config(config) + + +@when('I select all strategies for plan type "{plan_type}"') +def step_select_all_strategies(context: Any, plan_type: str) -> None: + """Select all strategies for plan type.""" + plan_type_enum = PlanType(plan_type) + context.selected_strategies = context.selector.select_strategies(plan_type_enum) + + +@then('I should get {count:d} strategies in order: "{strategies}"') +def step_verify_strategy_order(context: Any, count: int, strategies: str) -> None: + """Verify strategy order.""" + expected = _strip_quoted_csv(strategies) + assert len(context.selected_strategies) == count + for i, expected_name in enumerate(expected): + assert context.selected_strategies[i].name == expected_name + + +@when('I try to register a strategy named "{name}" again') +def step_try_duplicate_registration(context: Any, name: str) -> None: + """Try to register duplicate strategy.""" + try: + strategy = MockStrategy(name) + context.selector.register_strategy(name, strategy) + context.error = None + except ValueError as e: + context.error = str(e) + + +@then("I should get an error about duplicate registration") +def step_verify_duplicate_error(context: Any) -> None: + """Verify duplicate registration error.""" + assert context.error is not None + assert "already registered" in context.error + + +@when('I try to register configuration with unregistered primary strategy "{strategy}"') +def step_try_unregistered_primary(context: Any, strategy: str) -> None: + """Try to register config with unregistered primary strategy.""" + try: + config = AdaptiveStrategyConfig( + plan_type=PlanType.CODING, + primary_strategy=strategy, + ) + context.selector.register_config(config) + context.error = None + except ValueError as e: + context.error = str(e) + + +@then("I should get an error about unregistered strategy") +def step_verify_unregistered_error(context: Any) -> None: + """Verify unregistered strategy error.""" + assert context.error is not None + assert "not registered" in context.error + + +@when( + 'I try to register configuration with primary "{primary}" and fallback "{fallback}"' +) +def step_try_unregistered_fallback(context: Any, primary: str, fallback: str) -> None: + """Try to register config with unregistered fallback.""" + try: + config = AdaptiveStrategyConfig( + plan_type=PlanType.CODING, + primary_strategy=primary, + fallback_strategies=[fallback], + ) + context.selector.register_config(config) + context.error = None + except ValueError as e: + context.error = str(e) + + +@then("I should get an error about unregistered fallback strategy") +def step_verify_unregistered_fallback_error(context: Any) -> None: + """Verify unregistered fallback error.""" + assert context.error is not None + assert "not registered" in context.error + + +@given("I have strategy results:") +def step_have_strategy_results(context: Any) -> None: + """Parse strategy results from table.""" + context.strategy_results = {} + for row in context.table: + strategy = row["strategy"] + files_str = row["files"] + + ranked_files = [] + for file_pair in files_str.split(", "): + file_path, score = file_pair.split(":") + ranked_files.append((file_path, float(score))) + + context.strategy_results[strategy] = MockStrategyResult(ranked_files) + + +@when('I fuse the results for plan type "{plan_type}" with equal weights') +def step_fuse_equal_weights(context: Any, plan_type: str) -> None: + """Fuse results with equal weights.""" + plan_type_enum = PlanType(plan_type) + context.fused_result = context.fusion.fuse_results( + plan_type_enum, + context.strategy_results, + ) + + +@when('I fuse the results for plan type "{plan_type}"') +def step_fuse_for_plan_type(context: Any, plan_type: str) -> None: + """Fuse results using the configured weights for ``plan_type``.""" + plan_type_enum = PlanType(plan_type) + context.fused_result = context.fusion.fuse_results( + plan_type_enum, + context.strategy_results, + ) + + +@then("the fused result should have ranked files:") +def step_verify_ranked_files(context: Any) -> None: + """Verify ranked files in fused result.""" + expected_files = {} + for row in context.table: + file_path = row["file"] + score = float(row["score"]) + expected_files[file_path] = score + + for file_path, expected_score in expected_files.items(): + actual_score = context.fused_result.get_file_score(file_path) + assert actual_score is not None, f"File {file_path} not in results" + assert abs(actual_score - expected_score) < 0.0001, ( + f"Score mismatch for {file_path}: " + f"expected {expected_score}, got {actual_score}" + ) + + +@when("I fuse the results with custom weights:") +def step_fuse_custom_weights(context: Any) -> None: + """Fuse results with custom weights (headerless 2-column table).""" + weights: dict[str, float] = {} + for strategy, weight_str in _table_pairs(context.table): + weights[strategy] = float(weight_str) + + plan_type_enum = PlanType.CODING + context.fused_result = context.fusion.fuse_results( + plan_type_enum, + context.strategy_results, + custom_weights=weights, + ) + + +@given("I have a fused result with ranked files:") +def step_have_fused_result(context: Any) -> None: + """Create a fused result from table.""" + ranked_files = [] + strategy_contributions: dict[str, list[tuple[str, float]]] = {} + + for row in context.table: + file_path = row["file"] + score = float(row["score"]) + ranked_files.append((file_path, score)) + + context.fused_result = FusedResult( + ranked_files=ranked_files, + strategy_contributions=strategy_contributions, + ) + + +@when("I get the top {count:d} files") +def step_get_top_files(context: Any, count: int) -> None: + """Get top N files from fused result.""" + context.top_files = context.fused_result.get_top_files(limit=count) + + +@then('I should get: "{files}"') +def step_verify_top_files(context: Any, files: str) -> None: + """Verify top files.""" + expected = _strip_quoted_csv(files) + assert context.top_files == expected + + +@when('I get the score for "{file_path}"') +def step_get_file_score(context: Any, file_path: str) -> None: + """Get score for a file.""" + context.file_score = context.fused_result.get_file_score(file_path) + + +@then("the file score should be {value}") +def step_verify_file_score(context: Any, value: str) -> None: + """Verify file score.""" + if value == "None": + assert context.file_score is None + else: + expected = float(value) + assert abs(context.file_score - expected) < 0.0001 + + +@when("I try to fuse with empty results") +def step_try_fuse_empty(context: Any) -> None: + """Try to fuse with empty results.""" + try: + plan_type_enum = PlanType.CODING + context.fusion.fuse_results(plan_type_enum, {}) + context.error = None + except ValueError as e: + context.error = str(e) + + +@then("I should get an error about no results provided") +def step_verify_no_results_error(context: Any) -> None: + """Verify no results error.""" + assert context.error is not None + assert "No strategy results" in context.error + + +@given('I have strategy results with "{strategy1}" and "{strategy2}"') +def step_have_two_strategies(context: Any, strategy1: str, strategy2: str) -> None: + """Create strategy results with two strategies.""" + context.strategy_results = { + strategy1: MockStrategyResult([(f"{strategy1}_file.py", 0.8)]), + strategy2: MockStrategyResult([(f"{strategy2}_file.py", 0.6)]), + } + + +@when('I try to fuse with negative weight for "{strategy}"') +def step_try_negative_weight(context: Any, strategy: str) -> None: + """Try to fuse with negative weight.""" + try: + plan_type_enum = PlanType.CODING + context.fusion.fuse_results( + plan_type_enum, + context.strategy_results, + custom_weights={strategy: -0.5}, + ) + context.error = None + except ValueError as e: + context.error = str(e) + + +@then("I should get an error about invalid weight") +def step_verify_invalid_weight_error(context: Any) -> None: + """Verify invalid weight error.""" + assert context.error is not None + assert "positive" in context.error.lower() + + +@when("I normalize weights: {weights}") +def step_normalize_weights(context: Any, weights: str) -> None: + """Normalize weights.""" + weight_dict = {} + for pair in weights.split(", "): + strategy, weight = pair.split("=") + weight_dict[strategy] = float(weight) + + context.normalized = context.fusion._normalize_weights( + weight_dict, + weight_dict.keys(), + ) + + +@then("the normalized weights should be:") +def step_verify_normalized_weights(context: Any) -> None: + """Verify normalized weights (headerless 2-column table).""" + for strategy, weight_str in _table_pairs(context.table): + expected = float(weight_str) + actual = context.normalized[strategy] + assert abs(actual - expected) < 0.0001, ( + f"Weight mismatch for {strategy}: expected {expected}, got {actual}" + ) + + +@then("the fusion metadata should contain:") +def step_verify_fusion_metadata(context: Any) -> None: + """Verify fusion metadata (headerless 2-column table).""" + for key, value in _table_pairs(context.table): + if key in ("num_strategies", "num_files"): + expected: Any = int(value) + else: + expected = value + actual = context.fused_result.fusion_metadata[key] + assert actual == expected, ( + f"Metadata mismatch for {key}: expected {expected}, got {actual}" + ) + + +@given('I have registered configuration for plan types: "{plan_types}"') +def step_register_multiple_configs(context: Any, plan_types: str) -> None: + """Register configurations for multiple plan types.""" + for plan_type_str in _strip_quoted_csv(plan_types): + plan_type_enum = PlanType(plan_type_str) + strategy = MockStrategy(f"{plan_type_str}_strategy") + context.selector.register_strategy(f"{plan_type_str}_strategy", strategy) + + config = AdaptiveStrategyConfig( + plan_type=plan_type_enum, + primary_strategy=f"{plan_type_str}_strategy", + ) + context.selector.register_config(config) + + +@when("I list all configured plan types") +def step_list_plan_types(context: Any) -> None: + """List all configured plan types.""" + context.plan_types = context.selector.list_configured_plan_types() + + +@then('I should get plan types: "{plan_types}"') +def step_verify_plan_types(context: Any, plan_types: str) -> None: + """Verify plan types.""" + expected = [PlanType(pt) for pt in _strip_quoted_csv(plan_types)] + assert context.plan_types == expected + + +@when('I get the configuration for plan type "{plan_type}"') +def step_get_config(context: Any, plan_type: str) -> None: + """Get configuration for plan type. + + Stored on ``fetched_config`` because behave reserves ``context.config`` + for its own runtime configuration object. + """ + plan_type_enum = PlanType(plan_type) + context.fetched_config = context.selector.get_config(plan_type_enum) + + +@then('the configuration should have primary strategy "{strategy}"') +def step_verify_config_primary(context: Any, strategy: str) -> None: + """Verify configuration primary strategy.""" + assert context.fetched_config is not None + assert context.fetched_config.primary_strategy == strategy + + +@then("the configuration should be None") +def step_verify_config_none(context: Any) -> None: + """Verify configuration is None.""" + assert context.fetched_config is None + + +@when("I fuse with selector configuration weights") +def step_fuse_selector_weights(context: Any) -> None: + """Fuse with selector configuration weights.""" + plan_type_enum = PlanType.CODING + context.fused_result = context.fusion.fuse_with_selector( + plan_type_enum, + context.strategy_results, + ) + + +@then('the fused result should have file "{file_path}" with score {score}') +def step_verify_file_score_exact(context: Any, file_path: str, score: str) -> None: + """Verify exact file score.""" + expected = float(score) + actual = context.fused_result.get_file_score(file_path) + assert actual is not None + assert abs(actual - expected) < 0.0001 + + +@given('I have strategy results where "{strategy}" has no ranked_files attribute') +def step_have_result_without_ranked_files(context: Any, strategy: str) -> None: + """Create strategy result without ranked_files.""" + context.strategy_results = { + "semantic": MockStrategyResult([("file1.py", 0.8)]), + strategy: object(), # Object without ranked_files + } + + +@then("the fusion should skip the strategy without ranked_files") +def step_verify_skip_no_ranked_files(context: Any) -> None: + """Verify strategy without ranked_files is skipped.""" + assert context.fused_result is not None + assert len(context.fused_result.ranked_files) > 0 + + +@when("I try to create a strategy weight with negative weight") +def step_try_negative_strategy_weight(context: Any) -> None: + """Try to create strategy weight with negative weight.""" + try: + StrategyWeight(strategy_name="test", weight=-0.5) + context.error = None + except ValueError as e: + context.error = str(e) + + +@when("I try to create adaptive strategy config without primary strategy") +def step_try_config_no_primary(context: Any) -> None: + """Try to create config without primary strategy.""" + try: + AdaptiveStrategyConfig( + plan_type=PlanType.CODING, + primary_strategy="", + ) + context.error = None + except ValueError as e: + context.error = str(e) + + +@then("I should get an error about missing primary strategy") +def step_verify_missing_primary_error(context: Any) -> None: + """Verify missing primary strategy error.""" + assert context.error is not None + assert "primary_strategy" in context.error + + +@then('I should have plan types: "{plan_types}"') +def step_verify_plan_type_enum(context: Any, plan_types: str) -> None: + """Verify plan type enumeration.""" + expected = _strip_quoted_csv(plan_types) + actual = [pt.value for pt in PlanType] + assert actual == expected, f"PlanType mismatch: expected {expected}, got {actual}" + + +@given('I have registered a strategy named "{name}"') +def step_have_registered_strategy(context: Any, name: str) -> None: + """Register a strategy by name.""" + strategy = MockStrategy(name) + context.selector.register_strategy(name, strategy) + + +@when('I try to select a strategy for unconfigured plan type "{plan_type}"') +def step_try_select_unconfigured(context: Any, plan_type: str) -> None: + """Try to select a strategy when no configuration exists for plan_type.""" + try: + context.selector.select_strategy(PlanType(plan_type)) + context.error = None + except ValueError as e: + context.error = str(e) + + +@when('I try to select all strategies for unconfigured plan type "{plan_type}"') +def step_try_select_all_unconfigured(context: Any, plan_type: str) -> None: + """Try to select all strategies when no configuration exists for plan_type.""" + try: + context.selector.select_strategies(PlanType(plan_type)) + context.error = None + except ValueError as e: + context.error = str(e) + + +@when('I try to fuse results for unconfigured plan type "{plan_type}"') +def step_try_fuse_unconfigured(context: Any, plan_type: str) -> None: + """Try to fuse results when no configuration exists for plan_type.""" + try: + context.fusion.fuse_results(PlanType(plan_type), context.strategy_results) + context.error = None + except ValueError as e: + context.error = str(e) + + +@when('I try to fuse with selector for unconfigured plan type "{plan_type}"') +def step_try_fuse_with_selector_unconfigured(context: Any, plan_type: str) -> None: + """Try fuse_with_selector when no configuration exists for plan_type.""" + try: + context.fusion.fuse_with_selector(PlanType(plan_type), context.strategy_results) + context.error = None + except ValueError as e: + context.error = str(e) + + +@then("I should get an error about no configuration") +def step_verify_no_configuration_error(context: Any) -> None: + """Verify error mentions missing plan-type configuration.""" + assert context.error is not None + assert "No configuration" in context.error + + +@when('I create a strategy weight named "{name}" with weight {weight:f}') +def step_create_valid_strategy_weight(context: Any, name: str, weight: float) -> None: + """Create a StrategyWeight with a valid positive weight.""" + context.strategy_weight = StrategyWeight(strategy_name=name, weight=weight) + + +@then('the strategy weight should have name "{name}" and weight {weight:f}') +def step_verify_strategy_weight(context: Any, name: str, weight: float) -> None: + """Verify created StrategyWeight fields.""" + assert context.strategy_weight.strategy_name == name + assert abs(context.strategy_weight.weight - weight) < 0.0001 + + +@given( + 'I have registered configuration for plan type "{plan_type}" with primary strategy "{strategy}"' +) +def step_have_registered_config(context: Any, plan_type: str, strategy: str) -> None: + """Register configuration for plan type, auto-registering the strategy.""" + plan_type_enum = PlanType(plan_type) + if strategy not in context.selector.list_registered_strategies(): + context.selector.register_strategy(strategy, MockStrategy(strategy)) + config = AdaptiveStrategyConfig( + plan_type=plan_type_enum, + primary_strategy=strategy, + ) + context.selector.register_config(config) + + +@given('I have registered configuration for plan type "{plan_type}"') +def step_have_registered_config_no_strategy(context: Any, plan_type: str) -> None: + """Register configuration for plan type with a default strategy.""" + plan_type_enum = PlanType(plan_type) + strategy_name = f"{plan_type}_default" + strategy = MockStrategy(strategy_name) + context.selector.register_strategy(strategy_name, strategy) + config = AdaptiveStrategyConfig( + plan_type=plan_type_enum, + primary_strategy=strategy_name, + ) + context.selector.register_config(config) diff --git a/src/cleveragents/domain/models/acms/adaptive_selector.py b/src/cleveragents/domain/models/acms/adaptive_selector.py new file mode 100644 index 000000000..e1ce109ca --- /dev/null +++ b/src/cleveragents/domain/models/acms/adaptive_selector.py @@ -0,0 +1,340 @@ +"""Adaptive context strategy selector and context fusion implementation. + +This module provides intelligent strategy selection based on plan type and +context fusion to combine results from multiple strategies with configurable weights. +""" + +from __future__ import annotations + +from enum import StrEnum +from typing import Any + +from pydantic import BaseModel, Field, field_validator + +from cleveragents.domain.models.acms.strategy import ContextStrategy + + +class PlanType(StrEnum): + """Enumeration of plan types for strategy selection.""" + + CODING = "coding" + ANALYSIS = "analysis" + DOCUMENTATION = "documentation" + REFACTORING = "refactoring" + TESTING = "testing" + DEBUGGING = "debugging" + UNKNOWN = "unknown" + + +class StrategyWeight(BaseModel): + """Configuration for a strategy weight in fusion.""" + + strategy_name: str + weight: float = 1.0 + enabled: bool = True + + @field_validator("weight") + @classmethod + def _check_weight_positive(cls, v: float) -> float: + if v <= 0: + raise ValueError(f"Weight must be positive, got {v}") + return v + + +class AdaptiveStrategyConfig(BaseModel): + """Configuration for adaptive strategy selection.""" + + plan_type: PlanType + primary_strategy: str + fallback_strategies: list[str] = Field(default_factory=list) + fusion_weights: dict[str, float] = Field(default_factory=dict) + use_fusion: bool = False + + @field_validator("primary_strategy") + @classmethod + def _check_primary_strategy(cls, v: str) -> str: + if not v: + raise ValueError("primary_strategy must be specified") + return v + + +class FusedResult(BaseModel): + """Result of context fusion combining multiple strategy results.""" + + ranked_files: list[tuple[str, float]] # (file_path, combined_score) + # strategy -> [(file, score)] + strategy_contributions: dict[str, list[tuple[str, float]]] + fusion_metadata: dict[str, Any] = Field(default_factory=dict) + + def get_top_files(self, limit: int = 10) -> list[str]: + """Get top N files from fused results. + + Args: + limit: Maximum number of files to return + + Returns: + List of file paths sorted by combined score (descending) + """ + return [file_path for file_path, _ in self.ranked_files[:limit]] + + def get_file_score(self, file_path: str) -> float | None: + """Get combined score for a specific file. + + Args: + file_path: Path to the file + + Returns: + Combined score or None if file not in results + """ + for path, score in self.ranked_files: + if path == file_path: + return score + return None + + +class AdaptiveContextSelector: + """Selects the best context strategy based on plan type and configuration.""" + + def __init__(self) -> None: + """Initialize the adaptive selector.""" + self._strategy_registry: dict[str, ContextStrategy] = {} + self._config_map: dict[PlanType, AdaptiveStrategyConfig] = {} + + def register_strategy(self, name: str, strategy: ContextStrategy) -> None: + """Register a context strategy. + + Args: + name: Unique name for the strategy + strategy: The strategy implementation + + Raises: + ValueError: If strategy name is already registered + """ + if name in self._strategy_registry: + raise ValueError(f"Strategy '{name}' is already registered") + self._strategy_registry[name] = strategy + + def register_config(self, config: AdaptiveStrategyConfig) -> None: + """Register configuration for a plan type. + + Args: + config: Configuration for strategy selection + + Raises: + ValueError: If primary strategy is not registered + """ + if config.primary_strategy not in self._strategy_registry: + raise ValueError( + f"Primary strategy '{config.primary_strategy}' is not registered" + ) + + for fallback in config.fallback_strategies: + if fallback not in self._strategy_registry: + raise ValueError(f"Fallback strategy '{fallback}' is not registered") + + self._config_map[config.plan_type] = config + + def select_strategy(self, plan_type: PlanType) -> ContextStrategy: + """Select the best strategy for a plan type. + + Args: + plan_type: The type of plan + + Returns: + The selected strategy + + Raises: + ValueError: If no configuration exists for plan type + """ + if plan_type not in self._config_map: + raise ValueError(f"No configuration for plan type: {plan_type}") + + config = self._config_map[plan_type] + return self._strategy_registry[config.primary_strategy] + + def select_strategies(self, plan_type: PlanType) -> list[ContextStrategy]: + """Select all applicable strategies for a plan type (primary + fallbacks). + + Args: + plan_type: The type of plan + + Returns: + List of strategies in priority order + + Raises: + ValueError: If no configuration exists for plan type + """ + if plan_type not in self._config_map: + raise ValueError(f"No configuration for plan type: {plan_type}") + + config = self._config_map[plan_type] + strategies: list[ContextStrategy] = [ + self._strategy_registry[config.primary_strategy] + ] + + for fallback_name in config.fallback_strategies: + strategies.append(self._strategy_registry[fallback_name]) + + return strategies + + def get_config(self, plan_type: PlanType) -> AdaptiveStrategyConfig | None: + """Get configuration for a plan type. + + Args: + plan_type: The type of plan + + Returns: + Configuration or None if not found + """ + return self._config_map.get(plan_type) + + def list_registered_strategies(self) -> list[str]: + """List all registered strategy names. + + Returns: + List of strategy names + """ + return list(self._strategy_registry.keys()) + + def list_configured_plan_types(self) -> list[PlanType]: + """List all plan types with configurations. + + Returns: + List of configured plan types + """ + return list(self._config_map.keys()) + + +class ContextFusion: + """Combines results from multiple context strategies with configurable weights.""" + + def __init__(self, selector: AdaptiveContextSelector) -> None: + """Initialize context fusion. + + Args: + selector: The adaptive selector to use for strategy selection + """ + self._selector = selector + + def fuse_results( + self, + plan_type: PlanType, + strategy_results: dict[str, Any], + custom_weights: dict[str, float] | None = None, + ) -> FusedResult: + """Fuse results from multiple strategies. + + Args: + plan_type: The type of plan + strategy_results: Dictionary mapping strategy names to their results + custom_weights: Optional custom weights for strategies + + Returns: + Fused result with ranked files + + Raises: + ValueError: If no results provided or invalid weights + """ + if not strategy_results: + raise ValueError("No strategy results provided") + + config = self._selector.get_config(plan_type) + if not config: + raise ValueError(f"No configuration for plan type: {plan_type}") + + # Determine weights to use + weights = custom_weights or config.fusion_weights or {} + + # Normalize weights + normalized_weights = self._normalize_weights(weights, strategy_results.keys()) + + # Collect all files and their scores from each strategy + # file -> [(strategy, score)] + file_scores: dict[str, list[tuple[str, float]]] = {} + strategy_contributions: dict[str, list[tuple[str, float]]] = {} + + for strategy_name, result in strategy_results.items(): + strategy_contributions[strategy_name] = [] + + if not hasattr(result, "ranked_files"): + continue + + weight = normalized_weights.get(strategy_name, 1.0) + + for file_path, score in result.ranked_files: + if file_path not in file_scores: + file_scores[file_path] = [] + + weighted_score = score * weight + file_scores[file_path].append((strategy_name, weighted_score)) + strategy_contributions[strategy_name].append( + (file_path, weighted_score) + ) + + # Combine scores for each file + combined_scores: list[tuple[str, float]] = [] + for file_path, scores in file_scores.items(): + combined_score = sum(score for _, score in scores) + combined_scores.append((file_path, combined_score)) + + # Sort by combined score (descending) + ranked_files = sorted(combined_scores, key=lambda x: x[1], reverse=True) + + return FusedResult( + ranked_files=ranked_files, + strategy_contributions=strategy_contributions, + fusion_metadata={ + "plan_type": plan_type.value, + "num_strategies": len(strategy_results), + "num_files": len(ranked_files), + "weights_used": normalized_weights, + }, + ) + + def _normalize_weights( + self, weights: dict[str, float], strategy_names: Any + ) -> dict[str, float]: + """Normalize weights to sum to 1.0. + + Args: + weights: Raw weights dictionary + strategy_names: Available strategy names + + Returns: + Normalized weights dictionary + """ + if not weights: + # Equal weights for all strategies — weight 1.0 each. Test + # scenarios in features/adaptive_context_strategy.feature pin + # the "equal weights" semantics to "no scaling" (sum-of-scores + # behaviour), not 1/N normalisation. + return {name: 1.0 for name in strategy_names} + + # Validate all weights are positive + for weight in weights.values(): + if weight <= 0: + raise ValueError(f"All weights must be positive, got {weight}") + + # Normalize to sum to 1.0 + total = sum(weights.values()) + return {name: weight / total for name, weight in weights.items()} + + def fuse_with_selector( + self, + plan_type: PlanType, + strategy_results: dict[str, Any], + ) -> FusedResult: + """Fuse results using weights from selector configuration. + + Args: + plan_type: The type of plan + strategy_results: Dictionary mapping strategy names to their results + + Returns: + Fused result with ranked files + """ + config = self._selector.get_config(plan_type) + if not config: + raise ValueError(f"No configuration for plan type: {plan_type}") + + return self.fuse_results(plan_type, strategy_results, config.fusion_weights)