Send lk.agent.session messages with one-shot byte sends

.changeset/calm-dodos-send.md

@@ -0,0 +1,5 @@
+---
+'@livekit/agents': patch
+---
+
+Send `lk.agent.session` messages through the one-shot byte-stream path.

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.

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<pb.AgentSessionMessage>) => 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<string, unknown> = {
+      // 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');
+      });
     }
   }