test(core): fix ASV benchmark timing methodology and remove unused imports

- Remove unused `from typing import Any` import from core_retry_patterns_bench.py
- Refactor CircuitBreakerStateTransitionBench into three separate benchmark classes (CircuitBreakerClosedToOpenTransitionBench, CircuitBreakerOpenToHalfOpenTransitionBench, CircuitBreakerHalfOpenToClosedTransitionBench), each with a proper setup() method that pre-creates the CircuitBreaker in the correct initial state so only the transition itself is measured
- Move open-state CircuitBreaker creation for async fast-fail benchmark into setup() in CircuitBreakerAsyncBench, eliminating instance creation overhead from the timed method

Addresses reviewer feedback on PR #10961.

benchmarks/core_circuit_breaker_bench.py

@@ -0,0 +1,264 @@
+"""ASV benchmarks for circuit breaker pattern.
+
+Measures the overhead of circuit breaker state management,
+state transitions, and fast-fail latency in open state.
+"""
+
+from __future__ import annotations
+
+import asyncio
+import time
+
+from cleveragents.core.circuit_breaker import (
+    CircuitBreaker,
+    CircuitBreakerOpen,
+    CircuitBreakerState,
+)
+
+
+class CircuitBreakerClosedStateBench:
+    """Benchmark circuit breaker overhead in closed state."""
+
+    timeout = 60
+
+    def setup(self) -> None:
+        """Set up circuit breaker in closed state."""
+        self.breaker = CircuitBreaker(
+            failure_threshold=5,
+            recovery_timeout=60.0,
+            expected_exception=Exception,
+            half_open_max_successes=2,
+            cooldown_seconds=30.0,
+            name="test_service",
+        )
+
+    def time_call_success(self) -> None:
+        """Benchmark successful call overhead in closed state."""
+
+        def dummy_func() -> str:
+            return "success"
+
+        self.breaker.call(dummy_func)
+
+    def time_call_with_args(self) -> None:
+        """Benchmark call with arguments in closed state."""
+
+        def dummy_func(x: int, y: str) -> str:
+            return f"{x}:{y}"
+
+        self.breaker.call(dummy_func, 42, "test")
+
+    def time_state_check(self) -> None:
+        """Benchmark state property access."""
+        _ = self.breaker.state
+
+    def time_failure_count_check(self) -> None:
+        """Benchmark failure count property access."""
+        _ = self.breaker.failure_count
+
+
+class CircuitBreakerOpenStateBench:
+    """Benchmark circuit breaker fast-fail latency in open state."""
+
+    timeout = 60
+
+    def setup(self) -> None:
+        """Set up circuit breaker in open state."""
+        self.breaker = CircuitBreaker(
+            failure_threshold=2,
+            recovery_timeout=60.0,
+            expected_exception=Exception,
+            half_open_max_successes=2,
+            cooldown_seconds=30.0,
+            name="test_service",
+        )
+        # Force the breaker into open state
+        for _ in range(3):
+            try:
+                self.breaker.call(lambda: 1 / 0)  # Raise ZeroDivisionError
+            except ZeroDivisionError:
+                pass
+
+    def time_fast_fail(self) -> None:
+        """Benchmark fast-fail latency when circuit is open."""
+        try:
+            self.breaker.call(lambda: "should not execute")
+        except CircuitBreakerOpen:
+            pass
+
+    def time_open_state_check(self) -> None:
+        """Benchmark state check when open."""
+        assert self.breaker.state == CircuitBreakerState.OPEN
+
+
+class CircuitBreakerClosedToOpenTransitionBench:
+    """Benchmark closed-to-open state transition overhead."""
+
+    timeout = 60
+
+    def setup(self) -> None:
+        """Set up a fresh circuit breaker in closed state."""
+        self.breaker = CircuitBreaker(
+            failure_threshold=2,
+            recovery_timeout=60.0,
+            expected_exception=Exception,
+            half_open_max_successes=1,
+            cooldown_seconds=30.0,
+            name="test_service",
+        )
+
+    def time_closed_to_open_transition(self) -> None:
+        """Benchmark transition from closed to open state."""
+        # Trigger failures to transition to open
+        for _ in range(3):
+            try:
+                self.breaker.call(lambda: 1 / 0)
+            except ZeroDivisionError:
+                pass
+
+
+class CircuitBreakerOpenToHalfOpenTransitionBench:
+    """Benchmark open-to-half-open state transition overhead."""
+
+    timeout = 60
+
+    def setup(self) -> None:
+        """Set up circuit breaker already in open state."""
+        self.breaker = CircuitBreaker(
+            failure_threshold=2,
+            recovery_timeout=0.05,  # Very short timeout
+            expected_exception=Exception,
+            half_open_max_successes=1,
+            cooldown_seconds=0.01,
+            name="test_service",
+        )
+        # Force to open state
+        for _ in range(3):
+            try:
+                self.breaker.call(lambda: 1 / 0)
+            except ZeroDivisionError:
+                pass
+        # Wait for recovery timeout so transition to half-open is possible
+        time.sleep(0.1)
+
+    def time_open_to_half_open_transition(self) -> None:
+        """Benchmark transition from open to half-open state."""
+        # Attempt call to trigger half-open transition
+        try:
+            self.breaker.call(lambda: "success")
+        except CircuitBreakerOpen:
+            pass
+
+
+class CircuitBreakerHalfOpenToClosedTransitionBench:
+    """Benchmark half-open-to-closed state transition overhead."""
+
+    timeout = 60
+
+    def setup(self) -> None:
+        """Set up circuit breaker already in half-open state."""
+        self.breaker = CircuitBreaker(
+            failure_threshold=2,
+            recovery_timeout=0.05,
+            expected_exception=Exception,
+            half_open_max_successes=1,
+            cooldown_seconds=0.01,
+            name="test_service",
+        )
+        # Force to open state
+        for _ in range(3):
+            try:
+                self.breaker.call(lambda: 1 / 0)
+            except ZeroDivisionError:
+                pass
+        # Wait for recovery timeout so breaker enters half-open on next call
+        time.sleep(0.1)
+
+    def time_half_open_to_closed_transition(self) -> None:
+        """Benchmark transition from half-open to closed state."""
+        # Successful call in half-open should close the breaker
+        self.breaker.call(lambda: "success")
+
+
+class CircuitBreakerAsyncBench:
+    """Benchmark async circuit breaker operations."""
+
+    timeout = 60
+
+    def setup(self) -> None:
+        """Set up async circuit breakers."""
+        self.breaker = CircuitBreaker(
+            failure_threshold=5,
+            recovery_timeout=60.0,
+            expected_exception=Exception,
+            half_open_max_successes=2,
+            cooldown_seconds=30.0,
+            name="test_service",
+        )
+        # Set up a separate breaker already in open state for fast-fail benchmark
+        self.open_breaker = CircuitBreaker(
+            failure_threshold=2,
+            recovery_timeout=60.0,
+            expected_exception=Exception,
+            half_open_max_successes=2,
+            cooldown_seconds=30.0,
+            name="test_service_open",
+        )
+        for _ in range(3):
+            try:
+                self.open_breaker.call(lambda: 1 / 0)
+            except ZeroDivisionError:
+                pass
+
+    def time_async_call_success(self) -> None:
+        """Benchmark successful async call overhead."""
+
+        async def dummy_async() -> str:
+            return "success"
+
+        asyncio.run(self.breaker.async_call(dummy_async))
+
+    def time_async_call_with_args(self) -> None:
+        """Benchmark async call with arguments."""
+
+        async def dummy_async(x: int, y: str) -> str:
+            return f"{x}:{y}"
+
+        asyncio.run(self.breaker.async_call(dummy_async, 42, "test"))
+
+    def time_async_fast_fail(self) -> None:
+        """Benchmark async fast-fail when open."""
+
+        async def dummy_async() -> str:
+            return "should not execute"
+
+        try:
+            asyncio.run(self.open_breaker.async_call(dummy_async))
+        except CircuitBreakerOpen:
+            pass
+
+
+class CircuitBreakerInitializationBench:
+    """Benchmark circuit breaker initialization overhead."""
+
+    timeout = 60
+
+    def time_default_initialization(self) -> None:
+        """Benchmark default initialization."""
+        CircuitBreaker()
+
+    def time_custom_initialization(self) -> None:
+        """Benchmark initialization with custom parameters."""
+        CircuitBreaker(
+            failure_threshold=10,
+            recovery_timeout=120.0,
+            expected_exception=ValueError,
+            half_open_max_successes=3,
+            cooldown_seconds=45.0,
+            name="custom_service",
+        )
+
+    def time_multiple_breakers(self) -> None:
+        """Benchmark creating multiple breakers."""
+        for i in range(10):
+            CircuitBreaker(name=f"service_{i}")

benchmarks/core_retry_patterns_bench.py

@@ -0,0 +1,280 @@
+"""ASV benchmarks for retry patterns.
+
+Measures the overhead of retry decorator construction and
+happy-path invocation for the public retry factory functions.
+"""
+
+from __future__ import annotations
+
+import asyncio
+
+from cleveragents.core.retry_patterns import (
+    get_retry_decorator,
+    retry_on_result,
+    retry_with_exponential_backoff,
+    retry_with_jitter,
+    retry_with_timeout,
+    should_retry_result,
+)
+
+
+class RetryDecoratorConstructionBench:
+    """Benchmark retry decorator construction overhead."""
+
+    timeout = 60
+
+    def time_retry_with_exponential_backoff_construction(self) -> None:
+        """Benchmark constructing exponential backoff decorator."""
+
+        @retry_with_exponential_backoff(max_attempts=3)
+        def dummy_func() -> str:
+            return "success"
+
+    def time_retry_with_jitter_construction(self) -> None:
+        """Benchmark constructing jitter decorator."""
+
+        @retry_with_jitter(max_attempts=3)
+        def dummy_func() -> str:
+            return "success"
+
+    def time_retry_with_timeout_construction(self) -> None:
+        """Benchmark constructing timeout decorator."""
+
+        @retry_with_timeout(timeout_seconds=5.0)
+        def dummy_func() -> str:
+            return "success"
+
+    def time_retry_on_result_construction(self) -> None:
+        """Benchmark constructing result-based retry decorator."""
+
+        @retry_on_result(should_retry_result, max_attempts=3)
+        def dummy_func() -> str:
+            return "success"
+
+    def time_get_retry_decorator_network(self) -> None:
+        """Benchmark constructing decorator via factory for network category."""
+        get_retry_decorator("network")
+
+    def time_get_retry_decorator_provider(self) -> None:
+        """Benchmark constructing decorator via factory for provider category."""
+        get_retry_decorator("provider")
+
+    def time_get_retry_decorator_database(self) -> None:
+        """Benchmark constructing decorator via factory for database category."""
+        get_retry_decorator("database")
+
+    def time_get_retry_decorator_unknown(self) -> None:
+        """Benchmark constructing decorator via factory for unknown category."""
+        get_retry_decorator("unknown_category")
+
+
+class RetryDecoratorInvocationBench:
+    """Benchmark happy-path retry decorator invocation."""
+
+    timeout = 60
+
+    def setup(self) -> None:
+        """Set up decorated functions."""
+
+        @retry_with_exponential_backoff(max_attempts=3)
+        def exponential_func() -> str:
+            return "success"
+
+        @retry_with_jitter(max_attempts=3)
+        def jitter_func() -> str:
+            return "success"
+
+        @retry_with_timeout(timeout_seconds=5.0)
+        def timeout_func() -> str:
+            return "success"
+
+        @retry_on_result(should_retry_result, max_attempts=3)
+        def result_func() -> str:
+            return "success"
+
+        self.exponential_func = exponential_func
+        self.jitter_func = jitter_func
+        self.timeout_func = timeout_func
+        self.result_func = result_func
+
+    def time_exponential_backoff_invocation(self) -> None:
+        """Benchmark exponential backoff invocation (happy path)."""
+        self.exponential_func()
+
+    def time_jitter_invocation(self) -> None:
+        """Benchmark jitter invocation (happy path)."""
+        self.jitter_func()
+
+    def time_timeout_invocation(self) -> None:
+        """Benchmark timeout invocation (happy path)."""
+        self.timeout_func()
+
+    def time_result_based_invocation(self) -> None:
+        """Benchmark result-based retry invocation (happy path)."""
+        self.result_func()
+
+
+class RetryDecoratorAsyncBench:
+    """Benchmark async retry decorator operations."""
+
+    timeout = 60
+
+    def setup(self) -> None:
+        """Set up async decorated functions."""
+
+        @retry_with_exponential_backoff(max_attempts=3)
+        async def async_exponential() -> str:
+            return "success"
+
+        @retry_with_jitter(max_attempts=3)
+        async def async_jitter() -> str:
+            return "success"
+
+        @retry_with_timeout(timeout_seconds=5.0)
+        async def async_timeout() -> str:
+            return "success"
+
+        self.async_exponential = async_exponential
+        self.async_jitter = async_jitter
+        self.async_timeout = async_timeout
+
+    def time_async_exponential_invocation(self) -> None:
+        """Benchmark async exponential backoff invocation."""
+        asyncio.run(self.async_exponential())
+
+    def time_async_jitter_invocation(self) -> None:
+        """Benchmark async jitter invocation."""
+        asyncio.run(self.async_jitter())
+
+    def time_async_timeout_invocation(self) -> None:
+        """Benchmark async timeout invocation."""
+        asyncio.run(self.async_timeout())
+
+
+class RetryDecoratorWithArgumentsBench:
+    """Benchmark retry decorators with function arguments."""
+
+    timeout = 60
+
+    def setup(self) -> None:
+        """Set up decorated functions with arguments."""
+
+        @retry_with_exponential_backoff(max_attempts=3)
+        def func_with_args(x: int, y: str, z: float = 1.0) -> str:
+            return f"{x}:{y}:{z}"
+
+        @retry_with_jitter(max_attempts=3)
+        def func_with_kwargs(a: int, b: str = "default") -> str:
+            return f"{a}:{b}"
+
+        self.func_with_args = func_with_args
+        self.func_with_kwargs = func_with_kwargs
+
+    def time_exponential_with_positional_args(self) -> None:
+        """Benchmark exponential backoff with positional arguments."""
+        self.func_with_args(42, "test", 2.5)
+
+    def time_exponential_with_keyword_args(self) -> None:
+        """Benchmark exponential backoff with keyword arguments."""
+        self.func_with_args(x=42, y="test", z=2.5)
+
+    def time_jitter_with_mixed_args(self) -> None:
+        """Benchmark jitter with mixed positional and keyword arguments."""
+        self.func_with_kwargs(100, b="custom")
+
+
+class RetryDecoratorFactoryBench:
+    """Benchmark retry decorator factory function."""
+
+    timeout = 60
+
+    def time_factory_network_category(self) -> None:
+        """Benchmark factory with network category."""
+        decorator = get_retry_decorator("network")
+
+        @decorator
+        def dummy() -> str:
+            return "success"
+
+        dummy()
+
+    def time_factory_provider_category(self) -> None:
+        """Benchmark factory with provider category."""
+        decorator = get_retry_decorator("provider")
+
+        @decorator
+        def dummy() -> str:
+            return "success"
+
+        dummy()
+
+    def time_factory_database_category(self) -> None:
+        """Benchmark factory with database category."""
+        decorator = get_retry_decorator("database")
+
+        @decorator
+        def dummy() -> str:
+            return "success"
+
+        dummy()
+
+    def time_factory_file_operation_category(self) -> None:
+        """Benchmark factory with file_operation category."""
+        decorator = get_retry_decorator("file_operation")
+
+        @decorator
+        def dummy() -> str:
+            return "success"
+
+        dummy()
+
+
+class RetryDecoratorConfigurationBench:
+    """Benchmark retry decorator with various configurations."""
+
+    timeout = 60
+
+    def time_minimal_config(self) -> None:
+        """Benchmark decorator with minimal configuration."""
+
+        @retry_with_exponential_backoff()
+        def dummy() -> str:
+            return "success"
+
+        dummy()
+
+    def time_aggressive_config(self) -> None:
+        """Benchmark decorator with aggressive retry settings."""
+
+        @retry_with_exponential_backoff(max_attempts=10)
+        def dummy() -> str:
+            return "success"
+
+        dummy()
+
+    def time_conservative_config(self) -> None:
+        """Benchmark decorator with conservative retry settings."""
+
+        @retry_with_exponential_backoff(max_attempts=1)
+        def dummy() -> str:
+            return "success"
+
+        dummy()
+
+    def time_long_timeout_config(self) -> None:
+        """Benchmark decorator with long timeout."""
+
+        @retry_with_timeout(timeout_seconds=300.0)
+        def dummy() -> str:
+            return "success"
+
+        dummy()
+
+    def time_short_timeout_config(self) -> None:
+        """Benchmark decorator with short timeout."""
+
+        @retry_with_timeout(timeout_seconds=0.1)
+        def dummy() -> str:
+            return "success"
+
+        dummy()

benchmarks/core_retry_service_patterns_bench.py

@@ -0,0 +1,367 @@
+"""ASV benchmarks for service-level retry patterns.
+
+Measures the overhead of retry_service_operation decorator
+and retry_auto_debug happy-path latency.
+"""
+
+from __future__ import annotations
+
+import asyncio
+from typing import Any
+
+from cleveragents.core.circuit_breaker import CircuitBreaker
+from cleveragents.core.retry_service_patterns import (
+    RetryContext,
+    is_read_only_plan_operation,
+    retry_auto_debug,
+    retry_service_operation,
+)
+
+
+class RetryServiceOperationDecoratorBench:
+    """Benchmark retry_service_operation decorator overhead."""
+
+    timeout = 60
+
+    def time_decorator_construction_sync(self) -> None:
+        """Benchmark constructing sync service retry decorator."""
+
+        @retry_service_operation(
+            service_name="test_service",
+            operation_name="test_op",
+            max_attempts=3,
+        )
+        def dummy_func() -> str:
+            return "success"
+
+    def time_decorator_construction_async(self) -> None:
+        """Benchmark constructing async service retry decorator."""
+
+        @retry_service_operation(
+            service_name="test_service",
+            operation_name="test_op",
+            max_attempts=3,
+        )
+        async def dummy_func() -> str:
+            return "success"
+
+    def time_decorator_with_circuit_breaker(self) -> None:
+        """Benchmark decorator with circuit breaker instance."""
+        cb = CircuitBreaker(
+            failure_threshold=5,
+            recovery_timeout=60.0,
+            name="test_service",
+        )
+
+        @retry_service_operation(
+            service_name="test_service",
+            operation_name="test_op",
+            max_attempts=3,
+            circuit_breaker=cb,
+        )
+        def dummy_func() -> str:
+            return "success"
+
+    def time_decorator_with_exponential_strategy(self) -> None:
+        """Benchmark decorator with exponential backoff strategy."""
+
+        @retry_service_operation(
+            service_name="test_service",
+            operation_name="test_op",
+            max_attempts=3,
+            backoff_strategy="exponential",
+            base_delay=1.0,
+            max_delay=60.0,
+        )
+        def dummy_func() -> str:
+            return "success"
+
+
+class RetryServiceOperationInvocationBench:
+    """Benchmark happy-path service retry invocation."""
+
+    timeout = 60
+
+    def setup(self) -> None:
+        """Set up decorated service operations."""
+
+        @retry_service_operation(
+            service_name="test_service",
+            operation_name="test_op",
+            max_attempts=3,
+        )
+        def sync_op() -> str:
+            return "success"
+
+        @retry_service_operation(
+            service_name="test_service",
+            operation_name="test_op",
+            max_attempts=3,
+        )
+        async def async_op() -> str:
+            return "success"
+
+        cb = CircuitBreaker(
+            failure_threshold=5,
+            recovery_timeout=60.0,
+            name="test_service",
+        )
+
+        @retry_service_operation(
+            service_name="test_service",
+            operation_name="test_op",
+            max_attempts=3,
+            circuit_breaker=cb,
+        )
+        def breaker_op() -> str:
+            return "success"
+
+        self.sync_op = sync_op
+        self.async_op = async_op
+        self.breaker_op = breaker_op
+
+    def time_sync_operation_invocation(self) -> None:
+        """Benchmark sync service operation invocation (happy path)."""
+        self.sync_op()
+
+    def time_async_operation_invocation(self) -> None:
+        """Benchmark async service operation invocation (happy path)."""
+        asyncio.run(self.async_op())
+
+    def time_operation_with_circuit_breaker(self) -> None:
+        """Benchmark service operation with circuit breaker (happy path)."""
+        self.breaker_op()
+
+
+class RetryServiceOperationWithArgumentsBench:
+    """Benchmark service retry with function arguments."""
+
+    timeout = 60
+
+    def setup(self) -> None:
+        """Set up decorated operations with arguments."""
+
+        @retry_service_operation(
+            service_name="test_service",
+            operation_name="fetch_data",
+            max_attempts=3,
+        )
+        def fetch_data(
+            resource_id: int, include_metadata: bool = False
+        ) -> dict[str, Any]:
+            return {"id": resource_id, "data": "test"}
+
+        @retry_service_operation(
+            service_name="test_service",
+            operation_name="update_data",
+            max_attempts=3,
+        )
+        async def update_data(
+            resource_id: int, payload: dict[str, Any]
+        ) -> dict[str, Any]:
+            return {"id": resource_id, "updated": True}
+
+        self.fetch_data = fetch_data
+        self.update_data = update_data
+
+    def time_sync_with_positional_args(self) -> None:
+        """Benchmark sync operation with positional arguments."""
+        self.fetch_data(42, True)
+
+    def time_sync_with_keyword_args(self) -> None:
+        """Benchmark sync operation with keyword arguments."""
+        self.fetch_data(resource_id=42, include_metadata=True)
+
+    def time_async_with_complex_payload(self) -> None:
+        """Benchmark async operation with complex payload."""
+        payload = {"key": "value", "nested": {"data": [1, 2, 3]}}
+        asyncio.run(self.update_data(42, payload))
+
+
+class RetryAutoDebugBench:
+    """Benchmark retry_auto_debug decorator."""
+
+    timeout = 60
+
+    def time_auto_debug_construction(self) -> None:
+        """Benchmark constructing auto-debug decorator."""
+
+        @retry_auto_debug(max_debug_attempts=3)
+        async def dummy_func() -> str:
+            return "success"
+
+    def time_auto_debug_invocation(self) -> None:
+        """Benchmark auto-debug invocation (happy path)."""
+
+        @retry_auto_debug(max_debug_attempts=3)
+        async def dummy_func() -> str:
+            return "success"
+
+        asyncio.run(dummy_func())
+
+    def time_auto_debug_single_attempt(self) -> None:
+        """Benchmark auto-debug with single attempt."""
+
+        @retry_auto_debug(max_debug_attempts=1)
+        async def dummy_func() -> str:
+            return "success"
+
+        asyncio.run(dummy_func())
+
+    def time_auto_debug_multiple_attempts(self) -> None:
+        """Benchmark auto-debug with multiple attempts configured."""
+
+        @retry_auto_debug(max_debug_attempts=5)
+        async def dummy_func() -> str:
+            return "success"
+
+        asyncio.run(dummy_func())
+
+
+class RetryContextBench:
+    """Benchmark RetryContext operations."""
+
+    timeout = 60
+
+    def time_context_creation(self) -> None:
+        """Benchmark creating a retry context."""
+        RetryContext(
+            operation_name="test_op",
+            max_attempts=3,
+        )
+
+    def time_context_with_custom_wait(self) -> None:
+        """Benchmark context with custom wait strategy."""
+        from tenacity import wait_exponential
+
+        RetryContext(
+            operation_name="test_op",
+            max_attempts=3,
+            wait_strategy=wait_exponential(multiplier=1, min=1, max=30),
+        )
+
+    def time_context_property_access(self) -> None:
+        """Benchmark accessing context properties."""
+        ctx = RetryContext(
+            operation_name="test_op",
+            max_attempts=3,
+        )
+        _ = ctx.operation_name
+        _ = ctx.max_attempts
+        _ = ctx.attempt_count
+
+    def time_context_execute(self) -> None:
+        """Benchmark executing a function via RetryContext."""
+        ctx = RetryContext(
+            operation_name="test_op",
+            max_attempts=3,
+        )
+        ctx.execute(lambda: "success")
+
+
+class IsReadOnlyPlanOperationBench:
+    """Benchmark is_read_only_plan_operation utility."""
+
+    timeout = 60
+
+    def time_check_strategize(self) -> None:
+        """Benchmark checking strategize operation."""
+        is_read_only_plan_operation({"plan_phase": "strategize"})
+
+    def time_check_plan(self) -> None:
+        """Benchmark checking plan operation."""
+        is_read_only_plan_operation({"plan_phase": "plan"})
+
+    def time_check_validate(self) -> None:
+        """Benchmark checking validate operation."""
+        is_read_only_plan_operation({"plan_phase": "validate"})
+
+    def time_check_review(self) -> None:
+        """Benchmark checking review operation."""
+        is_read_only_plan_operation({"plan_phase": "review"})
+
+    def time_check_preview(self) -> None:
+        """Benchmark checking preview operation."""
+        is_read_only_plan_operation({"plan_phase": "preview"})
+
+    def time_check_check(self) -> None:
+        """Benchmark checking check operation."""
+        is_read_only_plan_operation({"plan_phase": "check"})
+
+    def time_check_execute(self) -> None:
+        """Benchmark checking execute operation (not read-only)."""
+        is_read_only_plan_operation({"plan_phase": "execute"})
+
+    def time_check_read_only_flag(self) -> None:
+        """Benchmark checking read_only flag."""
+        is_read_only_plan_operation({"read_only": True})
+
+    def time_check_unknown_operation(self) -> None:
+        """Benchmark checking unknown operation."""
+        is_read_only_plan_operation({"plan_phase": "unknown_op"})
+
+    def time_check_empty_kwargs(self) -> None:
+        """Benchmark checking empty kwargs."""
+        is_read_only_plan_operation({})
+
+
+class RetryServiceOperationStrategyBench:
+    """Benchmark different retry strategies in service operations."""
+
+    timeout = 60
+
+    def time_exponential_strategy(self) -> None:
+        """Benchmark exponential strategy."""
+
+        @retry_service_operation(
+            service_name="test_service",
+            operation_name="test_op",
+            max_attempts=3,
+            backoff_strategy="exponential",
+        )
+        def dummy() -> str:
+            return "success"
+
+        dummy()
+
+    def time_linear_strategy(self) -> None:
+        """Benchmark linear strategy."""
+
+        @retry_service_operation(
+            service_name="test_service",
+            operation_name="test_op",
+            max_attempts=3,
+            backoff_strategy="linear",
+        )
+        def dummy() -> str:
+            return "success"
+
+        dummy()
+
+    def time_fixed_strategy(self) -> None:
+        """Benchmark fixed strategy."""
+
+        @retry_service_operation(
+            service_name="test_service",
+            operation_name="test_op",
+            max_attempts=3,
+            backoff_strategy="fixed",
+        )
+        def dummy() -> str:
+            return "success"
+
+        dummy()
+
+    def time_jitter_strategy(self) -> None:
+        """Benchmark jitter strategy."""
+
+        @retry_service_operation(
+            service_name="test_service",
+            operation_name="test_op",
+            max_attempts=3,
+            backoff_strategy="jitter",
+        )
+        def dummy() -> str:
+            return "success"
+
+        dummy()