cleveragents-core/benchmarks/plan_cli_smoke_bench.py

"""ASV benchmarks for CLI argument parsing overhead.

Measures the performance of:
- Action CLI argument parsing (create, list, show, archive)
- Plan CLI argument parsing (use with various flag combinations)
- Plan lifecycle commands (execute, apply, status, cancel)

These benchmarks focus on CLI overhead (argument parsing + routing)
rather than service layer execution time, since the service is mocked.
"""

from __future__ import annotations

import importlib
import os
import sys
import tempfile
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 Action, ActionState  # noqa: E402
from cleveragents.domain.models.core.plan import (  # noqa: E402
    AutomationProfileProvenance,
    AutomationProfileRef,
    NamespacedName,
    Plan,
    PlanIdentity,
    PlanPhase,
    PlanTimestamps,
    ProcessingState,
    ProjectLink,
)

_runner = CliRunner()
_PLAN_ULID = "01KHDE6WWS2171PWW3GJEBXZ8S"

_VALID_YAML = """\
name: local/bench-smoke-action
description: Smoke bench action
strategy_actor: openai/gpt-4
execution_actor: openai/gpt-4
definition_of_done: Benchmarks pass
"""


def _mock_action(name: str = "local/bench-smoke-action") -> Action:
    return Action(
        namespaced_name=NamespacedName.parse(name),
        description="Smoke bench action",
        long_description=None,
        definition_of_done="Benchmarks pass",
        strategy_actor="openai/gpt-4",
        execution_actor="openai/gpt-4",
        reusable=True,
        read_only=False,
        state=ActionState.AVAILABLE,
        created_by=None,
        created_at=datetime.now(),
        updated_at=datetime.now(),
    )


def _mock_plan(
    name: str = "local/bench-smoke-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(name),
        description="Smoke bench plan",
        definition_of_done="Benchmarks pass",
        action_name="local/bench-smoke-action",
        phase=phase,
        processing_state=state,
        project_links=project_links or [],
        arguments={"target_coverage": 80},
        arguments_order=["target_coverage"],
        automation_profile=AutomationProfileRef(
            profile_name="trusted",
            provenance=AutomationProfileProvenance.PLAN,
        ),
        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 ActionCLIParsingSuite:
    """Benchmark action CLI argument parsing overhead."""

    def setup(self) -> None:
        """Set up mock service and temp config file."""
        self._mock_service = MagicMock()
        self._mock_service.create_action.return_value = _mock_action()
        self._mock_service.list_actions.return_value = [
            _mock_action(f"local/action-{i}") for i in range(20)
        ]
        self._mock_service.get_action_by_name.return_value = _mock_action()
        self._mock_service.archive_action.return_value = _mock_action()

        self._patcher = patch(
            "cleveragents.cli.commands.action._get_lifecycle_service",
            return_value=self._mock_service,
        )
        self._patcher.start()

        # Create temp YAML file
        fd, self._yaml_path = tempfile.mkstemp(suffix=".yaml")
        with os.fdopen(fd, "w") as fh:
            fh.write(_VALID_YAML)

    def teardown(self) -> None:
        self._patcher.stop()
        if os.path.exists(self._yaml_path):
            os.unlink(self._yaml_path)

    def time_action_create(self) -> None:
        """Benchmark action create with --config."""
        _runner.invoke(action_app, ["create", "--config", self._yaml_path])

    def time_action_list(self) -> None:
        """Benchmark action list."""
        _runner.invoke(action_app, ["list"])

    def time_action_list_with_filters(self) -> None:
        """Benchmark action list with namespace and state filters."""
        _runner.invoke(
            action_app, ["list", "--namespace", "local", "--state", "available"]
        )

    def time_action_show(self) -> None:
        """Benchmark action show."""
        _runner.invoke(action_app, ["show", "local/bench-smoke-action"])

    def time_action_archive(self) -> None:
        """Benchmark action archive."""
        _runner.invoke(action_app, ["archive", "local/bench-smoke-action"])


class PlanUseArgParsingSuite:
    """Benchmark plan use CLI argument parsing with various flag combinations."""

    def setup(self) -> None:
        """Set up mock service."""
        self._mock_service = MagicMock()
        self._mock_service.get_action_by_name.return_value = _mock_action()
        self._mock_service.use_action.return_value = _mock_plan(
            project_links=[
                ProjectLink(project_name="proj-a"),
                ProjectLink(project_name="proj-b"),
            ]
        )
        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_use_minimal(self) -> None:
        """Benchmark plan use with minimal args."""
        _runner.invoke(plan_app, ["use", "local/bench-smoke-action", "proj-a"])

    def time_use_multi_project(self) -> None:
        """Benchmark plan use with multiple projects."""
        _runner.invoke(
            plan_app,
            ["use", "local/bench-smoke-action", "proj-a", "proj-b", "proj-c"],
        )

    def time_use_with_all_flags(self) -> None:
        """Benchmark plan use with all optional flags."""
        _runner.invoke(
            plan_app,
            [
                "use",
                "local/bench-smoke-action",
                "proj-a",
                "--automation-profile",
                "trusted",
                "--invariant",
                "No warnings",
                "--invariant",
                "Keep compat",
                "--strategy-actor",
                "openai/gpt-4",
                "--execution-actor",
                "openai/gpt-4",
                "--estimation-actor",
                "openai/gpt-4",
                "--invariant-actor",
                "openai/gpt-4",
                "--automation-level",
                "full_automation",
                "--arg",
                "target_coverage=80",
                "--arg",
                "verbose=true",
            ],
        )

    def time_use_with_args_only(self) -> None:
        """Benchmark plan use with multiple --arg flags."""
        _runner.invoke(
            plan_app,
            [
                "use",
                "local/bench-smoke-action",
                "proj-a",
                "--arg",
                "coverage=80",
                "--arg",
                "verbose=true",
                "--arg",
                "ratio=3.14",
            ],
        )


class PlanLifecycleCmdParsingSuite:
    """Benchmark plan lifecycle command parsing overhead."""

    def setup(self) -> None:
        """Set up mock service."""
        self._mock_service = MagicMock()
        self._mock_service.execute_plan.return_value = _mock_plan(
            phase=PlanPhase.EXECUTE, state=ProcessingState.QUEUED
        )
        self._mock_service.apply_plan.return_value = _mock_plan(
            phase=PlanPhase.APPLY, state=ProcessingState.QUEUED
        )
        self._mock_service.get_plan.return_value = _mock_plan(
            project_links=[ProjectLink(project_name="proj-a", alias="api")]
        )
        self._mock_service.cancel_plan.return_value = _mock_plan(
            phase=PlanPhase.STRATEGIZE, state=ProcessingState.CANCELLED
        )
        self._mock_service.list_plans.return_value = [
            _mock_plan(f"local/plan-{i}") for i in range(10)
        ]
        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_execute(self) -> None:
        """Benchmark plan execute."""
        _runner.invoke(plan_app, ["execute", _PLAN_ULID])

    def time_apply(self) -> None:
        """Benchmark plan apply."""
        _runner.invoke(plan_app, ["apply", "--yes", _PLAN_ULID])

    def time_status_by_id(self) -> None:
        """Benchmark plan status with specific ID."""
        _runner.invoke(plan_app, ["status", _PLAN_ULID])

    def time_status_list_all(self) -> None:
        """Benchmark plan status listing all plans."""
        _runner.invoke(plan_app, ["status"])

    def time_cancel_with_reason(self) -> None:
        """Benchmark plan cancel with reason."""
        _runner.invoke(plan_app, ["cancel", _PLAN_ULID, "--reason", "Benchmarking"])

    def time_list(self) -> None:
        """Benchmark list."""
        _runner.invoke(plan_app, ["list"])

    def time_list_filtered(self) -> None:
        """Benchmark list with filters."""
        _runner.invoke(
            plan_app,
            ["list", "--phase", "strategize", "--state", "queued"],
        )