cleveragents-core/benchmarks/acms_context_cli_bench.py

"""ASV benchmarks for ACMS context CLI commands.

Measures the performance of:
- _format_budget_utilization at various utilization levels
- acms_context_show with varying fragment counts
- acms_context_clear with varying fragment counts
"""

from __future__ import annotations

import contextlib
import importlib
import sys
from pathlib import Path
from typing import Any
from unittest.mock import MagicMock, patch

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

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

import cleveragents  # noqa: E402

importlib.reload(cleveragents)

from features.mocks.acms_context_mocks import (  # noqa: E402
    make_mock_container,
    make_mock_tier_service,
    make_tiered_fragment,
)

from cleveragents.cli.commands.acms_context import (  # noqa: E402
    _format_budget_utilization,
    acms_context_clear,
    acms_context_show,
)
from cleveragents.domain.models.acms.tiers import ContextTier  # noqa: E402


def _make_fragments(count: int) -> list[Any]:
    """Create a list of sample TieredFragments for benchmarking."""
    tiers = [ContextTier.HOT, ContextTier.WARM, ContextTier.COLD]
    tags = ("default", "deprecated", "archived")
    return [
        make_tiered_fragment(
            fragment_id=f"bench-frag-{i}",
            resource_id=f"src/module{i}.py",
            tier=tiers[i % 3],
            token_count=1024 * (i + 1),
            project_name="bench_project",
            tag=tags[i % 3],
        )
        for i in range(count)
    ]


def _run_show_silent(view: str, fragments: list[Any]) -> None:
    """Run acms_context_show with output suppressed."""
    mock_service = make_mock_tier_service(
        fragments=fragments,
        scoped_fragments=fragments,
    )
    mock_container = make_mock_container(mock_service)
    mock_console = MagicMock()
    mock_console.print.return_value = None

    with (
        patch(
            "cleveragents.cli.commands.acms_context.get_container",
            return_value=mock_container,
        ),
        patch(
            "cleveragents.cli.renderers._get_console",
            return_value=mock_console,
        ),
        contextlib.suppress(SystemExit),
    ):
        acms_context_show(view)


def _run_clear_silent(
    fragments: list[Any],
    tier: str | None = None,
) -> None:
    """Run acms_context_clear with output suppressed."""
    mock_service = make_mock_tier_service(fragments=fragments)
    mock_container = make_mock_container(mock_service)
    mock_console = MagicMock()
    mock_console.print.return_value = None

    with (
        patch(
            "cleveragents.cli.commands.acms_context.get_container",
            return_value=mock_container,
        ),
        patch(
            "cleveragents.cli.renderers._get_console",
            return_value=mock_console,
        ),
        contextlib.suppress(SystemExit),
    ):
        acms_context_clear(tier=tier, yes=True)


class FormatBudgetUtilizationSuite:
    """Benchmark _format_budget_utilization throughput."""

    def time_format_zero_total(self) -> None:
        """Benchmark formatting with zero total."""
        _format_budget_utilization(0, 0)

    def time_format_half_utilization(self) -> None:
        """Benchmark formatting at 50% utilization."""
        _format_budget_utilization(500, 1000)

    def time_format_full_utilization(self) -> None:
        """Benchmark formatting at 100% utilization."""
        _format_budget_utilization(1000, 1000)

    def time_format_negative_values(self) -> None:
        """Benchmark formatting with negative values."""
        _format_budget_utilization(-1, 100)


class ContextShowSuite:
    """Benchmark acms_context_show with varying fragment counts."""

    def setup(self) -> None:
        """Set up fragment lists for benchmarking."""
        self._frags_10 = _make_fragments(10)
        self._frags_100 = _make_fragments(100)
        self._frags_empty: list[Any] = []

    def time_show_10_fragments(self) -> None:
        """Benchmark context show with 10 fragments."""
        _run_show_silent("bench_project", self._frags_10)

    def time_show_100_fragments(self) -> None:
        """Benchmark context show with 100 fragments."""
        _run_show_silent("bench_project", self._frags_100)

    def time_show_empty_context(self) -> None:
        """Benchmark context show with empty context."""
        _run_show_silent("bench_project", self._frags_empty)


class ContextClearSuite:
    """Benchmark acms_context_clear with varying fragment counts."""

    def setup(self) -> None:
        """Set up fragment lists for benchmarking."""
        self._frags_10 = _make_fragments(10)
        self._frags_100 = _make_fragments(100)
        self._frags_empty: list[Any] = []

    def time_clear_10_fragments(self) -> None:
        """Benchmark context clear with 10 fragments."""
        _run_clear_silent(self._frags_10)

    def time_clear_100_fragments(self) -> None:
        """Benchmark context clear with 100 fragments."""
        _run_clear_silent(self._frags_100)

    def time_clear_no_fragments(self) -> None:
        """Benchmark context clear with no matching fragments."""
        _run_clear_silent(self._frags_empty)

    def time_clear_by_tier(self) -> None:
        """Benchmark context clear filtered by tier."""
        _run_clear_silent(self._frags_10, tier="hot")