diff --git a/.changeset/calm-dodos-send.md b/.changeset/calm-dodos-send.md new file mode 100644 index 000000000..5cf479f77 --- /dev/null +++ b/.changeset/calm-dodos-send.md @@ -0,0 +1,5 @@ +--- +'@livekit/agents': patch +--- + +Send `lk.agent.session` messages through the one-shot byte-stream path. diff --git a/agents/src/voice/remote_session.test.ts b/agents/src/voice/remote_session.test.ts index bf2e93cdd..ded57d423 100644 --- a/agents/src/voice/remote_session.test.ts +++ b/agents/src/voice/remote_session.test.ts @@ -3,8 +3,10 @@ // SPDX-License-Identifier: Apache-2.0 import { AgentSession as pb } from '@livekit/protocol'; import type { Room } from '@livekit/rtc-node'; +import { readFile } from 'node:fs/promises'; import * as net from 'node:net'; import { afterEach, beforeAll, describe, expect, it, vi } from 'vitest'; +import { TOPIC_SESSION_MESSAGES } from '../constants.js'; import type { InferenceExecutor } from '../ipc/inference_executor.js'; import { JobContext, @@ -264,6 +266,85 @@ function createConnectedTransportPair(): [FakeTransport, FakeTransport] { return [client, host]; } +describe('RoomSessionTransport sends', () => { + it('serializes concurrent room sends', async () => { + let inFlight = 0; + let overlapping = false; + const sendFile = vi.fn(async () => { + inFlight += 1; + if (inFlight > 1) overlapping = true; + await new Promise((resolve) => setTimeout(resolve, 10)); + inFlight -= 1; + }); + const room = { isConnected: true, localParticipant: { sendFile } }; + const transport = new RoomSessionTransport( + room as unknown as Room, + { + linkedParticipant: undefined, + } as unknown as RoomIO, + ); + + await Promise.all( + Array.from({ length: 8 }, (_, index) => transport.sendMessage(pingMessage(`r${index}`))), + ); + + expect(overlapping).toBe(false); + }); + + it('sends each room message with one one-shot call', async () => { + // A one-shot send lets the SDK inline small messages and compress large ones. An + // incremental stream always costs a header, chunk, and trailer for each message. + let sentData: Uint8Array | undefined; + const sendFile = vi.fn(async (path: string) => { + sentData = await readFile(path); + }); + const streamBytes = vi.fn(); + const room = { + isConnected: true, + localParticipant: { sendFile, streamBytes }, + }; + const transport = new RoomSessionTransport( + room as unknown as Room, + { + linkedParticipant: { identity: 'user-1' }, + } as unknown as RoomIO, + ); + + const msg = pingMessage('r1'); + await transport.sendMessage(msg); + + expect(streamBytes).not.toHaveBeenCalled(); + expect(sendFile).toHaveBeenCalledOnce(); + expect([...sentData!]).toEqual([...msg.toBinary()]); + expect(sendFile).toHaveBeenCalledWith( + expect.any(String), + expect.objectContaining({ + topic: TOPIC_SESSION_MESSAGES, + destinationIdentities: ['user-1'], + }), + ); + }); + + it('reports one-shot send failures instead of dropping the message', async () => { + const room = { + isConnected: true, + localParticipant: { + sendFile: vi.fn().mockRejectedValue(new Error('stream refused')), + }, + }; + const transport = new RoomSessionTransport( + room as unknown as Room, + { + linkedParticipant: undefined, + } as unknown as RoomIO, + ); + + await expect(transport.sendMessage(pingMessage('r1'))).rejects.toThrow( + /failed to send binary stream message/, + ); + }); +}); + // A message that cannot be sent is a request somebody waits out in full, so the // transport reports the failure and each caller decides what to do with it, as // the Python SessionTransport does. Events are the one caller nobody awaits. diff --git a/agents/src/voice/remote_session.ts b/agents/src/voice/remote_session.ts index 7868326ee..c334b87df 100644 --- a/agents/src/voice/remote_session.ts +++ b/agents/src/voice/remote_session.ts @@ -2,12 +2,16 @@ // // SPDX-License-Identifier: Apache-2.0 import { Duration, Timestamp } from '@bufbuild/protobuf'; +import { Mutex } from '@livekit/mutex'; import { SimulationRun, AgentSession as pb } from '@livekit/protocol'; import type { ByteStreamReader, Room, TextStreamInfo } from '@livekit/rtc-node'; import { ThrowsPromise } from '@livekit/throws-transformer/throws'; import type { TypedEventEmitter } from '@livekit/typed-emitter'; import EventEmitter from 'events'; +import { rm, writeFile } from 'node:fs/promises'; import * as net from 'node:net'; +import { tmpdir } from 'node:os'; +import { join } from 'node:path'; import { TOPIC_SESSION_MESSAGES } from '../constants.js'; import type { OverlappingSpeechEvent } from '../inference/interruption/types.js'; import { getJobContext } from '../job.js'; @@ -110,6 +114,7 @@ export class RoomSessionTransport extends SessionTransport { private pendingMessages: pb.AgentSessionMessage[] = []; private waitingResolve: ((value: IteratorResult) => void) | null = null; private roomIO: RoomIO; + private sendLock = new Mutex(); constructor(room: Room, roomIO: RoomIO) { super(); @@ -166,21 +171,30 @@ export class RoomSessionTransport extends SessionTransport { throw new Error('room session transport is closed'); } + const unlock = await this.sendLock.lock(); + const name = shortuuid('AS_'); + const path = join(tmpdir(), name); try { const data = msg.toBinary(); - const opts: Record = { + // rtc-node currently exposes one-shot byte streams through its file-backed API. + await writeFile(path, data, { flag: 'wx', mode: 0o600 }); + const opts = { topic: TOPIC_SESSION_MESSAGES, - name: shortuuid('AS_'), + name, + destinationIdentities: undefined as string[] | undefined, }; const remoteIdentity = this.getRemoteIdentity(); if (remoteIdentity) { opts.destinationIdentities = [remoteIdentity]; } - const writer = await this.room.localParticipant!.streamBytes(opts); - await writer.write(new Uint8Array(data)); - await writer.close(); + await this.room.localParticipant!.sendFile(path, opts); } catch (e) { throw new Error(`failed to send binary stream message: ${e}`, { cause: e }); + } finally { + unlock(); + await rm(path, { force: true }).catch((e) => { + log().warn({ error: e, path }, 'failed to remove temporary session message'); + }); } }