fix(openai): preserve realtime response rejection code

.changeset/fail-fast-openai-active-response.md

@@ -0,0 +1,6 @@
+---
+'@livekit/agents': minor
+'@livekit/agents-plugin-openai': patch
+---
+
+Fail OpenAI Realtime reply generation immediately with a typed provider error when a response is already active.

agents/src/llm/index.ts

@@ -72,6 +72,7 @@ export {
 } from './llm.js';
 
 export {
+  RealtimeError,
   RealtimeModel,
   RealtimeSession,
   type GenerationCreatedEvent,

agents/src/llm/realtime.ts

@@ -43,6 +43,19 @@ export interface RealtimeModelError {
   recoverable: boolean;
 }
 
+/** Error raised by a realtime session when a request fails. */
+export class RealtimeError extends Error {
+  /** Provider error code when the failure mirrors one. */
+  readonly code?: string;
+
+  constructor(message: string, { code }: { code?: string } = {}) {
+    super(message);
+    this.name = 'RealtimeError';
+    this.code = code;
+    Error.captureStackTrace(this, RealtimeError);
+  }
+}
+
 export interface RealtimeCapabilities {
   /** Whether generated assistant messages can be truncated after interruption. */
   messageTruncation: boolean;

agents/src/voice/agent_session.ts

@@ -1032,6 +1032,12 @@ export class AgentSession<
     }
   }
 
+  /**
+   * Generate an agent reply and return a handle to the speech.
+   *
+   * Awaiting the handle waits for the reply to finish and never throws; check
+   * `handle.exception()` for the failure instead.
+   */
   generateReply(options?: {
     userInput?: string | ChatMessage;
     chatCtx?: ChatContext;

plugins/openai/src/realtime/realtime_model.test.ts

@@ -13,8 +13,16 @@ import {
 
 type RealtimeSessionInternals = {
   generateReply: RealtimeSession['generateReply'];
+  handleError: (event: api_proto.ErrorEvent) => void;
+  on: (event: 'error', listener: (error: llm.RealtimeModelError) => void) => void;
   updateInstructions: RealtimeSession['updateInstructions'];
-  responseCreatedFutures: Record<string, unknown>;
+  responseCreatedFutures: Record<
+    string,
+    {
+      doneFut: Future<llm.GenerationCreatedEvent>;
+      timeout?: ReturnType<typeof setTimeout>;
+    }
+  >;
   sendEvent: ReturnType<typeof vi.fn>;
   textModeRecoveryRetries: number;
   instructions?: string;
@@ -37,6 +45,13 @@ type ResponseDoneSessionInternals = {
   };
 };
 
+type ErrorSessionInternals = {
+  generateReply: RealtimeSession['generateReply'];
+  handleError: (event: api_proto.ErrorEvent) => void;
+  on: (event: 'error', listener: (error: llm.RealtimeModelError) => void) => void;
+  responseCreatedFutures: RealtimeSessionInternals['responseCreatedFutures'];
+};
+
 function createSessionForTest(): RealtimeSessionInternals {
   const session = Object.create(RealtimeSession.prototype) as RealtimeSessionInternals;
   session.responseCreatedFutures = {};
@@ -55,6 +70,20 @@ function stubTaskRuntime(): void {
   } as unknown as Task<void>);
 }
 
+function activeResponseRejection(eventId: string): api_proto.ErrorEvent {
+  return {
+    type: 'error',
+    event_id: eventId,
+    error: {
+      message: 'Conversation already has an active response',
+      type: 'invalid_request_error',
+      code: 'conversation_already_has_active_response',
+      event_id: eventId,
+      param: '',
+    },
+  };
+}
+
 describe('RealtimeSession.generateReply', () => {
   it('preserves session instructions when generating with per-response instructions', async () => {
     const session = createSessionForTest();
@@ -103,6 +132,56 @@ describe('RealtimeSession.generateReply', () => {
   });
 });
 
+describe('RealtimeSession error handling', () => {
+  afterEach(() => {
+    vi.restoreAllMocks();
+  });
+
+  function createErrorSession(): ErrorSessionInternals {
+    stubTaskRuntime();
+    const model = new RealtimeModel({ apiKey: 'test-key' });
+    return model.session() as unknown as ErrorSessionInternals;
+  }
+
+  it('fails fast when an active response rejection matches the pending response', async () => {
+    const session = createErrorSession();
+    const errors: llm.RealtimeModelError[] = [];
+    session.on('error', (error) => errors.push(error));
+    const promise = session.generateReply();
+    const eventId = Object.keys(session.responseCreatedFutures)[0]!;
+    const handle = session.responseCreatedFutures[eventId]!;
+
+    session.handleError(activeResponseRejection(eventId));
+
+    expect(handle.doneFut.done).toBe(true);
+    await expect(promise).rejects.toBeInstanceOf(llm.RealtimeError);
+    await expect(promise).rejects.toMatchObject({
+      code: 'conversation_already_has_active_response',
+    });
+    expect(session.responseCreatedFutures[eventId]).toBeUndefined();
+    expect(errors[0]?.recoverable).toBe(true);
+  });
+
+  it('leaves pending responses untouched when an error has an unknown event ID', async () => {
+    const session = createErrorSession();
+    const errors: llm.RealtimeModelError[] = [];
+    session.on('error', (error) => errors.push(error));
+    const abortController = new AbortController();
+    const promise = session.generateReply(undefined, { signal: abortController.signal });
+    const eventId = Object.keys(session.responseCreatedFutures)[0]!;
+    const handle = session.responseCreatedFutures[eventId]!;
+
+    session.handleError(activeResponseRejection('response_create_other'));
+
+    expect(handle.doneFut.done).toBe(false);
+    expect(session.responseCreatedFutures[eventId]).toBe(handle);
+    expect(errors[0]?.recoverable).toBe(true);
+
+    abortController.abort();
+    await expect(promise).rejects.toThrow('generateReply aborted');
+  });
+});
+
 describe('RealtimeSession response.done status handling', () => {
   afterEach(() => {
     vi.restoreAllMocks();

plugins/openai/src/realtime/realtime_model.ts

@@ -1969,7 +1969,9 @@ export class RealtimeSession extends llm.RealtimeSession {
         delete this.responseCreatedFutures[eventId];
         if (handle.timeout) clearTimeout(handle.timeout);
         if (!handle.doneFut.done) {
-          handle.doneFut.reject(new Error(event.error.message));
+          handle.doneFut.reject(
+            new llm.RealtimeError(event.error.message, { code: event.error.code }),
+          );
         }
       }
     }