fix(voice): wire debug messages over remote sessions (#1645)

Co-authored-by: rosetta-livekit-bot[bot] <282703043+rosetta-livekit-bot[bot]@users.noreply.github.com>
Co-authored-by: Brian Yin <brian.yin@livekit.io>

Author: rosetta-livekit-bot[bot]

.changeset/debug-message-remote-session.md

@@ -0,0 +1,5 @@
+---
+"@livekit/agents": patch
+---
+
+Wire internal debug messages through remote sessions.

agents/package.json

@@ -53,9 +53,9 @@
     "@bufbuild/protobuf": "^1.10.0",
     "@ffmpeg-installer/ffmpeg": "^1.1.0",
     "@livekit/mutex": "^1.1.1",
-    "@livekit/protocol": "^1.45.7",
-    "@livekit/typed-emitter": "^3.0.0",
+    "@livekit/protocol": "^1.46.4",
     "@livekit/throws-transformer": "0.1.8",
+    "@livekit/typed-emitter": "^3.0.0",
     "@opentelemetry/api": "^1.9.0",
     "@opentelemetry/api-logs": "^0.54.0",
     "@opentelemetry/core": "^2.2.0",

agents/src/voice/agent_session.ts

@@ -1,7 +1,9 @@
 // SPDX-FileCopyrightText: 2024 LiveKit, Inc.
 //
 // SPDX-License-Identifier: Apache-2.0
+import { type JsonObject, Struct } from '@bufbuild/protobuf';
 import { Mutex } from '@livekit/mutex';
+import { AgentSession as pb } from '@livekit/protocol';
 import type { AudioFrame, Room } from '@livekit/rtc-node';
 import { ThrowsPromise } from '@livekit/throws-transformer/throws';
 import type { TypedEventEmitter as TypedEmitter } from '@livekit/typed-emitter';
@@ -153,6 +155,7 @@ export type AgentSessionCallbacks = {
   [AgentSessionEventTypes.FunctionToolsExecuted]: (ev: FunctionToolsExecutedEvent) => void;
   [AgentSessionEventTypes.MetricsCollected]: (ev: MetricsCollectedEvent) => void;
   [AgentSessionEventTypes.SessionUsageUpdated]: (ev: SessionUsageUpdatedEvent) => void;
+  [AgentSessionEventTypes.DebugMessage]: (ev: pb.DebugMessage) => void;
   [AgentSessionEventTypes.SpeechCreated]: (ev: SpeechCreatedEvent) => void;
   [AgentSessionEventTypes.AgentFalseInterruption]: (ev: AgentFalseInterruptionEvent) => void;
   [AgentSessionEventTypes.Error]: (ev: ErrorEvent) => void;
@@ -740,6 +743,12 @@ export class AgentSession<
     return this.activity.interrupt(options);
   }
 
+  /** @internal — emit a debug/trace payload to the debugger/recorder. */
+  _emitDebugMessage(payload: JsonObject): void {
+    const debugMessage = new pb.DebugMessage({ payload: Struct.fromJson(payload) });
+    super.emit(AgentSessionEventTypes.DebugMessage, debugMessage);
+  }
+
   /**
    * The currently bound `AMD` instance, or `null` if AMD is not in use.
    * Mirrors python `AgentSession.amd`.

agents/src/voice/events.ts

@@ -30,6 +30,7 @@ export enum AgentSessionEventTypes {
   FunctionToolsExecuted = 'function_tools_executed',
   MetricsCollected = 'metrics_collected',
   SessionUsageUpdated = 'session_usage_updated',
+  DebugMessage = 'debug_message',
   SpeechCreated = 'speech_created',
   AgentFalseInterruption = 'agent_false_interruption',
   OverlappingSpeech = 'overlapping_speech',

agents/src/voice/remote_session.ts

@@ -63,6 +63,7 @@ export type RemoteSessionEventTypes =
   | 'overlapping_speech'
   | 'amd_prediction'
   | 'session_usage'
+  | 'debug_message'
   | 'error';
 
 /** @experimental */
@@ -75,6 +76,7 @@ export type RemoteSessionCallbacks = {
   overlapping_speech: (ev: pb.AgentSessionEvent_OverlappingSpeech) => void;
   amd_prediction: (ev: pb.AgentSessionEvent_AmdPrediction) => void;
   session_usage: (ev: pb.AgentSessionEvent_SessionUsageUpdated) => void;
+  debug_message: (ev: pb.DebugMessage) => void;
   error: (ev: pb.AgentSessionEvent_Error) => void;
 };
 
@@ -522,6 +524,7 @@ export class SessionHost {
       session.on(AgentSessionEventTypes.MetricsCollected, this.onMetricsCollected);
       session.on(AgentSessionEventTypes.OverlappingSpeech, this.onOverlappingSpeech);
       session.on(AgentSessionEventTypes.Error, this.onHostError);
+      session.on(AgentSessionEventTypes.DebugMessage, this.onDebugMessage);
     }
   }
 
@@ -550,6 +553,7 @@ export class SessionHost {
       this.session.off(AgentSessionEventTypes.MetricsCollected, this.onMetricsCollected);
       this.session.off(AgentSessionEventTypes.OverlappingSpeech, this.onOverlappingSpeech);
       this.session.off(AgentSessionEventTypes.Error, this.onHostError);
+      this.session.off(AgentSessionEventTypes.DebugMessage, this.onDebugMessage);
     }
 
     if (this.recvTask) {
@@ -713,6 +717,10 @@ export class SessionHost {
     );
   };
 
+  private onDebugMessage = (event: pb.DebugMessage): void => {
+    this.emitEvent({ case: 'debugMessage', value: event });
+  };
+
   /**
    * @internal — forwards an AMD prediction to the connected
    * {@link RemoteSession} peer. Mirrors python
@@ -1004,6 +1012,9 @@ export class RemoteSession extends (EventEmitter as new () => TypedEventEmitter<
       case 'sessionUsageUpdated':
         this.emit('session_usage', ev.value);
         break;
+      case 'debugMessage':
+        this.emit('debug_message', ev.value);
+        break;
       case 'error':
         this.emit('error', ev.value);
         break;