Repository: livekit/agents
File: livekit-agents/livekit/agents/job.py (Python)

Code before:
    def add_participant_entrypoint(
        self,
        entrypoint_fnc: Callable[[JobContext, rtc.RemoteParticipant], Coroutine[None, None, None]],
        *_: Any,
        kind: list[rtc.ParticipantKind.ValueType]
        | rtc.ParticipantKind.ValueType = DEFAULT_PARTICIPANT_KINDS,
    ) -> None:
        """Adds an entrypoint function to be run when a participant joins the room. In cases where
        the participant has already joined, the entrypoint will be run immediately. Multiple unique entrypoints can be
        added and they will each be run in parallel for each participant.
        """  # noqa: E501

        if entrypoint_fnc in [e for (e, _) in self._participant_entrypoints]:
            raise ValueError("entrypoints cannot be added more than once")

        self._participant_entrypoints.append((entrypoint_fnc, kind))

    def init_recording(self, options: RecordingOptions) -> None:
        if self._recording_initialized or not is_cloud(self._info.url):
            self._stop_log_buffering()
            return

        self._recording_initialized = True
        cloud_hostname = urlparse(self._info.url).hostname
        logger.debug("configuring session recording", extra={"hostname": cloud_hostname})
        if cloud_hostname:
            _setup_cloud_tracer(
                room_id=self.job.room.sid,
                job_id=self.job.id,
                cloud_hostname=cloud_hostname,
                enable_traces=options["traces"],
                enable_logs=options["logs"],
            )
            self._flush_early_log_buffer(replay=True)

    def _participant_available(self, p: rtc.RemoteParticipant) -> None:
        for coro, kind in self._participant_entrypoints:
            if isinstance(kind, list):
                if p.kind not in kind:
                    continue
            else:
                if p.kind != kind:
                    continue

            if (p.identity, coro) in self._participant_tasks:
                logger.warning(
                    f"a participant has joined before a prior participant task matching the same identity has finished: '{p.identity}'"  # noqa: E501
                )
            task_name = f"part-entry-{p.identity}-{coro.__name__}"
            task = asyncio.create_task(coro(self, p), name=task_name)
            self._participant_tasks[(p.identity, coro)] = task

Review comment (bug):
Do we need this early log buffer only because `init_recording` is called too late? (can crash before).

Can you add a comment for that 🙏

Code after:

        if entrypoint_fnc in [e for (e, _) in self._participant_entrypoints]:
            raise ValueError("entrypoints cannot be added more than once")

        self._participant_entrypoints.append((entrypoint_fnc, kind))

    def init_recording(self, options: RecordingOptions) -> None:
        if self._recording_initialized or not is_cloud(self._info.url):
            self._stop_log_buffering()
            return

        self._recording_initialized = True
        cloud_hostname = urlparse(self._info.url).hostname
        logger.debug("configuring session recording", extra={"hostname": cloud_hostname})
        if cloud_hostname:
            _setup_cloud_tracer(
                room_id=self.job.room.sid,
                job_id=self.job.id,
                cloud_hostname=cloud_hostname,
                enable_traces=options["traces"],
                enable_logs=options["logs"],
            )
            # init_recording is typically called during session.start(), at which point a bunch of
            # the logs would have already been emitted. we want to capture all of the logs as it
            # relates to the job
            self._flush_early_log_buffer(replay=True)

    def _participant_available(self, p: rtc.RemoteParticipant) -> None:
        for coro, kind in self._participant_entrypoints:
            if isinstance(kind, list):
                if p.kind not in kind:
                    continue
            else:
                if p.kind != kind:
                    continue

            if (p.identity, coro) in self._participant_tasks:
                logger.warning(
                    f"a participant has joined before a prior participant task matching the same identity has finished: '{p.identity}'"  # noqa: E501
                )
            task_name = f"part-entry-{p.identity}-{coro.__name__}"
            task = asyncio.create_task(coro(self, p), name=task_name)
            self._participant_tasks[(p.identity, coro)] = task
            task.add_done_callback(
                lambda _, coro=coro: self._participant_tasks.pop((p.identity, coro))  # type: ignore
            )

    def token_claims(self) -> Claims:
        return api.TokenVerifier().verify(self._info.token, verify_signature=False)