From c2bd33dfafc5089ade02b21aa770e94342abb8a9 Mon Sep 17 00:00:00 2001 From: HAL9000 Date: Sat, 18 Apr 2026 20:08:41 +0000 Subject: [PATCH 1/6] feat(context): implement adaptive context strategy selector and fusion Implements adaptive context strategy selector that chooses the best context strategy based on plan type, and context fusion that combines results from multiple strategies with configurable weights. Features: - AdaptiveContextSelector: Intelligent strategy selection per plan type - ContextFusion: Weighted combination of multiple strategy results - PlanType enumeration: coding, analysis, documentation, refactoring, testing, debugging - AdaptiveStrategyConfig: YAML-compatible configuration for strategy selection - FusedResult: Ranked file list with strategy contributions and metadata - Full type annotations and comprehensive Behave BDD tests Closes #5255 --- features/adaptive_context_strategy.feature | 185 ++++++ .../steps/adaptive_context_strategy_steps.py | 545 ++++++++++++++++++ .../domain/models/acms/adaptive_selector.py | 332 +++++++++++ 3 files changed, 1062 insertions(+) create mode 100644 features/adaptive_context_strategy.feature create mode 100644 features/steps/adaptive_context_strategy_steps.py create mode 100644 src/cleveragents/domain/models/acms/adaptive_selector.py diff --git a/features/adaptive_context_strategy.feature b/features/adaptive_context_strategy.feature new file mode 100644 index 000000000..353975739 --- /dev/null +++ b/features/adaptive_context_strategy.feature @@ -0,0 +1,185 @@ +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 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 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: "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" diff --git a/features/steps/adaptive_context_strategy_steps.py b/features/steps/adaptive_context_strategy_steps.py new file mode 100644 index 000000000..9e16ce9ab --- /dev/null +++ b/features/steps/adaptive_context_strategy_steps.py @@ -0,0 +1,545 @@ +"""Step definitions for adaptive context strategy selector and fusion tests.""" + +from __future__ import annotations + +from typing import Any, Dict, List, Optional, Tuple + +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.strategy import ContextStrategy, StrategyResult + + +class MockStrategy(ContextStrategy): + """Mock strategy for testing.""" + + def __init__(self, name: str) -> None: + """Initialize mock strategy.""" + self.name = name + + def execute(self, *args: Any, **kwargs: Any) -> StrategyResult: + """Execute mock strategy.""" + return StrategyResult(ranked_files=[]) + + +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 strategies.split(", "): + 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.""" + plan_type_enum = PlanType(plan_type) + + config_data: Dict[str, Any] = {} + for row in context.table: + key = row["key"] if "key" in row.headings else row.headings[0] + value = row[key] if key in row.headings else row[row.headings[0]] + + 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 = [s.strip('"') for s in strategies.split(", ")] + 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, + ) + + +@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}: 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.""" + weights = {} + for row in context.table: + strategy = row["strategy"] if "strategy" in row.headings else row.headings[0] + weight = float(row[strategy] if strategy in row.headings else row[row.headings[1]]) + weights[strategy] = weight + + 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 = {} + + 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 = [f.strip('"') for f in files.split(", ")] + 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 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.""" + for row in context.table: + strategy = row["strategy"] if "strategy" in row.headings else row.headings[0] + expected = float(row[strategy] if strategy in row.headings else row[row.headings[1]]) + 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.""" + for row in context.table: + key = row["key"] if "key" in row.headings else row.headings[0] + value = row[key] if key in row.headings else row[row.headings[1]] + + if key == "num_strategies" or key == "num_files": + expected = int(value) + actual = context.fused_result.fusion_metadata[key] + 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 plan_types.split(", "): + 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}"') +def step_verify_plan_types(context: Any, plan_types: str) -> None: + """Verify plan types.""" + expected = [PlanType(pt.strip()) for pt in plan_types.split(", ")] + 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.""" + plan_type_enum = PlanType(plan_type) + context.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.config is not None + assert context.config.primary_strategy == strategy + + +@then('the configuration should be None') +def step_verify_config_none(context: Any) -> None: + """Verify configuration is None.""" + assert context.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 = [pt.strip() for pt in plan_types.split(", ")] + actual = [pt.value for pt in PlanType] + assert actual == expected 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..232f5bc20 --- /dev/null +++ b/src/cleveragents/domain/models/acms/adaptive_selector.py @@ -0,0 +1,332 @@ +"""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 dataclasses import dataclass, field +from enum import Enum +from typing import Any, Dict, List, Optional, Tuple + +from cleveragents.domain.models.acms.strategy import ContextStrategy, StrategyResult + + +class PlanType(str, Enum): + """Enumeration of plan types for strategy selection.""" + + CODING = "coding" + ANALYSIS = "analysis" + DOCUMENTATION = "documentation" + REFACTORING = "refactoring" + TESTING = "testing" + DEBUGGING = "debugging" + UNKNOWN = "unknown" + + +@dataclass +class StrategyWeight: + """Configuration for a strategy weight in fusion.""" + + strategy_name: str + weight: float = 1.0 + enabled: bool = True + + def __post_init__(self) -> None: + """Validate weight is positive.""" + if self.weight <= 0: + raise ValueError(f"Weight must be positive, got {self.weight}") + + +@dataclass +class AdaptiveStrategyConfig: + """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 + + def __post_init__(self) -> None: + """Validate configuration.""" + if not self.primary_strategy: + raise ValueError("primary_strategy must be specified") + + +@dataclass +class FusedResult: + """Result of context fusion combining multiple strategy results.""" + + ranked_files: List[Tuple[str, float]] # (file_path, combined_score) + strategy_contributions: Dict[str, List[Tuple[str, float]]] # strategy -> [(file, score)] + 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) -> Optional[float]: + """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) -> Optional[AdaptiveStrategyConfig]: + """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, StrategyResult], + custom_weights: Optional[Dict[str, float]] = 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_scores: Dict[str, List[Tuple[str, float]]] = {} # file -> [(strategy, score)] + 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 + num_strategies = len(list(strategy_names)) + return {name: 1.0 / num_strategies 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, StrategyResult], + ) -> 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) -- 2.52.0 From da32f49b7a5899bc116e82a253e49676c280128e Mon Sep 17 00:00:00 2001 From: HAL9000 Date: Thu, 23 Apr 2026 12:21:23 +0000 Subject: [PATCH 2/6] fix(context): resolve lint, typecheck, and unit test failures in adaptive selector - Replace deprecated typing.Dict/List/Tuple/Optional with built-in types - Replace str+Enum with StrEnum for PlanType - Replace Optional[X] with X | None syntax - Remove non-existent StrategyResult import; use Any for strategy results - Fix MockStrategy to properly implement ContextStrategy protocol - Fix ambiguous Behave step definitions (plan types vs files, score steps) - Fix trailing whitespace on blank lines - Fix line length violations --- features/adaptive_context_strategy.feature | 6 +- .../steps/adaptive_context_strategy_steps.py | 200 +++++++++++++----- .../domain/models/acms/adaptive_selector.py | 58 ++--- 3 files changed, 178 insertions(+), 86 deletions(-) diff --git a/features/adaptive_context_strategy.feature b/features/adaptive_context_strategy.feature index 353975739..03194c464 100644 --- a/features/adaptive_context_strategy.feature +++ b/features/adaptive_context_strategy.feature @@ -96,14 +96,14 @@ Feature: Adaptive Context Strategy Selector and Context Fusion | file1.py | 1.5 | | file2.py | 0.6 | When I get the score for "file1.py" - Then the score should be 1.5 + 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 score should be None + Then the file score should be None Scenario: Reject fusion with no results Given I have registered configuration for plan type "coding" @@ -142,7 +142,7 @@ Feature: Adaptive Context Strategy Selector and Context Fusion 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: "coding", "analysis", "testing" + 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" diff --git a/features/steps/adaptive_context_strategy_steps.py b/features/steps/adaptive_context_strategy_steps.py index 9e16ce9ab..7fef9f4ad 100644 --- a/features/steps/adaptive_context_strategy_steps.py +++ b/features/steps/adaptive_context_strategy_steps.py @@ -2,7 +2,7 @@ from __future__ import annotations -from typing import Any, Dict, List, Optional, Tuple +from typing import Any from behave import given, then, when @@ -14,25 +14,64 @@ from cleveragents.domain.models.acms.adaptive_selector import ( PlanType, StrategyWeight, ) -from cleveragents.domain.models.acms.strategy import ContextStrategy, StrategyResult +from cleveragents.domain.models.acms.crp import ContextFragment +from cleveragents.domain.models.acms.strategy import ( + BackendSet, + ContextRequest, + ContextStrategy, + PlanContext, + StrategyCapabilities, +) -class MockStrategy(ContextStrategy): +class MockStrategy: """Mock strategy for testing.""" - def __init__(self, name: str) -> None: + def __init__(self, strategy_name: str) -> None: """Initialize mock strategy.""" - self.name = name + self._name = strategy_name - def execute(self, *args: Any, **kwargs: Any) -> StrategyResult: + @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 StrategyResult(ranked_files=[]) + 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: + def __init__(self, ranked_files: list[tuple[str, float]]) -> None: """Initialize mock result.""" self.ranked_files = ranked_files @@ -72,7 +111,9 @@ def step_verify_strategy_in_list(context: Any, name: str) -> None: assert name in strategies -@when('I register configuration for plan type "{plan_type}" with primary strategy "{strategy}"') +@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) @@ -125,12 +166,12 @@ def step_register_multiple_strategies(context: Any, strategies: str) -> None: def step_register_config_with_table(context: Any, plan_type: str) -> None: """Register configuration with table data.""" plan_type_enum = PlanType(plan_type) - - config_data: Dict[str, Any] = {} + + config_data: dict[str, Any] = {} for row in context.table: key = row["key"] if "key" in row.headings else row.headings[0] value = row[key] if key in row.headings else row[row.headings[0]] - + if key == "fallback_strategies": config_data["fallback_strategies"] = [s.strip() for s in value.split(",")] elif key == "fusion_weights": @@ -141,7 +182,7 @@ def step_register_config_with_table(context: Any, plan_type: str) -> None: config_data["fusion_weights"] = weights else: config_data[key] = value - + config = AdaptiveStrategyConfig( plan_type=plan_type_enum, primary_strategy=config_data.get("primary_strategy", ""), @@ -178,14 +219,16 @@ def step_try_duplicate_registration(context: Any, name: str) -> None: context.error = str(e) -@then('I should get an error about duplicate registration') +@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}"') +@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: @@ -199,14 +242,16 @@ def step_try_unregistered_primary(context: Any, strategy: str) -> None: context.error = str(e) -@then('I should get an error about unregistered strategy') +@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}"') +@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: @@ -221,26 +266,26 @@ def step_try_unregistered_fallback(context: Any, primary: str, fallback: str) -> context.error = str(e) -@then('I should get an error about unregistered fallback strategy') +@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:') +@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) @@ -254,7 +299,7 @@ def step_fuse_equal_weights(context: Any, plan_type: str) -> None: ) -@then('the fused result should have ranked files:') +@then("the fused result should have ranked files:") def step_verify_ranked_files(context: Any) -> None: """Verify ranked files in fused result.""" expected_files = {} @@ -262,23 +307,27 @@ def step_verify_ranked_files(context: Any) -> None: 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}: expected {expected_score}, got {actual_score}" + 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:') +@when("I fuse the results with custom weights:") def step_fuse_custom_weights(context: Any) -> None: """Fuse results with custom weights.""" weights = {} for row in context.table: strategy = row["strategy"] if "strategy" in row.headings else row.headings[0] - weight = float(row[strategy] if strategy in row.headings else row[row.headings[1]]) + weight = float( + row[strategy] if strategy in row.headings else row[row.headings[1]] + ) weights[strategy] = weight - + plan_type_enum = PlanType.CODING context.fused_result = context.fusion.fuse_results( plan_type_enum, @@ -287,24 +336,24 @@ def step_fuse_custom_weights(context: Any) -> None: ) -@given('I have a fused result with ranked files:') +@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 = {} - + 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') +@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) @@ -323,7 +372,7 @@ def step_get_file_score(context: Any, file_path: str) -> None: context.file_score = context.fused_result.get_file_score(file_path) -@then('the score should be {value}') +@then("the file score should be {value}") def step_verify_file_score(context: Any, value: str) -> None: """Verify file score.""" if value == "None": @@ -333,7 +382,7 @@ def step_verify_file_score(context: Any, value: str) -> None: assert abs(context.file_score - expected) < 0.0001 -@when('I try to fuse with empty results') +@when("I try to fuse with empty results") def step_try_fuse_empty(context: Any) -> None: """Try to fuse with empty results.""" try: @@ -344,7 +393,7 @@ def step_try_fuse_empty(context: Any) -> None: context.error = str(e) -@then('I should get an error about no results provided') +@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 @@ -375,53 +424,58 @@ def step_try_negative_weight(context: Any, strategy: str) -> None: context.error = str(e) -@then('I should get an error about invalid weight') +@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}') +@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:') +@then("the normalized weights should be:") def step_verify_normalized_weights(context: Any) -> None: """Verify normalized weights.""" for row in context.table: strategy = row["strategy"] if "strategy" in row.headings else row.headings[0] - expected = float(row[strategy] if strategy in row.headings else row[row.headings[1]]) + expected = float( + row[strategy] if strategy in row.headings else row[row.headings[1]] + ) actual = context.normalized[strategy] - assert abs(actual - expected) < 0.0001, \ + assert abs(actual - expected) < 0.0001, ( f"Weight mismatch for {strategy}: expected {expected}, got {actual}" + ) -@then('the fusion metadata should contain:') +@then("the fusion metadata should contain:") def step_verify_fusion_metadata(context: Any) -> None: """Verify fusion metadata.""" for row in context.table: key = row["key"] if "key" in row.headings else row.headings[0] value = row[key] if key in row.headings else row[row.headings[1]] - + if key == "num_strategies" or key == "num_files": - expected = int(value) + expected: Any = int(value) actual = context.fused_result.fusion_metadata[key] else: expected = value actual = context.fused_result.fusion_metadata[key] - - assert actual == expected, f"Metadata mismatch for {key}: expected {expected}, got {actual}" + + assert actual == expected, ( + f"Metadata mismatch for {key}: expected {expected}, got {actual}" + ) @given('I have registered configuration for plan types: "{plan_types}"') @@ -431,7 +485,7 @@ def step_register_multiple_configs(context: Any, plan_types: str) -> None: 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", @@ -439,13 +493,13 @@ def step_register_multiple_configs(context: Any, plan_types: str) -> None: context.selector.register_config(config) -@when('I list all configured plan types') +@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}"') +@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.strip()) for pt in plan_types.split(", ")] @@ -466,13 +520,13 @@ def step_verify_config_primary(context: Any, strategy: str) -> None: assert context.config.primary_strategy == strategy -@then('the configuration should be None') +@then("the configuration should be None") def step_verify_config_none(context: Any) -> None: """Verify configuration is None.""" assert context.config is None -@when('I fuse with selector configuration weights') +@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 @@ -500,14 +554,14 @@ def step_have_result_without_ranked_files(context: Any, strategy: str) -> None: } -@then('the fusion should skip the strategy 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') +@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: @@ -517,7 +571,7 @@ def step_try_negative_strategy_weight(context: Any) -> None: context.error = str(e) -@when('I try to create adaptive strategy config without primary strategy') +@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: @@ -530,7 +584,7 @@ def step_try_config_no_primary(context: Any) -> None: context.error = str(e) -@then('I should get an error about missing primary strategy') +@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 @@ -543,3 +597,37 @@ def step_verify_plan_type_enum(context: Any, plan_types: str) -> None: expected = [pt.strip() for pt in plan_types.split(", ")] actual = [pt.value for pt in PlanType] assert actual == expected + + +@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) + + +@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.""" + plan_type_enum = PlanType(plan_type) + 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 index 232f5bc20..ed4189a41 100644 --- a/src/cleveragents/domain/models/acms/adaptive_selector.py +++ b/src/cleveragents/domain/models/acms/adaptive_selector.py @@ -7,13 +7,13 @@ context fusion to combine results from multiple strategies with configurable wei from __future__ import annotations from dataclasses import dataclass, field -from enum import Enum -from typing import Any, Dict, List, Optional, Tuple +from enum import StrEnum +from typing import Any -from cleveragents.domain.models.acms.strategy import ContextStrategy, StrategyResult +from cleveragents.domain.models.acms.strategy import ContextStrategy -class PlanType(str, Enum): +class PlanType(StrEnum): """Enumeration of plan types for strategy selection.""" CODING = "coding" @@ -45,8 +45,8 @@ class AdaptiveStrategyConfig: plan_type: PlanType primary_strategy: str - fallback_strategies: List[str] = field(default_factory=list) - fusion_weights: Dict[str, float] = field(default_factory=dict) + fallback_strategies: list[str] = field(default_factory=list) + fusion_weights: dict[str, float] = field(default_factory=dict) use_fusion: bool = False def __post_init__(self) -> None: @@ -59,11 +59,12 @@ class AdaptiveStrategyConfig: class FusedResult: """Result of context fusion combining multiple strategy results.""" - ranked_files: List[Tuple[str, float]] # (file_path, combined_score) - strategy_contributions: Dict[str, List[Tuple[str, float]]] # strategy -> [(file, score)] - fusion_metadata: Dict[str, Any] = field(default_factory=dict) + 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]: + def get_top_files(self, limit: int = 10) -> list[str]: """Get top N files from fused results. Args: @@ -74,7 +75,7 @@ class FusedResult: """ return [file_path for file_path, _ in self.ranked_files[:limit]] - def get_file_score(self, file_path: str) -> Optional[float]: + def get_file_score(self, file_path: str) -> float | None: """Get combined score for a specific file. Args: @@ -94,8 +95,8 @@ class AdaptiveContextSelector: def __init__(self) -> None: """Initialize the adaptive selector.""" - self._strategy_registry: Dict[str, ContextStrategy] = {} - self._config_map: Dict[PlanType, AdaptiveStrategyConfig] = {} + 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. @@ -149,7 +150,7 @@ class AdaptiveContextSelector: config = self._config_map[plan_type] return self._strategy_registry[config.primary_strategy] - def select_strategies(self, plan_type: PlanType) -> List[ContextStrategy]: + def select_strategies(self, plan_type: PlanType) -> list[ContextStrategy]: """Select all applicable strategies for a plan type (primary + fallbacks). Args: @@ -165,7 +166,7 @@ class AdaptiveContextSelector: raise ValueError(f"No configuration for plan type: {plan_type}") config = self._config_map[plan_type] - strategies: List[ContextStrategy] = [ + strategies: list[ContextStrategy] = [ self._strategy_registry[config.primary_strategy] ] @@ -174,7 +175,7 @@ class AdaptiveContextSelector: return strategies - def get_config(self, plan_type: PlanType) -> Optional[AdaptiveStrategyConfig]: + def get_config(self, plan_type: PlanType) -> AdaptiveStrategyConfig | None: """Get configuration for a plan type. Args: @@ -185,7 +186,7 @@ class AdaptiveContextSelector: """ return self._config_map.get(plan_type) - def list_registered_strategies(self) -> List[str]: + def list_registered_strategies(self) -> list[str]: """List all registered strategy names. Returns: @@ -193,7 +194,7 @@ class AdaptiveContextSelector: """ return list(self._strategy_registry.keys()) - def list_configured_plan_types(self) -> List[PlanType]: + def list_configured_plan_types(self) -> list[PlanType]: """List all plan types with configurations. Returns: @@ -216,8 +217,8 @@ class ContextFusion: def fuse_results( self, plan_type: PlanType, - strategy_results: Dict[str, StrategyResult], - custom_weights: Optional[Dict[str, float]] = None, + strategy_results: dict[str, Any], + custom_weights: dict[str, float] | None = None, ) -> FusedResult: """Fuse results from multiple strategies. @@ -246,8 +247,9 @@ class ContextFusion: normalized_weights = self._normalize_weights(weights, strategy_results.keys()) # Collect all files and their scores from each strategy - file_scores: Dict[str, List[Tuple[str, float]]] = {} # file -> [(strategy, score)] - strategy_contributions: Dict[str, List[Tuple[str, float]]] = {} + # 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] = [] @@ -263,10 +265,12 @@ class ContextFusion: weighted_score = score * weight file_scores[file_path].append((strategy_name, weighted_score)) - strategy_contributions[strategy_name].append((file_path, weighted_score)) + strategy_contributions[strategy_name].append( + (file_path, weighted_score) + ) # Combine scores for each file - combined_scores: List[Tuple[str, float]] = [] + 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)) @@ -286,8 +290,8 @@ class ContextFusion: ) def _normalize_weights( - self, weights: Dict[str, float], strategy_names: Any - ) -> Dict[str, float]: + self, weights: dict[str, float], strategy_names: Any + ) -> dict[str, float]: """Normalize weights to sum to 1.0. Args: @@ -314,7 +318,7 @@ class ContextFusion: def fuse_with_selector( self, plan_type: PlanType, - strategy_results: Dict[str, StrategyResult], + strategy_results: dict[str, Any], ) -> FusedResult: """Fuse results using weights from selector configuration. -- 2.52.0 From a990b935df7743a1f91434764f9ab486d885b76c Mon Sep 17 00:00:00 2001 From: CleverThis Date: Thu, 4 Jun 2026 14:48:46 -0400 Subject: [PATCH 3/6] fix(acms): repair adaptive-context step definitions and equal-weights fusion The adaptive_context_strategy.feature suite was failing on 13 scenarios (2 failed, 11 errored) and ruff format was rejecting the step file: * step_register_config_with_table, step_fuse_custom_weights, step_verify_normalized_weights, and step_verify_fusion_metadata read no-header 2-column Gherkin tables as if they had key/value headers; behave promotes the first row to headings, so the first key/value pair was lost and the second-row lookups erroneously fed table data through float() / dict keys. Added a _table_pairs helper that recovers the promoted-heading pair and iterates the remaining rows. * step_register_multiple_strategies, step_register_multiple_configs, step_verify_plan_types, and step_verify_plan_type_enum captured the inner quotes of multi-token quoted-CSV placeholders (e.g. '"coding"' vs 'coding'). Added _strip_quoted_csv to normalise them. * step_have_registered_config validated against the strategy registry but never registered the strategy it was passed; the "Get configuration for plan type" scenario calls it without a prior registration. Auto-register on first use. * step_get_config wrote to context.config, which behave reserves for its own runtime configuration object; the assignment raised KeyError. Renamed to context.fetched_config. * No When step matched the bare 'I fuse the results for plan type "{plan_type}"' (scenarios 127/169). Added the matching step. * ContextFusion._normalize_weights returned 1/N when no weights were supplied; the "equal weights" scenarios pin the semantics to unscaled 1.0-per-strategy. Switched the empty-weights branch accordingly. Explicit non-empty weights still normalise to sum 1.0 so the custom-weights and selector-weights scenarios continue to produce the same scores. * Reformatted the over-wrapped @when decorator on step_try_unregistered_primary to satisfy ruff format. ISSUES CLOSED: #5255 --- .../steps/adaptive_context_strategy_steps.py | 112 +++++++++++------- .../domain/models/acms/adaptive_selector.py | 8 +- 2 files changed, 73 insertions(+), 47 deletions(-) diff --git a/features/steps/adaptive_context_strategy_steps.py b/features/steps/adaptive_context_strategy_steps.py index 7fef9f4ad..4fefe505a 100644 --- a/features/steps/adaptive_context_strategy_steps.py +++ b/features/steps/adaptive_context_strategy_steps.py @@ -24,6 +24,31 @@ from cleveragents.domain.models.acms.strategy import ( ) +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.""" @@ -157,21 +182,18 @@ def step_verify_selected_strategy(context: Any, name: str) -> None: @given('I have registered strategies: "{strategies}"') def step_register_multiple_strategies(context: Any, strategies: str) -> None: """Register multiple strategies.""" - for strategy_name in strategies.split(", "): + 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.""" + """Register configuration with table data (headerless 2-column table).""" plan_type_enum = PlanType(plan_type) config_data: dict[str, Any] = {} - for row in context.table: - key = row["key"] if "key" in row.headings else row.headings[0] - value = row[key] if key in row.headings else row[row.headings[0]] - + 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": @@ -202,7 +224,7 @@ def step_select_all_strategies(context: Any, plan_type: str) -> None: @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 = [s.strip('"') for s in strategies.split(", ")] + 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 @@ -226,9 +248,7 @@ def step_verify_duplicate_error(context: Any) -> None: assert "already registered" in context.error -@when( - 'I try to register configuration with unregistered primary strategy "{strategy}"' -) +@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: @@ -299,6 +319,16 @@ def step_fuse_equal_weights(context: Any, plan_type: str) -> None: ) +@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.""" @@ -319,14 +349,10 @@ def step_verify_ranked_files(context: Any) -> None: @when("I fuse the results with custom weights:") def step_fuse_custom_weights(context: Any) -> None: - """Fuse results with custom weights.""" - weights = {} - for row in context.table: - strategy = row["strategy"] if "strategy" in row.headings else row.headings[0] - weight = float( - row[strategy] if strategy in row.headings else row[row.headings[1]] - ) - weights[strategy] = weight + """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( @@ -362,7 +388,7 @@ def step_get_top_files(context: Any, count: int) -> None: @then('I should get: "{files}"') def step_verify_top_files(context: Any, files: str) -> None: """Verify top files.""" - expected = [f.strip('"') for f in files.split(", ")] + expected = _strip_quoted_csv(files) assert context.top_files == expected @@ -447,12 +473,9 @@ def step_normalize_weights(context: Any, weights: str) -> None: @then("the normalized weights should be:") def step_verify_normalized_weights(context: Any) -> None: - """Verify normalized weights.""" - for row in context.table: - strategy = row["strategy"] if "strategy" in row.headings else row.headings[0] - expected = float( - row[strategy] if strategy in row.headings else row[row.headings[1]] - ) + """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}" @@ -461,18 +484,13 @@ def step_verify_normalized_weights(context: Any) -> None: @then("the fusion metadata should contain:") def step_verify_fusion_metadata(context: Any) -> None: - """Verify fusion metadata.""" - for row in context.table: - key = row["key"] if "key" in row.headings else row.headings[0] - value = row[key] if key in row.headings else row[row.headings[1]] - - if key == "num_strategies" or key == "num_files": + """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) - actual = context.fused_result.fusion_metadata[key] else: expected = value - actual = context.fused_result.fusion_metadata[key] - + actual = context.fused_result.fusion_metadata[key] assert actual == expected, ( f"Metadata mismatch for {key}: expected {expected}, got {actual}" ) @@ -481,7 +499,7 @@ def step_verify_fusion_metadata(context: Any) -> None: @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 plan_types.split(", "): + 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) @@ -502,28 +520,32 @@ def step_list_plan_types(context: Any) -> None: @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.strip()) for pt in plan_types.split(", ")] + 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.""" + """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.config = context.selector.get_config(plan_type_enum) + 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.config is not None - assert context.config.primary_strategy == 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.config is None + assert context.fetched_config is None @when("I fuse with selector configuration weights") @@ -594,9 +616,9 @@ def step_verify_missing_primary_error(context: Any) -> None: @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 = [pt.strip() for pt in plan_types.split(", ")] + expected = _strip_quoted_csv(plan_types) actual = [pt.value for pt in PlanType] - assert actual == expected + assert actual == expected, f"PlanType mismatch: expected {expected}, got {actual}" @given('I have registered a strategy named "{name}"') @@ -610,8 +632,10 @@ def step_have_registered_strategy(context: Any, name: str) -> None: '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.""" + """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, diff --git a/src/cleveragents/domain/models/acms/adaptive_selector.py b/src/cleveragents/domain/models/acms/adaptive_selector.py index ed4189a41..f76827050 100644 --- a/src/cleveragents/domain/models/acms/adaptive_selector.py +++ b/src/cleveragents/domain/models/acms/adaptive_selector.py @@ -302,9 +302,11 @@ class ContextFusion: Normalized weights dictionary """ if not weights: - # Equal weights for all strategies - num_strategies = len(list(strategy_names)) - return {name: 1.0 / num_strategies for name in strategy_names} + # 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(): -- 2.52.0 From 590c11a3fc9b6d3c9ffdeaeb4ce14da1fb449e76 Mon Sep 17 00:00:00 2001 From: controller-ci-rerun Date: Thu, 4 Jun 2026 15:24:26 -0400 Subject: [PATCH 4/6] chore: re-trigger CI [controller] -- 2.52.0 From 17f1e2c55510ea3fef6eec5687ea8dc3aa48436a Mon Sep 17 00:00:00 2001 From: CleverThis Date: Mon, 15 Jun 2026 03:55:31 -0400 Subject: [PATCH 5/6] fix(acms): convert adaptive_selector dataclasses to Pydantic BaseModel Architecture test requires all @dataclass-decorated classes to inherit from Pydantic BaseModel. Replace StrategyWeight, AdaptiveStrategyConfig, and FusedResult plain dataclasses with BaseModel subclasses, using Field(default_factory=...) for mutable defaults and @field_validator for validation logic that was previously in __post_init__. ISSUES CLOSED: #5255 --- .../domain/models/acms/adaptive_selector.py | 36 ++++++++++--------- 1 file changed, 19 insertions(+), 17 deletions(-) diff --git a/src/cleveragents/domain/models/acms/adaptive_selector.py b/src/cleveragents/domain/models/acms/adaptive_selector.py index f76827050..e1ce109ca 100644 --- a/src/cleveragents/domain/models/acms/adaptive_selector.py +++ b/src/cleveragents/domain/models/acms/adaptive_selector.py @@ -6,10 +6,11 @@ context fusion to combine results from multiple strategies with configurable wei from __future__ import annotations -from dataclasses import dataclass, field from enum import StrEnum from typing import Any +from pydantic import BaseModel, Field, field_validator + from cleveragents.domain.models.acms.strategy import ContextStrategy @@ -25,44 +26,45 @@ class PlanType(StrEnum): UNKNOWN = "unknown" -@dataclass -class StrategyWeight: +class StrategyWeight(BaseModel): """Configuration for a strategy weight in fusion.""" strategy_name: str weight: float = 1.0 enabled: bool = True - def __post_init__(self) -> None: - """Validate weight is positive.""" - if self.weight <= 0: - raise ValueError(f"Weight must be positive, got {self.weight}") + @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 -@dataclass -class AdaptiveStrategyConfig: +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) + fallback_strategies: list[str] = Field(default_factory=list) + fusion_weights: dict[str, float] = Field(default_factory=dict) use_fusion: bool = False - def __post_init__(self) -> None: - """Validate configuration.""" - if not self.primary_strategy: + @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 -@dataclass -class FusedResult: +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) + 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. -- 2.52.0 From 21ba6dbc9eda5f22f2ab05662c1829d7f249f4c8 Mon Sep 17 00:00:00 2001 From: CleverThis Date: Mon, 15 Jun 2026 04:25:52 -0400 Subject: [PATCH 6/6] test(acms): cover adaptive_selector error paths and StrategyWeight validator Add five behave scenarios exercising the previously-uncovered error branches in AdaptiveContextSelector (select_strategy and select_strategies with an unconfigured plan_type) and ContextFusion (fuse_results and fuse_with_selector with an unconfigured plan_type), plus a scenario constructing a valid-weight StrategyWeight to cover the success path of the field_validator. ISSUES CLOSED: #5255 --- features/adaptive_context_strategy.feature | 26 ++++++++ .../steps/adaptive_context_strategy_steps.py | 60 +++++++++++++++++++ 2 files changed, 86 insertions(+) diff --git a/features/adaptive_context_strategy.feature b/features/adaptive_context_strategy.feature index 03194c464..616e48083 100644 --- a/features/adaptive_context_strategy.feature +++ b/features/adaptive_context_strategy.feature @@ -183,3 +183,29 @@ Feature: Adaptive Context Strategy Selector and Context Fusion 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 index 4fefe505a..d190c88f3 100644 --- a/features/steps/adaptive_context_strategy_steps.py +++ b/features/steps/adaptive_context_strategy_steps.py @@ -628,6 +628,66 @@ def step_have_registered_strategy(context: Any, name: str) -> None: 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}"' ) -- 2.52.0