Code import - branch 0.6.0

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,484 @@
+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
+                    )
+
+                    task_queue_name = self._build_task_queue_name(workflow_type, schedule)
+                    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
+
+    @staticmethod
+    def _build_task_queue_name(workflow_type: str, schedule: dict[str, Any]) -> str:
+        """
+        Build the Temporal task queue name for a schedule.
+
+        Only workflow types listed in ``RUNTIME_WORKFLOWS`` get an environment/tenant
+        specific ``runtime`` suffix; every other workflow type gets a plain queue name.
+        """
+        runtime_name = (
+            schedule.get('runtime', 'legacy') if workflow_type in RUNTIME_WORKFLOWS else None
+        )
+        return build_queue_name(workflow_type, runtime_name)
+
+    def _make_schedule_updater(self, schedule: dict[str, Any], metadata: dict[str, Any]):
+        """Build the ``ScheduleUpdate`` callback used by ``handler.update`` for one schedule."""
+
+        # 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
+        return update_schedule
+
+    async def _update_single_schedule(
+        self,
+        client: Client,
+        schedule_name: str,
+        schedule: dict[str, Any],
+        metadata: dict[str, Any],
+    ) -> None:
+        """Update a single schedule in Temporal, raising if the schedule handle is missing."""
+        handler = client.get_schedule_handle(schedule_name)
+
+        if not handler:
+            raise ValueError(f'Schedule {schedule_name} not found')
+
+        workflow_type = schedule['workflow_type']
+        schedule['task_queue'] = self._build_task_queue_name(workflow_type, schedule)
+
+        update_schedule = self._make_schedule_updater(schedule, metadata)
+        await handler.update(update_schedule)
+
+    @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:
+                    await self._update_single_schedule(client, schedule_name, schedule, metadata)
+
+                    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