cleveragents-core/benchmarks/m6_acceptance_bench.py

"""ASV benchmarks for M6 autonomy acceptance suite runtime.

Measures the performance of:
- A2A local facade dispatch operations
- Automation guard evaluation
- AutomationProfileService resolution precedence
- A2A event queue publish/subscribe
- Fixture loading overhead
"""

from __future__ import annotations

import importlib
import json
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)

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
from cleveragents.application.services.automation_profile_service import (  # noqa: E402
    AutomationProfileService,
)
from cleveragents.domain.models.core.automation_guard import (  # noqa: E402
    AutomationGuard,
)
from cleveragents.domain.models.core.automation_profile import (  # noqa: E402
    AutomationProfile,
)

_FIXTURES_DIR = Path(__file__).resolve().parents[1] / "features" / "fixtures" / "m6"


class M6FacadeDispatchSuite:
    """Benchmark A2A local facade dispatch operations."""

    def setup(self) -> None:
        self._facade = A2aLocalFacade()

    def time_session_create(self) -> None:
        """Benchmark session.create dispatch."""
        self._facade.dispatch(A2aRequest(method="session.create", params={}))

    def time_plan_create(self) -> None:
        """Benchmark plan.create dispatch."""
        self._facade.dispatch(A2aRequest(method="plan.create", params={}))

    def time_plan_execute(self) -> None:
        """Benchmark plan.execute dispatch."""
        self._facade.dispatch(
            A2aRequest(
                method="plan.execute",
                params={"plan_id": "01M6SM0KE00000000000000001"},
            )
        )

    def time_list_operations(self) -> None:
        """Benchmark list_operations call."""
        self._facade.list_operations()


class M6GuardEvaluationSuite:
    """Benchmark automation guard evaluation."""

    def setup(self) -> None:
        self._profile_denylist = AutomationProfile(
            name="bench-deny",
            guards=AutomationGuard(
                tool_denylist=["rm_rf", "drop_database", "format_disk"],
            ),
        )
        self._profile_allowlist = AutomationProfile(
            name="bench-allow",
            guards=AutomationGuard(
                tool_allowlist=["read_file", "search", "list_dir"],
            ),
        )
        self._profile_budget = AutomationProfile(
            name="bench-budget",
            guards=AutomationGuard(
                max_total_cost=100.0,
                max_tool_calls_per_step=10,
            ),
        )
        self._profile_no_guards = AutomationProfile(name="bench-noguard")

    def time_denylist_check_allowed(self) -> None:
        """Benchmark denylist check for allowed tool."""
        self._profile_denylist.check_guard(tool_name="read_file")

    def time_denylist_check_denied(self) -> None:
        """Benchmark denylist check for denied tool."""
        self._profile_denylist.check_guard(tool_name="rm_rf")

    def time_allowlist_check_allowed(self) -> None:
        """Benchmark allowlist check for allowed tool."""
        self._profile_allowlist.check_guard(tool_name="read_file")

    def time_allowlist_check_denied(self) -> None:
        """Benchmark allowlist check for unlisted tool."""
        self._profile_allowlist.check_guard(tool_name="write_file")

    def time_budget_check_under(self) -> None:
        """Benchmark budget check under limit."""
        self._profile_budget.check_guard(
            tool_name="llm_call",
            cost_so_far=10.0,
            calls_so_far=3,
        )

    def time_no_guards_check(self) -> None:
        """Benchmark check with no guards configured."""
        self._profile_no_guards.check_guard(tool_name="anything")


class M6ProfileResolutionSuite:
    """Benchmark AutomationProfileService resolution."""

    def setup(self) -> None:
        self._service = AutomationProfileService()

    def time_resolve_plan_level(self) -> None:
        """Benchmark resolution with plan-level override."""
        self._service.resolve_profile(
            plan_profile="ci",
            action_profile="auto",
            project_profile="manual",
        )

    def time_resolve_action_level(self) -> None:
        """Benchmark resolution with action-level override."""
        self._service.resolve_profile(
            plan_profile=None,
            action_profile="auto",
            project_profile="manual",
        )

    def time_resolve_global_default(self) -> None:
        """Benchmark resolution falling back to global default."""
        self._service.resolve_profile(
            plan_profile=None,
            action_profile=None,
            project_profile=None,
        )

    def time_get_builtin_profile(self) -> None:
        """Benchmark looking up a built-in profile."""
        self._service.get_profile("manual")

    def time_list_profiles(self) -> None:
        """Benchmark listing all profiles."""
        self._service.list_profiles()

    def time_evaluate_guard(self) -> None:
        """Benchmark guard evaluation through service."""
        self._service.evaluate_guard(
            profile_name="manual",
            tool_name="read_file",
        )

    def time_version_negotiation(self) -> None:
        """Benchmark A2A version negotiation."""
        negotiator = A2aVersionNegotiator()
        negotiator.negotiate("1.0")


class M6EventQueueSuite:
    """Benchmark A2A event queue operations."""

    def setup(self) -> None:
        self._queue = A2aEventQueue()

    def teardown(self) -> None:
        if not self._queue.is_closed:
            self._queue.close()

    def time_publish_event(self) -> None:
        """Benchmark publishing a single event."""
        self._queue.publish(A2aEvent(event_type="plan.progress", data={"step": 1}))

    def time_subscribe_and_publish(self) -> None:
        """Benchmark subscribe + publish cycle."""
        q = A2aEventQueue()
        sub_id = q.subscribe_local(lambda _e: None)
        q.publish(A2aEvent(event_type="test.event", data={}))
        q.unsubscribe(sub_id)
        q.close()

    def time_get_events(self) -> None:
        """Benchmark retrieving events from queue."""
        self._queue.get_events(limit=50)


class M6FixtureLoadSuite:
    """Benchmark loading M6 fixture files."""

    def time_load_a2a_facade_flows(self) -> None:
        """Benchmark loading a2a_facade_flows.json."""
        with open(_FIXTURES_DIR / "a2a_facade_flows.json") as f:
            json.load(f)

    def time_load_autonomy_guardrails(self) -> None:
        """Benchmark loading autonomy_guardrails.json."""
        with open(_FIXTURES_DIR / "autonomy_guardrails.json") as f:
            json.load(f)

    def time_load_automation_profiles(self) -> None:
        """Benchmark loading automation_profiles.json."""
        with open(_FIXTURES_DIR / "automation_profiles.json") as f:
            json.load(f)

    def time_load_all_fixtures(self) -> None:
        """Benchmark loading all M6 fixture files."""
        for fname in (
            "a2a_facade_flows.json",
            "autonomy_guardrails.json",
            "automation_profiles.json",
        ):
            with open(_FIXTURES_DIR / fname) as f:
                json.load(f)