forked from cleveragents/cleveragents-core
610 lines
21 KiB
Python
610 lines
21 KiB
Python
"""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
|