cleveragents-core/robot/helper_acms_fusion.py

"""Robot Framework helper for ACMS Fusion integration tests.

Provides a CLI-style interface for Robot to invoke StrategyCoordinator
and FusionEngine operations and verify the results.

Uses dynamic test data generation via helper_test_data_factory for realistic
test data instead of hardcoded values.

Usage:
    python robot/helper_acms_fusion.py coord-basic
    python robot/helper_acms_fusion.py coord-budget
    python robot/helper_acms_fusion.py coord-caps
    python robot/helper_acms_fusion.py coord-circuit
    python robot/helper_acms_fusion.py fuse-dedup
    python robot/helper_acms_fusion.py fuse-depth
    python robot/helper_acms_fusion.py fuse-pack
    python robot/helper_acms_fusion.py fuse-overage
    python robot/helper_acms_fusion.py integration
"""

from __future__ import annotations

import sys
from collections.abc import Callable, Sequence
from pathlib import Path
from typing import Any

_SRC = str(Path(__file__).resolve().parents[1] / "src")
if _SRC not in sys.path:
    sys.path.insert(0, _SRC)

from cleveragents.application.services.acms_pipeline import (  # noqa: E402, I001
    CircuitBreaker,
    ParallelStrategyExecutor,
)
from cleveragents.application.services.acms_service import (  # noqa: E402
    StrategyCapabilities,
)
from cleveragents.application.services.fusion_engine import (  # noqa: E402
    FusionConfig,
    FusionEngine,
)
from cleveragents.application.services.strategy_coordinator import (  # noqa: E402
    CoordinatorConfig,
    StrategyCoordinator,
)
from cleveragents.domain.models.core.context_fragment import (  # noqa: E402
    ContextBudget,
    ContextFragment,
)

from helper_test_data_factory import (  # noqa: E402
    RobotContextBudgetFactory,
    RobotContextFragmentFactory,
    RobotTestDataGenerator,
)


class _TestStrategy:
    """Minimal strategy for robot tests."""

    def __init__(self, name: str = "robot_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"Robot test strategy '{self._name}'."


def _cmd_coord_basic() -> int:
    """StrategyCoordinator basic coordination with realistic test data."""
    coordinator = StrategyCoordinator()
    strategies = [_TestStrategy("a", 0.8), _TestStrategy("b", 0.6)]
    # Use factory to generate realistic fragments instead of hardcoded "alpha"/"beta"
    frags = [
        RobotContextFragmentFactory.create_object(
            token_count=50,
            relevance_score=0.9,
        ),
        RobotContextFragmentFactory.create_object(
            token_count=50,
            relevance_score=0.5,
        ),
    ]
    b = RobotContextBudgetFactory.create_object(max_tokens=200)
    result = coordinator.coordinate(
        request={},
        strategies=strategies,
        budget=b,
        fragments=frags,
    )
    assert len(result.fragments) > 0, "Expected fragments"
    assert len(result.strategies_used) > 0, "Expected strategies used"
    print("coord-basic-ok")
    return 0


def _cmd_coord_budget() -> int:
    """StrategyCoordinator proportional budget allocation."""
    coordinator = StrategyCoordinator()
    strategies = [_TestStrategy("a", 0.8), _TestStrategy("b", 0.2)]
    frags = [
        RobotContextFragmentFactory.create_object(
            token_count=50,
            relevance_score=0.9,
        ),
    ]
    b = RobotContextBudgetFactory.create_object(max_tokens=1000)
    result = coordinator.coordinate(
        request={},
        strategies=strategies,
        budget=b,
        fragments=frags,
    )
    total_alloc = sum(a[2] for a in result.allocations)
    assert total_alloc == 1000, f"Total {total_alloc} != 1000"
    first_alloc = result.allocations[0][2]
    assert abs(first_alloc - 800) <= 1, f"First {first_alloc} != ~800"
    print("coord-budget-ok")
    return 0


def _cmd_coord_caps() -> int:
    """StrategyCoordinator per-strategy max caps."""
    config = CoordinatorConfig(per_strategy_max_cap=300)
    coordinator = StrategyCoordinator(config=config)
    strategies = [_TestStrategy("a", 0.9), _TestStrategy("b", 0.1)]
    frags = [
        RobotContextFragmentFactory.create_object(
            token_count=50,
            relevance_score=0.9,
        ),
    ]
    b = RobotContextBudgetFactory.create_object(max_tokens=1000)
    result = coordinator.coordinate(
        request={},
        strategies=strategies,
        budget=b,
        fragments=frags,
    )
    for name, _, tokens in result.allocations:
        assert tokens <= 300, f"Strategy {name} got {tokens} > 300"
    print("coord-caps-ok")
    return 0


def _cmd_coord_circuit() -> int:
    """StrategyCoordinator circuit breaker reporting."""
    cb = CircuitBreaker(failure_threshold=1)
    cb.record_failure("broken")
    executor = ParallelStrategyExecutor(circuit_breaker=cb)
    coordinator = StrategyCoordinator(executor=executor)
    strategies = [_TestStrategy("broken", 0.7)]
    frags = [
        RobotContextFragmentFactory.create_object(
            token_count=50,
            relevance_score=0.9,
        ),
    ]
    b = RobotContextBudgetFactory.create_object(max_tokens=200)
    result = coordinator.coordinate(
        request={},
        strategies=strategies,
        budget=b,
        fragments=frags,
    )
    assert "broken" in result.circuit_broken, "Expected 'broken' in circuit_broken"
    print("coord-circuit-ok")
    return 0


def _cmd_fuse_dedup() -> int:
    """FusionEngine deduplication with realistic test data."""
    engine = FusionEngine(config=FusionConfig(min_fragment_tokens=1))
    # Create fragments with same content for deduplication testing
    shared_content = RobotTestDataGenerator.python_code()
    shared_uko = RobotTestDataGenerator.uko_node_uri()
    frags = [
        RobotContextFragmentFactory.create_object(
            uko_node=shared_uko,
            content=shared_content,
            token_count=50,
            relevance_score=0.9,
        ),
        RobotContextFragmentFactory.create_object(
            uko_node=shared_uko,
            content=shared_content,
            token_count=50,
            relevance_score=0.7,
        ),
        RobotContextFragmentFactory.create_object(
            token_count=50,
            relevance_score=0.5,
        ),
    ]
    b = RobotContextBudgetFactory.create_object(max_tokens=500)
    result = engine.fuse(frags, b)
    assert result.dedup_count > 0, f"dedup_count={result.dedup_count}"
    assert len(result.fragments) < len(frags), "Expected fewer fragments"
    print("fuse-dedup-ok")
    return 0


def _cmd_fuse_depth() -> int:
    """FusionEngine depth resolution with realistic test data."""
    engine = FusionEngine(config=FusionConfig(min_fragment_tokens=1))
    shared_uko = RobotTestDataGenerator.uko_node_uri()
    frags = [
        RobotContextFragmentFactory.create_object(
            uko_node=shared_uko,
            token_count=50,
            detail_depth=2,
            relevance_score=0.8,
        ),
        RobotContextFragmentFactory.create_object(
            uko_node=shared_uko,
            token_count=80,
            detail_depth=5,
            relevance_score=0.7,
        ),
        RobotContextFragmentFactory.create_object(
            token_count=50,
            detail_depth=3,
            relevance_score=0.6,
        ),
    ]
    b = RobotContextBudgetFactory.create_object(max_tokens=500)
    result = engine.fuse(frags, b)
    assert result.depth_resolved_count > 0, (
        f"depth_resolved={result.depth_resolved_count}"
    )
    deep_frags = [f for f in result.fragments if f.detail_depth >= 5]
    if deep_frags:
        assert max(f.detail_depth for f in deep_frags) == 5
    print("fuse-depth-ok")
    return 0


def _cmd_fuse_pack() -> int:
    """FusionEngine knapsack packing with realistic test data."""
    engine = FusionEngine(config=FusionConfig(min_fragment_tokens=1))
    frags = [
        RobotContextFragmentFactory.create_object(
            token_count=100,
            relevance_score=round(0.9 - i * 0.1, 2),
        )
        for i in range(6)
    ]
    b = RobotContextBudgetFactory.create_object(max_tokens=400)
    result = engine.fuse(frags, b)
    assert result.total_tokens <= 400, f"total_tokens={result.total_tokens}"
    print("fuse-pack-ok")
    return 0


def _cmd_fuse_overage() -> int:
    """FusionEngine budget overage guard with realistic test data."""
    config = FusionConfig(overage_guard_enabled=True, min_fragment_tokens=1)
    engine = FusionEngine(config=config)
    frags = [
        RobotContextFragmentFactory.create_object(
            token_count=40,
            relevance_score=round(0.9 - i * 0.2, 2),
        )
        for i in range(4)
    ]
    b = RobotContextBudgetFactory.create_object(max_tokens=100)
    result = engine.fuse(frags, b)
    assert result.total_tokens <= 100, f"total_tokens={result.total_tokens}"
    print("fuse-overage-ok")
    return 0


def _cmd_integration() -> int:
    """Integration: coordinator -> fusion with realistic test data."""
    coordinator = StrategyCoordinator()
    strategies = [_TestStrategy("a", 0.8)]
    frags = [
        RobotContextFragmentFactory.create_object(
            token_count=50,
            relevance_score=0.9,
        ),
        RobotContextFragmentFactory.create_object(
            token_count=50,
            relevance_score=0.5,
        ),
    ]
    b = RobotContextBudgetFactory.create_object(max_tokens=200)
    coord_result = coordinator.coordinate(
        request={},
        strategies=strategies,
        budget=b,
        fragments=frags,
    )
    engine = FusionEngine(config=FusionConfig(min_fragment_tokens=1))
    fuse_result = engine.fuse(coord_result.fragments, b)
    assert fuse_result.total_tokens <= 200
    assert len(fuse_result.fragments) > 0
    print("integration-ok")
    return 0


# ---------------------------------------------------------------------------
# CLI dispatcher
# ---------------------------------------------------------------------------

_COMMANDS: dict[str, Callable[[], int]] = {
    "coord-basic": _cmd_coord_basic,
    "coord-budget": _cmd_coord_budget,
    "coord-caps": _cmd_coord_caps,
    "coord-circuit": _cmd_coord_circuit,
    "fuse-dedup": _cmd_fuse_dedup,
    "fuse-depth": _cmd_fuse_depth,
    "fuse-pack": _cmd_fuse_pack,
    "fuse-overage": _cmd_fuse_overage,
    "integration": _cmd_integration,
}


def main() -> int:
    if len(sys.argv) < 2 or sys.argv[1] not in _COMMANDS:
        print(f"Usage: {sys.argv[0]} <{'|'.join(_COMMANDS)}>", file=sys.stderr)
        return 1
    try:
        return _COMMANDS[sys.argv[1]]()
    except Exception as exc:
        print(f"FAIL: {exc}", file=sys.stderr)
        import traceback

        traceback.print_exc()
        return 1


if __name__ == "__main__":
    sys.exit(main())