cleveragents-core/features/mocks/advanced_context_strategies_mocks.py

"""Mock implementations for advanced context strategies tests.

FakeEmbeddings provides deterministic word-overlap embeddings so tests never
hit a real embedding API.  The three strategy classes are test-only
implementations that satisfy the strategy duck-type contract used by the
Behave and Robot Framework test layers.
"""

from __future__ import annotations

from typing import Any

from cleveragents.application.services.context_strategies import (
    BreadthDepthNavigatorStrategy,
    SemanticEmbeddingStrategy,
)
from cleveragents.domain.models.core.context_fragment import (
    ContextBudget,
    ContextFragment,
)


class FakeEmbeddings:
    """Deterministic fake embeddings for testing without real API calls."""

    def __init__(self) -> None:
        self._cache: dict[str, list[float]] = {}

    def embed_query(self, text: str) -> list[float]:
        if text not in self._cache:
            hash_val = hash(text) % 1000
            self._cache[text] = [float((hash_val + i) % 100) / 100.0 for i in range(10)]
        return self._cache[text]

    def embed_documents(self, texts: list[str]) -> list[list[float]]:
        return [self.embed_query(text) for text in texts]


def _pack_budget(
    fragments: list[ContextFragment], budget: ContextBudget
) -> list[ContextFragment]:
    result: list[ContextFragment] = []
    used_tokens = budget.reserved_tokens

    for frag in fragments:
        if used_tokens + frag.token_count <= budget.max_tokens:
            result.append(frag)
            used_tokens += frag.token_count
        else:
            break

    return result


class RelevanceScoringStrategy:
    """Strategy that ranks fragments purely by relevance score."""

    def __init__(self) -> None:
        pass

    @property
    def name(self) -> str:
        return "relevance-scoring"

    def can_handle(self, request: dict[str, Any]) -> float:
        return 0.5

    def assemble(
        self,
        fragments: list[ContextFragment],
        budget: ContextBudget,
    ) -> list[ContextFragment]:
        if not fragments:
            return []
        sorted_frags = sorted(fragments, key=lambda f: f.relevance_score, reverse=True)
        return _pack_budget(sorted_frags, budget)

    def explain(self) -> str:
        return "Ranks fragments purely by relevance score."


class AdaptiveContextSelector:
    """Selects the best strategy based on request characteristics."""

    def __init__(self) -> None:
        self._strategies: dict[str, Any] = {
            "semantic-embedding": SemanticEmbeddingStrategy(),
            "relevance-scoring": RelevanceScoringStrategy(),
            "breadth-depth-navigator": BreadthDepthNavigatorStrategy(),
        }

    @property
    def name(self) -> str:
        return "adaptive-selector"

    def select_strategy(self, request: dict[str, Any]) -> tuple[str, Any]:
        best_name = "relevance-scoring"
        best_confidence = 0.0

        for name, strategy in self._strategies.items():
            confidence = strategy.can_handle(request)
            if confidence > best_confidence:
                best_confidence = confidence
                best_name = name

        return best_name, self._strategies[best_name]


class ContextFusionStrategy:
    """Fuses results from multiple strategies."""

    def __init__(self, strategy_names: list[str]) -> None:
        self._strategy_names = strategy_names
        self._strategies: dict[str, Any] = {
            "semantic-embedding": SemanticEmbeddingStrategy(),
            "relevance-scoring": RelevanceScoringStrategy(),
            "breadth-depth-navigator": BreadthDepthNavigatorStrategy(),
        }

    @property
    def name(self) -> str:
        return "context-fusion"

    def assemble(
        self,
        fragments: list[ContextFragment],
        budget: ContextBudget,
        query: str = "",
    ) -> list[ContextFragment]:
        if not fragments:
            return []

        all_results: dict[str, ContextFragment] = {}
        remaining_budget = budget.max_tokens - budget.reserved_tokens

        for strategy_name in self._strategy_names:
            if remaining_budget <= 0:
                break

            strategy = self._strategies.get(strategy_name)
            if not strategy:
                continue

            if hasattr(strategy, "set_query"):
                strategy.set_query(query)

            strategy_budget = ContextBudget(
                max_tokens=remaining_budget,
                reserved_tokens=0,
            )

            results = strategy.assemble(fragments, strategy_budget)

            for frag in results:
                if frag.uko_node not in all_results:
                    all_results[frag.uko_node] = frag
                    remaining_budget -= frag.token_count

        return list(all_results.values())