diff --git a/backend/adapter_processor_v2/views.py b/backend/adapter_processor_v2/views.py
index 81f537f619..00c1bfe8dc 100644
--- a/backend/adapter_processor_v2/views.py
+++ b/backend/adapter_processor_v2/views.py
@@ -22,6 +22,7 @@
     diff,
     record_platform_event,
 )
+from platform_audit.resources import record_resource_event, resource_ref
 from plugins import get_plugin
 from rest_framework import status
 from rest_framework.decorators import action
@@ -319,6 +320,7 @@ def create(self, request: Any) -> Response:
             raise DuplicateAdapterNameError(
                 name=serializer.validated_data.get(AdapterKeys.ADAPTER_NAME)
             )
+        record_resource_event(request, Action.RESOURCE_CREATED, instance)
         headers = self.get_success_headers(serializer.data)
         return Response(serializer.data, status=status.HTTP_201_CREATED, headers=headers)
 
@@ -396,6 +398,7 @@ def destroy(
             )
             raise DeleteAdapterInUseError(adapter_name=adapter_instance.adapter_name)
 
+        deleted = resource_ref(adapter_instance)
         try:
             super().perform_destroy(adapter_instance)
         except ProtectedError:
@@ -405,6 +408,7 @@ def destroy(
             )
             # TODO: Provide details of adpter usage with exception object
             raise DeleteAdapterInUseError(adapter_name=adapter_instance.adapter_name)
+        record_resource_event(request, Action.RESOURCE_DELETED, deleted)
         return Response(status=status.HTTP_204_NO_CONTENT)
 
     @action(detail=True, methods=["post"], url_path="share")
diff --git a/backend/api_v2/api_deployment_views.py b/backend/api_v2/api_deployment_views.py
index 739fc2e703..73c90ff784 100644
--- a/backend/api_v2/api_deployment_views.py
+++ b/backend/api_v2/api_deployment_views.py
@@ -14,6 +14,8 @@
 )
 from permissions.resource_share_views import ResourceShareManagementMixin
 from platform_api.openapi_schema import PlatformKeyAutoSchema
+from platform_audit.actions import Action
+from platform_audit.resources import record_resource_event, resource_ref
 from plugins import get_plugin
 from prompt_studio.prompt_studio_registry_v2.models import PromptStudioRegistry
 from rest_framework import serializers, status, views, viewsets
@@ -389,6 +391,7 @@ def create(
         # OWNER membership row (UN-2202 co-owners).
         serializer.instance.grant_owner(request.user)
         api_key = DeploymentHelper.create_api_key(serializer=serializer, request=request)
+        record_resource_event(request, Action.RESOURCE_CREATED, serializer.instance)
         response_serializer = DeploymentResponseSerializer(
             {"api_key": api_key.api_key, **serializer.data}
         )
@@ -408,6 +411,11 @@ def create(
             headers=headers,
         )
 
+    def perform_destroy(self, instance: APIDeployment) -> None:
+        deleted = resource_ref(instance)
+        super().perform_destroy(instance)
+        record_resource_event(self.request, Action.RESOURCE_DELETED, deleted)
+
     @action(detail=False, methods=["get"])
     def by_prompt_studio_tool(self, request: Request) -> Response:
         """Get API deployments for a specific prompt studio tool."""
diff --git a/backend/connector_v2/views.py b/backend/connector_v2/views.py
index 3a869e500c..785d09fedb 100644
--- a/backend/connector_v2/views.py
+++ b/backend/connector_v2/views.py
@@ -11,6 +11,8 @@
 from permissions.membership_views import OwnerManagementMixin
 from permissions.permission import IsOwner, IsOwnerOrSharedUserOrSharedToOrg
 from permissions.resource_share_views import ResourceShareManagementMixin
+from platform_audit.actions import Action
+from platform_audit.resources import record_resource_event, resource_ref
 from plugins import get_plugin
 from rest_framework import status, viewsets
 from rest_framework.decorators import action
@@ -245,6 +247,7 @@ 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.grant_owner(request.user)
+        record_resource_event(request, Action.RESOURCE_CREATED, serializer.instance)
         headers = self.get_success_headers(serializer.data)
         return Response(serializer.data, status=status.HTTP_201_CREATED, headers=headers)
 
@@ -262,6 +265,7 @@ def perform_destroy(self, instance: ConnectorInstance) -> None:
         Raises:
             DeleteConnectorInUseError: If the connector is being used in workflows
         """
+        deleted = resource_ref(instance)
         try:
             super().perform_destroy(instance)
         except ProtectedError:
@@ -270,3 +274,4 @@ def perform_destroy(self, instance: ConnectorInstance) -> None:
                 f" named {instance.connector_name}"
             )
             raise DeleteConnectorInUseError(connector_name=instance.connector_name)
+        record_resource_event(self.request, Action.RESOURCE_DELETED, deleted)
diff --git a/backend/permissions/membership_serializers.py b/backend/permissions/membership_serializers.py
index 5a4e4d114e..3022a08392 100644
--- a/backend/permissions/membership_serializers.py
+++ b/backend/permissions/membership_serializers.py
@@ -36,14 +36,34 @@ def validate_user_id(self, value: int) -> int:
             raise serializers.ValidationError("User is already an owner.")
         return value
 
-    def save(self, **kwargs) -> None:
+    def save(self, **kwargs) -> bool:
+        """Promote a viewer or create the owner row; whether anything changed.
+
+        A concurrent request may have made the user an owner since validation.
+        """
         resource = self.context["resource"]
-        # Reverse manager auto-scopes to this resource; promote a viewer or
-        # create the owner row.
-        resource.memberships.update_or_create(
-            user_id=self.validated_data["user_id"],
-            defaults={"role": ResourceRole.OWNER},
-        )
+        user_id = self.validated_data["user_id"]
+        with transaction.atomic():
+            # Reverse manager auto-scopes to this resource. Lock the user's row
+            # by key (a concurrent share/ may add, remove or recreate it);
+            # create one only when there is none, then lock what exists.
+            for _ in range(3):
+                row = (
+                    resource.memberships.select_for_update()
+                    .filter(user_id=user_id)
+                    .first()
+                )
+                if row is not None:
+                    if row.role == ResourceRole.OWNER:
+                        return False
+                    resource.memberships.filter(pk=row.pk).update(role=ResourceRole.OWNER)
+                    return True
+                _, created = resource.memberships.get_or_create(
+                    user_id=user_id, defaults={"role": ResourceRole.OWNER}
+                )
+                if created:
+                    return True
+        raise serializers.ValidationError("Could not add the co-owner; please retry.")
 
 
 class RemoveOwnerSerializer(serializers.Serializer):
@@ -65,7 +85,8 @@ def validate(self, attrs: dict) -> dict:
             )
         return attrs
 
-    def save(self, **kwargs) -> None:
+    def save(self, **kwargs) -> bool:
+        """Remove the OWNER row; whether there was one to remove."""
         resource = self.context["resource"]
         user = self.context["user_to_remove"]
         with transaction.atomic():
@@ -93,4 +114,7 @@ def save(self, **kwargs) -> None:
                 raise serializers.ValidationError(
                     "Cannot remove the last owner. Add another owner first."
                 )
-            resource.memberships.filter(user=user, role=ResourceRole.OWNER).delete()
+            deleted, _ = resource.memberships.filter(
+                user=user, role=ResourceRole.OWNER
+            ).delete()
+            return deleted > 0
diff --git a/backend/permissions/membership_views.py b/backend/permissions/membership_views.py
index 6c296beb30..16507b4dad 100644
--- a/backend/permissions/membership_views.py
+++ b/backend/permissions/membership_views.py
@@ -3,6 +3,8 @@
 
 from account_v2.models import User
 from django.shortcuts import get_object_or_404
+from platform_audit.actions import Action
+from platform_audit.resources import record_resource_event
 from plugins import get_plugin
 from rest_framework import status
 from rest_framework.decorators import action
@@ -40,8 +42,12 @@ def add_co_owner(self, request: Request, pk: str | None = None) -> Response:
             data=request.data, context={"request": request, "resource": resource}
         )
         serializer.is_valid(raise_exception=True)
-        serializer.save()
+        changed = serializer.save()
         added_user = User.objects.get(id=serializer.validated_data["user_id"])
+        if changed:
+            _record_co_owner(
+                request, Action.RESOURCE_CO_OWNER_ADDED, resource, added_user
+            )
         self._notify_owner_added(resource, added_user, request.user)
         return Response(
             {"id": str(resource.pk), "co_owners": self._owner_refs(resource)},
@@ -63,7 +69,10 @@ def remove_co_owner(
             },
         )
         serializer.is_valid(raise_exception=True)
-        serializer.save()
+        if serializer.save():
+            _record_co_owner(
+                request, Action.RESOURCE_CO_OWNER_REMOVED, resource, user_to_remove
+            )
         self.on_owner_removed(resource, user_to_remove)
         self._notify_owner_removed(resource, user_to_remove, request.user)
         return Response(status=status.HTTP_204_NO_CONTENT)
@@ -139,3 +148,14 @@ def _notify_owner_removed(self, resource: Any, user: User, actor: Any) -> None:
             )
         except Exception as e:
             logger.exception("Failed to send co-owner removed notification: %s", e)
+
+
+def _record_co_owner(request: Request, action: str, resource: Any, user: User) -> None:
+    """Searchable by the co-owner's e-mail, like a group membership change."""
+    record_resource_event(
+        request,
+        action,
+        resource,
+        summary=user.email,
+        metadata={"user_id": user.pk, "email": user.email},
+    )
diff --git a/backend/permissions/resource_share_views.py b/backend/permissions/resource_share_views.py
index 43c440f369..d57ca39736 100644
--- a/backend/permissions/resource_share_views.py
+++ b/backend/permissions/resource_share_views.py
@@ -202,6 +202,11 @@ def share(self, request: Request, pk: str | None = None) -> Response:
             if "shared_groups" in desired
             else set()
         )
+        org_before = (
+            bool(getattr(resource, "shared_to_org", False))
+            if "shared_to_org" in desired
+            else None
+        )
         ShareAuthorizationService.authorize_and_commit(
             actor=request.user, resource=resource, desired=desired
         )
@@ -223,6 +228,19 @@ def share(self, request: Request, pk: str | None = None) -> Response:
             if "shared_groups" in desired
             else groups_before
         )
+        from platform_audit.resources import record_sharing_change
+
+        record_sharing_change(
+            request,
+            resource,
+            users=(users_before, users_after) if "shared_users" in desired else None,
+            groups=(groups_before, groups_after) if "shared_groups" in desired else None,
+            shared_to_org=(
+                (org_before, bool(resource.shared_to_org))
+                if org_before is not None
+                else None
+            ),
+        )
         notify_resource_group_share_changed(
             resource=resource,
             added=groups_after - groups_before,
diff --git a/backend/pipeline_v2/views.py b/backend/pipeline_v2/views.py
index 2171472e9d..e6e021cb09 100644
--- a/backend/pipeline_v2/views.py
+++ b/backend/pipeline_v2/views.py
@@ -16,6 +16,8 @@
     is_activation_only_patch,
 )
 from permissions.resource_share_views import ResourceShareManagementMixin
+from platform_audit.actions import Action
+from platform_audit.resources import record_resource_event, resource_ref
 from plugins import get_plugin
 from rest_framework import serializers, status, viewsets
 from rest_framework.decorators import action
@@ -179,11 +181,14 @@ def create(self, request: Request) -> Response:
             raise DuplicateData(
                 f"{PipelineErrors.PIPELINE_EXISTS}, {PipelineErrors.DUPLICATE_API}"
             )
+        record_resource_event(request, Action.RESOURCE_CREATED, pipeline_instance)
         return Response(data=serializer.data, status=status.HTTP_201_CREATED)
 
     def perform_destroy(self, instance: Pipeline) -> None:
         pipeline_to_remove = str(instance.pk)
+        deleted = resource_ref(instance)
         super().perform_destroy(instance)
+        record_resource_event(self.request, Action.RESOURCE_DELETED, deleted)
         return SchedulerHelper.remove_job(pipeline_to_remove)
 
     @action(detail=True, methods=["get"], url_path="users")
diff --git a/backend/platform_audit/README.md b/backend/platform_audit/README.md
index 03f7f1339e..39aae350cd 100644
--- a/backend/platform_audit/README.md
+++ b/backend/platform_audit/README.md
@@ -223,10 +223,23 @@ Expected emitters outside this PR:
 | `platform_setting.changed` | Setting changed | settings | `notifications/settings/` PATCH |
 | `capability.changed` | Capability changed | capabilities | `organization/capabilities/` PATCH and `organization/settings` PATCH (its two flags are the storage of two capabilities), one event per changed capability; target `capability`, id = capability key, name = label, `changes` `{"before": {"setting": …}, "after": {"setting": …}}` with values `admins_only`/`any_member` |
 | `execution.aborted` | Execution aborted | executions | PB-T08 |
+| `resource.created` | Resource created | resources | create of an adapter, connector, workflow, API deployment, pipeline or Prompt Studio project |
+| `resource.deleted` | Resource deleted | resources | delete of the same; a refused delete records nothing |
+| `resource.sharing_changed` | Sharing changed | resources | `<resource>/<id>/share/` POST; `changes` holds only the axes that changed (`shared_users` e-mails, `shared_groups` names, `shared_to_org`) |
+| `resource.co_owner_added` | Co-owner added | resources | `<resource>/<id>/owners/` POST; co-owner e-mail in summary |
+| `resource.co_owner_removed` | Co-owner removed | resources | `<resource>/<id>/owners/<user_id>/` DELETE |
+
+The `resource.*` target type is the resource's kind: `adapter_instance`,
+`connector_instance`, `workflow`, `api_deployment`, `pipeline`, `custom_tool`
+(a Prompt Studio project), and on cloud `agentic_project` and `lookup`, whose
+sharing is recorded through the same endpoints. The summary is the adapter type
+(`LLM`, `EMBEDDING`, ...) or pipeline type (`ETL`, `TASK`) where there is one.
 
 Target types: `member`, `invitation`, `group`, `platform_api_key`,
 `global_api_deployment_key`, `platform_key`, `default_profile`,
-`platform_setting`, `capability`, `execution`.
+`platform_setting`, `capability`, `execution`, and the resource kinds
+`adapter_instance`, `connector_instance`, `workflow`, `api_deployment`,
+`pipeline`, `custom_tool`, `agentic_project`, `lookup`.
 
 Per-deployment API keys (`api_v2`) are not recorded.
 
@@ -257,6 +270,14 @@ made any other way are not recorded: identity-provider or SCIM changes, cloud
 staff and admin routes, Django admin, invitation acceptance, and cloud-only
 features (until cloud code emits them).
 
+`resource.created` and `resource.deleted` come from the OSS viewsets only.
+These cloud paths create or delete the same kinds of resource without an
+event: Prompt Studio clone (`prompt-studio/clone/`, also mounted on-prem), the
+Agentic Studio connector views that create and delete adapters, and
+frictionless onboarding (the default adapters and the sample project). A
+project cloned and later deleted through the OSS viewset therefore shows a
+`resource.deleted` with no matching `resource.created`.
+
 Side effects that change a member's default adapter are not recorded. Among
 them: creating an adapter when the member has no default of that type makes it
 the default; revoking a share, or removing a co-owner who loses all access,
@@ -268,6 +289,10 @@ active a moment before; two admins toggling keys at the same instant can make
 that list miss or name the wrong key, and if reading that list fails, the
 deactivated keys are not recorded.
 
+Editing a resource (renaming it, changing its settings) is not recorded. Nor is
+anything a resource delete removes with it: deleting a workflow also deletes its
+pipelines and API deployments, with no event of their own.
+
 Each invitation is recorded as the identity provider accepts it, so the ones
 sent before a later one fails are kept. The role recorded is the one the
 request asked for; an identity provider that coerces a missing or unknown role
diff --git a/backend/platform_audit/actions.py b/backend/platform_audit/actions.py
index 0d47a50178..21766ce4f2 100644
--- a/backend/platform_audit/actions.py
+++ b/backend/platform_audit/actions.py
@@ -33,6 +33,7 @@ class ActionCategory(StrEnum):
     SETTINGS = "settings"
     CAPABILITIES = "capabilities"
     EXECUTIONS = "executions"
+    RESOURCES = "resources"
 
 
 class TargetType(StrEnum):
@@ -73,6 +74,12 @@ class Action(StrEnum):
     PLATFORM_SETTING_CHANGED = "platform_setting.changed"
     CAPABILITY_CHANGED = "capability.changed"
     EXECUTION_ABORTED = "execution.aborted"
+    # Target: the resource, typed by its SHAREABLE_RESOURCES kind.
+    RESOURCE_CREATED = "resource.created"
+    RESOURCE_DELETED = "resource.deleted"
+    RESOURCE_SHARING_CHANGED = "resource.sharing_changed"
+    RESOURCE_CO_OWNER_ADDED = "resource.co_owner_added"
+    RESOURCE_CO_OWNER_REMOVED = "resource.co_owner_removed"
 
 
 # The width of PlatformAuditEvent.action; a longer key could never be stored.
@@ -133,5 +140,10 @@ def all_actions() -> list[AuditAction]:
     (Action.PLATFORM_SETTING_CHANGED, "Setting changed", ActionCategory.SETTINGS),
     (Action.CAPABILITY_CHANGED, "Capability changed", ActionCategory.CAPABILITIES),
     (Action.EXECUTION_ABORTED, "Execution aborted", ActionCategory.EXECUTIONS),
+    (Action.RESOURCE_CREATED, "Resource created", ActionCategory.RESOURCES),
+    (Action.RESOURCE_DELETED, "Resource deleted", ActionCategory.RESOURCES),
+    (Action.RESOURCE_SHARING_CHANGED, "Sharing changed", ActionCategory.RESOURCES),
+    (Action.RESOURCE_CO_OWNER_ADDED, "Co-owner added", ActionCategory.RESOURCES),
+    (Action.RESOURCE_CO_OWNER_REMOVED, "Co-owner removed", ActionCategory.RESOURCES),
 ):
     register_action(_key, _label, _category)
diff --git a/backend/platform_audit/resources.py b/backend/platform_audit/resources.py
new file mode 100644
index 0000000000..2ba667b875
--- /dev/null
+++ b/backend/platform_audit/resources.py
@@ -0,0 +1,118 @@
+"""Create, delete and share events for shareable resources.
+
+Adapters, connectors, workflows, API deployments, pipelines and Prompt Studio
+projects (plus any other model in ``SHAREABLE_RESOURCES``). The target type is
+the resource's ``kind`` from that registry, so the read API and the share
+endpoints name a resource type the same way.
+"""
+
+from dataclasses import dataclass
+from typing import Any
+
+from tenant_account_v2.shareable_resources import descriptor_for_kind, kind_for_instance
+
+from platform_audit.recorder import best_effort, record_platform_event
+
+# The field that says what sort of resource it is, where the kind alone does
+# not: an LLM vs an embedding adapter, an ETL vs a Task pipeline.
+_SUBTYPE_FIELDS = {
+    "adapter_instance": "adapter_type",
+    "pipeline": "pipeline_type",
+}
+
+
+@dataclass(frozen=True)
+class ResourceRef:
+    kind: str
+    id: str
+    name: str
+    subtype: str
+    organization: Any
+
+
+@best_effort
+def resource_ref(resource: Any) -> ResourceRef | None:
+    """What an event needs to name ``resource``, read now.
+
+    Read it before deleting: a deleted row has no primary key. Best effort,
+    because it runs before the delete: a failed read (``organization`` can be a
+    query) gives ``None``, which records nothing, rather than failing the delete.
+    """
+    kind = kind_for_instance(resource)
+    if kind is None:
+        return None
+    descriptor = descriptor_for_kind(kind)
+    subtype_field = _SUBTYPE_FIELDS.get(kind)
+    return ResourceRef(
+        kind=kind,
+        id=str(getattr(resource, descriptor.id_field)),
+        name=str(getattr(resource, descriptor.name_field, "") or ""),
+        subtype=str(getattr(resource, subtype_field, "") or "") if subtype_field else "",
+        organization=getattr(resource, "organization", None),
+    )
+
+
+@best_effort
+def record_resource_event(
+    request: Any,
+    action: str,
+    resource: Any,
+    *,
+    summary: str = "",
+    changes: dict | None = None,
+    metadata: dict | None = None,
+) -> None:
+    """Record ``action`` on ``resource`` (a model instance or a ``ResourceRef``).
+
+    The summary defaults to the resource's subtype (``LLM``, ``ETL``).
+    """
+    ref = resource if isinstance(resource, ResourceRef) else resource_ref(resource)
+    if ref is None:
+        return
+    record_platform_event(
+        action=action,
+        target_type=ref.kind,
+        target_id=ref.id,
+        target_name=ref.name,
+        summary=summary or ref.subtype,
+        changes=changes,
+        metadata=metadata,
+        request=request,
+        organization=ref.organization,
+    )
+
+
+@best_effort
+def record_sharing_change(
+    request: Any,
+    resource: Any,
+    *,
+    users: tuple[set, set] | None = None,
+    groups: tuple[set, set] | None = None,
+    shared_to_org: tuple[bool, bool] | None = None,
+) -> None:
+    """One ``resource.sharing_changed`` for a share request, if it changed anything.
+
+    Each argument is the ``(before, after)`` of an axis the request touched:
+    users and groups as model instances (stored as sorted e-mails and group
+    names), ``shared_to_org`` as booleans. Untouched axes are ``None``.
+    """
+    from platform_audit.actions import Action
+
+    sides: dict[str, dict] = {"before": {}, "after": {}}
+    for axis, pair, label in (
+        ("shared_users", users, lambda u: u.email),
+        ("shared_groups", groups, lambda g: g.name),
+    ):
+        if pair is None or pair[0] == pair[1]:
+            continue
+        sides["before"][axis] = sorted(label(x) for x in pair[0])
+        sides["after"][axis] = sorted(label(x) for x in pair[1])
+    if shared_to_org is not None and shared_to_org[0] != shared_to_org[1]:
+        sides["before"]["shared_to_org"] = shared_to_org[0]
+        sides["after"]["shared_to_org"] = shared_to_org[1]
+    if not sides["after"]:
+        return
+    record_resource_event(
+        request, Action.RESOURCE_SHARING_CHANGED, resource, changes=sides
+    )
diff --git a/backend/platform_audit/tests/test_resource_events.py b/backend/platform_audit/tests/test_resource_events.py
new file mode 100644
index 0000000000..0cd5df2d17
--- /dev/null
+++ b/backend/platform_audit/tests/test_resource_events.py
@@ -0,0 +1,561 @@
+"""Create, delete, share and co-owner events for shareable resources.
+
+Every flow goes through the real viewset, so each emit site is exercised.
+"""
+
+import json
+from unittest.mock import MagicMock, patch
+
+from adapter_processor_v2.models import AdapterInstance
+from adapter_processor_v2.views import AdapterInstanceViewSet
+from api_v2.api_deployment_views import APIDeploymentViewSet
+from api_v2.models import APIDeployment
+from connector_v2.models import ConnectorInstance
+from connector_v2.views import ConnectorInstanceViewSet
+from django.core.files.uploadedfile import SimpleUploadedFile
+from permissions.membership_serializers import AddOwnerSerializer, RemoveOwnerSerializer
+from permissions.roles import ResourceRole
+from pipeline_v2.models import Pipeline
+from pipeline_v2.views import PipelineViewSet
+from prompt_studio.prompt_studio_core_v2.models import CustomTool
+from prompt_studio.prompt_studio_core_v2.prompt_studio_helper import PromptStudioHelper
+from prompt_studio.prompt_studio_core_v2.views import PromptStudioCoreView
+from rest_framework.test import force_authenticate
+from tenant_account_v2.models import OrganizationGroup, OrganizationMember
+from workflow_manager.endpoint_v2.models import WorkflowEndpoint
+from workflow_manager.workflow_v2.models.workflow import Workflow
+from workflow_manager.workflow_v2.views import WorkflowViewSet
+
+from platform_audit.actions import Action
+from platform_audit.tests.base import AuditTestBase, make_user
+from platform_audit.views import PlatformAuditEventListView
+
+
+class ResourceEventTestBase(AuditTestBase):
+    def assert_resource_event(
+        self, action: str, kind: str, resource_id, name: str, summary: str = ""
+    ):
+        event = self.only_event(action=action, target_type=kind)
+        self.assert_admin_actor(event)
+        self.assertEqual(
+            (event.target_type, event.target_id, event.target_name, event.summary),
+            (kind, str(resource_id), name, summary),
+        )
+        return event
+
+    def create_workflow(self, name: str = "wf") -> Workflow:
+        response = self.call(
+            WorkflowViewSet.as_view({"post": "create"}), "post", data={"workflow_name": name}
+        )
+        self.assertEqual(response.status_code, 201, response.data)
+        return Workflow.objects.get(workflow_name=name)
+
+    def create_adapter(self, name: str = "my-llm") -> AdapterInstance:
+        with patch.object(AdapterInstance, "get_context_window_size", return_value=4096):
+            response = self.call(
+                AdapterInstanceViewSet.as_view({"post": "create"}),
+                "post",
+                data={
+                    "adapter_id": "openai|test",
+                    "adapter_name": name,
+                    "adapter_type": "LLM",
+                    "adapter_metadata": {"api_key": "sk-test", "model": "gpt-4o-mini"},
+                },
+            )
+        self.assertEqual(response.status_code, 201, response.data)
+        return AdapterInstance.objects.get(adapter_name=name)
+
+    def create_connector(self) -> ConnectorInstance:
+        # The connector and payload of connector_v2's register test; creating
+        # stores the metadata without contacting the storage.
+        response = self.call(
+            ConnectorInstanceViewSet.as_view({"post": "create"}),
+            "post",
+            data={
+                "connector_id": "minio|c799f6e3-2b57-434e-aaac-b5daa415da19",
+                "connector_name": "minio-in",
+                "connector_metadata": {
+                    "connectorName": "minio",
+                    "key": "access-key",
+                    "secret": "secret-key",
+                    "endpoint_url": "http://minio.invalid:9000",
+                    "region_name": "",
+                    "bucket": "bucket",
+                    "path": "/",
+                },
+            },
+        )
+        self.assertEqual(response.status_code, 201, response.data)
+        return ConnectorInstance.objects.get(connector_name="minio-in")
+
+
+class CreateAndDeleteTests(ResourceEventTestBase):
+    def _delete(self, viewset, resource, pk) -> None:
+        response = self.call(
+            viewset.as_view({"delete": "destroy"}), "delete", pk=str(pk)
+        )
+        self.assertIn(response.status_code, (200, 204), getattr(response, "data", None))
+
+    def test_adapter(self) -> None:
+        adapter = self.create_adapter()
+        self.assert_resource_event(
+            Action.RESOURCE_CREATED, "adapter_instance", adapter.id, "my-llm", "LLM"
+        )
+        adapter_id = adapter.id
+        # The creator's first LLM becomes their default, which cannot be
+        # deleted; delete a second one.
+        second = self.create_adapter("second-llm")
+        self._delete(AdapterInstanceViewSet, second, second.id)
+        self.assert_resource_event(
+            Action.RESOURCE_DELETED, "adapter_instance", second.id, "second-llm", "LLM"
+        )
+        self.assertTrue(AdapterInstance.objects.filter(id=adapter_id).exists())
+
+    def test_connector(self) -> None:
+        connector = self.create_connector()
+        self.assert_resource_event(
+            Action.RESOURCE_CREATED, "connector_instance", connector.id, "minio-in"
+        )
+        self._delete(ConnectorInstanceViewSet, connector, connector.id)
+        self.assert_resource_event(
+            Action.RESOURCE_DELETED, "connector_instance", connector.id, "minio-in"
+        )
+
+    def test_a_failing_snapshot_does_not_stop_the_delete(self) -> None:
+        connector = self.create_connector()
+        with (
+            patch(
+                "platform_audit.resources.kind_for_instance",
+                side_effect=RuntimeError("db"),
+            ),
+            self.assertLogs("platform_audit.recorder", "ERROR"),
+        ):
+            self._delete(ConnectorInstanceViewSet, connector, connector.id)
+        self.assertFalse(ConnectorInstance.objects.filter(id=connector.id).exists())
+        self.assertEqual(self.events(action=Action.RESOURCE_DELETED), [])
+
+    def test_workflow(self) -> None:
+        workflow = self.create_workflow()
+        self.assert_resource_event(Action.RESOURCE_CREATED, "workflow", workflow.id, "wf")
+        self._delete(WorkflowViewSet, workflow, workflow.id)
+        self.assert_resource_event(Action.RESOURCE_DELETED, "workflow", workflow.id, "wf")
+
+    def test_a_workflow_whose_endpoint_setup_fails_is_still_recorded(self) -> None:
+        # With ATOMIC_REQUESTS off (the default) the workflow row stays behind.
+        with patch(
+            "workflow_manager.workflow_v2.views.WorkflowEndpointUtils"
+            ".create_endpoints_for_workflow",
+            side_effect=RuntimeError("endpoint setup"),
+        ):
+            response = self.call(
+                WorkflowViewSet.as_view({"post": "create"}),
+                "post",
+                data={"workflow_name": "half-made"},
+            )
+        self.assertGreaterEqual(response.status_code, 400)
+        workflow = Workflow.objects.get(workflow_name="half-made")
+        self.assert_resource_event(
+            Action.RESOURCE_CREATED, "workflow", workflow.id, "half-made"
+        )
+
+    def test_api_deployment(self) -> None:
+        workflow = self.create_workflow()
+        for endpoint in WorkflowEndpoint.objects.filter(workflow=workflow):
+            endpoint.connection_type = WorkflowEndpoint.ConnectionType.API
+            endpoint.save()
+        response = self.call(
+            APIDeploymentViewSet.as_view({"post": "create"}),
+            "post",
+            data={
+                "workflow": str(workflow.id),
+                "display_name": "my api",
+                "api_name": "my-api",
+                "description": "audited",
+            },
+        )
+        self.assertEqual(response.status_code, 201, response.data)
+        api = APIDeployment.objects.get(api_name="my-api")
+        self.assert_resource_event(Action.RESOURCE_CREATED, "api_deployment", api.id, "my api")
+        self._delete(APIDeploymentViewSet, api, api.id)
+        self.assert_resource_event(Action.RESOURCE_DELETED, "api_deployment", api.id, "my api")
+
+    def test_pipeline(self) -> None:
+        workflow = self.create_workflow()
+        response = self.call(
+            PipelineViewSet.as_view({"post": "create"}),
+            "post",
+            data={
+                "workflow": str(workflow.id),
+                "pipeline_name": "nightly",
+                "pipeline_type": Pipeline.PipelineType.ETL,
+            },
+        )
+        self.assertEqual(response.status_code, 201, response.data)
+        pipeline = Pipeline.objects.get(pipeline_name="nightly")
+        self.assert_resource_event(
+            Action.RESOURCE_CREATED, "pipeline", pipeline.id, "nightly", "ETL"
+        )
+        with patch("pipeline_v2.views.SchedulerHelper.remove_job"):
+            self._delete(PipelineViewSet, pipeline, pipeline.id)
+        self.assert_resource_event(
+            Action.RESOURCE_DELETED, "pipeline", pipeline.id, "nightly", "ETL"
+        )
+
+    def test_prompt_studio_project(self) -> None:
+        response = self.call(
+            PromptStudioCoreView.as_view({"post": "create"}),
+            "post",
+            data={"tool_name": "invoice-parser", "description": "d", "author": "test"},
+        )
+        self.assertEqual(response.status_code, 201, response.data)
+        tool = CustomTool.objects.get(tool_name="invoice-parser")
+        self.assert_resource_event(
+            Action.RESOURCE_CREATED, "custom_tool", tool.tool_id, "invoice-parser"
+        )
+        # Deleting also removes the project's files from remote storage.
+        models = "prompt_studio.prompt_studio_core_v2.models"
+        with (
+            patch(f"{models}.EnvHelper.get_storage", return_value=MagicMock()),
+            patch(
+                f"{models}.PromptStudioFileHelper.get_or_create_prompt_studio_subdirectory",
+                return_value="unused",
+            ),
+        ):
+            self._delete(PromptStudioCoreView, tool, tool.tool_id)
+        self.assert_resource_event(
+            Action.RESOURCE_DELETED, "custom_tool", tool.tool_id, "invoice-parser"
+        )
+
+    def _import_project(self):
+        # The export's top-level shape (prompt_studio_helper.validate_import_file).
+        export = {
+            "tool_metadata": {"tool_name": "imported", "author": "a", "description": "d"},
+            "tool_settings": {},
+            "prompts": [],
+        }
+        upload = SimpleUploadedFile(
+            "project.json", json.dumps(export).encode(), content_type="application/json"
+        )
+        request = self.factory.post("/", {"file": upload}, format="multipart")
+        request.id = "req-123"  # as self.call sets it
+        force_authenticate(request, user=self.admin)
+        helper = "prompt_studio.prompt_studio_core_v2.views.PromptStudioHelper"
+        # The steps after the project row exists are not under test.
+        with (
+            patch(f"{helper}.import_prompts"),
+            patch(f"{helper}.validate_adapter_configuration", return_value=(False, "")),
+            self.captureOnCommitCallbacks(execute=True),
+        ):
+            return PromptStudioCoreView.as_view({"post": "import_project"})(request)
+
+    def test_prompt_studio_import(self) -> None:
+        with patch(
+            "prompt_studio.prompt_studio_core_v2.views.PromptStudioHelper"
+            ".create_profile_manager"
+        ):
+            response = self._import_project()
+        self.assertEqual(response.status_code, 201, response.data)
+        tool = CustomTool.objects.get(tool_id=response.data["tool_id"])
+        self.assert_resource_event(
+            Action.RESOURCE_CREATED, "custom_tool", tool.tool_id, tool.tool_name
+        )
+
+    def test_a_failed_import_still_records_the_project_it_left(self) -> None:
+        # The view has no transaction: a later failure leaves the project row.
+        with patch(
+            "prompt_studio.prompt_studio_core_v2.views.PromptStudioHelper"
+            ".create_profile_manager",
+            side_effect=ValueError("adapter not found"),
+        ):
+            response = self._import_project()
+        self.assertEqual(response.status_code, 400, response.data)
+        tool = CustomTool.objects.get()
+        self.assert_resource_event(
+            Action.RESOURCE_CREATED, "custom_tool", tool.tool_id, tool.tool_name
+        )
+
+    def test_a_sync_prompts_backup_copy_is_recorded(self) -> None:
+        self.call(
+            PromptStudioCoreView.as_view({"post": "create"}),
+            "post",
+            data={"tool_name": "source", "description": "d", "author": "test"},
+        )
+        source = CustomTool.objects.get(tool_name="source")
+        export = PromptStudioHelper.export_project_settings(source)
+        # Syncing into the source needs its default profile; the backup is
+        # made before that and is what is under test.
+        with patch(
+            "prompt_studio.prompt_studio_core_v2.views.PromptStudioHelper.sync_prompts",
+            return_value={"prompts_created": 0},
+        ):
+            response = self.call(
+                PromptStudioCoreView.as_view({"post": "sync_prompts"}),
+                "post",
+                data={"data": export, "create_copy": True},
+                pk=str(source.tool_id),
+            )
+        self.assertEqual(response.status_code, 200, response.data)
+        backup = CustomTool.objects.get(tool_id=response.data["backup_tool_id"])
+        event = self.only_event(
+            action=Action.RESOURCE_CREATED, target_id=str(backup.tool_id)
+        )
+        self.assert_admin_actor(event)
+        self.assertEqual(
+            (event.target_type, event.target_name, event.summary),
+            ("custom_tool", backup.tool_name, "backup of source"),
+        )
+
+    def test_an_in_use_connector_is_not_recorded_deleted(self) -> None:
+        connector = self.create_connector()
+        workflow = self.create_workflow()
+        WorkflowEndpoint.objects.filter(workflow=workflow).update(
+            connector_instance=connector
+        )
+        response = self.call(
+            ConnectorInstanceViewSet.as_view({"delete": "destroy"}),
+            "delete",
+            pk=str(connector.id),
+        )
+        self.assertGreaterEqual(response.status_code, 400)
+        self.assertTrue(ConnectorInstance.objects.filter(id=connector.id).exists())
+        self.assertEqual(self.events(action=Action.RESOURCE_DELETED), [])
+
+    def test_a_refused_delete_records_nothing(self) -> None:
+        adapter = self.create_adapter()  # the creator's default LLM
+        response = self.call(
+            AdapterInstanceViewSet.as_view({"delete": "destroy"}),
+            "delete",
+            pk=str(adapter.id),
+        )
+        self.assertGreaterEqual(response.status_code, 400)
+        self.assertEqual(self.events(action=Action.RESOURCE_DELETED), [])
+
+
+class SharingTests(ResourceEventTestBase):
+    def setUp(self) -> None:
+        super().setUp()
+        self.workflow = self.create_workflow()
+        self.group = OrganizationGroup.objects.create(organization=self.org, name="Ops")
+
+    def _share(self, viewset, pk, data):
+        response = self.call(viewset.as_view({"post": "share"}), "post", data=data, pk=str(pk))
+        self.assertEqual(response.status_code, 200, getattr(response, "data", None))
+
+    def test_records_each_changed_axis(self) -> None:
+        self._share(
+            WorkflowViewSet,
+            self.workflow.id,
+            {
+                "shared_users": [self.member.id],
+                "shared_groups": [self.group.id],
+                "shared_to_org": True,
+            },
+        )
+        event = self.assert_resource_event(
+            Action.RESOURCE_SHARING_CHANGED, "workflow", self.workflow.id, "wf"
+        )
+        self.assertEqual(
+            event.changes,
+            {
+                "before": {"shared_users": [], "shared_groups": [], "shared_to_org": False},
+                "after": {
+                    "shared_users": ["member@example.com"],
+                    "shared_groups": ["Ops"],
+                    "shared_to_org": True,
+                },
+            },
+        )
+
+    def test_unchanged_axes_and_no_op_shares_are_left_out(self) -> None:
+        self._share(WorkflowViewSet, self.workflow.id, {"shared_users": [self.member.id]})
+        self._share(
+            WorkflowViewSet,
+            self.workflow.id,
+            {"shared_users": [self.member.id], "shared_to_org": True},
+        )
+        self._share(WorkflowViewSet, self.workflow.id, {"shared_to_org": True})
+        events = self.events(action=Action.RESOURCE_SHARING_CHANGED)
+        self.assertEqual(
+            [e.changes for e in events],
+            [
+                {"before": {"shared_users": []}, "after": {"shared_users": ["member@example.com"]}},
+                {"before": {"shared_to_org": False}, "after": {"shared_to_org": True}},
+            ],
+        )
+
+    def test_adapter_share_override_records_too(self) -> None:
+        adapter = self.create_adapter()
+        self._share(AdapterInstanceViewSet, adapter.id, {"shared_to_org": True})
+        self.assert_resource_event(
+            Action.RESOURCE_SHARING_CHANGED, "adapter_instance", adapter.id, "my-llm", "LLM"
+        )
+
+    def test_co_owner_added_and_removed(self) -> None:
+        coowner = make_user("co@example.com")
+        from tenant_account_v2.models import OrganizationMember
+
+        OrganizationMember.objects.create(organization=self.org, user=coowner, role="user")
+        response = self.call(
+            WorkflowViewSet.as_view({"post": "add_co_owner"}),
+            "post",
+            data={"user_id": coowner.id},
+            pk=str(self.workflow.id),
+        )
+        self.assertEqual(response.status_code, 200, response.data)
+        added = self.assert_resource_event(
+            Action.RESOURCE_CO_OWNER_ADDED, "workflow", self.workflow.id, "wf", "co@example.com"
+        )
+        self.assertEqual(added.metadata, {"user_id": coowner.id, "email": "co@example.com"})
+        response = self.call(
+            WorkflowViewSet.as_view({"delete": "remove_co_owner"}),
+            "delete",
+            pk=str(self.workflow.id),
+            user_id=str(coowner.id),
+        )
+        self.assertEqual(response.status_code, 204)
+        self.assert_resource_event(
+            Action.RESOURCE_CO_OWNER_REMOVED, "workflow", self.workflow.id, "wf", "co@example.com"
+        )
+
+
+class CoOwnerRaceTests(ResourceEventTestBase):
+    """Two requests can pass validation for the same change; only one makes it."""
+
+    def setUp(self) -> None:
+        super().setUp()
+        self.workflow = self.create_workflow()
+        self.coowner = make_user("co@example.com")
+        OrganizationMember.objects.create(
+            organization=self.org, user=self.coowner, role="user"
+        )
+
+    def _context(self) -> dict:
+        request = self.factory.post("/")
+        request.user = self.admin
+        return {"request": request, "resource": self.workflow}
+
+    def test_an_add_racing_a_share_still_makes_the_user_an_owner(self) -> None:
+        serializer = AddOwnerSerializer(
+            data={"user_id": self.coowner.id}, context=self._context()
+        )
+        self.assertTrue(serializer.is_valid(), serializer.errors)
+        manager_cls = type(self.workflow.memberships)
+        real_get_or_create = manager_cls.get_or_create
+
+        def share_lands_first(manager, *args, **kwargs):
+            # A concurrent share/ makes them a viewer just before this write.
+            real_get_or_create(
+                manager, user_id=kwargs["user_id"], defaults={"role": ResourceRole.VIEWER}
+            )
+            return real_get_or_create(manager, *args, **kwargs)
+
+        with patch.object(manager_cls, "get_or_create", share_lands_first):
+            self.assertTrue(serializer.save())
+        self.assertEqual(
+            self.workflow.memberships.get(user=self.coowner).role, ResourceRole.OWNER
+        )
+
+    def test_an_add_racing_an_unshare_still_makes_the_user_an_owner(self) -> None:
+        viewer = self.workflow.memberships.create(
+            user=self.coowner, role=ResourceRole.VIEWER
+        )
+        serializer = AddOwnerSerializer(
+            data={"user_id": self.coowner.id}, context=self._context()
+        )
+        self.assertTrue(serializer.is_valid(), serializer.errors)
+        # A concurrent share/ removes the viewer row after validation saw it, so
+        # save() finds no row to promote and has to create the owner row.
+        viewer.delete()
+        manager_cls = type(self.workflow.memberships)
+        real_get_or_create = manager_cls.get_or_create
+        calls = []
+
+        def counted(manager, *args, **kwargs):
+            calls.append(kwargs.get("user_id"))
+            return real_get_or_create(manager, *args, **kwargs)
+
+        with patch.object(manager_cls, "get_or_create", counted):
+            self.assertTrue(serializer.save())
+        self.assertEqual(calls, [self.coowner.id])
+        self.assertEqual(
+            self.workflow.memberships.get(user=self.coowner).role, ResourceRole.OWNER
+        )
+
+    def test_an_add_racing_an_unshare_and_reshare_still_makes_the_user_an_owner(
+        self,
+    ) -> None:
+        serializer = AddOwnerSerializer(
+            data={"user_id": self.coowner.id}, context=self._context()
+        )
+        self.assertTrue(serializer.is_valid(), serializer.errors)
+        manager_cls = type(self.workflow.memberships)
+        real_get_or_create = manager_cls.get_or_create
+
+        def share_unshare_reshare(manager, *args, **kwargs):
+            # A share/ makes them a viewer; the read returns that row; an
+            # unshare removes it and another share recreates it (a new pk).
+            real_get_or_create(
+                manager, user_id=kwargs["user_id"], defaults={"role": ResourceRole.VIEWER}
+            )
+            membership, created = real_get_or_create(manager, *args, **kwargs)
+            manager.filter(pk=membership.pk).delete()
+            manager.create(user_id=kwargs["user_id"], role=ResourceRole.VIEWER)
+            return membership, created
+
+        with patch.object(manager_cls, "get_or_create", share_unshare_reshare):
+            self.assertTrue(serializer.save())
+        self.assertEqual(
+            self.workflow.memberships.get(user=self.coowner).role, ResourceRole.OWNER
+        )
+
+    def test_an_add_another_request_already_made_changes_nothing(self) -> None:
+        serializer = AddOwnerSerializer(
+            data={"user_id": self.coowner.id}, context=self._context()
+        )
+        self.assertTrue(serializer.is_valid(), serializer.errors)
+        self.workflow.memberships.create(user=self.coowner, role=ResourceRole.OWNER)
+        self.assertFalse(serializer.save())
+        self.assertEqual(
+            self.workflow.memberships.filter(user=self.coowner).count(), 1
+        )
+
+    def test_a_remove_another_request_already_made_changes_nothing(self) -> None:
+        self.workflow.memberships.create(user=self.coowner, role=ResourceRole.OWNER)
+        serializer = RemoveOwnerSerializer(
+            data={}, context={**self._context(), "user_to_remove": self.coowner}
+        )
+        self.assertTrue(serializer.is_valid(), serializer.errors)
+        self.workflow.memberships.filter(user=self.coowner).delete()
+        self.assertFalse(serializer.save())
+
+    def test_a_save_that_changed_nothing_records_nothing(self) -> None:
+        with patch(
+            "permissions.membership_views.AddOwnerSerializer.save", return_value=False
+        ):
+            response = self.call(
+                WorkflowViewSet.as_view({"post": "add_co_owner"}),
+                "post",
+                data={"user_id": self.coowner.id},
+                pk=str(self.workflow.id),
+            )
+        self.assertEqual(response.status_code, 200, response.data)
+        self.assertEqual(self.events(action=Action.RESOURCE_CO_OWNER_ADDED), [])
+
+
+class OrgScopingTests(ResourceEventTestBase):
+    def test_events_belong_to_the_resources_org_and_only_it_reads_them(self) -> None:
+        workflow = self.create_workflow()
+        event = self.only_event(action=Action.RESOURCE_CREATED)
+        self.assertEqual(event.organization, self.org)
+        self.assertEqual(workflow.organization, self.org)
+
+        other_admin = make_user("admin@other.com")
+        from tenant_account_v2.models import OrganizationMember
+
+        OrganizationMember.objects.create(
+            organization=self.other_org, user=other_admin, role="admin"
+        )
+        self.use_org(self.other_org)
+        response = self.call(PlatformAuditEventListView.as_view(), "get", user=other_admin)
+        self.assertEqual(response.status_code, 200, response.data)
+        self.assertEqual(response.data["results"], [])
diff --git a/backend/prompt_studio/prompt_studio_core_v2/views.py b/backend/prompt_studio/prompt_studio_core_v2/views.py
index 2c638c87ae..ee84d88cb5 100644
--- a/backend/prompt_studio/prompt_studio_core_v2/views.py
+++ b/backend/prompt_studio/prompt_studio_core_v2/views.py
@@ -19,6 +19,8 @@
 from permissions.membership_views import OwnerManagementMixin
 from permissions.permission import IsOwner, IsOwnerOrSharedUserOrSharedToOrg
 from permissions.resource_share_views import ResourceShareManagementMixin
+from platform_audit.actions import Action
+from platform_audit.resources import record_resource_event, resource_ref
 from plugins import get_plugin
 from rest_framework import status, viewsets
 from rest_framework.decorators import action
@@ -219,6 +221,7 @@ def create(self, request: HttpRequest) -> Response:
         PromptStudioHelper.create_default_profile_manager(
             request.user, serializer.data["tool_id"]
         )
+        record_resource_event(request, Action.RESOURCE_CREATED, serializer.instance)
 
         # Notify HubSpot if this is the first Prompt Studio project for the org
         # (count == 1 means the one we just created is the first)
@@ -233,7 +236,9 @@ def create(self, request: HttpRequest) -> Response:
 
     def perform_destroy(self, instance: CustomTool) -> None:
         organization_id = UserSessionUtils.get_organization_id(self.request)
+        deleted = resource_ref(instance)
         instance.delete(organization_id)
+        record_resource_event(self.request, Action.RESOURCE_DELETED, deleted)
 
     def _check_tool_usage_in_workflows(self, instance: CustomTool) -> tuple[bool, set]:
         """Whether ``instance``'s exported tool is in use, and by which workflows.
@@ -1298,6 +1303,8 @@ def import_project(self, request: Request) -> Response:
             new_tool = PromptStudioHelper.create_tool_from_import_data(
                 import_data, tool_name, organization, request.user
             )
+            # Recorded now: the project exists even if a later step fails.
+            record_resource_event(request, Action.RESOURCE_CREATED, new_tool)
 
             try:
                 PromptStudioHelper.create_profile_manager(
@@ -1375,6 +1382,12 @@ def sync_prompts(self, request: Request, pk: Any = None) -> Response:
             backup_tool = PromptStudioHelper.create_tool_from_import_data(
                 export_data, backup_name, organization, request.user
             )
+            record_resource_event(
+                request,
+                Action.RESOURCE_CREATED,
+                backup_tool,
+                summary=f"backup of {tool.tool_name}",
+            )
             # Copy profiles from original to backup
             for profile in ProfileManager.objects.filter(prompt_studio_tool=tool):
                 profile.pk = None
diff --git a/backend/workflow_manager/workflow_v2/views.py b/backend/workflow_manager/workflow_v2/views.py
index e2eb0a8be6..344a332afb 100644
--- a/backend/workflow_manager/workflow_v2/views.py
+++ b/backend/workflow_manager/workflow_v2/views.py
@@ -12,6 +12,8 @@
 from permissions.resource_share_views import ResourceShareManagementMixin
 from pipeline_v2.models import Pipeline
 from pipeline_v2.pipeline_processor import PipelineProcessor
+from platform_audit.actions import Action
+from platform_audit.resources import record_resource_event, resource_ref
 from plugins import get_plugin
 from rest_framework import serializers, status, viewsets
 from rest_framework.decorators import action, api_view
@@ -168,6 +170,9 @@ def perform_create(self, serializer: WorkflowSerializer) -> Workflow:
         # ``created_by`` is audit-only; the creator's access flows through an
         # OWNER membership row (UN-2202 co-owners).
         workflow.grant_owner(self.request.user)
+        # Recorded as soon as it exists (as import_project does): with
+        # ATOMIC_REQUESTS off, a failure below leaves the workflow in place.
+        record_resource_event(self.request, Action.RESOURCE_CREATED, workflow)
         try:
             # Create empty WorkflowEndpoints for UI compatibility
             # ConnectorInstances will be created when users actually configure connectors
@@ -177,6 +182,11 @@ def perform_create(self, serializer: WorkflowSerializer) -> Workflow:
             raise WorkflowGenerationError
         return workflow
 
+    def perform_destroy(self, instance: Workflow) -> None:
+        deleted = resource_ref(instance)
+        super().perform_destroy(instance)
+        record_resource_event(self.request, Action.RESOURCE_DELETED, deleted)
+
     def get_execution(self, request: Request, pk: str) -> Response:
         execution = WorkflowHelper.get_current_execution(pk)
         return Response(make_execution_response(execution), status=status.HTTP_200_OK)