cleveragents-core/benchmarks/bench_context_tier_runtime.py

"""ASV benchmarks for context tier runtime logic (issue #821).

Measures the overhead of auto-promotion on access, staleness
enforcement, and budget-based eviction.
"""

from __future__ import annotations

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

# Ensure source tree is importable
_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.context_tiers import (  # noqa: E402
    ContextTierService,
)
from cleveragents.domain.models.acms.tiers import (  # noqa: E402
    ContextTier,
    TierBudget,
    TieredFragment,
)


def _make_frag(
    fid: str,
    tier: ContextTier,
    tokens: int = 50,
    last_accessed: datetime | None = None,
) -> TieredFragment:
    frag = TieredFragment(
        fragment_id=fid,
        content="bench",
        tier=tier,
        token_count=tokens,
        project_name="bench",
    )
    if last_accessed is not None:
        frag.last_accessed = last_accessed
    return frag


class AutoPromotionSuite:
    """Benchmark auto-promotion on repeated ``get()`` calls."""

    timeout = 60

    def setup(self) -> None:
        self.service = ContextTierService(settings=None)
        frag = _make_frag("bench-promote", ContextTier.COLD)
        self.service.store(frag)

    def time_auto_promote_5_accesses(self) -> None:
        for _ in range(5):
            self.service.get("bench-promote")


class StalenessEnforcementSuite:
    """Benchmark ``enforce_staleness()`` with varying fragment counts."""

    timeout = 60

    def setup(self) -> None:
        self.service = ContextTierService(settings=None)
        stale = datetime.now(tz=UTC) - timedelta(hours=25)
        for i in range(100):
            frag = _make_frag(f"stale-{i:04d}", ContextTier.HOT, last_accessed=stale)
            self.service.store(frag)

    def time_enforce_staleness_100_fragments(self) -> None:
        self.service.enforce_staleness()


class BudgetEvictionSuite:
    """Benchmark budget-based eviction on hot-tier overflow."""

    timeout = 60

    def setup(self) -> None:
        self.service = ContextTierService(settings=None)
        self.service._budget = TierBudget(max_tokens_hot=500)
        old_time = datetime.now(tz=UTC) - timedelta(minutes=100)
        for i in range(10):
            ts = old_time + timedelta(minutes=i)
            frag = _make_frag(
                f"budget-{i:03d}",
                ContextTier.HOT,
                tokens=50,
                last_accessed=ts,
            )
            self.service.store(frag)
        self.overflow_frag = _make_frag(
            "overflow",
            ContextTier.HOT,
            tokens=50,
        )

    def time_store_with_budget_eviction(self) -> None:
        self.service.store(self.overflow_frag)


class TierEventEmissionSuite:
    """Benchmark tier transition event emission overhead."""

    timeout = 60

    def setup(self) -> None:
        from collections.abc import Callable

        from cleveragents.infrastructure.events.models import DomainEvent
        from cleveragents.infrastructure.events.types import EventType

        class _NullBus:
            def emit(self, event: DomainEvent) -> None:
                pass

            def subscribe(
                self,
                event_type: EventType,
                handler: Callable[[DomainEvent], None],
            ) -> None:
                pass

        self.service = ContextTierService(
            settings=None,
            event_bus=_NullBus(),
        )
        frag = _make_frag("evt-bench", ContextTier.COLD)
        self.service.store(frag)

    def time_promote_with_event(self) -> None:
        self.service.promote("evt-bench")