cleveragents-core/benchmarks/context_tiers_bench.py

"""ASV benchmarks for ACMS Context Tiers.

Measures the performance of:
- TieredFragment construction
- ContextTierService store/get operations
- Tier promotion and demotion
- LRU eviction
- Actor view filtering
- Project scoping
"""

from __future__ import annotations

import importlib
import sys
from pathlib import Path

# Ensure the local *source* tree is importable even when ASV has an
# older build of the package installed.
_SRC = str(Path(__file__).resolve().parents[1] / "src")
if _SRC not in sys.path:
    sys.path.insert(0, _SRC)

# Force-reload so ASV picks up the source tree version.
import cleveragents  # noqa: E402

importlib.reload(cleveragents)

from cleveragents.application.services.context_tiers import (  # noqa: E402
    ContextTierService,
)
from cleveragents.domain.models.acms.tiers import (  # noqa: E402
    ActorRole,
    ContextTier,
    ScopedBackendView,
    TierBudget,
    TieredFragment,
    TierMetrics,
)

# ---------------------------------------------------------------------------
# Model construction benchmarks
# ---------------------------------------------------------------------------


class TierModelConstructionSuite:
    """Benchmark tier model object creation overhead."""

    timeout = 60

    def time_create_tiered_fragment(self) -> None:
        TieredFragment(fragment_id="bench-001", content="benchmark data")

    def time_create_tier_budget(self) -> None:
        TierBudget()

    def time_create_tier_metrics(self) -> None:
        TierMetrics()

    def time_create_scoped_view(self) -> None:
        ScopedBackendView(allowed_projects=["proj-a", "proj-b"])


# ---------------------------------------------------------------------------
# Service operation benchmarks
# ---------------------------------------------------------------------------


class TierServiceStoreSuite:
    """Benchmark ContextTierService store/get operations."""

    timeout = 60

    def setup(self) -> None:
        self.service = ContextTierService()
        self.fragment = TieredFragment(
            fragment_id="bench-store",
            content="benchmark content",
            tier=ContextTier.HOT,
        )

    def time_store_fragment(self) -> None:
        self.service.store(self.fragment)

    def time_get_fragment(self) -> None:
        self.service.store(self.fragment)
        self.service.get("bench-store")

    def time_get_miss(self) -> None:
        self.service.get("nonexistent")


class TierServicePromotionSuite:
    """Benchmark tier promotion and demotion."""

    timeout = 60

    def setup(self) -> None:
        self.service = ContextTierService()

    def time_promote_cold_to_warm(self) -> None:
        frag = TieredFragment(
            fragment_id="promote-bench",
            content="data",
            tier=ContextTier.COLD,
        )
        self.service.store(frag)
        self.service.promote("promote-bench")

    def time_demote_hot_to_warm(self) -> None:
        frag = TieredFragment(
            fragment_id="demote-bench",
            content="data",
            tier=ContextTier.HOT,
        )
        self.service.store(frag)
        self.service.demote("demote-bench")


class TierServiceEvictionSuite:
    """Benchmark LRU eviction performance."""

    timeout = 60

    def setup(self) -> None:
        self.service = ContextTierService()
        for i in range(100):
            self.service.store(
                TieredFragment(
                    fragment_id=f"evict-{i}",
                    content=f"data-{i}",
                    tier=ContextTier.HOT,
                )
            )

    def time_evict_10_from_100(self) -> None:
        self.service.evict_lru(ContextTier.HOT, 10)


class TierServiceActorViewSuite:
    """Benchmark per-actor filtered view performance."""

    timeout = 60

    def setup(self) -> None:
        self.service = ContextTierService()
        for i in range(50):
            tier = [ContextTier.HOT, ContextTier.WARM, ContextTier.COLD][i % 3]
            self.service.store(
                TieredFragment(
                    fragment_id=f"actor-{i}",
                    content=f"data-{i}",
                    tier=tier,
                    project_name="bench-proj",
                )
            )

    def time_strategist_view(self) -> None:
        self.service.get_for_actor(ActorRole.STRATEGIST, ["bench-proj"])

    def time_executor_view(self) -> None:
        self.service.get_for_actor(ActorRole.EXECUTOR, ["bench-proj"])

    def time_reviewer_view(self) -> None:
        self.service.get_for_actor(ActorRole.REVIEWER, ["bench-proj"])

    def time_scoped_view(self) -> None:
        self.service.get_scoped_view(["bench-proj"])