cleveragents-core/benchmarks/cli_robot_flow_bench.py

"""ASV benchmarks for CLI Robot flow fixture setup overhead.

Measures the performance overhead of:
- Fixture setup for action creation (YAML write + parse)
- Mock service initialisation for plan lifecycle commands
- ChangeSet store creation and entry recording
- Full lifecycle fixture setup (action + plan + changeset + sandbox)
"""

from __future__ import annotations

import os
import sys
import tempfile
from datetime import datetime
from pathlib import Path
from unittest.mock import MagicMock, patch

try:
    from cleveragents.cli.commands.action import app as action_app
    from cleveragents.cli.commands.plan import app as plan_app
    from cleveragents.domain.models.core.action import Action, ActionState
    from cleveragents.domain.models.core.change import (
        ChangeEntry,
        ChangeOperation,
        InMemoryChangeSetStore,
    )
    from cleveragents.domain.models.core.plan import (
        AutomationProfileProvenance,
        AutomationProfileRef,
        NamespacedName,
        Plan,
        PlanIdentity,
        PlanPhase,
        PlanTimestamps,
        ProcessingState,
        ProjectLink,
    )
except ModuleNotFoundError:
    sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "src"))
    from cleveragents.cli.commands.action import app as action_app
    from cleveragents.cli.commands.plan import app as plan_app
    from cleveragents.domain.models.core.action import Action, ActionState
    from cleveragents.domain.models.core.change import (
        ChangeEntry,
        ChangeOperation,
        InMemoryChangeSetStore,
    )
    from cleveragents.domain.models.core.plan import (
        AutomationProfileProvenance,
        AutomationProfileRef,
        NamespacedName,
        Plan,
        PlanIdentity,
        PlanPhase,
        PlanTimestamps,
        ProcessingState,
        ProjectLink,
    )

from typer.testing import CliRunner

_PLAN_ULID = "01KHDE6WWS2171PWW3GJEBXZBN"

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

_runner = CliRunner()


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


def _mock_plan(
    phase: PlanPhase = PlanPhase.STRATEGIZE,
    state: ProcessingState = ProcessingState.QUEUED,
) -> Plan:
    """Create a minimal Plan for benchmarks."""
    now = datetime.now()
    return Plan(
        identity=PlanIdentity(plan_id=_PLAN_ULID),
        namespaced_name=NamespacedName.parse("local/bench-plan"),
        description="Benchmark plan",
        definition_of_done="Benchmarks pass",
        action_name="local/bench-lifecycle-action",
        phase=phase,
        processing_state=state,
        project_links=[ProjectLink(project_name="proj-bench")],
        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 CLIRobotFixtureSetupSuite:
    """Benchmark fixture setup overhead for CLI Robot flow tests."""

    def setup(self) -> None:
        """Prepare reusable mock objects."""
        self._mock_service = MagicMock()
        self._mock_service.create_action.return_value = _mock_action()
        self._mock_service.get_action_by_name.return_value = _mock_action()
        self._mock_service.use_action.return_value = _mock_plan()
        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
        )

    def time_yaml_fixture_write(self) -> None:
        """Benchmark writing a YAML fixture file."""
        fd, path = tempfile.mkstemp(suffix=".yaml")
        with os.fdopen(fd, "w") as fh:
            fh.write(_VALID_YAML)
        Path(path).unlink(missing_ok=True)

    def time_mock_service_creation(self) -> None:
        """Benchmark creating and configuring a mock service."""
        svc = MagicMock()
        svc.create_action.return_value = _mock_action()
        svc.get_action_by_name.return_value = _mock_action()
        svc.use_action.return_value = _mock_plan()

    def time_changeset_store_creation(self) -> None:
        """Benchmark creating an InMemoryChangeSetStore and starting a changeset."""
        store = InMemoryChangeSetStore()
        store.start(_PLAN_ULID)

    def time_changeset_entry_recording(self) -> None:
        """Benchmark recording a change entry in the store."""
        store = InMemoryChangeSetStore()
        cid = store.start(_PLAN_ULID)
        store.record(
            cid,
            ChangeEntry(
                plan_id=_PLAN_ULID,
                resource_id="res-bench",
                tool_name="builtin/file-write",
                operation=ChangeOperation.CREATE,
                path="src/bench.py",
            ),
        )


class CLIRobotLifecycleSuite:
    """Benchmark the full CLI lifecycle flow with mocked service."""

    def setup(self) -> None:
        """Prepare mock service and YAML file."""
        self._mock_service = MagicMock()
        self._mock_service.create_action.return_value = _mock_action()
        self._mock_service.get_action_by_name.return_value = _mock_action()
        self._mock_service.use_action.return_value = _mock_plan()
        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
        )
        fd, self._yaml_path = tempfile.mkstemp(suffix=".yaml")
        with os.fdopen(fd, "w") as fh:
            fh.write(_VALID_YAML)
        self._action_patcher = patch(
            "cleveragents.cli.commands.action._get_lifecycle_service",
            return_value=self._mock_service,
        )
        self._plan_patcher = patch(
            "cleveragents.cli.commands.plan._get_lifecycle_service",
            return_value=self._mock_service,
        )
        self._action_patcher.start()
        self._plan_patcher.start()

    def teardown(self) -> None:
        """Clean up patchers and temp file."""
        self._action_patcher.stop()
        self._plan_patcher.stop()
        Path(self._yaml_path).unlink(missing_ok=True)

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

    def time_plan_use(self) -> None:
        """Benchmark plan use via CLI."""
        _runner.invoke(plan_app, ["use", "local/bench-lifecycle-action", "proj-bench"])

    def time_plan_execute(self) -> None:
        """Benchmark plan execute via CLI."""
        _runner.invoke(plan_app, ["execute", _PLAN_ULID])

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

    def time_full_lifecycle_flow(self) -> None:
        """Benchmark the full create -> use -> execute -> apply flow."""
        _runner.invoke(action_app, ["create", "--config", self._yaml_path])
        _runner.invoke(plan_app, ["use", "local/bench-lifecycle-action", "proj-bench"])
        _runner.invoke(plan_app, ["execute", _PLAN_ULID])
        _runner.invoke(plan_app, ["apply", "--yes", _PLAN_ULID])