fix(acms): repair adaptive-context step definitions and equal-weights fusion
CI / push-validation (pull_request) Successful in 34s
CI / lint (pull_request) Successful in 42s
CI / helm (pull_request) Successful in 35s
CI / build (pull_request) Successful in 41s
CI / quality (pull_request) Successful in 49s
CI / typecheck (pull_request) Successful in 1m1s
CI / security (pull_request) Successful in 1m25s
CI / integration_tests (pull_request) Failing after 15m42s
CI / unit_tests (pull_request) Failing after 15m42s
CI / coverage (pull_request) Has been cancelled
CI / docker (pull_request) Has been cancelled
CI / status-check (pull_request) Has been cancelled

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
This commit is contained in:
2026-06-04 14:48:46 -04:00
parent 6e0b49d1be
commit e1bab8f419
2 changed files with 73 additions and 47 deletions
@@ -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,
@@ -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():