cleveragents-core/benchmarks/acms_pipeline_bench.py

"""ASV benchmarks for ACMS v1 context assembly pipeline.

Measures the performance of:
- ContextFragment creation (with and without metadata)
- ACMSPipeline.assemble with varying fragment counts
- Tiered fusion strategy ranking
- Recency strategy ranking
"""

from __future__ import annotations

import importlib
import sys
from datetime import UTC, datetime
from pathlib import Path

_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_service import ACMSPipeline  # noqa: E402
from cleveragents.domain.models.core.context_fragment import (  # noqa: E402
    ContextBudget,
    ContextFragment,
    FragmentProvenance,
    compute_context_hash,
)

# Default provenance for benchmark fragments.
_DEFAULT_PROV = FragmentProvenance(resource_uri="bench://default")


class ContextFragmentSuite:
    """Benchmark ContextFragment creation throughput."""

    def time_fragment_creation(self) -> None:
        """Benchmark creating a ContextFragment with defaults."""
        ContextFragment(
            uko_node="bench://file",
            content="benchmark content",
            token_count=10,
            provenance=_DEFAULT_PROV,
        )

    def time_fragment_with_metadata(self) -> None:
        """Benchmark creating a ContextFragment with metadata."""
        ContextFragment(
            uko_node="bench://decision",
            content="benchmark with metadata",
            relevance_score=0.85,
            token_count=150,
            tier="hot",
            metadata={"author": "bench", "priority": "high"},
            provenance=_DEFAULT_PROV,
        )


class ACMSPipelineSuite:
    """Benchmark ACMSPipeline.assemble throughput."""

    def setup(self) -> None:
        """Set up pipeline and fragment lists for assembly benchmarks."""
        self._pipeline = ACMSPipeline()
        self._budget = ContextBudget(max_tokens=100_000, reserved_tokens=0)
        self._frags_10 = [
            ContextFragment(
                uko_node=f"bench://file/{i}",
                content=f"fragment {i}",
                relevance_score=round(i / 10, 1),
                token_count=50,
                provenance=_DEFAULT_PROV,
            )
            for i in range(10)
        ]
        self._frags_100 = [
            ContextFragment(
                uko_node=f"bench://file/{i}",
                content=f"fragment {i}",
                relevance_score=round((i % 10) / 10, 1),
                token_count=50,
                provenance=_DEFAULT_PROV,
            )
            for i in range(100)
        ]
        self._frags_1000 = [
            ContextFragment(
                uko_node=f"bench://file/{i}",
                content=f"fragment {i}",
                relevance_score=round((i % 10) / 10, 1),
                token_count=50,
                provenance=_DEFAULT_PROV,
            )
            for i in range(1000)
        ]
        tiers = ("hot", "warm", "cold")
        self._tiered_frags = [
            ContextFragment(
                uko_node=f"bench://file/{i}",
                content=f"fragment {i}",
                relevance_score=round((i % 10) / 10, 1),
                token_count=50,
                tier=tiers[i % 3],
                provenance=_DEFAULT_PROV,
            )
            for i in range(100)
        ]
        # Recency benchmark fragments with distinct timestamps
        self._recency_frags = [
            ContextFragment(
                uko_node=f"bench://file/{i}",
                content=f"fragment {i}",
                token_count=50,
                created_at=datetime(2024, 1, 1 + (i % 28), tzinfo=UTC),
                provenance=_DEFAULT_PROV,
            )
            for i in range(100)
        ]

    def time_assemble_10_fragments(self) -> None:
        """Benchmark assembling 10 fragments with relevance strategy."""
        self._pipeline.assemble(
            plan_id="01JQBENCHM00000000000000AA",
            fragments=self._frags_10,
            budget=self._budget,
        )

    def time_assemble_100_fragments(self) -> None:
        """Benchmark assembling 100 fragments with relevance strategy."""
        self._pipeline.assemble(
            plan_id="01JQBENCHM00000000000000AA",
            fragments=self._frags_100,
            budget=self._budget,
        )

    def time_assemble_1000_fragments(self) -> None:
        """Benchmark assembling 1000 fragments with relevance strategy."""
        self._pipeline.assemble(
            plan_id="01JQBENCHM00000000000000AA",
            fragments=self._frags_1000,
            budget=self._budget,
        )

    def time_tiered_strategy(self) -> None:
        """Benchmark assembling with tiered strategy."""
        self._pipeline.assemble(
            plan_id="01JQBENCHM00000000000000AA",
            fragments=self._tiered_frags,
            budget=self._budget,
            strategy="tiered",
        )

    def time_recency_strategy(self) -> None:
        """Benchmark assembling with recency strategy."""
        self._pipeline.assemble(
            plan_id="01JQBENCHM00000000000000AA",
            fragments=self._recency_frags,
            budget=self._budget,
            strategy="recency",
        )


class ContextHashSuite:
    """Benchmark compute_context_hash in isolation.

    Measures the length-prefixed SHA-256 hash function at various
    fragment counts to detect performance regressions.
    """

    def setup(self) -> None:
        """Build fragment tuples of varying sizes."""
        self._frags_10 = tuple(
            ContextFragment(
                uko_node=f"bench://hash/{i}",
                content=f"fragment content {i}" * 10,
                token_count=50,
                provenance=_DEFAULT_PROV,
            )
            for i in range(10)
        )
        self._frags_100 = tuple(
            ContextFragment(
                uko_node=f"bench://hash/{i}",
                content=f"fragment content {i}" * 10,
                token_count=50,
                provenance=_DEFAULT_PROV,
            )
            for i in range(100)
        )
        self._frags_1000 = tuple(
            ContextFragment(
                uko_node=f"bench://hash/{i}",
                content=f"fragment content {i}" * 10,
                token_count=50,
                provenance=_DEFAULT_PROV,
            )
            for i in range(1000)
        )

    def time_hash_10_fragments(self) -> None:
        """Benchmark hashing 10 fragments."""
        compute_context_hash(self._frags_10)

    def time_hash_100_fragments(self) -> None:
        """Benchmark hashing 100 fragments."""
        compute_context_hash(self._frags_100)

    def time_hash_1000_fragments(self) -> None:
        """Benchmark hashing 1000 fragments."""
        compute_context_hash(self._frags_1000)