fix: wait for a participant to become active, not just connected

.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`.

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<T>(
 
 /**
  * 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,15 +1211,17 @@ export async function waitForParticipant({
     }
 
     for (const p of room.remoteParticipants.values()) {
-      onParticipantConnected(p);
+      if (p.state === ParticipantState.ACTIVE) {
+        onParticipantActive(p);
+      }
       if (fut.done) {
         break;
       }
     }
 
     return await fut.await;
   } finally {
-    room.off(RoomEvent.ParticipantConnected, onParticipantConnected);
+    room.off(RoomEvent.ParticipantActive, onParticipantActive);
     room.off(RoomEvent.Disconnected, onDisconnected);
     signal?.removeEventListener('abort', onAbort);
   }

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<Room['localParticipant']>;
+
+    await expect(waitForParticipant({ room, identity: 'agent', includeLocal: true })).resolves.toBe(
+      local,
+    );
+  });
+});