cleveragents-core/benchmarks/acms_pipeline_phase3_bench.py

"""ASV benchmarks for ACMS pipeline Phase 3 — Context Finalization and
advanced strategy components.

Measures the performance of:
- RelevanceCoherenceOrderer with varying fragment counts
- ProvenancePreambleGenerator with varying fragment counts
- ArceStrategy with varying fragment counts and iteration limits
- TemporalArchaeologyStrategy with varying fragment counts
- PlanDecisionContextStrategy with varying fragment counts
"""

from __future__ import annotations

import importlib
import sys
from pathlib import Path
from typing import ClassVar

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

import cleveragents  # noqa: E402

importlib.reload(cleveragents)

from cleveragents.application.services.acms_advanced_strategies import (  # noqa: E402
    ArceStrategy,
    PlanDecisionContextStrategy,
    TemporalArchaeologyStrategy,
)
from cleveragents.application.services.acms_phase3 import (  # noqa: E402
    ProvenancePreambleGenerator,
    RelevanceCoherenceOrderer,
)
from cleveragents.domain.models.core.context_fragment import (  # noqa: E402
    ContextBudget,
    ContextFragment,
    FragmentProvenance,
)

_DEFAULT_PROV = FragmentProvenance(resource_uri="bench://phase3")


def _make_fragments(
    n: int,
    *,
    unique_nodes: bool = True,
    vary_tiers: bool = False,
) -> list[ContextFragment]:
    """Create *n* fragments for benchmarks."""
    tiers = ["hot", "warm", "cold"]
    frags: list[ContextFragment] = []
    for i in range(n):
        node = f"bench://file{i}" if unique_nodes else "bench://same-node"
        tier = tiers[i % 3] if vary_tiers else "warm"
        frags.append(
            ContextFragment(
                uko_node=node,
                content=f"content-{i} " * max(1, (i % 10)),
                relevance_score=round(0.1 + (i % 10) * 0.09, 2),
                token_count=50 + (i % 5) * 10,
                detail_depth=i % 10,
                tier=tier,
                strategy_source=f"strategy-{i % 3}",
                provenance=_DEFAULT_PROV,
            )
        )
    return frags


class RelevanceCoherenceOrdererSuite:
    """Benchmark RelevanceCoherenceOrderer throughput."""

    params: ClassVar[list[int]] = [10, 100, 500]
    param_names: ClassVar[list[str]] = ["num_fragments"]

    def setup(self, n: int) -> None:
        self.orderer = RelevanceCoherenceOrderer()
        self.fragments = _make_fragments(n)

    def time_order(self, n: int) -> None:
        self.orderer.order(self.fragments)


class ProvenancePreambleGeneratorSuite:
    """Benchmark ProvenancePreambleGenerator throughput."""

    params: ClassVar[list[int]] = [10, 100, 500]
    param_names: ClassVar[list[str]] = ["num_fragments"]

    def setup(self, n: int) -> None:
        self.generator = ProvenancePreambleGenerator()
        self.fragments = _make_fragments(n)

    def time_generate(self, n: int) -> None:
        self.generator.generate(self.fragments)


class ArceStrategySuite:
    """Benchmark ArceStrategy throughput."""

    params: ClassVar[list[int]] = [10, 100, 500]
    param_names: ClassVar[list[str]] = ["num_fragments"]

    def setup(self, n: int) -> None:
        self.strategy = ArceStrategy(max_iterations=3)
        self.fragments = _make_fragments(n)
        self.budget = ContextBudget(max_tokens=100000, reserved_tokens=0)

    def time_assemble(self, n: int) -> None:
        self.strategy.assemble(self.fragments, self.budget)


class TemporalArchaeologyStrategySuite:
    """Benchmark TemporalArchaeologyStrategy throughput."""

    params: ClassVar[list[int]] = [10, 100, 500]
    param_names: ClassVar[list[str]] = ["num_fragments"]

    def setup(self, n: int) -> None:
        self.strategy = TemporalArchaeologyStrategy()
        self.fragments = _make_fragments(n, vary_tiers=True)
        self.budget = ContextBudget(max_tokens=100000, reserved_tokens=0)

    def time_assemble(self, n: int) -> None:
        self.strategy.assemble(self.fragments, self.budget)


class PlanDecisionContextStrategySuite:
    """Benchmark PlanDecisionContextStrategy throughput."""

    params: ClassVar[list[int]] = [10, 100, 500]
    param_names: ClassVar[list[str]] = ["num_fragments"]

    def setup(self, n: int) -> None:
        self.strategy = PlanDecisionContextStrategy()
        self.fragments = _make_fragments(n, vary_tiers=True)
        self.budget = ContextBudget(max_tokens=100000, reserved_tokens=0)

    def time_assemble(self, n: int) -> None:
        self.strategy.assemble(self.fragments, self.budget)