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():