wip: replace created_at by started_at. may want to revert

CHANGELOG.md

@@ -15,7 +15,6 @@ and this project adheres to
 ### Fixed
 
 - 🔒️(backend) fix email disclosure in room invitation endpoint #1200
-- 🐛(backend) fix regression in update-participant endpoint #1204
 
 ## [1.12.0] - 2026-03-24
 

Makefile

@@ -125,10 +125,15 @@ run-summary: ## start only the summary application and all needed services
 	@$(COMPOSE) up --force-recreate -d celery-summary-summarize
 .PHONY: run-summary
 
+run-agents: ## start the LiveKit agents (metadata collector)
+	@$(COMPOSE) up --force-recreate -d metadata-collector-dev
+.PHONY: run-agents
+
 run:
 run: ## start the wsgi (production) and development server
 	@$(MAKE) run-backend
 	@$(MAKE) run-summary
+	@$(MAKE) run-agents
 	@$(COMPOSE) up --force-recreate -d frontend
 .PHONY: run
 

compose.yml

@@ -246,6 +246,29 @@ services:
       depends_on:
         - redis
 
+  metadata-collector-dev:
+    build:
+      context: ./src/agents
+    command: ["python", "metadata_collector.py", "dev"]
+    environment:
+      - LIVEKIT_URL=ws://livekit:7880
+      - LIVEKIT_API_KEY=devkey
+      - LIVEKIT_API_SECRET=secret
+      - AWS_S3_ENDPOINT_URL=minio:9000
+      - AWS_S3_ACCESS_KEY_ID=meet
+      - AWS_S3_SECRET_ACCESS_KEY=password
+      - AWS_STORAGE_BUCKET_NAME=meet-media-storage
+      - AWS_S3_SECURE_ACCESS=False
+    volumes:
+      - ./src/agents:/app
+    depends_on:
+      - livekit
+      - minio
+    develop:
+      watch:
+        - action: rebuild
+          path: ./src/agents
+
   redis-summary:
     image: redis
     ports:

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/Dockerfile

@@ -15,19 +15,30 @@ COPY pyproject.toml .
 RUN mkdir /install && \
   pip install --prefix=/install .
 
+FROM base AS development
+
+WORKDIR /app
+
+COPY pyproject.toml .
+RUN pip install --no-cache-dir ".[dev]"
+
+COPY . .
+
+CMD ["python", "metadata_collector.py", "dev"]
+
 FROM base AS production
 
 WORKDIR /app
 
+COPY --from=builder /install /usr/local
+
 # Remove pip to reduce attack surface in production
 RUN pip uninstall -y pip
 
+# Un-privileged user running the application
 ARG DOCKER_USER
 USER ${DOCKER_USER}
 
-# Un-privileged user running the application
-COPY --from=builder /install /usr/local
+COPY ./*.py /app/
 
-COPY . .
-
-CMD ["python", "multi-user-transcriber.py", "start"]
+CMD ["python", "multi_user_transcriber.py", "start"]

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,384 @@
+"""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: '%s'", participant_identity, full_text
+        )
+
+        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/serializers.py

@@ -303,9 +303,6 @@ class MuteParticipantSerializer(BaseParticipantsManagementSerializer):
     )
 
 
-TrackSource = Literal["SCREEN_SHARE", "SCREEN_SHARE_AUDIO", "CAMERA", "MICROPHONE"]
-
-
 class ParticipantPermission(BaseModel):
     """Mirror the LiveKit ParticipantPermission protobuf.
 
@@ -316,7 +313,9 @@ class ParticipantPermission(BaseModel):
     can_subscribe: bool | None = None
     can_publish: bool | None = None
     can_publish_data: bool | None = None
-    can_publish_sources: list[TrackSource] = Field(default_factory=list)
+    can_publish_sources: list[int] = Field(
+        default_factory=list
+    )  # TrackSource enum values
     hidden: bool | None = None
     recorder: bool | None = None
     can_update_metadata: bool | None = None
@@ -367,6 +366,14 @@ class UpdateParticipantSerializer(BaseParticipantsManagementSerializer):
                 f"Setting the following participant permissions is not allowed: "
                 f"{', '.join(suspicious_fields)}."
             )
+        if permission.can_subscribe_metrics is not None:
+            raise serializers.ValidationError(
+                {
+                    "permission": {
+                        "can_subscribe_metrics": "This permission is not implemented."
+                    }
+                }
+            )
 
         return permission
 

src/backend/core/api/viewsets.py

@@ -44,6 +44,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,
@@ -336,6 +340,16 @@ 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)
+                logger.info("Started MetadataCollectorService")
+            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/event/notification.py

@@ -1,6 +1,7 @@
 """Service to notify external services when a new recording is ready."""
 
 import logging
+import os
 import smtplib
 
 from django.conf import settings
@@ -150,22 +151,23 @@ class NotificationService:
             .first()
         )
 
+        # TODO: change how we get metadata_filename
+        output_folder = os.getenv("AWS_S3_OUTPUT_FOLDER", "metadata")
+        metadata_filename = f"{output_folder}/{recording.id}-metadata.json"
+
         if not owner_access:
             logger.error("No owner found for recording %s", recording.id)
             return False
         payload = {
             "owner_id": str(owner_access.user.id),
-            "filename": recording.key,
+            "recording_filename": recording.key,
+            "metadata_filename": metadata_filename,
             "email": owner_access.user.email,
             "sub": owner_access.user.sub,
             "room": recording.room.name,
             "language": recording.options.get("language"),
-            "recording_date": recording.created_at.astimezone(
-                owner_access.user.timezone
-            ).strftime("%Y-%m-%d"),
-            "recording_time": recording.created_at.astimezone(
-                owner_access.user.timezone
-            ).strftime("%H:%M"),
+            "worker_id": recording.worker_id,
+            "owner_timezone": str(owner_access.user.timezone),
             "download_link": f"{get_recording_download_base_url()}/{recording.id}",
             "context_language": owner_access.user.language,
         }

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_participants_management.py

@@ -130,11 +130,10 @@ def test_update_participant_success(mock_livekit_client):
             "can_publish": True,
             "can_publish_data": True,
             "can_publish_sources": [
-                "CAMERA",
-                "MICROPHONE",
-            ],
+                1,
+                2,
+            ],  # [TrackSource.CAMERA, TrackSource.MICROPHONE]
             "can_update_metadata": True,
-            "can_subscribe_metrics": True,
         },
         "name": "John Doe",
     }
@@ -156,14 +155,8 @@ def test_update_participant_success(mock_livekit_client):
         {"can_subscribe": True},
         {"can_publish": True},
         {"can_publish_data": True},
-        {
-            "can_publish_sources": [
-                "CAMERA",
-                "MICROPHONE",
-            ]
-        },
+        {"can_publish_sources": [1, 2]},
         {"can_update_metadata": True},
-        {"can_subscribe_metrics": False},
     ],
 )
 def test_update_participant_permission_fields_are_optional(
@@ -271,6 +264,35 @@ def test_update_participant_suspicious_permission_multiple(mock_suspicious):
     )
 
 
+@pytest.mark.parametrize("value", (False, True))
+def test_update_participant_unimplemented_can_subscribe_metrics(value):
+    """Test update participant raises 400 when can_subscribe_metrics is set."""
+    client = APIClient()
+    room = RoomFactory()
+    user = UserFactory()
+    UserResourceAccessFactory(
+        resource=room, user=user, role=random.choice(["administrator", "owner"])
+    )
+    client.force_authenticate(user=user)
+
+    payload = {
+        "participant_identity": str(uuid4()),
+        "permission": {
+            "can_subscribe": True,
+            "can_publish": True,
+            "can_publish_data": True,
+            "can_update_metadata": False,
+            "can_subscribe_metrics": value,
+        },
+    }
+
+    url = reverse("rooms-update-participant", kwargs={"pk": room.id})
+    response = client.post(url, payload, format="json")
+
+    assert response.status_code == status.HTTP_400_BAD_REQUEST
+    assert "can_subscribe_metrics" in str(response.data)
+
+
 def test_update_participant_forbidden_without_access():
     """Test update participant returns 403 when user lacks room privileges."""
     client = APIClient()

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,87 @@ 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}},
+            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,transcribe",
+    [
+        (False, True),
+        (False, False),
+        (True, False),
+        (True, None),
+    ],
+)
+def test_start_recording_does_not_call_metadata_collector_start_when_conditions_not_met(
+    settings,
+    mock_worker_service_factory,
+    mock_worker_manager,
+    metadata_enabled,
+    transcribe,
+):
+    """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)
+
+    options = {}
+    if transcribe is not None:
+        options["transcribe"] = transcribe
+
+    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": options}
+            if options
+            else {"mode": "screen_recording"},
+            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,

src/helm/env.d/dev-dinum/values.meet.yaml.gotmpl

@@ -1,29 +1,3 @@
-_summaryEnvVars: &summaryEnvVars
-  APP_NAME: summary-microservice
-  APP_API_TOKEN: password
-  AWS_STORAGE_BUCKET_NAME: meet-media-storage
-  AWS_S3_ENDPOINT_URL: http://minio.meet.svc.cluster.local:9000/
-  AWS_S3_ACCESS_KEY_ID: meet
-  AWS_S3_SECRET_ACCESS_KEY: password
-  AWS_S3_SECURE_ACCESS: False
-  WHISPERX_API_KEY: your-secret-value
-  WHISPERX_BASE_URL: https://configure-your-url.com
-  WHISPERX_ASR_MODEL: large-v2
-  WHISPERX_DEFAULT_LANGUAGE: fr
-  LLM_BASE_URL: https://configure-your-url.com
-  LLM_API_KEY: your-secret-value
-  LLM_MODEL: meta-llama/Llama-3.1-8B-Instruct
-  WEBHOOK_API_TOKEN: password
-  WEBHOOK_URL: https://www.mock-impress.com/webhook/
-  CELERY_BROKER_URL: redis://default:pass@redis-master:6379/1
-  CELERY_RESULT_BACKEND: redis://default:pass@redis-master:6379/1
-  TASK_TRACKER_REDIS_URL: redis://default:pass@redis-master:6379/1
-
-_summaryImage: &summaryImage
-  repository: localhost:5001/meet-summary
-  pullPolicy: Always
-  tag: "latest"
-
 image:
   repository: localhost:5001/meet-backend
   pullPolicy: Always
@@ -168,15 +142,66 @@ posthog:
 
 summary:
   replicas: 1
-  image: *summaryImage
   envVars:
-    <<: *summaryEnvVars
+    APP_NAME: summary-microservice
+    APP_API_TOKEN: password
+    AWS_STORAGE_BUCKET_NAME: meet-media-storage
+    AWS_S3_ENDPOINT_URL: minio.meet.svc.cluster.local:9000
+    AWS_S3_ACCESS_KEY_ID: meet
+    AWS_S3_SECRET_ACCESS_KEY: password
+    AWS_S3_SECURE_ACCESS: False
+    WHISPERX_API_KEY: your-secret-value
+    WHISPERX_BASE_URL: https://configure-your-url.com
+    WHISPERX_ASR_MODEL: large-v2
+    LLM_BASE_URL: https://configure-your-url.com
+    LLM_API_KEY: your-secret-value
+    LLM_MODEL: meta-llama/Llama-3.1-8B-Instruct
+    WEBHOOK_API_TOKEN: password
+    WEBHOOK_URL: https://www.mock-impress.com/webhook/
+    CELERY_BROKER_URL: redis://default:pass@redis-master:6379/1
+    CELERY_RESULT_BACKEND: redis://default:pass@redis-master:6379/1
+  
+  image:
+    repository: localhost:5001/meet-summary
+    pullPolicy: Always
+    tag: "latest"
+  
+  command:
+    - "uvicorn"
+    - "summary.main:app"
+    - "--host"
+    - "0.0.0.0"
+    - "--port"
+    - "8000"
+    - "--reload"
 
 celeryTranscribe:
   replicas: 1
-  image: *summaryImage
   envVars:
-    <<: *summaryEnvVars
+    APP_NAME: summary-microservice
+    APP_API_TOKEN: password
+    AWS_STORAGE_BUCKET_NAME: meet-media-storage
+    AWS_S3_ENDPOINT_URL: minio.meet.svc.cluster.local:9000
+    AWS_S3_ACCESS_KEY_ID: meet
+    AWS_S3_SECRET_ACCESS_KEY: password
+    AWS_S3_SECURE_ACCESS: False
+    WHISPERX_API_KEY: your-secret-value
+    WHISPERX_BASE_URL: https://configure-your-url.com
+    WHISPERX_ASR_MODEL: large-v2
+    LLM_BASE_URL: https://configure-your-url.com
+    LLM_API_KEY: your-secret-value
+    LLM_MODEL: meta-llama/Llama-3.1-8B-Instruct
+    WEBHOOK_API_TOKEN: password
+    WEBHOOK_URL: https://www.mock-impress.com/webhook/
+    CELERY_BROKER_URL: redis://default:pass@redis-master:6379/1
+    CELERY_RESULT_BACKEND: redis://default:pass@redis-master:6379/1
+    TASK_TRACKER_REDIS_URL: redis://default:pass@redis-master:6379/1
+
+  image:
+    repository: localhost:5001/meet-summary
+    pullPolicy: Always
+    tag: "latest"
+
   command:
     - "celery"
     - "-A"
@@ -188,9 +213,31 @@ celeryTranscribe:
 
 celerySummarize:
   replicas: 1
-  image: *summaryImage
   envVars:
-    <<: *summaryEnvVars
+    APP_NAME: summary-microservice
+    APP_API_TOKEN: password
+    AWS_STORAGE_BUCKET_NAME: meet-media-storage
+    AWS_S3_ENDPOINT_URL: minio.meet.svc.cluster.local:9000
+    AWS_S3_ACCESS_KEY_ID: meet
+    AWS_S3_SECRET_ACCESS_KEY: password
+    AWS_S3_SECURE_ACCESS: False
+    WHISPERX_API_KEY: your-secret-value
+    WHISPERX_BASE_URL: https://configure-your-url.com
+    WHISPERX_ASR_MODEL: large-v2
+    LLM_BASE_URL: https://configure-your-url.com
+    LLM_API_KEY: your-secret-value
+    LLM_MODEL: meta-llama/Llama-3.1-8B-Instruct
+    WEBHOOK_API_TOKEN: password
+    WEBHOOK_URL: https://www.mock-impress.com/webhook/
+    CELERY_BROKER_URL: redis://default:pass@redis-master:6379/1
+    CELERY_RESULT_BACKEND: redis://default:pass@redis-master:6379/1
+    TASK_TRACKER_REDIS_URL: redis://default:pass@redis-master:6379/1
+
+  image:
+    repository: localhost:5001/meet-summary
+    pullPolicy: Always
+    tag: "latest"
+
   command:
     - "celery"
     - "-A"

src/helm/env.d/dev-keycloak/values.meet.yaml.gotmpl

@@ -1,29 +1,3 @@
-_summaryEnvVars: &summaryEnvVars
-  APP_NAME: summary-microservice
-  APP_API_TOKEN: password
-  AWS_STORAGE_BUCKET_NAME: meet-media-storage
-  AWS_S3_ENDPOINT_URL: http://minio.meet.svc.cluster.local:9000/
-  AWS_S3_ACCESS_KEY_ID: meet
-  AWS_S3_SECRET_ACCESS_KEY: password
-  AWS_S3_SECURE_ACCESS: False
-  WHISPERX_API_KEY: your-secret-value
-  WHISPERX_BASE_URL: https://configure-your-url.com
-  WHISPERX_ASR_MODEL: large-v2
-  WHISPERX_DEFAULT_LANGUAGE: fr
-  LLM_BASE_URL: https://configure-your-url.com
-  LLM_API_KEY: your-secret-value
-  LLM_MODEL: meta-llama/Llama-3.1-8B-Instruct
-  WEBHOOK_API_TOKEN: password
-  WEBHOOK_URL: https://www.mock-impress.com/webhook/
-  CELERY_BROKER_URL: redis://default:pass@redis-master:6379/1
-  CELERY_RESULT_BACKEND: redis://default:pass@redis-master:6379/1
-  TASK_TRACKER_REDIS_URL: redis://default:pass@redis-master:6379/1
-
-_summaryImage: &summaryImage
-  repository: localhost:5001/meet-summary
-  pullPolicy: Always
-  tag: "latest"
-
 image:
   repository: localhost:5001/meet-backend
   pullPolicy: Always
@@ -111,6 +85,7 @@ backend:
     AWS_S3_DOMAIN_REPLACE: https://minio.127.0.0.1.nip.io
     CELERY_ENABLED: True
     CELERY_BROKER_URL: redis://default:pass@redis-master:6379/1
+    METADATA_COLLECTOR_ENABLED: True
 
 
   migrate:
@@ -186,15 +161,69 @@ posthog:
 
 summary:
   replicas: 1
-  image: *summaryImage
   envVars:
-    <<: *summaryEnvVars
+    APP_NAME: summary-microservice
+    APP_API_TOKEN: password
+    AWS_STORAGE_BUCKET_NAME: meet-media-storage
+    AWS_S3_ENDPOINT_URL: http://minio.meet.svc.cluster.local:9000/
+    AWS_S3_ACCESS_KEY_ID: meet
+    AWS_S3_SECRET_ACCESS_KEY: password
+    AWS_S3_SECURE_ACCESS: False
+    WHISPERX_API_KEY: your-secret-value
+    WHISPERX_BASE_URL: https://configure-your-url.com
+    WHISPERX_ASR_MODEL: large-v2
+    WHISPERX_DEFAULT_LANGUAGE: fr
+    LLM_BASE_URL: https://configure-your-url.com
+    LLM_API_KEY: your-secret-value
+    LLM_MODEL: meta-llama/Llama-3.1-8B-Instruct
+    WEBHOOK_API_TOKEN: password
+    WEBHOOK_URL: https://www.mock-impress.com/webhook/
+    CELERY_BROKER_URL: redis://default:pass@redis-master:6379/1
+    CELERY_RESULT_BACKEND: redis://default:pass@redis-master:6379/1
+    TASK_TRACKER_REDIS_URL: redis://default:pass@redis-master:6379/1
+  
+  image:
+    repository: localhost:5001/meet-summary
+    pullPolicy: Always
+    tag: "latest"
+  
+  command:
+    - "uvicorn"
+    - "summary.main:app"
+    - "--host"
+    - "0.0.0.0"
+    - "--port"
+    - "8000"
+    - "--reload"
 
 celeryTranscribe:
   replicas: 1
-  image: *summaryImage
   envVars:
-    <<: *summaryEnvVars
+    APP_NAME: summary-microservice
+    APP_API_TOKEN: password
+    AWS_STORAGE_BUCKET_NAME: meet-media-storage
+    AWS_S3_ENDPOINT_URL: http://minio.meet.svc.cluster.local:9000/
+    AWS_S3_ACCESS_KEY_ID: meet
+    AWS_S3_SECRET_ACCESS_KEY: password
+    AWS_S3_SECURE_ACCESS: False
+    WHISPERX_API_KEY: your-secret-value
+    WHISPERX_BASE_URL: https://configure-your-url.com
+    WHISPERX_ASR_MODEL: large-v2
+    WHISPERX_DEFAULT_LANGUAGE: fr
+    LLM_BASE_URL: https://configure-your-url.com
+    LLM_API_KEY: your-secret-value
+    LLM_MODEL: meta-llama/Llama-3.1-8B-Instruct
+    WEBHOOK_API_TOKEN: password
+    WEBHOOK_URL: https://www.mock-impress.com/webhook/
+    CELERY_BROKER_URL: redis://default:pass@redis-master:6379/1
+    CELERY_RESULT_BACKEND: redis://default:pass@redis-master:6379/1
+    TASK_TRACKER_REDIS_URL: redis://default:pass@redis-master:6379/1
+  
+  image:
+    repository: localhost:5001/meet-summary
+    pullPolicy: Always
+    tag: "latest"
+
   command:
     - "celery"
     - "-A"
@@ -207,9 +236,32 @@ celeryTranscribe:
 
 celerySummarize:
   replicas: 1
-  image: *summaryImage
   envVars:
-    <<: *summaryEnvVars
+    APP_NAME: summary-microservice
+    APP_API_TOKEN: password
+    AWS_STORAGE_BUCKET_NAME: meet-media-storage
+    AWS_S3_ENDPOINT_URL: http://minio.meet.svc.cluster.local:9000/
+    AWS_S3_ACCESS_KEY_ID: meet
+    AWS_S3_SECRET_ACCESS_KEY: password
+    AWS_S3_SECURE_ACCESS: False
+    WHISPERX_API_KEY: your-secret-value
+    WHISPERX_BASE_URL: https://configure-your-url.com
+    WHISPERX_ASR_MODEL: large-v2
+    WHISPERX_DEFAULT_LANGUAGE: fr
+    LLM_BASE_URL: https://configure-your-url.com
+    LLM_API_KEY: your-secret-value
+    LLM_MODEL: meta-llama/Llama-3.1-8B-Instruct
+    WEBHOOK_API_TOKEN: password
+    WEBHOOK_URL: https://www.mock-impress.com/webhook/
+    CELERY_BROKER_URL: redis://default:pass@redis-master:6379/1
+    CELERY_RESULT_BACKEND: redis://default:pass@redis-master:6379/1
+    TASK_TRACKER_REDIS_URL: redis://default:pass@redis-master:6379/1
+
+  image:
+    repository: localhost:5001/meet-summary
+    pullPolicy: Always
+    tag: "latest"
+
   command:
     - "celery"
     - "-A"
@@ -231,12 +283,23 @@ agents:
     {{- end }}
     {{- end }}
     ENABLE_SILERO_VAD: "false"
+    AWS_S3_ENDPOINT_URL: minio.meet.svc.cluster.local:9000
+    AWS_S3_ACCESS_KEY_ID: meet
+    AWS_S3_SECRET_ACCESS_KEY: password
+    AWS_S3_SECURE_ACCESS: False
+    AWS_STORAGE_BUCKET_NAME: meet-media-storage
+    AWS_S3_OUTPUT_FOLDER: recordings
 
   image:
     repository: localhost:5001/meet-agents
     pullPolicy: Always
     tag: "latest"
 
+  command:
+    - "python"
+    - "metadata_collector.py"
+    - "start"
+
   # Extra volume mounts to manage our local custom CA and avoid to disable ssl
   extraVolumeMounts:
     - name: certs

src/helm/env.d/dev/values.egress.yaml.gotmpl

@@ -0,0 +1,43 @@
+replicaCount: 1
+terminationGracePeriodSeconds: 18000
+
+egress:
+  log_level: debug
+  ws_url: ws://livekit-livekit-server:80
+  insecure: true
+  enable_chrome_sandbox: true
+  {{- with .Values.livekit.keys }}
+  {{- range $key, $value := . }}
+  api_key: {{ $key }}
+  api_secret: {{ $value }}
+  {{- end }}
+  {{- end }}
+  redis:
+    address: redis-master:6379
+    password: pass
+  s3:
+    access_key: meet
+    secret: password
+    region: local
+    bucket: meet-media-storage
+    endpoint: http://minio:9000
+    force_path_style: true
+
+loadBalancer:
+  type: nginx
+  annotations:
+    nginx.ingress.kubernetes.io/rewrite-target: /
+    nginx.ingress.kubernetes.io/ssl-redirect: "true"
+    cert-manager.io/cluster-issuer: "letsencrypt-prod"
+  tls:
+    - hosts:
+        - livekit-egress.127.0.0.1.nip.io
+      secretName: livekit-egress-dinum-cert
+
+autoscaling:
+  enabled: false
+  minReplicas: 1
+  maxReplicas: 5
+
+nodeSelector: {}
+resources: {}

src/helm/env.d/dev/values.livekit.yaml.gotmpl

@@ -0,0 +1,49 @@
+replicaCount: 1
+terminationGracePeriodSeconds: 18000
+
+image:
+    repository: localhost:5001/meet-livekit
+    pullPolicy: Always
+    tag: "latest"
+
+livekit:
+  log_level: debug
+  rtc:
+    use_external_ip: false
+    port_range_start: 50000
+    port_range_end: 60000
+    tcp_port: 7881
+  redis:
+    address: redis-master:6379
+    password: pass
+  keys:
+  turn:
+    enabled: true
+    udp_port: 443
+    domain: livekit.127.0.0.1.nip.io
+    loadBalancerAnnotations: {}
+
+  webhook:
+    api_key:
+    urls:
+        - https://meet.127.0.0.1.nip.io/api/v1.0/rooms/webhooks-livekit/
+
+loadBalancer:
+  type: nginx
+  annotations:
+    nginx.ingress.kubernetes.io/rewrite-target: /
+    nginx.ingress.kubernetes.io/ssl-redirect: "true"
+    cert-manager.io/cluster-issuer: "letsencrypt-prod"
+  tls:
+    - hosts:
+        - livekit.127.0.0.1.nip.io
+      secretName: livekit-dinum-cert
+
+autoscaling:
+  enabled: false
+  minReplicas: 1
+  maxReplicas: 5
+  targetCPUUtilizationPercentage: 60
+
+nodeSelector: {}
+resources: {}

src/helm/env.d/dev/values.meet.yaml.gotmpl

@@ -0,0 +1,290 @@
+secrets:
+  - name: oidcLogin
+    itemId: a25effec-eaea-4ce1-9ed8-3a3cc1c734db
+    field: username
+    podVariable: OIDC_RP_CLIENT_ID
+    clusterSecretStore: bitwarden-login-meet
+  - name: oidcPass
+    itemId: a25effec-eaea-4ce1-9ed8-3a3cc1c734db
+    field: password
+    podVariable: OIDC_RP_CLIENT_SECRET
+    clusterSecretStore: bitwarden-login-meet
+  - name: brevoApiKey
+    itemId: 99107889-6124-4436-97cc-a5193f28443f
+    field: password
+    podVariable: BREVO_API_KEY
+    clusterSecretStore: bitwarden-login-meet
+image:
+  repository: localhost:5001/meet-backend
+  pullPolicy: Always
+  tag: "latest"
+
+backend:
+  replicas: 1
+  envVars:
+    DJANGO_CSRF_TRUSTED_ORIGINS: https://meet.127.0.0.1.nip.io,http://meet.127.0.0.1.nip.io
+    DJANGO_CONFIGURATION: Production
+    DJANGO_ALLOWED_HOSTS: meet.127.0.0.1.nip.io
+    DJANGO_SECRET_KEY: {{ .Values.djangoSecretKey }}
+    DJANGO_SETTINGS_MODULE: meet.settings
+    DJANGO_SILENCED_SYSTEM_CHECKS: security.W004, security.W008
+    DJANGO_SUPERUSER_PASSWORD: admin
+    DJANGO_EMAIL_HOST: "mailcatcher"
+    DJANGO_EMAIL_PORT: 1025
+    DJANGO_EMAIL_USE_SSL: False
+    DJANGO_EMAIL_BRAND_NAME: "La Suite Numérique"
+    DJANGO_EMAIL_SUPPORT_EMAIL: "test@yopmail.com"
+    DJANGO_EMAIL_LOGO_IMG: https://meet.127.0.0.1.nip.io/assets/logo-suite-numerique.png
+    DJANGO_EMAIL_DOMAIN: meet.127.0.0.1.nip.io
+    DJANGO_EMAIL_APP_BASE_URL: https://meet.127.0.0.1.nip.io
+    OIDC_OP_JWKS_ENDPOINT: https://fca.integ01.dev-agentconnect.fr/api/v2/jwks
+    OIDC_OP_AUTHORIZATION_ENDPOINT: https://fca.integ01.dev-agentconnect.fr/api/v2/authorize
+    OIDC_OP_TOKEN_ENDPOINT: https://fca.integ01.dev-agentconnect.fr/api/v2/token
+    OIDC_OP_USER_ENDPOINT: https://fca.integ01.dev-agentconnect.fr/api/v2/userinfo
+    OIDC_OP_LOGOUT_ENDPOINT: https://fca.integ01.dev-agentconnect.fr/api/v2/session/end
+    OIDC_RP_CLIENT_ID:
+      secretKeyRef:
+        name: backend
+        key: OIDC_RP_CLIENT_ID
+    OIDC_RP_CLIENT_SECRET:
+      secretKeyRef:
+        name: backend
+        key: OIDC_RP_CLIENT_SECRET
+    OIDC_RP_SIGN_ALGO: RS256
+    OIDC_RP_SCOPES: "openid email given_name usual_name"
+    OIDC_REDIRECT_ALLOWED_HOSTS: https://meet.127.0.0.1.nip.io
+    OIDC_AUTH_REQUEST_EXTRA_PARAMS: "{'acr_values': 'eidas1'}"
+    LOGIN_REDIRECT_URL: https://meet.127.0.0.1.nip.io
+    LOGIN_REDIRECT_URL_FAILURE: https://meet.127.0.0.1.nip.io
+    LOGOUT_REDIRECT_URL: https://meet.127.0.0.1.nip.io
+    DB_HOST: postgres
+    DB_NAME: meet
+    DB_USER: dinum
+    DB_PASSWORD: pass
+    DB_PORT: 5432
+    REDIS_URL: redis://default:pass@redis-master:6379/1
+    STORAGES_STATICFILES_BACKEND: django.contrib.staticfiles.storage.StaticFilesStorage
+    {{- with .Values.livekit.keys }}
+    {{- range $key, $value := . }}
+    LIVEKIT_API_SECRET: {{ $value }}
+    LIVEKIT_API_KEY: {{ $key }}
+    {{- end }}
+    {{- end }}
+    LIVEKIT_API_URL: https://livekit.127.0.0.1.nip.io/
+    LIVEKIT_WEBHOOK_EVENTS_FILTER_REGEX: "[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}"
+    ALLOW_UNREGISTERED_ROOMS: False
+    FRONTEND_SILENCE_LIVEKIT_DEBUG: False
+    FRONTEND_SUPPORT: "{'id': '58ea6697-8eba-4492-bc59-ad6562585041'}"
+    AWS_S3_ENDPOINT_URL: http://minio.meet.svc.cluster.local:9000
+    AWS_S3_ACCESS_KEY_ID: meet
+    AWS_S3_SECRET_ACCESS_KEY: password
+    AWS_STORAGE_BUCKET_NAME: meet-media-storage
+    AWS_S3_REGION_NAME: local
+    MEDIA_BASE_URL: https://meet.127.0.0.1.nip.io
+    RECORDING_ENABLE: True
+    RECORDING_STORAGE_EVENT_ENABLE: True
+    RECORDING_STORAGE_EVENT_TOKEN: password
+    SUMMARY_SERVICE_ENDPOINT: http://meet-summary:80/api/v1/tasks/
+    SUMMARY_SERVICE_API_TOKEN: password
+    RECORDING_DOWNLOAD_BASE_URL: https://meet.127.0.0.1.nip.io/recording
+    SIGNUP_NEW_USER_TO_MARKETING_EMAIL: True
+    BREVO_API_KEY:
+      secretKeyRef:
+        name: backend
+        key: BREVO_API_KEY
+    BREVO_API_CONTACT_LIST_IDS: 8
+    ROOM_TELEPHONY_ENABLED: True
+    SSL_CERT_FILE: /app/.venv/lib/python3.13/site-packages/certifi/cacert.pem
+
+
+  migrate:
+    command:
+      - "/bin/sh"
+      - "-c"
+      - |
+        python manage.py migrate --no-input
+    restartPolicy: Never
+
+  command:
+    - "gunicorn"
+    - "-c"
+    - "/usr/local/etc/gunicorn/meet.py"
+    - "meet.wsgi:application"
+    - "--reload"
+
+  createsuperuser:
+    command:
+      - "/bin/sh"
+      - "-c"
+      - |
+        python manage.py createsuperuser --email admin@example.com --password admin
+    restartPolicy: Never
+
+  # Extra volume mounts to manage our local custom CA and avoid to set ssl_verify: false
+  extraVolumeMounts:
+    - name: certs
+      mountPath: /app/.venv/lib/python3.13/site-packages/certifi/cacert.pem
+      subPath: cacert.pem
+
+  # Extra volumes to manage our local custom CA and avoid to set ssl_verify: false
+  extraVolumes:
+    - name: certs
+      configMap:
+        name: certifi
+        items:
+        - key: cacert.pem
+          path: cacert.pem
+
+frontend:
+  envVars:
+    VITE_APP_TITLE: "LaSuite Meet"
+    VITE_PORT: 8080
+    VITE_HOST: 0.0.0.0
+    VITE_API_BASE_URL: https://meet.127.0.0.1.nip.io/
+
+  replicas: 1
+
+  image:
+    repository: localhost:5001/meet-frontend
+    pullPolicy: Always
+    tag: "latest"
+
+ingress:
+  enabled: true
+  host: meet.127.0.0.1.nip.io
+
+ingressAdmin:
+  enabled: true
+  host: meet.127.0.0.1.nip.io
+
+ingressWebhook:
+  enabled: true
+  host: meet.127.0.0.1.nip.io
+
+posthog:
+  ingress:
+    enabled: false
+
+  ingressAssets:
+    enabled: false
+
+summary:
+  replicas: 1
+  envVars:
+    APP_NAME: summary-microservice
+    APP_API_TOKEN: password
+    AWS_STORAGE_BUCKET_NAME: meet-media-storage
+    AWS_S3_ENDPOINT_URL: minio.meet.svc.cluster.local:9000
+    AWS_S3_ACCESS_KEY_ID: meet
+    AWS_S3_SECRET_ACCESS_KEY: password
+    AWS_S3_SECURE_ACCESS: False
+    WHISPERX_API_KEY: your-secret-value
+    WHISPERX_BASE_URL: https://configure-your-url.com
+    WHISPERX_ASR_MODEL: large-v2
+    LLM_BASE_URL: https://configure-your-url.com
+    LLM_API_KEY: your-secret-value
+    LLM_MODEL: meta-llama/Llama-3.1-8B-Instruct
+    WEBHOOK_API_TOKEN: password
+    WEBHOOK_URL: https://www.mock-impress.com/webhook/
+    CELERY_BROKER_URL: redis://default:pass@redis-master:6379/1
+    CELERY_RESULT_BACKEND: redis://default:pass@redis-master:6379/1
+
+  image:
+    repository: localhost:5001/meet-summary
+    pullPolicy: Always
+    tag: "latest"
+
+  command:
+    - "uvicorn"
+    - "summary.main:app"
+    - "--host"
+    - "0.0.0.0"
+    - "--port"
+    - "8000"
+    - "--reload"
+
+celeryTranscribe:
+  replicas: 1
+  envVars:
+    APP_NAME: summary-microservice
+    APP_API_TOKEN: password
+    AWS_STORAGE_BUCKET_NAME: meet-media-storage
+    AWS_S3_ENDPOINT_URL: minio.meet.svc.cluster.local:9000
+    AWS_S3_ACCESS_KEY_ID: meet
+    AWS_S3_SECRET_ACCESS_KEY: password
+    AWS_S3_SECURE_ACCESS: False
+    WHISPERX_API_KEY: your-secret-value
+    WHISPERX_BASE_URL: https://configure-your-url.com
+    WHISPERX_ASR_MODEL: large-v2
+    LLM_BASE_URL: https://configure-your-url.com
+    LLM_API_KEY: your-secret-value
+    LLM_MODEL: meta-llama/Llama-3.1-8B-Instruct
+    WEBHOOK_API_TOKEN: password
+    WEBHOOK_URL: https://www.mock-impress.com/webhook/
+    CELERY_BROKER_URL: redis://default:pass@redis-master:6379/1
+    CELERY_RESULT_BACKEND: redis://default:pass@redis-master:6379/1
+    TASK_TRACKER_REDIS_URL: redis://default:pass@redis-master:6379/1
+
+  image:
+    repository: localhost:5001/meet-summary
+    pullPolicy: Always
+    tag: "latest"
+
+  command:
+    - "celery"
+    - "-A"
+    - "summary.core.celery_worker"
+    - "worker"
+    - "--pool=solo"
+    - "--loglevel=info"
+    - "-Q transcribe-queue"
+
+celerySummarize:
+  replicas: 1
+  envVars:
+    APP_NAME: summary-microservice
+    APP_API_TOKEN: password
+    AWS_STORAGE_BUCKET_NAME: meet-media-storage
+    AWS_S3_ENDPOINT_URL: minio.meet.svc.cluster.local:9000
+    AWS_S3_ACCESS_KEY_ID: meet
+    AWS_S3_SECRET_ACCESS_KEY: password
+    AWS_S3_SECURE_ACCESS: False
+    WHISPERX_API_KEY: your-secret-value
+    WHISPERX_BASE_URL: https://configure-your-url.com
+    WHISPERX_ASR_MODEL: large-v2
+    LLM_BASE_URL: https://configure-your-url.com
+    LLM_API_KEY: your-secret-value
+    LLM_MODEL: meta-llama/Llama-3.1-8B-Instruct
+    WEBHOOK_API_TOKEN: password
+    WEBHOOK_URL: https://www.mock-impress.com/webhook/
+    CELERY_BROKER_URL: redis://default:pass@redis-master:6379/1
+    CELERY_RESULT_BACKEND: redis://default:pass@redis-master:6379/1
+    TASK_TRACKER_REDIS_URL: redis://default:pass@redis-master:6379/1
+
+  image:
+    repository: localhost:5001/meet-summary
+    pullPolicy: Always
+    tag: "latest"
+
+  command:
+    - "celery"
+    - "-A"
+    - "summary.core.celery_worker"
+    - "worker"
+    - "--pool=solo"
+    - "--loglevel=info"
+    - "-Q summarize-queue"
+
+ingressMedia:
+  enabled: true
+  host: meet.127.0.0.1.nip.io
+
+  annotations:
+    nginx.ingress.kubernetes.io/auth-url: https://meet.127.0.0.1.nip.io/api/v1.0/recordings/media-auth/
+    nginx.ingress.kubernetes.io/auth-response-headers: "Authorization, X-Amz-Date, X-Amz-Content-SHA256"
+    nginx.ingress.kubernetes.io/upstream-vhost: minio.meet.svc.cluster.local:9000
+    nginx.ingress.kubernetes.io/rewrite-target: /meet-media-storage/$1
+
+serviceMedia:
+  host: minio.meet.svc.cluster.local
+  port: 9000

src/helm/env.d/dev/values.secrets.yaml

@@ -0,0 +1,10 @@
+djangoSecretKey: 7K9mQ2xR8pL3vN6tY1sW4jH5cE0zF9bM2qA7uI3oP6rT1wErt12te12
+livekit:
+    keys:
+        devkey: secret
+    webhook:
+        api_key: devkey
+livekitApi:
+    key: devkey
+    secret: secret
+

src/helm/helmfile.yaml.gotmpl

@@ -7,6 +7,10 @@ environments:
     values:
       - version: 0.0.1
       - env.d/dev-dinum/values.secrets.yaml
+  dev:
+    values:
+      - version: 0.0.1
+      - env.d/dev/values.secrets.yaml
 
 ---
 repositories:

src/summary/summary/api/route/tasks.py

@@ -19,13 +19,14 @@ class TranscribeSummarizeTaskCreation(BaseModel):
     """Transcription and summarization parameters."""
 
     owner_id: str
-    filename: str
+    recording_filename: str
+    metadata_filename: str
     email: str
     sub: str
     version: Optional[int] = 2
     room: Optional[str]
-    recording_date: Optional[str]
-    recording_time: Optional[str]
+    worker_id: Optional[str]
+    owner_timezone: Optional[str]
     language: Optional[str]
     download_link: Optional[str]
     context_language: Optional[str] = None
@@ -51,13 +52,14 @@ async def create_transcribe_summarize_task(request: TranscribeSummarizeTaskCreat
     task = process_audio_transcribe_summarize_v2.apply_async(
         args=[
             request.owner_id,
-            request.filename,
+            request.recording_filename,
+            request.metadata_filename,
             request.email,
             request.sub,
             time.time(),
             request.room,
-            request.recording_date,
-            request.recording_time,
+            request.worker_id,
+            request.owner_timezone,
             request.language,
             request.download_link,
             request.context_language,

src/summary/summary/core/analytics.py

@@ -112,7 +112,7 @@ class MetadataManager:
         if self._is_disabled or self.has_task_id(task_id):
             return
 
-        _, filename, email, _, received_at, *_ = task_args
+        _, filename, _, email, _, received_at, *_ = task_args
 
         start_time = time.time()
         initial_metadata = {

src/summary/summary/core/celery_worker.py

@@ -3,7 +3,9 @@
 # ruff: noqa: PLR0913
 
 import json
+import os
 import time
+from datetime import datetime, timezone
 from typing import Optional
 
 import openai
@@ -28,6 +30,7 @@ from summary.core.prompt import (
     PROMPT_USER_PART,
 )
 from summary.core.transcript_formatter import TranscriptFormatter
+from summary.core.user_assign import assign_speakers
 from summary.core.webhook_service import submit_content
 
 settings = get_settings()
@@ -58,7 +61,7 @@ if settings.sentry_dsn and settings.sentry_is_enabled:
 file_service = FileService()
 
 
-def transcribe_audio(task_id, filename, language):
+def transcribe_audio(task_id, recording_filename, language):
     """Transcribe an audio file using WhisperX.
 
     Downloads the audio from MinIO, sends it to WhisperX for transcription,
@@ -75,9 +78,13 @@ def transcribe_audio(task_id, filename, language):
 
     # Transcription
     try:
-        with file_service.prepare_audio_file(filename) as (audio_file, metadata):
+        with file_service.prepare_audio_file(recording_filename) as (
+            audio_file,
+            metadata,
+        ):
             metadata_manager.track(task_id, {"audio_length": metadata["duration"]})
 
+            # Compute language parameter
             if language is None:
                 language = settings.whisperx_default_language
                 logger.info(
@@ -90,22 +97,25 @@ def transcribe_audio(task_id, filename, language):
                     language,
                 )
 
+            # Call remote service for transcription
             transcription_start_time = time.time()
-
             transcription = whisperx_client.audio.transcriptions.create(
                 model=settings.whisperx_asr_model, file=audio_file, language=language
             )
 
-            transcription_time = round(time.time() - transcription_start_time, 2)
+            # Logging
+            transcription_duration = round(time.time() - transcription_start_time, 2)
             metadata_manager.track(
                 task_id,
-                {"transcription_time": transcription_time},
+                {"transcription_time": transcription_duration},
+            )
+            logger.info(
+                "Transcription received in %.2f seconds.", transcription_duration
             )
-            logger.info("Transcription received in %.2f seconds.", transcription_time)
             logger.debug("Transcription: \n %s", transcription)
 
     except FileServiceException:
-        logger.exception("Unexpected error for filename: %s", filename)
+        logger.exception("Unexpected error for recording: %s", recording_filename)
         return None
 
     metadata_manager.track_transcription_metadata(task_id, transcription)
@@ -117,8 +127,8 @@ def format_transcript(
     context_language,
     language,
     room,
-    recording_date,
-    recording_time,
+    recording_datetime,
+    owner_timezone,
     download_link,
 ):
     """Format a transcription into readable content with a title.
@@ -134,8 +144,8 @@ def format_transcript(
     return formatter.format(
         transcription,
         room=room,
-        recording_date=recording_date,
-        recording_time=recording_time,
+        recording_datetime=recording_datetime,
+        owner_timezone=owner_timezone,
         download_link=download_link,
     )
 
@@ -167,13 +177,14 @@ def format_actions(llm_output: dict) -> str:
 def process_audio_transcribe_summarize_v2(
     self,
     owner_id: str,
-    filename: str,
+    recording_filename: str,
+    metadata_filename: str,
     email: str,
     sub: str,
     received_at: float,
     room: Optional[str],
-    recording_date: Optional[str],
-    recording_time: Optional[str],
+    worker_id: Optional[str],
+    owner_timezone: Optional[str],
     language: Optional[str],
     download_link: Optional[str],
     context_language: Optional[str] = None,
@@ -189,13 +200,14 @@ def process_audio_transcribe_summarize_v2(
     Args:
         self: Celery task instance (passed on with bind=True)
         owner_id: Unique identifier of the recording owner.
-        filename: Name of the audio file in MinIO storage.
+        recording_filename: Name of the audio file in MinIO storage.
+        metadata_filename: Name of the audio file in MinIO storage.
         email: Email address of the recording owner.
         sub: OIDC subject identifier of the recording owner.
         received_at: Unix timestamp when the recording was received.
         room: room name where the recording took place.
-        recording_date: Date of the recording (localized display string).
-        recording_time: Time of the recording (localized display string).
+        worker_id: LiveKit egress ID used to fetch the egress JSON from S3.
+        owner_timezone: IANA timezone of the recording owner (e.g. "Europe/Paris").
         language: ISO 639-1 language code for transcription.
         download_link: URL to download the original recording.
         context_language: ISO 639-1 language code of the meeting summary context text.
@@ -208,17 +220,46 @@ def process_audio_transcribe_summarize_v2(
 
     task_id = self.request.id
 
-    transcription = transcribe_audio(task_id, filename, language)
+    # Transcribe the audio
+    transcription = transcribe_audio(task_id, recording_filename, language)
     if transcription is None:
         return
 
+    # Fetch recording start time from LiveKit egress JSON
+    recording_datetime = None
+    recording_start_dt = None
+    if worker_id:
+        folder = os.path.dirname(recording_filename)
+        egress_json_key = f"{folder}/{worker_id}.json"
+        try:
+            egress_info = file_service.read_json(egress_json_key)
+            started_at_ns = egress_info.get("started_at")
+            if started_at_ns:
+                recording_start_dt = datetime.fromtimestamp(
+                    started_at_ns / 1e9, tz=timezone.utc
+                )
+                recording_datetime = recording_start_dt.isoformat()
+        except Exception:
+            logger.warning(
+                "Could not read egress JSON %s, falling back to no recording datetime",
+                egress_json_key,
+            )
+
+    # Assign user names from metadata and transcription
+    metadata = file_service.read_json(metadata_filename)
+    if recording_start_dt is not None:
+        assign_speakers(
+            metadata, transcription, recording_start_dt, overlap_threshold=0.5
+        )
+
+    # Format output
     content, title = format_transcript(
         transcription,
         context_language,
         language,
         room,
-        recording_date,
-        recording_time,
+        recording_datetime,
+        owner_timezone,
         download_link,
     )
 

src/summary/summary/core/file_service.py

@@ -1,5 +1,6 @@
 """File service to encapsulate files' manipulations."""
 
+import json
 import logging
 import os
 import subprocess
@@ -156,6 +157,26 @@ class FileService:
                 os.remove(output_path)
             raise RuntimeError("Failed to extract audio.") from e
 
+    def read_json(self, object_name: str) -> dict:
+        """Read and parse a JSON file from MinIO storage."""
+        logger.info("Reading JSON: %s", object_name)
+
+        if not object_name:
+            raise ValueError("Invalid object_name")
+
+        response = None
+        try:
+            response = self._minio_client.get_object(self._bucket_name, object_name)
+            return json.loads(response.read())
+        except (MinioException, S3Error) as e:
+            raise FileServiceException(
+                "Unexpected error while reading JSON object."
+            ) from e
+        finally:
+            if response:
+                response.close()
+                response.release_conn()
+
     @contextmanager
     def prepare_audio_file(self, remote_object_key: str):
         """Download and prepare audio file for processing.

src/summary/summary/core/transcript_formatter.py

@@ -1,7 +1,9 @@
 """Transcript formatting into readable conversation format with speaker labels."""
 
 import logging
+from datetime import datetime
 from typing import Optional, Tuple
+from zoneinfo import ZoneInfo
 
 from summary.core.config import get_settings
 from summary.core.locales import LocaleStrings
@@ -40,8 +42,8 @@ class TranscriptFormatter:
         self,
         transcription,
         room: Optional[str] = None,
-        recording_date: Optional[str] = None,
-        recording_time: Optional[str] = None,
+        recording_datetime: Optional[str] = None,
+        owner_timezone: Optional[str] = None,
         download_link: Optional[str] = None,
     ) -> Tuple[str, str]:
         """Format transcription into the final document and its title."""
@@ -54,7 +56,7 @@ class TranscriptFormatter:
             content = self._remove_hallucinations(content)
             content = self._add_header(content, download_link)
 
-        title = self._generate_title(room, recording_date, recording_time)
+        title = self._generate_title(room, recording_datetime, owner_timezone)
 
         return content, title
 
@@ -98,15 +100,19 @@ class TranscriptFormatter:
     def _generate_title(
         self,
         room: Optional[str] = None,
-        recording_date: Optional[str] = None,
-        recording_time: Optional[str] = None,
+        recording_datetime: Optional[str] = None,
+        owner_timezone: Optional[str] = None,
     ) -> str:
         """Generate title from context or return default."""
-        if not room or not recording_date or not recording_time:
+        if not room or not recording_datetime:
             return self._locale.document_default_title
 
+        dt = datetime.fromisoformat(recording_datetime)
+        if owner_timezone:
+            dt = dt.astimezone(ZoneInfo(owner_timezone))
+
         return self._locale.document_title_template.format(
             room=room,
-            room_recording_date=recording_date,
-            room_recording_time=recording_time,
+            room_recording_date=dt.strftime("%Y-%m-%d"),
+            room_recording_time=dt.strftime("%H:%M"),
         )

src/summary/summary/core/user_assign.py

@@ -0,0 +1,281 @@
+"""Assign WhisperX diarization speakers to participant identities.
+
+Uses per-stream VAD events to match generic SPEAKER_XX labels provided
+by diarization to real user id's by computing time overlap between
+diarization segments and VAD intervals.
+
+Multiple speakers can map to the same participant (e.g. two people sharing
+one microphone). A participant with no matching speaker gets no assignment.
+"""
+
+import logging
+from dataclasses import dataclass, field
+from datetime import datetime
+from typing import Any
+
+logger = logging.getLogger(__name__)
+
+# Minimum fraction of a speaker's total duration that must overlap with a
+# participant's VAD to accept the assignment.
+DEFAULT_OVERLAP_THRESHOLD = 0.5
+
+
+@dataclass
+class Interval:
+    """A time interval in seconds relative to recording start."""
+
+    start: float
+    end: float
+
+
+@dataclass
+class SpeakerAssignment:
+    """Maps a diarization speaker label to a participant."""
+
+    speaker_label: str
+    participant_id: str
+    participant_name: str
+    score: float
+
+
+@dataclass
+class AssignmentResult:
+    """Result of speaker-to-participant assignment."""
+
+    assignments: list[SpeakerAssignment] = field(default_factory=list)
+    unassigned_speakers: list[str] = field(default_factory=list)
+
+    def apply(self, diarization: dict[str, Any]) -> dict[str, Any]:
+        """Return a copy of diarization with speaker labels replaced by names.
+
+        Replaces `"speaker"` fields in segments and word_segments with the
+        assigned participant name.  Unassigned speakers are left as-is.
+
+        Args:
+            diarization: WhisperX dict with `segments` and optionally
+                `word_segments`.
+
+        Returns:
+            New dict with speaker labels replaced.
+        """
+        speaker_to_name = {
+            a.speaker_label: a.participant_name for a in self.assignments
+        }
+
+        def _replace_speaker(item: dict[str, Any]) -> dict[str, Any]:
+            if "speaker" in item and item["speaker"] in speaker_to_name:
+                return {
+                    **item,
+                    "speaker": f"{speaker_to_name[item['speaker']]}:{item['speaker']}",  # We still use the original diarization
+                }
+            return item
+
+        result: dict[str, Any] = {}
+        for key, value in diarization.items():
+            if key in ("segments", "word_segments"):
+                new_items = []
+                for item in value:
+                    new_item = _replace_speaker(item)
+                    if key == "segments" and "words" in item:
+                        new_item["words"] = [_replace_speaker(w) for w in item["words"]]
+                    new_items.append(new_item)
+                result[key] = new_items
+            else:
+                result[key] = value
+        return result
+
+
+def _merge_intervals(intervals: list[Interval]) -> list[Interval]:
+    """Return a list of non-overlapping intervals sorted by start time."""
+    if not intervals:
+        return []
+    sorted_intervals = sorted(intervals, key=lambda interval: interval.start)
+    merged: list[Interval] = [
+        Interval(sorted_intervals[0].start, sorted_intervals[0].end)
+    ]
+    for interval in sorted_intervals[1:]:
+        if interval.start <= merged[-1].end:
+            merged[-1].end = max(merged[-1].end, interval.end)
+        else:
+            merged.append(Interval(interval.start, interval.end))
+    return merged
+
+
+def _total_duration(intervals: list[Interval]) -> float:
+    """Return the sum of all interval durations."""
+    return sum(interval.end - interval.start for interval in intervals)
+
+
+def _overlap_duration(
+    a_intervals: list[Interval],
+    b_intervals: list[Interval],
+) -> float:
+    """Compute total overlap between two merged interval lists, sorted by start time.
+
+    Uses a sweep-line approach in O(n + m).
+    """
+    overlap = 0.0
+    i = j = 0
+    while i < len(a_intervals) and j < len(b_intervals):
+        a = a_intervals[i]
+        b = b_intervals[j]
+        lo = max(a.start, b.start)
+        hi = min(a.end, b.end)
+        if lo < hi:
+            overlap += hi - lo
+        if a.end <= b.end:
+            i += 1
+        else:
+            j += 1
+    return overlap
+
+
+def _parse_iso(ts: str) -> datetime:
+    """Parse an ISO-formatted timestamp string."""
+    return datetime.fromisoformat(ts)
+
+
+def _build_participant_timelines(
+    metadata: dict[str, Any],
+    recording_start_datetime: datetime,
+) -> tuple[dict[str, list[Interval]], dict[str, str]]:
+    """Build VAD interval timelines for each participant.
+
+    Args:
+        metadata: Dict with `events` and `participants` keys.
+        recording_start_datetime: UTC datetime used as t=0 reference.
+
+    Returns:
+        participant_id → merged VAD intervals
+            (seconds relative to recording_start_datetime).
+        participant_id → display name.
+        Intervals are in seconds relative to recording_start_datetime.
+        Events before recording start are clamped to 0.
+    """
+    events = metadata.get("events", [])
+    participants_info = {
+        p["participantId"]: p.get("name", p["participantId"])
+        for p in metadata.get("participants", [])
+    }
+
+    ref_epoch = recording_start_datetime.timestamp()
+
+    open_starts: dict[str, float] = {}
+    intervals: dict[str, list[Interval]] = {}
+
+    for event in events:
+        pid = event["participant_id"]
+        ts = _parse_iso(event["timestamp"]).timestamp() - ref_epoch
+        etype = event["type"]
+
+        # TODO: close open speech_start
+        if etype == "speech_start":
+            open_starts[pid] = max(ts, 0.0)
+        elif etype == "speech_end":
+            start = open_starts.pop(pid, None)
+            if start is not None:
+                end = max(ts, 0.0)
+                if end > start:
+                    intervals.setdefault(pid, []).append(Interval(start, end))
+
+    for pid, pid_intervals in intervals.items():
+        intervals[pid] = _merge_intervals(pid_intervals)
+
+    return intervals, participants_info
+
+
+def _build_speaker_timelines(
+    transcription: Any,
+) -> dict[str, list[Interval]]:
+    """Build interval timelines from WhisperX transcription segments."""
+    intervals: dict[str, list[Interval]] = {}
+
+    segments = transcription.segments if hasattr(transcription, "segments") else []
+    for segment in segments:
+        speaker = segment.get("speaker")
+        if speaker is None:
+            continue
+        intervals.setdefault(speaker, []).append(
+            Interval(segment["start"], segment["end"])
+        )
+
+    for speaker, speaker_intervals in intervals.items():
+        intervals[speaker] = _merge_intervals(speaker_intervals)
+
+    return intervals
+
+
+def assign_speakers(
+    metadata: dict[str, Any],
+    transcription: Any,
+    recording_start_datetime: datetime,
+    overlap_threshold: float = DEFAULT_OVERLAP_THRESHOLD,
+) -> AssignmentResult:
+    """Match WhisperX speaker labels to participants.
+
+    Args:
+        metadata: User metadata with `events` and `participants`.
+        transcription: WhisperX Transcription object with a `segments` attribute.
+        recording_start_datetime: UTC datetime for t=0 reference.
+        overlap_threshold: Minimum overlap/speaker_duration to accept.
+
+    Returns:
+        AssignmentResult with per-speaker assignments and unassigned
+        speakers.
+    """
+    participant_timelines, participant_names = _build_participant_timelines(
+        metadata, recording_start_datetime
+    )
+    speaker_timelines = _build_speaker_timelines(transcription)
+
+    logger.info("Info info info")
+    logger.info(metadata)
+    logger.info(recording_start_datetime)
+    logger.info("participant_timelines")
+    logger.info(participant_timelines)
+    logger.info("participant_names")
+    logger.info(participant_names)
+
+    result = AssignmentResult()
+
+    for speaker, speaker_intervals in speaker_timelines.items():
+        speaker_duration = _total_duration(speaker_intervals)
+        if speaker_duration == 0:
+            result.unassigned_speakers.append(speaker)
+            continue
+
+        best_pid: str | None = None
+        best_score: float = 0.0
+
+        for pid, part_intervals in participant_timelines.items():
+            overlap = _overlap_duration(speaker_intervals, part_intervals)
+            score = overlap / speaker_duration
+            if score > best_score:
+                best_score = score
+                best_pid = pid
+
+        if best_pid is not None and best_score >= overlap_threshold:
+            result.assignments.append(
+                SpeakerAssignment(
+                    speaker_label=speaker,
+                    participant_id=best_pid,
+                    participant_name=participant_names.get(best_pid, best_pid),
+                    score=best_score,
+                )
+            )
+            logger.info(
+                "Assigned %s -> %s (score=%.3f)",
+                speaker,
+                participant_names.get(best_pid, best_pid),
+                best_score,
+            )
+        else:
+            result.unassigned_speakers.append(speaker)
+            logger.info(
+                "Speaker %s unassigned (best=%.3f, threshold=%.3f)",
+                speaker,
+                best_score,
+                overlap_threshold,
+            )
+
+    return result

src/summary/tests/api/test_api_tasks.py

@@ -19,12 +19,13 @@ class TestTasks:
             headers={"Authorization": "Bearer test-api-token"},
             json={
                 "owner_id": "owner-123",
-                "filename": "recording.mp4",
+                "recording_filename": "recording.mp4",
+                "metadata_filename": "metadata.json",  # TODO: change
                 "email": "user@example.com",
                 "sub": "sub-123",
                 "room": "room-abc",
-                "recording_date": "2026-01-01",
-                "recording_time": "10:00:00",
+                "worker_id": "EG_test123",
+                "owner_timezone": "UTC",
                 "language": None,
                 "download_link": "http://example.com/file.mp4",
             },
@@ -36,13 +37,14 @@ class TestTasks:
         args = mock_apply_async.call_args.kwargs["args"]
         assert args == [
             "owner-123",  # owner_id
-            "recording.mp4",  # filename
+            "recording.mp4",  # recording_filename
+            "metadata.json",  # metadata_filename
             "user@example.com",  # email
             "sub-123",  # sub
             1735725600.0,  # frozen time
             "room-abc",  # room
-            "2026-01-01",  # recording_date
-            "10:00:00",  # recording_time
+            "EG_test123",  # worker_id
+            "UTC",  # owner_timezone
             None,  # language
             "http://example.com/file.mp4",  # download_link
             None,  # context_language

src/summary/tests/unit/test_user_assign.py

@@ -0,0 +1,477 @@
+"""Tests for the speaker-to-user assignment service."""
+
+from dataclasses import dataclass, field
+from datetime import datetime
+
+from summary.core.user_assign import (
+    AssignmentResult,
+    Interval,
+    SpeakerAssignment,
+    _merge_intervals,
+    _overlap_duration,
+    _total_duration,
+    assign_speakers,
+)
+
+
+@dataclass
+class FakeTranscription:
+    """Mimics the OpenAI Transcription pydantic model for testing."""
+
+    segments: list = field(default_factory=list)
+
+RECORDING_START = datetime.fromisoformat("2026-03-17T15:30:33.000001")
+
+METADATA_SINGLE_USER = {
+    "events": [
+        {
+            "participant_id": "da8d39ff-3b1c-4e8d-9a70-c630c9871bcb",
+            "type": "participant_connected",
+            "timestamp": "2026-03-17T15:30:33.000001",
+        },
+        {
+            "participant_id": "da8d39ff-3b1c-4e8d-9a70-c630c9871bcb",
+            "type": "speech_start",
+            "timestamp": "2026-03-17T15:30:36.039456",
+        },
+        {
+            "participant_id": "da8d39ff-3b1c-4e8d-9a70-c630c9871bcb",
+            "type": "speech_end",
+            "timestamp": "2026-03-17T15:30:36.589114",
+        },
+        {
+            "participant_id": "da8d39ff-3b1c-4e8d-9a70-c630c9871bcb",
+            "type": "speech_start",
+            "timestamp": "2026-03-17T15:30:38.887518",
+        },
+        {
+            "participant_id": "da8d39ff-3b1c-4e8d-9a70-c630c9871bcb",
+            "type": "speech_end",
+            "timestamp": "2026-03-17T15:30:39.438141",
+        },
+        {
+            "participant_id": "da8d39ff-3b1c-4e8d-9a70-c630c9871bcb",
+            "type": "participant_disconnected",
+            "timestamp": "2026-03-17T15:30:43.223255",
+        },
+    ],
+    "participants": [
+        {
+            "participantId": "da8d39ff-3b1c-4e8d-9a70-c630c9871bcb",
+            "name": "cameledev",
+        }
+    ],
+}
+
+DIARIZATION_SINGLE_SPEAKER = FakeTranscription(
+    segments=[
+        {
+            "start": 1.363,
+            "end": 3.545,
+            "text": " The stale smell.",
+            "speaker": "SPEAKER_00",
+        },
+        {
+            "start": 4.466,
+            "end": 6.247,
+            "text": "It takes heat.",
+            "speaker": "SPEAKER_00",
+        },
+    ],
+)
+
+USER_ID = "da8d39ff-3b1c-4e8d-9a70-c630c9871bcb"
+
+
+class TestMergeIntervals:
+    """Tests for _merge_intervals."""
+
+    def test_empty(self):
+        """Empty input returns empty list."""
+        assert _merge_intervals([]) == []
+
+    def test_no_overlap(self):
+        """Non-overlapping intervals stay separate."""
+        result = _merge_intervals([Interval(1, 2), Interval(3, 4)])
+        assert len(result) == 2
+
+    def test_overlap(self):
+        """Overlapping intervals are merged."""
+        result = _merge_intervals([Interval(1, 3), Interval(2, 4)])
+        assert len(result) == 1
+        assert result[0].start == 1
+        assert result[0].end == 4
+
+    def test_adjacent(self):
+        """Adjacent intervals are merged."""
+        result = _merge_intervals([Interval(1, 2), Interval(2, 3)])
+        assert len(result) == 1
+        assert result[0].start == 1
+        assert result[0].end == 3
+
+    def test_unsorted(self):
+        """Unsorted input is sorted before merging."""
+        result = _merge_intervals([Interval(5, 6), Interval(1, 2)])
+        assert len(result) == 2
+        assert result[0].start == 1
+
+
+class TestOverlapDuration:
+    """Tests for _overlap_duration."""
+
+    def test_no_overlap(self):
+        """Disjoint intervals have zero overlap."""
+        a = [Interval(1, 2)]
+        b = [Interval(3, 4)]
+        assert _overlap_duration(a, b) == 0.0
+
+    def test_full_overlap(self):
+        """Identical intervals have full overlap."""
+        a = [Interval(1, 3)]
+        b = [Interval(1, 3)]
+        assert _overlap_duration(a, b) == 2.0
+
+    def test_partial_overlap(self):
+        """Partially overlapping intervals."""
+        a = [Interval(1, 3)]
+        b = [Interval(2, 4)]
+        assert _overlap_duration(a, b) == 1.0
+
+    def test_multiple_intervals(self):
+        """Multiple intervals with a spanning interval."""
+        a = [Interval(1, 3), Interval(5, 7)]
+        b = [Interval(2, 6)]
+        assert _overlap_duration(a, b) == 2.0
+
+    def test_empty(self):
+        """Empty input yields zero overlap."""
+        assert _overlap_duration([], [Interval(1, 2)]) == 0.0
+        assert _overlap_duration([Interval(1, 2)], []) == 0.0
+
+
+class TestTotalDuration:
+    """Tests for _total_duration."""
+
+    def test_basic(self):
+        """Sum of durations for multiple intervals."""
+        ivs = [Interval(0, 1), Interval(2, 5)]
+        assert _total_duration(ivs) == 4.0
+
+    def test_empty(self):
+        """Empty input returns zero."""
+        assert _total_duration([]) == 0.0
+
+
+class TestAssignSpeakers:
+    """Tests for assign_speakers."""
+
+    def test_single_speaker_single_user(self):
+        """Single speaker assigned to single user with low threshold."""
+        result = assign_speakers(
+            METADATA_SINGLE_USER,
+            DIARIZATION_SINGLE_SPEAKER,
+            RECORDING_START,
+            overlap_threshold=0.2,
+        )
+        assert len(result.assignments) == 1
+        assert result.assignments[0].speaker_label == "SPEAKER_00"
+        assert result.assignments[0].participant_id == USER_ID
+        assert result.assignments[0].participant_name == "cameledev"
+        assert result.assignments[0].score > 0
+        assert result.unassigned_speakers == []
+
+    def test_no_vad_events(self):
+        """Participant with no speech leaves speakers unassigned."""
+        metadata = {
+            "events": [
+                {
+                    "participant_id": "user-a",
+                    "type": "participant_connected",
+                    "timestamp": "2026-03-17T15:30:33.000000",
+                },
+            ],
+            "participants": [{"participantId": "user-a", "name": "Silent User"}],
+        }
+        result = assign_speakers(
+            metadata,
+            DIARIZATION_SINGLE_SPEAKER,
+            RECORDING_START,
+        )
+        assert len(result.assignments) == 0
+        assert "SPEAKER_00" in result.unassigned_speakers
+
+    def test_multiple_speakers_same_user(self):
+        """Two speakers from same mic both assigned to same user."""
+        metadata = {
+            "events": [
+                {
+                    "participant_id": "user-a",
+                    "type": "speech_start",
+                    "timestamp": "2026-03-17T15:30:35.000000",
+                },
+                {
+                    "participant_id": "user-a",
+                    "type": "speech_end",
+                    "timestamp": "2026-03-17T15:30:50.000000",
+                },
+            ],
+            "participants": [{"participantId": "user-a", "name": "Shared Mic"}],
+        }
+        transcription = FakeTranscription(
+            segments=[
+                {"start": 1.0, "end": 3.0, "speaker": "SPEAKER_00"},
+                {"start": 5.0, "end": 7.0, "speaker": "SPEAKER_01"},
+            ],
+        )
+        result = assign_speakers(metadata, transcription, RECORDING_START)
+        assert len(result.assignments) == 2
+        pids = {a.participant_id for a in result.assignments}
+        assert pids == {"user-a"}
+
+    def test_two_users_two_speakers(self):
+        """Each speaker maps to correct user by VAD overlap."""
+        metadata = {
+            "events": [
+                {
+                    "participant_id": "user-a",
+                    "type": "speech_start",
+                    "timestamp": "2026-03-17T15:30:34.000001",
+                },
+                {
+                    "participant_id": "user-a",
+                    "type": "speech_end",
+                    "timestamp": "2026-03-17T15:30:37.000001",
+                },
+                {
+                    "participant_id": "user-b",
+                    "type": "speech_start",
+                    "timestamp": "2026-03-17T15:30:38.000001",
+                },
+                {
+                    "participant_id": "user-b",
+                    "type": "speech_end",
+                    "timestamp": "2026-03-17T15:30:41.000001",
+                },
+            ],
+            "participants": [
+                {"participantId": "user-a", "name": "Alice"},
+                {"participantId": "user-b", "name": "Bob"},
+            ],
+        }
+        transcription = FakeTranscription(
+            segments=[
+                {"start": 1.5, "end": 3.5, "speaker": "SPEAKER_00"},
+                {"start": 5.5, "end": 7.5, "speaker": "SPEAKER_01"},
+            ],
+        )
+        result = assign_speakers(metadata, transcription, RECORDING_START)
+        assert len(result.assignments) == 2
+        by_speaker = {a.speaker_label: a for a in result.assignments}
+        assert by_speaker["SPEAKER_00"].participant_name == "Alice"
+        assert by_speaker["SPEAKER_01"].participant_name == "Bob"
+
+    def test_overlapping_speech_two_users(self):
+        """Simultaneous speech from two users still assigns each speaker correctly."""
+        # user-a speaks from t=1s to t=6s, user-b speaks from t=3s to t=8s
+        # (3s overlap where both are speaking)
+        # SPEAKER_00 diarization covers t=1.5–5.5 (mostly user-a)
+        # SPEAKER_01 diarization covers t=4.0–7.5 (mostly user-b)
+        metadata = {
+            "events": [
+                {
+                    "participant_id": "user-a",
+                    "type": "speech_start",
+                    "timestamp": "2026-03-17T15:30:34.000001",
+                },
+                {
+                    "participant_id": "user-b",
+                    "type": "speech_start",
+                    "timestamp": "2026-03-17T15:30:36.000001",
+                },
+                {
+                    "participant_id": "user-a",
+                    "type": "speech_end",
+                    "timestamp": "2026-03-17T15:30:39.000001",
+                },
+                {
+                    "participant_id": "user-b",
+                    "type": "speech_end",
+                    "timestamp": "2026-03-17T15:30:41.000001",
+                },
+            ],
+            "participants": [
+                {"participantId": "user-a", "name": "Alice"},
+                {"participantId": "user-b", "name": "Bob"},
+            ],
+        }
+        transcription = FakeTranscription(
+            segments=[
+                {"start": 1.5, "end": 5.5, "speaker": "SPEAKER_00"},
+                {"start": 4.0, "end": 7.5, "speaker": "SPEAKER_01"},
+            ],
+        )
+        result = assign_speakers(
+            metadata, transcription, RECORDING_START, overlap_threshold=0.3
+        )
+        assert len(result.assignments) == 2
+        by_speaker = {a.speaker_label: a for a in result.assignments}
+        assert by_speaker["SPEAKER_00"].participant_name == "Alice"
+        assert by_speaker["SPEAKER_01"].participant_name == "Bob"
+        assert result.unassigned_speakers == []
+
+    def test_below_threshold(self):
+        """Speaker with minimal overlap stays unassigned."""
+        metadata = {
+            "events": [
+                {
+                    "participant_id": "user-a",
+                    "type": "speech_start",
+                    "timestamp": "2026-03-17T15:30:34.000001",
+                },
+                {
+                    "participant_id": "user-a",
+                    "type": "speech_end",
+                    "timestamp": "2026-03-17T15:30:35.169950",
+                },
+            ],
+            "participants": [{"participantId": "user-a", "name": "Brief User"}],
+        }
+        transcription = FakeTranscription(
+            segments=[
+                {"start": 1.0, "end": 10.0, "speaker": "SPEAKER_00"},
+            ],
+        )
+        result = assign_speakers(
+            metadata,
+            transcription,
+            RECORDING_START,
+            overlap_threshold=0.5,
+        )
+        assert len(result.assignments) == 0
+        assert "SPEAKER_00" in result.unassigned_speakers
+
+    def test_events_before_recording_start_clamped(self):
+        """Speech events before recording start are clamped to t=0."""
+        metadata = {
+            "events": [
+                {
+                    "participant_id": "user-a",
+                    "type": "speech_start",
+                    "timestamp": "2026-03-17T15:30:31.000001",  # before RECORDING_START
+                },
+                {
+                    "participant_id": "user-a",
+                    "type": "speech_end",
+                    "timestamp": "2026-03-17T15:30:36.000001",  # after RECORDING_START
+                },
+            ],
+            "participants": [{"participantId": "user-a", "name": "Early User"}],
+        }
+        transcription = FakeTranscription(
+            segments=[
+                {"start": 0.0, "end": 3.0, "speaker": "SPEAKER_00"},
+            ],
+        )
+        result = assign_speakers(metadata, transcription, RECORDING_START)
+        assert len(result.assignments) == 1
+        assert result.assignments[0].participant_name == "Early User"
+
+    def test_empty_diarization(self):
+        """No segments produces empty result."""
+        result = assign_speakers(
+            METADATA_SINGLE_USER, FakeTranscription(segments=[]), RECORDING_START
+        )
+        assert result == AssignmentResult()
+
+    def test_segment_without_speaker_ignored(self):
+        """Segments missing speaker key are skipped."""
+        transcription = FakeTranscription(
+            segments=[
+                {"start": 1.0, "end": 3.0, "text": "no speaker"},
+            ],
+        )
+        result = assign_speakers(METADATA_SINGLE_USER, transcription, RECORDING_START)
+        assert result == AssignmentResult()
+
+
+class TestApply:
+    """Tests for AssignmentResult.apply."""
+
+    def test_replaces_speakers_in_segments_and_words(self):
+        """Speaker labels replaced in segments, words, and word_segments."""
+        diarization = {
+            "segments": [
+                {
+                    "start": 1.0,
+                    "end": 3.0,
+                    "text": "Hello",
+                    "speaker": "SPEAKER_00",
+                    "words": [
+                        {
+                            "word": "Hello",
+                            "start": 1.0,
+                            "end": 1.5,
+                            "speaker": "SPEAKER_00",
+                        },
+                    ],
+                },
+                {
+                    "start": 4.0,
+                    "end": 6.0,
+                    "text": "World",
+                    "speaker": "SPEAKER_01",
+                    "words": [
+                        {
+                            "word": "World",
+                            "start": 4.0,
+                            "end": 4.5,
+                            "speaker": "SPEAKER_01",
+                        },
+                    ],
+                },
+            ],
+            "word_segments": [
+                {"word": "Hello", "start": 1.0, "end": 1.5, "speaker": "SPEAKER_00"},
+                {"word": "World", "start": 4.0, "end": 4.5, "speaker": "SPEAKER_01"},
+            ],
+        }
+        assignment = AssignmentResult(
+            assignments=[
+                SpeakerAssignment("SPEAKER_00", "id-a", "Alice", 0.9),
+                SpeakerAssignment("SPEAKER_01", "id-b", "Bob", 0.8),
+            ],
+        )
+        result = assignment.apply(diarization)
+
+        assert result["segments"][0]["speaker"] == "Alice"
+        assert result["segments"][0]["words"][0]["speaker"] == "Alice"
+        assert result["segments"][1]["speaker"] == "Bob"
+        assert result["word_segments"][0]["speaker"] == "Alice"
+        assert result["word_segments"][1]["speaker"] == "Bob"
+
+    def test_unassigned_speakers_unchanged(self):
+        """Unassigned speaker labels are left as-is."""
+        diarization = {
+            "segments": [
+                {"start": 1.0, "end": 3.0, "speaker": "SPEAKER_02"},
+            ],
+        }
+        assignment = AssignmentResult(
+            assignments=[
+                SpeakerAssignment("SPEAKER_00", "id-a", "Alice", 0.9),
+            ],
+            unassigned_speakers=["SPEAKER_02"],
+        )
+        result = assignment.apply(diarization)
+        assert result["segments"][0]["speaker"] == "SPEAKER_02"
+
+    def test_preserves_extra_keys(self):
+        """Non-segment keys in diarization are preserved."""
+        diarization = {
+            "segments": [],
+            "language": "en",
+            "custom_field": 42,
+        }
+        result = AssignmentResult().apply(diarization)
+        assert result["language"] == "en"
+        assert result["custom_field"] == 42