feat(notifications): add support for notifications and notification hooks

Author: wakonig

bec_lib/bec_lib/endpoints.py

@@ -1995,6 +1995,38 @@ def available_messaging_services():
             message_op=MessageOp.STREAM,
         )
 
+    @staticmethod
+    def notification(event_type: str):
+        """
+        Endpoint for transient notification events that SciHub can route to
+        configured messaging services.
+
+        Args:
+            event_type (str): Notification event name such as ``new_scan``.
+
+        Returns:
+            EndpointInfo: Endpoint for notification events.
+        """
+        endpoint = f"{EndpointType.INTERNAL.value}/messaging_services/notification/{event_type}"
+        return EndpointInfo(
+            endpoint=endpoint, message_type=messages.NotificationMessage, message_op=MessageOp.SEND
+        )
+
+    @staticmethod
+    def notification_config():
+        """
+        Endpoint for persisted notification routing configuration.
+
+        Returns:
+            EndpointInfo: Endpoint for notification routing config.
+        """
+        endpoint = f"{EndpointType.USER.value}/messaging_services/notification_config"
+        return EndpointInfo(
+            endpoint=endpoint,
+            message_type=messages.NotificationConfigMessage,
+            message_op=MessageOp.SET_PUBLISH,
+        )
+
     @staticmethod
     def message_service_ingest(deployment_name: str):
         """

bec_lib/bec_lib/messages.py

@@ -1728,6 +1728,39 @@ class MessagingConfig(BaseModel):
     scilog: MessagingServiceScopeConfig
 
 
+class NotificationServiceTarget(BaseModel):
+    service_name: Literal["signal", "teams", "scilog"]
+    scope: str | list[str] | None = None
+
+
+class NotificationMessage(BECMessage):
+    """
+    Message for notification events that should be routed through configured
+    messaging services.
+
+    Args:
+        message (str): Notification body text.
+    """
+
+    msg_type: ClassVar[str] = "notification_message"
+    message: str
+
+
+class NotificationConfigMessage(BECMessage):
+    """
+    Routing configuration for notification events.
+
+    Args:
+        routes: Mapping of event name to messaging service targets.
+    """
+
+    msg_type: ClassVar[str] = "notification_config_message"
+    routes: dict[
+        Literal["new_scan", "scan_completed", "alarm", "scan_interlock"] | str,
+        list[NotificationServiceTarget],
+    ] = Field(default_factory=dict)
+
+
 AvailableMessagingServices = Annotated[
     Union[SignalServiceInfo, SciLogServiceInfo, TeamsServiceInfo],
     Field(discriminator="service_type"),

bec_lib/bec_lib/messaging_hooks.py

@@ -0,0 +1,114 @@
+from __future__ import annotations
+
+import enum
+from typing import TYPE_CHECKING, cast
+
+from bec_lib import messages
+from bec_lib.connector import MessageObject
+from bec_lib.endpoints import MessageEndpoints
+from bec_lib.logger import bec_logger
+from bec_lib.messaging_services import (
+    SciLogMessagingService,
+    SignalMessagingService,
+    TeamsMessagingService,
+)
+
+if TYPE_CHECKING:
+    from bec_lib.redis_connector import RedisConnector
+
+logger = bec_logger.logger
+
+
+class MessagingEvent(str, enum.Enum):
+    """
+    Enumeration of messaging events that can trigger configured hooks.
+    """
+
+    SCAN = "new_scan"
+    SCAN_COMPLETED = "scan_completed"
+    ALARM = "alarm"
+    SCAN_INTERLOCK = "scan_interlock"
+
+
+class MessagingManager:
+    """
+    Manage notification routing from internal events to concrete messaging
+    services.
+    """
+
+    def __init__(self, connector: RedisConnector):
+        self.connector = connector
+        self.config: dict[str, list[messages.NotificationServiceTarget]] = {}
+        signal = SignalMessagingService(self.connector)
+        scilog = SciLogMessagingService(self.connector)
+        teams = TeamsMessagingService(self.connector)
+        self._service_by_name = {"signal": signal, "scilog": scilog, "teams": teams}
+
+        self.connector.register(
+            patterns=MessageEndpoints.notification("*"), cb=self._handle_notification
+        )
+        self.connector.register(
+            topics=MessageEndpoints.notification_config(), cb=self._handle_notification_config
+        )
+
+        config_msg = self.connector.get(MessageEndpoints.notification_config())
+        if config_msg is not None:
+            self.on_notification_config(config_msg)
+
+    def _handle_notification(self, msg_obj: MessageObject[messages.NotificationMessage], **_kwargs):
+        prefix = MessageEndpoints.notification("").endpoint
+        event_type_str = msg_obj.topic.removeprefix(prefix)
+        self.on_notification(event_type_str, cast(messages.NotificationMessage, msg_obj.value))
+
+    def _handle_notification_config(
+        self, msg_obj: MessageObject[messages.NotificationConfigMessage], **_kwargs
+    ):
+        self.on_notification_config(cast(messages.NotificationConfigMessage, msg_obj.value))
+
+    def on_notification(self, event_type: str, message: messages.NotificationMessage) -> None:
+        """
+        Handle a notification event by routing it to the configured messaging services.
+
+        Args:
+            event_type(str): The type of the event that triggered the notification.
+            message(messages.NotificationMessage): The notification message containing the content to be sent.
+        """
+        routes = self.config.get(event_type, [])
+        for route in routes:
+            service = self._service_by_name.get(route.service_name)
+            if service is None:
+                logger.warning(f"Unknown messaging service: {route.service_name}")
+                continue
+            try:
+                logger.info(
+                    f"Routing notification for {event_type} to {route.service_name}: {message.message}"
+                )
+                service.new().add_text(message.message).send(scope=route.scope)
+            except RuntimeError as exc:
+                logger.warning(
+                    f"Failed to send notification for {event_type} via {route.service_name}: {exc}"
+                )
+
+    def on_notification_config(self, message: messages.NotificationConfigMessage) -> None:
+        """
+        Update the notification routing configuration based on the received message.
+
+        Args:
+            message(NotificationConfigMessage): The message containing the new routing configuration.
+        """
+        config: dict[str, list[messages.NotificationServiceTarget]] = {}
+        for event_name, targets in message.routes.items():
+            config[event_name] = targets
+        self.config = config
+
+    def shutdown(self) -> None:
+        """
+        Shutdown the messaging manager by unregistering all notification handlers.
+        """
+
+        self.connector.unregister(
+            patterns=MessageEndpoints.notification("*"), cb=self._handle_notification
+        )
+        self.connector.unregister(
+            topics=MessageEndpoints.notification_config(), cb=self._handle_notification_config
+        )

bec_lib/bec_lib/redis_connector.py

@@ -57,7 +57,9 @@
     DynamicMetricDict,
     DynamicMetricMessage,
     ErrorInfo,
+    NotificationMessage,
 )
+from bec_lib.messaging_hooks import MessagingEvent
 from bec_lib.serialization import MsgpackSerialization
 
 logger = bec_logger.logger
@@ -655,6 +657,23 @@ def raise_alarm(self, severity: Alarms, info: ErrorInfo, metadata: dict | None =
         """
         alarm_msg = AlarmMessage(severity=severity, info=info, metadata=metadata or {})
         self.set_and_publish(MessageEndpoints.alarm(), alarm_msg)
+        compact_message = info.compact_error_message or info.error_message or info.exception_type
+        self.notify(MessagingEvent.ALARM, compact_message)
+
+    def notify(self, event: MessagingEvent, message: str, pipe: Pipeline | None = None) -> None:
+        """
+        Publish a notification event for downstream routing by SciHub.
+
+        Args:
+            event(MessagingEvent): The type of the event that triggered the notification.
+            message(str): The notification message containing the content to be sent.
+            pipe(Pipeline, optional): Optional pipeline to enqueue the publish operation into.
+        """
+        self.send(
+            MessageEndpoints.notification(event.value),
+            NotificationMessage(message=message),
+            pipe=pipe,
+        )
 
     def pipeline(self) -> redis.client.Pipeline:
         """Create a new pipeline"""

bec_lib/bec_lib/tests/utils.py

@@ -561,6 +561,13 @@ def send(self, topic, msg, pipe=None):
             raise TypeError("Message must be a BECMessage")
         return self.raw_send(topic, msg, pipe)
 
+    def notify(self, event, message: str, pipe=None):
+        return self.send(
+            MessageEndpoints.notification(event.value),
+            messages.NotificationMessage(message=message),
+            pipe=pipe,
+        )
+
     def set_and_publish(self, topic, msg, pipe=None, expire: int = None):
         if pipe:
             pipe._pipe_buffer.append(("set_and_publish", (topic.endpoint, msg), {"expire": expire}))

bec_lib/tests/test_bec_messages.py

@@ -134,6 +134,31 @@ def test_ClientInfoMessage_raises():
         )
 
 
+def test_NotificationMessage():
+    msg = messages.NotificationMessage(message="Scan started")
+    res = MsgpackSerialization.dumps(msg)
+    res_loaded = MsgpackSerialization.loads(res)
+    assert res_loaded == msg
+
+
+def test_NotificationConfigMessage():
+    msg = messages.NotificationConfigMessage(
+        routes={
+            "new_scan": [
+                messages.NotificationServiceTarget(service_name="scilog", scope="logbook")
+            ],
+            "alarm": [
+                messages.NotificationServiceTarget(
+                    service_name="signal", scope=["+41791234567", "+41797654321"]
+                )
+            ],
+        }
+    )
+    res = MsgpackSerialization.dumps(msg)
+    res_loaded = MsgpackSerialization.loads(res)
+    assert res_loaded == msg
+
+
 def test_DeviceRPCMessage():
     msg = messages.DeviceRPCMessage(
         device="samx", return_val=1, out="done", success=True, metadata={"RID": "1234"}

bec_lib/tests/test_messaging_hooks.py

@@ -0,0 +1,86 @@
+from bec_lib import messages
+from bec_lib.endpoints import MessageEndpoints, MessageOp
+from bec_lib.messaging_hooks import MessagingEvent, MessagingManager
+
+
+def _available_services_message():
+    return messages.AvailableMessagingServicesMessage(
+        config=messages.MessagingConfig(
+            signal=messages.MessagingServiceScopeConfig(enabled=True, default=None),
+            teams=messages.MessagingServiceScopeConfig(enabled=False, default=None),
+            scilog=messages.MessagingServiceScopeConfig(enabled=True, default=None),
+        ),
+        deployment_services=[
+            messages.SciLogServiceInfo(
+                id="scilog-default", scope="logbook", enabled=True, logbook_id="lb-1"
+            )
+        ],
+        session_services=[],
+    )
+
+
+def test_notification_endpoints():
+    event_endpoint = MessageEndpoints.notification("new_scan")
+    config_endpoint = MessageEndpoints.notification_config()
+
+    assert event_endpoint.endpoint == "internal/messaging_services/notification/new_scan"
+    assert event_endpoint.message_type is messages.NotificationMessage
+    assert event_endpoint.message_op == MessageOp.SEND
+
+    assert config_endpoint.endpoint == "user/messaging_services/notification_config"
+    assert config_endpoint.message_type is messages.NotificationConfigMessage
+    assert config_endpoint.message_op == MessageOp.SET_PUBLISH
+
+
+def test_messaging_manager_loads_initial_config(connected_connector):
+    config_msg = messages.NotificationConfigMessage(
+        routes={
+            "new_scan": [messages.NotificationServiceTarget(service_name="scilog", scope="logbook")]
+        }
+    )
+    connected_connector.set_and_publish(MessageEndpoints.notification_config(), config_msg)
+
+    manager = MessagingManager(connected_connector)
+    try:
+        assert manager.config == {
+            MessagingEvent.SCAN: [
+                messages.NotificationServiceTarget(service_name="scilog", scope="logbook")
+            ]
+        }
+    finally:
+        manager.shutdown()
+
+
+def test_messaging_manager_routes_notifications_to_message_service_queue(connected_connector):
+    manager = MessagingManager(connected_connector)
+    try:
+        available_services = _available_services_message()
+        manager.scilog._on_new_scope_change_msg({"data": available_services})
+        manager.signal._on_new_scope_change_msg({"data": available_services})
+        manager.teams._on_new_scope_change_msg({"data": available_services})
+
+        manager.on_notification_config(
+            messages.NotificationConfigMessage(
+                routes={
+                    "new_scan": [
+                        messages.NotificationServiceTarget(service_name="scilog", scope="logbook")
+                    ]
+                }
+            )
+        )
+
+        manager.on_notification(
+            MessagingEvent.SCAN, messages.NotificationMessage(message="Scan started")
+        )
+
+        out = connected_connector.xread(MessageEndpoints.message_service_queue(), from_start=True)
+        assert len(out) == 1
+        sent_message = out[0]["data"]
+        assert sent_message.service_name == "scilog"
+        assert sent_message.scope == "logbook"
+        assert isinstance(sent_message.message[0], messages.MessagingServiceTextContent)
+        assert sent_message.message[0].content == "Scan started"
+        assert isinstance(sent_message.message[1], messages.MessagingServiceTagsContent)
+        assert sent_message.message[1].tags == ["bec"]
+    finally:
+        manager.shutdown()

bec_lib/tests/test_redis_connector.py

@@ -15,6 +15,7 @@
     ClientInfoMessage,
     ProcedureExecutionMessage,
 )
+from bec_lib.messaging_hooks import MessagingEvent
 from bec_lib.redis_connector import (
     IncompatibleMessageForEndpoint,
     IncompatibleRedisOperation,
@@ -71,7 +72,10 @@ def test_redis_connector_send_client_info(connector):
     ],
 )
 def test_redis_connector_raise_alarm(connector, severity, alarm_type, msg, compact_msg, metadata):
-    with mock.patch.object(connector, "set_and_publish", return_value=None):
+    with (
+        mock.patch.object(connector, "set_and_publish", return_value=None),
+        mock.patch.object(connector, "notify", return_value=None),
+    ):
         info = messages.ErrorInfo(
             error_message=msg, compact_error_message=compact_msg, exception_type=alarm_type
         )
@@ -80,6 +84,7 @@ def test_redis_connector_raise_alarm(connector, severity, alarm_type, msg, compa
         connector.set_and_publish.assert_called_once_with(
             MessageEndpoints.alarm(), AlarmMessage(severity=severity, info=info, metadata=metadata)
         )
+        connector.notify.assert_called_once_with(MessagingEvent.ALARM, compact_msg)
 
 
 @pytest.mark.parametrize(