SIENTIAPDE-1646 Sync laborious/ from local release/SIENTIAPDE-1646 (272e02d)

laborious/workflows/drift.py

@@ -0,0 +1,107 @@
+from temporalio import workflow
+
+with workflow.unsafe.imports_passed_through():
+    from datetime import timedelta
+    from typing import Any
+
+    from sientia_do.temporal.constants import DATETIME_FORMAT_WITH_TZ
+    from sientia_do.temporal.policies import retry_policy
+
+    from laborious.activities.activities import Activities
+
+
+@workflow.defn(name='drift')
+class Drift:
+    @workflow.run
+    async def run(self, input_data: dict[str, Any]):
+        """
+        Execute the drift workflow.
+
+        This method orchestrates the complete drift process by:
+        1. Loading data using the provided custom SQL query
+        2. Preparing prediction configuration and filters
+        3. Delegating to the PredictionProcess workflow for ML operations
+        """
+
+        metadata = {
+            'metadata': {
+                'schedule_name': input_data['schedule_name'],
+                'model_name': input_data['model_name'],
+                'model_id': input_data['model_id'],
+                'workflow_name': 'drift',
+            }
+        }
+
+        print(f'Input data: {input_data}', metadata)
+
+        model_config = input_data['model_config']
+        target_name = model_config['target']
+
+        gathering_query = f"""
+            SELECT *
+            FROM "{input_data['schema']}"."{input_data['source_table_name']}"
+            WHERE
+                model_id = '{input_data['model_id']}' AND
+                timestamp > NOW() - INTERVAL '{input_data['interval']} minutes'
+            ORDER BY timestamp ASC
+        """  # nosec B608 - values come from internal Temporal workflow config, not user input
+
+        target_data_handler = workflow.start_activity_method(
+            Activities.load_custom_query,
+            {
+                **metadata,
+                'query': gathering_query,
+                'datetime_columns': ['timestamp', 'created_at'],
+                'orient': 'records',
+            },
+            retry_policy=retry_policy,
+            start_to_close_timeout=timedelta(seconds=300),
+        )
+
+        reference_data_handler = workflow.start_activity_method(
+            Activities.get_reference_data,
+            {**metadata, 'model_name': input_data['model_name']},
+            retry_policy=retry_policy,
+            start_to_close_timeout=timedelta(seconds=300),
+        )
+
+        target_data = await target_data_handler
+        reference_data = await reference_data_handler
+
+        if not target_data:
+            return
+
+        drift_data = await workflow.execute_local_activity_method(
+            Activities.calculate_drift,
+            {
+                **metadata,
+                'target_data': target_data,
+                'reference_data': reference_data,
+                'model_name': input_data['model_name'],
+                'model_id': input_data['model_id'],
+                'target_name': target_name,
+                'drift_metrics': input_data.get(
+                    'drift_metrics', ['kolmogorov_smirnov', 'jensen_shannon', 'wasserstein']
+                ),
+                'chunk_period': input_data.get('chunk_period', 'min'),
+            },
+            retry_policy=retry_policy,
+            start_to_close_timeout=timedelta(seconds=300),
+        )
+
+        if drift_data:
+            await workflow.execute_activity_method(
+                Activities.export_data_to_postgres,
+                {
+                    **metadata,
+                    'data': drift_data,
+                    'schema': input_data['schema'],
+                    'table_name': input_data['target_table_name'],
+                    'timestamp_conversion': {
+                        'column': 'timestamp',
+                        'format': DATETIME_FORMAT_WITH_TZ,
+                    },
+                },
+                retry_policy=retry_policy,
+                start_to_close_timeout=timedelta(seconds=300),
+            )

laborious/workflows/minimal_retrain.py

@@ -1,14 +1,17 @@
 from temporalio import workflow
 
 with workflow.unsafe.imports_passed_through():
-    from laborious.activities.activities import Activities
-    from typing import Any
-    from sientia_do.temporal.policies import retry_policy
     from datetime import timedelta
+    from typing import Any
+
+    from sientia_do.temporal.policies import retry_policy
+
+    from laborious.activities.activities import Activities
+    from laborious.utils.models.minio_dataframe_payload import MinioDataFramePayload
 
 
-@workflow.defn(name="minimal_retrain")
-class MinimalRetrain():
+@workflow.defn(name='minimal_retrain')
+class MinimalRetrain:
     """
     Automated model retraining workflow for the Laborious system.
 
@@ -63,44 +66,62 @@ class MinimalRetrain():
                 'schedule_name': input_data['schedule_name'],
                 'model_name': input_data['model_name'],
                 'model_id': input_data['model_id'],
-                'workflow_name': 'minimal_retrain'
+                'workflow_name': 'minimal_retrain',
             }
         }
 
         model_name = input_data['model_name']
+        model_config = input_data.get('model_config', {})
 
-        data = await workflow.execute_local_activity_method(
-            Activities.load_custom_query,
+        storage_result = await workflow.execute_activity_method(
+            Activities.load_query_with_minio_offload,
             {
                 **metadata,
                 'query': input_data['query'],
-                'datetime_columns': input_data.get('datetime_columns', [])
+                'datetime_columns': input_data.get('datetime_columns', []),
+                'model_name': model_name,
             },
             retry_policy=retry_policy,
-            start_to_close_timeout=timedelta(seconds=60)
+            start_to_close_timeout=timedelta(seconds=600),
         )
 
+        storage_payload = MinioDataFramePayload.from_dict(storage_result)
+        if not storage_payload.has_data():
+            raise ValueError('No data returned from query')
+
         experiment_response = await workflow.execute_activity_method(
             Activities.retrain_model,
             {
                 **metadata,
-                'data': data,
-                'model_name': model_name
+                'data': storage_result,
+                'model_name': model_name,
+                'model_config': model_config,
             },
             retry_policy=retry_policy,
-            start_to_close_timeout=timedelta(seconds=60)
+            start_to_close_timeout=timedelta(hours=1),
         )
 
-        report = await workflow.execute_activity_method(
-            Activities.update_production_model,
+        if experiment_response['success']:
+            update_report = await workflow.execute_activity_method(
+                Activities.update_production_model,
+                {**metadata, 'model_name': model_name, **experiment_response},
+                retry_policy=retry_policy,
+                start_to_close_timeout=timedelta(seconds=60),
+            )
+        else:
+            update_report = {}
+
+        report = await workflow.execute_local_activity_method(
+            Activities.format_retrain_report,
             {
                 **metadata,
+                'experiment_response': experiment_response,
                 'model_name': model_name,
                 'model_id': input_data['model_id'],
-                **experiment_response
+                'update_report': update_report,
             },
             retry_policy=retry_policy,
-            start_to_close_timeout=timedelta(seconds=60)
+            start_to_close_timeout=timedelta(seconds=60),
         )
 
         await workflow.execute_activity_method(
@@ -109,8 +130,8 @@ class MinimalRetrain():
                 **metadata,
                 'data': report,
                 'schema': input_data['schema'],
-                'table_name': input_data['table_name']
+                'table_name': input_data['table_name'],
             },
             retry_policy=retry_policy,
-            start_to_close_timeout=timedelta(seconds=60)
+            start_to_close_timeout=timedelta(seconds=600),
         )

laborious/workflows/predictions_batch.py

@@ -1,14 +1,16 @@
 from temporalio import workflow
 
 with workflow.unsafe.imports_passed_through():
-    from laborious.activities.activities import Activities
-    from typing import Any
-    from sientia_do.temporal.policies import retry_policy
     from datetime import timedelta
+    from typing import Any
+
+    from sientia_do.temporal.policies import retry_policy
+
+    from laborious.activities.activities import Activities
 
 
-@workflow.defn(name="predictions_batch")
-class PredictionsBatch():
+@workflow.defn(name='predictions_batch')
+class PredictionsBatch:
     """
     Main batch prediction workflow for the Laborious system.
 
@@ -59,7 +61,10 @@ class PredictionsBatch():
                     - model_retention (int, optional): Model retention period in minutes
                     - path_priority (list[str]): Decision path priority configuration
                     - opc_output_config (dict, optional): OPC server export configuration
+                    - pi_web_api_output_config (dict, optional): PI Web API export configuration
                     - datetime_columns (list[str], optional): Columns to treat as datetime
+                    - save_transform (bool, optional): Whether to save transformed data (default: True)
+                    - prediction_store_policy (str, optional): Data retention policy (default: 'lts:1')
 
         Returns:
             None: The workflow completes successfully when the child workflow finishes
@@ -74,20 +79,21 @@ class PredictionsBatch():
                 'schedule_name': input_data['schedule_name'],
                 'model_name': input_data['model_name'],
                 'model_id': input_data['model_id'],
-                'workflow_name': 'predictions_batch'
+                'workflow_name': 'predictions_batch',
             }
         }
 
-        # Load data using custom query
-        data = await workflow.execute_local_activity_method(
-            Activities.load_custom_query,
+        # Load data using custom query with optional MinIO offload for large frames
+        data = await workflow.execute_activity_method(
+            Activities.load_query_with_minio_offload,
             {
                 **metadata,
                 'query': input_data['query'],
-                'datetime_columns': input_data.get('datetime_columns', [])
+                'datetime_columns': input_data.get('datetime_columns', []),
+                'model_name': input_data['model_name'],
             },
             retry_policy=retry_policy,
-            start_to_close_timeout=timedelta(seconds=300)
+            start_to_close_timeout=timedelta(seconds=300),
         )
 
         # Prepare input for prediction_process workflow
@@ -96,30 +102,26 @@ class PredictionsBatch():
             'data': data,
             'schema': input_data['schema'],
             'table_name': input_data['table_name'],
+            'transform_table_name': input_data['transform_table_name'],
             'model_id': input_data['model_id'],
             'model_name': input_data['model_name'],
-            'input_filters': input_data.get('input_filters', {
-                'EMPTY_DATA': {
-                    'POLICY': 'STOP'
-                }
-            }),
-            'mlflow_transform_filters': input_data.get('mlflow_transform_filters', {
-                'API_ERROR': {
-                    'POLICY': 'STOP'
-                }
-            }),
-            'mlflow_predict_filters': input_data.get('mlflow_predict_filters', {
-                'API_ERROR': {
-                    'POLICY': 'STOP'
-                }
-            }),
+            'input_filters': input_data.get(
+                'input_filters', {'EMPTY_DATA': {'POLICY': 'STOP', 'CONFIG': {}}}
+            ),
+            'mlflow_transform_filters': input_data.get(
+                'mlflow_transform_filters', {'API_ERROR': {'POLICY': 'STOP', 'CONFIG': {}}}
+            ),
+            'mlflow_predict_filters': input_data.get(
+                'mlflow_predict_filters', {'API_ERROR': {'POLICY': 'STOP', 'CONFIG': {}}}
+            ),
             'model_config': input_data.get('model_config', {}),
             'path_priority': input_data.get('path_priority', ['STOP', 'CONTINUE', 'REPEAT']),
             'opc_output_config': input_data.get('opc_output_config', {}),
-            'prediction_store_policy': input_data.get(
-                'prediction_store_policy', 'lts:1')
+            'on_conflict': input_data.get('on_conflict', 'error'),
+            'pi_web_api_output_config': input_data.get('pi_web_api_output_config', {}),
+            'prediction_store_policy': input_data.get('prediction_store_policy', 'lts:1'),
+            'save_transform': input_data.get('save_transform', True),
         }
 
         # Execute prediction process workflow
-        await workflow.execute_child_workflow(
-            'prediction_process', prediction_input)
+        await workflow.execute_child_workflow('subworkflow.prediction_process', prediction_input)

laborious/workflows/simple_metrics.py

@@ -0,0 +1,95 @@
+from temporalio import workflow
+
+with workflow.unsafe.imports_passed_through():
+    from datetime import timedelta
+    from typing import Any
+
+    from sientia_do.temporal.constants import DATETIME_FORMAT_WITH_TZ
+    from sientia_do.temporal.policies import retry_policy
+
+    from laborious.activities.activities import Activities
+
+
+@workflow.defn(name='simple_metrics')
+class SimpleMetrics:
+    @workflow.run
+    async def run(self, input_data: dict[str, Any]):
+        """
+        Execute the simple metrics workflow.
+        """
+        metadata = {
+            'metadata': {
+                'model_id': input_data['model_id'],
+                'model_name': input_data['model_name'],
+                'workflow_name': 'simple_metrics',
+                'schedule_name': input_data['schedule_name'],
+            }
+        }
+
+        model_id = input_data['model_id']
+        interval_minutes = input_data['interval_minutes']
+
+        model_config = input_data['model_config']
+        target_name = model_config['target']
+
+        query = f"""
+            select p."timestamp", p.prediction, ld.value as "target"
+            from "{input_data['schema']}"."{input_data['predictions_table_name']}" p
+            inner join "{input_data['schema']}"."{input_data['data_table_name']}" ld
+                on p."timestamp" = ld."timestamp"
+            where
+                p.model_id = '{model_id}' and
+                p.prediction is not null and
+                ld.variable = '{target_name}' and
+                ld.value is not null and
+                p."timestamp" >= NOW() - INTERVAL '{interval_minutes} minutes'
+            order by
+                p."timestamp" desc;
+        """  # nosec B608 - values come from internal Temporal workflow config, not user input
+
+        target_data = await workflow.execute_activity_method(
+            Activities.load_custom_query,
+            {
+                **metadata,
+                'query': query,
+                'datetime_columns': ['timestamp'],
+                'orient': 'records',
+            },
+            retry_policy=retry_policy,
+            start_to_close_timeout=timedelta(seconds=300),
+        )
+
+        if not target_data:
+            return
+
+        simple_metrics = await workflow.execute_local_activity_method(
+            Activities.calculate_simple_metrics,
+            {
+                **metadata,
+                'model_id': model_id,
+                'target_data': target_data,
+                'metrics': input_data.get('metrics', ['rmse', 'mse', 'mae', 'r2']),
+                'interval_minutes': interval_minutes,
+            },
+            retry_policy=retry_policy,
+            start_to_close_timeout=timedelta(seconds=300),
+        )
+
+        if not simple_metrics:
+            return
+
+        await workflow.execute_activity_method(
+            Activities.export_data_to_postgres,
+            {
+                **metadata,
+                'data': simple_metrics,
+                'schema': input_data['schema'],
+                'table_name': input_data['target_table_name'],
+                'timestamp_conversion': {
+                    'column': 'timestamp',
+                    'format': DATETIME_FORMAT_WITH_TZ,
+                },
+            },
+            retry_policy=retry_policy,
+            start_to_close_timeout=timedelta(seconds=300),
+        )

laborious/workflows/sub_workflows/format_and_export_prediction.py

@@ -1,15 +1,17 @@
 from temporalio import workflow
 
 with workflow.unsafe.imports_passed_through():
-    from laborious.activities.activities import Activities
-    from typing import Any
     from datetime import timedelta
-    from sientia_do.temporal.policies import retry_policy
+    from typing import Any
+
     from sientia_do.temporal.constants import DATETIME_FORMAT_WITH_TZ
+    from sientia_do.temporal.policies import retry_policy
+
+    from laborious.activities.activities import Activities
 
 
-@workflow.defn(name="format_and_export_prediction")
-class FormatAndExportPrediction():
+@workflow.defn(name='subworkflow.format_and_export_prediction')
+class FormatAndExportPrediction:
     """
     Data formatting and export workflow for prediction results.
 
@@ -24,6 +26,7 @@ class FormatAndExportPrediction():
 
     Export Destinations:
     - PostgreSQL Database: Persistent storage with timestamp conversion
+    - PI Web API: Real-time industrial system integration for prediction and confidence values
     - OPC Servers: Real-time industrial system integration
     - Prometheus Metrics: Performance monitoring and operational visibility
     """
@@ -36,9 +39,10 @@ class FormatAndExportPrediction():
         This method orchestrates the complete data export process by:
         1. Determining the appropriate formatting strategy based on path_flag
         2. Formatting prediction data according to quality and requirements
-        3. Exporting data to OPC servers for real-time industrial access
-        4. Persisting data to PostgreSQL database with comprehensive metadata
-        5. Recording performance metrics for operational monitoring
+        3. Exporting data to PI Web API for real-time industrial access (if configured)
+        4. Exporting data to OPC servers for real-time industrial access (if configured)
+        5. Persisting data to PostgreSQL database with comprehensive metadata
+        6. Recording performance metrics for operational monitoring
 
         The method implements flexible formatting strategies:
         - Normal predictions: Full data formatting with confidence scores
@@ -48,29 +52,50 @@ class FormatAndExportPrediction():
         Args:
             input_data: Complete configuration for the export workflow
                 Required keys:
+                    - metadata (dict): Workflow execution metadata
                     - path_flag (str | None): Decision path flag for formatting strategy
+                        - None: Normal prediction path with full formatting
+                        - Any other value: Default prediction path for error conditions
                     - data (dict[str, Any]): Prediction data to format and export
                     - prediction_confidence (float): Confidence score for the prediction
                     - timestamp (str): ISO-formatted timestamp for the prediction
                     - model_id (int): Unique identifier for the ML model
                     - model_name (str): Name of the ML model
-                    - model_retention (str): Model retention policy configuration
-                    - comment (str): Operational comment or error description
                     - schema (str): Database schema for data storage
                     - table_name (str): Target table for data persistence
+                Optional keys:
                     - opc_output_config (dict[str, Any]): OPC server export configuration
-                    - prediction_store_policy (str, optional): Data retention policy
+                    - pi_web_api_output_config (dict[str, Any]): PI Web API export configuration
+                        Contains endpoint, prediction_tags, and confidence_tags mappings
+                    - transformed_data (dict[str, Any]): Transformed data to export separately
+                        Only processed when path_flag is None
+                    - transform_table_name (str): Target table for transformed data export
+                        Required if transformed_data is provided
+                    - prediction_store_policy (str): Data retention policy (e.g., 'lts:1', 'erl:2')
+                        Required when path_flag is None
+                    - comment (str): Operational comment or error description
+                        Required when path_flag is not None
 
         Returns:
-            bool: True if the workflow completes successfully, False otherwise
+            None: The workflow completes successfully when all export operations finish
+
+        Note:
+            When transformed_data is provided and path_flag is None, the workflow will:
+            1. Format the transformed data using format_transformed_data
+            2. Export it to a separate table (transform_table_name) asynchronously
+            3. Wait for both prediction and transformed data exports to complete
         """
         metadata = input_data['metadata']
         path_flag = input_data['path_flag']
         data = input_data['data']
+        transformed_data = input_data.get('transformed_data', None)
         prediction_confidence = input_data['prediction_confidence']
 
+        opc_output_config = input_data.get('opc_output_config', None)
+        pi_web_api_output_config = input_data.get('pi_web_api_output_config', None)
+
         if path_flag is None:
-            # proceed with formatting and exporting
+            # Normal prediction path: format prediction data with full metadata
             prediction = await workflow.execute_local_activity_method(
                 Activities.format_prediction,
                 {
@@ -79,14 +104,48 @@ class FormatAndExportPrediction():
                     'timestamp': input_data['timestamp'],
                     'model_id': input_data['model_id'],
                     'prediction_confidence': prediction_confidence,
-                    'prediction_store_policy': input_data['prediction_store_policy']
+                    'prediction_store_policy': input_data['prediction_store_policy'],
+                    'model_name': input_data['model_name'],
                 },
                 retry_policy=retry_policy,
-                start_to_close_timeout=timedelta(seconds=60)
+                start_to_close_timeout=timedelta(seconds=60),
             )
 
+            # Optionally format and export transformed data to separate table
+            if transformed_data is not None:
+                transformed = await workflow.execute_local_activity_method(
+                    Activities.format_transformed_data,
+                    {
+                        **metadata,
+                        'data': transformed_data,
+                        'model_id': input_data['model_id'],
+                        'model_name': input_data['model_name'],
+                    },
+                    retry_policy=retry_policy,
+                    start_to_close_timeout=timedelta(seconds=60),
+                )
+
+                write_transformed_handler = workflow.start_activity_method(
+                    Activities.export_payload_to_postgres,
+                    {
+                        **metadata,
+                        'schema': input_data['schema'],
+                        'table_name': input_data['transform_table_name'],
+                        'data': transformed,
+                        'timestamp_conversion': {
+                            'column': 'timestamp',
+                            'format': DATETIME_FORMAT_WITH_TZ,
+                        },
+                    },
+                    retry_policy=retry_policy,
+                    start_to_close_timeout=timedelta(seconds=60),
+                )
+
+            else:
+                write_transformed_handler = None
+
         else:
-            # create default prediction
+            # Error path: create default prediction with error indicators
             prediction = await workflow.execute_local_activity_method(
                 Activities.format_default_prediction,
                 {
@@ -94,23 +153,41 @@ class FormatAndExportPrediction():
                     'timestamp': input_data['timestamp'],
                     'model_id': input_data['model_id'],
                     'prediction_confidence': prediction_confidence,
-                    'comment': input_data['comment']
+                    'comment': input_data['comment'],
                 },
                 retry_policy=retry_policy,
-                start_to_close_timeout=timedelta(seconds=60)
+                start_to_close_timeout=timedelta(seconds=60),
+            )
+
+            write_transformed_handler = None
+
+        opc_metrics: dict[str, dict[str, float | None]] = {}
+
+        # write to pi web api
+        if pi_web_api_output_config:
+            prediction = await workflow.execute_activity_method(
+                Activities.write_pi_web_api_data,
+                {
+                    'pi_web_api_output_config': pi_web_api_output_config,
+                    'data': prediction,
+                    **metadata,
+                },
+                retry_policy=retry_policy,
+                start_to_close_timeout=timedelta(seconds=60),
             )
 
         # write to opc
-        prediction = await workflow.execute_activity_method(
-            Activities.write_opc_data,
-            {
-                **metadata,
-                'opc_output_config': input_data['opc_output_config'],
-                'data': prediction
-            },
-            retry_policy=retry_policy,
-            start_to_close_timeout=timedelta(seconds=60)
-        )
+        if opc_output_config:
+            prediction, opc_metrics = await workflow.execute_activity_method(
+                Activities.write_opc_data,
+                {
+                    'opc_output_config': opc_output_config,
+                    'data': prediction,
+                    **metadata,
+                },
+                retry_policy=retry_policy,
+                start_to_close_timeout=timedelta(seconds=60),
+            )
 
         # write to postgres
         await workflow.execute_activity_method(
@@ -120,21 +197,24 @@ class FormatAndExportPrediction():
                 'schema': input_data['schema'],
                 'table_name': input_data['table_name'],
                 'data': prediction,
-                'timestamp_conversion': {
-                    'column': 'timestamp',
-                    'format': DATETIME_FORMAT_WITH_TZ
-                }
+                'timestamp_conversion': {'column': 'timestamp', 'format': DATETIME_FORMAT_WITH_TZ},
+                'on_conflict': input_data.get('on_conflict', 'error'),
+                'unique_columns': ['model_id', 'timestamp'],
             },
             retry_policy=retry_policy,
-            start_to_close_timeout=timedelta(seconds=60)
+            start_to_close_timeout=timedelta(seconds=180),
         )
 
+        if write_transformed_handler is not None:
+            await write_transformed_handler
+
         await workflow.execute_activity_method(
             Activities.write_metrics,
             {
                 **metadata,
-                'prediction': prediction
+                'prediction': prediction,
+                'opc_metrics': opc_metrics,
             },
             retry_policy=retry_policy,
-            start_to_close_timeout=timedelta(seconds=60)
+            start_to_close_timeout=timedelta(seconds=60),
         )

laborious/workflows/sub_workflows/prediction_process.py

@@ -1,14 +1,16 @@
 from temporalio import workflow
 
 with workflow.unsafe.imports_passed_through():
-    from laborious.activities.activities import Activities
-    from typing import Any
-    from sientia_do.temporal.policies import retry_policy
     from datetime import timedelta
+    from typing import Any
+
+    from sientia_do.temporal.policies import retry_policy
+
+    from laborious.activities.activities import Activities
 
 
-@workflow.defn(name="prediction_process")
-class PredictionProcess():
+@workflow.defn(name='subworkflow.prediction_process')
+class PredictionProcess:
     """
     Core prediction processing workflow for the Laborious system.
 
@@ -64,7 +66,10 @@ class PredictionProcess():
                     - mlflow_predict_filters (dict): MLFlow prediction filters
                     - model_retention (int): Model retention period in minutes
                     - path_priority (list[str]): Decision path priority configuration
-                    - opc_output_config (dict): OPC server export configuration
+                    - opc_output_config (dict, optional): OPC server export configuration
+                    - pi_web_api_output_config (dict, optional): PI Web API export configuration
+                    - save_transform (bool, optional): Whether to save transformed data (default: True)
+                    - prediction_store_policy (str, optional): Data retention policy (default: 'lts:1')
 
         Returns:
             None: The workflow completes successfully when export workflow finishes
@@ -80,24 +85,51 @@ class PredictionProcess():
         model_id = input_data['model_id']
         model_name = input_data['model_name']
         model_config = input_data.get('model_config', {})
+        save_transform = input_data.get('save_transform', True)
 
-        # Get last timestamp for incremental processing
-        last_timestamp = await workflow.execute_local_activity_method(
-            Activities.get_last_timestamp,
-            {
-                **metadata,
-                'data': data
-            },
-            retry_policy=retry_policy,
-            start_to_close_timeout=timedelta(minutes=1),
-        )
+        try:
+            await self._run_prediction_pipeline(
+                input_data,
+                metadata,
+                data,
+                model_id,
+                model_name,
+                model_config,
+                save_transform,
+            )
+            await workflow.execute_activity_method(
+                Activities.cleanup_minio_objects_expired,
+                {**metadata, 'data': data},
+                retry_policy=retry_policy,
+                start_to_close_timeout=timedelta(minutes=5),
+            )
+        except Exception as e:
+            await workflow.execute_activity_method(
+                Activities.cleanup_minio_objects_expired,
+                {**metadata, 'data': data},
+                retry_policy=retry_policy,
+                start_to_close_timeout=timedelta(minutes=5),
+            )
+            raise e
+
+    async def _run_prediction_pipeline(
+        self,
+        input_data: dict[str, Any],
+        metadata: dict[str, Any],
+        data: dict[str, Any],
+        model_id: str,
+        model_name: str,
+        model_config: dict[str, Any],
+        save_transform: bool,
+    ) -> None:
+        last_timestamp = data['last_timestamp']
 
         # Apply input data quality gates
         gate_input = {
             **metadata,
             'filters': input_data['input_filters'],
             'data': data,
-            'path_priority': input_data['path_priority']
+            'path_priority': input_data['path_priority'],
         }
 
         path_flag, confidence, comment = await workflow.execute_local_activity_method(
@@ -114,14 +146,9 @@ class PredictionProcess():
             return
 
         # Request MLFlow model transformation
-        response_data = await workflow.execute_local_activity_method(
+        transformed_data = await workflow.execute_activity_method(
             Activities.request_transform,
-            {
-                **metadata,
-                'data': data,
-                'model_name': model_name,
-                'model_config': model_config
-            },
+            {**metadata, 'data': data, 'model_name': model_name, 'model_config': model_config},
             retry_policy=retry_policy,
             start_to_close_timeout=timedelta(minutes=5),
         )
@@ -132,9 +159,9 @@ class PredictionProcess():
             {
                 **metadata,
                 'filters': input_data['mlflow_transform_filters'],
-                'data': response_data,
+                'data': transformed_data,
                 'type': 'transform',
-                'path_priority': input_data['path_priority']
+                'path_priority': input_data['path_priority'],
             },
             retry_policy=retry_policy,
             start_to_close_timeout=timedelta(minutes=1),
@@ -146,8 +173,6 @@ class PredictionProcess():
         ):
             return
 
-        transformed_data = response_data['content']
-
         path_flag, confidence, comment = await workflow.execute_local_activity_method(
             Activities.mlflow_content_gate,
             {
@@ -155,7 +180,7 @@ class PredictionProcess():
                 'filters': input_data['mlflow_transform_filters'],
                 'data': transformed_data,
                 'type': 'transform',
-                'path_priority': input_data['path_priority']
+                'path_priority': input_data['path_priority'],
             },
             retry_policy=retry_policy,
             start_to_close_timeout=timedelta(minutes=1),
@@ -166,13 +191,13 @@ class PredictionProcess():
         ):
             return
 
-        response_data = await workflow.execute_local_activity_method(
+        predicted_data = await workflow.execute_activity_method(
             Activities.request_predict,
             {
                 **metadata,
                 'data': transformed_data,
                 'model_name': model_name,
-                'model_config': model_config
+                'model_config': model_config,
             },
             retry_policy=retry_policy,
             start_to_close_timeout=timedelta(minutes=5),
@@ -184,9 +209,9 @@ class PredictionProcess():
             {
                 **metadata,
                 'filters': input_data['mlflow_predict_filters'],
-                'data': response_data,
+                'data': predicted_data,
                 'type': 'predict',
-                'path_priority': input_data['path_priority']
+                'path_priority': input_data['path_priority'],
             },
             retry_policy=retry_policy,
             start_to_close_timeout=timedelta(minutes=1),
@@ -200,26 +225,37 @@ class PredictionProcess():
 
         # Delegate to export workflow for data persistence
         await workflow.execute_child_workflow(
-            'format_and_export_prediction',
+            'subworkflow.format_and_export_prediction',
             {
                 'metadata': metadata,
+                'on_conflict': input_data.get('on_conflict', 'error'),
                 'path_flag': path_flag,
-                'data': response_data['content'],
+                'data': predicted_data,
+                'transformed_data': transformed_data if save_transform else None,
                 'prediction_confidence': confidence,
                 'timestamp': last_timestamp,
                 'model_id': model_id,
                 'model_name': model_name,
                 'model_config': model_config,
                 'opc_output_config': input_data['opc_output_config'],
+                'pi_web_api_output_config': input_data['pi_web_api_output_config'],
                 'schema': input_data['schema'],
                 'table_name': input_data['table_name'],
+                'transform_table_name': input_data['transform_table_name'],
                 'comment': comment,
-                'prediction_store_policy': input_data['prediction_store_policy']
-            }
+                'prediction_store_policy': input_data['prediction_store_policy'],
+            },
         )
 
-    async def path_flag_handler(self, data: dict, path_flag: str, input_data: dict,
-                                confidence: int, last_timestamp: str, comment: str) -> bool:
+    async def path_flag_handler(
+        self,
+        data: dict[str, Any],
+        path_flag: str | None,
+        input_data: dict,
+        confidence: int,
+        last_timestamp: str,
+        comment: str,
+    ) -> bool:
         """
         Handle path decisions based on filter results and confidence levels.
 
@@ -230,7 +266,17 @@ class PredictionProcess():
         Args:
             data: Input data for processing
             path_flag: Path decision from filter (STOP, CONTINUE, REPEAT)
-            input_data: Complete workflow input configuration
+            input_data: Complete workflow input configuration including:
+                - metadata (dict): Workflow execution metadata
+                - schema (str): Database schema
+                - table_name (str): Target table for predictions
+                - transform_table_name (str): Target table for transformed data
+                - model_id (str): ML model identifier
+                - model_name (str): ML model name
+                - model_config (dict, optional): Model configuration
+                - opc_output_config (dict, optional): OPC server export configuration
+                - pi_web_api_output_config (dict, optional): PI Web API export configuration
+                - prediction_store_policy (str, optional): Data retention policy
             confidence: Confidence level from filter validation
             last_timestamp: Last processed timestamp
             comment: Additional information about the filter result
@@ -240,13 +286,14 @@ class PredictionProcess():
 
         Path Handling:
             - STOP: Terminates workflow execution
-            - CONTINUE: Proceeds with normal processing
+            - CONTINUE: Delegates to FormatAndExportPrediction workflow with current data
             - REPEAT: Repeats last prediction if available
         """
         metadata = input_data['metadata']
 
         schema = input_data['schema']
         table_name = input_data['table_name']
+        transform_table_name = input_data['transform_table_name']
         model_id = input_data['model_id']
         model_name = input_data['model_name']
         model_config = input_data.get('model_config', {})
@@ -265,7 +312,7 @@ class PredictionProcess():
                     'schema': schema,
                     'table_name': table_name,
                     'model': model_id,
-                    'last_timestamp': last_timestamp
+                    'last_timestamp': last_timestamp,
                 },
                 retry_policy=retry_policy,
                 start_to_close_timeout=timedelta(minutes=1),
@@ -274,7 +321,7 @@ class PredictionProcess():
         elif path_flag == 'CONTINUE':
             # call write workflow
             await workflow.execute_child_workflow(
-                'format_and_export_prediction',
+                'subworkflow.format_and_export_prediction',
                 {
                     'metadata': metadata,
                     'path_flag': path_flag,
@@ -286,10 +333,13 @@ class PredictionProcess():
                     'model_config': model_config,
                     'schema': schema,
                     'table_name': table_name,
+                    'transform_table_name': transform_table_name,
                     'comment': comment,
                     'opc_output_config': input_data['opc_output_config'],
-                    'prediction_store_policy': input_data['prediction_store_policy']
-                }
+                    'pi_web_api_output_config': input_data['pi_web_api_output_config'],
+                    'prediction_store_policy': input_data['prediction_store_policy'],
+                    'on_conflict': input_data.get('on_conflict', 'error'),
+                },
             )
             return True