"""Step definitions for features/acms_fusion.feature. Tests the ACMS StrategyCoordinator and FusionEngine facades, covering parallel strategy coordination, budget allocation, dedup, depth resolution, greedy knapsack packing, and budget overage guard. """ 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_pipeline import ( CircuitBreaker, ParallelStrategyExecutor, ) from cleveragents.application.services.acms_service import StrategyCapabilities from cleveragents.application.services.fusion_engine import ( FusionConfig, FusionEngine, FusionResult, ) from cleveragents.application.services.strategy_coordinator import ( CoordinatorConfig, StrategyCoordinator, ) from cleveragents.domain.models.core.context_fragment import ( ContextBudget, ContextFragment, FragmentProvenance, ) # --------------------------------------------------------------------------- # Helpers # --------------------------------------------------------------------------- _DEFAULT_PROV = FragmentProvenance(resource_uri="test://fusion") def _make_frag(**kwargs: Any) -> ContextFragment: """Create a ContextFragment with test defaults.""" kwargs.setdefault("uko_node", "test://fusion/default") kwargs.setdefault("token_count", 10) kwargs.setdefault("provenance", _DEFAULT_PROV) return ContextFragment(**kwargs) class _FusionTestStrategy: """Test strategy with configurable confidence.""" def __init__(self, name: str = "test_strat", confidence: float = 0.7) -> None: self._name = name self._confidence = confidence @property def name(self) -> str: return self._name @property def capabilities(self) -> StrategyCapabilities: return StrategyCapabilities() def can_handle(self, request: dict[str, Any]) -> float: return self._confidence def assemble( self, fragments: Sequence[ContextFragment], budget: ContextBudget, ) -> Sequence[ContextFragment]: 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 f"Test strategy '{self._name}' with confidence {self._confidence}." # --------------------------------------------------------------------------- # Given steps # --------------------------------------------------------------------------- @given("the fusion modules are available") def step_fusion_modules_available(context: Context) -> None: """Ensure all fusion modules are importable.""" pass @given("fusion test strategies with confidences {c1:g} and {c2:g}") def step_fusion_strategies(context: Context, c1: float, c2: float) -> None: context.fusion_strategies = [ _FusionTestStrategy("strat_a", c1), _FusionTestStrategy("strat_b", c2), ] @given("fusion test fragments with token counts {t1:d} and {t2:d}") def step_fusion_fragments(context: Context, t1: int, t2: int) -> None: context.fusion_fragments = [ _make_frag( uko_node="project://fusion/a.py", content="alpha content", token_count=t1, relevance_score=0.9, ), _make_frag( uko_node="project://fusion/b.py", content="beta content", token_count=t2, relevance_score=0.5, ), ] @given('a coordinator with a pre-broken circuit for "{name}"') def step_coordinator_broken_circuit(context: Context, name: str) -> None: cb = CircuitBreaker(failure_threshold=1) cb.record_failure(name) executor = ParallelStrategyExecutor(circuit_breaker=cb) context.broken_coordinator = StrategyCoordinator(executor=executor) context.broken_strategy_name = name @given("a coordinator with per-strategy max cap of {cap:d}") def step_coordinator_with_cap(context: Context, cap: int) -> None: config = CoordinatorConfig(per_strategy_max_cap=cap) context.capped_coordinator = StrategyCoordinator(config=config) @given("fusion fragments with duplicates by URI and content") def step_fusion_dup_fragments(context: Context) -> None: context.fusion_dup_fragments = [ _make_frag( uko_node="project://fusion/dup.py", content="same content here", token_count=50, relevance_score=0.9, ), _make_frag( uko_node="project://fusion/dup.py", content="same content here", token_count=50, relevance_score=0.7, ), _make_frag( uko_node="project://fusion/other.py", content="different content", token_count=50, relevance_score=0.5, ), ] @given("fusion fragments with same UKO node at depths {d1:d} and {d2:d}") def step_fusion_depth_fragments(context: Context, d1: int, d2: int) -> None: context.fusion_depth_fragments = [ _make_frag( uko_node="project://fusion/deep.py", content="shallow content", token_count=50, detail_depth=d1, relevance_score=0.8, ), _make_frag( uko_node="project://fusion/deep.py", content="deep content with more detail", token_count=80, detail_depth=d2, relevance_score=0.7, ), _make_frag( uko_node="project://fusion/other.py", content="other content", token_count=50, detail_depth=3, relevance_score=0.6, ), ] @given("fusion fragments totaling {total:d} tokens") def step_fusion_total_fragments(context: Context, total: int) -> None: count = total // 100 if count < 1: count = 1 per_token = total // count context.fusion_total_fragments = [ _make_frag( uko_node=f"project://fusion/file{i}.py", content=f"content_{i}", token_count=per_token, relevance_score=round(0.9 - i * 0.1, 2), ) for i in range(count) ] @given("fusion fragments with equal relevance but different depths") def step_fusion_tiebreaker_fragments(context: Context) -> None: context.fusion_tiebreaker_fragments = [ _make_frag( uko_node="project://fusion/t1.py", content="content_t1", token_count=50, detail_depth=3, relevance_score=0.8, ), _make_frag( uko_node="project://fusion/t2.py", content="content_t2", token_count=30, detail_depth=7, relevance_score=0.8, ), _make_frag( uko_node="project://fusion/t3.py", content="content_t3", token_count=40, detail_depth=5, relevance_score=0.8, ), ] @given("a fusion engine with overage guard enabled") def step_fusion_engine_overage(context: Context) -> None: config = FusionConfig(overage_guard_enabled=True, min_fragment_tokens=1) context.overage_engine = FusionEngine(config=config) @given("fusion fragments that cause budget overage") def step_fusion_overage_fragments(context: Context) -> None: # Create fragments where packer might allow overage. # The overage guard is a safety net for when fragments pass through # the packer but still exceed budget. We test this by using a custom # FusionEngine with a pass-through packer that doesn't enforce budget. context.overage_fragments = [ _make_frag( uko_node=f"project://fusion/ov{i}.py", content=f"overage_content_{i}", token_count=40, relevance_score=round(0.9 - i * 0.2, 2), ) for i in range(4) ] @given("fusion fragments totaling exactly {total:d} tokens") def step_fusion_exact_fragments(context: Context, total: int) -> None: context.exact_fragments = [ _make_frag( uko_node="project://fusion/exact.py", content="exact content", token_count=total, relevance_score=0.9, ), ] # --------------------------------------------------------------------------- # When steps — StrategyCoordinator # --------------------------------------------------------------------------- @when("I coordinate with a budget of {budget:d} tokens") def step_coordinate(context: Context, budget: int) -> None: coordinator = StrategyCoordinator() b = ContextBudget(max_tokens=budget, reserved_tokens=0) context.coord_result = coordinator.coordinate( request={}, strategies=context.fusion_strategies, budget=b, fragments=context.fusion_fragments, ) @when("I coordinate with no strategies and a budget of {budget:d} tokens") def step_coordinate_no_strategies(context: Context, budget: int) -> None: coordinator = StrategyCoordinator() b = ContextBudget(max_tokens=budget, reserved_tokens=0) context.coord_result = coordinator.coordinate( request={}, strategies=[], budget=b, fragments=context.fusion_fragments, ) @when("I coordinate with the circuit-broken strategy") def step_coordinate_broken(context: Context) -> None: strategy = _FusionTestStrategy(context.broken_strategy_name, 0.7) b = ContextBudget(max_tokens=200, reserved_tokens=0) context.coord_result = context.broken_coordinator.coordinate( request={}, strategies=[strategy], budget=b, fragments=context.fusion_fragments, ) @when("I coordinate with a budget of {budget:d} tokens using the capped coordinator") def step_coordinate_capped(context: Context, budget: int) -> None: b = ContextBudget(max_tokens=budget, reserved_tokens=0) context.coord_result = context.capped_coordinator.coordinate( request={}, strategies=context.fusion_strategies, budget=b, fragments=context.fusion_fragments, ) # --------------------------------------------------------------------------- # When steps — FusionEngine # --------------------------------------------------------------------------- @when("I fuse with a budget of {budget:d} tokens") def step_fuse(context: Context, budget: int) -> None: engine = FusionEngine(config=FusionConfig(min_fragment_tokens=1)) b = ContextBudget(max_tokens=budget, reserved_tokens=0) # Use whichever fragments are available frags = getattr( context, "fusion_dup_fragments", getattr( context, "fusion_depth_fragments", getattr( context, "fusion_total_fragments", getattr(context, "fusion_tiebreaker_fragments", []), ), ), ) context.fusion_result = engine.fuse(frags, b) context.fusion_input_count = len(frags) @when("I fuse with the overage-prone fragments and a budget of {budget:d} tokens") def step_fuse_overage(context: Context, budget: int) -> None: b = ContextBudget(max_tokens=budget, reserved_tokens=0) # Test the overage guard directly since the packer already respects budget. # We call _budget_overage_guard on fragments that exceed the budget. surviving, dropped = FusionEngine._budget_overage_guard( list(context.overage_fragments), b ) total_tokens = sum(f.token_count for f in surviving) available = b.available_tokens context.fusion_result = FusionResult( fragments=surviving, input_count=len(context.overage_fragments), dedup_count=0, depth_resolved_count=0, packed_count=len(surviving), dropped_by_overage_guard=dropped, total_tokens=total_tokens, budget_utilization=round( min(total_tokens / available if available > 0 else 0.0, 1.0), 4 ), ) @when("I fuse with the exact fragments and a budget of {budget:d} tokens") def step_fuse_exact(context: Context, budget: int) -> None: b = ContextBudget(max_tokens=budget, reserved_tokens=0) context.fusion_result = context.overage_engine.fuse(context.exact_fragments, b) @when("I fuse with no fragments and a budget of {budget:d} tokens") def step_fuse_empty(context: Context, budget: int) -> None: engine = FusionEngine() b = ContextBudget(max_tokens=budget, reserved_tokens=0) context.fusion_result = engine.fuse([], b) @when("I coordinate then fuse with a budget of {budget:d} tokens") def step_coordinate_then_fuse(context: Context, budget: int) -> None: coordinator = StrategyCoordinator() b = ContextBudget(max_tokens=budget, reserved_tokens=0) coord_result = coordinator.coordinate( request={}, strategies=context.fusion_strategies, budget=b, fragments=context.fusion_fragments, ) engine = FusionEngine(config=FusionConfig(min_fragment_tokens=1)) context.fusion_result = engine.fuse(coord_result.fragments, b) context.integration_budget = budget # --------------------------------------------------------------------------- # When steps — Config # --------------------------------------------------------------------------- @when("I create a default CoordinatorConfig") def step_create_coord_config(context: Context) -> None: context.coord_config = CoordinatorConfig() @when("I create a default FusionConfig") def step_create_fusion_config(context: Context) -> None: context.fusion_config = FusionConfig() # --------------------------------------------------------------------------- # Then steps — StrategyCoordinator # --------------------------------------------------------------------------- @then("the coordination result should contain fragments") def step_coord_has_fragments(context: Context) -> None: assert len(context.coord_result.fragments) > 0, ( "Expected fragments from coordination" ) @then("the coordination result should list strategies used") def step_coord_strategies_used(context: Context) -> None: assert len(context.coord_result.strategies_used) > 0, ( "Expected strategies used in result" ) @then("the coordination result should contain {count:d} fragments") def step_coord_count(context: Context, count: int) -> None: assert len(context.coord_result.fragments) == count, ( f"Expected {count} fragments, got {len(context.coord_result.fragments)}" ) @then("the coordination result should report circuit-broken strategies") def step_coord_circuit_broken(context: Context) -> None: assert len(context.coord_result.circuit_broken) > 0, ( "Expected circuit-broken strategies in result" ) @then("the allocation for the first strategy should be approximately {tokens:d}") def step_coord_alloc_first(context: Context, tokens: int) -> None: first_alloc = context.coord_result.allocations[0][2] assert abs(first_alloc - tokens) <= 1, ( f"First allocation {first_alloc}, expected ~{tokens}" ) @then("the allocation for the second strategy should be approximately {tokens:d}") def step_coord_alloc_second(context: Context, tokens: int) -> None: second_alloc = context.coord_result.allocations[1][2] assert abs(second_alloc - tokens) <= 1, ( f"Second allocation {second_alloc}, expected ~{tokens}" ) @then("the total allocation should equal {total:d}") def step_coord_alloc_total(context: Context, total: int) -> None: actual = sum(a[2] for a in context.coord_result.allocations) assert actual == total, f"Total allocation {actual} != {total}" @then("no strategy allocation should exceed {cap:d} tokens") def step_coord_cap_enforced(context: Context, cap: int) -> None: for name, _, tokens in context.coord_result.allocations: assert tokens <= cap, f"Strategy {name} allocated {tokens} > cap {cap}" # --------------------------------------------------------------------------- # Then steps — FusionEngine # --------------------------------------------------------------------------- @then("the fusion result dedup count should be greater than 0") def step_fusion_dedup_count(context: Context) -> None: assert context.fusion_result.dedup_count > 0, ( f"Expected dedup_count > 0, got {context.fusion_result.dedup_count}" ) @then("the fusion result should contain fewer fragments than input") def step_fusion_fewer(context: Context) -> None: assert len(context.fusion_result.fragments) < context.fusion_input_count, ( f"Expected fewer fragments ({len(context.fusion_result.fragments)}) " f"than input ({context.fusion_input_count})" ) @then("the fusion result should retain the depth {depth:d} fragment") def step_fusion_depth_retained(context: Context, depth: int) -> None: depths = [f.detail_depth for f in context.fusion_result.fragments] # After scoring, the depth values should still be present # The max-depth fragment should survive for the node node_frags = [f for f in context.fusion_result.fragments if "deep" in f.uko_node] if node_frags: max_depth = max(f.detail_depth for f in node_frags) assert max_depth == depth, ( f"Expected depth {depth} retained, got max {max_depth} in {depths}" ) @then("the fusion result depth resolved count should be greater than 0") def step_fusion_depth_resolved(context: Context) -> None: assert context.fusion_result.depth_resolved_count > 0, ( f"Expected depth_resolved > 0, got {context.fusion_result.depth_resolved_count}" ) @then("the fusion result total tokens should be at most {max_tokens:d}") def step_fusion_within_budget(context: Context, max_tokens: int) -> None: assert context.fusion_result.total_tokens <= max_tokens, ( f"Total tokens {context.fusion_result.total_tokens} > {max_tokens}" ) @then( "the fusion result fragments should be ordered by depth descending " "then token count ascending" ) def step_fusion_tiebreakers(context: Context) -> None: frags = context.fusion_result.fragments if len(frags) <= 1: return # Verify tie-breaker ordering within same relevance group for i in range(len(frags) - 1): a, b = frags[i], frags[i + 1] if ( abs(a.relevance_score - b.relevance_score) < 0.001 and a.detail_depth == b.detail_depth ): # Same relevance and depth: token count ascending assert a.token_count <= b.token_count, ( f"Token count tie-breaker failed: {a.token_count} > {b.token_count}" ) @then("the fusion result dropped by overage guard should be greater than 0") def step_fusion_overage_dropped(context: Context) -> None: assert context.fusion_result.dropped_by_overage_guard > 0, ( f"Expected drops > 0, got {context.fusion_result.dropped_by_overage_guard}" ) @then("the fusion result dropped by overage guard should be {count:d}") def step_fusion_overage_exact(context: Context, count: int) -> None: assert context.fusion_result.dropped_by_overage_guard == count, ( f"Expected {count} drops, got {context.fusion_result.dropped_by_overage_guard}" ) @then("the fusion result should contain {count:d} fragments") def step_fusion_count(context: Context, count: int) -> None: assert len(context.fusion_result.fragments) == count, ( f"Expected {count} fragments, got {len(context.fusion_result.fragments)}" ) @then("the fusion result input count should be {count:d}") def step_fusion_input(context: Context, count: int) -> None: assert context.fusion_result.input_count == count, ( f"Expected input_count {count}, got {context.fusion_result.input_count}" ) @then("the final fused result should contain fragments within budget") def step_integration_result(context: Context) -> None: assert len(context.fusion_result.fragments) > 0, ( "Expected fragments from integration" ) assert context.fusion_result.total_tokens <= context.integration_budget, ( f"Total tokens {context.fusion_result.total_tokens} " f"> budget {context.integration_budget}" ) # --------------------------------------------------------------------------- # Then steps — Config # --------------------------------------------------------------------------- @then("the config min_useful_budget should be {val:d}") def step_config_min_budget(context: Context, val: int) -> None: assert context.coord_config.min_useful_budget == val @then("the config executor_timeout should be {val:g}") def step_config_timeout(context: Context, val: float) -> None: assert context.coord_config.executor_timeout == val @then("the config per_strategy_max_cap should be None") def step_config_cap_none(context: Context) -> None: assert context.coord_config.per_strategy_max_cap is None @then("the fusion config overage_guard_enabled should be True") def step_fusion_config_overage(context: Context) -> None: assert context.fusion_config.overage_guard_enabled is True @then("the fusion config min_fragment_tokens should be {val:d}") def step_fusion_config_min_tokens(context: Context, val: int) -> None: assert context.fusion_config.min_fragment_tokens == val