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