cleveragents-core/benchmarks/m4_smoke_bench.py

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