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