UN-3853 [FIX] Attribute platform-key-created resources to the API key's creator

backend/adapter_processor_v2/views.py

@@ -16,6 +16,7 @@
 )
 from permissions.resource_share_views import ResourceShareManagementMixin
 from permissions.roles import ResourceRole
+from platform_api.services import owner_user_for
 from plugins import get_plugin
 from rest_framework import status
 from rest_framework.decorators import action
@@ -272,7 +273,8 @@ def create(self, request: Any) -> Response:
             # ``created_by`` is audit-only; the creator's access flows through
             # an OWNER membership row (UN-2202 co-owners).
             instance.memberships.get_or_create(
-                user_id=request.user.id, defaults={"role": ResourceRole.OWNER}
+                user_id=owner_user_for(request.user).id,
+                defaults={"role": ResourceRole.OWNER},
             )
             organization_member = OrganizationMemberService.get_user_by_id(
                 request.user.id

backend/api_v2/api_deployment_views.py

@@ -11,6 +11,7 @@
 from permissions.resource_share_views import ResourceShareManagementMixin
 from permissions.roles import ResourceRole
 from platform_api.openapi_schema import PlatformKeyAutoSchema
+from platform_api.services import owner_user_for
 from plugins import get_plugin
 from prompt_studio.prompt_studio_registry_v2.models import PromptStudioRegistry
 from rest_framework import serializers, status, views, viewsets
@@ -361,7 +362,8 @@ def create(
         # ``created_by`` is audit-only; the creator's access flows through an
         # OWNER membership row (UN-2202 co-owners).
         serializer.instance.memberships.get_or_create(
-            user_id=request.user.id, defaults={"role": ResourceRole.OWNER}
+            user_id=owner_user_for(request.user).id,
+            defaults={"role": ResourceRole.OWNER},
         )
         api_key = DeploymentHelper.create_api_key(serializer=serializer, request=request)
         response_serializer = DeploymentResponseSerializer(

backend/api_v2/serializers.py

@@ -510,6 +510,7 @@ class APIDeploymentListSerializer(ModelSerializer):
     last_run_time = SerializerMethodField()
     is_owner = SerializerMethodField()
     co_owners_count = SerializerMethodField()
+    owner_emails = SerializerMethodField()
 
     class Meta:
         model = APIDeployment
@@ -529,6 +530,7 @@ class Meta:
             "last_run_time",
             "is_owner",
             "co_owners_count",
+            "owner_emails",
         ]
 
     def get_created_by_email(self, obj) -> str | None:
@@ -542,6 +544,11 @@ def get_is_owner(self, obj) -> bool:
     def get_co_owners_count(self, obj) -> int:
         return obj.co_owners_count()
 
+    def get_owner_emails(self, obj) -> list[str]:
+        # Names the actual owner in "Owned By"; ``created_by`` is audit-only
+        # (UN-2202) and stays the service account on platform-key creates.
+        return obj.owner_emails()
+
     # Both read the list view's annotations when they are there, and fall back
     # to a query for the callers that serialize a plain queryset. A deployment
     # that has never run annotates to `None`, so absence is what decides, not

backend/connector_v2/views.py

@@ -12,6 +12,7 @@
 from permissions.permission import IsOwner, IsOwnerOrSharedUserOrSharedToOrg
 from permissions.resource_share_views import ResourceShareManagementMixin
 from permissions.roles import ResourceRole
+from platform_api.services import owner_user_for
 from plugins import get_plugin
 from rest_framework import status, viewsets
 from rest_framework.decorators import action
@@ -259,7 +260,8 @@ def create(self, request: Any) -> Response:
         # ``created_by`` is audit-only; the creator's access flows through an
         # OWNER membership row (UN-2202 co-owners).
         serializer.instance.memberships.get_or_create(
-            user_id=request.user.id, defaults={"role": ResourceRole.OWNER}
+            user_id=owner_user_for(request.user).id,
+            defaults={"role": ResourceRole.OWNER},
         )
         headers = self.get_success_headers(serializer.data)
         return Response(serializer.data, status=status.HTTP_201_CREATED, headers=headers)

backend/pipeline_v2/serializers/crud.py

@@ -32,6 +32,7 @@ class PipelineSerializer(IntegrityErrorMixin, AuditSerializer):
     next_run_time = SerializerMethodField()
     is_owner = SerializerMethodField()
     co_owners_count = SerializerMethodField()
+    owner_emails = SerializerMethodField()
     # ``shared_groups`` is no longer an M2M on Pipeline — declare it
     # explicitly so ``fields = "__all__"`` continues to expose it. Share
     # mutations go through ``POST /pipeline/{id}/share/`` (UN-2977 plan §B).
@@ -224,6 +225,11 @@ def get_is_owner(self, obj) -> bool:
     def get_co_owners_count(self, obj) -> int:
         return obj.co_owners_count()
 
+    def get_owner_emails(self, obj) -> list[str]:
+        # Names the actual owner in "Owned By"; ``created_by`` is audit-only
+        # (UN-2202) and stays the service account on platform-key creates.
+        return obj.owner_emails()
+
     def get_last_5_run_statuses(self, instance: Pipeline) -> list[dict]:
         """Fetch the last 5 execution statuses with timestamps for this pipeline."""
         return WorkflowExecution.get_last_run_statuses(instance.id, limit=5)

backend/pipeline_v2/views.py

@@ -13,6 +13,7 @@
 from permissions.permission import IsOwner, IsOwnerOrSharedUserOrSharedToOrg
 from permissions.resource_share_views import ResourceShareManagementMixin
 from permissions.roles import ResourceRole
+from platform_api.services import owner_user_for
 from plugins import get_plugin
 from rest_framework import serializers, status, viewsets
 from rest_framework.decorators import action
@@ -159,7 +160,8 @@ def create(self, request: Request) -> Response:
                 # Grant before the API key so the creator's access is committed
                 # with the row itself, matching api_deployment_views.create().
                 pipeline_instance.memberships.get_or_create(
-                    user_id=request.user.id, defaults={"role": ResourceRole.OWNER}
+                    user_id=owner_user_for(request.user).id,
+                    defaults={"role": ResourceRole.OWNER},
                 )
                 # Create API key using the created instance
                 KeyHelper.create_api_key(pipeline_instance, request)

backend/platform_api/services.py

@@ -15,6 +15,10 @@
 
     from platform_api.models import PlatformApiKey
 
+# Reserved domain for service-account addresses. The frontend matches on it to
+# label an ownerless resource "Platform key" instead of naming a machine.
+SERVICE_ACCOUNT_EMAIL_DOMAIN = "platform.internal"
+
 # Business app labels whose models may carry created_by / membership rows.
 # Restricts transfer_ownership to avoid scanning Django built-in and third-party models.
 _BUSINESS_APP_LABELS = {
@@ -45,7 +49,7 @@ def create_api_user_for_key(
         name_slug = _slugify_for_email(platform_api_key.name)
         user = User(
             username=f"svc-{name_slug}-{uid[:8]}",
-            email=f"{name_slug}-{uid[:8]}@platform.internal",
+            email=f"{name_slug}-{uid[:8]}@{SERVICE_ACCOUNT_EMAIL_DOMAIN}",
             user_id=uid,
             is_service_account=True,
         )
@@ -63,6 +67,37 @@ def create_api_user_for_key(
     return user
 
 
+def owner_user_for(user: User) -> User:
+    """Resolve the human who should own a resource created by ``user``.
+
+    A platform key authenticates as a service account, and service accounts are
+    filtered out of every owner surface (``HasMembersMixin``), so a resource
+    granted to one has no human owner: it is invisible to its creator and only
+    an org admin can manage it. Attribute it to the key's creator instead — the
+    same successor :func:`delete_api_user_for_key` already hands ownership to.
+
+    Returns ``user`` unchanged for a normal session, and for the residual case
+    where the key's creator has since been deleted (``created_by`` is
+    ``SET_NULL``) — such a resource stays deliberately ownerless and the UI
+    labels it "Platform key".
+
+    Org membership of the creator is deliberately not re-checked: a key can
+    outlive its creator's membership, and granting to an ex-member matches what
+    :func:`delete_api_user_for_key` already does. The row is inert until they
+    rejoin, which beats leaving the resource with no owner at all.
+    """
+    if not getattr(user, "is_service_account", False):
+        return user
+
+    # Imported here so the module keeps its models import behind TYPE_CHECKING.
+    from platform_api.models import PlatformApiKey
+
+    key = (
+        PlatformApiKey.objects.filter(api_user=user).select_related("created_by").first()
+    )
+    return key.created_by if key and key.created_by else user
+
+
 def _get_user_fk_fields(model: type) -> list[str]:
     """Return names of all ForeignKey fields pointing to User."""
     return [

backend/platform_api/tests/test_owner_user_for.py

@@ -0,0 +1,74 @@
+"""The `owner_user_for` resolver in isolation.
+
+End-to-end coverage of the create sites that call it lives in
+`test_platform_key_resource_ownership.py`.
+"""
+
+import secrets
+import uuid
+
+from account_v2.models import Organization, User
+from django.db import connection
+from django.test.utils import CaptureQueriesContext
+from platform_api.models import ApiKeyPermission, PlatformApiKey
+from platform_api.services import create_api_user_for_key, owner_user_for
+from rest_framework.test import APITestCase
+
+ORG = "org-owner-test"
+
+
+def _make_user() -> User:
+    email = f"user-{uuid.uuid4().hex[:8]}@example.com"
+    return User.objects.create_user(
+        username=email, email=email, password=secrets.token_urlsafe()
+    )
+
+
+class OwnerUserForTest(APITestCase):
+    """The resolver in isolation."""
+
+    def setUp(self) -> None:
+        self.org = Organization.objects.create(
+            name=ORG, display_name="Owner Test", organization_id=ORG
+        )
+
+    def _make_key(self, created_by: User | None) -> PlatformApiKey:
+        key = PlatformApiKey.objects.create(
+            name=f"key-{uuid.uuid4().hex[:8]}",
+            description="test key",
+            organization=self.org,
+            permission=ApiKeyPermission.FULL_ACCESS,
+            created_by=created_by,
+        )
+        # The minting path, for the `is_service_account` flag it sets.
+        create_api_user_for_key(key, self.org)
+        key.refresh_from_db()
+        return key
+
+    def test_a_normal_user_is_returned_unchanged(self) -> None:
+        user = _make_user()
+        self.assertEqual(owner_user_for(user), user)
+
+    def test_a_normal_user_costs_no_query(self) -> None:
+        """The early return is the hot path — every create site calls this."""
+        user = _make_user()
+        with CaptureQueriesContext(connection) as queries:
+            owner_user_for(user)
+        self.assertEqual(len(queries), 0)
+
+    def test_a_service_account_resolves_to_the_keys_creator(self) -> None:
+        creator = _make_user()
+        key = self._make_key(created_by=creator)
+        self.assertEqual(owner_user_for(key.api_user), creator)
+
+    def test_a_deleted_creator_leaves_the_resource_ownerless(self) -> None:
+        """`created_by` is SET_NULL, so the key can outlive its creator."""
+        key = self._make_key(created_by=_make_user())
+        PlatformApiKey.objects.filter(pk=key.pk).update(created_by=None)
+        self.assertEqual(owner_user_for(key.api_user), key.api_user)
+
+    def test_a_service_account_with_no_key_stays_itself(self) -> None:
+        key = self._make_key(created_by=_make_user())
+        service_account = key.api_user
+        PlatformApiKey.objects.filter(pk=key.pk).delete()
+        self.assertEqual(owner_user_for(service_account), service_account)