cleveragents-core/benchmarks/a2a_facade_bench.py

"""ASV benchmarks for A2A facade dispatch, version negotiation, and event queue.

Measures the performance of:
- A2aLocalFacade dispatch overhead per operation
- A2aVersionNegotiator negotiation throughput
- A2aEventQueue publish and get_events throughput
"""

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.a2a.events import A2aEventQueue  # noqa: E402
from cleveragents.a2a.facade import A2aLocalFacade  # noqa: E402
from cleveragents.a2a.models import A2aEvent, A2aRequest  # noqa: E402
from cleveragents.a2a.versioning import A2aVersionNegotiator  # noqa: E402

# ---------------------------------------------------------------------------
# Facade dispatch benchmarks
# ---------------------------------------------------------------------------


class FacadeDispatchSuite:
    """Benchmark A2aLocalFacade.dispatch() overhead."""

    timeout = 60

    def setup(self) -> None:
        self.facade = A2aLocalFacade()
        self.session_req = A2aRequest(method="session.create", params={})
        self.plan_req = A2aRequest(
            method="plan.execute", params={"plan_id": "BENCH001"}
        )
        self.context_req = A2aRequest(method="context.get", params={})

    def time_dispatch_session_create(self) -> None:
        self.facade.dispatch(self.session_req)

    def time_dispatch_plan_execute(self) -> None:
        self.facade.dispatch(self.plan_req)

    def time_dispatch_context_get(self) -> None:
        self.facade.dispatch(self.context_req)

    def time_dispatch_all_operations(self) -> None:
        for op in self.facade.list_operations():
            req = A2aRequest(method=op, params={})
            self.facade.dispatch(req)

    def time_list_operations(self) -> None:
        self.facade.list_operations()


# ---------------------------------------------------------------------------
# Version negotiation benchmarks
# ---------------------------------------------------------------------------


class VersionNegotiationSuite:
    """Benchmark A2aVersionNegotiator throughput."""

    timeout = 60

    def setup(self) -> None:
        self.negotiator = A2aVersionNegotiator()

    def time_negotiate_supported(self) -> None:
        self.negotiator.negotiate("1.0")

    def time_is_supported_true(self) -> None:
        self.negotiator.is_supported("1.0")

    def time_is_supported_false(self) -> None:
        self.negotiator.is_supported("99.0")

    def time_get_current(self) -> None:
        self.negotiator.get_current()


# ---------------------------------------------------------------------------
# Event queue benchmarks
# ---------------------------------------------------------------------------


class EventQueueSuite:
    """Benchmark A2aEventQueue throughput."""

    timeout = 60

    def setup(self) -> None:
        self.queue = A2aEventQueue()
        self.event = A2aEvent(event_type="bench.event")

    def time_publish_single(self) -> None:
        self.queue.publish(self.event)

    def time_publish_100(self) -> None:
        for _ in range(100):
            self.queue.publish(self.event)

    def time_get_events_default(self) -> None:
        self.queue.get_events()

    def time_subscribe_local(self) -> None:
        self.queue.subscribe_local(lambda e: None)

    def time_publish_with_subscriber(self) -> None:
        q = A2aEventQueue()
        q.subscribe_local(lambda e: None)
        for _ in range(100):
            q.publish(self.event)