"""ASV benchmarks for M1 source-code plan lifecycle CLI overhead. Measures the performance of: - Action create from YAML config - Plan use with arguments and project links - Plan execute phase transition - Plan apply terminal transition - Full lifecycle (action create -> plan use -> execute -> apply) """ from __future__ import annotations import importlib import json import sys from datetime import datetime from pathlib import Path from unittest.mock import MagicMock, patch # 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 typer.testing import CliRunner # noqa: E402 from cleveragents.cli.commands.action import app as action_app # noqa: E402 from cleveragents.cli.commands.plan import app as plan_app # noqa: E402 from cleveragents.domain.models.core.action import ( # noqa: E402 Action, ActionState, ) from cleveragents.domain.models.core.plan import ( # noqa: E402 NamespacedName, Plan, PlanIdentity, PlanPhase, PlanTimestamps, ProcessingState, ProjectLink, ) _runner = CliRunner() _PLAN_ULID = "01M1SM0KE00000000000000001" _FIXTURES_DIR = Path(__file__).resolve().parents[1] / "features" / "fixtures" / "m1" def _mock_action(name: str = "local/m1-source-review") -> Action: return Action( namespaced_name=NamespacedName.parse(name), description="Minimal source-code review action", long_description=None, definition_of_done="Source code reviewed", strategy_actor="openai/gpt-4", execution_actor="openai/gpt-4", reusable=True, read_only=True, state=ActionState.AVAILABLE, created_by=None, created_at=datetime.now(), updated_at=datetime.now(), ) def _mock_plan( *, phase: PlanPhase = PlanPhase.STRATEGIZE, state: ProcessingState = ProcessingState.QUEUED, project_links: list[ProjectLink] | None = None, ) -> Plan: now = datetime.now() return Plan( identity=PlanIdentity(plan_id=_PLAN_ULID), namespaced_name=NamespacedName.parse("local/m1-smoke-plan"), description="M1 bench plan", definition_of_done="Source code reviewed", action_name="local/m1-source-review", phase=phase, processing_state=state, project_links=project_links or [], arguments={}, arguments_order=[], strategy_actor="openai/gpt-4", execution_actor="openai/gpt-4", reusable=True, read_only=False, created_by=None, timestamps=PlanTimestamps(created_at=now, updated_at=now), ) class M1ActionCreateSuite: """Benchmark action create from YAML config.""" def setup(self) -> None: self._mock_service = MagicMock() self._mock_service.create_action.return_value = _mock_action() self._patcher = patch( "cleveragents.cli.commands.action._get_lifecycle_service", return_value=self._mock_service, ) self._patcher.start() def teardown(self) -> None: self._patcher.stop() def time_action_create_from_yaml(self) -> None: """Benchmark action create with YAML config.""" config_path = _FIXTURES_DIR / "action_sourcecode.yaml" _runner.invoke(action_app, ["create", "--config", str(config_path)]) class M1PlanUseSuite: """Benchmark plan use with various arguments.""" def setup(self) -> None: self._mock_service = MagicMock() self._mock_service.get_action_by_name.return_value = _mock_action() self._mock_service.use_action.return_value = _mock_plan() self._patcher = patch( "cleveragents.cli.commands.plan._get_lifecycle_service", return_value=self._mock_service, ) self._patcher.start() def teardown(self) -> None: self._patcher.stop() def time_plan_use_minimal(self) -> None: """Benchmark plan use with minimal arguments.""" _runner.invoke(plan_app, ["use", "local/m1-source-review"]) def time_plan_use_with_project(self) -> None: """Benchmark plan use with project argument.""" _runner.invoke( plan_app, ["use", "local/m1-source-review", "local/m1-smoke-proj"] ) def time_plan_use_with_args(self) -> None: """Benchmark plan use with --arg flags.""" _runner.invoke( plan_app, [ "use", "local/m1-source-review", "--arg", "target_path=src/", "--arg", "depth=3", ], ) def time_plan_use_plain_format(self) -> None: """Benchmark plan use with --format plain.""" _runner.invoke( plan_app, ["use", "local/m1-source-review", "--format", "plain"], ) class M1PlanExecuteSuite: """Benchmark plan execute phase transition.""" def setup(self) -> None: self._mock_service = MagicMock() strategize_plan = _mock_plan( phase=PlanPhase.STRATEGIZE, state=ProcessingState.COMPLETE ) self._mock_service.list_plans.return_value = [strategize_plan] self._mock_service.execute_plan.return_value = _mock_plan( phase=PlanPhase.EXECUTE, state=ProcessingState.QUEUED ) self._patcher = patch( "cleveragents.cli.commands.plan._get_lifecycle_service", return_value=self._mock_service, ) self._patcher.start() def teardown(self) -> None: self._patcher.stop() def time_plan_execute(self) -> None: """Benchmark plan execute command.""" _runner.invoke(plan_app, ["execute"]) class M1PlanApplySuite: """Benchmark plan apply terminal transition.""" def setup(self) -> None: self._mock_service = MagicMock() self._mock_service.apply_plan.return_value = _mock_plan( phase=PlanPhase.APPLY, state=ProcessingState.APPLIED ) self._patcher = patch( "cleveragents.cli.commands.plan._get_lifecycle_service", return_value=self._mock_service, ) self._patcher.start() def teardown(self) -> None: self._patcher.stop() def time_plan_apply(self) -> None: """Benchmark plan apply command.""" _runner.invoke(plan_app, ["apply", "--yes", _PLAN_ULID]) class M1FixtureLoadSuite: """Benchmark loading M1 fixture files.""" def time_load_git_repo_fixture(self) -> None: """Benchmark loading git_repo.json fixture.""" fixture_path = _FIXTURES_DIR / "git_repo.json" with open(fixture_path) as f: json.load(f) def time_load_git_checkout_fixture(self) -> None: """Benchmark loading git_checkout_resource.json fixture.""" fixture_path = _FIXTURES_DIR / "git_checkout_resource.json" with open(fixture_path) as f: json.load(f)