Un 3290 email notification on co owner addition

backend/adapter_processor_v2/views.py

@@ -138,6 +138,18 @@ def test(self, request: Request) -> Response:
 
 class AdapterInstanceViewSet(CoOwnerManagementMixin, ModelViewSet):
     serializer_class = AdapterInstanceSerializer
+    notification_resource_name_field = "adapter_name"
+
+    def get_notification_resource_type(self, resource: Any) -> str | None:
+        from plugins.notification.constants import ResourceType
+
+        adapter_type_to_resource = {
+            "LLM": ResourceType.LLM.value,
+            "EMBEDDING": ResourceType.EMBEDDING.value,
+            "VECTOR_DB": ResourceType.VECTOR_DB.value,
+            "X2TEXT": ResourceType.X2TEXT.value,
+        }
+        return adapter_type_to_resource.get(resource.adapter_type)
 
     def get_permissions(self) -> list[Any]:
         if self.action in ["update", "retrieve"]:

backend/api_v2/api_deployment_views.py

@@ -244,6 +244,13 @@ def get(
 
 
 class APIDeploymentViewSet(CoOwnerManagementMixin, viewsets.ModelViewSet):
+    notification_resource_name_field = "display_name"
+
+    def get_notification_resource_type(self, resource: Any) -> str | None:
+        from plugins.notification.constants import ResourceType
+
+        return ResourceType.API_DEPLOYMENT.value  # type: ignore
+
     def get_permissions(self) -> list[Any]:
         if self.action in [
             "destroy",

backend/connector_v2/views.py

@@ -38,6 +38,12 @@
 class ConnectorInstanceViewSet(CoOwnerManagementMixin, viewsets.ModelViewSet):
     versioning_class = URLPathVersioning
     serializer_class = ConnectorInstanceSerializer
+    notification_resource_name_field = "connector_name"
+
+    def get_notification_resource_type(self, resource: Any) -> str | None:
+        from plugins.notification.constants import ResourceType
+
+        return ResourceType.CONNECTOR.value  # type: ignore
 
     def get_permissions(self) -> list[Any]:
         if self.action in ["update", "destroy", "partial_update"]:

backend/permissions/co_owner_views.py

@@ -21,8 +21,60 @@ class CoOwnerManagementMixin:
     Adds:
         - POST <pk>/owners/     -> add_co_owner
         - DELETE <pk>/owners/<user_id>/ -> remove_co_owner
+
+    Subclasses can opt in to co-owner email notifications by setting:
+        notification_resource_name_field = "<model_field>"
+    and overriding:
+        get_notification_resource_type(resource) -> str | None
     """
 
+    notification_resource_name_field: str | None = None
+
+    def get_notification_resource_type(self, resource: Any) -> str | None:
+        """Return the ResourceType value for notifications, or None to skip."""
+        return None
+
+    def _send_co_owner_notification(
+        self, resource: Any, user: User, request: Request
+    ) -> None:
+        """Send an email notification to a newly added co-owner.
+
+        Does nothing when the notification plugin is not installed (OSS)
+        or when the ViewSet has not opted in.
+        """
+        from plugins import get_plugin
+
+        plugin = get_plugin("notification")
+        if not plugin:
+            return
+
+        resource_type = self.get_notification_resource_type(resource)
+        if not resource_type:
+            return
+
+        name_field = self.notification_resource_name_field
+        if not name_field:
+            return
+
+        resource_name = getattr(resource, name_field, None) or ""
+
+        try:
+            service = plugin["service_class"]()
+            service.send_sharing_notification(
+                resource_type=resource_type,
+                resource_name=resource_name,
+                resource_id=str(resource.pk),
+                shared_by=request.user,
+                shared_to=[user],
+                resource_instance=resource,
+            )
+        except Exception:
+            logger.exception(
+                "Failed to send co-owner notification for %s %s",
+                resource.__class__.__name__,
+                resource.pk,
+            )
+
     @action(detail=True, methods=["post"], url_path="owners")
     def add_co_owner(self, request: Request, pk: Any = None) -> Response:
         """Add a co-owner to the resource."""
@@ -44,6 +96,8 @@ def add_co_owner(self, request: Request, pk: Any = None) -> Response:
             request.user.email,
         )
 
+        self._send_co_owner_notification(resource, user, request)
+
         co_owners = [{"id": u.id, "email": u.email} for u in resource.co_owners.all()]
         return Response(
             {"id": str(resource.pk), "co_owners": co_owners},

backend/permissions/tests/test_co_owners.py

@@ -5,6 +5,7 @@
 
 from django.test import RequestFactory, TestCase
 from permissions.co_owner_serializers import AddCoOwnerSerializer, RemoveCoOwnerSerializer
+from permissions.co_owner_views import CoOwnerManagementMixin
 from permissions.permission import (
     IsOwner,
     IsOwnerOrSharedUser,
@@ -362,3 +363,114 @@ def test_save_removes_creator_without_promotion(self) -> None:
 
         resource.co_owners.remove.assert_called_once_with(creator)
         resource.save.assert_not_called()
+
+
+class TestCoOwnerNotification(TestCase):
+    """Tests for co-owner addition email notifications."""
+
+    def _make_mixin(
+        self, resource_name_field: str | None = None, resource_type: str | None = None
+    ) -> CoOwnerManagementMixin:
+        """Create a mixin instance with optional notification opt-in."""
+        mixin = CoOwnerManagementMixin()
+        if resource_name_field is not None:
+            mixin.notification_resource_name_field = resource_name_field
+        if resource_type is not None:
+            # Override the method to return the given type
+            mixin.get_notification_resource_type = lambda self_unused: resource_type  # type: ignore[assignment]
+        return mixin
+
+    def _make_resource(self, name_value: str = "My Resource") -> Mock:
+        resource = Mock()
+        resource.pk = uuid4()
+        resource.__class__ = type("FakeModel", (), {})
+        resource.workflow_name = name_value
+        return resource
+
+    def _make_request(self) -> Mock:
+        request = Mock()
+        request.user = Mock()
+        request.user.email = "actor@example.com"
+        return request
+
+    @patch("plugins.get_plugin")
+    def test_add_co_owner_sends_notification(self, mock_get_plugin: Mock) -> None:
+        """Notification is sent when plugin is available and ViewSet opts in."""
+        mock_service_instance = Mock()
+        mock_get_plugin.return_value = {"service_class": Mock(return_value=mock_service_instance)}
+
+        mixin = self._make_mixin(
+            resource_name_field="workflow_name",
+            resource_type="workflow",
+        )
+        resource = self._make_resource("Test Workflow")
+        user = Mock()
+        user.email = "newowner@example.com"
+        request = self._make_request()
+
+        mixin._send_co_owner_notification(resource, user, request)
+
+        mock_service_instance.send_sharing_notification.assert_called_once_with(
+            resource_type="workflow",
+            resource_name="Test Workflow",
+            resource_id=str(resource.pk),
+            shared_by=request.user,
+            shared_to=[user],
+            resource_instance=resource,
+        )
+
+    @patch("plugins.get_plugin")
+    def test_add_co_owner_no_notification_when_plugin_missing(
+        self, mock_get_plugin: Mock
+    ) -> None:
+        """No-op when notification plugin is not installed (OSS)."""
+        mock_get_plugin.return_value = {}
+
+        mixin = self._make_mixin(
+            resource_name_field="workflow_name",
+            resource_type="workflow",
+        )
+        resource = self._make_resource()
+        user = Mock()
+        request = self._make_request()
+
+        # Should not raise
+        mixin._send_co_owner_notification(resource, user, request)
+
+    @patch("plugins.get_plugin")
+    def test_add_co_owner_notification_failure_does_not_break(
+        self, mock_get_plugin: Mock
+    ) -> None:
+        """Notification errors are logged but do not propagate."""
+        mock_service_instance = Mock()
+        mock_service_instance.send_sharing_notification.side_effect = RuntimeError("email down")
+        mock_get_plugin.return_value = {"service_class": Mock(return_value=mock_service_instance)}
+
+        mixin = self._make_mixin(
+            resource_name_field="workflow_name",
+            resource_type="workflow",
+        )
+        resource = self._make_resource()
+        user = Mock()
+        request = self._make_request()
+
+        # Should not raise
+        mixin._send_co_owner_notification(resource, user, request)
+
+    @patch("plugins.get_plugin")
+    def test_add_co_owner_no_notification_when_resource_type_none(
+        self, mock_get_plugin: Mock
+    ) -> None:
+        """No notification when ViewSet does not opt in (returns None)."""
+        mock_get_plugin.return_value = {"service_class": Mock()}
+
+        # Default mixin — no opt-in
+        mixin = CoOwnerManagementMixin()
+        resource = self._make_resource()
+        user = Mock()
+        request = self._make_request()
+
+        mixin._send_co_owner_notification(resource, user, request)
+
+        # service_class should never have been instantiated
+        mock_get_plugin.return_value["service_class"].assert_not_called()

backend/pipeline_v2/views.py

@@ -44,6 +44,16 @@
 class PipelineViewSet(CoOwnerManagementMixin, viewsets.ModelViewSet):
     versioning_class = URLPathVersioning
     queryset = Pipeline.objects.all()
+    notification_resource_name_field = "pipeline_name"
+
+    def get_notification_resource_type(self, resource: Any) -> str | None:
+        from plugins.notification.constants import ResourceType
+
+        if resource.pipeline_type == ResourceType.ETL.value:
+            return ResourceType.ETL.value  # type: ignore
+        if resource.pipeline_type == ResourceType.TASK.value:
+            return ResourceType.TASK.value  # type: ignore
+        return None
 
     def get_permissions(self) -> list[Any]:
         if self.action in [

backend/prompt_studio/prompt_studio_core_v2/views.py

@@ -88,6 +88,12 @@ class PromptStudioCoreView(CoOwnerManagementMixin, viewsets.ModelViewSet):
     versioning_class = URLPathVersioning
 
     serializer_class = CustomToolSerializer
+    notification_resource_name_field = "tool_name"
+
+    def get_notification_resource_type(self, resource: Any) -> str | None:
+        from plugins.notification.constants import ResourceType
+
+        return ResourceType.TEXT_EXTRACTOR.value  # type: ignore
 
     def get_permissions(self) -> list[Any]:
         if self.action == "destroy":

backend/workflow_manager/workflow_v2/views.py

@@ -72,6 +72,12 @@ def make_execution_response(response: ExecutionResponse) -> Any:
 
 class WorkflowViewSet(CoOwnerManagementMixin, viewsets.ModelViewSet):
     versioning_class = URLPathVersioning
+    notification_resource_name_field = "workflow_name"
+
+    def get_notification_resource_type(self, resource: Any) -> str | None:
+        from plugins.notification.constants import ResourceType
+
+        return ResourceType.WORKFLOW.value  # type: ignore
 
     def get_permissions(self) -> list[Any]:
         if self.action in [