✨(backend) add metadata collection of VAD, connection and chat events

Introduce MetadataCollector and MetadataCollectorService classes to
centralize the collection and storage of user connections, VAD events,
and chat messages. This creates a structured foundation for future speaker
assignment logic based on voice activity detection. Add tests for this new
feature.

CHANGELOG.md

@@ -8,6 +8,10 @@ and this project adheres to
 
 ## [Unreleased]
 
+### Added
+
+- ✨(backend) add metadata collection of VAD, connection and chat events
+
 ## [1.14.0] - 2026-04-16
 
 ### Added

env.d/development/common.dist

@@ -71,6 +71,9 @@ RECORDING_DOWNLOAD_BASE_URL=http://localhost:3000/recording
 # Telephony
 ROOM_TELEPHONY_ENABLED=True
 
+# Metadata
+METADATA_COLLECTOR_ENABLED=True
+
 FRONTEND_USE_FRENCH_GOV_FOOTER=False
 FRONTEND_USE_PROCONNECT_BUTTON=False
 

src/agents/exceptions.py

@@ -0,0 +1,5 @@
+"""Storage parsers specific exceptions."""
+
+
+class MissingConfigError(Exception):
+    """Raised when a variable is not set in configuration."""

src/agents/metadata_collector.py

@@ -0,0 +1,382 @@
+"""Metadata agent that extracts metadata from active room."""
+
+import asyncio
+import json
+import logging
+import os
+from dataclasses import asdict, dataclass
+from datetime import datetime, timezone
+from io import BytesIO
+from typing import List, Optional
+
+from dotenv import load_dotenv
+from livekit import api, rtc
+from livekit.agents import (
+    Agent,
+    AgentServer,
+    AgentSession,
+    AutoSubscribe,
+    JobContext,
+    JobProcess,
+    JobRequest,
+    RoomInputOptions,
+    RoomIO,
+    RoomOutputOptions,
+    WorkerPermissions,
+    cli,
+    utils,
+)
+from livekit.plugins import silero
+from minio import Minio
+from minio.error import S3Error
+
+from exceptions import MissingConfigError
+
+load_dotenv()
+
+logger = logging.getLogger("metadata-collector")
+
+AGENT_NAME = os.getenv("METADATA_COLLECTOR_AGENT_NAME", "metadata-collector")
+
+
+def prewarm(proc: JobProcess):
+    """Preload voice activity detection model."""
+    proc.userdata["vad"] = silero.VAD.load()
+
+
+server = AgentServer(
+    permissions=WorkerPermissions(
+        can_publish=False,
+        can_publish_data=False,
+        can_subscribe=True,
+        hidden=True,
+    ),
+)
+server.setup_fnc = prewarm
+
+
+@dataclass
+class MetadataEvent:
+    """A single timestamped event recorded during a meeting."""
+
+    participant_id: str
+    type: str
+    timestamp: datetime
+    data: Optional[str] = None
+
+    def serialize(self) -> dict:
+        """Return a JSON-serializable dictionary representation of the event."""
+        data = asdict(self)
+        data["timestamp"] = self.timestamp.isoformat()
+        return data
+
+
+class VADAgent(Agent):
+    """Agent that monitors voice activity for a specific participant."""
+
+    def __init__(self, participant_identity: str, events: List):
+        """Initialize with a participant identity and shared events list."""
+        super().__init__(
+            instructions="not-needed",
+        )
+        self.participant_identity = participant_identity
+        self.events = events
+
+    async def on_enter(self) -> None:
+        """Initialize VAD monitoring for this participant."""
+
+        @self.session.on("user_state_changed")
+        def on_user_state(event):
+            timestamp = datetime.now(timezone.utc)
+
+            if event.new_state == "speaking":
+                event = MetadataEvent(
+                    participant_id=self.participant_identity,
+                    type="speech_start",
+                    timestamp=timestamp,
+                )
+                self.events.append(event)
+
+            elif event.old_state == "speaking":
+                event = MetadataEvent(
+                    participant_id=self.participant_identity,
+                    type="speech_end",
+                    timestamp=timestamp,
+                )
+                self.events.append(event)
+
+
+class MetadataCollector:
+    """Collect meeting events across all participants in a room.
+
+    Creates one AgentSession per participant to capture VAD events
+    (speech start/end), and listens for connection, disconnection,
+    and chat events. Persists all collected events as JSON to S3
+    on shutdown.
+    """
+
+    def __init__(self, ctx: JobContext, recording_id: str):
+        """Initialize metadata agent."""
+        self.minio_client = Minio(
+            endpoint=os.getenv("AWS_S3_ENDPOINT_URL"),
+            access_key=os.getenv("AWS_S3_ACCESS_KEY_ID"),
+            secret_key=os.getenv("AWS_S3_SECRET_ACCESS_KEY"),
+            secure=os.getenv("AWS_S3_SECURE_ACCESS", "False").lower() == "true",
+        )
+
+        if (bucket_name := os.getenv("AWS_STORAGE_BUCKET_NAME")) is not None:
+            self.bucket_name = bucket_name
+        else:
+            raise MissingConfigError
+
+        self.ctx = ctx
+        self._sessions: dict[str, AgentSession] = {}
+        self._tasks: set[asyncio.Task] = set()
+
+        output_folder = os.getenv("AWS_S3_OUTPUT_FOLDER", "metadata")
+        self.output_filename = f"{output_folder}/{recording_id}-metadata.json"
+
+        # Storage for events
+        self.events = []
+        self.participants = {}
+
+        logger.info("MetadataCollector initialized")
+
+    def start(self):
+        """Start listening for room-level events."""
+        self.ctx.room.on("participant_disconnected", self.on_participant_disconnected)
+        self.ctx.room.on("participant_name_changed", self.on_participant_name_changed)
+
+        self.ctx.room.register_text_stream_handler("lk.chat", self.handle_chat_stream)
+
+        logger.info("Started listening for participant events")
+
+    async def on_chat_message_received(
+        self, reader: rtc.TextStreamReader, participant_identity: str
+    ):
+        """Read a complete chat message and record it as an event."""
+        full_text = await reader.read_all()
+        logger.info("Received chat message from %s", participant_identity)
+
+        self.events.append(
+            MetadataEvent(
+                participant_id=participant_identity,
+                type="chat_received",
+                timestamp=datetime.now(timezone.utc),
+                data=full_text,
+            )
+        )
+
+    def handle_chat_stream(self, reader, participant_identity):
+        """Schedule async processing of an incoming chat stream."""
+        task = asyncio.create_task(
+            self.on_chat_message_received(reader, participant_identity)
+        )
+        self._tasks.add(task)
+        task.add_done_callback(lambda _: self._tasks.remove(task))
+
+    def save(self):
+        """Serialize collected events and upload as JSON to S3."""
+        logger.info("Persisting metadata…")
+
+        participants = []
+        for k, v in self.participants.items():
+            participants.append({"participantId": k, "name": v})
+
+        sorted_events = sorted(self.events, key=lambda e: e.timestamp)
+
+        payload = {
+            "events": [event.serialize() for event in sorted_events],
+            "participants": participants,
+        }
+
+        data = json.dumps(payload, indent=2).encode("utf-8")
+        stream = BytesIO(data)
+
+        try:
+            self.minio_client.put_object(
+                self.bucket_name,
+                self.output_filename,
+                stream,
+                length=len(data),
+                content_type="application/json",
+            )
+            logger.info(
+                "Uploaded speaker meeting metadata",
+            )
+        except S3Error:
+            logger.exception(
+                "Failed to upload meeting metadata",
+            )
+
+    async def aclose(self):
+        """Close all sessions and cleanup resources."""
+        logger.info("Closing all VAD monitoring sessions…")
+
+        await utils.aio.cancel_and_wait(*self._tasks)
+
+        await asyncio.gather(
+            *[self._close_session(session) for session in self._sessions.values()],
+            return_exceptions=True,
+        )
+
+        self.ctx.room.off("participant_disconnected", self.on_participant_disconnected)
+        self.ctx.room.off("participant_name_changed", self.on_participant_name_changed)
+
+        logger.info("All VAD sessions closed")
+        self.save()
+
+    async def on_participant_entrypoint(
+        self, ctx: JobContext, participant: rtc.RemoteParticipant
+    ):
+        """Handle new participant by starting a VAD monitoring session."""
+        if participant.identity in self._sessions:
+            logger.debug("Session already exists for %s", participant.identity)
+            return
+
+        self.events.append(
+            MetadataEvent(
+                participant_id=participant.identity,
+                type="participant_connected",
+                timestamp=datetime.now(timezone.utc),
+            )
+        )
+
+        self.participants[participant.identity] = participant.name
+
+        logger.info("New participant connected: %s", participant.identity)
+        try:
+            session = await self._start_session(participant)
+            self._sessions[participant.identity] = session
+        except Exception:
+            logger.exception("Failed to start session for %s", participant.identity)
+
+    def on_participant_disconnected(self, participant: rtc.RemoteParticipant):
+        """Handle participant disconnection by closing VAD monitoring."""
+        self.events.append(
+            MetadataEvent(
+                participant_id=participant.identity,
+                type="participant_disconnected",
+                timestamp=datetime.now(timezone.utc),
+            )
+        )
+
+        session = self._sessions.pop(participant.identity, None)
+        if session is None:
+            logger.debug("No session found for %s", participant.identity)
+            return
+
+        logger.info("Participant disconnected: %s", participant.identity)
+        task = asyncio.create_task(self._close_session(session))
+        self._tasks.add(task)
+
+        def on_close_done(_):
+            self._tasks.discard(task)
+            logger.info(
+                "VAD session closed for %s (remaining sessions: %d)",
+                participant.identity,
+                len(self._sessions),
+            )
+
+        task.add_done_callback(on_close_done)
+
+    def on_participant_name_changed(self, participant: rtc.RemoteParticipant):
+        """Update stored participant name when it changes."""
+        logger.info("Participant's name changed: %s", participant.identity)
+        self.participants[participant.identity] = participant.name
+
+    async def _start_session(self, participant: rtc.RemoteParticipant) -> AgentSession:
+        """Create and start VAD monitoring session for participant."""
+        if participant.identity in self._sessions:
+            return self._sessions[participant.identity]
+
+        # Create session with VAD only - no STT, LLM, or TTS
+        session = AgentSession(
+            vad=self.ctx.proc.userdata["vad"],
+            turn_detection="vad",
+            user_away_timeout=30.0,
+        )
+
+        # Set up room IO to receive audio from this specific participant
+        room_io = RoomIO(
+            agent_session=session,
+            room=self.ctx.room,
+            participant=participant,
+            input_options=RoomInputOptions(
+                audio_enabled=True,
+                text_enabled=False,
+            ),
+            output_options=RoomOutputOptions(
+                audio_enabled=False,
+                transcription_enabled=False,
+            ),
+        )
+
+        await room_io.start()
+        await session.start(
+            agent=VADAgent(
+                participant_identity=participant.identity, events=self.events
+            )
+        )
+
+        return session
+
+    async def _close_session(self, session: AgentSession) -> None:
+        """Close and cleanup VAD monitoring session."""
+        try:
+            await session.drain()
+            await session.aclose()
+        except Exception:
+            logger.exception("Error closing session")
+
+
+async def handle_job_request(job_req: JobRequest) -> None:
+    """Accept or reject the job request based on agent presence in the room."""
+    room_name = job_req.room.name
+    recording_id = job_req.job.metadata
+    agent_identity = f"{AGENT_NAME}-{room_name}"
+
+    async with api.LiveKitAPI() as lk:
+        try:
+            resp = await lk.room.list_participants(
+                list=api.ListParticipantsRequest(room=room_name)
+            )
+            already_present = any(
+                p.kind == rtc.ParticipantKind.PARTICIPANT_KIND_AGENT
+                and p.identity == agent_identity
+                for p in resp.participants
+            )
+            if already_present:
+                logger.info("Agent already in the room '%s' — reject", room_name)
+                await job_req.reject()
+            else:
+                logger.info(
+                    "Accept job for '%s' — identity=%s", room_name, agent_identity
+                )
+                await job_req.accept(identity=agent_identity, metadata=recording_id)
+        except Exception:
+            logger.exception("Error treating the job for '%s'", room_name)
+            await job_req.reject()
+
+
+@server.rtc_session(agent_name=AGENT_NAME, on_request=handle_job_request)
+async def entrypoint(ctx: JobContext):
+    """Initialize and run the metadata collector."""
+    logger.info("Starting metadata agent in room: %s", ctx.room.name)
+    recording_id = ctx.job.metadata
+    metadata_collector = MetadataCollector(ctx, recording_id)
+    metadata_collector.start()
+
+    ctx.add_participant_entrypoint(metadata_collector.on_participant_entrypoint)
+
+    await ctx.connect(auto_subscribe=AutoSubscribe.AUDIO_ONLY)
+
+    async def cleanup():
+        logger.info("Shutting down metadata collector…")
+        await metadata_collector.aclose()
+
+    ctx.add_shutdown_callback(cleanup)
+
+
+if __name__ == "__main__":
+    cli.run_app(server)

src/agents/pyproject.toml

@@ -9,7 +9,8 @@ dependencies = [
     "livekit-plugins-silero==1.4.5",
     "livekit-plugins-kyutai-lasuite==0.0.6",
     "python-dotenv==1.2.2",
-    "protobuf==6.33.5"
+    "protobuf==6.33.5",
+    "minio==7.2.15"
 ]
 
 [project.optional-dependencies]
@@ -17,6 +18,9 @@ dev = [
     "ruff==0.15.6",
 ]
 
+[tool.setuptools]
+py-modules = ["multi_user_transcriber", "metadata_collector", "exceptions"]
+
 [build-system]
 requires = ["setuptools>=61.0"]
 build-backend = "setuptools.build_meta"

src/backend/core/api/viewsets.py

@@ -45,6 +45,10 @@ from core.recording.event.exceptions import (
 )
 from core.recording.event.notification import notification_service
 from core.recording.event.parsers import get_parser
+from core.recording.services.metadata_collector import (
+    MetadataCollectorException,
+    MetadataCollectorService,
+)
 from core.recording.worker.exceptions import (
     RecordingStartError,
     RecordingStopError,
@@ -338,6 +342,15 @@ class RoomViewSet(
                 status=drf_status.HTTP_500_INTERNAL_SERVER_ERROR,
             )
 
+        if settings.METADATA_COLLECTOR_ENABLED and (
+            recording.mode == models.RecordingModeChoices.TRANSCRIPT
+            or recording.options.get("transcribe", False)
+        ):
+            try:
+                MetadataCollectorService().start(recording)
+            except MetadataCollectorException:
+                logger.warning("Failed to start MetadataCollectorService")
+
         return drf_response.Response(
             {"message": f"Recording successfully started for room {room.slug}"},
             status=drf_status.HTTP_201_CREATED,

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

@@ -0,0 +1,91 @@
+"""Meeting metadata collection service."""
+
+from logging import getLogger
+
+from django.conf import settings
+
+from asgiref.sync import async_to_sync, sync_to_async
+from livekit.protocol.agent_dispatch import (
+    CreateAgentDispatchRequest,
+)
+
+from core import utils
+from core.models import Recording
+
+logger = getLogger(__name__)
+
+
+class MetadataCollectorException(Exception):
+    """Generic exception in the metadata collector."""
+
+
+class MetadataCollectorService:
+    """Service for dispatching and managing the metadata collector agent."""
+
+    @async_to_sync
+    async def start(self, recording: Recording):
+        """Explicitly dispatch the metadata collector agent to a room."""
+
+        lkapi = utils.create_livekit_client()
+        room_id = str(recording.room.id)
+
+        try:
+            response = await lkapi.agent_dispatch.create_dispatch(
+                CreateAgentDispatchRequest(
+                    agent_name=settings.METADATA_COLLECTOR_AGENT_NAME,
+                    room=room_id,
+                    metadata=str(recording.id),
+                )
+            )
+        except Exception as e:
+            logger.exception(
+                "Failed to create metadata collector agent for room %s", room_id
+            )
+            raise MetadataCollectorException(
+                "Failed to create metadata collector agent"
+            ) from e
+        finally:
+            await lkapi.aclose()
+
+        dispatch_id = getattr(response, "id", None)
+
+        if not dispatch_id:
+            logger.error("LiveKit response missing dispatch ID for room %s", room_id)
+            raise MetadataCollectorException(
+                f"LiveKit did not return a dispatch_id for room {room_id}"
+            )
+
+        recording.options["metadata_collector_dispatch_id"] = dispatch_id
+        await sync_to_async(recording.save)(update_fields=["options"])
+
+        return dispatch_id
+
+    @async_to_sync
+    async def stop(self, recording: Recording):
+        """Stop and delete the agent dispatch associated to the room."""
+
+        room_id = str(recording.room.id)
+        dispatch_id = recording.options.get("metadata_collector_dispatch_id")
+        lkapi = utils.create_livekit_client()
+
+        try:
+            if not dispatch_id:
+                logger.warning(
+                    "No metadata collector dispatch ID stored for room %s", room_id
+                )
+                return None
+
+            await lkapi.agent_dispatch.delete_dispatch(
+                dispatch_id=str(dispatch_id), room_name=room_id
+            )
+
+        except Exception as e:
+            logger.exception(
+                "Failed to stop metadata collector agent dispatch for room %s",
+                room_id,
+            )
+            raise MetadataCollectorException(
+                f"Failed to stop metadata collector agent for room {room_id}"
+            ) from e
+        finally:
+            await lkapi.aclose()

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

@@ -7,6 +7,7 @@ from logging import getLogger
 from livekit import api
 
 from core import models, utils
+from core.models import Recording
 
 logger = getLogger(__name__)
 
@@ -19,7 +20,7 @@ class RecordingEventsService:
     """Handles recording-related LiveKit webhook events."""
 
     @staticmethod
-    def handle_update(recording, egress_status):
+    def handle_update(recording: Recording, egress_status):
         """Handle egress status updates and sync recording state to room metadata."""
 
         room_name = str(recording.room.id)
@@ -40,7 +41,7 @@ class RecordingEventsService:
                 logger.exception("Failed to update room's metadata: %s", e)
 
     @staticmethod
-    def handle_limit_reached(recording):
+    def handle_limit_reached(recording: Recording):
         """Stop recording and notify participants when limit is reached."""
 
         recording.status = models.RecordingStatusChoices.STOPPED

src/backend/core/services/livekit_events.py

@@ -12,6 +12,10 @@ from django.conf import settings
 from livekit import api
 
 from core import models, utils
+from core.recording.services.metadata_collector import (
+    MetadataCollectorException,
+    MetadataCollectorService,
+)
 from core.recording.services.recording_events import (
     RecordingEventsError,
     RecordingEventsService,
@@ -158,7 +162,7 @@ class LiveKitEventsService:
         """Handle 'egress_ended' event."""
 
         try:
-            recording = models.Recording.objects.get(
+            recording = models.Recording.objects.select_related("room").get(
                 worker_id=data.egress_info.egress_id
             )
         except models.Recording.DoesNotExist as err:
@@ -174,6 +178,14 @@ class LiveKitEventsService:
         except utils.MetadataUpdateException as e:
             logger.exception("Failed to update room's metadata: %s", e)
 
+        if settings.METADATA_COLLECTOR_ENABLED and recording.options.get(
+            "metadata_collector_dispatch_id"
+        ):
+            try:
+                MetadataCollectorService().stop(recording)
+            except MetadataCollectorException:
+                logger.warning("Failed to stop the MetadataCollectorService")
+
         if (
             data.egress_info.status == api.EgressStatus.EGRESS_LIMIT_REACHED
             and recording.status == models.RecordingStatusChoices.ACTIVE

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

@@ -277,6 +277,7 @@ def test_start_recording_options_transcribe_valid_true(
 ):
     """Should accept transcribe with any valid pydantic true values."""
     settings.RECORDING_ENABLE = True
+    settings.METADATA_COLLECTOR_ENABLED = False
     room = RoomFactory()
     user = UserFactory()
     room.accesses.create(user=user, role="owner")
@@ -487,6 +488,93 @@ def test_start_recording_options_original_mode_omitted(
     assert recording.options == {}
 
 
+def test_start_recording_calls_metadata_collector_start(
+    settings, mock_worker_service_factory, mock_worker_manager
+):
+    """Should call MetadataCollectorService.start when conditions are met."""
+    settings.RECORDING_ENABLE = True
+    settings.METADATA_COLLECTOR_ENABLED = True
+
+    room = RoomFactory()
+    user = UserFactory()
+    room.accesses.create(user=user, role="owner")
+
+    client = APIClient()
+    client.force_login(user)
+
+    with mock.patch(
+        "core.api.viewsets.MetadataCollectorService"
+    ) as mock_collector_class:
+        mock_collector = mock.Mock()
+        mock_collector_class.return_value = mock_collector
+
+        response = client.post(
+            f"/api/v1.0/rooms/{room.id}/start-recording/",
+            {
+                "mode": "screen_recording",
+                "options": {"transcribe": True, "collect_metadata": True},
+            },
+            format="json",
+        )
+
+    assert response.status_code == 201
+
+    recording = Recording.objects.get(room=room)
+    mock_collector.start.assert_called_once_with(recording)
+
+
+@pytest.mark.parametrize(
+    "metadata_enabled,options",
+    [
+        # Metadata collector disabled, regardless of transcribe option
+        (False, {"transcribe": True}),
+        (False, {"transcribe": False}),
+        (False, None),
+        # Metadata collector enabled, but transcribe is False or missing
+        (True, {"transcribe": False}),
+        (True, None),
+        # Metadata collector enabled, transcribe True, but collect_metadata explicitly False
+        (True, {"transcribe": True, "collect_metadata": False}),
+    ],
+)
+def test_start_recording_does_not_call_metadata_collector_start_when_conditions_not_met(
+    settings,
+    mock_worker_service_factory,
+    mock_worker_manager,
+    metadata_enabled,
+    options,
+):
+    """Should not call MetadataCollectorService.start when conditions are not met."""
+    settings.RECORDING_ENABLE = True
+    settings.METADATA_COLLECTOR_ENABLED = metadata_enabled
+
+    room = RoomFactory()
+    user = UserFactory()
+    room.accesses.create(user=user, role="owner")
+
+    client = APIClient()
+    client.force_login(user)
+
+    payload = {"mode": "screen_recording"}
+    if options is not None:
+        payload["options"] = options
+
+    with mock.patch(
+        "core.api.viewsets.MetadataCollectorService"
+    ) as mock_collector_class:
+        mock_collector = mock.Mock()
+        mock_collector_class.return_value = mock_collector
+
+        response = client.post(
+            f"/api/v1.0/rooms/{room.id}/start-recording/",
+            payload,
+            format="json",
+        )
+
+    assert response.status_code == 201
+    mock_collector.start.assert_not_called()
+
+
 @pytest.mark.parametrize("value", ["invalid_mode", "foo", 123, "SCREEN_RECORDING"])
 def test_start_recording_options_original_mode_invalid(settings, value):
     """Should reject invalid recording mode values for original_mode."""

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

@@ -269,6 +269,63 @@ def test_handle_egress_ended_recording_not_limit_reached(
     assert recording.status == "stopped"
 
 
+@mock.patch("core.services.livekit_events.MetadataCollectorService")
+@mock.patch("core.utils.update_room_metadata")
+def test_handle_egress_ended_calls_metadata_collector_stop_when_conditions_are_met(
+    mock_update_room_metadata, mock_collector_class, service, settings
+):
+    """Should call MetadataCollectorService.stop when it exists."""
+    settings.METADATA_COLLECTOR_ENABLED = True
+
+    recording = RecordingFactory(
+        worker_id="worker-1",
+        status="active",
+        options={"metadata_collector_dispatch_id": "dispatch-123"},
+    )
+    mock_data = mock.MagicMock()
+    mock_data.egress_info.egress_id = recording.worker_id
+    mock_data.egress_info.status = EgressStatus.EGRESS_COMPLETE
+
+    mock_collector = mock.Mock()
+    mock_collector_class.return_value = mock_collector
+
+    service._handle_egress_ended(mock_data)
+
+    mock_collector.stop.assert_called_once_with(recording)
+
+
+@pytest.mark.parametrize(
+    "metadata_enabled,options",
+    [
+        (True, {}),
+        (False, {}),
+    ],
+)
+@mock.patch("core.services.livekit_events.MetadataCollectorService")
+@mock.patch("core.utils.update_room_metadata")
+def test_handle_egress_ended_does_not_call_metadata_collector_stop_when_conditions_not_met(
+    _, mock_collector_class, metadata_enabled, options, service, settings
+):  # pylint: disable=too-many-arguments,too-many-positional-arguments
+    """Should not call MetadataCollectorService.stop when it does not exist."""
+    settings.METADATA_COLLECTOR_ENABLED = metadata_enabled
+
+    recording = RecordingFactory(
+        worker_id="worker-1",
+        status="active",
+        options=options,
+    )
+    mock_data = mock.MagicMock()
+    mock_data.egress_info.egress_id = recording.worker_id
+    mock_data.egress_info.status = EgressStatus.EGRESS_COMPLETE
+
+    mock_collector = mock.Mock()
+    mock_collector_class.return_value = mock_collector
+
+    service._handle_egress_ended(mock_data)
+
+    mock_collector.stop.assert_not_called()
+
+
 @mock.patch.object(LobbyService, "clear_room_cache")
 @mock.patch.object(TelephonyService, "delete_dispatch_rule")
 def test_handle_room_finished_clears_cache_and_deletes_dispatch_rule(

src/backend/meet/settings.py

@@ -808,6 +808,16 @@ class Base(Configuration):
         environ_prefix=None,
     )
 
+    # Metadata collector settings
+    METADATA_COLLECTOR_ENABLED = values.BooleanValue(
+        False, environ_name="METADATA_COLLECTOR_ENABLED", environ_prefix=None
+    )
+    METADATA_COLLECTOR_AGENT_NAME = values.Value(
+        "metadata-collector",
+        environ_name="METADATA_COLLECTOR_AGENT_NAME",
+        environ_prefix=None,
+    )
+
     # External Applications
     APPLICATION_CLIENT_ID_LENGTH = values.PositiveIntegerValue(
         40,