cleveragents-core/benchmarks/phase_reversion_bench.py

"""ASV benchmarks for phase reversion state machine.

Measures the performance of:
- Plan.can_revert_to() validation
- PlanLifecycleService.revert_plan() overhead
- PlanLifecycleService.try_auto_revert_from_apply() overhead
- Decision recording during reversion
- Reversion loop guard check overhead
"""

from __future__ import annotations

import sys
from pathlib import Path

try:
    from cleveragents.application.services.plan_lifecycle_service import (
        PlanLifecycleService,
    )
    from cleveragents.config.settings import Settings
    from cleveragents.domain.models.core.plan import (
        AutomationProfileProvenance,
        AutomationProfileRef,
        PlanPhase,
        ProcessingState,
    )
except ModuleNotFoundError:
    sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "src"))
    from cleveragents.application.services.plan_lifecycle_service import (
        PlanLifecycleService,
    )
    from cleveragents.config.settings import Settings
    from cleveragents.domain.models.core.plan import (
        AutomationProfileProvenance,
        AutomationProfileRef,
        PlanPhase,
        ProcessingState,
    )


def _make_service() -> PlanLifecycleService:
    return PlanLifecycleService(settings=Settings())


def _create_plan_in_execute(
    service: PlanLifecycleService,
    action_name: str = "local/bench-action",
    profile: str = "manual",
) -> str:
    service.create_action(
        name=action_name,
        description="Benchmark action",
        definition_of_done="Tests pass",
        strategy_actor="local/strategist",
        execution_actor="local/executor",
    )
    plan = service.use_action(action_name)
    plan.automation_profile = AutomationProfileRef(
        profile_name=profile,
        provenance=AutomationProfileProvenance.PLAN,
    )
    plan_id = plan.identity.plan_id
    service.start_strategize(plan_id)
    service.complete_strategize(plan_id)
    service.execute_plan(plan_id)
    return plan_id


class CanRevertToSuite:
    """Benchmark Plan.can_revert_to() validation."""

    def setup(self) -> None:
        service = _make_service()
        plan_id = _create_plan_in_execute(service, "local/bench-can-revert")
        self.plan = service.get_plan(plan_id)

    def time_can_revert_to_valid(self) -> None:
        self.plan.can_revert_to(PlanPhase.STRATEGIZE)

    def time_can_revert_to_invalid(self) -> None:
        self.plan.can_revert_to(PlanPhase.APPLY)


class RevertPlanSuite:
    """Benchmark PlanLifecycleService.revert_plan() overhead."""

    def setup(self) -> None:
        self.service = _make_service()
        self.counter = 0

    def time_revert_plan(self) -> None:
        self.counter += 1
        name = f"local/bench-revert-{self.counter}"
        plan_id = _create_plan_in_execute(self.service, name)
        self.service.revert_plan(plan_id, PlanPhase.STRATEGIZE, reason="benchmark")


class AutoRevertSuite:
    """Benchmark auto-reversion from constrained apply."""

    def setup(self) -> None:
        self.service = _make_service()
        self.counter = 0

    def time_auto_revert_from_apply(self) -> None:
        self.counter += 1
        name = f"local/bench-auto-{self.counter}"
        self.service.create_action(
            name=name,
            description="Benchmark action",
            definition_of_done="Tests pass",
            strategy_actor="local/strategist",
            execution_actor="local/executor",
        )
        plan = self.service.use_action(name)
        plan.automation_profile = AutomationProfileRef(
            profile_name="ci",
            provenance=AutomationProfileProvenance.PLAN,
        )
        plan_id = plan.identity.plan_id
        self.service.start_strategize(plan_id)
        self.service.complete_strategize(plan_id)
        plan = self.service.get_plan(plan_id)
        if plan.phase != PlanPhase.EXECUTE:
            self.service.execute_plan(plan_id)
        self.service.start_execute(plan_id)
        self.service.complete_execute(plan_id)
        self.service.apply_plan(plan_id)
        self.service.start_apply(plan_id)
        plan = self.service.get_plan(plan_id)
        plan.processing_state = ProcessingState.CONSTRAINED
        self.service.try_auto_revert_from_apply(plan_id, "benchmark")


class LoopGuardSuite:
    """Benchmark loop guard check overhead."""

    def setup(self) -> None:
        self.service = _make_service()
        plan_id = _create_plan_in_execute(self.service, "local/bench-guard")
        self.plan = self.service.get_plan(plan_id)

    def time_loop_guard_under_limit(self) -> None:
        self.plan.reversion_count = 2
        self.plan.can_revert_to(PlanPhase.STRATEGIZE)

    def time_loop_guard_at_limit(self) -> None:
        self.plan.reversion_count = 3
        self.plan.can_revert_to(PlanPhase.STRATEGIZE)