diff --git a/.changeset/wait-for-active-participant.md b/.changeset/wait-for-active-participant.md new file mode 100644 index 000000000..53d1efcb6 --- /dev/null +++ b/.changeset/wait-for-active-participant.md @@ -0,0 +1,14 @@ +--- +'@livekit/agents': patch +--- + +`waitForParticipant` now waits for a participant to become _active_ rather than merely connected. + +A remote participant can only receive data messages once it reaches `ParticipantState.ACTIVE`. +Resolving on `ParticipantConnected` handed back a participant that is present in +`room.remoteParticipants` but not yet reachable, so anything sent to it was silently dropped — +most visibly in `DataStreamAudioOutput`, where avatar audio could be published before the avatar +worker could receive it. This matches the Python SDK's `wait_for_participant`, which has always +waited on `participant_active`. + +Requires `@livekit/rtc-node` with `RoomEvent.ParticipantActive` and `Participant.state`. diff --git a/agents/src/utils.ts b/agents/src/utils.ts index 034004ff7..3ce8891dd 100644 --- a/agents/src/utils.ts +++ b/agents/src/utils.ts @@ -9,7 +9,13 @@ import type { Room, TrackKind, } from '@livekit/rtc-node'; -import { AudioFrame, AudioResampler, RemoteParticipant, RoomEvent } from '@livekit/rtc-node'; +import { + AudioFrame, + AudioResampler, + ParticipantState, + RemoteParticipant, + RoomEvent, +} from '@livekit/rtc-node'; import { type Throws, ThrowsPromise } from '@livekit/throws-transformer/throws'; import { AsyncLocalStorage } from 'node:async_hooks'; import { randomUUID } from 'node:crypto'; @@ -1096,8 +1102,18 @@ export async function waitUntilAborted( /** * Returns a participant that matches the given identity. If identity is None, the first - * participant that joins the room will be returned. - * If the participant has already joined, the function will return immediately. + * participant that becomes active in the room will be returned. + * If the participant is already active, the function will return immediately. + * + * A remote participant is only considered a match once it reaches + * `ParticipantState.ACTIVE` — the point at which it can receive data messages. + * A participant that has connected but is still JOINING/JOINED is present in + * `room.remoteParticipants` yet not reachable, so waiting on connection alone would + * hand back a participant that silently drops anything sent to it. + * + * The local participant (via `includeLocal`) is exempt: it has no remote lifecycle to + * wait on. + * * @param room - The room to wait for a participant in. * @param identity - The identity of the participant to wait for. * @param kind - The kind of the participant to wait for. @@ -1161,7 +1177,7 @@ export async function waitForParticipant({ return participant.kind === kind; }; - const onParticipantConnected = (p: RemoteParticipant) => { + const onParticipantActive = (p: RemoteParticipant) => { if ((identity === undefined || p.identity === identity) && kindMatch(p)) { if (!fut.done) { fut.resolve(p); @@ -1179,7 +1195,7 @@ export async function waitForParticipant({ } }; - room.on(RoomEvent.ParticipantConnected, onParticipantConnected); + room.on(RoomEvent.ParticipantActive, onParticipantActive); room.on(RoomEvent.Disconnected, onDisconnected); signal?.addEventListener('abort', onAbort, { once: true }); @@ -1195,7 +1211,9 @@ export async function waitForParticipant({ } for (const p of room.remoteParticipants.values()) { - onParticipantConnected(p); + if (p.state === ParticipantState.ACTIVE) { + onParticipantActive(p); + } if (fut.done) { break; } @@ -1203,7 +1221,7 @@ export async function waitForParticipant({ return await fut.await; } finally { - room.off(RoomEvent.ParticipantConnected, onParticipantConnected); + room.off(RoomEvent.ParticipantActive, onParticipantActive); room.off(RoomEvent.Disconnected, onDisconnected); signal?.removeEventListener('abort', onAbort); } diff --git a/agents/src/wait_for_participant.test.ts b/agents/src/wait_for_participant.test.ts new file mode 100644 index 000000000..aa5b4018e --- /dev/null +++ b/agents/src/wait_for_participant.test.ts @@ -0,0 +1,126 @@ +// SPDX-FileCopyrightText: 2026 LiveKit, Inc. +// +// SPDX-License-Identifier: Apache-2.0 +import type { RemoteParticipant } from '@livekit/rtc-node'; +import { ParticipantKind, ParticipantState, Room, RoomEvent } from '@livekit/rtc-node'; +import { describe, expect, it, vi } from 'vitest'; +import { isPending } from './utils.js'; +import { waitForParticipant } from './utils.js'; + +/** A Room stub exposing only what waitForParticipant touches. */ +const mockRoom = () => { + const room = new Room(); + vi.spyOn(room, 'isConnected', 'get').mockReturnValue(true); + return room; +}; + +const makeParticipant = ( + identity: string, + state: ParticipantState, + kind: ParticipantKind = ParticipantKind.STANDARD, +): RemoteParticipant => ({ identity, state, kind }) as unknown as RemoteParticipant; + +/** Add a participant to the room the way the SDK's participantConnected handler would. */ +const connect = (room: Room, p: RemoteParticipant) => { + room.remoteParticipants.set(p.identity, p); +}; + +/** Promote a participant the way the SDK's participantActive handler would. */ +const activate = (room: Room, p: RemoteParticipant) => { + (p as { state: ParticipantState }).state = ParticipantState.ACTIVE; + room.emit(RoomEvent.ParticipantActive, p); +}; + +describe('waitForParticipant', () => { + it('returns immediately for an already-active participant', async () => { + const room = mockRoom(); + const alice = makeParticipant('alice', ParticipantState.ACTIVE); + connect(room, alice); + + await expect(waitForParticipant({ room, identity: 'alice' })).resolves.toBe(alice); + }); + + it('keeps waiting for a participant that has connected but is not active yet', async () => { + const room = mockRoom(); + const alice = makeParticipant('alice', ParticipantState.JOINED); + connect(room, alice); + + const pending = waitForParticipant({ room, identity: 'alice' }); + expect(await isPending(pending)).toBe(true); + + activate(room, alice); + await expect(pending).resolves.toBe(alice); + }); + + it('resolves on the ParticipantActive event for a participant that joins later', async () => { + const room = mockRoom(); + const pending = waitForParticipant({ room, identity: 'alice' }); + expect(await isPending(pending)).toBe(true); + + const alice = makeParticipant('alice', ParticipantState.JOINED); + connect(room, alice); + // connecting alone must not resolve the wait + expect(await isPending(pending)).toBe(true); + + activate(room, alice); + await expect(pending).resolves.toBe(alice); + }); + + it('ignores an active participant with a different identity', async () => { + const room = mockRoom(); + connect(room, makeParticipant('bob', ParticipantState.ACTIVE)); + + const pending = waitForParticipant({ room, identity: 'alice' }); + expect(await isPending(pending)).toBe(true); + + const alice = makeParticipant('alice', ParticipantState.JOINED); + connect(room, alice); + activate(room, alice); + await expect(pending).resolves.toBe(alice); + }); + + it('honours the kind filter', async () => { + const room = mockRoom(); + const standard = makeParticipant('standard', ParticipantState.ACTIVE); + connect(room, standard); + + const pending = waitForParticipant({ room, kind: ParticipantKind.AGENT }); + expect(await isPending(pending)).toBe(true); + + const agent = makeParticipant('agent', ParticipantState.JOINED, ParticipantKind.AGENT); + connect(room, agent); + activate(room, agent); + await expect(pending).resolves.toBe(agent); + }); + + it('rejects when the room disconnects while waiting', async () => { + const room = mockRoom(); + connect(room, makeParticipant('alice', ParticipantState.JOINED)); + + const pending = waitForParticipant({ room, identity: 'alice' }); + room.emit(RoomEvent.Disconnected, undefined as never); + + await expect(pending).rejects.toThrow('Got disconnected from room while waiting'); + }); + + it('rejects when the abort signal fires', async () => { + const room = mockRoom(); + connect(room, makeParticipant('alice', ParticipantState.JOINED)); + + const controller = new AbortController(); + const pending = waitForParticipant({ room, identity: 'alice', signal: controller.signal }); + controller.abort(); + + await expect(pending).rejects.toThrow('waitForParticipant aborted'); + }); + + it('returns the local participant without waiting on remote state', async () => { + const room = mockRoom(); + const local = { identity: 'agent', kind: ParticipantKind.AGENT }; + room.localParticipant = local as unknown as NonNullable; + + await expect(waitForParticipant({ room, identity: 'agent', includeLocal: true })).resolves.toBe( + local, + ); + }); +});