cleveragents-core/benchmarks/component_resolver_bench.py

"""ASV benchmarks for ComponentResolver scope chain resolution.

Measures the performance of:
- Global default resolution
- Project override resolution
- Plan override resolution (3-level chain)
- Cache hit performance
- Registration throughput
- Introspection API latency
"""

from __future__ import annotations

import importlib
import sys
from pathlib import Path
from typing import Any, Protocol, runtime_checkable

# 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)

import cleveragents  # noqa: E402

importlib.reload(cleveragents)

from cleveragents.application.services.component_resolver import (  # noqa: E402
    ComponentResolver,
)

# ---------------------------------------------------------------------------
# Test protocol and implementations
# ---------------------------------------------------------------------------


@runtime_checkable
class _BenchProtocol(Protocol):
    def run(self) -> str: ...


class _BenchGlobal:
    def run(self) -> str:
        return "global"


class _BenchProject:
    def run(self) -> str:
        return "project"


class _BenchPlan:
    def run(self) -> str:
        return "plan"


# ---------------------------------------------------------------------------
# Timing suite
# ---------------------------------------------------------------------------


class ComponentResolverTimeSuite:
    """Benchmark timing for component resolution across scope levels."""

    def setup(self) -> None:
        self._resolver = ComponentResolver()
        self._resolver.register_global(_BenchProtocol, _BenchGlobal())
        self._resolver.register_project("proj-1", _BenchProtocol, _BenchProject())
        self._resolver.register_plan("plan-1", _BenchProtocol, _BenchPlan())

    def time_resolve_global(self) -> None:
        """Resolve from global scope (no plan/project context)."""
        self._resolver.invalidate_cache()
        self._resolver.resolve(_BenchProtocol)

    def time_resolve_project(self) -> None:
        """Resolve from project scope."""
        self._resolver.invalidate_cache()
        self._resolver.resolve(_BenchProtocol, project_id="proj-1")

    def time_resolve_plan(self) -> None:
        """Resolve from plan scope (full 3-level chain)."""
        self._resolver.invalidate_cache()
        self._resolver.resolve(_BenchProtocol, plan_id="plan-1", project_id="proj-1")

    def time_resolve_cache_hit(self) -> None:
        """Resolve with cache hit (pre-warmed)."""
        # Prime cache
        self._resolver.resolve(_BenchProtocol)
        # Measure cache hit
        self._resolver.resolve(_BenchProtocol)

    def time_resolve_fallthrough(self) -> None:
        """Resolve with fallthrough (unknown plan/project)."""
        self._resolver.invalidate_cache()
        self._resolver.resolve(_BenchProtocol, plan_id="unknown", project_id="unknown")

    def time_register_global(self) -> None:
        """Register a global implementation."""
        resolver = ComponentResolver()
        resolver.register_global(_BenchProtocol, _BenchGlobal())

    def time_register_project(self) -> None:
        """Register a project implementation."""
        resolver = ComponentResolver()
        resolver.register_project("p", _BenchProtocol, _BenchProject())

    def time_register_plan(self) -> None:
        """Register a plan implementation."""
        resolver = ComponentResolver()
        resolver.register_plan("p", _BenchProtocol, _BenchPlan())

    def time_has_global(self) -> None:
        """Check has_global."""
        self._resolver.has_global(_BenchProtocol)

    def time_list_global_types(self) -> None:
        """List global type names."""
        self._resolver.list_global_types()


# ---------------------------------------------------------------------------
# Memory suite
# ---------------------------------------------------------------------------


class ComponentResolverMemSuite:
    """Benchmark memory consumption for component resolver operations."""

    def setup(self) -> None:
        self._resolver = ComponentResolver()
        self._resolver.register_global(_BenchProtocol, _BenchGlobal())
        for i in range(100):
            self._resolver.register_project(
                f"proj-{i}", _BenchProtocol, _BenchProject()
            )
            self._resolver.register_plan(f"plan-{i}", _BenchProtocol, _BenchPlan())

    def mem_resolve_many(self) -> list[Any]:
        """Memory for resolving 100 plan-level lookups."""
        results = []
        for i in range(100):
            results.append(
                self._resolver.resolve(
                    _BenchProtocol, plan_id=f"plan-{i}", project_id=f"proj-{i}"
                )
            )
        return results