UN-3769 [FEAT] Sortable resource lists with co-owner ownership

backend/adapter_processor_v2/serializers.py

@@ -215,6 +215,7 @@ def to_representation(self, instance: AdapterInstance) -> dict[str, str]:
         request = self.context.get("request")
         rep["is_owner"] = instance.is_owner(request.user) if request else False
         rep["co_owners_count"] = instance.co_owners_count()
+        rep["owner_email"] = instance.owner_email()
 
         return rep
 

backend/adapter_processor_v2/views.py

@@ -28,6 +28,7 @@
 from tenant_account_v2.organization_member_service import OrganizationMemberService
 from tool_instance_v2.models import ToolInstance
 from utils.filtering import FilterHelper
+from utils.list_query import apply_search_and_sort
 from utils.pagination import OptionalPagination
 from utils.user_context import UserContext
 
@@ -190,14 +191,16 @@ def get_queryset(self) -> QuerySet | None:
         ):
             queryset = queryset.filter(**filter_args)
 
-        search = self.request.query_params.get("search")
-        if search:
-            queryset = queryset.filter(adapter_name__icontains=search)
-
-        # Order by the DISTINCT ON field so pagination is deterministic and the
-        # admin/service branch (no distinct) is ordered too. Not modified_at:
-        # that would conflict with the DISTINCT ON in for_user().
-        return queryset.order_by("id")
+        # Owner-inclusive search + per-column sort (name/owner/created); re-wraps
+        # via pk__in to drop the DISTINCT ON in for_user() so any column sorts.
+        return apply_search_and_sort(
+            queryset,
+            model=AdapterInstance,
+            name_field="adapter_name",
+            request=self.request,
+            select_related=("created_by",),
+            prefetch_related=("memberships__user",),
+        )
 
     def get_serializer_class(
         self,

backend/connector_v2/serializers.py

@@ -176,6 +176,7 @@ def to_representation(self, instance: ConnectorInstance) -> dict[str, str]:
         request = self.context.get("request")
         rep["is_owner"] = instance.is_owner(request.user) if request else False
         rep["co_owners_count"] = instance.co_owners_count()
+        rep["owner_email"] = instance.owner_email()
 
         return rep
 

backend/connector_v2/views.py

@@ -21,6 +21,7 @@
 from rest_framework.versioning import URLPathVersioning
 from tenant_account_v2.organization_member_service import OrganizationMemberService
 from utils.filtering import FilterHelper
+from utils.list_query import apply_search_and_sort
 from utils.pagination import OptionalPagination
 from utils.user_context import UserContext
 
@@ -105,10 +106,6 @@ def get_queryset(self) -> QuerySet | None:
         if filter_args:
             queryset = queryset.filter(**filter_args)
 
-        search = self.request.query_params.get("search")
-        if search:
-            queryset = queryset.filter(connector_name__icontains=search)
-
         # Filter by connector_mode
         connector_mode_param = self.request.query_params.get("connector_mode")
         if connector_mode_param:
@@ -127,10 +124,16 @@ def get_queryset(self) -> QuerySet | None:
                 )
                 queryset = queryset.none()
 
-        # Order by the DISTINCT ON field so pagination is deterministic and the
-        # admin/service branch (no distinct) is ordered too. Not modified_at:
-        # that would conflict with the DISTINCT ON in for_user().
-        return queryset.order_by("id")
+        # Owner-inclusive search + per-column sort (name/owner/created); re-wraps
+        # via pk__in to drop the DISTINCT ON in for_user() so any column sorts.
+        return apply_search_and_sort(
+            queryset,
+            model=ConnectorInstance,
+            name_field="connector_name",
+            request=self.request,
+            select_related=("created_by",),
+            prefetch_related=("memberships__user",),
+        )
 
     def _get_connector_metadata(self, connector_id: str) -> dict[str, str] | None:
         """Gets connector metadata for the ConnectorInstance.

backend/permissions/models.py

@@ -43,6 +43,21 @@ def co_owners_count(self) -> int:
             for m in self.memberships.all()  # type: ignore[attr-defined]
         )
 
+    def owner_email(self) -> str | None:
+        # Email for the "Owned By" label. ``created_by`` is audit-only (UN-2202)
+        # and the creator can be removed as owner, so it must not name the owner.
+        # Reads the prefetched ``memberships`` (list views set ``memberships__user``)
+        # to stay query-free; earliest live OWNER wins so the label names the same
+        # roster as ``co_owners_count()`` and is stable across page loads.
+        owners = [
+            m
+            for m in self.memberships.all()  # type: ignore[attr-defined]
+            if m.role == ResourceRole.OWNER and not m.user.is_service_account
+        ]
+        if not owners:
+            return None
+        return min(owners, key=lambda m: m.created_at).user.email
+
     def is_owner(self, user: Any) -> bool:
         if user is None:
             return False

backend/prompt_studio/prompt_studio_core_v2/serializers.py

@@ -48,6 +48,7 @@ class CustomToolListSerializer(serializers.ModelSerializer):
     prompt_count = serializers.SerializerMethodField()
     is_owner = serializers.SerializerMethodField()
     co_owners_count = serializers.SerializerMethodField()
+    owner_email = serializers.SerializerMethodField()
 
     class Meta:
         model = CustomTool
@@ -65,6 +66,7 @@ class Meta:
             "prompt_count",
             "is_owner",
             "co_owners_count",
+            "owner_email",
         ]
 
     def get_created_by_email(self, instance):
@@ -77,6 +79,9 @@ def get_is_owner(self, instance) -> bool:
     def get_co_owners_count(self, instance) -> int:
         return instance.co_owners_count()
 
+    def get_owner_email(self, instance) -> str | None:
+        return instance.owner_email()
+
     def get_prompt_count(self, instance):
         if hasattr(instance, "_prompt_count"):
             return instance._prompt_count or 0

backend/prompt_studio/prompt_studio_core_v2/views.py

@@ -32,6 +32,7 @@
 from tool_instance_v2.models import ToolInstance
 from utils.file_storage.helpers.prompt_studio_file_helper import PromptStudioFileHelper
 from utils.hubspot_notify import notify_hubspot_event
+from utils.list_query import apply_search_and_sort
 from utils.pagination import OptionalPagination
 from utils.user_context import UserContext
 from utils.user_session import UserSessionUtils
@@ -161,27 +162,31 @@ def get_queryset(self) -> QuerySet | None:
             "memberships__user"
         )
         if self.action == "list":
-            # Subquery avoids conflict with distinct("tool_id") from for_user()
+            # Owner-inclusive search + per-column sort (name/owner/created);
+            # re-wraps via pk__in to drop the DISTINCT ON in for_user() so any
+            # column sorts.
+            qs = apply_search_and_sort(
+                qs,
+                model=CustomTool,
+                name_field="tool_name",
+                request=self.request,
+                select_related=("created_by",),
+                prefetch_related=("memberships__user",),
+            )
+            # apply_search_and_sort returns a fresh CustomTool.objects chain, so
+            # the prompt_count annotation goes on afterwards. Subquery keeps the
+            # count out of the outer GROUP BY, which would otherwise collide with
+            # the sort column.
             prompt_count_sq = (
                 ToolStudioPrompt.objects.filter(tool_id=OuterRef("pk"))
                 .order_by()
                 .values("tool_id")
                 .annotate(cnt=Count("prompt_id"))
                 .values("cnt")
             )
-            # modified_at needs no annotation: prompt writes bump the parent
-            # row at the source (ToolStudioPrompt.save/delete, sync_prompts),
-            # keeping the plain field orderable. Only prompt writes bump —
-            # profile/document edits and queryset-level prompt writes do not;
-            # any new write path must bump CustomTool itself
-            qs = qs.select_related("created_by").annotate(
-                _prompt_count=Subquery(prompt_count_sq),
-            )
-            search = self.request.query_params.get("search")
-            if search:
-                qs = qs.filter(tool_name__icontains=search)
-        # Order by the DISTINCT ON field so pagination is deterministic and the
-        # admin/service branch (no distinct) is ordered too.
+            return qs.annotate(_prompt_count=Subquery(prompt_count_sq))
+        # Order by the DISTINCT ON field so pagination is deterministic for the
+        # admin/service (non-list) branch.
         return qs.order_by("tool_id")
 
     def get_object(self):

backend/utils/list_query.py

@@ -0,0 +1,75 @@
+"""Shared list-query helpers for resource list endpoints.
+
+Provides owner-inclusive ``?search`` and per-column sorting (name / owner /
+created) with a ``pk__in`` re-wrap so managers ending in Postgres ``DISTINCT
+ON`` can still be ordered by an arbitrary column.
+"""
+
+from typing import Any
+
+from django.db.models import Model, Q, QuerySet
+from rest_framework.request import Request
+
+# ``name`` maps to the resource-specific name field each caller passes;
+# owner/created are shared across every list endpoint.
+OWNER_SORT_FIELD = "created_by__email"
+CREATED_SORT_FIELD = "created_at"
+
+
+def apply_search_and_sort(
+    queryset: QuerySet[Any],
+    *,
+    model: type[Model],
+    name_field: str,
+    request: Request,
+    select_related: tuple[str, ...] = (),
+    prefetch_related: tuple[str, ...] = (),
+    default_sort_by: str = "name",
+) -> QuerySet[Any]:
+    """Apply owner-inclusive ``?search`` and ``?sort_by``/``?order`` to a list
+    queryset.
+
+    ``sort_by`` is ``name`` | ``owner`` | ``created`` (default ``name``);
+    ``order`` is ``asc`` | ``desc`` (default ``asc``). The queryset is re-wrapped
+    via ``pk__in`` to drop any ``DISTINCT ON`` (so ordering by a non-distinct
+    column is legal) and a ``pk`` tiebreaker is appended for stable pagination.
+    ``select_related`` / ``prefetch_related`` are re-attached to the re-wrapped
+    queryset to keep the list free of N+1 owner/co-owner lookups.
+
+    Args:
+        queryset: The already org-scoped, ``for_user``-filtered list queryset.
+        model: The concrete resource model, used to re-wrap via ``pk__in``.
+        name_field: The resource's name column (e.g. ``adapter_name``).
+        request: DRF request carrying ``search`` / ``sort_by`` / ``order``.
+        select_related: FK joins to re-attach after the re-wrap.
+        prefetch_related: Reverse/M2M prefetches to re-attach after the re-wrap.
+        default_sort_by: Sort key used when ``?sort_by`` is absent.
+
+    Returns:
+        An ordered queryset ready for pagination.
+    """
+    params = request.query_params
+
+    search = params.get("search")
+    if search:
+        queryset = queryset.filter(
+            Q(**{f"{name_field}__icontains": search})
+            | Q(**{f"{OWNER_SORT_FIELD}__icontains": search})
+        )
+
+    sort_field = {
+        "name": name_field,
+        "owner": OWNER_SORT_FIELD,
+        "created": CREATED_SORT_FIELD,
+    }.get((params.get("sort_by") or default_sort_by).lower(), name_field)
+    order_prefix = "-" if (params.get("order") or "asc").lower() == "desc" else ""
+
+    # Ordering the source by ``pk`` keeps the DISTINCT ON (always the pk) valid
+    # while stripping the model's default ordering, so the outer query is free
+    # to sort by any column.
+    rewrapped = model.objects.filter(pk__in=queryset.order_by("pk").values("pk"))
+    if select_related:
+        rewrapped = rewrapped.select_related(*select_related)
+    if prefetch_related:
+        rewrapped = rewrapped.prefetch_related(*prefetch_related)
+    return rewrapped.order_by(f"{order_prefix}{sort_field}", "pk")

backend/workflow_manager/workflow_v2/serializers.py

@@ -85,6 +85,7 @@ def to_representation(self, instance: Workflow) -> dict[str, str]:
         request = self.context.get("request")
         representation["is_owner"] = instance.is_owner(request.user) if request else False
         representation["co_owners_count"] = instance.co_owners_count()
+        representation["owner_email"] = instance.owner_email()
         return representation
 
     def create(self, validated_data: dict[str, Any]) -> Any:

backend/workflow_manager/workflow_v2/views.py

@@ -21,6 +21,7 @@
 from rest_framework.versioning import URLPathVersioning
 from rest_framework.views import APIView
 from utils.filtering import FilterHelper
+from utils.list_query import apply_search_and_sort
 from utils.organization_utils import filter_queryset_by_organization, resolve_organization
 from utils.pagination import OptionalPagination
 
@@ -110,24 +111,18 @@ def get_queryset(self) -> QuerySet:
             if filter_args
             else Workflow.objects.for_user(self.request.user)
         )
-        # Avoid per-row queries for owner/co-owner + creator fields in list views
-        queryset = queryset.select_related("created_by").prefetch_related(
-            "memberships__user"
-        )
-
-        search = self.request.query_params.get("search")
-        if search:
-            queryset = queryset.filter(workflow_name__icontains=search)
 
-        # `id` tiebreaker keeps ordering deterministic across paginated requests
-        # (the for_user() manager uses plain .distinct(), so there is no default)
-        order_by = self.request.query_params.get("order_by")
-        if order_by == "asc":
-            queryset = queryset.order_by("modified_at", "id")
-        else:
-            queryset = queryset.order_by("-modified_at", "id")
-
-        return queryset
+        # Owner-inclusive search + per-column sort (name/owner/created); re-wraps
+        # via pk__in (harmless here — for_user() uses plain .distinct()) and
+        # re-attaches the owner/co-owner joins to avoid N+1 in list views.
+        return apply_search_and_sort(
+            queryset,
+            model=Workflow,
+            name_field="workflow_name",
+            request=self.request,
+            select_related=("created_by",),
+            prefetch_related=("memberships__user",),
+        )
 
     def get_serializer_class(self) -> serializers.Serializer:
         if self.action == "execute":