feat(cfg,pipeline,cli,utils): introduce nested RetryConfig and migrate retry fields

- add RetryConfig and embed it as RunConfig.retry while keeping top-level retry_* fields for backward compatibility
- normalize and convert retry_exceptions (strings → exception classes) in RunConfig.__post_init__
- update builders and RunConfigBuilder to initialize/use nested RetryConfig when setting retry-related values
- adapt merge_run_config_with_kwargs to map deprecated flat retry fields into the nested retry config
- update Pipeline to prefer the nested RetryConfig for setup/validation and handle exceptions as list/tuple
- update CLI to construct and pass a RetryConfig when invoking runs

Centralizes retry logic, keeps backward compatibility, and simplifies retry handling across components.

Author: legout

src/flowerpower/cfg/pipeline/builder.py

@@ -4,7 +4,7 @@
 
 from fsspec_utils import AbstractFileSystem, BaseStorageOptions, filesystem
 
-from .run import RunConfig
+from .run import RunConfig, RetryConfig
 from ..project.adapter import AdapterConfig as ProjectAdapterConfig
 from .builder_executor import ExecutorBuilder
 from .builder_adapter import AdapterBuilder
@@ -194,12 +194,14 @@ def with_retries(
         Returns:
             Self for method chaining
         """
-        self._config.max_retries = max_attempts
-        self._config.retry_delay = delay
-        self._config.jitter_factor = jitter
-
+        # Ensure nested retry exists
+        if self._config.retry is None:
+            self._config.retry = RetryConfig()
+        self._config.retry.max_retries = max_attempts
+        self._config.retry.retry_delay = delay
+        self._config.retry.jitter_factor = jitter
         if exceptions:
-            self._config.retry_exceptions = exceptions
+            self._config.retry.retry_exceptions = exceptions
 
         return self
 
@@ -358,13 +360,14 @@ def _validate_config(self, config: RunConfig):
             ValueError: If configuration is invalid
         """
         # Validate retry configuration
-        if config.max_retries < 0:
+        retry_cfg = config.retry or RetryConfig()
+        if retry_cfg.max_retries < 0:
             raise ValueError("max_retries must be non-negative")
 
-        if config.retry_delay < 0:
+        if retry_cfg.retry_delay < 0:
             raise ValueError("retry_delay must be non-negative")
 
-        if config.jitter_factor is not None and config.jitter_factor < 0:
+        if retry_cfg.jitter_factor is not None and retry_cfg.jitter_factor < 0:
             raise ValueError("jitter_factor must be non-negative")
 
         # Validate executor configuration

src/flowerpower/cfg/pipeline/run.py

@@ -31,64 +31,18 @@ class CallbackSpec(msgspec.Struct):
     kwargs: dict | None = None
 
 
-class RunConfig(BaseConfig):
-    inputs: dict | None = msgspec.field(default_factory=dict)
-    final_vars: list[str] | None = msgspec.field(default_factory=list)
-    config: dict | None = msgspec.field(default_factory=dict)
-    cache: dict | bool | None = msgspec.field(default=False)
-    with_adapter: WithAdapterConfig = msgspec.field(default_factory=WithAdapterConfig)
-    executor: ExecutorConfig = msgspec.field(default_factory=ExecutorConfig)
-    log_level: str | None = msgspec.field(default="INFO")
+class RetryConfig(BaseConfig):
+    """Retry configuration for pipeline execution."""
     max_retries: int = msgspec.field(default=3)
-    retry_delay: int | float = msgspec.field(default=1)
+    retry_delay: float = msgspec.field(default=1.0)
     jitter_factor: float | None = msgspec.field(default=0.1)
-    retry_exceptions: list[str] = msgspec.field(default_factory=lambda: ["Exception"])
-    # New fields for comprehensive configuration
-    pipeline_adapter_cfg: dict | None = msgspec.field(default=None)
-    project_adapter_cfg: dict | None = msgspec.field(default=None)
-    adapter: dict[str, Any] | None = msgspec.field(default=None)
-    reload: bool = msgspec.field(default=False)
-    on_success: CallbackSpec | None = msgspec.field(default=None)
-    on_failure: CallbackSpec | None = msgspec.field(default=None)
+    # Accept strings or classes; will be converted to exception classes in __post_init__
+    retry_exceptions: list[Any] = msgspec.field(default_factory=lambda: ["Exception"])  # type: ignore[assignment]
 
     def __post_init__(self):
-        # if isinstance(self.inputs, dict):
-        #     self.inputs = munchify(self.inputs)
-        if isinstance(self.config, dict):
-            self.config = munchify(self.config)
-        if isinstance(self.cache, (dict)):
-            self.cache = munchify(self.cache)
-        if isinstance(self.with_adapter, dict):
-            self.with_adapter = WithAdapterConfig.from_dict(self.with_adapter)
-        if isinstance(self.executor, dict):
-            self.executor = ExecutorConfig.from_dict(self.executor)
-        if isinstance(self.pipeline_adapter_cfg, dict):
-            from ..pipeline.adapter import AdapterConfig as PipelineAdapterConfig
-            self.pipeline_adapter_cfg = PipelineAdapterConfig.from_dict(self.pipeline_adapter_cfg)
-        if isinstance(self.project_adapter_cfg, dict):
-            from ..project.adapter import AdapterConfig as ProjectAdapterConfig
-            self.project_adapter_cfg = ProjectAdapterConfig.from_dict(self.project_adapter_cfg)
-        if isinstance(self.adapter, dict):
-            # Convert adapter instances if needed
-            pass
         if isinstance(self.retry_exceptions, list):
-            # Convert string exceptions to actual exception classes using dynamic import
             self.retry_exceptions = self._convert_exception_strings(self.retry_exceptions)
 
-        # Handle callback conversions
-        if self.on_success is not None and not isinstance(self.on_success, CallbackSpec):
-            if callable(self.on_success):
-                self.on_success = CallbackSpec(func=self.on_success)
-            elif isinstance(self.on_success, tuple) and len(self.on_success) == 3:
-                func, args, kwargs = self.on_success
-                self.on_success = CallbackSpec(func=func, args=args, kwargs=kwargs)
-            else:
-                self.on_success = None
-                warnings.warn(
-                    "Invalid on_success format, must be Callable or (Callable, args, kwargs)",
-                    RuntimeWarning
-                )
-    
     def _convert_exception_strings(self, exception_list: list) -> list:
         """Convert exception strings to actual exception classes using dynamic import.
         
@@ -99,44 +53,29 @@ def _convert_exception_strings(self, exception_list: list) -> list:
             List of exception classes.
         """
         converted_exceptions = []
-        
         for exc in exception_list:
             if isinstance(exc, str):
                 try:
-                    # Try to dynamically import the exception class
                     exc_class = self._import_exception_class(exc)
                     converted_exceptions.append(exc_class)
                 except (ImportError, AttributeError) as e:
                     warnings.warn(
                         f"Could not import exception class '{exc}': {e}. Using Exception instead.",
-                        RuntimeWarning
+                        RuntimeWarning,
                     )
                     converted_exceptions.append(Exception)
             elif isinstance(exc, type) and issubclass(exc, BaseException):
                 converted_exceptions.append(exc)
             else:
                 warnings.warn(
                     f"Invalid exception type: {type(exc)}. Using Exception instead.",
-                    RuntimeWarning
+                    RuntimeWarning,
                 )
                 converted_exceptions.append(Exception)
-        
         return converted_exceptions
-    
+
     def _import_exception_class(self, exception_name: str) -> type:
-        """Dynamically import an exception class by name.
-        
-        Args:
-            exception_name: Name of the exception class to import.
-            
-        Returns:
-            The imported exception class.
-            
-        Raises:
-            ImportError: If the module cannot be imported.
-            AttributeError: If the exception class is not found in the module.
-        """
-        # Handle built-in exceptions first
+        """Dynamically import an exception class by name."""
         built_in_exceptions = {
             'Exception': Exception,
             'ValueError': ValueError,
@@ -149,17 +88,14 @@ def _import_exception_class(self, exception_name: str) -> type:
             'ImportError': ImportError,
             'TimeoutError': TimeoutError,
         }
-        
         if exception_name in built_in_exceptions:
             return built_in_exceptions[exception_name]
-        
-        # Handle module-qualified exceptions (e.g., 'requests.exceptions.HTTPError')
+
         if '.' in exception_name:
             module_name, class_name = exception_name.rsplit('.', 1)
             module = importlib.import_module(module_name)
             return getattr(module, class_name)
-        
-        # Try to import from common modules
+
         common_modules = [
             'requests.exceptions',
             'urllib.error',
@@ -169,18 +105,92 @@ def _import_exception_class(self, exception_name: str) -> type:
             'os',
             'io',
         ]
-        
         for module_name in common_modules:
             try:
                 module = importlib.import_module(module_name)
                 if hasattr(module, exception_name):
                     return getattr(module, exception_name)
             except ImportError:
                 continue
-        
-        # If not found in common modules, raise an error
         raise ImportError(f"Could not find exception class: {exception_name}")
 
+
+class RunConfig(BaseConfig):
+    inputs: dict | None = msgspec.field(default_factory=dict)
+    final_vars: list[str] | None = msgspec.field(default_factory=list)
+    config: dict | None = msgspec.field(default_factory=dict)
+    cache: dict | bool | None = msgspec.field(default=False)
+    with_adapter: WithAdapterConfig = msgspec.field(default_factory=WithAdapterConfig)
+    executor: ExecutorConfig = msgspec.field(default_factory=ExecutorConfig)
+    log_level: str | None = msgspec.field(default="INFO")
+    # New nested retry configuration
+    retry: RetryConfig | None = msgspec.field(default=None)
+    # Deprecated top-level retry fields (kept for backward compatibility)
+    max_retries: int = msgspec.field(default=3)
+    retry_delay: int | float = msgspec.field(default=1)
+    jitter_factor: float | None = msgspec.field(default=0.1)
+    retry_exceptions: list[str] = msgspec.field(default_factory=lambda: ["Exception"])  # type: ignore[assignment]
+    # New fields for comprehensive configuration
+    pipeline_adapter_cfg: dict | None = msgspec.field(default=None)
+    project_adapter_cfg: dict | None = msgspec.field(default=None)
+    adapter: dict[str, Any] | None = msgspec.field(default=None)
+    reload: bool = msgspec.field(default=False)
+    on_success: CallbackSpec | None = msgspec.field(default=None)
+    on_failure: CallbackSpec | None = msgspec.field(default=None)
+
+    def __post_init__(self):
+        # if isinstance(self.inputs, dict):
+        #     self.inputs = munchify(self.inputs)
+        if isinstance(self.config, dict):
+            self.config = munchify(self.config)
+        if isinstance(self.cache, (dict)):
+            self.cache = munchify(self.cache)
+        if isinstance(self.with_adapter, dict):
+            self.with_adapter = WithAdapterConfig.from_dict(self.with_adapter)
+        if isinstance(self.executor, dict):
+            self.executor = ExecutorConfig.from_dict(self.executor)
+        if isinstance(self.pipeline_adapter_cfg, dict):
+            from ..pipeline.adapter import AdapterConfig as PipelineAdapterConfig
+            self.pipeline_adapter_cfg = PipelineAdapterConfig.from_dict(self.pipeline_adapter_cfg)
+        if isinstance(self.project_adapter_cfg, dict):
+            from ..project.adapter import AdapterConfig as ProjectAdapterConfig
+            self.project_adapter_cfg = ProjectAdapterConfig.from_dict(self.project_adapter_cfg)
+        if isinstance(self.adapter, dict):
+            # Convert adapter instances if needed
+            pass
+
+        # Normalize retry configuration (prefer nested RetryConfig)
+        if isinstance(self.retry, dict):
+            self.retry = RetryConfig.from_dict(self.retry)
+        if self.retry is None:
+            # Build nested retry from (deprecated) top-level fields
+            self.retry = RetryConfig(
+                max_retries=self.max_retries,
+                retry_delay=float(self.retry_delay),
+                jitter_factor=self.jitter_factor,
+                retry_exceptions=self.retry_exceptions,
+            )
+        # Keep top-level fields in sync for backward compatibility
+        self.max_retries = self.retry.max_retries
+        self.retry_delay = self.retry.retry_delay
+        self.jitter_factor = self.retry.jitter_factor
+        # Ensure top-level exceptions reflect converted classes
+        self.retry_exceptions = list(self.retry.retry_exceptions)
+
+        # Handle callback conversions
+        if self.on_success is not None and not isinstance(self.on_success, CallbackSpec):
+            if callable(self.on_success):
+                self.on_success = CallbackSpec(func=self.on_success)
+            elif isinstance(self.on_success, tuple) and len(self.on_success) == 3:
+                func, args, kwargs = self.on_success
+                self.on_success = CallbackSpec(func=func, args=args, kwargs=kwargs)
+            else:
+                self.on_success = None
+                warnings.warn(
+                    "Invalid on_success format, must be Callable or (Callable, args, kwargs)",
+                    RuntimeWarning
+                )
+        # Handle on_failure callback conversions (mirror on_success behavior)
         if self.on_failure is not None and not isinstance(self.on_failure, CallbackSpec):
             if callable(self.on_failure):
                 self.on_failure = CallbackSpec(func=self.on_failure)
@@ -191,5 +201,5 @@ def _import_exception_class(self, exception_name: str) -> type:
                 self.on_failure = None
                 warnings.warn(
                     "Invalid on_failure format, must be Callable or (Callable, args, kwargs)",
-                    RuntimeWarning
+                    RuntimeWarning,
                 )

src/flowerpower/cli/pipeline.py

@@ -6,7 +6,7 @@
 
 from ..flowerpower import FlowerPowerProject
 from ..pipeline.manager import HookType, PipelineManager
-from ..cfg.pipeline.run import RunConfig
+from ..cfg.pipeline.run import RunConfig, RetryConfig
 from ..utils.logging import setup_logging
 from .utils import parse_dict_or_list_param
 
@@ -168,9 +168,11 @@ def run(
             config=parsed_config,
             cache=parsed_cache,
             with_adapter=with_adapter_config,  # type: ignore
-            max_retries=max_retries,
-            retry_delay=retry_delay,
-            jitter_factor=jitter_factor,
+            retry=RetryConfig(
+                max_retries=max_retries,
+                retry_delay=retry_delay,
+                jitter_factor=jitter_factor,
+            ),
         )
 
         # Handle executor configuration

src/flowerpower/pipeline/pipeline.py

@@ -124,12 +124,13 @@ def run(self, run_config: RunConfig | None = None, **kwargs) -> dict[str, Any]:
         if run_config.reload:
             self._reload_module()
 
-        # Set up retry configuration
+        # Set up retry configuration (use nested RetryConfig)
+        retry_cfg = run_config.retry or self.config.run.retry
         retry_config = self._setup_retry_config(
-            run_config.max_retries,
-            run_config.retry_delay,
-            run_config.jitter_factor,
-            run_config.retry_exceptions,
+            retry_cfg.max_retries,
+            retry_cfg.retry_delay,
+            retry_cfg.jitter_factor,
+            tuple(retry_cfg.retry_exceptions) if isinstance(retry_cfg.retry_exceptions, (list, tuple)) else retry_cfg.retry_exceptions,
         )
         max_retries = retry_config["max_retries"]
         retry_delay = retry_config["retry_delay"]
@@ -151,12 +152,13 @@ def _setup_retry_config(
         max_retries: int | None,
         retry_delay: float | None,
         jitter_factor: float | None,
-        retry_exceptions: tuple | None,
+        retry_exceptions: tuple | list | None,
     ) -> dict:
         """Set up retry configuration with defaults and validation."""
-        max_retries = max_retries or self.config.run.max_retries or 0
-        retry_delay = retry_delay or self.config.run.retry_delay or 1.0
-        jitter_factor = jitter_factor or self.config.run.jitter_factor or 0.1
+        cfg = self.config.run.retry
+        max_retries = max_retries or (cfg.max_retries if cfg else 0) or 0
+        retry_delay = retry_delay or (cfg.retry_delay if cfg else 1.0) or 1.0
+        jitter_factor = jitter_factor or (cfg.jitter_factor if cfg else 0.1) or 0.1
 
         # Convert string exceptions to actual exception classes
         if retry_exceptions and isinstance(retry_exceptions, (list, tuple)):