"""Airspeed Velocity benchmarks for subplan model validation. Measures construction, serialization, and failure-handler performance for SubplanConfig, SubplanStatus, SubplanFailureHandler, and Plan hierarchy properties. """ from __future__ import annotations from cleveragents.domain.models.core.plan import ( ExecutionMode, NamespacedName, Plan, PlanIdentity, ProcessingState, SubplanConfig, SubplanFailureHandler, SubplanMergeStrategy, SubplanStatus, ) # Valid ULID constants _ROOT_ULID = "01HGZ6FE0AQDYTR4BXVQZ6EA00" _PARENT_ULID = "01HGZ6FE0AQDYTR4BXVQZ6EB00" _CHILD_ULID = "01HGZ6FE0AQDYTR4BXVQZ6EC00" _STATUS_ULID = "01KH72MDGQMPRW3R5WPZVB8KC1" def _root_plan() -> Plan: """Create a root plan with no parent.""" return Plan( identity=PlanIdentity(plan_id=_ROOT_ULID), namespaced_name=NamespacedName( server=None, namespace="local", name="bench-root" ), description="Benchmark root plan", action_name="local/bench-action", ) def _child_plan() -> Plan: """Create a child plan with parent and root references.""" return Plan( identity=PlanIdentity( plan_id=_CHILD_ULID, parent_plan_id=_PARENT_ULID, root_plan_id=_ROOT_ULID, ), namespaced_name=NamespacedName( server=None, namespace="local", name="bench-child" ), description="Benchmark child plan", action_name="local/bench-action", ) def _plan_with_subplans() -> Plan: """Create a plan with subplan config and statuses.""" return Plan( identity=PlanIdentity(plan_id=_ROOT_ULID), namespaced_name=NamespacedName( server=None, namespace="local", name="bench-parent" ), description="Benchmark parent plan with subplans", action_name="local/bench-action", subplan_config=SubplanConfig( execution_mode=ExecutionMode.PARALLEL, max_parallel=5, fail_fast=True, ), subplan_statuses=[ SubplanStatus( subplan_id=_STATUS_ULID, action_name="local/bench-sub", target_resources=["src/main.py"], status=ProcessingState.QUEUED, ), ], ) class SubplanConfigValidationSuite: """Benchmark SubplanConfig construction and validation.""" def time_default_config_creation(self) -> None: """Time creating a SubplanConfig with all defaults.""" SubplanConfig() def time_custom_config_creation(self) -> None: """Time creating a SubplanConfig with custom settings.""" SubplanConfig( execution_mode=ExecutionMode.PARALLEL, merge_strategy=SubplanMergeStrategy.FAIL_ON_CONFLICT, max_parallel=10, fail_fast=True, timeout_per_subplan_seconds=600, retry_failed=True, max_retries=3, ) def time_config_model_dump(self) -> None: """Time serializing a SubplanConfig to dict.""" config = SubplanConfig( execution_mode=ExecutionMode.PARALLEL, max_parallel=8, ) config.model_dump() def time_config_model_dump_json(self) -> None: """Time serializing a SubplanConfig to JSON string.""" config = SubplanConfig( execution_mode=ExecutionMode.DEPENDENCY_ORDERED, merge_strategy=SubplanMergeStrategy.LAST_WINS, ) config.model_dump_json() class SubplanStatusValidationSuite: """Benchmark SubplanStatus construction and serialization.""" def time_minimal_status_creation(self) -> None: """Time creating a SubplanStatus with minimal fields.""" SubplanStatus( subplan_id=_STATUS_ULID, action_name="local/bench-sub", ) def time_full_status_creation(self) -> None: """Time creating a SubplanStatus with all fields populated.""" SubplanStatus( subplan_id=_STATUS_ULID, action_name="local/bench-sub", target_resources=["src/a.py", "src/b.py"], status=ProcessingState.ERRORED, error="TimeoutError: timed out", changeset_summary="Modified 5 files", files_changed=5, attempt_number=2, ) def time_status_model_dump(self) -> None: """Time serializing a SubplanStatus to dict.""" status = SubplanStatus( subplan_id=_STATUS_ULID, action_name="local/bench-sub", status=ProcessingState.PROCESSING, ) status.model_dump() class SubplanFailureHandlerSuite: """Benchmark SubplanFailureHandler decisions.""" def setup(self) -> None: """Prepare handler and test fixtures.""" self.handler = SubplanFailureHandler() self.config_fail_fast = SubplanConfig( execution_mode=ExecutionMode.PARALLEL, fail_fast=True, retry_failed=True, max_retries=2, ) self.config_no_fail_fast = SubplanConfig( execution_mode=ExecutionMode.PARALLEL, fail_fast=False, retry_failed=True, max_retries=2, ) self.retriable_status = SubplanStatus( subplan_id=_STATUS_ULID, action_name="local/bench-sub", status=ProcessingState.ERRORED, error="TimeoutError: timed out", attempt_number=1, ) self.non_retriable_status = SubplanStatus( subplan_id=_STATUS_ULID, action_name="local/bench-sub", status=ProcessingState.ERRORED, error="ConfigurationError: bad config", attempt_number=1, ) def time_should_stop_others_fail_fast(self) -> None: """Time should_stop_others with fail_fast=True.""" self.handler.should_stop_others(self.config_fail_fast, self.retriable_status) def time_should_stop_others_no_fail_fast(self) -> None: """Time should_stop_others with fail_fast=False.""" self.handler.should_stop_others(self.config_no_fail_fast, self.retriable_status) def time_should_retry_retriable(self) -> None: """Time should_retry for a retriable error.""" self.handler.should_retry(self.config_fail_fast, self.retriable_status) def time_should_retry_non_retriable(self) -> None: """Time should_retry for a non-retriable error.""" self.handler.should_retry(self.config_fail_fast, self.non_retriable_status) class SubplanHierarchySuite: """Benchmark Plan hierarchy property access.""" def setup(self) -> None: """Prepare plans for hierarchy property benchmarks.""" self.root = _root_plan() self.child = _child_plan() self.parent_with_subs = _plan_with_subplans() def time_is_root_plan(self) -> None: """Time is_root_plan property check on root.""" _ = self.root.is_root_plan def time_is_subplan(self) -> None: """Time is_subplan property check on child.""" _ = self.child.is_subplan def time_depth_root(self) -> None: """Time depth property on root plan.""" _ = self.root.depth def time_depth_child(self) -> None: """Time depth property on child plan.""" _ = self.child.depth def time_has_subplans(self) -> None: """Time has_subplans property check.""" _ = self.parent_with_subs.has_subplans def time_cli_dict_with_subplans(self) -> None: """Time as_cli_dict for plan with subplan statuses.""" self.parent_with_subs.as_cli_dict() def time_child_plan_creation(self) -> None: """Time creating a child plan with hierarchy fields.""" _child_plan() def time_plan_with_subplans_creation(self) -> None: """Time creating a plan with subplan config and statuses.""" _plan_with_subplans()