"""Step definitions for features/acms_service_coverage_boost.feature. Targets uncovered lines in src/cleveragents/application/services/acms_service.py: - Line 413: DefaultBudgetAllocator.allocate with empty candidates - Line 436: DefaultBudgetAllocator.allocate remainder token distribution - Line 463: DefaultStrategyExecutor.execute with empty allocations - Line 467: DefaultStrategyExecutor.execute skipping zero-token allocations - Lines 616-620: ACMSPipeline.__init__ unknown default_strategy ValueError """ from __future__ import annotations from collections.abc import Sequence from typing import Any from behave import given, then, when from behave.runner import Context from cleveragents.application.services.acms_service import ( ACMSPipeline, DefaultBudgetAllocator, DefaultStrategyExecutor, RecencyStrategy, RelevanceStrategy, StrategyCapabilities, TieredStrategy, ) from cleveragents.domain.models.core.context_fragment import ( ContextBudget, ContextFragment, FragmentProvenance, ) # --------------------------------------------------------------------------- # Helpers # --------------------------------------------------------------------------- _DEFAULT_PROVENANCE = FragmentProvenance(resource_uri="test://coverage-boost") def _make_frag( uko_node: str = "test://cov", content: str = "test", token_count: int = 10, relevance_score: float = 0.5, **kwargs: Any, ) -> ContextFragment: kwargs.setdefault("provenance", _DEFAULT_PROVENANCE) return ContextFragment( uko_node=uko_node, content=content, token_count=token_count, relevance_score=relevance_score, **kwargs, ) class _SimpleStrategy: """Minimal strategy for testing the executor.""" def __init__(self, name: str = "simple") -> None: self._name = name @property def name(self) -> str: return self._name @property def capabilities(self) -> StrategyCapabilities: return StrategyCapabilities(quality_score=1.0) def can_handle(self, request: dict[str, Any]) -> float: return 0.9 def assemble( self, fragments: Sequence[ContextFragment], budget: ContextBudget, ) -> Sequence[ContextFragment]: # Just return all fragments that fit. result: list[ContextFragment] = [] total = 0 for frag in fragments: if total + frag.token_count <= budget.available_tokens: result.append(frag) total += frag.token_count return result def explain(self) -> str: return "Simple test strategy." # --------------------------------------------------------------------------- # Given — DefaultBudgetAllocator # --------------------------------------------------------------------------- @given("the default budget allocator") def step_given_default_allocator(context: Context) -> None: context.allocator = DefaultBudgetAllocator() # --------------------------------------------------------------------------- # Given — DefaultStrategyExecutor # --------------------------------------------------------------------------- @given("the default strategy executor") def step_given_default_executor(context: Context) -> None: context.executor = DefaultStrategyExecutor() @given("a test strategy and fragments for execution") def step_given_test_strategy_and_fragments(context: Context) -> None: context.test_strategy = _SimpleStrategy("test_exec") context.exec_fragments = [ _make_frag(uko_node="test://a", content="alpha", token_count=10), _make_frag(uko_node="test://b", content="beta", token_count=20), ] context.exec_budget = ContextBudget(max_tokens=200, reserved_tokens=0) # --------------------------------------------------------------------------- # When — DefaultBudgetAllocator # --------------------------------------------------------------------------- @when("I allocate budget {budget:d} to an empty candidate list") def step_allocate_empty_candidates(context: Context, budget: int) -> None: """Exercises line 413: return [] for empty candidates.""" context.alloc_result = context.allocator.allocate([], budget) @when( "I allocate budget {budget:d} across two candidates with confidences {c1:g} and {c2:g}" ) def step_allocate_two_candidates_fractional( context: Context, budget: int, c1: float, c2: float ) -> None: """Exercises line 436: remainder distribution in proportional allocation. Budget 100 with confidences 0.5 and 0.3 (total 0.8): raw = [62.5, 37.5], floors = [62, 37], remainder = 1 -> line 436 awards the extra token to the first candidate. """ s1 = RelevanceStrategy() s2 = RecencyStrategy() context.alloc_result = context.allocator.allocate([(s1, c1), (s2, c2)], budget) @when( "I allocate budget {budget:d} across three candidates with confidences {conf_str}" ) def step_allocate_three_uneven(context: Context, budget: int, conf_str: str) -> None: """Exercises line 436 with three candidates and non-zero remainder.""" confidences = [float(c) for c in conf_str.split()] strategies = [RelevanceStrategy(), RecencyStrategy(), TieredStrategy()] candidates = [ (s, c) for s, c in zip(strategies[: len(confidences)], confidences, strict=True) ] context.alloc_result = context.allocator.allocate(candidates, budget) @when("I allocate budget {budget:d} across three zero-confidence candidates") def step_allocate_zero_conf_remainder(context: Context, budget: int) -> None: """Exercises lines 418-423: zero-confidence equal split with remainder. Budget 10, 3 candidates, all confidence 0.0: share = 10 // 3 = 3, remainder = 10 - 9 = 1 -> first candidate gets 3+1=4, others get 3. """ strategies = [RelevanceStrategy(), RecencyStrategy(), TieredStrategy()] candidates = [(s, 0.0) for s in strategies] context.alloc_result = context.allocator.allocate(candidates, budget) # --------------------------------------------------------------------------- # When — DefaultStrategyExecutor # --------------------------------------------------------------------------- @when("I execute with an empty allocations list") def step_execute_empty_allocations(context: Context) -> None: """Exercises line 463: return [] for empty allocations.""" dummy_budget = ContextBudget(max_tokens=100, reserved_tokens=0) context.exec_result = context.executor.execute([], [], dummy_budget) @when("I execute with a zero-token allocation for the test strategy") def step_execute_zero_token(context: Context) -> None: """Exercises line 467: continue when allocated_tokens <= 0.""" allocations = [(context.test_strategy, 0.8, 0)] context.exec_result = context.executor.execute( allocations, context.exec_fragments, context.exec_budget ) @when("I execute with mixed allocations including zero and positive tokens") def step_execute_mixed_allocations(context: Context) -> None: """Exercises line 467 (skip) and line 472 (process positive).""" zero_strategy = _SimpleStrategy("zero_strat") positive_strategy = _SimpleStrategy("pos_strat") allocations = [ (zero_strategy, 0.5, 0), # should be skipped (line 467) (positive_strategy, 0.8, 100), # should be processed ] context.exec_result = context.executor.execute( allocations, context.exec_fragments, context.exec_budget ) # --------------------------------------------------------------------------- # When — ACMSPipeline.__init__ unknown default_strategy # --------------------------------------------------------------------------- @when('I create a pipeline with default strategy "{strategy}"') def step_create_pipeline_unknown_strategy(context: Context, strategy: str) -> None: """Exercises lines 616-620: ValueError for unknown default_strategy.""" context.pipeline_error = None try: context.test_pipeline = ACMSPipeline(default_strategy=strategy) except ValueError as exc: context.pipeline_error = exc @when("I create a pipeline with an empty default strategy") def step_create_pipeline_empty_strategy(context: Context) -> None: """Exercises lines 616-620: ValueError for empty string default_strategy.""" context.pipeline_error = None try: context.test_pipeline = ACMSPipeline(default_strategy="") except ValueError as exc: context.pipeline_error = exc # --------------------------------------------------------------------------- # When/Given — Full pipeline with multi-candidate selector # --------------------------------------------------------------------------- class _MultiCandidateSelector: """Returns multiple candidates to force proportional allocation.""" def select( self, strategies: Sequence[Any], request: dict[str, Any], ) -> list[tuple[Any, float]]: # Return first two strategies with different confidences # so the allocator exercises the remainder path. result = [] for s in strategies[:2]: conf = 0.7 if s.name == "relevance" else 0.3 result.append((s, conf)) return result @given("a pipeline with a custom selector that returns multiple candidates") def step_pipeline_multi_candidate_selector(context: Context) -> None: context.pipeline = ACMSPipeline(strategy_selector=_MultiCandidateSelector()) @given("a context budget with {max_tok:d} max tokens") def step_budget_simple(context: Context, max_tok: int) -> None: context.budget = ContextBudget(max_tokens=max_tok, reserved_tokens=0) @given("simple test fragments totalling {total:d} tokens") def step_simple_fragments(context: Context, total: int) -> None: half = total // 2 context.fragments = [ _make_frag( uko_node="test://frag1", content="fragment one", token_count=half, relevance_score=0.9, ), _make_frag( uko_node="test://frag2", content="fragment two", token_count=total - half, relevance_score=0.7, ), ] @when("I assemble the fragments through the full pipeline") def step_assemble_full_pipeline(context: Context) -> None: context.assemble_error = None try: context.payload = context.pipeline.assemble( plan_id="01JQTESTPN00000000000000AA", fragments=context.fragments, budget=context.budget, strategy="relevance", ) except Exception as exc: context.assemble_error = exc # --------------------------------------------------------------------------- # Then — DefaultBudgetAllocator assertions # --------------------------------------------------------------------------- @then("the allocation result should be an empty list") def step_alloc_result_empty(context: Context) -> None: assert context.alloc_result == [], ( f"Expected empty list, got {context.alloc_result}" ) @then("the total allocated tokens should equal exactly {budget:d}") def step_total_alloc_exact(context: Context, budget: int) -> None: total = sum(a[2] for a in context.alloc_result) assert total == budget, ( f"Expected total allocation {budget}, got {total}. " f"Allocations: {[a[2] for a in context.alloc_result]}" ) @then("each candidate should receive a positive allocation") def step_each_positive(context: Context) -> None: for _strategy, _confidence, tokens in context.alloc_result: assert tokens > 0, f"Expected positive allocation, got {tokens}" @then("allocations should be {high:d} or {low:d} tokens each") def step_allocs_bounded(context: Context, high: int, low: int) -> None: valid = {high, low} for _strategy, _confidence, tokens in context.alloc_result: assert tokens in valid, f"Expected allocation in {valid}, got {tokens}" # --------------------------------------------------------------------------- # Then — DefaultStrategyExecutor assertions # --------------------------------------------------------------------------- @then("the executor result should be an empty list") def step_exec_result_empty(context: Context) -> None: assert list(context.exec_result) == [], ( f"Expected empty result, got {list(context.exec_result)}" ) @then("the executor result should contain only fragments from the positive allocation") def step_exec_result_positive_only(context: Context) -> None: result = list(context.exec_result) assert len(result) > 0, "Expected non-empty result from positive allocation" # The positive strategy should have returned fragments assert len(result) == len(context.exec_fragments), ( f"Expected {len(context.exec_fragments)} fragments, got {len(result)}" ) # --------------------------------------------------------------------------- # Then — ACMSPipeline.__init__ error assertions # --------------------------------------------------------------------------- @then('the pipeline creation should fail with a ValueError about "{keyword}"') def step_pipeline_value_error_raised(context: Context, keyword: str) -> None: assert context.pipeline_error is not None, "Expected ValueError but none was raised" assert isinstance(context.pipeline_error, ValueError), ( f"Expected ValueError, got {type(context.pipeline_error).__name__}" ) assert keyword.lower() in str(context.pipeline_error).lower(), ( f"Expected '{keyword}' in error: {context.pipeline_error}" ) @then("the pipeline error message should list available strategies") def step_pipeline_error_lists_strategies(context: Context) -> None: msg = str(context.pipeline_error) # The error message should mention at least the built-in strategies for name in ("recency", "relevance", "tiered"): assert name in msg, f"Expected strategy '{name}' listed in error: {msg}" # --------------------------------------------------------------------------- # Then — Full pipeline assertions # --------------------------------------------------------------------------- @then("the assembly should succeed without error") def step_assembly_success(context: Context) -> None: assert context.assemble_error is None, ( f"Assembly failed with error: {context.assemble_error}" ) assert context.payload is not None, "Payload is None"