Code import - branch feature/SIENTIAPDE-1646

orchestrator/__init__.py

@@ -0,0 +1 @@
+

orchestrator/activities/activities.py

@@ -0,0 +1,130 @@
+from temporalio import workflow
+
+with workflow.unsafe.imports_passed_through():
+    from logging import Logger
+    from typing import Any
+
+    from sientia_do.notifications.handlers import CoreNotificationHandler as NotificationHandler
+    from sientia_do.observability.metrics_controller import MetricsController
+    from sientia_do.temporal.activities.postgres_sync import Postgres
+
+    from orchestrator.activities.email import Email
+    from orchestrator.activities.formatters import Formatters
+    from orchestrator.activities.mongo_db import MongoDB
+    from orchestrator.activities.slot_manager import SlotManager
+    from orchestrator.activities.temporal_manager import TemporalManager
+
+
+class Activities(TemporalManager, SlotManager, Formatters, MongoDB, Email, Postgres):
+    """
+    Central activities orchestrator for Temporal workflow operations.
+
+    This class combines multiple activity components including temporal management,
+    slot management, data formatting, MongoDB operations, email services, and
+    PostgreSQL operations. It provides a unified interface for all activity
+    operations required by the orchestration workflows.
+
+    Args:
+        temporal_config (dict[str, Any]): Temporal server configuration
+        redis_config (dict[str, Any]): Redis server configuration
+        mongodb_config (dict[str, Any]): MongoDB connection configuration
+        email_config (dict[str, Any]): Email service configuration
+        postgres_config (dict[str, Any]): PostgreSQL database configuration
+        logger (Logger): Application logger instance
+        notification_handler (NotificationHandler): Notification management handler
+    """
+
+    def __init__(
+        self,
+        temporal_config: dict[str, Any],
+        # couchbase_config: dict[str, Any],
+        redis_config: dict[str, Any],
+        mongodb_config: dict[str, Any],
+        email_config: dict[str, Any],
+        postgres_config: dict[str, Any],
+        logger: Logger,
+        notification_handler: NotificationHandler,
+    ):
+        # Initialize parent classes
+
+        metrics_controller = MetricsController(logger=logger)
+
+        TemporalManager.__init__(
+            self,
+            host=temporal_config['temporal_host'],
+            scouter_namespace=temporal_config['temporal_scouter_namespace'],
+            laborious_namespace=temporal_config['temporal_laborious_namespace'],
+            logger=logger,
+            notification_handler=notification_handler,
+            metrics_controller=metrics_controller,
+        )
+
+        SlotManager.__init__(
+            self,
+            host=redis_config['host'],
+            port=redis_config['port'],
+            username=redis_config['username'],
+            password=redis_config['password'],
+            logger=logger,
+            notification_handler=notification_handler,
+            metrics_controller=metrics_controller,
+        )
+
+        Formatters.__init__(
+            self,
+            scouter_namespace=temporal_config['temporal_scouter_namespace'],
+            laborious_namespace=temporal_config['temporal_laborious_namespace'],
+            logger=logger,
+            notification_handler=notification_handler,
+            metrics_controller=metrics_controller,
+        )
+
+        MongoDB.__init__(
+            self,
+            connection_string=mongodb_config['connection_string'],
+            database_name=mongodb_config['database_name'],
+            ttl_index_seconds=mongodb_config['ttl_index_seconds'],
+            logger=logger,
+            notification_handler=notification_handler,
+            metrics_controller=metrics_controller,
+        )
+
+        Email.__init__(
+            self,
+            sender_email=email_config['sender_email'],
+            sender_password=email_config['sender_password'],
+            smtp_server=email_config['smtp_server'],
+            smtp_port=email_config['smtp_port'],
+            logger=logger,
+            notification_handler=notification_handler,
+            metrics_controller=metrics_controller,
+        )
+
+        Postgres.__init__(
+            self,
+            host=postgres_config['host'],
+            port=postgres_config['port'],
+            user=postgres_config['user'],
+            password=postgres_config['password'],
+            dbname=postgres_config['dbname'],
+            min_connections=postgres_config['min_connections'],
+            max_connections=postgres_config['max_connections'],
+            logger=logger,
+            notification_handler=notification_handler,
+            metrics_controller=metrics_controller,
+        )
+
+    def shutdown(self):
+        """
+        Shutdown all connections and clean up resources.
+
+        This method gracefully shuts down all database connections, email
+        services, and other resources to ensure proper cleanup when the
+        application terminates.
+        """
+        MongoDB.close(self)
+        Postgres.close(self)
+        Email.close(self)
+        Formatters.close(self)
+        SlotManager.close(self)
+        TemporalManager.close(self)

orchestrator/activities/email.py

@@ -0,0 +1,269 @@
+from temporalio import activity, workflow
+
+with workflow.unsafe.imports_passed_through():
+    import smtplib
+    import traceback
+    from email import encoders
+    from email.mime.base import MIMEBase
+    from email.mime.multipart import MIMEMultipart
+    from email.mime.text import MIMEText
+    from smtplib import SMTPServerDisconnected
+    from typing import Any
+
+    from sientia_do.notifications.handlers import NotificationHandler
+    from sientia_do.observability.logger import Logger
+    from sientia_do.observability.sientia_monitoring import MetricsController, SientiaMonitoring
+
+    from orchestrator import metrics
+    from orchestrator.utils.email_builder import EmailBuilder
+
+
+class Email(SientiaMonitoring):
+    """
+    Email service activity for sending workflow notifications.
+
+    This class provides email sending capabilities including HTML email
+    generation, attachment handling, and SMTP connection management with
+    automatic reconnection for workflow notification delivery.
+
+    Args:
+        sender_email (str): Email address for sending messages
+        sender_password (str): SMTP authentication password
+        smtp_server (str): SMTP server hostname
+        smtp_port (int): SMTP server port number
+        logger (Logger): Application logger instance
+        notification_handler (NotificationHandler): Notification management handler
+    """
+
+    def __init__(
+        self,
+        sender_email: str,
+        sender_password: str,
+        smtp_server: str,
+        smtp_port: int,
+        logger: Logger,
+        notification_handler: NotificationHandler,
+        metrics_controller: MetricsController,
+    ):
+        self.email_builder = EmailBuilder(logger=logger)
+
+        self.sender_email = sender_email
+        self.sender_password = sender_password
+        self.smtp_port = smtp_port
+        self.smtp_server = smtp_server
+
+        logger.info(f'Initializing Email with {smtp_server}:{smtp_port}')
+
+        if smtp_server is not None:
+            self.server = smtplib.SMTP(smtp_server, smtp_port, timeout=20)
+
+            if self.sender_password:
+                self.server.starttls()
+                self.server.login(self.sender_email, self.sender_password)
+
+        SientiaMonitoring.__init__(
+            self,
+            logger=logger,
+            notification_handler=notification_handler,
+            metrics_controller=metrics_controller,
+        )
+
+    def close(self):
+        """
+        Close the Email connection and clean up resources.
+
+        Closes the SMTP server connection and shuts down the SientiaMonitoring instance.
+        """
+        self.server.quit()
+        SientiaMonitoring.shutdown(self)
+
+    def __del__(self):
+        """
+        Ensure the Email connection is closed when the object is garbage-collected.
+        """
+        self.close()
+
+    @activity.defn(name='build_email_html')
+    def build_email_html(self, input_data: dict[str, Any]) -> dict[str, Any]:
+        """
+        Build HTML email content for configured receiver groups.
+
+        This activity generates HTML email content for each receiver group
+        based on notification data and mail type. It processes notification
+        data through the email builder to create formatted HTML messages.
+
+        Args:
+            input_data (dict[str, Any]): Activity input parameters.
+                Required fields:
+                - metadata (dict[str, Any]): Workflow execution metadata
+                - receiver_groups (dict[str, Any]): Receiver group configurations with notifications
+                - mail_type (str): Type of email (Alerts/Reports)
+
+        Returns:
+            dict[str, Any]: Updated receiver groups with generated HTML content
+
+        Raises:
+            Exception: If HTML generation fails
+        """
+        metadata = input_data['metadata']
+        receiver_groups = input_data['receiver_groups']
+        mail_type = input_data['mail_type']
+
+        self.info(f'Building email html for {mail_type} mail type.', metadata=metadata)
+
+        for _group_name, group_config in receiver_groups.items():
+            html = self.email_builder.build_email(group_config['notifications'], mail_type)
+
+            group_config['html'] = html
+
+        self.info(f'Email html built for {mail_type} mail type.', metadata=metadata)
+
+        return receiver_groups
+
+    def handle_attachments(self, attachments: list[dict], msg: MIMEMultipart) -> MIMEMultipart:
+        """
+        Attach a list of attachments to an email message.
+
+        Processes notification attachments and adds them to the email message
+        as base64-encoded MIME parts. Each attachment contains error details
+        or additional context for the notification.
+
+        Args:
+            attachments (list[dict]): A list of dictionaries where each dictionary contains:
+                - filename (str): Name of the attachment file
+                - attachment_content (str): Content of the attachment
+            msg (MIMEMultipart): The email message object to which the attachments will be added
+
+        Returns:
+            MIMEMultipart: The email message object with the attachments added
+
+        Raises:
+            Exception: If an attachment cannot be added, an error is logged and raised
+        """
+
+        for attachment in attachments:
+            att_name = attachment['filename']
+            try:
+                # Create the attachment as a MIMEBase object
+                part = MIMEBase('application', 'octet-stream')
+                part.set_payload(attachment['attachment_content'].encode('utf-8'))
+                encoders.encode_base64(part)
+                part.add_header('Content-Disposition', f'attachment; filename="{att_name}"')
+                msg.attach(part)
+            except Exception as e:
+                self.logger.error(f'Failed to attach content of {att_name}: {e}')
+
+                raise e
+
+        return msg
+
+    def try_send_email(self, msg: MIMEMultipart, receivers: str):
+        """
+        Send an email to the receivers with automatic reconnection handling.
+
+        Attempts to send an email and automatically reconnects to the SMTP server
+        if a disconnection occurs during transmission.
+
+        Args:
+            msg (MIMEMultipart): The email message to send
+            receivers (str): Comma-separated list of email addresses to send to
+
+        Raises:
+            Exception: If email sending fails after reconnection attempts
+        """
+        try:
+            self.server.sendmail(self.sender_email, receivers, msg.as_string())
+        except SMTPServerDisconnected as e:
+            self.logger.error(f'SMTP server disconnected: {e}')
+            self.logger.info(f'Reconnecting to {self.smtp_server}:{self.smtp_port}')
+
+            if self.server:
+                try:
+                    self.server.quit()
+                except SMTPServerDisconnected as e:
+                    self.logger.info(f'Server already disconnected: {e}')
+                except Exception as e:
+                    self.logger.error(f'Failed to quit server: {e}')
+                    raise e
+
+            self.server = smtplib.SMTP(self.smtp_server, self.smtp_port, timeout=20)
+            if self.sender_password:
+                self.server.starttls()
+                self.server.login(self.sender_email, self.sender_password)
+            self.server.sendmail(self.sender_email, receivers, msg.as_string())
+
+    @activity.defn(name='send_email')
+    def send_email(self, input_data: dict[str, Any]) -> dict[str, Any]:
+        """
+        Send email notifications to configured receiver groups.
+
+        This activity sends HTML emails with attachments to all configured
+        receiver groups. It handles SMTP connection management, attachment
+        processing, and error reporting with automatic reconnection support.
+
+        Args:
+            input_data (dict[str, Any]): Activity input parameters.
+                Required fields:
+                - metadata (dict[str, Any]): Workflow execution metadata
+                - receiver_groups (dict[str, Any]): Receiver groups with HTML content
+                - mail_type (str): Type of email being sent (Alerts/Reports)
+
+        Returns:
+            dict[str, Any]: Updated receiver groups with sending status
+
+        Raises:
+            Exception: If email sending fails for all groups
+        """
+        metadata = input_data['metadata']
+        receiver_groups = input_data['receiver_groups']
+        mail_type = input_data['mail_type']
+
+        if self.smtp_server is None:
+            self.info(f'Skipping email sending for {mail_type} mail type.', metadata=metadata)
+            return {}
+
+        self.info(f'Sending email for {mail_type} mail type.', metadata=metadata)
+
+        for group_name, group_config in receiver_groups.items():
+            try:
+                receivers = ', '.join(group_config['members'])
+
+                self.info(f'Sending email to {group_name}: {receivers}', metadata=metadata)
+
+                msg = MIMEMultipart()
+                msg.attach(MIMEText(group_config['html'], 'html'))
+                msg['From'] = self.sender_email
+                msg['To'] = receivers
+                msg['Subject'] = f'SIENTIA™ {mail_type}'
+
+                msg = self.handle_attachments(
+                    [
+                        {
+                            'filename': f'{notification["trigger"]}_{notification["notification_id"]}.txt',
+                            'attachment_content': notification['attachment_content'],
+                        }
+                        for notification in group_config['notifications']
+                        if notification.get('attachment_content') is not None
+                    ],
+                    msg,
+                )
+
+                self.try_send_email(msg, receivers)
+            except Exception as e:
+                self.error(f'Failed to send email to {group_name}: {e}', metadata=metadata)
+                traceback.print_exc()
+                group_config['status'] = 'failed'
+            else:
+                group_config['status'] = 'sent'
+                metrics.EMAIL_SENT_COUNT.labels(
+                    pod_id=self.pod_id,
+                    model_name=metadata['model_name'],
+                    pipeline_name=metadata['workflow_name'],
+                    email_group=group_name,
+                ).inc()
+
+                self.info(f'Email sent to {group_name}: {receivers}', metadata=metadata)
+
+        self.info(f'Email sent for {mail_type} mail type.', metadata=metadata)
+
+        return receiver_groups

orchestrator/activities/formatters.py

@@ -0,0 +1,854 @@
+from temporalio import activity, workflow
+
+with workflow.unsafe.imports_passed_through():
+    import json
+    from collections.abc import Callable
+    from logging import Logger
+    from math import ceil
+    from typing import Any, TypedDict, cast
+
+    from pandas import DataFrame
+    from sientia_do.notifications.handlers import CoreNotificationHandler as NotificationHandler
+    from sientia_do.notifications.models import NotificationLevel
+    from sientia_do.observability.sientia_monitoring import MetricsController, SientiaMonitoring
+    from sientia_do.temporal.constants import DATETIME_FORMAT_MS_WITH_TZ, now
+
+    from orchestrator.utils.orchestrator_functions import (
+        build_tag_config,
+        drift,
+        gather_read_tags,
+        minimal_retrain,
+        pi_web_api_scouter,
+        predictions_batch,
+        scouter,
+        simple_metrics,
+    )
+
+topic_separator = '\n ========== \n'
+
+
+class ScheduleType(TypedDict):
+    """
+    Type definition for schedule type configuration entries.
+
+    Each schedule type entry maps a workflow type string to its corresponding
+    namespace and configuration builder function.
+
+    Attributes:
+        namespace (str): The Temporal namespace where workflows of this type execute.
+            Valid values: 'scouter', 'laborious'
+        function (Callable): Configuration builder function that transforms pipeline
+            configuration into Temporal-compatible workflow arguments
+    """
+
+    namespace: str
+    function: Callable
+
+
+schedule_types: dict[str, ScheduleType] = {
+    'scouter': {
+        'namespace': 'scouter',
+        'function': scouter,
+    },
+    'pi_web_api_scouter': {
+        'namespace': 'scouter',
+        'function': pi_web_api_scouter,
+    },
+    'predictions_batch': {
+        'namespace': 'laborious',
+        'function': predictions_batch,
+    },
+    'xgboost_predictions_batch': {
+        'namespace': 'laborious',
+        'function': predictions_batch,
+    },
+    'minimal_retrain': {
+        'namespace': 'laborious',
+        'function': minimal_retrain,
+    },
+    'xgboost_minimal_retrain': {
+        'namespace': 'laborious',
+        'function': minimal_retrain,
+    },
+    'drift': {
+        'namespace': 'laborious',
+        'function': drift,
+    },
+    'simple_metrics': {
+        'namespace': 'laborious',
+        'function': simple_metrics,
+    },
+}
+"""
+Registry mapping workflow types to their namespace and configuration builder functions.
+
+Supported workflow types:
+    - scouter: OPC data collection using OPC UA protocol
+    - pi_web_api_scouter: Data collection using PI Web API
+    - predictions_batch: ML model prediction workflows with OPC write-back
+    - minimal_retrain: Model retraining workflows using SQL queries
+    - drift: Data drift detection and monitoring workflows
+    - simple_metrics: Model performance metrics computation workflows
+
+Each entry specifies:
+    - namespace: Target Temporal namespace for workflow execution
+    - function: Configuration builder that transforms MongoDB pipeline config
+                into Temporal workflow arguments
+"""
+
+
+class Formatters(SientiaMonitoring):
+    """
+    Schedule and slot configuration formatting and notification filtering activity.
+
+    This class provides comprehensive formatting operations for schedules and OPC slots,
+    converting pipeline configurations into Temporal-compatible formats, managing slot
+    distribution across active ingestors, and implementing notification filtering for
+    scheduled reports.
+
+    Key features:
+    - Pipeline schedule configuration formatting ("scouter", "predictions_batch", "minimal_retrain", "drift")
+    - OPC slot distribution across active ingestors
+    - Notification filtering for comprehensive scheduled reports
+    - Group-based report filtering with ignore list support
+    - Configuration validation and transformation
+
+    Args:
+        scouter_namespace (str): Scouter workflow namespace
+        laborious_namespace (str): Laborious workflow namespace
+        logger (Logger): Application logger instance
+        notification_handler (NotificationHandler): Notification management handler
+    """
+
+    def __init__(
+        self,
+        scouter_namespace: str,
+        laborious_namespace: str,
+        logger: Logger,
+        notification_handler: NotificationHandler,
+        metrics_controller: MetricsController,
+    ):
+        self.scouter_namespace = scouter_namespace
+        self.laborious_namespace = laborious_namespace
+        SientiaMonitoring.__init__(
+            self,
+            logger=logger,
+            notification_handler=notification_handler,
+            metrics_controller=metrics_controller,
+        )
+
+    def close(self):
+        """
+        Close the Formatters connection and clean up resources.
+
+        Shuts down the SientiaMonitoring instance and releases all resources.
+        """
+        SientiaMonitoring.shutdown(self)
+
+    def __del__(self):
+        """
+        Ensure the Formatters connection is closed when the object is garbage-collected.
+        """
+        self.close()
+
+    @activity.defn(name='process_schedules')
+    def process_schedules(self, input_data: dict[str, Any]) -> dict[str, Any]:
+        """
+        Process pipeline configurations into Temporal-compatible schedule configurations.
+
+        This method transforms pipeline configurations from MongoDB into properly formatted
+        Temporal schedule configurations, organizing them by workflow type (scouter and
+        laborious) and applying the appropriate configuration builders for each pipeline type.
+
+        Pipeline types supported:
+        - "scouter": Data collection workflows with OPC tag configurations
+        - "predictions_batch": ML prediction workflows with OPC write configurations
+        - "minimal_retrain": Model retraining workflows with SQL query configurations
+
+        Args:
+            - input_data (dict[str, Any]): The input data containing
+              the schedules to process.
+                - pipelines (list[dict[str, Any]]): The schedules to process.
+
+        Returns:
+            - dict[str, Any]: The schedule configuration dictionary keyed by namespace
+        """
+
+        metadata = input_data.get('metadata', {})
+
+        self.info('Processing schedules...', metadata=metadata)
+
+        pipelines = input_data['pipelines']
+
+        schedule_config: dict[str, dict[str, Any]] = {
+            self.scouter_namespace: {},
+            self.laborious_namespace: {},
+        }
+
+        for pipeline in pipelines:
+            workflow_type = pipeline['workflow_type']
+
+            if workflow_type not in schedule_types:
+                self.error(f'Workflow type {workflow_type} not supported', metadata=metadata)
+                continue
+
+            schedule_type = schedule_types[workflow_type]
+            namespace = schedule_type['namespace']
+            function = schedule_type['function']
+
+            schedule_config[namespace][pipeline['schedule_name']] = {
+                **function(pipeline),
+                'updated_at': pipeline.get(
+                    'updated_at', now().strftime(DATETIME_FORMAT_MS_WITH_TZ)
+                ),
+            }
+
+        self.info('Processed schedules', metadata=metadata)
+        self.debug(json.dumps(schedule_config, indent=4, sort_keys=True), metadata=metadata)
+
+        return schedule_config
+
+    @activity.defn(name='process_slots')
+    def process_slots(self, input_data: dict[str, Any]) -> dict[str, Any]:
+        """
+        Extracts all read tags from input pipelines, divides them into slots and
+        returns a slot config dictionary. If no ingestor is available, only one slot
+        is created.
+
+        Args:
+            - input_data (dict[str, Any]): The input data containing
+              the schedules to process.
+                - pipelines (list[dict[str, Any]]): The schedules to process.
+                - opc_servers (list[str]): The OPC servers to create ingestor config.
+                - active_ingestors (list[str]): The active ingestors to divide into slots.
+
+        Returns:
+            - dict[str, Any]: The slot configuration dictionary keyed by slot id (as string)
+        """
+
+        metadata = input_data.get('metadata', {})
+
+        self.info('Processing slots...', metadata=metadata)
+
+        pipelines = input_data['pipelines']
+        opc_servers_list = input_data['opc_servers']
+        active_ingestors = input_data['active_ingestors']
+
+        opc_servers = {}
+        for server in opc_servers_list:
+            opc_servers[server['id']] = {
+                **server,
+            }
+
+        tags = list(gather_read_tags(pipelines).values())
+
+        number_of_tags = len(tags)
+        number_of_slots = len(active_ingestors) if active_ingestors else 1
+        tags_per_slot = ceil(number_of_tags / number_of_slots)
+
+        slot_config: dict[str, Any] = {}
+        last_index = 0
+
+        for i in range(1, number_of_slots):
+            slot_tags = tags[last_index : last_index + tags_per_slot]
+            slot_config[f'{i}'], notifications = build_tag_config(slot_tags, opc_servers)
+
+            if notifications:
+                self.send_notification(
+                    metadata=metadata,
+                    notification_id='ORCHESTRATOR_SERVER_NOT_FOUND_DURING_SLOT_CONFIGURATION',
+                    message=f'Servers {", ".join(notifications)} not found in opc_servers',
+                    block='orchestrator',
+                    level=NotificationLevel.ERROR,
+                )
+
+            last_index += tags_per_slot
+
+        slot_tags = tags[last_index:]
+        slot_config[f'{number_of_slots}'], notifications = build_tag_config(slot_tags, opc_servers)
+        if notifications:
+            self.send_notification(
+                metadata=metadata,
+                notification_id='ORCHESTRATOR_SERVER_NOT_FOUND_DURING_SLOT_CONFIGURATION',
+                message=f'Servers {", ".join(notifications)} not found in opc_servers',
+                block='orchestrator',
+                level=NotificationLevel.ERROR,
+            )
+
+        self.info('Processed slots', metadata=metadata)
+        self.debug(json.dumps(slot_config, indent=4, sort_keys=True), metadata=metadata)
+
+        return slot_config
+
+    @activity.defn(name='format_schedule_config')
+    def format_schedule_config(self, input_data: dict[str, Any]) -> dict[str, Any]:
+        """
+        Format the schedule config to a dictionary with the schedule name as the key.
+
+        Transforms a list of schedule configurations into a nested dictionary structure
+        organized by namespace and schedule name for efficient lookup and comparison.
+
+        Args:
+            input_data (dict[str, Any]): The input data containing:
+                - schedule_config (list[dict[str, Any]]): The schedule config to format
+                - metadata (dict): Metadata for logging purposes
+
+        Returns:
+            dict[str, Any]: The formatted schedule config organized by namespace and schedule name
+        """
+        metadata = input_data['metadata']
+
+        self.info('Formatting schedule config...', metadata=metadata)
+
+        schedule_config = input_data['schedule_config']
+
+        config: dict[str, dict[str, str]] = {}
+        for schedule in schedule_config:
+            namespace = schedule['namespace']
+            schedule_name = schedule['schedule_name']
+            updated_at = schedule['updated_at']
+
+            if namespace not in config:
+                config[namespace] = {}
+
+            config[namespace][schedule_name] = updated_at
+
+        self.info('Formatted schedule config', metadata=metadata)
+        self.debug(json.dumps(config, indent=4, sort_keys=True), metadata=metadata)
+
+        return config
+
+    def compare_config_timestamps(
+        self,
+        schedules: dict[str, Any],
+        current_schedules: dict[str, Any],
+        to_update: dict[str, Any],
+        to_create: dict[str, Any],
+        namespace: str,
+        metadata: dict[str, Any],
+    ):
+        """
+        Compare new and current schedules to determine which should be updated or created.
+
+        Compares schedule timestamps to identify schedules that need updating (newer timestamp)
+        or creating (schedule doesn't exist). Results are accumulated in the provided dictionaries.
+
+        Args:
+            schedules (dict[str, Any]): New schedules to compare
+            current_schedules (dict[str, Any]): Existing schedules to compare against
+            to_update (dict[str, Any]): Output accumulator for schedules that need updating
+            to_create (dict[str, Any]): Output accumulator for schedules that need creating
+            namespace (str): Namespace for the schedules being compared
+            metadata (dict[str, Any]): Metadata for logging purposes
+        """
+        for schedule_name, schedule in schedules.items():
+            if schedule_name in current_schedules:
+                update_timestamp = schedule.get('updated_at', now())
+
+                old_timestamp = current_schedules[schedule_name]
+
+                self.debug(
+                    f'Comparing schedule {schedule_name}:{update_timestamp} vs {old_timestamp}',
+                    metadata=metadata,
+                )
+
+                if update_timestamp > old_timestamp:
+                    to_update[namespace][schedule_name] = schedule
+            else:
+                to_create[namespace][schedule_name] = schedule
+
+    @activity.defn(name='create_schedule_config')
+    def create_schedule_config(self, input_data: dict[str, Any]) -> dict[str, Any]:
+        """
+        Create a schedule config dictionary based on the input data.
+
+        Compares the current schedule configuration in Temporal with the new schedule
+        configuration to determine which schedules need to be created, updated, or deleted.
+        Uses timestamp comparison to identify schedules that have changed.
+
+        Args:
+            input_data (dict[str, Any]): The input data containing:
+                - current_schedule_config (dict[str, Any]): The current schedule config in Temporal server
+                - schedule_config (dict[str, Any]): The schedule config to process
+                - metadata (dict): Metadata for logging purposes
+
+        Returns:
+            dict[str, Any]: A dictionary with keys:
+                - to_update (dict): Schedules that need updating
+                - to_create (dict): Schedules that need creating
+                - to_delete (dict): Schedules that need deleting
+        """
+
+        metadata = input_data.get('metadata', {})
+
+        self.info('Creating schedule config...', metadata=metadata)
+
+        current_schedule_config = input_data['current_schedule_config']
+        schedule_config = input_data['schedule_config']
+
+        to_update: dict[str, dict[str, Any]] = {
+            self.scouter_namespace: {},
+            self.laborious_namespace: {},
+        }
+        to_create: dict[str, dict[str, Any]] = {
+            self.scouter_namespace: {},
+            self.laborious_namespace: {},
+        }
+        to_delete: dict[str, list[str]] = {
+            self.scouter_namespace: [],
+            self.laborious_namespace: [],
+        }
+
+        for namespace, schedules in schedule_config.items():
+            current_schedules = current_schedule_config.get(namespace, {})
+            self.compare_config_timestamps(
+                schedules, current_schedules, to_update, to_create, namespace, metadata
+            )
+
+        for namespace, schedules in current_schedule_config.items():
+            for schedule_name in schedules:
+                if schedule_name not in schedule_config[namespace]:
+                    to_delete[namespace].append(schedule_name)
+
+        output = {'to_update': to_update, 'to_create': to_create, 'to_delete': to_delete}
+
+        self.info('Created schedule config', metadata=metadata)
+        self.debug(json.dumps(output, indent=4, sort_keys=True), metadata=metadata)
+
+        return output
+
+    @activity.defn(name='create_slot_config')
+    def create_slot_config(self, input_data: dict[str, Any]) -> dict[str, Any]:
+        """
+        Create a slot config dictionary based on the input data.
+
+        Compares the current slot configuration in Redis with the new slot configuration
+        to determine which slots need to be inserted or deleted. Slots are identified
+        by numeric IDs, and excess slots are marked for deletion.
+
+        Args:
+            input_data (dict[str, Any]): The input data containing:
+                - current_slot_config (dict[str, Any]): The current slot config in Redis
+                - slot_config (dict[str, Any]): The slot config to process
+                - metadata (dict): Metadata for logging purposes
+
+        Returns:
+            dict[str, Any]: A dictionary containing:
+                - to_insert (dict): Slots that need to be inserted/updated
+                - to_delete (list[str]): Slot IDs that need to be deleted
+        """
+
+        metadata = input_data.get('metadata', {})
+
+        self.info('Creating slot config...', metadata=metadata)
+
+        current_slot_config = input_data['current_slot_config']
+        slot_config = input_data['slot_config']
+        to_delete = []
+
+        number_of_current_slots = len(current_slot_config)
+        number_of_slots = len(slot_config)
+
+        if number_of_current_slots > number_of_slots:
+            to_delete = [str(i) for i in range(number_of_slots + 1, number_of_current_slots + 1)]
+
+        output = {'to_delete': to_delete, 'to_insert': slot_config}
+
+        self.info('Created slot config', metadata=metadata)
+        self.debug(json.dumps(output, indent=4, sort_keys=True), metadata=metadata)
+
+        return output
+
+    def send_success_report(
+        self,
+        metadata: dict[str, Any],
+        message: str,
+        notification_id: str,
+        attachment: Any | None = None,
+    ) -> None:
+        """
+        Send a success notification report.
+
+        Sends an INFO-level notification to the notification handler with success
+        details about orchestration operations.
+
+        Args:
+            metadata (dict[str, Any]): Metadata for the notification
+            message (str): The success message to send
+            notification_id (str): The ID of the notification to send
+            attachment (Any | None, optional): Optional attachment content to include
+        """
+        self.send_notification(
+            metadata=metadata,
+            notification_id=notification_id,
+            message=message,
+            block='report_orchestration',
+            level=NotificationLevel.INFO,
+            attachment_content=json.dumps(attachment, indent=4, sort_keys=True),
+        )
+
+    def send_error_report(
+        self, metadata: dict[str, Any], message: str, notification_id: str, attachment: str
+    ) -> None:
+        """
+        Send an error notification report.
+
+        Sends an ERROR-level notification to the notification handler with error
+        details about orchestration operation failures.
+
+        Args:
+            metadata (dict[str, Any]): Metadata for the notification
+            message (str): The error message to send
+            notification_id (str): The ID of the notification
+            attachment (str): The attachment content for the notification
+        """
+        self.send_notification(
+            metadata=metadata,
+            notification_id=notification_id,
+            message=message,
+            block='report_orchestration',
+            level=NotificationLevel.ERROR,
+            attachment_content=attachment,
+        )
+
+    def parse_report_schedule(
+        self, input_data: list[dict[str, Any]]
+    ) -> tuple[list[str], dict[str, Any]]:
+        """
+        Parse the report schedule data to extract success and error information.
+
+        Processes schedule operation reports and separates successful operations
+        from failed ones, formatting keys as "namespace/schedule_name" for consistency.
+
+        Args:
+            input_data (list[dict[str, Any]]): The schedule reports. Each item must
+                contain 'namespace', 'schedule_name', 'success', 'message', and optionally
+                'attachment'
+
+        Returns:
+            tuple[list[str], dict[str, Any]]: A tuple of:
+                - Successful schedule keys in the form "namespace/schedule_name"
+                - Error map keyed by the same string to error details
+        """
+        success_keys = [
+            f'{value["namespace"]}/{value["schedule_name"]}'
+            for value in input_data
+            if value['success']
+        ]
+
+        error_keys = {
+            f'{value["namespace"]}/{value["schedule_name"]}': {
+                'message': value['message'],
+                'attachment': value.get('attachment', None),
+            }
+            for value in input_data
+            if not value['success']
+        }
+
+        return success_keys, error_keys
+
+    def parse_report(self, input_data: dict[str, dict[str, Any]]) -> tuple[list[str], list[str]]:
+        """
+        Parse the report data to extract success and error keys.
+
+        Processes operation reports and separates successful operations from failed ones
+        based on the 'success' field in each report item.
+
+        Args:
+            input_data (dict[str, dict[str, Any]]): The input data containing report items.
+                Each item should have a 'success' field indicating success/failure
+
+        Returns:
+            tuple[list[str], list[str]]: A tuple containing:
+                - List of successful keys
+                - List of error keys
+        """
+        success_keys = [key for key, value in input_data.items() if value['success']]
+
+        error_keys = [key for key, value in input_data.items() if not value['success']]
+
+        return success_keys, error_keys
+
+    def manage_and_send_report(
+        self,
+        metadata: dict[str, Any],
+        success_keys: list[str],
+        error_keys: dict[str, Any],
+        schedule_type: str,
+        schedule_data: dict[str, Any],
+    ):
+        """
+        Manage and send success and error reports based on the provided keys and data.
+
+        Sends separate notifications for successful and failed operations, formatting
+        error messages with attachments when available.
+
+        Args:
+            metadata (dict[str, Any]): Metadata for logging and notifications
+            success_keys (list[str]): List of keys that were successful
+            error_keys (dict[str, Any]): Dictionary of error keys mapped to error details
+            schedule_type (str): The type of schedule being reported (e.g., 'created schedules')
+            schedule_data (dict[str, Any]): The schedule data containing items and notification ID
+        """
+        if len(success_keys) > 0:
+            self.send_success_report(
+                metadata=metadata,
+                message=f'Successfully {schedule_type}: \n {", ".join(success_keys)}',
+                notification_id=schedule_data['id'],
+                attachment=schedule_data['items'],
+            )
+
+        if len(error_keys) > 0:
+            attachment = []
+            for key, value in error_keys.items():
+                if value['attachment'] is not None:
+                    attachment.append(f'{key}:\n{value["message"]}\n{value["attachment"]}')
+                else:
+                    attachment.append(f'{key}:\n{value["message"]}')
+
+            self.send_error_report(
+                metadata=metadata,
+                message=f'Fails on {schedule_type}: \n {", ".join(error_keys)}',
+                notification_id=f'{schedule_data["id"]}_ERROR',
+                attachment=topic_separator.join(attachment),
+            )
+
+    @activity.defn(name='report_schedule_orchestration')
+    def report_schedule_orchestration(self, input_data: dict[str, Any]) -> None:
+        """
+        Report the orchestration result to the notification handler.
+
+        Processes schedule orchestration results and sends notifications for
+        created, updated, and deleted schedules with success and error details.
+
+        Args:
+            input_data (dict[str, Any]): The input data containing:
+                - created_schedules (list[dict[str, Any]]): The created schedules
+                - updated_schedules (list[dict[str, Any]]): The updated schedules
+                - deleted_schedules (list[dict[str, Any]]): The deleted schedules
+                - metadata (dict): Metadata for logging purposes
+        """
+        metadata = input_data.get('metadata', {})
+
+        self.info('Reporting orchestration...', metadata=metadata)
+
+        created_schedules = input_data['created_schedules']
+        updated_schedules = input_data['updated_schedules']
+        deleted_schedules = input_data['deleted_schedules']
+
+        schedules_report = {
+            'created schedules': {
+                'items': created_schedules,
+                'id': 'REPORT_ORCHESTRATION_CREATED_SCHEDULES',
+            },
+            'updated schedules': {
+                'items': updated_schedules,
+                'id': 'REPORT_ORCHESTRATION_UPDATED_SCHEDULES',
+            },
+            'deleted schedules': {
+                'items': deleted_schedules,
+                'id': 'REPORT_ORCHESTRATION_DELETED_SCHEDULES',
+            },
+        }
+
+        for schedule_type, schedule_data in schedules_report.items():
+            if len(schedule_data['items']) > 0:
+                success_keys, error_keys = self.parse_report_schedule(schedule_data['items'])
+
+                self.manage_and_send_report(
+                    metadata=metadata,
+                    success_keys=success_keys,
+                    error_keys=error_keys,
+                    schedule_type=schedule_type,
+                    schedule_data=schedule_data,
+                )
+
+    @activity.defn(name='report_slot_orchestration')
+    def report_slot_orchestration(self, input_data: dict[str, Any]) -> None:
+        """
+        Report the slot orchestration result to the notification handler.
+
+        Processes slot orchestration results and sends notifications for
+        inserted and deleted slots with success and error details.
+
+        Args:
+            input_data (dict[str, Any]): The input data containing:
+                - inserted_slots (dict[str, Any]): The inserted slots
+                - deleted_slots (dict[str, Any]): The deleted slots
+                - metadata (dict): Metadata for logging purposes
+        """
+
+        metadata = input_data.get('metadata', {})
+
+        self.info('Reporting orchestration...', metadata=metadata)
+
+        inserted_slots = input_data['inserted_slots']
+        deleted_slots = input_data['deleted_slots']
+
+        if len(inserted_slots) > 0:
+            success_keys, error_keys = self.parse_report(inserted_slots)
+
+            if len(success_keys) > 0:
+                self.send_success_report(
+                    metadata=metadata,
+                    message=f'Inserted slots: \n {", ".join(success_keys)}',
+                    notification_id='REPORT_ORCHESTRATION_INSERTED_SLOTS',
+                )
+
+            if len(error_keys) > 0:
+                self.send_error_report(
+                    metadata=metadata,
+                    message=f'Failed to insert slots: \n {", ".join(error_keys)}',
+                    notification_id='REPORT_ORCHESTRATION_INSERTED_SLOTS',
+                    attachment=inserted_slots,
+                )
+
+        if len(deleted_slots) > 0:
+            success_keys, error_keys = self.parse_report(deleted_slots)
+
+            if len(success_keys) > 0:
+                self.send_success_report(
+                    metadata=metadata,
+                    message=f'Deleted slots: \n {", ".join(success_keys)}',
+                    notification_id='REPORT_ORCHESTRATION_DELETED_SLOTS',
+                )
+
+            if len(error_keys) > 0:
+                self.send_error_report(
+                    metadata=metadata,
+                    message=f'Failed to delete slots: \n {", ".join(error_keys)}',
+                    notification_id='REPORT_ORCHESTRATION_DELETED_SLOTS',
+                    attachment=deleted_slots,
+                )
+
+    @activity.defn(name='format_log_report')
+    def format_log_report(self, input_data: dict[str, Any]) -> dict[str, Any]:
+        """
+        Format the receiver_groups status to a dataframe to be stored in the database.
+
+        Transforms receiver group notification data into a structured format suitable
+        for database storage, aggregating notifications by unique notification ID and trigger.
+
+        Args:
+            input_data (dict[str, Any]): The input data containing:
+                - receiver_groups (dict): The receiver groups configuration with notifications
+                - mail_type (str): The type of mail for the report (Alerts/Reports)
+                - metadata (dict): Metadata for logging purposes
+
+        Returns:
+            dict[str, Any]: The formatted log report as a dictionary representation of a DataFrame
+        """
+        metadata = input_data['metadata']
+        mail_type = input_data['mail_type']
+
+        self.info('Formatting log report...', metadata=metadata)
+
+        receiver_groups = input_data['receiver_groups']
+
+        data = {}
+
+        for group_name, group_config in receiver_groups.items():
+            for notification in group_config['notifications']:
+                notification_id = notification['notification_id']
+                trigger = notification['trigger']
+
+                key = f'{notification_id}:{trigger}'
+
+                if key not in data:
+                    data[key] = {
+                        'status': group_config['status'],
+                        'timestamp': notification['timestamp'],
+                        'groups': [group_name],
+                        'message': notification['message'],
+                        'level': notification['level'],
+                        'notification_id': notification_id,
+                        'block': notification['block'],
+                        'schedule': trigger,
+                        'pipeline': notification['pipeline'],
+                        'project': notification['project'],
+                        'model_name': notification['model_name'],
+                        'model_id': notification['model_id'],
+                        'mail_type': mail_type,
+                    }
+                else:
+                    if group_name not in data[key]['groups']:
+                        data[key]['groups'].append(group_name)
+
+        data_values: DataFrame = DataFrame(list(data.values()))
+
+        # ``DataFrame.to_dict()`` is typed as ``dict[Hashable, Any]`` in pandas
+        # stubs, but default orientation uses column names (str keys).
+        return cast(dict[str, Any], data_values.to_dict())
+
+    @activity.defn(name='filter_notification_reports')
+    def filter_notification_reports(self, input_data: dict[str, Any]) -> dict[str, Any]:
+        """
+        Filter notifications for comprehensive scheduled reports.
+
+        This method filters notifications of all levels (ERROR, WARNING, INFO, DEBUG)
+        for scheduled report generation. Unlike alert filtering, this method does not
+        implement TTL-based duplicate prevention since reports are meant to provide
+        comprehensive coverage of system activity within a time window.
+
+        Filtering Logic:
+        - Processes all notification levels (not just ERROR)
+        - Applies group-specific content filtering for "reports" type
+        - Respects ignore lists for each receiver group
+        - Prevents duplicate notifications within the same report
+        - Groups notifications by receiver group configurations
+
+        Args:
+            input_data (dict[str, Any]): The input data containing:
+                - metadata (dict): Workflow execution metadata for logging
+                - notification_package (list): All notifications to filter (any level)
+                - sending_configs (list): Receiver group configurations with:
+                    - group_name (str): Name of the receiver group
+                    - contents (list): Content types to include (must contain "reports")
+                    - ignore (list, optional): Notification IDs to exclude
+
+        Returns:
+            dict[str, Any]: Filtered receiver groups with their notifications, keyed by group_name.
+                Each group contains:
+                - All receiver group configuration fields
+                - notifications (list): Filtered notifications for this group
+        """
+        metadata = input_data['metadata']
+        notification_package = input_data['notification_package']
+        sending_configs = input_data['sending_configs']
+
+        self.info('Filtering notification reports...', metadata=metadata)
+
+        receiver_groups = {}
+
+        for receiver_group in sending_configs:
+            group_name = receiver_group['group_name']
+            receiver_groups[group_name] = {**receiver_group, 'notifications': []}
+            receiver_groups[group_name]['notifications'] = []
+
+            already_added_keys = []
+
+            ignore_list = receiver_group.get('ignore', [])
+
+            for notification in notification_package:
+                alert_type = 'reports'
+                notification_id = notification['notification_id']
+
+                key = f'{notification["trigger"]}:{notification_id}'
+
+                # Check if this group must be notified
+                if (
+                    alert_type in receiver_group['contents']
+                    and notification_id not in ignore_list
+                    and key not in already_added_keys
+                ):
+                    receiver_groups[group_name]['notifications'].append(notification)
+                    already_added_keys.append(key)
+
+        # Remove groups with no notifications
+        receiver_groups = {
+            group_name: group
+            for group_name, group in receiver_groups.items()
+            if group['notifications']
+        }
+
+        return receiver_groups

orchestrator/activities/mongo_db.py

@@ -0,0 +1,511 @@
+from temporalio import activity, workflow
+
+with workflow.unsafe.imports_passed_through():
+    import traceback
+    from datetime import UTC
+    from logging import Logger
+    from typing import Any
+
+    from sientia_do.notifications.handlers import CoreNotificationHandler as NotificationHandler
+    from sientia_do.notifications.models import NotificationLevel
+    from sientia_do.observability.sientia_monitoring import MetricsController, SientiaMonitoring
+    from sientia_do.repository.mongodb_repository_sync import MongoDBRepository
+    from sientia_do.temporal.constants import (
+        DATETIME_FORMAT_MS_WITH_TZ,
+        now,
+    )
+
+
+class MongoDB(SientiaMonitoring):
+    """
+    MongoDB operations activity for Temporal workflows.
+
+    This class provides MongoDB database operations including document
+    querying, aggregation, timestamp management, and collection management
+    with TTL indexes. It centralizes all MongoDB interactions required by
+    the orchestration system lifecycle.
+
+    Args:
+        connection_string (str): MongoDB connection string
+        database_name (str): Target database name
+        ttl_index_seconds (int): TTL index duration in seconds
+        logger (Logger): Application logger instance
+        notification_handler (NotificationHandler): Notification management handler
+    """
+
+    def __init__(
+        self,
+        connection_string: str,
+        database_name: str,
+        ttl_index_seconds: int,
+        logger: Logger,
+        notification_handler: NotificationHandler,
+        metrics_controller: MetricsController,
+    ):
+        SientiaMonitoring.__init__(
+            self,
+            logger=logger,
+            notification_handler=notification_handler,
+            metrics_controller=metrics_controller,
+        )
+
+        self.connection_string = connection_string
+        self.database_name = database_name
+
+        self.mongo_db_repository = MongoDBRepository(
+            connection_string=connection_string,
+            database_name=database_name,
+            logger=logger,
+            notification_handler=notification_handler,
+            metrics_controller=metrics_controller,
+        )
+
+        self.ttl_index_seconds = ttl_index_seconds
+
+        # Initialize MongoDB client here (omitted for brevity)
+        logger.info('MongoDB connection initialized')
+
+    def close(self):
+        """
+        Shutdown the MongoDB client and clean up resources.
+
+        Closes the MongoDB repository connection and shuts down the SientiaMonitoring instance.
+        """
+        self.mongo_db_repository.close()
+        SientiaMonitoring.shutdown(self)
+
+    def __del__(self):
+        """
+        Ensure the MongoDB client is closed when the object is garbage-collected.
+        """
+        self.close()
+
+    @activity.defn(
+        name='find_documents_in_mongodb',
+    )
+    def find_documents_in_mongodb(self, input_data: dict[str, Any]) -> list[dict[str, Any]]:
+        """
+        Find documents in a MongoDB collection based on the provided query parameters.
+
+        Args:
+            - input_data (dict): Input data containing query parameters. Contains:
+                - query (dict): Query parameters to filter documents.
+                - timestamp_fields (list[str], optional): Fields to format as RFC3339 with TZ.
+
+        Returns:
+            list[dict]: Documents matching the query with timestamp fields normalized.
+        """
+
+        query = input_data.get('query', {})
+        metadata = input_data.get('metadata', {})
+        timestamp_fields = input_data.get('timestamp_fields', [])
+
+        collection_name = query.get('collection')
+        if not collection_name:
+            raise ValueError('Collection name must be provided in the query.')
+
+        filters = query.get('filters', {})
+
+        self.info(
+            f"Loading documents from collection '{collection_name}' with filters: {filters}",
+            metadata=metadata,
+        )
+
+        try:
+            documents = self.mongo_db_repository.find(collection_name, filters, metadata)
+
+            self.info(
+                f"Loaded {len(documents)} documents from collection '{collection_name}'",
+                metadata=metadata,
+            )
+
+            for document in documents:
+                for timestamp_field in timestamp_fields:
+                    if timestamp_field in document:
+                        document[timestamp_field] = (
+                            document[timestamp_field]
+                            .replace(tzinfo=UTC)
+                            .strftime(DATETIME_FORMAT_MS_WITH_TZ)
+                        )
+
+            self.debug(f'Documents loaded: {documents}', metadata=metadata)
+
+            return documents
+
+        except Exception as e:
+            trace = traceback.format_exc()
+            self.send_notification(
+                metadata=metadata,
+                notification_id='MONGODB_QUERY_ERROR',
+                message=f'Failed to execute MongoDB query: {e}',
+                block='load_query_from_mongodb',
+                level=NotificationLevel.ERROR,
+                attachment_content=trace,
+            )
+            self.error(trace, metadata=metadata)
+
+            raise e
+
+    @activity.defn(name='aggregate_documents_in_mongodb')
+    def aggregate_documents_in_mongodb(self, input_data: dict[str, Any]) -> list[dict[str, Any]]:
+        """
+        Aggregate documents in a MongoDB collection based on the provided aggregation pipeline.
+
+        Args:
+            - input_data (dict): Input data containing aggregation parameters. Contains:
+                - query (dict): Aggregation parameters including 'collection' and 'aggregation'.
+                - timestamp_fields (list[str], optional): Fields to format as RFC3339 with TZ.
+
+        Returns:
+            list[dict]: Aggregated documents with timestamp fields normalized.
+        """
+
+        query = input_data.get('query', {})
+        metadata = input_data.get('metadata', {})
+        timestamp_fields = input_data.get('timestamp_fields', [])
+
+        collection_name = query.get('collection')
+        if not collection_name:
+            raise ValueError('Collection name must be provided in the query.')
+        aggregation = query.get('aggregation')
+        if not aggregation:
+            raise ValueError('Aggregation must be provided.')
+        aggregation.append({'$project': {'_id': 0}})
+
+        self.info(
+            f"Aggregating documents from collection '{collection_name}' with aggregation: {aggregation}",
+            metadata=metadata,
+        )
+
+        try:
+            aggregated_documents = self.mongo_db_repository.aggregate(
+                collection_name, aggregation, metadata
+            )
+
+            self.info(
+                f"Aggregated {len(aggregated_documents)} documents from collection '{collection_name}'",
+                metadata=metadata,
+            )
+
+            for document in aggregated_documents:
+                for timestamp_field in timestamp_fields:
+                    if timestamp_field in document:
+                        document[timestamp_field] = (
+                            document[timestamp_field]
+                            .replace(tzinfo=UTC)
+                            .strftime(DATETIME_FORMAT_MS_WITH_TZ)
+                        )
+
+            self.debug(f'Aggregation result: {aggregated_documents}', metadata=metadata)
+
+            return aggregated_documents
+
+        except Exception as e:
+            trace = traceback.format_exc()
+            self.send_notification(
+                metadata=metadata,
+                notification_id='MONGODB_AGGREGATION_ERROR',
+                message=f'Failed to execute MongoDB aggregation: {e}',
+                block='aggregate_documents_in_mongodb',
+                level=NotificationLevel.ERROR,
+                attachment_content=trace,
+            )
+            self.error(trace, metadata=metadata)
+
+            raise e
+
+    @activity.defn(name='update_pipelines_timestamps')
+    def update_pipelines_timestamps(self, input_data: dict[str, Any]) -> None:
+        """
+        Update `updated_at` timestamps for successfully updated pipelines.
+
+        Updates the `updated_at` field in the `orchestrated_schedules` collection
+        for all pipelines that were successfully updated in Temporal.
+
+        Args:
+            input_data (dict[str, Any]): Input data containing:
+                - updated_pipelines (list[dict]): Pipelines with success flags to consider
+                - metadata (dict): Metadata for logging purposes
+        """
+        updated_pipelines = input_data.get('updated_pipelines', [])
+        metadata = input_data.get('metadata', {})
+        date_now = now()
+
+        self.info('Updating pipelines timestamps...', metadata=metadata)
+
+        argument = [
+            {'schedule_name': pipeline['schedule_name'], 'namespace': pipeline['namespace']}
+            for pipeline in updated_pipelines
+            if pipeline['success']
+        ]
+        data_filter = {'$or': argument} if argument else {}
+
+        try:
+            self.mongo_db_repository.update_many(
+                'orchestrated_schedules', data_filter, {'$set': {'updated_at': date_now}}, metadata
+            )
+        except Exception as e:
+            trace = traceback.format_exc()
+            self.send_notification(
+                metadata=metadata,
+                notification_id='MONGODB_UPDATE_PIPELINES_ERROR',
+                message=f'Failed to update pipelines timestamps: {e}',
+                block='update_pipelines_timestamps',
+                level=NotificationLevel.ERROR,
+                attachment_content=trace,
+            )
+            self.error(trace, metadata=metadata)
+            raise e
+
+        self.info(
+            f'Updated {len(updated_pipelines)} pipelines timestamps',
+            metadata=metadata,
+        )
+
+    @activity.defn(name='create_pipelines_timestamps')
+    def create_pipelines_timestamps(self, input_data: dict[str, Any]) -> None:
+        """
+        Insert `updated_at` timestamps for newly created pipelines.
+
+        Inserts new documents into the `orchestrated_schedules` collection
+        for all pipelines that were successfully created in Temporal.
+
+        Args:
+            input_data (dict[str, Any]): Input data containing:
+                - created_pipelines (list[dict]): Pipelines with success flags to consider
+                - metadata (dict): Metadata for logging purposes
+        """
+        created_pipelines = input_data.get('created_pipelines', [])
+        metadata = input_data.get('metadata', {})
+
+        self.info('Creating pipelines timestamps...', metadata=metadata)
+
+        date_now = now()
+
+        argument = [
+            {
+                'schedule_name': pipeline['schedule_name'],
+                'namespace': pipeline['namespace'],
+                'updated_at': date_now,
+            }
+            for pipeline in created_pipelines
+            if pipeline['success']
+        ]
+        data_filter = argument if argument else {}
+
+        try:
+            if data_filter:
+                self.mongo_db_repository.insert_many(
+                    'orchestrated_schedules', data_filter, metadata
+                )
+        except Exception as e:
+            trace = traceback.format_exc()
+            self.send_notification(
+                metadata=metadata,
+                notification_id='MONGODB_CREATE_PIPELINES_ERROR',
+                message=f'Failed to create pipelines timestamps: {e}',
+                block='create_pipelines_timestamps',
+                level=NotificationLevel.ERROR,
+                attachment_content=trace,
+            )
+            self.error(trace, metadata=metadata)
+            raise e
+
+        self.info(
+            f'Created {len(created_pipelines)} pipelines timestamps',
+            metadata=metadata,
+        )
+
+    @activity.defn(name='delete_pipelines_timestamps')
+    def delete_pipelines_timestamps(self, input_data: dict[str, Any]) -> None:
+        """
+        Delete timestamp rows for successfully deleted pipelines.
+
+        Removes documents from the `orchestrated_schedules` collection
+        for all pipelines that were successfully deleted from Temporal.
+
+        Args:
+            input_data (dict[str, Any]): Input data containing:
+                - deleted_pipelines (list[dict]): Pipelines with success flags to consider
+                - metadata (dict): Metadata for logging purposes
+        """
+        deleted_pipelines = input_data.get('deleted_pipelines', [])
+        metadata = input_data.get('metadata', {})
+
+        self.info('Deleting pipelines timestamps...', metadata=metadata)
+
+        argument = [
+            {'schedule_name': pipeline['schedule_name'], 'namespace': pipeline['namespace']}
+            for pipeline in deleted_pipelines
+            if pipeline['success']
+        ]
+        data_filter = {'$or': argument} if argument else {}
+
+        try:
+            self.mongo_db_repository.delete_many('orchestrated_schedules', data_filter, metadata)
+        except Exception as e:
+            trace = traceback.format_exc()
+            self.send_notification(
+                metadata=metadata,
+                notification_id='MONGODB_DELETE_PIPELINES_ERROR',
+                message=f'Failed to delete pipelines timestamps: {e}',
+                block='delete_pipelines_timestamps',
+                level=NotificationLevel.ERROR,
+                attachment_content=trace,
+            )
+            self.error(trace, metadata=metadata)
+            raise e
+
+        self.info(
+            f'Deleted {len(deleted_pipelines)} pipelines timestamps',
+            metadata=metadata,
+        )
+
+    @activity.defn(name='create_collection_with_ttl_index')
+    def create_collection_with_ttl_index(self, input_data: dict[str, Any]) -> None:
+        """
+        Create collections with TTL indexes for pipeline topics.
+
+        Creates MongoDB collections for scouter pipeline topics and sets up
+        TTL indexes on the `inserted_at` field to automatically expire old documents.
+
+        Args:
+            input_data (dict[str, Any]): Input data containing:
+                - pipelines (dict[str, Any]): Pipeline configurations with topic names
+                - metadata (dict): Metadata for logging purposes
+        """
+        pipelines = input_data.get('pipelines', {})
+        metadata = input_data.get('metadata', {})
+
+        self.info(
+            f'Creating collection with TTL index for pipelines: {list(pipelines.keys())}',
+            metadata=metadata,
+        )
+
+        collection_names = self.mongo_db_repository.database.list_collection_names()
+
+        created_collections = []
+        created_indexes = []
+
+        for _pipeline_name, pipeline_config in pipelines.items():
+            collection = pipeline_config.get('topic', None)
+            if not collection:
+                continue
+
+            try:
+                # Check if collection exists
+                if collection not in collection_names:
+                    self.mongo_db_repository.database.create_collection(collection)
+                    created_collections.append(collection)
+
+                collection = self.mongo_db_repository.database[collection]
+                # Check if TTL index exists
+                existing_indexes = collection.list_indexes()
+                ttl_index_exists = False
+                for index in existing_indexes:
+                    if (
+                        'inserted_at' in index['key']
+                        and index.get('expireAfterSeconds') is not None
+                    ):
+                        ttl_index_exists = True
+                        break
+
+                # Create TTL index if it doesn't exist
+                if not ttl_index_exists:
+                    collection.create_index(
+                        'inserted_at', expireAfterSeconds=self.ttl_index_seconds, background=True
+                    )
+                    created_indexes.append(collection)
+
+            except Exception as e:
+                trace = traceback.format_exc()
+                self.send_notification(
+                    metadata=metadata,
+                    notification_id='MONGODB_CREATE_COLLECTION_ERROR',
+                    message=f'Failed to create collection {collection} with TTL index: {e}',
+                    block='create_collection_with_ttl_index',
+                    level=NotificationLevel.ERROR,
+                    attachment_content=trace,
+                )
+                self.error(trace, metadata=metadata)
+                raise e
+
+        self.info(
+            f'Created {len(created_collections)} collections and {len(created_indexes)} indexes',
+            metadata=metadata,
+        )
+
+        self.debug(f'Created collections: {created_collections}', metadata=metadata)
+        self.debug(f'Created indexes: {created_indexes}', metadata=metadata)
+
+    @activity.defn(name='load_latest_data')
+    def load_latest_data(self, input_data: dict[str, Any]) -> list[dict[str, Any]]:
+        """
+        Load the latest data from MongoDB collection since a specified timestamp.
+
+        This activity retrieves data from a MongoDB collection, optionally
+        filtering by timestamp to enable incremental data processing. It
+        handles connection management and provides comprehensive error reporting.
+
+        Args:
+            input_data (dict[str, Any]): Activity input parameters.
+                Required fields:
+                - metadata (dict[str, Any]): Workflow execution metadata
+                - collection_name (str): Name of the MongoDB collection
+                - last_data_timestamp (str | None): Last processed timestamp for filtering
+                - base_data_filter (dict[str, Any]): Base query filter conditions
+
+        Returns:
+            list[dict[str, Any]]: Retrieved data, or empty list if no data found
+
+        Raises:
+            Exception: If MongoDB operation fails
+        """
+        metadata = input_data['metadata']
+        collection_name = input_data['collection_name']
+        last_data_timestamp = input_data['last_data_timestamp']
+        base_data_filter = input_data['base_data_filter']
+
+        self.debug(f'Loading data from MongoDB: {input_data}', metadata=metadata)
+
+        try:
+            if last_data_timestamp is None:
+                data_filter = base_data_filter
+            else:
+                # ``notification_queue.timestamp`` is stored as a string in
+                # ``DATETIME_FORMAT_WITH_TZ`` (``Notification`` writes it as
+                # ``now().strftime(DATETIME_FORMAT_WITH_TZ)``). Coercing
+                # ``last_data_timestamp`` to ``datetime`` here would force a
+                # BSON ``String`` vs ``Date`` comparison, which always yields
+                # ``False`` (``String < Date`` in BSON sort order) and breaks
+                # incremental loading entirely. Comparing strings preserves the
+                # intended chronological filter because the format is
+                # lexicographically ordered when the timezone is fixed
+                # (``Notification.timestamp`` always uses UTC).
+                data_filter = {
+                    **base_data_filter,
+                    'timestamp': {'$gt': last_data_timestamp},
+                }
+
+            self.debug(f'Data filter: {data_filter}', metadata=metadata)
+
+            data = self.mongo_db_repository.find(collection_name, data_filter, metadata)
+
+            self.debug(f'Collected: {data}', metadata=metadata)
+
+            self.info(f'Loaded {len(data)} documents from MongoDB', metadata=metadata)
+
+            self.debug(f'Loaded data: {data}', metadata=metadata)
+
+            return data
+        except Exception as e:
+            trace = traceback.format_exc()
+            self.send_notification(
+                metadata=metadata,
+                notification_id='MONGO_LOAD_ERROR',
+                message=f'Error loading data from MongoDB: {e}',
+                block='load_latest_data',
+                level=NotificationLevel.ERROR,
+                attachment_content=trace,
+            )
+            raise e

orchestrator/activities/slot_manager.py

@@ -0,0 +1,468 @@
+from temporalio import activity, workflow
+
+with workflow.unsafe.imports_passed_through():
+    import json
+    import traceback
+    from datetime import datetime, timedelta
+    from logging import Logger
+    from typing import Any
+
+    from pandas import DataFrame
+    from sientia_do.notifications.handlers import CoreNotificationHandler as NotificationHandler
+    from sientia_do.notifications.models import NotificationLevel
+    from sientia_do.observability.sientia_monitoring import MetricsController, SientiaMonitoring
+    from sientia_do.repository.redis_repository_sync import RedisRepository
+    from sientia_do.temporal.constants import DATETIME_FORMAT_MS_WITH_TZ, now
+
+
+class SlotManager(SientiaMonitoring):
+    """
+    Redis-based OPC slot management and notification filtering activity.
+
+    This class manages OPC server slots and provides advanced notification
+    filtering capabilities through Redis operations. It handles OPC slot
+    lifecycle management, active ingestor tracking, and implements intelligent
+    notification filtering with TTL-based duplicate prevention.
+
+    Key Features:
+    - OPC slot loading, updating, and deletion
+    - Active ingestor management
+    - Notification filtering for alerts with TTL management
+    - Persistent alert detection for ongoing issues
+    - Notification caching with configurable expiration
+    - Group-based filtering with ignore list support
+
+    Args:
+        host (str): Redis server hostname
+        port (int): Redis server port number
+        username (str): Redis authentication username
+        password (str): Redis authentication password
+        logger (Logger): Application logger instance
+        notification_handler (NotificationHandler): Notification management handler
+    """
+
+    def __init__(
+        self,
+        host: str,
+        port: int,
+        username: str,
+        password: str,
+        logger: Logger,
+        notification_handler: NotificationHandler,
+        metrics_controller: MetricsController,
+    ):
+        SientiaMonitoring.__init__(
+            self,
+            logger=logger,
+            notification_handler=notification_handler,
+            metrics_controller=metrics_controller,
+        )
+
+        self.redis_repository = RedisRepository(
+            host=host,
+            port=port,
+            username=username,
+            password=password,
+            logger=logger,
+            notification_handler=notification_handler,
+            metrics_controller=metrics_controller,
+        )
+
+    def close(self):
+        """
+        Close the SlotManager connection and clean up resources.
+
+        Closes the Redis repository connection and shuts down the SientiaMonitoring instance.
+        """
+        self.redis_repository.close()
+        SientiaMonitoring.shutdown(self)
+
+    def __del__(self):
+        """
+        Ensure the SlotManager connection is closed when the object is garbage-collected.
+        """
+        self.close()
+
+    @activity.defn(name='load_opc_slots')
+    def load_opc_slots(self, input_data: dict[str, Any]) -> dict[str, Any]:
+        """
+        Load all OPC slots from Redis for current system state assessment.
+
+        This method retrieves all OPC server slot configurations from Redis,
+        which are used to determine current resource allocation and identify
+        changes needed for pipeline orchestration.
+
+        Args:
+            input_data (dict[str, Any]): Activity input containing metadata
+
+        Returns:
+            dict[str, Any]: Dictionary of OPC slots keyed by server ID, where each slot contains:
+                - Configuration parameters for OPC server connections
+                - Active pipeline assignments
+                - Resource allocation details
+
+        Raises:
+            Exception: If Redis connection fails or data retrieval errors occur
+        """
+
+        metadata = input_data.get('metadata', {})
+
+        self.info('Loading OPC slots...', metadata=metadata)
+
+        opc_slots = {}
+
+        try:
+            slot_keys = self.redis_repository.keys('slot:opc_tags:*')
+
+            self.debug(f'Slot keys: {slot_keys}', metadata=metadata)
+
+            if slot_keys:
+                if isinstance(slot_keys[0], bytes):
+                    decoded_keys = [key.decode('utf-8') for key in slot_keys]
+                else:
+                    decoded_keys = slot_keys
+
+                for key in decoded_keys:
+                    opc_slots[key] = self.redis_repository.get(key)
+        except Exception as e:
+            trace = traceback.format_exc()
+            self.send_notification(
+                metadata=metadata,
+                notification_id='REDIS_GET_ERROR',
+                message=f'Failed to load OPC slots: {e}',
+                block='load_opc_slots',
+                level=NotificationLevel.ERROR,
+                attachment_content=trace,
+            )
+            self.error(trace, metadata=metadata)
+            raise e
+
+        self.info(f'Loaded {len(opc_slots)} OPC slots', metadata=metadata)
+
+        return opc_slots
+
+    @activity.defn(name='load_active_ingestors')
+    def load_active_ingestors(self, input_data: dict[str, Any]) -> list[str]:
+        """
+        Load all active ingestors from Redis.
+
+        Retrieves all active ingestor heartbeat keys from Redis to determine
+        which ingestors are currently available for slot assignment.
+
+        Args:
+            input_data (dict[str, Any]): Activity input containing metadata
+
+        Returns:
+            list[str]: A list of active ingestor keys from Redis
+        """
+
+        metadata = input_data.get('metadata', {})
+
+        self.info('Loading active ingestors...', metadata=metadata)
+
+        try:
+            active_ingestors = self.redis_repository.keys('heartbeat:ingestor:*')
+
+            self.info(f'Loaded {len(active_ingestors)} active ingestors', metadata=metadata)
+
+            self.debug(f'Active ingestors: \n {active_ingestors}', metadata=metadata)
+
+            ingestors = []
+
+            for ingestor in active_ingestors:
+                if isinstance(ingestor, bytes):
+                    ingestors.append(ingestor.decode('utf-8'))
+                else:
+                    ingestors.append(ingestor)
+
+            return ingestors
+        except Exception as e:
+            trace = traceback.format_exc()
+            self.send_notification(
+                metadata=metadata,
+                notification_id='REDIS_GET_ERROR',
+                message=f'Failed to load active ingestors: {e}',
+                block='load_active_ingestors',
+                level=NotificationLevel.ERROR,
+                attachment_content=trace,
+            )
+            self.error(trace, metadata=metadata)
+            raise e
+
+    @activity.defn(name='update_slots')
+    def update_slots(self, input_data: dict[str, Any]) -> dict[str, Any]:
+        """
+        Update OPC slots in Redis
+
+        Args:
+            - input_data (dict[str, Any]): The input data containing
+              the slots to update.
+                - to_insert (dict[str, Any]): The slots to insert.
+
+        Returns:
+            - report (dict[str, Any]): A report of the updated slots.
+        """
+
+        to_insert = input_data['to_insert']
+        metadata = input_data.get('metadata', {})
+        self.info('Updating OPC slots...', metadata=metadata)
+
+        report = {}
+
+        success_count = 0
+
+        for slot in to_insert:
+            try:
+                self.redis_repository.set(f'slot:opc_tags:{slot}', to_insert[slot], ttl=None)
+                report[slot] = {'success': True, 'message': 'Slot updated successfully'}
+                success_count += 1
+            except Exception as e:
+                self.error(f'Failed to update slot {slot}: {str(e)}', metadata=metadata)
+                report[slot] = {'success': False, 'message': str(e)}
+
+        self.info(f'Updated {success_count} of {len(to_insert)} OPC slots', metadata=metadata)
+
+        self.debug(f'Report: \n {json.dumps(report, indent=4, sort_keys=True)}', metadata=metadata)
+
+        return report
+
+    @activity.defn(name='delete_slots')
+    def delete_slots(self, input_data: dict[str, Any]) -> dict[str, Any]:
+        """
+        Delete OPC slots from Redis
+
+        Args:
+            - input_data (dict[str, Any]): The input data containing
+              the slots to delete.
+                - to_delete (list[str]): The slots to delete.
+
+        Returns:
+            - report (dict[str, Any]): A report of the deleted slots.
+        """
+
+        to_delete = input_data['to_delete']
+        metadata = input_data.get('metadata', {})
+
+        self.info('Deleting OPC slots...', metadata=metadata)
+
+        report = {}
+
+        success_count = 0
+
+        for slot in to_delete:
+            try:
+                self.redis_repository.delete(f'slot:opc_tags:{slot}')
+                report[slot] = {'success': True, 'message': 'Slot deleted successfully'}
+                success_count += 1
+            except Exception as e:
+                self.error(f'Failed to delete slot {slot}: {str(e)}', metadata=metadata)
+                report[slot] = {'success': False, 'message': str(e)}
+
+        self.info(f'Deleted {success_count} of {len(to_delete)} OPC slots', metadata=metadata)
+
+        self.debug(f'Report: \n {json.dumps(report, indent=4, sort_keys=True)}', metadata=metadata)
+
+        return report
+
+    @activity.defn(name='get_last_data_timestamp')
+    def get_last_data_timestamp(self, input_data: dict[str, Any]) -> str | None:
+        """
+        Get the last data timestamp from Redis.
+
+        Retrieves the last processed timestamp for a specific mail type from Redis.
+        This timestamp is used for incremental data loading to avoid reprocessing
+        already processed notifications.
+
+        Args:
+            input_data (dict[str, Any]): The input data containing:
+                - metadata (dict): Metadata for logging purposes
+                - mail_type (str): The type of mail to get timestamp for (Alerts/Reports)
+
+        Returns:
+            str | None: The last data timestamp as a string, or None if no timestamp exists
+        """
+        metadata = input_data['metadata']
+        key = f'notification_last_timestamp:{input_data["mail_type"]}'
+
+        try:
+            data_hold = self.redis_repository.get(key)
+        except Exception as e:
+            self.send_notification(
+                metadata=metadata,
+                notification_id='REDIS_GET_ERROR',
+                message=f'Error getting last data timestamp: {e}',
+                block='get_last_data_timestamp',
+                level=NotificationLevel.ERROR,
+                attachment_content=traceback.format_exc(),
+            )
+            raise e
+
+        self.debug(f'Last collected timestamp: {data_hold}', metadata=metadata)
+
+        if not data_hold:
+            return None
+
+        return data_hold
+
+    @activity.defn(name='put_last_data_timestamp')
+    def put_last_data_timestamp(self, input_data: dict[str, Any]) -> str | None:
+        """
+        Store the last data timestamp in Redis.
+
+        Extracts the maximum timestamp from the provided data and stores it in Redis
+        with a TTL for the specified mail type. This enables incremental processing
+        in subsequent workflow executions.
+
+        Args:
+            input_data (dict[str, Any]): The input data containing:
+                - metadata (dict): Metadata for logging purposes
+                - data (list[dict]): The data to extract timestamp from
+                - mail_type (str): The type of mail to store timestamp for (Alerts/Reports)
+
+        Returns:
+            str | None: The last data timestamp that was stored, or None if no data exists
+        """
+        metadata = input_data['metadata']
+        key = f'notification_last_timestamp:{input_data["mail_type"]}'
+
+        data = DataFrame(input_data['data'])
+
+        if data.empty:
+            self.warning('No data to insert', metadata=metadata)
+            return None
+
+        last_data_timestamp = data['timestamp'].max()
+
+        self.debug(f'Last collected timestamp to insert: {last_data_timestamp}', metadata=metadata)
+
+        try:
+            self.redis_repository.set(key, last_data_timestamp, ttl=60 * 60 * 5)
+        except Exception as e:
+            self.send_notification(
+                metadata=metadata,
+                notification_id='REDIS_SET_ERROR',
+                message=f'Error setting last data timestamp: {e}',
+                block='put_last_data_timestamp',
+                level=NotificationLevel.ERROR,
+                attachment_content=traceback.format_exc(),
+            )
+            raise e
+
+        return last_data_timestamp
+
+    @activity.defn(name='filter_notification_alerts')
+    def filter_notification_alerts(self, input_data: dict[str, Any]) -> dict[str, Any]:
+        """
+        Filter notification alerts with intelligent TTL-based duplicate prevention.
+
+        This method implements advanced notification filtering for ERROR-level alerts,
+        preventing spam through TTL management and detecting persistent issues that
+        require escalation. It applies user group-based filtering with configurable
+        ignore lists and content policies.
+
+        Filtering Logic:
+        - Checks Redis cache for recently sent notifications
+        - Identifies "core_alerts" for new notifications
+        - Detects "persistent_alerts" for ongoing issues beyond TTL
+        - Applies group-specific content filtering and ignore lists
+        - Prevents duplicate notifications within the same TTL window
+
+        Args:
+            input_data (dict[str, Any]): The input data containing:
+                - metadata (dict): Workflow execution metadata for logging
+                - notification_package (list): ERROR-level notifications to filter
+                - sending_configs (list): Receiver group configurations with:
+                    - group_name (str): Name of the receiver group
+                    - contents (list): Alert types to include (core_alerts, persistent_alerts)
+                    - ignore (list, optional): Notification IDs to exclude
+                - notification_ttl (int): Seconds before considering notification persistent
+
+        Returns:
+            dict[str, Any]: Filtered receiver groups with their notifications, keyed by group_name.
+                Each group contains:
+                - All receiver group configuration fields
+                - notifications (list): Filtered notifications for this group
+        """
+        metadata = input_data['metadata']
+        notification_package = input_data['notification_package']
+        sending_configs = input_data['sending_configs']
+        notification_ttl = input_data['notification_ttl']
+
+        self.info('Filtering notification alerts...', metadata=metadata)
+
+        receiver_groups = {}
+
+        for receiver_group in sending_configs:
+            group_name = receiver_group['group_name']
+            receiver_groups[group_name] = {**receiver_group, 'notifications': []}
+            receiver_groups[group_name]['notifications'] = []
+
+            already_added_keys = []
+
+            ignore_list = receiver_group.get('ignore', [])
+
+            for notification in notification_package:
+                alert_type = 'do_nothing'
+                notification_id = notification['notification_id']
+                # Check if notification was recently sent
+                key = f'{notification["trigger"]}:{notification_id}'
+
+                last_sent = self.redis_repository.get(key)
+
+                if last_sent is None:
+                    alert_type = 'core_alerts'
+
+                else:
+                    last_sent = datetime.strptime(last_sent, DATETIME_FORMAT_MS_WITH_TZ)
+
+                    # Check if "notification_ttl" seconds has passed since last sent
+                    if (now() - last_sent) > timedelta(seconds=notification_ttl):
+                        alert_type = 'persistent_alerts'
+
+                # Check if this group must be notified
+                if (
+                    alert_type in receiver_group['contents']
+                    and notification_id not in ignore_list
+                    and key not in already_added_keys
+                ):
+                    receiver_groups[group_name]['notifications'].append(notification)
+                    already_added_keys.append(key)
+
+        # Remove groups with no notifications
+        receiver_groups = {
+            group_name: group
+            for group_name, group in receiver_groups.items()
+            if group['notifications']
+        }
+
+        return receiver_groups
+
+    @activity.defn(name='store_notification_cache')
+    def store_notification_cache(self, input_data: dict[str, Any]) -> None:
+        """
+        Store notification cache in Redis to track recently sent notifications.
+
+        Stores successfully sent notifications in Redis with a TTL to prevent
+        duplicate alert delivery. Only notifications with status 'sent' are cached.
+
+        Args:
+            input_data (dict[str, Any]): The input data containing:
+                - metadata (dict): Metadata for logging purposes
+                - log_report (list[dict]): The log report containing notification statuses
+                - sent_ttl (int): Time to live for sent notification cache in seconds
+        """
+        metadata = input_data['metadata']
+        log_report = DataFrame(input_data['log_report'])
+        sent_ttl = input_data['sent_ttl']
+
+        self.info('Storing notification cache...', metadata=metadata)
+
+        date_now = now().strftime(DATETIME_FORMAT_MS_WITH_TZ)
+
+        for _index, row in log_report.iterrows():
+            status = row['status']
+            if status == 'sent':
+                key = f'{row["schedule"]}:{row["notification_id"]}'
+                self.redis_repository.set(key, date_now, ttl=sent_ttl)
+
+        self.info('Notification cache stored...', metadata=metadata)

orchestrator/activities/temporal_manager.py

@@ -0,0 +1,458 @@
+from temporalio import activity, workflow
+from temporalio.client import (
+    Client,
+    Schedule,
+    ScheduleActionStartWorkflow,
+    ScheduleIntervalSpec,
+    ScheduleSpec,
+    ScheduleUpdate,
+    ScheduleUpdateInput,
+)
+from temporalio.common import SearchAttributeKey, SearchAttributePair, TypedSearchAttributes
+
+with workflow.unsafe.imports_passed_through():
+    import json
+    import traceback
+    from datetime import timedelta
+    from logging import Logger
+    from typing import Any
+
+    from sientia_do.notifications.handlers import CoreNotificationHandler as NotificationHandler
+    from sientia_do.notifications.models import NotificationLevel
+    from sientia_do.observability.sientia_monitoring import MetricsController, SientiaMonitoring
+    from sientia_do.temporal.worker.prepare_worker import build_queue_name
+
+    from orchestrator.utils.converters import parse_frequency
+
+
+RUNTIME_WORKFLOWS = ['predictions_batch', 'minimal_retrain', 'drift', 'simple_metrics']
+
+class TemporalManager(SientiaMonitoring):
+    """
+    Temporal workflow and schedule management activity.
+
+    This class manages Temporal schedules across multiple namespaces,
+    providing operations for schedule creation, updates, deletion, and
+    normalization. It handles connections to both scouter and laborious
+    namespaces for comprehensive workflow orchestration.
+
+    Args:
+        host (str): Temporal server host address
+        scouter_namespace (str): Scouter workflow namespace
+        laborious_namespace (str): Laborious workflow namespace
+        logger (Logger): Application logger instance
+        notification_handler (NotificationHandler): Notification management handler
+    """
+
+    def __init__(
+        self,
+        host: str,
+        scouter_namespace: str,
+        laborious_namespace: str,
+        logger: Logger,
+        notification_handler: NotificationHandler,
+        metrics_controller: MetricsController,
+    ):
+        SientiaMonitoring.__init__(
+            self,
+            logger=logger,
+            notification_handler=notification_handler,
+            metrics_controller=metrics_controller,
+        )
+        self.temporal_host = host
+        self.scouter_namespace = scouter_namespace
+        self.laborious_namespace = laborious_namespace
+        self.temporal_clients: dict[str, Client] = {}
+
+        self.model_id_id_key = SearchAttributeKey.for_keyword('model_id')
+        self.model_name_id_key = SearchAttributeKey.for_keyword('model_name')
+        self.orchestrated_id_key = SearchAttributeKey.for_keyword('orchestrated')
+
+    def close(self):
+        """
+        Close the TemporalManager connection and clean up resources.
+
+        Shuts down the SientiaMonitoring instance and releases all resources.
+        """
+        SientiaMonitoring.shutdown(self)
+
+    def __del__(self):
+        """
+        Ensure the TemporalManager connection is closed when the object is garbage-collected.
+        """
+        self.close()
+
+    async def connect_to_temporal(self):
+        """
+        Connect to Temporal server namespaces used by scouter and laborious workflows.
+
+        Creates and caches `Client` connections for both namespaces for later use.
+        The connections are stored in `temporal_clients` dictionary for efficient access.
+        """
+        self.logger.info(f'Connecting to Temporal side namespaces at {self.temporal_host}')
+        self.logger.info(f'Scouter namespace: {self.scouter_namespace}')
+
+        scouter_client = await Client.connect(
+            target_host=self.temporal_host, namespace=self.scouter_namespace
+        )
+
+        self.logger.info(f'Laborious namespace: {self.laborious_namespace}')
+
+        laborious_client = await Client.connect(
+            target_host=self.temporal_host, namespace=self.laborious_namespace
+        )
+
+        self.temporal_clients = {
+            self.scouter_namespace: scouter_client,
+            self.laborious_namespace: laborious_client,
+        }
+
+    @activity.defn(name='normalize_schedules')
+    async def normalize_schedules(self, input_data: dict[str, Any]):
+        """
+        Normalize schedules by removing orphaned schedules from Temporal.
+
+        Any schedule marked with search attribute `orchestrated=true` that does not
+        exist in MongoDB collection `orchestrated_schedules` will be deleted.
+
+        Args:
+            input_data (dict[str, Any]): Input data containing:
+                - orchestrated_schedules (dict[str, Any]): Current orchestrated schedules from MongoDB.
+        """
+        metadata = input_data['metadata']
+
+        remove_count = 0
+
+        self.info('Getting orchestrated schedules...', metadata=metadata)
+
+        orchestrated_schedules = input_data.get('orchestrated_schedules', {})
+
+        for namespace, client in self.temporal_clients.items():
+            try:
+                schedules = orchestrated_schedules.get(namespace, {})
+
+                self.info(f'Getting orchestrated schedules for {namespace}', metadata=metadata)
+
+                async for schedule in await client.list_schedules():
+                    search_attrs = getattr(schedule, 'search_attributes', {})
+                    if search_attrs.get('orchestrated', ['false']) == ['true']:
+                        schedule_id = schedule.id
+
+                        if schedule_id not in schedules:
+                            self.info(
+                                f'Schedule {schedule_id} not found in mongo db, cleaning up',
+                                metadata=metadata,
+                            )
+
+                            handle = client.get_schedule_handle(schedule_id)
+
+                            await handle.delete()
+
+                            remove_count += 1
+
+            except Exception as e:
+                trace = traceback.format_exc()
+                await self.send_notification_async(
+                    metadata=metadata,
+                    notification_id='SCHEDULER_NORMALIZE_SCHEDULES_ERROR',
+                    message=f'Failed to normalize schedules: {e}',
+                    block='normalize_schedules',
+                    level=NotificationLevel.ERROR,
+                    attachment_content=trace,
+                )
+                self.error(trace, metadata=metadata)
+                raise e
+
+        self.info(f'Removed {remove_count} schedules', metadata=metadata)
+
+    @activity.defn(name='create_schedules')
+    async def create_schedules(self, input_data: dict[str, Any]) -> list[dict[str, Any]]:
+        """
+        Create schedules in Temporal.
+
+        Args:
+            - input_data (dict[str, Any]): The input data containing
+              the schedules to create.
+                - schedules (dict[str, Any]): The schedules to create.
+
+        Returns:
+            - list[dict[str, Any]]: Report entries for each attempted schedule creation.
+        """
+
+        schedules_to_create = input_data['schedules']
+        metadata = input_data.get('metadata', {})
+
+        report = []
+
+        success_count = 0
+
+        self.info('Creating schedules...', metadata=metadata)
+
+        for namespace, schedules in schedules_to_create.items():
+            client = self.temporal_clients.get(namespace)
+
+            if not client:
+                raise ValueError(
+                    f'Temporal client for {namespace} not found, clients: {self.temporal_clients}'
+                )
+
+            for schedule_name, schedule in schedules.items():
+                search_attributes = TypedSearchAttributes(
+                    [
+                        SearchAttributePair(key=self.model_id_id_key, value=schedule['model_id']),
+                        SearchAttributePair(
+                            key=self.model_name_id_key, value=schedule['model_name']
+                        ),
+                        SearchAttributePair(key=self.orchestrated_id_key, value='true'),
+                    ]
+                )
+                workflow_type = schedule['workflow_type']
+
+                try:
+                    execution_timeout_seconds = schedule.get('execution_timeout_seconds', 300)
+                    task_timeout_seconds = schedule.get('task_timeout_seconds', 300)
+                    self.debug(f'Creating schedule {schedule_name}:', metadata=metadata)
+                    self.debug(
+                        f'{json.dumps(schedule, indent=4, sort_keys=True)}', metadata=metadata
+                    )
+                    
+                    runtime_name = schedule.get('runtime', 'legacy') if workflow_type in RUNTIME_WORKFLOWS else None
+                    task_queue_name = build_queue_name(workflow_type, runtime_name)
+                    schedule['task_queue'] = task_queue_name
+
+                    await client.create_schedule(
+                        schedule_name,
+                        Schedule(
+                            action=ScheduleActionStartWorkflow(
+                                workflow_type,
+                                schedule,
+                                id=schedule_name,
+                                task_queue=task_queue_name,
+                                execution_timeout=timedelta(seconds=execution_timeout_seconds),
+                                run_timeout=timedelta(seconds=execution_timeout_seconds),
+                                task_timeout=timedelta(seconds=task_timeout_seconds),
+                                typed_search_attributes=search_attributes,
+                            ),
+                            spec=ScheduleSpec(
+                                intervals=[
+                                    ScheduleIntervalSpec(
+                                        every=timedelta(
+                                            seconds=parse_frequency(schedule.get('frequency', '1m'))
+                                        ),
+                                        offset=timedelta(
+                                            seconds=parse_frequency(schedule.get('offset', '0m'))
+                                        ),
+                                    )
+                                ]
+                            ),
+                        ),
+                        search_attributes=search_attributes,
+                    )
+
+                    report.append(
+                        {
+                            'namespace': namespace,
+                            'schedule_name': schedule_name,
+                            'success': True,
+                            'message': 'Schedule created successfully',
+                        }
+                    )
+
+                    success_count += 1
+                except Exception as e:
+                    self.error(
+                        f'Failed to create schedule {schedule_name}: {str(e)}', metadata=metadata
+                    )
+                    report.append(
+                        {
+                            'namespace': namespace,
+                            'schedule_name': schedule_name,
+                            'success': False,
+                            'message': str(e),
+                        }
+                    )
+
+        self.info(
+            f'Created {success_count} of {len(schedules_to_create)} schedules', metadata=metadata
+        )
+
+        self.debug(f'\n {json.dumps(report, indent=4, sort_keys=True)}', metadata=metadata)
+
+        return report
+
+    @activity.defn(name='update_schedules')
+    async def update_schedules(self, input_data: dict[str, Any]) -> list[dict[str, Any]]:
+        """
+        Update schedules in Temporal.
+
+        Does not modify ``task_queue``. The ``ScheduleUpdate`` callback only patches
+        workflow ``args`` and schedule ``intervals``. A ``runtime`` change requires
+        delete-then-create on the next orchestrator tick.
+
+        Args:
+            - input_data (dict[str, Any]): The input data containing
+              the schedules to update.
+                - schedules (dict[str, Any]): The schedules to update.
+
+        Returns:
+            - list[dict[str, Any]]: Report entries for each attempted schedule update.
+        """
+
+        schedules_to_update = input_data['schedules']
+        metadata = input_data.get('metadata', {})
+        report = []
+
+        success_count = 0
+
+        self.info('Updating schedules...', metadata=metadata)
+
+        for namespace, schedules in schedules_to_update.items():
+            client = self.temporal_clients.get(namespace)
+
+            if not client:
+                raise ValueError(
+                    f'Temporal client for {namespace} not found, clients: {self.temporal_clients}'
+                )
+
+            for schedule_name, schedule in schedules.items():
+                try:
+                    handler = client.get_schedule_handle(schedule_name)
+
+                    if not handler:
+                        raise ValueError(f'Schedule {schedule_name} not found')
+
+                    workflow_type = schedule['workflow_type']
+                    runtime_name = schedule.get('runtime', 'legacy') if workflow_type in RUNTIME_WORKFLOWS else None
+                    schedule['task_queue'] = build_queue_name(workflow_type, runtime_name)
+
+                    # fmt: off
+                    async def update_schedule(input_data: ScheduleUpdateInput) -> ScheduleUpdate: # NOSONAR
+                        schedule_action = input_data.description.schedule.action
+
+                        self.debug("Updating schedule:", metadata=metadata)
+
+                        if hasattr(schedule_action, "args"):
+                            self.debug("New schedule:", metadata=metadata)
+                            self.debug(
+                                f"{json.dumps(schedule, indent=4, sort_keys=True)}", metadata=metadata)  # NOSONAR
+
+                            schedule_action.args = [schedule]
+
+                        input_data.description.schedule.spec.intervals = [
+                            ScheduleIntervalSpec(
+                                every=timedelta(
+                                    seconds=parse_frequency(schedule.get('frequency', '1m'))),
+                                offset=timedelta(
+                                    seconds=parse_frequency(schedule.get('offset', '0m'))),
+                            )
+                        ]
+
+                        return ScheduleUpdate(schedule=input_data.description.schedule)
+
+                    # fmt: on
+                    await handler.update(update_schedule)
+
+                    del update_schedule
+
+                    report.append(
+                        {
+                            'namespace': namespace,
+                            'schedule_name': schedule_name,
+                            'success': True,
+                            'message': 'Schedule updated successfully',
+                        }
+                    )
+
+                    success_count += 1
+                except Exception as e:
+                    self.error(
+                        f'Failed to update schedule {schedule_name}: {str(e)}', metadata=metadata
+                    )
+                    report.append(
+                        {
+                            'namespace': namespace,
+                            'schedule_name': schedule_name,
+                            'success': False,
+                            'message': str(e),
+                        }
+                    )
+
+        self.info(
+            f'Updated {success_count} of {len(schedules_to_update)} schedules', metadata=metadata
+        )
+
+        self.debug(f'\n {json.dumps(report, indent=4, sort_keys=True)}', metadata=metadata)
+
+        return report
+
+    @activity.defn(name='delete_schedules')
+    async def delete_schedules(self, input_data: dict[str, Any]) -> list[dict[str, Any]]:
+        """
+        Delete schedules in Temporal.
+
+        Args:
+            - input_data (dict[str, Any]): The input data containing
+              the schedules to delete.
+                - schedules (list[str]): The schedules to delete.
+
+        Returns:
+            - list[dict[str, Any]]: Report entries for each attempted schedule deletion.
+        """
+
+        schedules_to_delete = input_data['schedules']
+        metadata = input_data.get('metadata', {})
+        report = []
+
+        success_count = 0
+
+        self.info('Deleting schedules...', metadata=metadata)
+
+        for namespace, schedules in schedules_to_delete.items():
+            client = self.temporal_clients.get(namespace)
+
+            if not client:
+                raise ValueError(
+                    f'Temporal client for {namespace} not found, clients: {self.temporal_clients}'
+                )
+
+            for schedule_name in schedules:
+                try:
+                    handler = client.get_schedule_handle(schedule_name)
+
+                    if not handler:
+                        raise ValueError(f'Schedule {schedule_name} not found')
+
+                    await handler.delete()
+
+                    report.append(
+                        {
+                            'namespace': namespace,
+                            'schedule_name': schedule_name,
+                            'success': True,
+                            'message': 'Schedule deleted successfully',
+                        }
+                    )
+
+                    success_count += 1
+                except Exception as e:
+                    trace = traceback.format_exc()
+                    self.error(
+                        f'Failed to delete schedule {schedule_name}: {str(e)}', metadata=metadata
+                    )
+                    report.append(
+                        {
+                            'namespace': namespace,
+                            'schedule_name': schedule_name,
+                            'success': False,
+                            'message': str(e),
+                            'attachment': trace,
+                        }
+                    )
+
+        self.info(
+            f'Deleted {success_count} of {len(schedules_to_delete)} schedules', metadata=metadata
+        )
+
+        self.debug(f'\n {json.dumps(report, indent=4, sort_keys=True)}', metadata=metadata)
+
+        return report

orchestrator/metrics.py

@@ -0,0 +1,23 @@
+"""
+Prometheus metric definitions for the orchestrator application.
+
+This module exposes Prometheus counters and gauges for monitoring the
+orchestrator, including application health and email delivery metrics.
+"""
+
+from prometheus_client import Counter, Gauge
+
+APP_UP = Gauge(
+    'app_up',
+    'Indicates if the application is running (1) or shutting down (0)',
+    ['pod_id'],
+)
+
+CORE_LABELS = ['pod_id', 'model_name', 'pipeline_name']
+
+
+EMAIL_SENT_COUNT = Counter(
+    'email_sent_count',
+    'Total number of emails sent by the orchestrator',
+    [*CORE_LABELS, 'email_group'],
+)

orchestrator/utils/connectors_config.py

@@ -0,0 +1,97 @@
+from os import getenv
+
+
+def build_redis_config():
+    """
+    Build Redis configuration from environment variables.
+
+    Returns:
+        dict: Redis configuration with host, port, username, and password.
+    """
+    return {
+        'host': getenv('REDIS_HOST', 'localhost'),
+        'port': int(getenv('REDIS_PORT', '6379')),
+        'username': getenv('REDIS_USERNAME', 'default'),
+        'password': getenv('REDIS_PASSWORD', 'bdnZOpcyiL'),
+    }
+
+
+def build_mongodb_config():
+    """
+    Build MongoDB configuration from environment variables.
+
+    Returns:
+        dict: MongoDB configuration with connection string, database name, and TTL index seconds.
+    """
+    username = getenv('MONGODB_USERNAME', 'root')
+    password = getenv('MONGODB_PASSWORD', 'wKZDbMNU1c')
+    uri = getenv('MONGODB_URL', 'localhost:27018')
+
+    connection_string = f'mongodb://{username}:{password}@{uri}'
+    return {
+        'connection_string': connection_string,
+        'database_name': getenv('MONGODB_DATABASE_NAME', 'sientia'),
+        'ttl_index_seconds': int(getenv('MONGODB_TTL_INDEX_HOURS', '1')) * 3600,
+    }
+
+
+def build_couchbase_config():
+    """
+    Build Couchbase configuration from environment variables.
+
+    Returns:
+        dict: Couchbase configuration with connection string, username, and password.
+    """
+    return {
+        'connection_string': getenv('COUCHBASE_CONNECTION_STRING', 'couchbase://localhost'),
+        'username': getenv('COUCHBASE_USERNAME', 'sientia'),
+        'password': getenv('COUCHBASE_PASSWORD', 'sientia'),
+    }
+
+
+def build_temporal_config():
+    """
+    Build Temporal configuration from environment variables.
+
+    Returns:
+        dict: Temporal configuration with host and namespace settings.
+    """
+    return {
+        'temporal_host': getenv('TEMPORAL_HOST', 'localhost:7233'),
+        'temporal_namespace': getenv('TEMPORAL_NAMESPACE', 'default'),
+        'temporal_scouter_namespace': getenv('TEMPORAL_SCOUTER_NAMESPACE', 'scouter'),
+        'temporal_laborious_namespace': getenv('TEMPORAL_LABORIOUS_NAMESPACE', 'laborious'),
+    }
+
+
+def build_postgres_config():
+    """
+    Build PostgreSQL configuration from environment variables.
+
+    Returns:
+        dict: PostgreSQL configuration with connection details and connection pool settings.
+    """
+    return {
+        'host': getenv('POSTGRES_HOST', 'localhost'),
+        'port': int(getenv('POSTGRES_PORT', '5432')),
+        'user': getenv('POSTGRES_USER', 'sientia'),
+        'password': getenv('POSTGRES_PASSWORD', 'sientia'),
+        'dbname': getenv('POSTGRES_DBNAME', 'sientia'),
+        'min_connections': int(getenv('POSTGRES_MIN_CONNECTIONS', '5')),
+        'max_connections': int(getenv('POSTGRES_MAX_CONNECTIONS', '20')),
+    }
+
+
+def build_email_config():
+    """
+    Build email configuration from environment variables.
+
+    Returns:
+        dict: Email configuration with SMTP server settings and sender credentials.
+    """
+    return {
+        'sender_email': getenv('EMAIL_SENDER', 'sientia-alerts@aignosi.com'),
+        'sender_password': getenv('EMAIL_SENDER_PASSWORD', 'sientia'),
+        'smtp_server': getenv('EMAIL_SMTP_SERVER', None),
+        'smtp_port': int(getenv('EMAIL_SMTP_PORT', '587')),
+    }

orchestrator/utils/converters.py

@@ -0,0 +1,31 @@
+def parse_frequency(frequency: str) -> int:
+    """
+    Parse frequency string into seconds for Temporal schedule intervals.
+
+    This function converts human-readable frequency strings into seconds
+    for use in Temporal schedule configurations. Supports seconds, minutes,
+    hours, and days notation.
+
+    Args:
+        frequency (str): Frequency string with suffix:
+            - 's' for seconds (e.g., '30s')
+            - 'm' for minutes (e.g., '5m')
+            - 'h' for hours (e.g., '2h')
+            - 'd' for days (e.g., '1d')
+
+    Returns:
+        int: Frequency converted to seconds
+
+    Raises:
+        ValueError: If frequency format is invalid
+    """
+    if frequency.endswith('s'):
+        return int(frequency[:-1])
+    elif frequency.endswith('m'):
+        return int(frequency[:-1]) * 60
+    elif frequency.endswith('h'):
+        return int(frequency[:-1]) * 60 * 60
+    elif frequency.endswith('d'):
+        return int(frequency[:-1]) * 60 * 60 * 24
+    else:
+        raise ValueError('Invalid frequency')

orchestrator/utils/email_builder.py

@@ -0,0 +1,132 @@
+from typing import Any
+
+from jinja2 import Template
+from sientia_do.observability.logger import Logger
+
+
+class EmailBuilder:
+    """
+    HTML email template builder for notification emails.
+
+    This class handles the generation of HTML email content from notification
+    data using Jinja2 templates. It supports different email types (alerts,
+    reports) and notification levels (ERROR, WARNING, INFO) with customizable
+    templates and parameter replacement.
+
+    Args:
+        logger (Logger): Application logger instance for error reporting
+    """
+
+    def __init__(self, logger: Logger):
+        self.logger = logger
+
+        self.report_template_file = './orchestrator/utils/templates/email_template.html'
+        self.general_template_file = './orchestrator/utils/templates/general_template.html'
+
+        with open(self.report_template_file) as file:
+            self.report_template = file.read()
+        with open(self.general_template_file) as file:
+            self.general_template = file.read()
+
+    def replace_parameters(self, template: str, parameters: dict) -> str:
+        """
+        Replace parameters in a Jinja2 template with provided values.
+
+        Renders a Jinja2 template string with the provided parameter dictionary,
+        replacing all template variables with their corresponding values.
+
+        Args:
+            template (str): The Jinja2 template string
+            parameters (dict): Dictionary of parameters to replace in the template
+
+        Returns:
+            str: The rendered template with parameters replaced
+        """
+        # Create a Jinja2 template from the provided string
+        template_obj = Template(template)
+
+        return template_obj.render(parameters)
+
+    def parameters(self, general_events: dict, mail_type: str) -> dict:
+        """
+        Build parameters dictionary for email templates based on general events and mail type.
+
+        Processes notification events organized by level and model, rendering HTML
+        sections for each notification level using the general template.
+
+        Args:
+            general_events (dict): Dictionary containing events categorized by level (ERROR, WARNING, INFO).
+                Each level contains a 'models' key with model-specific event data
+            mail_type (str): The type of email being sent (Alerts/Reports)
+
+        Returns:
+            dict: Dictionary with mail_type and rendered event sections for each notification level
+        """
+        error_events = general_events.get('ERROR', {})
+        warning_events = general_events.get('WARNING', {})
+        info_events = general_events.get('INFO', {})
+
+        error_models = error_events.get('models', [])
+        warning_models = warning_events.get('models', [])
+        info_models = info_events.get('models', [])
+
+        return {
+            'mail_type': mail_type,
+            'error_events': self.replace_parameters(self.general_template, error_events)
+            if error_models
+            else '',
+            'warning_events': self.replace_parameters(self.general_template, warning_events)
+            if warning_models
+            else '',
+            'info_events': self.replace_parameters(self.general_template, info_events)
+            if info_models
+            else '',
+        }
+
+    def build_email(self, report_data: list[dict[str, Any]], mail_type: str) -> str:
+        """
+        Build the email HTML by organizing report data by notification level and model.
+
+        Organizes notification data by level and model, then renders the complete
+        HTML email using the report template with all event sections.
+
+        Args:
+            report_data (list[dict[str, Any]]): List of notification reports, each containing:
+                - level (str): Notification level (ERROR, WARNING, INFO)
+                - model_name (str): Name of the model
+                - Additional notification details
+            mail_type (str): The type of email being built (Alerts/Reports)
+
+        Returns:
+            str: Complete HTML email content ready for sending
+        """
+        general_events: dict[str, dict[str, Any]] = {}
+
+        for report in report_data:
+            level = report['level']
+            model_name = report['model_name']
+
+            if level not in general_events:
+                general_events[level] = {
+                    'section_name': f'{level.capitalize()}s detected:',
+                    'models': {},
+                }
+
+            # Type assertion to help the type checker understand the structure
+            level_data = general_events[level]
+            models_dict = level_data['models']
+
+            if model_name not in models_dict:
+                models_dict[model_name] = {
+                    'model_name': model_name,
+                    'events': [],
+                }
+
+            models_dict[model_name]['events'].append(report)
+
+        for _type, content in general_events.items():
+            content['models'] = list(content['models'].values())
+
+        return self.replace_parameters(
+            self.report_template, self.parameters(general_events, mail_type)
+        )

orchestrator/utils/orchestrator_functions.py

@@ -0,0 +1,505 @@
+from typing import Any
+
+from orchestrator.utils.converters import parse_frequency
+
+
+def common_config(config: dict[str, Any]):
+    """
+    Extract common configuration parameters from a pipeline configuration.
+
+    Args:
+        config (dict[str, Any]): Pipeline configuration containing:
+            - workflow_type (str): Type of workflow (e.g., 'scouter', 'predictions_batch', 'drift')
+            - schedule_name (str): Unique name identifier for this schedule
+            - model_id (str): MongoDB ID of the associated model
+            - model (dict): Model configuration containing:
+                - name (str): Human-readable name of the model
+                - model_config (dict, optional): Additional model-specific configuration
+            - frequency (str, optional): Execution frequency (default: '1m')
+                Format: '{number}{unit}' where unit is 's', 'm', 'h', or 'd'
+            - offset (str, optional): Schedule offset/delay (default: '0m')
+            - max_retry_policy (int, optional): Maximum retry attempts on failure (default: 1)
+            - execution_timeout_seconds (int, optional): Workflow execution timeout (default: 300)
+            - task_timeout_seconds (int, optional): Individual task timeout (default: 300)
+
+    Returns:
+        dict[str, Any]: Common configuration dictionary with standardized parameters
+            for Temporal workflow execution
+    """
+    model = config['model']
+    return {
+        'workflow_type': config['workflow_type'],
+        'schedule_name': config['schedule_name'],
+        'frequency': config.get('frequency', '1m'),
+        'offset': config.get('offset', '0m'),
+        'max_retry_policy': config.get('max_retry_policy', 1),
+        'model_id': config['model_id'],
+        'model_name': model['name'],
+        'model_config': model.get('model_config', {}),
+        'execution_timeout_seconds': config.get('execution_timeout_seconds', 300),
+        'task_timeout_seconds': config.get('task_timeout_seconds', 300),
+        'on_conflict': config.get('on_conflict', 'error'),
+        'runtime': config.get('runtime', 'legacy'),
+    }
+
+
+def drift(config: dict[str, Any]):
+    """
+    Build drift configuration from pipeline config.
+
+    Creates a drift detection workflow configuration with database table mappings
+    and drift metric specifications for monitoring data distribution changes.
+
+    Args:
+        config (dict[str, Any]): Pipeline configuration containing:
+            - interval_minutes (int, optional): Time window for data comparison in minutes (default: 60)
+            - drift_metrics (list[str], optional): Statistical metrics to compute (default:
+                ['kolmogorov_smirnov', 'jensen_shannon', 'wasserstein'])
+            - Additional fields from common_config
+
+    Returns:
+        dict[str, Any]: Drift detection configuration with source/target tables,
+            time interval, and metrics specifications. Results stored in 'drift_metrics' table
+    """
+    return {
+        **common_config(config),
+        'schema': 'sientia_data',
+        'source_table_name': 'laborious_data',
+        'target_table_name': 'drift_metrics',
+        'interval': config.get('interval_minutes', 60),
+        'drift_metrics': config.get(
+            'drift_metrics', ['kolmogorov_smirnov', 'jensen_shannon', 'wasserstein']
+        ),
+    }
+
+
+def simple_metrics(config: dict[str, Any]):
+    """
+    Build simple metrics configuration from pipeline config.
+
+    Creates a simple metrics computation workflow configuration for calculating
+    model performance metrics like RMSE, MSE, MAE, and R².
+
+    Args:
+        config (dict[str, Any]): Pipeline configuration containing:
+            - interval_minutes (int, optional): Computation interval in minutes (default: 60)
+            - metrics (list[str], optional): List of metrics to compute
+                (default: ['rmse', 'mse', 'mae', 'r2'])
+            - Additional fields from common_config
+
+    Returns:
+        dict[str, Any]: Simple metrics configuration with source tables (predictions and
+            actual data), target table, time interval, and metrics list. Results stored in
+            'simple_metrics' table
+    """
+    return {
+        **common_config(config),
+        'schema': 'sientia_data',
+        'predictions_table_name': 'predictions',
+        'data_table_name': 'laborious_data',
+        'target_table_name': 'simple_metrics',
+        'interval_minutes': config.get('interval_minutes', 60),
+        'metrics': config.get('metrics', ['rmse', 'mse', 'mae', 'r2']),
+    }
+
+
+def minimal_retrain(config: dict[str, Any]):
+    """
+    Build minimal retrain configuration from pipeline config.
+
+    Args:
+        config (dict[str, Any]): Pipeline configuration containing:
+            - query (str): SQL query to retrieve training data. Should return features
+                and target variable in expected format
+            - datetime_columns (list[str], optional): Column names to parse as datetime
+                for proper temporal handling (default: [])
+            - Additional fields from common_config
+
+    Returns:
+        dict[str, Any]: Minimal retrain configuration with SQL query, database settings,
+            and datetime column specifications. Retraining logs are stored in 'log_retrain' table
+    """
+    return {
+        **common_config(config),
+        'query': config['query'],
+        'schema': 'sientia_data',
+        'table_name': 'log_retrain',
+        'datetime_columns': config.get('datetime_columns', []),
+    }
+
+
+def base_scouter(config: dict[str, Any]):
+    """
+    Build base scouter configuration shared by all scouter workflow types.
+
+    Creates the foundational configuration for OPC data collection workflows,
+    including filter policies, database settings, and data retention parameters.
+    This configuration is extended by specific scouter implementations (OPC UA, PI Web API).
+
+    Args:
+        config (dict[str, Any]): Pipeline configuration containing:
+            - filters (list[dict], optional): List of filter configurations with:
+                - filter_name (str): Name of the filter
+                - policy (str): Filter policy to apply
+            - tag_retention_minutes (int, optional): Tag retention time in minutes (default: 60)
+            - debug_data_package (bool, optional): Enable debug data package logging (default: False)
+            - fill_missing_tags (bool, optional): Fill missing tags with interpolation (default: False)
+            - Additional fields from common_config
+
+    Returns:
+        dict[str, Any]: Base scouter configuration with filters, database settings,
+            and retention policies
+    """
+
+    filters = {}
+    for f in config.get('filters', []):
+        filters[f['filter_name']] = {'policy': f['policy']}
+
+    return {
+        **common_config(config),
+        'trigger_laborious': False,
+        'filters': filters,
+        'schema': 'sientia_data',
+        'table_name': 'laborious_data',
+        'retention_time': config.get('tag_retention_minutes', 60) * 60,
+        'debug_data_package': config.get('debug_data_package', False),
+        'fill_missing_tags': config.get('fill_missing_tags', False),
+    }
+
+
+def scouter(config: dict[str, Any]):
+    """
+    Build scouter configuration from pipeline config.
+
+    Args:
+        config (dict[str, Any]): Pipeline configuration containing:
+            - schedule_name (str): Name of the schedule (used for topic generation)
+            - read_tags (list[dict]): List of tag configurations with:
+                - tag_name (str): Name of the tag to read
+                - aggr_func (str, optional): Aggregation function for data collection (default: 'lts')
+                    Common values: 'lts' (last), 'avg' (average), 'min', 'max', 'sum'
+                - data_range (list[int], optional): Valid data range [min, max] (default: [-100, 100])
+            - Additional fields from base_scouter
+
+    Returns:
+        dict[str, Any]: OPC UA scouter configuration with Kafka topic, tag mappings,
+            filters, and retention settings. Topic name follows pattern: 'raw_{schedule_name}'
+    """
+
+    tags = {}
+    for tag in config['read_tags']:
+        tags[tag['tag_name']] = {
+            'aggr_func': tag.get('aggr_func', 'lts'),
+            'data_range': tag.get('data_range', [-100, 100]),
+        }
+
+    return {
+        **base_scouter(config),
+        'topic': f'raw_{config["schedule_name"]}',
+        'model_tags': tags,
+    }
+
+
+def pi_web_api_scouter(config: dict[str, Any]):
+    """
+    Build PI Web API scouter configuration from pipeline config.
+
+    Creates a data collection workflow configuration for OSIsoft PI servers using
+    the PI Web API REST interface. Configures tag mappings with WebIDs, aggregation
+    functions, and API query parameters including timeout management.
+
+    Args:
+        config (dict[str, Any]): Pipeline configuration containing:
+            - read_tags (list[dict]): List of tag configurations with:
+                - tag_name (str): Name of the tag
+                - webid (str): PI Web API WebID for the tag
+                - aggr_func (str, optional): Aggregation function (default: 'lts')
+                - data_range (list[int], optional): Valid data range (default: [-100, 100])
+            - pi_web_api_config (dict): PI Web API connection settings with:
+                - endpoint (str): PI Web API endpoint URL
+                - period (str, optional): Time period for data retrieval (default: '*-1d')
+                - max_count (int, optional): Maximum number of values to retrieve (default: 1)
+                - api_timeout (int, optional): API request timeout in seconds
+            - Additional fields from base_scouter
+
+    Returns:
+        dict[str, Any]: PI Web API scouter configuration with tag mappings and query settings.
+            API timeout is automatically adjusted to not exceed workflow frequency.
+    """
+    tags = {}
+    for tag, tag_config in config['read_tags'].items():
+        tags[tag] = {
+            'webid': tag_config['webid'],
+            'aggr_func': tag_config.get('aggr_func', 'lts'),
+            'data_range': tag_config.get('data_range', [-100, 100]),
+        }
+
+    base_config = base_scouter(config)
+
+    pi_web_api_config = config['pi_web_api_config']
+
+    config_timeout = pi_web_api_config.get('api_timeout', None)
+    frequency = parse_frequency(base_config['frequency'])
+
+    if config_timeout is None or config_timeout > frequency:
+        config_timeout = frequency
+
+    return {
+        **base_config,
+        'model_tags': tags,
+        'pi_web_api_query': {
+            'endpoint': pi_web_api_config['endpoint'],
+            'period': pi_web_api_config.get('period', '*-1d'),
+            'max_count': pi_web_api_config.get('max_count', 1),
+            'api_timeout': config_timeout,
+        },
+    }
+
+
+def overlap_filter_config(base_filter_config: dict[str, Any], config: list[dict[str, Any]]):
+    """
+    Merge filter configurations with base filter configuration.
+
+    Extends the base filter configuration dictionary by adding or overwriting
+    filters from the provided configuration list. Used in predictions_batch
+    workflows to combine default filters with user-defined custom filters.
+
+    Args:
+        base_filter_config (dict[str, Any]): Base filter configuration dictionary to extend.
+            Each filter entry contains 'policy' and optionally 'config' keys.
+        config (list[dict[str, Any]]): List of filter configurations to merge, each containing:
+            - filter_name (str): Name of the filter to add or update
+            - policy (str): Filter policy (e.g., 'STOP', 'CONTINUE', 'REPEAT')
+            - config (dict, optional): Additional filter-specific configuration
+
+    Returns:
+        dict[str, Any]: Extended filter configuration dictionary with merged filters.
+            Filters from config list overwrite or add to base_filter_config entries.
+    """
+    for fil in config:
+        base_filter_config[fil['filter_name']] = {
+            'policy': fil['policy'],
+            'config': fil.get('config', {}),
+        }
+
+    return base_filter_config
+
+
+def process_path_priority(path_priority: list[str]):
+    """
+    Process and normalize path priority list for filter policy execution order.
+
+    Validates and normalizes the priority list used to determine the order in which
+    filter policies are evaluated in prediction workflows. Invalid priorities are
+    removed, missing required priorities are appended, and the list is truncated to
+    exactly 3 elements.
+
+    Valid priorities define workflow behavior when filters are triggered:
+        - STOP: Halt workflow execution immediately
+        - CONTINUE: Proceed to next step despite filter trigger
+        - REPEAT: Retry the current step
+
+    Args:
+        path_priority (list[str]): User-provided list of path priorities. May contain
+            invalid values or be incomplete.
+
+    Returns:
+        list[str]: Normalized path priority list with exactly 3 elements in user-specified
+            or default order. Default order when priorities are missing: ["STOP", "CONTINUE", "REPEAT"]
+    """
+    for priority in path_priority[:]:
+        if priority not in ['STOP', 'CONTINUE', 'REPEAT']:
+            path_priority.remove(priority)
+
+    for priority in ['STOP', 'CONTINUE', 'REPEAT']:
+        if priority not in path_priority:
+            path_priority.append(priority)
+
+    return path_priority[0:3]
+
+
+def predictions_batch(config: dict[str, Any]):
+    """
+    Build predictions batch configuration from pipeline config.
+
+    Args:
+        config (dict[str, Any]): Pipeline configuration containing:
+            - query (str): SQL query to retrieve input data for predictions
+            - write_tags (list[dict]): List of OPC tag configurations for write-back with:
+                - server_id (str): ID of the target OPC server
+                - type (str): Tag type - 'prediction' (model output) or 'confidence' (prediction confidence)
+                - addr (str): OPC tag address/path
+                - data_type (str, optional): OPC data type (default: 'float')
+            - datetime_columns (list[str], optional): Column names to parse as datetime (default: [])
+            - path_priority (list[str], optional): Filter policy execution order (default: ["STOP", "CONTINUE", "REPEAT"])
+            - input_filters (list[dict], optional): Input data validation filters
+            - mlflow_transform_filters (list[dict], optional): Transform stage filters
+            - mlflow_predict_filters (list[dict], optional): Prediction stage filters
+            - model_retention_minutes (int, optional): Data retention time in minutes (default: 60)
+            - save_transform (bool, optional): Save transformed data to database (default: True)
+            - predictions_storage_policy (str, optional): Prediction storage policy (default: 'lts:1')
+            - pi_web_api_output_config (dict, optional): PI Web API output configuration for write-back (default: {})
+            - Additional fields from common_config
+
+    Returns:
+        dict[str, Any]: Complete predictions batch configuration with OPC output mappings,
+            PI Web API output configuration, multi-stage filters, SQL query, and retention policies
+    """
+    tags: dict[str, Any] = {}
+    for tag in config.get('write_tags', []):
+        if tag['server_id'] not in tags:
+            tags[tag['server_id']] = {}
+
+        tag_type = tag['type']
+
+        if tag_type == 'prediction' or tag_type == 'confidence':
+            tag_type_str = f'{tag_type}_tags'
+
+            if tag_type_str not in tags[tag['server_id']]:
+                tags[tag['server_id']][tag_type_str] = {}
+
+            tags[tag['server_id']][tag_type_str][tag['addr']] = {
+                'data_type': tag.get('data_type', 'float'),
+            }
+
+    path_priority = process_path_priority(
+        config.get('path_priority', ['STOP', 'CONTINUE', 'REPEAT'])
+    )
+
+    return {
+        **common_config(config),
+        'query': config['query'],
+        'datetime_columns': config.get('datetime_columns', []),
+        'schema': 'sientia_data',
+        'table_name': 'predictions',
+        'save_transform': config.get('save_transform', True),
+        'transform_table_name': 'transformed_data',
+        'retention_time': config.get('model_retention_minutes', 60) * 60,
+        'opc_output_config': tags,
+        'pi_web_api_output_config': config.get('pi_web_api_output_config', {}),
+        'input_filters': overlap_filter_config(
+            {'EMPTY_DATA': {'policy': 'STOP', 'config': {}}}, config.get('input_filters', [])
+        ),
+        'mlflow_transform_filters': overlap_filter_config(
+            {
+                'EMPTY_DATA': {'policy': 'STOP', 'config': {}},
+                'API_ERROR': {'policy': 'STOP', 'config': {}},
+            },
+            config.get('mlflow_transform_filters', []),
+        ),
+        'mlflow_predict_filters': overlap_filter_config(
+            {'API_ERROR': {'policy': 'STOP', 'config': {}}},
+            config.get('mlflow_predict_filters', []),
+        ),
+        'path_priority': path_priority,
+        'predictions_storage_policy': config.get('predictions_storage_policy', 'lts:1'),
+    }
+
+
+def gather_read_tags(pipelines: list[dict[str, Any]]) -> dict[str, Any]:
+    """
+    Gather all read tags from scouter pipeline configurations.
+
+    Collects all read tags from scouter-type pipelines and organizes them by
+    server_id and tag_address, tracking which Kafka topics each tag is associated with.
+    This function is used during slot configuration to aggregate tags across multiple
+    scouter pipelines for efficient OPC server slot allocation.
+
+    Args:
+        pipelines (list[dict[str, Any]]): List of pipeline configurations to process.
+            Only pipelines with workflow_type 'scouter' are processed. Each scouter
+            pipeline should contain a 'read_tags' list with tag configurations.
+
+    Returns:
+        dict[str, Any]: Dictionary of read tags keyed by "server_id:tag_address",
+            where each entry contains:
+            - All original tag configuration fields
+            - topics (list[str]): List of Kafka topic names associated with this tag
+                (format: 'raw_{schedule_name}')
+    """
+
+    tags = {}
+
+    # Get all read tags from pipelines
+    for pipeline in pipelines:
+        if pipeline['workflow_type'] != 'scouter':
+            continue
+
+        for tag in pipeline.get('read_tags', []):
+            tag_string = f'{tag["server_id"]}:{tag["tag_address"]}'
+            if tag_string not in tags:
+                tags[tag_string] = {**tag, 'topics': []}
+
+            tags[tag_string]['topics'].append(f'raw_{pipeline["schedule_name"]}')
+
+    return tags
+
+
+def build_tag_config(
+    tags: list[dict[str, Any]], opc_servers: dict[str, Any]
+) -> tuple[dict[str, Any], list]:
+    """
+    Build tag configuration for a specific slot and OPC server.
+
+    Organizes tags by OPC server name and calculates the minimum subscription period
+    based on tag frequencies. Validates that all server IDs exist in the OPC
+    servers configuration. The subscription period is set to half of the minimum
+    tag frequency to ensure efficient data collection.
+
+    Args:
+        tags (list[dict[str, Any]]): List of tag configurations containing:
+            - server_id (str): ID of the OPC server
+            - tag_address (str): Address/path of the OPC tag
+            - frequency (int): Tag read frequency in milliseconds
+            - Additional tag-specific configuration fields
+        opc_servers (dict[str, Any]): Dictionary of OPC server configurations keyed by server_id.
+            Each server configuration should contain:
+            - server_name (str): Human-readable server name
+            - url (str): OPC server URL
+            - uri (str): OPC server URI
+            - cert_path (str, optional): Certificate file path
+            - private_key_path (str, optional): Private key file path
+            - server_cert_path (str, optional): Server certificate file path
+
+    Returns:
+        tuple[dict[str, Any], list]: A tuple containing:
+            - Slot configuration dictionary organized by server_name, where each server
+              contains connection details, tags dictionary, and subscription_period_ms
+            - List of server IDs (str) that were not found in opc_servers configuration
+    """
+
+    slot_config = {}
+    notifications = []
+
+    for tag in tags:
+        server_id = tag['server_id']
+
+        if server_id not in opc_servers:
+            notifications.append(server_id)
+            continue
+
+        server_name = opc_servers[server_id]['server_name']
+        if server_name not in slot_config:
+            slot_config[server_name] = {
+                'server_id': server_id,
+                'name': server_name,
+                'url': opc_servers[server_id]['url'],
+                'server_uri': opc_servers[server_id]['uri'],
+                'cert_path': opc_servers[server_id].get('cert_path', None),
+                'private_key_path': opc_servers[server_id].get('private_key_path', None),
+                'server_cert_path': opc_servers[server_id].get('server_cert_path', None),
+                'tags': {},
+            }
+
+        slot_config[server_name]['tags'][tag['tag_address']] = {
+            **tag,
+        }
+
+    for server_name in slot_config:
+        frequencies = [int(x['frequency']) for x in slot_config[server_name]['tags'].values()]
+
+        min_frequency = min(frequencies) if frequencies else 1000
+
+        slot_config[server_name]['subscription_period_ms'] = min_frequency / 2
+
+    return slot_config, notifications

orchestrator/utils/templates/email_template.html

@@ -0,0 +1,24 @@
+<!DOCTYPE html>
+<html lang="pt-BR">
+<head>
+    <meta charset="UTF-8">
+    <meta name="viewport" content="width=device-width, initial-scale=1.0">
+    <title>SIENTIA™ Report</title>
+    <style>
+        body { font-family: Arial, sans-serif; margin: 0; padding: 20px; }
+        h1, h2 { color: #333; }
+        table { width: 100%; border-collapse: collapse; margin-top: 20px; }
+        th, td { border: 1px solid #ddd; padding: 8px; text-align: left; }
+        th { background-color: #f4f4f4; }
+    </style>
+</head>
+<body>
+    <h1>SIENTIA™ {{ mail_type }}</h1>
+    
+    {{ error_events }}
+    {{ warning_events }}
+    {{ info_events }}
+
+    {{ special_events }}
+</body>
+</html>

orchestrator/utils/templates/general_template.html

@@ -0,0 +1,26 @@
+<h3>{{ section_name }}</h3>
+{% for model in models %}
+<h4>Model: <span>{{ model.model_name }}</span></h4>
+<table>
+    <thead>
+        <tr>
+            <th>Notification ID</th>
+            <th>Schedule</th>
+            <th>Block</th>
+            <th>Timestamp</th>
+            <th>Message</th>
+        </tr>
+    </thead>
+    <tbody>
+        {% for event in model.events %}
+        <tr>
+            <td>{{ event.notification_id }}</td>
+            <td>{{ event.trigger }}</td>
+            <td>{{ event.block }}</td>
+            <td>{{ event.timestamp }}</td>
+            <td>{{ event.message }}</td>
+        </tr>
+        {% endfor %}
+    </tbody>
+</table>
+{% endfor %}

orchestrator/worker/worker.py

@@ -0,0 +1,228 @@
+from temporalio import client, workflow
+from temporalio.runtime import PrometheusConfig, Runtime, TelemetryConfig
+
+with workflow.unsafe.imports_passed_through():
+    import asyncio
+    import os
+    import sys
+
+    from prometheus_client import start_http_server
+    from sientia_do.notifications.handlers import CoreNotificationHandler as NotificationHandler
+    from sientia_do.observability.logger import get_logger
+    from sientia_do.temporal.worker.prepare_worker import prepare_worker
+
+    from orchestrator import metrics
+    from orchestrator.activities.activities import Activities
+    from orchestrator.utils.connectors_config import (
+        build_email_config,
+        build_mongodb_config,
+        build_postgres_config,
+        build_redis_config,
+        build_temporal_config,
+    )
+    from orchestrator.workflows.alerts import Alerts
+    from orchestrator.workflows.orchestrator import Orchestrator
+    from orchestrator.workflows.reports import Reports
+    from orchestrator.workflows.subworkflows.load_notification_package import (
+        LoadNotificationPackage,
+    )
+    from orchestrator.workflows.subworkflows.process_notifications import ProcessNotifications
+
+POD_ID = os.getenv('POD_ID')
+SDK_METRICS_PORT = int(os.getenv('HTTP_SDK_METRICS_PORT', '9091'))
+
+
+async def main():
+    """
+    Main function to initialize and run the Temporal worker.
+
+    Sets up MongoDB connection, notification handler, Temporal client, and starts
+    multiple workers for different task queues (orchestrator, alerts, reports).
+    Handles graceful shutdown and error handling. Initializes Prometheus metrics
+    server and SDK metrics for monitoring.
+    """
+    host = os.getenv('TEMPORAL_HOST', 'localhost:7233')
+    namespace = os.getenv('TEMPORAL_NAMESPACE', 'default')
+    logger = get_logger(__name__)
+
+    metadata = {
+        'pod_id': POD_ID,
+        'model_name': '-',
+        'model_id': '-',
+        'workflow_name': '-',
+        'schedule_name': '-',
+    }
+
+    logger.custom_info(f'Starting Worker with POD_ID: {POD_ID}', metadata=metadata)
+
+    logger.custom_info('Starting prometheus client...', metadata=metadata)
+    start_prometheus_server()
+
+    logger.custom_info('Starting Notification Handler...', metadata=metadata)
+
+    mongo_config = build_mongodb_config()
+    notification_handler = NotificationHandler(
+        connection_string=mongo_config['connection_string'],
+        database=mongo_config['database_name'],
+        logger=logger,
+        project_name=os.getenv('PROJECT_NAME', 'orchestrator'),
+    )
+
+    logger.custom_info(
+        f'Starting SDK Metrics Server on port {SDK_METRICS_PORT}...', metadata=metadata
+    )
+
+    new_runtime = Runtime(
+        telemetry=TelemetryConfig(
+            metrics=PrometheusConfig(bind_address=f'0.0.0.0:{SDK_METRICS_PORT}')
+        )
+    )
+
+    logger.custom_info(f'Starting Temporal Client at {host}:{namespace}', metadata=metadata)
+
+    temporal_client = await client.Client.connect(
+        target_host=host,
+        namespace=os.getenv('TEMPORAL_NAMESPACE', 'laborious'),
+        runtime=new_runtime,
+    )
+
+    logger.custom_info('Starting Activities...', metadata=metadata)
+
+    activities = Activities(
+        temporal_config=build_temporal_config(),
+        redis_config=build_redis_config(),
+        mongodb_config=build_mongodb_config(),
+        email_config=build_email_config(),
+        postgres_config=build_postgres_config(),
+        logger=logger,
+        notification_handler=notification_handler,
+    )
+
+    await activities.connect_to_temporal()
+
+    logger.custom_info('Starting Workers...', metadata=metadata)
+
+    workers = [
+        prepare_worker(
+            temporal_client=temporal_client,
+            main_workflow=Orchestrator,
+            other_workflows=[],
+            activities=[
+                # Redis
+                activities.load_active_ingestors,
+                activities.load_opc_slots,
+                activities.update_slots,
+                activities.delete_slots,
+                # Couchbase
+                # activities.load_query_from_couchbase,
+                # MongoDB
+                activities.aggregate_documents_in_mongodb,
+                activities.find_documents_in_mongodb,
+                activities.update_pipelines_timestamps,
+                activities.create_pipelines_timestamps,
+                activities.delete_pipelines_timestamps,
+                activities.create_collection_with_ttl_index,
+                # Temporal
+                activities.create_schedules,
+                activities.update_schedules,
+                activities.delete_schedules,
+                activities.normalize_schedules,
+                # Formatters
+                activities.process_schedules,
+                activities.process_slots,
+                activities.create_schedule_config,
+                activities.create_slot_config,
+                activities.report_schedule_orchestration,
+                activities.report_slot_orchestration,
+                activities.format_schedule_config,
+            ],
+            logger=logger,
+        ),
+        prepare_worker(
+            temporal_client=temporal_client,
+            main_workflow=Alerts,
+            other_workflows=[LoadNotificationPackage, ProcessNotifications],
+            activities=[
+                # Load notifications
+                activities.get_last_data_timestamp,
+                activities.find_documents_in_mongodb,
+                activities.load_latest_data,
+                activities.put_last_data_timestamp,
+                # Format and filter notifications
+                activities.filter_notification_alerts,
+                # Send email and export data to postgres
+                activities.build_email_html,
+                activities.send_email,
+                activities.format_log_report,
+                activities.export_data_to_postgres,
+                # Store notification cache
+                activities.store_notification_cache,
+            ],
+            logger=logger,
+        ),
+        prepare_worker(
+            temporal_client=temporal_client,
+            main_workflow=Reports,
+            other_workflows=[LoadNotificationPackage, ProcessNotifications],
+            activities=[
+                # Load notifications
+                activities.get_last_data_timestamp,
+                activities.find_documents_in_mongodb,
+                activities.load_latest_data,
+                activities.put_last_data_timestamp,
+                # Format and filter notifications
+                activities.filter_notification_reports,
+                # Send email and export data to postgres
+                activities.build_email_html,
+                activities.send_email,
+                activities.format_log_report,
+                activities.export_data_to_postgres,
+            ],
+            logger=logger,
+        ),
+    ]
+
+    handlers = []
+    for w in workers:
+        handlers.append(w.run())
+
+    logger.custom_info('Workers started successfully', metadata=metadata)
+
+    exit_code = 0
+    try:
+        await asyncio.gather(*handlers)
+    except BaseException as e:
+        logger.error(f'An unhandled exception occurred: {e}', exc_info=True)
+        exit_code = 1
+    finally:
+        if notification_handler:
+            notification_handler.shutdown()
+        if activities:
+            activities.shutdown()
+        metrics.APP_UP.labels(pod_id=POD_ID).set(0)
+        sys.exit(exit_code)
+
+
+def start_prometheus_server():
+    """
+    Start the Prometheus metrics server on the configured port.
+
+    Sets up HTTP server for metrics collection and marks the application as UP.
+    Exits the application if the server fails to start. The metrics server
+    exposes application metrics on the port specified by HTTP_METRICS_PORT.
+
+    Raises:
+        SystemExit: If the metrics server fails to start
+    """
+    try:
+        port = int(os.getenv('HTTP_METRICS_PORT', 9090))
+        start_http_server(port)
+        print(f'Prometheus server started on port {port}.')
+        metrics.APP_UP.labels(pod_id=POD_ID).set(1)
+    except Exception as e:
+        print(f'Failed to start Prometheus server: {e}')
+        os._exit(1)
+
+
+if __name__ == '__main__':
+    asyncio.run(main())

orchestrator/workflows/alerts.py

@@ -0,0 +1,116 @@
+from temporalio import workflow
+
+with workflow.unsafe.imports_passed_through():
+    from datetime import timedelta
+    from typing import Any
+
+    from sientia_do.temporal.policies import retry_policy
+
+    from orchestrator.activities.activities import Activities
+
+
+@workflow.defn(name='alerts')
+class Alerts:
+    """
+    Alerts workflow for real-time error notification delivery.
+
+    This workflow processes ERROR-level notifications from the notification queue
+    and sends immediate alerts to configured user groups. It implements intelligent
+    filtering with TTL-based duplicate prevention and persistent alert detection
+    for ongoing issues.
+    """
+
+    @workflow.run
+    async def run(self, input_data: dict[str, Any]):
+        """
+        Execute the alerts workflow for real-time error notification delivery.
+
+        This workflow loads ERROR-level notifications from MongoDB, applies
+        intelligent filtering with TTL management to prevent alert spam,
+        and sends immediate email alerts to configured receiver groups.
+
+        The workflow implements:
+        - TTL-based duplicate prevention for notifications
+        - Persistent alert detection for ongoing issues
+        - User group-based filtering with ignore lists
+        - Audit logging of alert delivery status
+
+        Args:
+            input_data (dict[str, Any]): Workflow input parameters.
+                Required fields:
+                - schedule_name (str): Name of the alert schedule
+                - notification_ttl (int): Seconds before considering notification persistent
+                - sent_ttl (int): Time-to-live for sent notification cache
+
+        Returns:
+            None: Workflow completes without return value
+
+        Raises:
+            Exception: If alert processing or delivery fails
+        """
+        metadata = {
+            'metadata': {
+                'schedule_name': input_data['schedule_name'],
+                'workflow_name': 'alerts',
+                'model_name': '-',
+                'model_id': '-',
+            }
+        }
+
+        mail_type = 'Alerts'
+
+        input_data['metadata'] = metadata
+        input_data['mail_type'] = mail_type
+
+        input_data['base_data_filter'] = {'level': 'ERROR'}
+
+        # Call subworkflow "load_notification_package" passing the static filters
+        # (level = "ERROR" and timestamp > last timestamp)
+
+        package = await workflow.execute_child_workflow(
+            'subworkflow.load_notification_package', input_data
+        )
+
+        if not package['notification_package'] or not package['sending_configs']:
+            return
+
+        # Filter notification package by groups custom configs, levels and
+        # timestamp cached
+
+        receiver_groups = await workflow.execute_local_activity_method(
+            Activities.filter_notification_alerts,
+            {
+                **metadata,
+                'notification_package': package['notification_package'],
+                'sending_configs': package['sending_configs'],
+                'notification_ttl': input_data['notification_ttl'],
+            },
+            schedule_to_close_timeout=timedelta(seconds=60),
+            retry_policy=retry_policy,
+        )
+
+        if not receiver_groups:
+            return
+
+        # Call subworkflow "process_notifications" passing the notification package
+        log_report = await workflow.execute_child_workflow(
+            'subworkflow.process_notifications',
+            {
+                'metadata': metadata,
+                'mail_type': mail_type,
+                'notification_package': receiver_groups,
+                'schema': 'sientia_data',
+                'table_name': 'log_report',
+            },
+        )
+
+        if not log_report:
+            return
+
+        # Store the notification_id sendings to avoid sending them again
+        await workflow.execute_activity_method(
+            Activities.store_notification_cache,
+            {**metadata, 'log_report': log_report, 'sent_ttl': input_data['sent_ttl']},
+            schedule_to_close_timeout=timedelta(seconds=60),
+            retry_policy=retry_policy,
+        )

orchestrator/workflows/orchestrator.py

@@ -0,0 +1,281 @@
+from temporalio import workflow
+
+with workflow.unsafe.imports_passed_through():
+    from datetime import timedelta
+    from typing import Any
+
+    from sientia_do.temporal.policies import retry_policy
+
+    from orchestrator.activities.activities import Activities
+
+
+@workflow.defn(name='orchestrator')
+class Orchestrator:
+    """
+    Main orchestrator workflow for pipeline and resource management.
+
+    This workflow coordinates pipeline deployment and OPC server slot
+    management by retrieving configurations from MongoDB and Redis,
+    processing schedules, and deploying them to the Temporal server
+    and Redis infrastructure.
+    """
+
+    @workflow.run
+    async def run(self, input_data: dict[str, Any]):
+        """
+        Execute the orchestration workflow for pipeline and slot management.
+
+        This workflow retrieves pipeline configurations and OPC server data,
+        processes schedules and slot configurations, and deploys them to
+        the appropriate services. It handles creation, updates, and deletion
+        of schedules and slots based on current system state.
+
+        Args:
+            input_data (dict[str, Any]): Workflow input parameters.
+                Required fields:
+                - schedule_name (str): Name of the orchestration schedule
+                - pipelines_query (dict[str, Any]): MongoDB query for pipeline configurations
+                - opc_servers_query (dict[str, Any]): MongoDB query for OPC server data
+
+        Returns:
+            None: Workflow completes without return value
+
+        Raises:
+            Exception: If orchestration operations fail
+        """
+
+        input_data['workflow_name'] = 'orchestrator'
+
+        metadata = {
+            'metadata': {
+                'schedule_name': input_data.get('schedule_name', 'orchestrator'),
+                'model_name': '-',
+                'model_id': '-',
+                'workflow_name': input_data['workflow_name'],
+            }
+        }
+
+        pipeline_config_handler = workflow.start_local_activity_method(
+            Activities.aggregate_documents_in_mongodb,
+            {
+                **metadata,
+                'query': input_data['pipelines_query'],
+                'timestamp_fields': ['updated_at'],
+            },
+            retry_policy=retry_policy,
+            start_to_close_timeout=timedelta(seconds=60),
+        )
+
+        opc_servers_handler = workflow.start_local_activity_method(
+            Activities.find_documents_in_mongodb,
+            {**metadata, 'query': input_data['opc_servers_query']},
+            retry_policy=retry_policy,
+            start_to_close_timeout=timedelta(seconds=60),
+        )
+
+        orchestrated_schedules_handler = workflow.start_local_activity_method(
+            Activities.find_documents_in_mongodb,
+            {
+                **metadata,
+                'query': {'collection': 'orchestrated_schedules'},
+                'timestamp_fields': ['updated_at'],
+            },
+            retry_policy=retry_policy,
+            start_to_close_timeout=timedelta(seconds=60),
+        )
+
+        current_slot_config_handler = workflow.start_local_activity_method(
+            Activities.load_opc_slots,
+            {
+                **metadata,
+            },
+            retry_policy=retry_policy,
+            start_to_close_timeout=timedelta(seconds=60),
+        )
+
+        active_ingestors_handler = workflow.start_local_activity_method(
+            Activities.load_active_ingestors,
+            {
+                **metadata,
+            },
+            retry_policy=retry_policy,
+            start_to_close_timeout=timedelta(seconds=60),
+        )
+
+        pipeline_config = await pipeline_config_handler
+        orchestrated_schedules = await orchestrated_schedules_handler
+        current_slot_config = await current_slot_config_handler
+        opc_servers = await opc_servers_handler
+        active_ingestors = await active_ingestors_handler
+
+        formatted_orchestrated_schedules_handler = workflow.start_local_activity_method(
+            Activities.format_schedule_config,
+            {**metadata, 'schedule_config': orchestrated_schedules},
+            retry_policy=retry_policy,
+            start_to_close_timeout=timedelta(seconds=60),
+        )
+
+        schedules_config_handler = workflow.start_local_activity_method(
+            Activities.process_schedules,
+            {**metadata, 'pipelines': pipeline_config},
+            retry_policy=retry_policy,
+            start_to_close_timeout=timedelta(seconds=60),
+        )
+
+        slot_config_handler = workflow.start_local_activity_method(
+            Activities.process_slots,
+            {
+                **metadata,
+                'opc_servers': opc_servers,
+                'active_ingestors': active_ingestors,
+                'pipelines': pipeline_config,
+            },
+            retry_policy=retry_policy,
+            start_to_close_timeout=timedelta(seconds=60),
+        )
+
+        schedules_config = await schedules_config_handler
+        slot_config = await slot_config_handler
+        formatted_orchestrated_schedules = await formatted_orchestrated_schedules_handler
+
+        schedule_actions_handler = workflow.start_local_activity_method(
+            Activities.create_schedule_config,
+            {
+                **metadata,
+                'current_schedule_config': formatted_orchestrated_schedules,
+                'schedule_config': schedules_config,
+            },
+            retry_policy=retry_policy,
+            start_to_close_timeout=timedelta(seconds=60),
+        )
+
+        slot_actions_handler = workflow.start_local_activity_method(
+            Activities.create_slot_config,
+            {**metadata, 'current_slot_config': current_slot_config, 'slot_config': slot_config},
+            retry_policy=retry_policy,
+            start_to_close_timeout=timedelta(seconds=60),
+        )
+
+        normalize_schedules_handler = workflow.start_activity_method(
+            Activities.normalize_schedules,
+            {**metadata, 'orchestrated_schedules': formatted_orchestrated_schedules},
+            retry_policy=retry_policy,
+            start_to_close_timeout=timedelta(seconds=60),
+        )
+
+        create_collection_with_ttl_index_handler = workflow.start_activity_method(
+            Activities.create_collection_with_ttl_index,
+            {**metadata, 'pipelines': schedules_config['scouter']},
+            retry_policy=retry_policy,
+            start_to_close_timeout=timedelta(seconds=60),
+        )
+
+        schedule_actions = await schedule_actions_handler
+        slot_actions = await slot_actions_handler
+        await normalize_schedules_handler
+        await create_collection_with_ttl_index_handler
+
+        slot_deletion_report_handler = workflow.start_activity_method(
+            Activities.delete_slots,
+            {**metadata, 'to_delete': slot_actions['to_delete']},
+            retry_policy=retry_policy,
+            start_to_close_timeout=timedelta(seconds=60),
+        )
+
+        slot_insertion_report_handler = workflow.start_activity_method(
+            Activities.update_slots,
+            {**metadata, 'to_insert': slot_actions['to_insert']},
+            retry_policy=retry_policy,
+            start_to_close_timeout=timedelta(seconds=60),
+        )
+
+        schedule_deletion_report_handler = workflow.start_activity_method(
+            Activities.delete_schedules,
+            {**metadata, 'schedules': schedule_actions['to_delete']},
+            retry_policy=retry_policy,
+            start_to_close_timeout=timedelta(seconds=60),
+        )
+
+        schedule_insertion_report_handler = workflow.start_activity_method(
+            Activities.create_schedules,
+            {**metadata, 'schedules': schedule_actions['to_create']},
+            retry_policy=retry_policy,
+            start_to_close_timeout=timedelta(seconds=60),
+        )
+
+        schedule_update_report_handler = workflow.start_activity_method(
+            Activities.update_schedules,
+            {**metadata, 'schedules': schedule_actions['to_update']},
+            retry_policy=retry_policy,
+            start_to_close_timeout=timedelta(seconds=60),
+        )
+
+        slot_deletion_report = await slot_deletion_report_handler
+        slot_insertion_report = await slot_insertion_report_handler
+        schedule_deletion_report = await schedule_deletion_report_handler
+        schedule_insertion_report = await schedule_insertion_report_handler
+        schedule_update_report = await schedule_update_report_handler
+
+        if schedule_insertion_report or schedule_update_report or schedule_deletion_report:
+            schedule_report_handler = workflow.start_activity_method(
+                Activities.report_schedule_orchestration,
+                {
+                    **metadata,
+                    'created_schedules': schedule_insertion_report,
+                    'updated_schedules': schedule_update_report,
+                    'deleted_schedules': schedule_deletion_report,
+                },
+                retry_policy=retry_policy,
+                start_to_close_timeout=timedelta(seconds=60),
+            )
+
+        if slot_insertion_report or slot_deletion_report:
+            slot_report_handler = workflow.start_activity_method(
+                Activities.report_slot_orchestration,
+                {
+                    **metadata,
+                    'inserted_slots': slot_insertion_report,
+                    'deleted_slots': slot_deletion_report,
+                },
+                retry_policy=retry_policy,
+                start_to_close_timeout=timedelta(seconds=60),
+            )
+
+        if schedule_update_report:
+            update_pipelines_timestamps_handler = workflow.start_activity_method(
+                Activities.update_pipelines_timestamps,
+                {**metadata, 'updated_pipelines': schedule_update_report},
+                retry_policy=retry_policy,
+                start_to_close_timeout=timedelta(seconds=60),
+            )
+
+        if schedule_insertion_report:
+            create_pipelines_timestamps_handler = workflow.start_activity_method(
+                Activities.create_pipelines_timestamps,
+                {**metadata, 'created_pipelines': schedule_insertion_report},
+                retry_policy=retry_policy,
+                start_to_close_timeout=timedelta(seconds=60),
+            )
+
+        if schedule_deletion_report:
+            delete_pipelines_timestamps_handler = workflow.start_activity_method(
+                Activities.delete_pipelines_timestamps,
+                {**metadata, 'deleted_pipelines': schedule_deletion_report},
+                retry_policy=retry_policy,
+                start_to_close_timeout=timedelta(seconds=60),
+            )
+
+        if schedule_insertion_report or schedule_update_report or schedule_deletion_report:
+            await schedule_report_handler
+
+        if slot_insertion_report or slot_deletion_report:
+            await slot_report_handler
+
+        if schedule_update_report:
+            await update_pipelines_timestamps_handler
+
+        if schedule_insertion_report:
+            await create_pipelines_timestamps_handler
+
+        if schedule_deletion_report:
+            await delete_pipelines_timestamps_handler

orchestrator/workflows/reports.py

@@ -0,0 +1,96 @@
+from temporalio import workflow
+
+with workflow.unsafe.imports_passed_through():
+    from datetime import timedelta
+    from typing import Any
+
+    from sientia_do.temporal.policies import retry_policy
+
+    from orchestrator.activities.activities import Activities
+
+
+@workflow.defn(name='reports')
+class Reports:
+    """
+    Reports workflow for sending scheduled notification summaries.
+
+    This workflow processes and sends scheduled reports to configured
+    user groups. It loads notification data from MongoDB, filters it
+    by receiver group configurations, and sends formatted HTML reports
+    via email.
+    """
+
+    @workflow.run
+    async def run(self, input_data: dict[str, Any]):
+        """
+        Execute the reports workflow for scheduled notification delivery.
+
+        This workflow loads all notifications from the notification queue,
+        applies receiver group filtering, and sends comprehensive HTML
+        reports to configured user groups.
+
+        Args:
+            input_data (dict[str, Any]): Workflow input parameters.
+                Required fields:
+                - schedule_name (str): Name of the report schedule
+
+        Returns:
+            None: Workflow completes without return value
+
+        Raises:
+            Exception: If report generation or delivery fails
+        """
+        metadata = {
+            'metadata': {
+                'schedule_name': input_data['schedule_name'],
+                'workflow_name': 'reports',
+                'model_name': '-',
+                'model_id': '-',
+            }
+        }
+
+        mail_type = 'Reports'
+
+        input_data['metadata'] = metadata
+        input_data['mail_type'] = mail_type
+
+        input_data['base_data_filter'] = {}
+
+        # Call subworkflow "load_notification_package" passing the static filters
+        # (timestamp > last timestamp)
+
+        package = await workflow.execute_child_workflow(
+            'subworkflow.load_notification_package', input_data
+        )
+
+        if not package['notification_package'] or not package['sending_configs']:
+            return
+
+        # Filter notification package by groups custom configs, levels and
+        # timestamp cached
+
+        receiver_groups = await workflow.execute_local_activity_method(
+            Activities.filter_notification_reports,
+            {
+                **metadata,
+                'notification_package': package['notification_package'],
+                'sending_configs': package['sending_configs'],
+            },
+            schedule_to_close_timeout=timedelta(seconds=60),
+            retry_policy=retry_policy,
+        )
+
+        if not receiver_groups:
+            return
+
+        # Call subworkflow "process_notifications" passing the notification package
+        await workflow.execute_child_workflow(
+            'subworkflow.process_notifications',
+            {
+                'metadata': metadata,
+                'mail_type': mail_type,
+                'notification_package': receiver_groups,
+                'schema': 'sientia_data',
+                'table_name': 'log_report',
+            },
+        )

orchestrator/workflows/subworkflows/load_notification_package.py

@@ -0,0 +1,108 @@
+from temporalio import workflow
+
+with workflow.unsafe.imports_passed_through():
+    from datetime import timedelta
+    from typing import Any
+
+    from sientia_do.temporal.policies import retry_policy
+
+    from orchestrator.activities.activities import Activities
+
+
+@workflow.defn(name='subworkflow.load_notification_package')
+class LoadNotificationPackage:
+    """
+    Subworkflow for loading notification data and configuration.
+
+    This subworkflow retrieves notification packages from MongoDB and
+    loads receiver group configurations. It handles timestamp-based
+    filtering for incremental data processing and manages the data
+    required for notification workflows.
+    """
+
+    @workflow.run
+    async def run(self, input_data: dict[str, Any]):
+        """
+        Load notification package and sending configurations.
+
+        This subworkflow loads notifications from the MongoDB notification
+        queue using timestamp-based filtering and retrieves active receiver
+        group configurations. It updates the last processed timestamp in Redis.
+
+        Args:
+            input_data (dict[str, Any]): Workflow input parameters.
+                Required fields:
+                - metadata (dict[str, Any]): Workflow execution metadata
+                - mail_type (str): Type of mail (Alerts/Reports)
+                - base_data_filter (dict[str, Any]): Base filter for notification query
+
+        Returns:
+            dict[str, Any]: Package containing:
+                - last_timestamp (str | None): Last processed timestamp
+                - notification_package (list[dict]): Retrieved notifications
+                - sending_configs (list[dict]): Active receiver group configurations
+
+        Raises:
+            Exception: If data loading fails
+        """
+        metadata = input_data['metadata']
+
+        # Load last timestamp from redis "notification_last_timestamp"
+        last_timestamp_handler = workflow.start_local_activity_method(
+            Activities.get_last_data_timestamp,
+            {**metadata, 'mail_type': input_data['mail_type']},
+            start_to_close_timeout=timedelta(seconds=60),
+            retry_policy=retry_policy,
+        )
+
+        # In parallel, load sending configs from collection "receiver_groups"
+        sending_configs_handler = workflow.start_local_activity_method(
+            Activities.find_documents_in_mongodb,
+            {**metadata, 'query': {'collection': 'receiver_groups', 'filters': {'active': True}}},
+            start_to_close_timeout=timedelta(seconds=60),
+            retry_policy=retry_policy,
+        )
+
+        last_timestamp = await last_timestamp_handler
+
+        # Load notification package from collection "notification_queue", using a
+        # static filter
+
+        notification_package = await workflow.start_local_activity_method(
+            Activities.load_latest_data,
+            {
+                **metadata,
+                'collection_name': 'notification_queue',
+                'last_data_timestamp': last_timestamp,
+                'base_data_filter': input_data['base_data_filter'],
+            },
+            start_to_close_timeout=timedelta(seconds=60),
+            retry_policy=retry_policy,
+        )
+        sending_configs = await sending_configs_handler
+
+        if not sending_configs or not notification_package:
+            return {
+                'last_timestamp': last_timestamp,
+                'notification_package': notification_package,
+                'sending_configs': sending_configs,
+            }
+
+        # Put last collected timestamp in redis "notification_last_timestamp"
+
+        await workflow.start_activity_method(
+            Activities.put_last_data_timestamp,
+            {**metadata, 'data': notification_package, 'mail_type': input_data['mail_type']},
+            start_to_close_timeout=timedelta(seconds=60),
+            retry_policy=retry_policy,
+        )
+
+        # Return a dict with the following keys:
+        # - last_timestamp
+        # - notification_package
+        # - sending_configs
+        return {
+            'last_timestamp': last_timestamp,
+            'notification_package': notification_package,
+            'sending_configs': sending_configs,
+        }

orchestrator/workflows/subworkflows/process_notifications.py

@@ -0,0 +1,99 @@
+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 orchestrator.activities.activities import Activities
+
+
+@workflow.defn(name='subworkflow.process_notifications')
+class ProcessNotifications:
+    """
+    Subworkflow for processing and sending notification emails.
+
+    This subworkflow handles the email delivery process including HTML
+    generation, email sending, and logging to PostgreSQL. It processes
+    receiver groups and generates delivery reports for monitoring.
+    """
+
+    @workflow.run
+    async def run(self, input_data: dict[str, Any]) -> dict[str, Any]:
+        """
+        Process notifications and send emails to configured receiver groups.
+
+        This subworkflow builds HTML email content, sends emails to all
+        receiver groups, and logs the delivery results to PostgreSQL for
+        monitoring and audit purposes.
+
+        Args:
+            input_data (dict[str, Any]): Workflow input parameters.
+                Required fields:
+                - metadata (dict[str, Any]): Workflow execution metadata
+                - mail_type (str): Type of email being sent
+                - schema (str): PostgreSQL schema name for logging
+                - table_name (str): PostgreSQL table name for logging
+                - notification_package (dict[str, Any]): Receiver groups with notifications
+
+        Returns:
+            dict[str, Any]: Log report of email delivery results
+
+        Raises:
+            Exception: If notification processing fails
+        """
+
+        metadata = input_data['metadata']
+
+        # Use notification package to create the report html for each group and each model
+        data_to_sent = await workflow.execute_local_activity_method(
+            Activities.build_email_html,
+            {
+                **metadata,
+                'receiver_groups': input_data['notification_package'],
+                'mail_type': input_data['mail_type'],
+            },
+            schedule_to_close_timeout=timedelta(seconds=60),
+            retry_policy=retry_policy,
+        )
+
+        # Send the report html to the receivers of each group
+        log_report = await workflow.execute_activity_method(
+            Activities.send_email,
+            {**metadata, 'receiver_groups': data_to_sent, 'mail_type': input_data['mail_type']},
+            schedule_to_close_timeout=timedelta(seconds=60),
+            retry_policy=retry_policy,
+        )
+
+        if not log_report:
+            return {}
+
+        # Format the log report to a dataframe to be stored in the database
+        log_report = await workflow.execute_local_activity_method(
+            Activities.format_log_report,
+            {**metadata, 'receiver_groups': log_report, 'mail_type': input_data['mail_type']},
+            schedule_to_close_timeout=timedelta(seconds=60),
+            retry_policy=retry_policy,
+        )
+
+        # Store sending log in postgres database "log_report"
+        await workflow.execute_activity_method(
+            Activities.export_data_to_postgres,
+            {
+                **metadata,
+                'schema': input_data['schema'],
+                'table_name': input_data['table_name'],
+                'data': log_report,
+                'timestamp_conversion': {
+                    'column': 'timestamp',
+                    'format': DATETIME_FORMAT_WITH_TZ,
+                },
+            },
+            schedule_to_close_timeout=timedelta(seconds=60),
+            retry_policy=retry_policy,
+        )
+
+        # Return the log report to the caller
+        return log_report