"""ASV benchmarks for M4 correction + subplan smoke suite runtime. Measures the performance of: - Correction CLI commands (revert, append, dry-run) - Subplan status rendering with config and statuses - SubplanFailureHandler decision logic - Fixture loading overhead """ from __future__ import annotations import importlib import json import sys from datetime import UTC, 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.plan import app as plan_app # noqa: E402 from cleveragents.domain.models.core.correction import ( # noqa: E402 CorrectionImpact, CorrectionResult, CorrectionStatus, ) from cleveragents.domain.models.core.plan import ( # noqa: E402 ExecutionMode, NamespacedName, Plan, PlanIdentity, PlanPhase, PlanTimestamps, ProcessingState, SubplanConfig, SubplanFailureHandler, SubplanMergeStrategy, SubplanStatus, ) _runner = CliRunner() _PLAN_ULID = "01M4SM0KE00000000000000001" _DECISION_ULID = "01M4DEC00000000000000000001" _FIXTURES_DIR = Path(__file__).resolve().parents[1] / "features" / "fixtures" / "m4" def _mock_plan( *, phase: PlanPhase = PlanPhase.EXECUTE, state: ProcessingState = ProcessingState.COMPLETE, subplan_config: SubplanConfig | None = None, subplan_statuses: list[SubplanStatus] | None = None, ) -> Plan: now = datetime.now() return Plan( identity=PlanIdentity(plan_id=_PLAN_ULID), namespaced_name=NamespacedName.parse("local/m4-smoke-plan"), description="M4 bench plan", definition_of_done="Corrections applied", action_name="local/m4-correction-test", phase=phase, processing_state=state, project_links=[], 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), decision_root_id=_DECISION_ULID, subplan_config=subplan_config, subplan_statuses=subplan_statuses or [], ) def _mock_correction_service() -> MagicMock: mock_svc = MagicMock() now = datetime.now(UTC) _correction_id = "01M4CORRID00000000000000001" # request_correction returns a CorrectionRequest-like mock mock_request = MagicMock() mock_request.correction_id = _correction_id mock_svc.request_correction.return_value = mock_request # execute_correction returns a CorrectionResult (non-dry-run path) mock_svc.execute_correction.return_value = CorrectionResult( correction_id=_correction_id, status=CorrectionStatus.APPLIED, reverted_decisions=[_DECISION_ULID], completed_at=now, ) # analyze_impact returns a CorrectionImpact (dry-run path) mock_svc.analyze_impact.return_value = CorrectionImpact( affected_decisions=[_DECISION_ULID], affected_files=["src/auth.py"], risk_level="medium", rollback_tier="phase", ) return mock_svc class M4CorrectionCLISuite: """Benchmark correction CLI commands.""" def setup(self) -> None: self._mock_service = MagicMock() self._mock_service.get_plan.return_value = _mock_plan() self._mock_correction = _mock_correction_service() self._plan_patcher = patch( "cleveragents.cli.commands.plan._get_lifecycle_service", return_value=self._mock_service, ) self._correction_patcher = patch( "cleveragents.application.services.correction_service.CorrectionService", return_value=self._mock_correction, ) self._plan_patcher.start() self._correction_patcher.start() def teardown(self) -> None: self._correction_patcher.stop() self._plan_patcher.stop() def time_correction_revert(self) -> None: """Benchmark correction revert command.""" _runner.invoke( plan_app, [ "correct", _DECISION_ULID, "--mode", "revert", "--guidance", "bench test", "--plan", _PLAN_ULID, "--yes", ], ) def time_correction_append(self) -> None: """Benchmark correction append command.""" _runner.invoke( plan_app, [ "correct", _DECISION_ULID, "--mode", "append", "--guidance", "bench test", "--plan", _PLAN_ULID, "--yes", ], ) def time_correction_dry_run(self) -> None: """Benchmark correction dry-run command.""" _runner.invoke( plan_app, [ "correct", _DECISION_ULID, "--mode", "revert", "--guidance", "Evaluate impact of revert", "--dry-run", "--plan", _PLAN_ULID, ], ) class M4SubplanStatusSuite: """Benchmark subplan status rendering.""" def setup(self) -> None: config = SubplanConfig( execution_mode=ExecutionMode.PARALLEL, merge_strategy=SubplanMergeStrategy.GIT_THREE_WAY, max_parallel=5, ) statuses = [ SubplanStatus( subplan_id="01J4S0BP000000000000000020", action_name="local/update-api", target_resources=["src/api.py"], status=ProcessingState.COMPLETE, ), SubplanStatus( subplan_id="01J4S0BP000000000000000021", action_name="local/update-tests", target_resources=["tests/test_api.py"], status=ProcessingState.COMPLETE, ), ] plan = _mock_plan(subplan_config=config, subplan_statuses=statuses) self._mock_service = MagicMock() self._mock_service.get_plan.return_value = plan self._mock_service.list_plans.return_value = [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_status_with_subplans_json(self) -> None: """Benchmark plan status JSON rendering with subplans.""" _runner.invoke(plan_app, ["status", _PLAN_ULID, "--format", "json"]) def time_status_with_subplans_rich(self) -> None: """Benchmark plan status rich rendering with subplans.""" _runner.invoke(plan_app, ["status", _PLAN_ULID]) class M4FailureHandlerSuite: """Benchmark SubplanFailureHandler decision logic.""" def setup(self) -> None: self._handler = SubplanFailureHandler() self._config_ff = SubplanConfig( execution_mode=ExecutionMode.PARALLEL, fail_fast=True ) self._config_retry = SubplanConfig( execution_mode=ExecutionMode.SEQUENTIAL, retry_failed=True, max_retries=2, ) self._failed_retriable = SubplanStatus( subplan_id="01J4S0BP000000000000000030", action_name="local/flaky-task", status=ProcessingState.ERRORED, error="ValidationError: schema mismatch", attempt_number=1, ) self._failed_non_retriable = SubplanStatus( subplan_id="01J4S0BP000000000000000031", action_name="local/broken-task", status=ProcessingState.ERRORED, error="ConfigurationError: permanent failure", attempt_number=1, ) def time_should_stop_others_fail_fast(self) -> None: """Benchmark should_stop_others with fail_fast.""" self._handler.should_stop_others(self._config_ff, self._failed_retriable) def time_should_retry_retriable(self) -> None: """Benchmark should_retry for retriable error.""" self._handler.should_retry(self._config_retry, self._failed_retriable) def time_should_retry_non_retriable(self) -> None: """Benchmark should_retry for non-retriable error.""" self._handler.should_retry(self._config_retry, self._failed_non_retriable) class M4FixtureLoadSuite: """Benchmark loading M4 fixture files.""" def time_load_correction_flows(self) -> None: """Benchmark loading correction_flows.json.""" with open(_FIXTURES_DIR / "correction_flows.json") as f: json.load(f) def time_load_subplan_execution(self) -> None: """Benchmark loading subplan_execution.json.""" with open(_FIXTURES_DIR / "subplan_execution.json") as f: json.load(f) def time_load_conflict_simulations(self) -> None: """Benchmark loading conflict_simulations.json.""" with open(_FIXTURES_DIR / "conflict_simulations.json") as f: json.load(f) def time_load_all_fixtures(self) -> None: """Benchmark loading all M4 fixture files.""" for fname in ( "correction_flows.json", "subplan_execution.json", "conflict_simulations.json", ): with open(_FIXTURES_DIR / fname) as f: json.load(f)