✨(backend) audit recordings, transcripts and summaries

Emit audit events across the recording lifecycle: recording.start,
recording.stop and recording.delete from the API, recording.end when
LiveKit reports the egress ended, recording.transcript.request when the
backend sends a recording to the summary service, and
recording.transcript.report and recording.summary.report when that
service calls back.

recording.end has an unknown outcome when LiveKit reports an egress
status the backend does not map, and recording.transcript.request names
the summary service as service.target.

Small bugfix: AuditViewMixin keeps a copy of a deleted target, since deleting an
instance clears its primary key.

CHANGELOG.md

@@ -53,6 +53,7 @@ and this project adheres to
 - ✨(backend) add structured audit logging facility
 - ✨(backend) audit external API token and room operations
 - 🔒️(backend) audit writes and bulk actions made in the Django admin
+- ✨(backend) audit recordings, transcripts and summaries
 
 ### Changed
 

docs/features/audit-logging.md

@@ -1,7 +1,7 @@
 # Audit logging
 
 La Suite Meet keeps an **audit log**: a record of the notable things that happen in the product, such as a room created
-through the external API, a sign-in, or a change made in the Django admin. Every entry answers the
+through the external API, a recording started, a sign-in, or a change made in the Django admin. Every entry answers the
 same questions: *who* did *what*, *to what*, *on whose behalf*, *from where*, and *did it work?*
 
 It is written for:
@@ -102,7 +102,7 @@ find each one in the JSON above.
 | **Application** | A registered third-party integration, identified by its client id, that calls the external API on behalf of users. | `lasuite.application.client_id` |
 | **Delegation** | An Application acting for one specific user, through the delegated token it obtained with `application.token.issue`. The event then names both: the Application as the Actor, and the user in `user.*`. | `lasuite.actor.type` = `application`, with `user.*` set |
 | **Provisional User** | An account an Application created from an email address before that person ever signed in. It has no OIDC `sub` yet, so its account id is what links its `user.provision` event to its later `user.login`. | `event.action` = `user.provision` |
-| **Machine User** | The identity of an internal service calling the backend with a shared secret: the LiveKit SIP bridge (`roomkit`) and the summary service (`summary`). It is an Actor of type `service`, never an account. | `service.origin.name` |
+| **Machine User** | The identity of an internal service calling the backend with a shared secret: the LiveKit SIP bridge (`roomkit`), the summary service (`summary`), and the LiveKit server reporting that a recording ended (`livekit`). It is an Actor of type `service`, never an account. | `service.origin.name` |
 | **Tenant** | The organisation an event belongs to: the calling Application when there is one, otherwise the Actor's email domain. | `organization.id` |
 
 ### What it was done to
@@ -174,8 +174,8 @@ what ECS does not define: `lasuite.*`, and `entity.target.raw.*` for the fields
 | `@timestamp` | When the event was emitted: ISO 8601, in UTC, with millisecond precision |
 | `data_stream.*`, `event.dataset` | `logs`, `<AUDIT_LOG_SERVICE_NAME>.audit` and `AUDIT_LOG_DATA_STREAM_NAMESPACE`: what sets the audit stream apart from the application logs |
 | `service.name`, `service.environment`, `service.version`, `service.node.name` | The Emitter: `AUDIT_LOG_SERVICE_NAME`, the current environment, the release and the host name (the pod, on Kubernetes). Never the caller |
-| `service.origin.name` | The internal service that called the backend, when the actor is a `service`: `roomkit` or `summary` |
-| `service.target.name` | The service the backend called, when one is named |
+| `service.origin.name` | The internal service that called the backend, when the actor is a `service`: `roomkit`, `summary` or `livekit` |
+| `service.target.name` | The service the backend called, such as `summary` for `recording.transcript.request` |
 | `event.id` | Unique id of the event, so that a shipper retrying it cannot create a duplicate |
 | `event.action` | What was attempted, from the [Catalogue](#catalogue) |
 | `event.category`, `event.type` | The ECS classification (`api`, `authentication`, `iam`... and `creation`, `change`, `access`, `denied`, `user`...). Always a combination ECS expects, see [Classification](#classification) |
@@ -223,7 +223,7 @@ an application acts for a user, the actor is the application, but `user.*` is th
 |---|---|---|
 | `user` | A person's account acting for itself: through a session, an OIDC or password login, an add-on token, or the LiveKit token of a known account | That account |
 | `application` | A client application acting on behalf of a user. Either a Meet application, through its client credentials or its delegated token, or another La Suite application, through the resource server. `lasuite.application.client_id` names it | The delegating user |
-| `service` | An internal service of the deployment, named by `service.origin.name`: the LiveKit SIP bridge (`roomkit`) or the summary service (`summary`) | Absent |
+| `service` | An internal service of the deployment, named by `service.origin.name`: the LiveKit SIP bridge (`roomkit`), the summary service (`summary`), or the LiveKit server reporting on a recording (`livekit`) | Absent |
 | `system` | The backend itself, with no inbound request | Absent |
 | `anonymous` | A caller that did not authenticate, or failed to | Absent |
 
@@ -272,6 +272,30 @@ Rooms are audited on the external API only.
 | `room.retrieve` | A room is read through the external API, or the attempt fails | `entity.target` is the room |
 | `room.list` | Rooms are listed through the external API, or the attempt fails | `lasuite.details.total` |
 
+### Recordings, transcripts and summaries
+
+| `event.action` | Emitted when | Notable fields |
+|---|---|---|
+| `recording.start` | A room owner or administrator starts a recording, or the attempt fails: a recording already in progress, a worker error | `entity.target` is the recording, or the room when none was created. `lasuite.details.collect_metadata` |
+| `recording.stop` | A room owner or administrator stops the recording in progress, or the attempt fails | `entity.target` is the recording, or the room when none is active |
+| `recording.end` | LiveKit reports that a recording ended (`egress_ended` webhook): `success` when its media file is available, `failure` when it was aborted or failed, `unknown` for a status the backend does not map | `service.origin.name` is `livekit`. `entity.target` is the recording as it was before the report was processed. `lasuite.details.worker_event` (`completed`, `limit reached`, `aborted`, `failed`) and `error_code` |
+| `recording.delete` | A recording is deleted, or the attempt fails | `entity.target` is the recording |
+| `recording.transcript.request` | The backend sends a recording to the summary service to be transcribed, or fails to | `lasuite.actor.type` is `system`, `service.target.name` is `summary`, and `entity.target` is the recording. `lasuite.details.summary_requested` (whether a summary of the transcript is to be made too) and `job_id` |
+| `recording.transcript.report` | The summary service reports on a transcript (`external-process-hook`), or a call to that hook is refused | `service.origin.name` is `summary`. `entity.target` is the recording, absent for an unknown job. `lasuite.details.job_id` and `status`. A failure the service reports is recorded as a `failure` |
+| `recording.summary.report` | The summary service reports on a summary | Same as `recording.transcript.report` |
+
+A recording's target says what was recorded:
+
+- `mode` is how the media was captured: `screen_recording` for a video, `transcript` for an audio track;
+- `requested_mode` is what the user asked for;
+- `is_transcribed` says whether it is sent to the summary service.
+
+`mode` and `requested_mode` differ for a transcript started with a screen capture: `mode` is then `screen_recording`,
+and `requested_mode` is `transcript`.
+
+Transcripts and summaries are not recordings, but what the summary service derives from one. They are audited under
+`recording.transcript.*` and `recording.summary.*`, with the recording as their target.
+
 ### Sign-in and the admin
 
 | `event.action` | Emitted when | Notable fields |

src/backend/core/api/viewsets.py

@@ -40,7 +40,7 @@ from rest_framework import (
 )
 from rest_framework.settings import api_settings
 
-from core import analytics, enums, models, utils
+from core import analytics, audit, auditing, enums, models, utils
 from core.api import throttling
 from core.api.filters import ListFileFilter
 from core.enums import MEDIA_STORAGE_URL_PATTERN
@@ -262,6 +262,7 @@ class UserViewSet(
 
 
 class RoomViewSet(
+    audit.AuditViewMixin,
     mixins.CreateModelMixin,
     mixins.DestroyModelMixin,
     mixins.UpdateModelMixin,
@@ -406,6 +407,7 @@ class RoomViewSet(
         permission_classes=[
             permissions.HasPrivilegesOnRoom,
         ],
+        audit_action=auditing.RECORDING_START,
     )
     @FeatureFlag.require("recording")
     def start_room_recording(self, request, pk=None):  # pylint: disable=unused-argument
@@ -443,6 +445,11 @@ class RoomViewSet(
                     recording=recording,
                 )
 
+            self.audit_target = recording
+            self.audit_details = {
+                "collect_metadata": bool(recording.options.get("collect_metadata")),
+            }
+
         except (DjangoValidationError, IntegrityError):
             # DjangoValidationError covers the Python-level check (full_clean);
             # IntegrityError covers the race where two concurrent requests both
@@ -487,6 +494,7 @@ class RoomViewSet(
         permission_classes=[
             permissions.HasPrivilegesOnRoom,
         ],
+        audit_action=auditing.RECORDING_STOP,
     )
     @FeatureFlag.require("recording")
     def stop_room_recording(self, request, pk=None):  # pylint: disable=unused-argument
@@ -503,6 +511,8 @@ class RoomViewSet(
                 "No active recording found for this room."
             ) from e
 
+        self.audit_target = recording
+
         worker_service = get_worker_service(mode=recording.mode)
         worker_manager = WorkerServiceMediator(worker_service=worker_service)
 
@@ -1041,6 +1051,7 @@ class ResourceAccessViewSet(
 
 
 class RecordingViewSet(
+    audit.AuditViewMixin,
     mixins.DestroyModelMixin,
     mixins.ListModelMixin,
     mixins.RetrieveModelMixin,
@@ -1055,6 +1066,23 @@ class RecordingViewSet(
     queryset = models.Recording.objects.all()
     serializer_class = serializers.RecordingSerializer
 
+    audit_actions = {"destroy": auditing.RECORDING_DELETE}
+    # What the summary service reports on, once its event is validated
+    audit_process_type = None
+
+    def get_audit_action(self):
+        """Audit a summary apart from the transcript it is made from."""
+        if self.audit_process_type == "summary":
+            return auditing.RECORDING_SUMMARY_REPORT
+        return super().get_audit_action()
+
+    def get_audit_fields(self, status_code, error=None):
+        """Report a failure of the summary service as one, though it is acknowledged."""
+        fields = super().get_audit_fields(status_code, error)
+        if status_code < 400 and (self.audit_details or {}).get("status") == "failure":
+            fields["outcome"] = audit.Outcome.FAILURE
+        return fields
+
     def get_queryset(self):
         """Restrict recordings to the user's ones."""
         user = self.request.user
@@ -1070,6 +1098,7 @@ class RecordingViewSet(
         url_path="external-process-hook",
         authentication_classes=[RecordingProcessWebhookAuthentication],
         serializer_class=serializers.ExternalProcessEventSerializer,
+        audit_action=auditing.RECORDING_TRANSCRIPT_REPORT,
     )
     def on_external_process_event_received(self, request, pk=None):  # pylint: disable=unused-argument
         """Handle incoming external process events for recordings."""
@@ -1083,12 +1112,16 @@ class RecordingViewSet(
 
         validated_data = serializer.validated_data
         job_id = validated_data["job_id"]
+        self.audit_process_type = validated_data.get("type")
+        self.audit_details = {"job_id": job_id, "status": validated_data.get("status")}
         try:
             recording = models.Recording.objects.get(external_process_id=job_id)
         except models.Recording.DoesNotExist as e:
             logger.warning("No recording found for job_id %s: %s", job_id, e)
             return ok_response
 
+        self.audit_target = recording
+
         if validated_data.get("type") == "transcript":
             if validated_data.get("status") == "success":
                 logger.info(

src/backend/core/audit/drf.py

@@ -18,6 +18,7 @@ A refusal is recorded under the action that was attempted, with its outcome
 and reason derived from the response status.
 """
 
+import copy
 import logging
 from collections.abc import Mapping
 from typing import Any
@@ -94,6 +95,11 @@ class AuditViewMixin:
         self.audit_target = obj
         super().check_object_permissions(request, obj)
 
+    def perform_destroy(self, instance):
+        """Keep a copy of the target, since deleting an instance clears its pk."""
+        self.audit_target = copy.copy(instance)
+        super().perform_destroy(instance)
+
     def finalize_response(self, request, response, *args, **kwargs):
         """Audit the response once DRF has built it."""
         response = super().finalize_response(request, response, *args, **kwargs)

src/backend/core/auditing.py

@@ -34,6 +34,19 @@ ROOM_CREATE = audit.Action("room.create")
 ROOM_LIST = audit.Action("room.list")
 ROOM_RETRIEVE = audit.Action("room.retrieve")
 ROOM_UPDATE = audit.Action("room.update")
+RECORDING_START = audit.Action("recording.start", types=(EventType.START,))
+RECORDING_STOP = audit.Action("recording.stop", types=(EventType.END,))
+RECORDING_END = audit.Action("recording.end", types=(EventType.END,))
+RECORDING_DELETE = audit.Action("recording.delete", types=(EventType.DELETION,))
+RECORDING_TRANSCRIPT_REQUEST = audit.Action(
+    "recording.transcript.request", types=(EventType.START,)
+)
+RECORDING_TRANSCRIPT_REPORT = audit.Action(
+    "recording.transcript.report", types=(EventType.END,)
+)
+RECORDING_SUMMARY_REPORT = audit.Action(
+    "recording.summary.report", types=(EventType.END,)
+)
 
 # Models
 
@@ -80,7 +93,7 @@ audit.register(
 )
 audit.register(
     models.Recording,
-    fields=("room_id", "status", "mode"),
+    fields=("room_id", "status", "mode", "requested_mode", "is_transcribed"),
     admin_values=("status", "mode"),
 )
 audit.register(models.File, admin_values=("title", "upload_state"))

src/backend/core/models.py

@@ -706,6 +706,22 @@ class Recording(BaseModel):
         """Check if the recording is in a saved state."""
         return self.status in RecordingStatusChoices.saved_statuses()
 
+    @property
+    def requested_mode(self) -> str:
+        """Return the mode the user asked for.
+
+        A transcript started with a screen capture is stored as a screen
+        recording, which keeps the transcript as its original mode.
+        """
+        return (self.options or {}).get("original_mode") or self.mode
+
+    @property
+    def is_transcribed(self) -> bool:
+        """Tell whether the recording is sent to the summary service."""
+        return self.mode == RecordingModeChoices.TRANSCRIPT or bool(
+            (self.options or {}).get("transcribe", False)
+        )
+
     @property
     def extension(self):
         """Get recording extension based on its mode."""

src/backend/core/recording/event/notification.py

@@ -16,7 +16,7 @@ from asgiref.sync import async_to_sync
 from celery.exceptions import OperationalError as CeleryOperationalError
 from livekit import api as livekit_api
 
-from core import models, utils
+from core import audit, auditing, models, utils
 from core.analytics import UserFeatureFlag, is_user_feature_flag_enabled
 from core.tasks.mail import send_email
 from core.utils import generate_download_s3_url
@@ -52,7 +52,7 @@ class NotificationService:
 
         if recording.mode == models.RecordingModeChoices.SCREEN_RECORDING:
             summary_success = True
-            if recording.options.get("transcribe", False):
+            if recording.is_transcribed:
                 summary_success = self._notify_summary_service(recording)
 
             email_success = self._notify_user_by_email(recording)
@@ -198,14 +198,32 @@ class NotificationService:
 
     @staticmethod
     def _notify_summary_service(recording: models.Recording):
-        if settings.SUMMARY_SERVICE_VERSION == 1:
-            return NotificationService._notify_summary_service_v1(recording)
-        if settings.SUMMARY_SERVICE_VERSION == 2:
-            return NotificationService._notify_summary_service_v2(recording)
-
-        raise NotImplementedError(
-            f"Unknown summary service version: {settings.SUMMARY_SERVICE_VERSION}"
-        )
+        succeeded = False
+        audit_details = {}
+        try:
+            if settings.SUMMARY_SERVICE_VERSION == 1:
+                succeeded = NotificationService._notify_summary_service_v1(recording)
+            elif settings.SUMMARY_SERVICE_VERSION == 2:
+                succeeded = NotificationService._notify_summary_service_v2(
+                    recording, audit_details
+                )
+            else:
+                raise NotImplementedError(
+                    "Unknown summary service version: "
+                    f"{settings.SUMMARY_SERVICE_VERSION}"
+                )
+            return succeeded
+        finally:
+            audit.log(
+                auditing.RECORDING_TRANSCRIPT_REQUEST,
+                actor=None,
+                actor_type=audit.ActorType.SYSTEM,
+                target=recording,
+                target_service="summary",
+                outcome=audit.Outcome.SUCCESS if succeeded else audit.Outcome.FAILURE,
+                job_id=recording.external_process_id,
+                **audit_details,
+            )
 
     @staticmethod
     def _notify_summary_service_v1(recording: models.Recording):
@@ -283,8 +301,11 @@ class NotificationService:
         return True
 
     @staticmethod
-    def _notify_summary_service_v2(recording: models.Recording):
-        """Notify summary service about a new recording."""
+    def _notify_summary_service_v2(recording: models.Recording, audit_details=None):
+        """Notify summary service about a new recording.
+
+        Whether a summary is asked for is added to ``audit_details``.
+        """
 
         if (
             not settings.SUMMARY_SERVICE_ENDPOINT
@@ -334,6 +355,12 @@ class NotificationService:
                 "ended_at": ended_at.isoformat(),
             }
 
+        summary_requested = is_user_feature_flag_enabled(
+            owner_access.user, UserFeatureFlag.TRANSCRIPT_SUMMARY_ENABLED
+        )
+        if audit_details is not None:
+            audit_details["summary_requested"] = summary_requested
+
         payload = {
             "user_sub": owner_access.user.sub,
             "user_email": owner_access.user.email,
@@ -357,9 +384,7 @@ class NotificationService:
                 ),
                 "download_link": f"{get_recording_download_base_url()}/{recording.id}",
                 "form_link": form_link,
-                "auto_create_summary": is_user_feature_flag_enabled(
-                    owner_access.user, UserFeatureFlag.TRANSCRIPT_SUMMARY_ENABLED
-                ),
+                "auto_create_summary": summary_requested,
             },
             "metadata": metadata_payload,
         }

src/backend/core/recording/services/recording_events.py

@@ -84,7 +84,7 @@ class RecordingEventsService:
     @staticmethod
     def _notify_participants(recording: Recording, event: RecordingWorkerEvent):
         """Notify the room's participants that a recording ended on the given event."""
-        recording_mode = recording.options.get("original_mode", None) or recording.mode
+        recording_mode = recording.requested_mode
 
         notification_type = get_notification_type(recording_mode, event)
         if not notification_type:

src/backend/core/recording/worker/mediator.py

@@ -68,7 +68,7 @@ class WorkerServiceMediator:
         finally:
             recording.save()
 
-        mode = recording.options.get("original_mode", None) or recording.mode
+        mode = recording.requested_mode
 
         try:
             RoomManagement.update_metadata(

src/backend/core/services/livekit_events.py

@@ -12,8 +12,9 @@ from django.utils import timezone
 
 from livekit import api
 
-from core import models
-from core.recording.enums import RecordingWorkerEvent
+from core import audit, auditing, models
+from core.recording.enums import UNSUCCESSFUL_EVENTS, RecordingWorkerEvent
+from core.recording.event.authentication import MachineUser
 from core.recording.services.metadata_collector import (
     MetadataCollectorException,
     MetadataCollectorService,
@@ -31,6 +32,8 @@ from .sip_management import SIPException, SIPManagement
 
 logger = getLogger(__name__)
 
+SERVICE_ORIGIN = "livekit"
+
 
 class LiveKitWebhookError(Exception):
     """Base exception for LiveKit webhook processing errors."""
@@ -219,6 +222,7 @@ class LiveKitEventsService:
             ) from err
 
         event = to_recording_event(data.egress_info.status)
+        self._audit_recording_end(recording, event, data.egress_info.error_code)
 
         # Log if/why the recording failed
         self.recording_events.log_worker_error(
@@ -254,6 +258,28 @@ class LiveKitEventsService:
 
         self.recording_events.handle_terminal_event(recording, event)
 
+    @staticmethod
+    def _audit_recording_end(recording, event, error_code=None):
+        """Audit the end of a recording, as LiveKit reports it.
+
+        An egress status the backend does not map has an unknown outcome.
+        """
+        if event is None:
+            outcome = audit.Outcome.UNKNOWN
+        elif event in UNSUCCESSFUL_EVENTS:
+            outcome = audit.Outcome.FAILURE
+        else:
+            outcome = audit.Outcome.SUCCESS
+        audit.log(
+            auditing.RECORDING_END,
+            actor=MachineUser(SERVICE_ORIGIN),
+            auth_method="shared_secret",
+            target=recording,
+            outcome=outcome,
+            worker_event=event.value if event is not None else None,
+            error_code=error_code or None,
+        )
+
     @staticmethod
     def _is_connection_test_room(room_name: str) -> bool:
         """Return True for ephemeral rooms created by the connection test endpoint."""

src/backend/core/tests/audit/test_ecs.py

@@ -1,12 +1,81 @@
 """Tests holding audit events to the Elastic Common Schema 9.5.0."""
 
+from collections.abc import Iterator, Mapping
+from typing import Any
+
 from django.apps import apps
+from django.test import RequestFactory
 
 import pytest
 
 from core import audit, auditing
 from core.audit import ecs
 from core.audit.admin import ADMIN_ACCESS_ACTION, AdminVerb, category_for, types_for
+from core.factories import ApplicationFactory, RoomFactory, UserFactory
+from core.recording.event.authentication import MachineUser
+
+pytestmark = pytest.mark.django_db
+
+# The ECS 9.5.0 fields audit events may carry, from
+# https://github.com/elastic/ecs/blob/v9.5.0/generated/ecs/ecs_flat.yml
+ECS_FIELDS = frozenset(
+    {
+        "@timestamp",
+        "client.ip",
+        "data_stream.dataset",
+        "data_stream.namespace",
+        "data_stream.type",
+        "ecs.version",
+        "entity.target.id",
+        "entity.target.name",
+        "entity.target.sub_type",
+        "entity.target.type",
+        "error.message",
+        "error.stack_trace",
+        "error.type",
+        "event.action",
+        "event.category",
+        "event.dataset",
+        "event.id",
+        "event.kind",
+        "event.outcome",
+        "event.reason",
+        "event.type",
+        "http.request.id",
+        "http.request.method",
+        "http.response.status_code",
+        "log.level",
+        "log.logger",
+        "message",
+        "organization.id",
+        "service.environment",
+        "service.name",
+        "service.node.name",
+        "service.origin.name",
+        "service.target.name",
+        "service.version",
+        "source.ip",
+        "url.path",
+        "user.domain",
+        "user.id",
+        "user.roles",
+        "user.target.domain",
+        "user.target.id",
+        "user_agent.original",
+    }
+)
+# What is not ECS lives in these namespaces: ``raw`` holds a target's own fields
+CUSTOM_NAMESPACES = ("lasuite.", "entity.target.raw.")
+
+
+def leaf_paths(document: Mapping[str, Any], prefix: str = "") -> Iterator[str]:
+    """Yield the dotted path of every value of a document."""
+    for key, value in document.items():
+        path = f"{prefix}{key}"
+        if isinstance(value, Mapping):
+            yield from leaf_paths(value, f"{path}.")
+        else:
+            yield path
 
 
 def declared_actions() -> list[audit.Action]:
@@ -55,3 +124,48 @@ def test_admin_writes_are_classified_as_ecs_expects(verb):
     """Writes made through the admin are classified as ECS expects, for any model."""
     for model in apps.get_models():
         ecs.check_classification([category_for(model)], types_for(model, verb))
+
+
+def _documents(audit_events) -> list[dict[str, Any]]:
+    """Emit the events covering every field the facility fills."""
+    user = UserFactory(is_staff=True)
+    request = RequestFactory().post(
+        "/external-api/v1.0/rooms/",
+        REMOTE_ADDR="1.2.3.4",
+        HTTP_USER_AGENT="Mozilla/5.0",
+    )
+    request.user = user
+    request.auth = {"client_id": "app-1"}
+
+    audit.log(
+        auditing.ROOM_UPDATE,
+        request=request,
+        target=RoomFactory(),
+        status_code=500,
+        outcome=audit.Outcome.FAILURE,
+        reason=audit.Reason.INTERNAL_ERROR,
+        error="boom",
+        error_type="builtins.RuntimeError",
+        message="anything",
+        updated_fields=["name"],
+    )
+    audit.log(auditing.USER_PROVISION, target=user)
+    audit.log(auditing.APPLICATION_TOKEN_ISSUE, target=ApplicationFactory())
+    audit.log(
+        auditing.RECORDING_TRANSCRIPT_REQUEST,
+        actor=MachineUser("livekit"),
+        target_service="summary",
+    )
+    return audit_events
+
+
+def test_documents_only_carry_ecs_fields_or_custom_namespaces(audit_events):
+    """No field outside ECS 9.5.0 lands anywhere but in a custom namespace."""
+    for document in _documents(audit_events):
+        stray = {
+            path
+            for path in leaf_paths(document)
+            if path not in ECS_FIELDS and not path.startswith(CUSTOM_NAMESPACES)
+        }
+
+        assert not stray, f"{document['event']['action']}: {sorted(stray)}"

src/backend/core/tests/recording/event/test_notification.py

@@ -16,6 +16,7 @@ from celery.exceptions import OperationalError as CeleryOperationalError
 
 from core import factories, models
 from core.analytics import UserFeatureFlag
+from core.audit.testing import find_events
 from core.recording.event.notification import NotificationService, notification_service
 
 pytestmark = pytest.mark.django_db
@@ -534,3 +535,64 @@ def test_notify_summary_service_v2_payload_json_serializable_without_timestamps(
     assert isinstance(title, str)
     # ...so the payload serializes exactly the way ``requests`` serializes it.
     json.dumps(payload)
+
+
+@pytest.mark.parametrize("summary_requested", [True, False])
+@mock.patch("core.recording.event.notification.requests.post")
+@mock.patch("core.recording.event.notification.generate_download_s3_url")
+@mock.patch.object(
+    NotificationService, "_get_recording_timestamps", new_callable=mock.AsyncMock
+)
+def test_notify_summary_service_is_audited(  # noqa: PLR0913, PLR0917
+    mock_get_recording_timestamps,
+    mock_generate_download_s3_url,
+    mock_post,
+    summary_requested,
+    settings,
+    audit_events,
+):  # pylint: disable=too-many-arguments,too-many-positional-arguments
+    """Sending a recording to be transcribed says whether a summary is asked for."""
+    settings.SUMMARY_SERVICE_VERSION = 2
+    settings.SUMMARY_SERVICE_ENDPOINT = "https://summary.test/api/v2/tasks"
+    settings.SUMMARY_SERVICE_API_TOKEN = "summary-token"
+    settings.METADATA_COLLECTOR_ENABLED = False
+
+    recording = factories.RecordingFactory(mode="transcript")
+    factories.UserRecordingAccessFactory(
+        recording=recording, role=models.RoleChoices.OWNER
+    )
+    mock_get_recording_timestamps.return_value = (None, None)
+    mock_generate_download_s3_url.return_value = "https://storage.test/recording.ogg"
+    mock_post.return_value.json.return_value = {"job_id": "job-7"}
+
+    with mock.patch(
+        "core.recording.event.notification.is_user_feature_flag_enabled",
+        return_value=summary_requested,
+    ):
+        assert NotificationService._notify_summary_service(recording) is True
+
+    [event] = find_events(audit_events, "recording.transcript.request")
+
+    assert event["lasuite"]["outcome"] == "success"
+    assert event["lasuite"]["actor"] == {"type": "system"}
+    assert event["entity"]["target"]["id"] == str(recording.id)
+    assert event["service"]["target"] == {"name": "summary"}
+    assert event["lasuite"]["details"] == {
+        "job_id": "job-7",
+        "summary_requested": summary_requested,
+    }
+
+
+def test_notify_summary_service_failure_is_audited(settings, audit_events):
+    """A recording the summary service never received is a failure."""
+    settings.SUMMARY_SERVICE_VERSION = 2
+    settings.SUMMARY_SERVICE_ENDPOINT = None
+
+    recording = factories.RecordingFactory(mode="transcript")
+
+    assert NotificationService._notify_summary_service(recording) is False
+
+    [event] = find_events(audit_events, "recording.transcript.request")
+
+    assert event["lasuite"]["outcome"] == "failure"
+    assert event["entity"]["target"]["id"] == str(recording.id)

src/backend/core/tests/recording/test_api_recordings_delete.py

@@ -5,6 +5,7 @@ Test recordings API endpoints in the Meet core app: delete.
 import pytest
 from rest_framework.test import APIClient
 
+from ...audit.testing import capture_audit, find_events
 from ...factories import RecordingFactory, UserFactory, UserRecordingAccessFactory
 from ...models import Recording
 
@@ -112,3 +113,24 @@ def test_api_recordings_delete_final(role):
 
     assert response.status_code == 204
     assert Recording.objects.count() == 0
+
+
+def test_api_recordings_delete_is_audited():
+    """A deleted recording is still identified by the event."""
+    user = UserFactory()
+    recording = RecordingFactory(status="saved", mode="transcript")
+    UserRecordingAccessFactory(role="owner", user=user, recording=recording)
+    client = APIClient()
+    client.force_login(user)
+
+    with capture_audit() as events:
+        response = client.delete(f"/api/v1.0/recordings/{recording.id}/")
+
+    assert response.status_code == 204
+
+    [event] = find_events(events, "recording.delete")
+
+    assert event["event"]["type"] == ["deletion"]
+    assert event["lasuite"]["outcome"] == "success"
+    assert event["entity"]["target"]["id"] == str(recording.id)
+    assert event["entity"]["target"]["raw"]["mode"] == "transcript"

src/backend/core/tests/recording/test_api_recordings_external_process_hook.py

@@ -6,6 +6,7 @@ Test recordings API endpoints: external process hook.
 
 import pytest
 
+from ...audit.testing import find_events
 from ...factories import RecordingFactory
 from ...models import RecordingStatusChoices
 
@@ -132,3 +133,83 @@ def test_external_process_event_non_transcript_event_does_not_change_status(
 
     recording.refresh_from_db()
     assert recording.status == RecordingStatusChoices.SAVED
+
+
+@pytest.mark.parametrize(
+    "process_type,action",
+    [
+        ("transcript", "recording.transcript.report"),
+        ("summary", "recording.summary.report"),
+    ],
+)
+@pytest.mark.parametrize(
+    "process_status,outcome", [("success", "success"), ("failure", "failure")]
+)
+def test_external_process_event_is_audited(  # noqa: PLR0913, PLR0917
+    external_process_settings,
+    client,
+    audit_events,
+    process_type,
+    action,
+    process_status,
+    outcome,
+):  # pylint: disable=too-many-arguments,too-many-positional-arguments
+    """A transcript and a summary are audited apart, as reported by the service."""
+    recording = RecordingFactory(
+        status=RecordingStatusChoices.SAVED, external_process_id="job-123"
+    )
+
+    response = client.post(
+        "/api/v1.0/recordings/external-process-hook/",
+        {"job_id": "job-123", "type": process_type, "status": process_status},
+        HTTP_AUTHORIZATION="Bearer testWebhookToken",
+    )
+
+    assert response.status_code == 200
+
+    [event] = find_events(audit_events, action)
+
+    assert event["lasuite"]["outcome"] == outcome
+    assert event["lasuite"]["actor"] == {"type": "service"}
+    assert event["service"]["origin"] == {"name": "summary"}
+    assert event["entity"]["target"]["id"] == str(recording.id)
+    assert event["lasuite"]["details"] == {
+        "job_id": "job-123",
+        "status": process_status,
+    }
+
+
+def test_external_process_event_unknown_recording_is_audited(
+    external_process_settings, client, audit_events
+):
+    """An event about an unknown job keeps its job id, with no target."""
+    response = client.post(
+        "/api/v1.0/recordings/external-process-hook/",
+        {"job_id": "job-unknown", "type": "summary", "status": "success"},
+        HTTP_AUTHORIZATION="Bearer testWebhookToken",
+    )
+
+    assert response.status_code == 200
+
+    [event] = find_events(audit_events, "recording.summary.report")
+
+    assert "target" not in event["lasuite"]
+    assert event["lasuite"]["details"]["job_id"] == "job-unknown"
+
+
+def test_external_process_event_refused_is_audited(
+    external_process_settings, client, audit_events
+):
+    """A caller without the token is denied under the transcript action."""
+    response = client.post(
+        "/api/v1.0/recordings/external-process-hook/",
+        {"job_id": "job-1", "type": "summary", "status": "success"},
+        HTTP_AUTHORIZATION="Bearer wrongToken",
+    )
+
+    assert response.status_code == 401
+
+    [event] = find_events(audit_events, "recording.transcript.report")
+
+    assert event["lasuite"]["outcome"] == "denied"
+    assert not find_events(audit_events, "recording.summary.report")

src/backend/core/tests/rooms/test_api_rooms_start_recording.py

@@ -2,7 +2,7 @@
 Test rooms API endpoints in the Meet core app: start recording.
 """
 
-# pylint: disable=redefined-outer-name,unused-argument,no-member
+# pylint: disable=redefined-outer-name,unused-argument,no-member,too-many-lines
 
 from unittest import mock
 
@@ -10,6 +10,7 @@ import pytest
 from livekit import api as livekit_api
 from rest_framework.test import APIClient
 
+from ...audit.testing import capture_audit, find_events
 from ...factories import RoomFactory, UserFactory
 from ...models import Recording
 from ...recording.worker.exceptions import RecordingStartError
@@ -879,3 +880,144 @@ def test_start_recording_options_original_mode_invalid(settings, value):
     )
 
     assert response.status_code == 400
+
+
+def test_start_recording_is_audited(
+    settings, mock_worker_service_factory, mock_worker_manager, audit_events
+):
+    """A started screen recording names the recording and its mode."""
+    settings.RECORDING_ENABLE = True
+    room = RoomFactory()
+    user = UserFactory()
+    room.accesses.create(user=user, role="owner")
+    client = APIClient()
+    client.force_login(user)
+
+    response = client.post(
+        f"/api/v1.0/rooms/{room.id}/start-recording/",
+        {"mode": "screen_recording"},
+    )
+
+    assert response.status_code == 201
+
+    recording = Recording.objects.get(room=room)
+    [event] = find_events(audit_events, "recording.start")
+
+    assert event["event"]["type"] == ["start"]
+    assert event["lasuite"]["outcome"] == "success"
+    assert event["user"]["id"] == str(user.pk)
+    assert event["entity"]["target"] == {
+        "id": str(recording.pk),
+        "sub_type": "recording",
+        "raw": {
+            "room_id": str(room.pk),
+            "status": "initiated",
+            "mode": "screen_recording",
+            "requested_mode": "screen_recording",
+            "is_transcribed": False,
+        },
+    }
+    assert event["lasuite"]["details"] == {"collect_metadata": False}
+
+
+def test_start_recording_transcript_with_screen_capture_is_audited(
+    settings, mock_worker_service_factory, mock_worker_manager, audit_events
+):
+    """A transcript recorded with the screen is told apart from a screen recording."""
+    settings.RECORDING_ENABLE = True
+    room = RoomFactory()
+    user = UserFactory()
+    room.accesses.create(user=user, role="owner")
+    client = APIClient()
+    client.force_login(user)
+
+    response = client.post(
+        f"/api/v1.0/rooms/{room.id}/start-recording/",
+        {
+            "mode": "screen_recording",
+            "options": {"transcribe": True, "original_mode": "transcript"},
+        },
+        format="json",
+    )
+
+    assert response.status_code == 201
+
+    [event] = find_events(audit_events, "recording.start")
+    target = event["entity"]["target"]["raw"]
+
+    assert target["mode"] == "screen_recording"
+    assert target["requested_mode"] == "transcript"
+    assert target["is_transcribed"] is True
+
+
+def test_start_recording_conflict_is_audited_on_the_room(
+    settings, mock_worker_service_factory, mock_worker_manager, audit_events
+):
+    """A conflicting start is a failure aimed at the room, no recording existing."""
+    settings.RECORDING_ENABLE = True
+    room = RoomFactory()
+    user = UserFactory()
+    room.accesses.create(user=user, role="owner")
+    Recording.objects.create(room=room, mode="screen_recording", status="active")
+    client = APIClient()
+    client.force_login(user)
+
+    response = client.post(
+        f"/api/v1.0/rooms/{room.id}/start-recording/",
+        {"mode": "transcript"},
+    )
+
+    assert response.status_code == 409
+
+    [event] = find_events(audit_events, "recording.start")
+
+    assert event["lasuite"]["outcome"] == "failure"
+    assert event["event"]["reason"] == "conflict"
+    assert event["entity"]["target"]["sub_type"] == "room"
+    assert event["entity"]["target"]["id"] == str(room.pk)
+
+
+def test_start_recording_worker_error_is_audited(
+    settings, mock_worker_service_factory, mock_worker_manager, audit_events
+):
+    """A recording the worker could not start is a failure on that recording."""
+    settings.RECORDING_ENABLE = True
+    room = RoomFactory()
+    user = UserFactory()
+    room.accesses.create(user=user, role="owner")
+    mock_worker_manager.start = mock.Mock(side_effect=RecordingStartError("boom"))
+    client = APIClient()
+    client.force_login(user)
+
+    response = client.post(
+        f"/api/v1.0/rooms/{room.id}/start-recording/",
+        {"mode": "transcript"},
+    )
+
+    assert response.status_code == 502
+
+    [event] = find_events(audit_events, "recording.start")
+
+    assert event["lasuite"]["outcome"] == "failure"
+    assert event["http"]["response"]["status_code"] == 502
+    assert event["entity"]["target"]["id"] == str(Recording.objects.get().pk)
+    assert event["entity"]["target"]["raw"]["mode"] == "transcript"
+    assert event["entity"]["target"]["raw"]["is_transcribed"] is True
+
+
+def test_start_recording_anonymous_is_audited():
+    """An anonymous attempt is denied."""
+    room = RoomFactory()
+
+    with capture_audit() as events:
+        response = APIClient().post(
+            f"/api/v1.0/rooms/{room.id}/start-recording/",
+            {"mode": "screen_recording"},
+        )
+
+    assert response.status_code == 401
+
+    [event] = find_events(events, "recording.start")
+
+    assert event["lasuite"]["outcome"] == "denied"
+    assert event["lasuite"]["actor"]["type"] == "anonymous"

src/backend/core/tests/rooms/test_api_rooms_stop_recording.py

@@ -9,6 +9,7 @@ from unittest import mock
 import pytest
 from rest_framework.test import APIClient
 
+from ...audit.testing import find_events
 from ...factories import RecordingFactory, RoomFactory, UserFactory
 from ...models import Recording, RecordingStatusChoices
 from ...recording.worker.exceptions import RecordingStopError
@@ -181,3 +182,48 @@ def test_stop_recording_success(
 
     # Verify the recording still exists
     assert Recording.objects.count() == 1
+
+
+def test_stop_recording_is_audited(
+    settings, mock_worker_service_factory, mock_worker_manager, audit_events
+):
+    """A stopped recording is the target of the event."""
+    settings.RECORDING_ENABLE = True
+    room = RoomFactory()
+    user = UserFactory()
+    room.accesses.create(user=user, role="owner")
+    recording = RecordingFactory(
+        room=room, mode="transcript", status=RecordingStatusChoices.ACTIVE
+    )
+    client = APIClient()
+    client.force_login(user)
+
+    response = client.post(f"/api/v1.0/rooms/{room.id}/stop-recording/")
+
+    assert response.status_code == 200
+
+    [event] = find_events(audit_events, "recording.stop")
+
+    assert event["event"]["type"] == ["end"]
+    assert event["lasuite"]["outcome"] == "success"
+    assert event["entity"]["target"]["id"] == str(recording.pk)
+    assert event["entity"]["target"]["raw"]["mode"] == "transcript"
+
+
+def test_stop_recording_without_active_recording_is_audited(settings, audit_events):
+    """Stopping a room that records nothing fails on the room."""
+    settings.RECORDING_ENABLE = True
+    room = RoomFactory()
+    user = UserFactory()
+    room.accesses.create(user=user, role="owner")
+    client = APIClient()
+    client.force_login(user)
+
+    response = client.post(f"/api/v1.0/rooms/{room.id}/stop-recording/")
+
+    assert response.status_code == 404
+
+    [event] = find_events(audit_events, "recording.stop")
+
+    assert event["event"]["reason"] == "not_found"
+    assert event["entity"]["target"]["sub_type"] == "room"

src/backend/core/tests/services/test_livekit_events.py

@@ -1,7 +1,7 @@
 """
 Test LiveKitEvents service.
 """
-# pylint: disable=W0621,W0613, W0212, E0611
+# pylint: disable=W0621,W0613, W0212, E0611, too-many-lines
 
 import logging
 import uuid
@@ -13,6 +13,7 @@ from django.utils import timezone
 import pytest
 from livekit.api import EgressStatus
 
+from core.audit.testing import find_events
 from core.factories import RecordingFactory, RoomFactory
 from core.models import Room
 from core.recording.enums import RecordingWorkerEvent
@@ -994,3 +995,76 @@ def test_participant_left_without_identity_is_ignored(mock_delete, service, sett
 
     service._handle_participant_left(data)  # pylint: disable=protected-access
     mock_delete.assert_not_called()
+
+
+@pytest.mark.parametrize(
+    ("egress_status", "worker_event", "outcome"),
+    (
+        (EgressStatus.EGRESS_COMPLETE, "completed", "success"),
+        (EgressStatus.EGRESS_LIMIT_REACHED, "limit reached", "success"),
+        (EgressStatus.EGRESS_ABORTED, "aborted", "failure"),
+        (EgressStatus.EGRESS_FAILED, "failed", "failure"),
+    ),
+)
+@mock.patch(
+    "core.recording.services.recording_events.notification_service."
+    "notify_external_services"
+)
+@mock.patch("core.utils.notify_participants")
+@mock.patch("core.services.room_management.RoomManagement.update_metadata")
+def test_handle_egress_ended_is_audited(  # noqa: PLR0913, PLR0917
+    mock_update_metadata,
+    mock_notify,
+    mock_notify_external_services,
+    egress_status,
+    worker_event,
+    outcome,
+    service,
+    audit_events,
+):  # pylint: disable=too-many-arguments,too-many-positional-arguments
+    """The end of a recording is audited as LiveKit reports it."""
+
+    recording = RecordingFactory(
+        worker_id="worker-1",
+        status="active",
+        mode="screen_recording",
+        options={"transcribe": True},
+    )
+    mock_data = mock.MagicMock()
+    mock_data.egress_info.egress_id = recording.worker_id
+    mock_data.egress_info.status = egress_status
+    mock_data.egress_info.error_code = 0
+
+    service._handle_egress_ended(mock_data)
+
+    [event] = find_events(audit_events, "recording.end")
+
+    assert event["event"]["type"] == ["end"]
+    assert event["lasuite"]["outcome"] == outcome
+    assert event["lasuite"]["actor"] == {"type": "service"}
+    assert event["service"]["origin"] == {"name": "livekit"}
+    assert event["lasuite"]["auth"] == {"method": "shared_secret"}
+    assert event["entity"]["target"]["id"] == str(recording.id)
+    assert event["entity"]["target"]["raw"]["is_transcribed"] is True
+    assert event["lasuite"]["details"] == {"worker_event": worker_event}
+
+
+@mock.patch("core.services.room_management.RoomManagement.update_metadata")
+def test_handle_egress_ended_with_an_unmapped_status_is_audited_as_unknown(
+    mock_update_metadata, service, audit_events
+):
+    """An egress status the backend does not map has an unknown outcome."""
+    recording = RecordingFactory(worker_id="worker-1", status="active")
+    mock_data = mock.MagicMock()
+    mock_data.egress_info.egress_id = recording.worker_id
+    mock_data.egress_info.status = 999
+    mock_data.egress_info.error_code = 0
+
+    service._handle_egress_ended(mock_data)
+
+    [event] = find_events(audit_events, "recording.end")
+
+    assert event["event"]["outcome"] == "unknown"
+    assert event["lasuite"]["outcome"] == "unknown"
+    assert event["log"]["level"] == "warning"
+    assert "details" not in event["lasuite"]