cleveragents-core/benchmarks/plan_phase_bench.py

"""ASV benchmarks for plan phase transition validation overhead.

Measures the cost of:
- ``can_transition()`` lookups
- ``Plan.get_next_phase()`` traversals
- ``Plan.is_terminal`` property checks
- ``Plan.can_transition_to_next_phase`` property checks
"""

from cleveragents.domain.models.core.plan import (
    NamespacedName,
    Plan,
    PlanIdentity,
    PlanPhase,
    ProcessingState,
    can_transition,
)


def _make_plan(
    phase: PlanPhase = PlanPhase.STRATEGIZE,
    processing_state: ProcessingState = ProcessingState.QUEUED,
) -> Plan:
    return Plan(
        identity=PlanIdentity(plan_id="01ARZ3NDEKTSV4RRFFQ69G5FAV"),
        namespaced_name=NamespacedName(server=None, namespace="local", name="bench"),
        description="Benchmark plan",
        action_name="local/bench-action",
        phase=phase,
        processing_state=processing_state,
    )


class TimeCanTransition:
    """Benchmark ``can_transition()`` function."""

    def time_valid_transition_strategize_to_execute(self) -> None:
        can_transition(PlanPhase.STRATEGIZE, PlanPhase.EXECUTE)

    def time_valid_transition_action_to_strategize(self) -> None:
        can_transition(PlanPhase.ACTION, PlanPhase.STRATEGIZE)

    def time_invalid_transition_strategize_to_apply(self) -> None:
        can_transition(PlanPhase.STRATEGIZE, PlanPhase.APPLY)

    def time_revert_execute_to_strategize(self) -> None:
        can_transition(PlanPhase.EXECUTE, PlanPhase.STRATEGIZE)

    def time_revert_apply_to_strategize(self) -> None:
        can_transition(PlanPhase.APPLY, PlanPhase.STRATEGIZE)


class TimeGetNextPhase:
    """Benchmark ``Plan.get_next_phase()``."""

    def setup(self) -> None:
        self.plan_action = _make_plan(PlanPhase.ACTION)
        self.plan_strategize = _make_plan(PlanPhase.STRATEGIZE)
        self.plan_execute = _make_plan(PlanPhase.EXECUTE)
        self.plan_apply = _make_plan(PlanPhase.APPLY)

    def time_next_from_action(self) -> None:
        self.plan_action.get_next_phase()

    def time_next_from_strategize(self) -> None:
        self.plan_strategize.get_next_phase()

    def time_next_from_execute(self) -> None:
        self.plan_execute.get_next_phase()

    def time_next_from_apply(self) -> None:
        self.plan_apply.get_next_phase()


class TimeIsTerminal:
    """Benchmark ``Plan.is_terminal`` property."""

    def setup(self) -> None:
        self.plan_queued = _make_plan(PlanPhase.STRATEGIZE, ProcessingState.QUEUED)
        self.plan_applied = _make_plan(PlanPhase.APPLY, ProcessingState.APPLIED)
        self.plan_constrained = _make_plan(PlanPhase.APPLY, ProcessingState.CONSTRAINED)
        self.plan_cancelled = _make_plan(
            PlanPhase.STRATEGIZE, ProcessingState.CANCELLED
        )

    def time_non_terminal(self) -> None:
        self.plan_queued.is_terminal

    def time_applied_terminal(self) -> None:
        self.plan_applied.is_terminal

    def time_constrained_terminal(self) -> None:
        self.plan_constrained.is_terminal

    def time_cancelled_terminal(self) -> None:
        self.plan_cancelled.is_terminal


class TimeCanTransitionToNextPhase:
    """Benchmark ``Plan.can_transition_to_next_phase`` property."""

    def setup(self) -> None:
        self.plan_action = _make_plan(PlanPhase.ACTION, ProcessingState.QUEUED)
        self.plan_strat_complete = _make_plan(
            PlanPhase.STRATEGIZE, ProcessingState.COMPLETE
        )
        self.plan_apply = _make_plan(PlanPhase.APPLY, ProcessingState.QUEUED)

    def time_action_can_transition(self) -> None:
        self.plan_action.can_transition_to_next_phase

    def time_strategize_complete_can_transition(self) -> None:
        self.plan_strat_complete.can_transition_to_next_phase

    def time_apply_cannot_transition(self) -> None:
        self.plan_apply.can_transition_to_next_phase