UN-2022 [FEAT] Add co-owner management for Adapters, API Deployments, Connectors, Pipelines, Workflows, Prompt Studio

backend/adapter_processor_v2/migrations/0004_adapterinstance_co_owners.py

@@ -0,0 +1,24 @@
+# Generated by Django 4.2.1 on 2026-02-17 08:24
+
+from django.conf import settings
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+    dependencies = [
+        migrations.swappable_dependency(settings.AUTH_USER_MODEL),
+        ("adapter_processor_v2", "0003_mark_deprecated_adapters"),
+    ]
+
+    operations = [
+        migrations.AddField(
+            model_name="adapterinstance",
+            name="co_owners",
+            field=models.ManyToManyField(
+                blank=True,
+                help_text="Users with full ownership privileges",
+                related_name="co_owned_adapters",
+                to=settings.AUTH_USER_MODEL,
+            ),
+        ),
+    ]

backend/adapter_processor_v2/migrations/0005_backfill_creator_to_co_owners.py

@@ -0,0 +1,21 @@
+from django.db import migrations
+
+
+def backfill_creator_to_co_owners(apps, schema_editor):
+    adapter_instance_model = apps.get_model("adapter_processor_v2", "AdapterInstance")
+    for adapter in adapter_instance_model.objects.filter(created_by__isnull=False):
+        if not adapter.co_owners.filter(id=adapter.created_by_id).exists():
+            adapter.co_owners.add(adapter.created_by)
+
+
+class Migration(migrations.Migration):
+    dependencies = [
+        ("adapter_processor_v2", "0004_adapterinstance_co_owners"),
+    ]
+
+    operations = [
+        migrations.RunPython(
+            backfill_creator_to_co_owners,
+            reverse_code=migrations.RunPython.noop,
+        ),
+    ]

backend/adapter_processor_v2/models.py

@@ -37,7 +37,7 @@ def for_user(self, user: User) -> QuerySet[Any]:
         return (
             self.get_queryset()
             .filter(
-                models.Q(created_by=user)
+                models.Q(co_owners=user)
                 | models.Q(shared_users=user)
                 | models.Q(shared_to_org=True)
                 | models.Q(is_friction_less=True)
@@ -131,6 +131,12 @@ class AdapterInstance(DefaultOrganizationMixin, BaseModel):
     # Introduced field to establish M2M relation between users and adapters.
     # This will introduce intermediary table which relates both the models.
     shared_users = models.ManyToManyField(User, related_name="shared_adapters_instance")
+    co_owners = models.ManyToManyField(
+        User,
+        related_name="co_owned_adapters",
+        blank=True,
+        help_text="Users with full ownership privileges",
+    )
     description = models.TextField(blank=True, null=True, default=None)
 
     objects = AdapterInstanceModelManager()

backend/adapter_processor_v2/serializers.py

@@ -4,6 +4,7 @@
 from account_v2.serializer import UserSerializer
 from cryptography.fernet import Fernet
 from django.conf import settings
+from permissions.co_owner_serializers import CoOwnerRepresentationMixin
 from rest_framework import serializers
 from rest_framework.serializers import ModelSerializer
 
@@ -126,7 +127,7 @@ def get_context_window_size(self, obj: AdapterInstance) -> int:
         return obj.get_context_window_size()
 
 
-class AdapterListSerializer(BaseAdapterSerializer):
+class AdapterListSerializer(CoOwnerRepresentationMixin, BaseAdapterSerializer):
     """Inherits BaseAdapterSerializer.
 
     Used for listing adapters
@@ -175,10 +176,10 @@ def to_representation(self, instance: AdapterInstance) -> dict[str, str]:
         if model:
             rep["model"] = model
 
+        request = self.context.get("request")
+        self.add_co_owner_fields(instance, rep, request)
         if instance.is_friction_less:
             rep["created_by_email"] = "Unstract"
-        else:
-            rep["created_by_email"] = instance.created_by.email
 
         return rep
 
@@ -190,6 +191,7 @@ class SharedUserListSerializer(BaseAdapterSerializer):
     """
 
     shared_users = UserSerializer(many=True)
+    co_owners = UserSerializer(many=True, read_only=True)
     created_by = UserSerializer()
 
     class Meta(BaseAdapterSerializer.Meta):
@@ -200,6 +202,7 @@ class Meta(BaseAdapterSerializer.Meta):
             "adapter_name",
             "adapter_type",
             "created_by",
+            "co_owners",
             "shared_users",
             "shared_to_org",
         )  # type: ignore

backend/adapter_processor_v2/urls.py

@@ -25,6 +25,8 @@
 
 adapter_users = AdapterInstanceViewSet.as_view({"get": "list_of_shared_users"})
 adapter_info = AdapterInstanceViewSet.as_view({"get": "adapter_info"})
+adapter_add_owner = AdapterInstanceViewSet.as_view({"post": "add_co_owner"})
+adapter_remove_owner = AdapterInstanceViewSet.as_view({"delete": "remove_co_owner"})
 urlpatterns = format_suffix_patterns(
     [
         path("adapter_schema/", adapter_schema, name="get_adapter_schema"),
@@ -39,5 +41,15 @@
             adapter_users,
             name="adapter-users",
         ),
+        path(
+            "adapter/<uuid:pk>/owners/",
+            adapter_add_owner,
+            name="adapter-add-owner",
+        ),
+        path(
+            "adapter/<uuid:pk>/owners/<int:user_id>/",
+            adapter_remove_owner,
+            name="adapter-remove-owner",
+        ),
     ]
 )

backend/adapter_processor_v2/views.py

@@ -6,6 +6,7 @@
 from django.db.models import ProtectedError, QuerySet
 from django.http import HttpRequest
 from django.http.response import HttpResponse
+from permissions.co_owner_views import CoOwnerManagementMixin
 from permissions.permission import (
     IsFrictionLessAdapter,
     IsFrictionLessAdapterDelete,
@@ -135,8 +136,26 @@ def test(self, request: Request) -> Response:
         )
 
 
-class AdapterInstanceViewSet(ModelViewSet):
+class AdapterInstanceViewSet(CoOwnerManagementMixin, ModelViewSet):
     serializer_class = AdapterInstanceSerializer
+    notification_resource_name_field = "adapter_name"
+
+    def get_notification_resource_type(self, resource: Any) -> str | None:
+        try:
+            from plugins.notification.constants import ResourceType
+        except ImportError:
+            logger.debug(
+                "Notification plugin not available, skipping resource type lookup"
+            )
+            return None
+
+        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"]:
@@ -148,6 +167,9 @@ def get_permissions(self) -> list[Any]:
         elif self.action in ["list_of_shared_users", "adapter_info"]:
             return [IsOwnerOrSharedUserOrSharedToOrg()]
 
+        elif self.action in ["add_co_owner", "remove_co_owner"]:
+            return [IsOwner()]
+
         # Hack for friction-less onboarding
         # User cant view/update metadata but can delete/share etc
         return [IsOwner()]
@@ -162,7 +184,7 @@ def get_queryset(self) -> QuerySet | None:
             )
         else:
             queryset = AdapterInstance.objects.for_user(self.request.user)
-        return queryset
+        return queryset.prefetch_related("co_owners")
 
     def get_serializer_class(
         self,

backend/api_v2/api_deployment_views.py

@@ -6,6 +6,7 @@
 from configuration.models import Configuration
 from django.db.models import F, OuterRef, QuerySet, Subquery
 from django.http import HttpResponse
+from permissions.co_owner_views import CoOwnerManagementMixin
 from permissions.permission import IsOwner, IsOwnerOrSharedUserOrSharedToOrg
 from plugins import get_plugin
 from prompt_studio.prompt_studio_registry_v2.models import PromptStudioRegistry
@@ -17,7 +18,6 @@
 from tool_instance_v2.models import ToolInstance
 from utils.enums import CeleryTaskState
 from utils.hubspot_notify import notify_hubspot_event
-from utils.pagination import CustomPagination
 from workflow_manager.workflow_v2.dto import ExecutionResponse
 from workflow_manager.workflow_v2.models.execution import WorkflowExecution
 
@@ -245,11 +245,22 @@ def get(
         )
 
 
-class APIDeploymentViewSet(viewsets.ModelViewSet):
-    pagination_class = CustomPagination
+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", "partial_update", "update"]:
+        if self.action in [
+            "destroy",
+            "partial_update",
+            "update",
+            "add_co_owner",
+            "remove_co_owner",
+        ]:
             return [IsOwner()]
         return [IsOwnerOrSharedUserOrSharedToOrg()]
 
@@ -277,7 +288,7 @@ def get_queryset(self) -> QuerySet | None:
         if search:
             queryset = queryset.filter(display_name__icontains=search)
 
-        return queryset
+        return queryset.prefetch_related("co_owners")
 
     def get_serializer_class(self) -> serializers.Serializer:
         if self.action in ["list"]:
@@ -345,7 +356,9 @@ def by_prompt_studio_tool(self, request: Request) -> Response:
                 workflow_id__in=workflow_ids, created_by=request.user
             )
 
-            serializer = APIDeploymentListSerializer(deployments, many=True)
+            serializer = APIDeploymentListSerializer(
+                deployments, many=True, context={"request": request}
+            )
             return Response(serializer.data, status=status.HTTP_200_OK)
 
         except PromptStudioRegistry.DoesNotExist:

backend/api_v2/migrations/0004_apideployment_co_owners.py

@@ -0,0 +1,24 @@
+# Generated by Django 4.2.1 on 2026-02-17 08:24
+
+from django.conf import settings
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+    dependencies = [
+        migrations.swappable_dependency(settings.AUTH_USER_MODEL),
+        ("api_v2", "0003_add_organization_rate_limit"),
+    ]
+
+    operations = [
+        migrations.AddField(
+            model_name="apideployment",
+            name="co_owners",
+            field=models.ManyToManyField(
+                blank=True,
+                help_text="Users with full ownership privileges",
+                related_name="co_owned_api_deployments",
+                to=settings.AUTH_USER_MODEL,
+            ),
+        ),
+    ]

backend/api_v2/migrations/0005_backfill_creator_to_co_owners.py

@@ -0,0 +1,21 @@
+from django.db import migrations
+
+
+def backfill_creator_to_co_owners(apps, schema_editor):
+    api_deployment_model = apps.get_model("api_v2", "APIDeployment")
+    for deployment in api_deployment_model.objects.filter(created_by__isnull=False):
+        if not deployment.co_owners.filter(id=deployment.created_by_id).exists():
+            deployment.co_owners.add(deployment.created_by)
+
+
+class Migration(migrations.Migration):
+    dependencies = [
+        ("api_v2", "0004_apideployment_co_owners"),
+    ]
+
+    operations = [
+        migrations.RunPython(
+            backfill_creator_to_co_owners,
+            reverse_code=migrations.RunPython.noop,
+        ),
+    ]

backend/api_v2/models.py

@@ -27,14 +27,14 @@
 class APIDeploymentModelManager(DefaultOrganizationManagerMixin, models.Manager):
     def for_user(self, user):
         """Filter API deployments that the user can access:
-        - API deployments created by the user
+        - API deployments co-owned by the user
         - API deployments shared with the user
         - API deployments shared with the entire organization
         """
         from django.db.models import Q
 
         return self.filter(
-            Q(created_by=user)  # Owned by user
+            Q(co_owners=user)  # Co-owned by user
             | Q(shared_users=user)  # Shared with user
             | Q(shared_to_org=True)  # Shared to entire organization
         ).distinct()
@@ -96,6 +96,12 @@ class APIDeployment(DefaultOrganizationMixin, BaseModel):
     shared_users = models.ManyToManyField(
         User, related_name="shared_api_deployments", blank=True
     )
+    co_owners = models.ManyToManyField(
+        User,
+        related_name="co_owned_api_deployments",
+        blank=True,
+        help_text="Users with full ownership privileges",
+    )
     shared_to_org = models.BooleanField(
         default=False,
         db_comment="Whether this API deployment is shared with the entire organization",