import { afterEach, beforeEach, describe, expect, it, Mock, Mocked, vi } from 'vitest';
import WebSocket from 'ws';
import { disableCache, enableCache } from '../../../src/cache';
import logger from '../../../src/logger';
import { OpenAiRealtimeProvider } from '../../../src/providers/openai/realtime';
import * as util from '../../../src/util/index';
import { mockProcessEnv } from '../../util/utils';
import { getOpenAiMissingApiKeyMessage, restoreEnvVar } from './shared';

import type { OpenAiRealtimeOptions } from '../../../src/providers/openai/realtime';

// Mock WebSocket
vi.mock('ws');
const MockWebSocket = WebSocket as Mocked<typeof WebSocket>;

// Mock logger
vi.mock('../../../src/logger', () => ({
  __esModule: true,
  default: {
    debug: vi.fn(),
    info: vi.fn(),
    warn: vi.fn(),
    error: vi.fn(),
  },
}));

/**
 * Flush enough microtasks to advance past the persistent-path serialization
 * queue. The provider chains: `await previousTurn` → `await openPersistentConnection`
 * → `setupMessageHandlers` (which itself awaits tool config) → message handler
 * is attached. Tests that need to invoke the attached handler synchronously
 * after callApi() must wait for all of those hops to settle.
 */
const MICROTASK_FLUSH_ITERATIONS = 8;

const flushMicrotasks = async () => {
  for (let i = 0; i < MICROTASK_FLUSH_ITERATIONS; i++) {
    await Promise.resolve();
  }
};

describe('OpenAI Realtime Provider', () => {
  let mockWs: any;
  let mockHandlers: { [key: string]: Function[] };
  const originalOpenAiApiKey = process.env.OPENAI_API_KEY;
  const originalCustomRealtimeApiKey = process.env.CUSTOM_REALTIME_API_KEY;
  const originalOpenAiApiBaseUrl = process.env.OPENAI_API_BASE_URL;
  const originalOpenAiBaseUrl = process.env.OPENAI_BASE_URL;

  const trackMockHandler = (event: string, handler: Function) => {
    (mockHandlers[event] ??= []).push(handler);
  };

  const removeMockHandler = (event: string, handler: Function) => {
    const arr = mockHandlers[event];
    if (!arr) {
      return;
    }
    const idx = arr.indexOf(handler);
    if (idx !== -1) {
      arr.splice(idx, 1);
    }
  };

  const createPersistentMockWebSocket = (owner?: OpenAiRealtimeProvider) =>
    ({
      on: vi.fn((event: string, handler: Function) => {
        trackMockHandler(event, handler);
        return owner?.persistentConnection;
      }),
      once: vi.fn((event: string, handler: Function) => {
        trackMockHandler(event, handler);
        return owner?.persistentConnection;
      }),
      send: vi.fn(),
      close: vi.fn(),
      removeListener: vi.fn(removeMockHandler),
      readyState: WebSocket.OPEN,
    }) as unknown as WebSocket;

  const lastMessageHandler = () => {
    const handler = mockHandlers.message[mockHandlers.message.length - 1];
    if (typeof handler !== 'function') {
      throw new Error('expected message handler attached');
    }
    return handler;
  };

  const emitRealtimeEvent = (handler: Function, event: Record<string, unknown>) =>
    handler(Buffer.from(JSON.stringify(event)));

  const emitUserItem = (handler: Function, id = 'u1') =>
    emitRealtimeEvent(handler, { type: 'conversation.item.added', item: { id, role: 'user' } });

  const emitFunctionCall = (handler: Function, name: string, callId = 'c1') =>
    emitRealtimeEvent(handler, {
      type: 'response.output_item.added',
      item: { type: 'function_call', call_id: callId, name, arguments: '{}' },
    });

  const emitFunctionCallArgumentsDone = (handler: Function, callId = 'c1') =>
    emitRealtimeEvent(handler, {
      type: 'response.function_call_arguments.done',
      call_id: callId,
      arguments: '{}',
    });

  const emitOutputTextDone = (handler: Function, text: string) =>
    emitRealtimeEvent(handler, { type: 'response.output_text.done', text });

  const emitResponseDone = (
    handler: Function,
    usage = { total_tokens: 1, input_tokens: 1, output_tokens: 0 },
  ) => emitRealtimeEvent(handler, { type: 'response.done', response: { usage } });

  const sentWebSocketEvents = (connection: WebSocket) =>
    vi.mocked(connection.send as Mock).mock.calls.map(([raw]) => JSON.parse(raw as string));

  const createDeferredString = () => {
    let resolveDeferred: ((value: string) => void) | undefined;
    const promise = new Promise<string>((resolve) => {
      resolveDeferred = resolve;
    });

    const resolve = (value: string) => {
      if (!resolveDeferred) {
        throw new Error('expected deferred string to be initialized');
      }
      resolveDeferred(value);
    };

    return { promise, resolve };
  };

  beforeEach(() => {
    vi.resetAllMocks();
    disableCache();
    mockProcessEnv({ OPENAI_API_KEY: 'test-api-key' });
    mockProcessEnv({ OPENAI_API_BASE_URL: undefined });
    mockProcessEnv({ OPENAI_BASE_URL: undefined });
    mockHandlers = {
      open: [],
      message: [],
      error: [],
      close: [],
    };

    // Create a mock WebSocket instance
    mockWs = {
      on: vi.fn((event: string, handler: Function) => {
        trackMockHandler(event, handler);
      }),
      send: vi.fn(),
      close: vi.fn(),
      once: vi.fn((event: string, handler: Function) => {
        trackMockHandler(event, handler);
      }),
      removeListener: vi.fn(removeMockHandler),
    };

    // Mock WebSocket constructor
    (MockWebSocket as any).mockImplementation(function () {
      return mockWs;
    });
  });

  afterEach(() => {
    enableCache();
    restoreEnvVar('OPENAI_API_KEY', originalOpenAiApiKey);
    restoreEnvVar('CUSTOM_REALTIME_API_KEY', originalCustomRealtimeApiKey);
    restoreEnvVar('OPENAI_API_BASE_URL', originalOpenAiApiBaseUrl);
    restoreEnvVar('OPENAI_BASE_URL', originalOpenAiBaseUrl);
  });

  describe('Basic Functionality', () => {
    it('should initialize with correct model and config', () => {
      const config = {
        modalities: ['text'],
        instructions: 'Test instructions',
        voice: 'alloy' as const,
      };

      const provider = new OpenAiRealtimeProvider('gpt-4o-realtime-preview', { config });

      expect(provider.modelName).toBe('gpt-4o-realtime-preview');
      expect(provider.config).toEqual(expect.objectContaining(config));
      expect(provider.config.maintainContext).toBe(true); // Default value
    });

    it('should initialize with gpt-realtime model and new voices', () => {
      const config = {
        modalities: ['text', 'audio'],
        instructions: 'Test instructions',
        voice: 'cedar' as const, // New voice for gpt-realtime
      };

      const provider = new OpenAiRealtimeProvider('gpt-realtime', { config });

      expect(provider.modelName).toBe('gpt-realtime');
      expect(provider.config).toEqual(expect.objectContaining(config));
      expect(provider.config.maintainContext).toBe(true); // Default value
    });

    it('should support marin voice for gpt-realtime model', () => {
      const config = {
        modalities: ['text', 'audio'],
        instructions: 'Test instructions',
        voice: 'marin' as const, // New voice for gpt-realtime
      };

      const provider = new OpenAiRealtimeProvider('gpt-realtime', { config });

      expect(provider.modelName).toBe('gpt-realtime');
      expect(provider.config.voice).toBe('marin');
    });

    it('should log warning for unknown model', () => {
      new OpenAiRealtimeProvider('unknown-model');
      expect(logger.debug).toHaveBeenCalledWith(
        'Using unknown OpenAI realtime model: unknown-model',
      );
    });

    it('rejects gpt-realtime-translate as a conversational realtime provider', () => {
      expect(() => new OpenAiRealtimeProvider('gpt-realtime-translate')).toThrow(
        /not a conversational Realtime model/,
      );
    });

    it('rejects gpt-realtime-whisper used as a standalone realtime provider', () => {
      expect(() => new OpenAiRealtimeProvider('gpt-realtime-whisper')).toThrow(
        /input_audio_transcription\.model/,
      );
    });

    it('returns ProviderResponse.error instead of a placeholder when the API yields nothing', async () => {
      const provider = new OpenAiRealtimeProvider('gpt-4o-realtime-preview', {
        config: { modalities: ['text'], maintainContext: false },
      });

      const responsePromise = provider.directWebSocketRequest('hi');
      // Allow ws constructor + handler registration to settle.
      await flushMicrotasks();
      mockHandlers.open.forEach((h) => h());
      await flushMicrotasks();

      const handler = mockHandlers.message[mockHandlers.message.length - 1];
      handler(
        Buffer.from(
          JSON.stringify({
            type: 'conversation.item.created',
            item: { id: 'u1', role: 'user' },
          }),
        ),
      );
      // No deltas, no audio, no function call — just response.done.
      handler(
        Buffer.from(
          JSON.stringify({
            type: 'response.done',
            response: { usage: { total_tokens: 1, input_tokens: 1, output_tokens: 0 } },
          }),
        ),
      );

      const inner = await responsePromise;
      expect(inner.output).toBe('');

      // Now drive a full callApi flow to assert the outer error wrapping.
      // Reset mocks for a clean second call.
      mockHandlers.open = [];
      mockHandlers.message = [];

      const callApiPromise = provider.callApi('hi');
      await flushMicrotasks();
      mockHandlers.open.forEach((h) => h());
      await flushMicrotasks();
      const handler2 = mockHandlers.message[mockHandlers.message.length - 1];
      handler2(
        Buffer.from(
          JSON.stringify({
            type: 'conversation.item.created',
            item: { id: 'u2', role: 'user' },
          }),
        ),
      );
      handler2(
        Buffer.from(
          JSON.stringify({
            type: 'response.done',
            response: { usage: { total_tokens: 1, input_tokens: 1, output_tokens: 0 } },
          }),
        ),
      );

      const outer = await callApiPromise;
      expect(outer.error).toMatch(/empty response/i);
      expect(outer.output).toBeUndefined();
    });

    it('should use default missing API key error message', async () => {
      mockProcessEnv({ OPENAI_API_KEY: undefined });
      const provider = new OpenAiRealtimeProvider('gpt-4o-realtime-preview', {
        env: {
          OPENAI_API_KEY: undefined,
        },
      });

      await expect(provider.callApi('Hello')).rejects.toThrow(getOpenAiMissingApiKeyMessage());
    });

    it('should use custom apiKeyEnvar in missing API key errors', async () => {
      mockProcessEnv({ OPENAI_API_KEY: undefined });
      mockProcessEnv({ CUSTOM_REALTIME_API_KEY: undefined });
      const provider = new OpenAiRealtimeProvider('gpt-4o-realtime-preview', {
        config: {
          apiKeyEnvar: 'CUSTOM_REALTIME_API_KEY',
        },
        env: {
          OPENAI_API_KEY: undefined,
          CUSTOM_REALTIME_API_KEY: undefined,
        },
      });

      await expect(provider.callApi('Hello')).rejects.toThrow(
        getOpenAiMissingApiKeyMessage('CUSTOM_REALTIME_API_KEY'),
      );
    });

    it('should generate valid session body', async () => {
      const config = {
        modalities: ['text'],
        voice: 'echo' as const,
        instructions: 'Test instructions',
        temperature: 0.7,
        max_response_output_tokens: 100,
      };

      const provider = new OpenAiRealtimeProvider('gpt-4o-realtime-preview', { config });
      const body = await provider.getRealtimeSessionBody();

      expect(body).toEqual({
        type: 'realtime',
        model: 'gpt-4o-realtime-preview',
        output_modalities: ['text'],
        instructions: 'Test instructions',
        audio: {
          input: {
            format: {
              type: 'audio/pcm',
              rate: 24000,
            },
          },
          output: {
            format: {
              type: 'audio/pcm',
              rate: 24000,
            },
            voice: 'echo',
          },
        },
        max_output_tokens: 100,
      });
    });

    it('should handle audio configuration', async () => {
      const config: OpenAiRealtimeOptions = {
        modalities: ['text', 'audio'],
        voice: 'alloy' as const,
        instructions: 'Test instructions',
        input_audio_format: 'pcm16' as const,
        output_audio_format: 'pcm16' as const,
        input_audio_transcription: {
          model: 'whisper-1',
          language: 'en',
          prompt: 'Transcribe the following audio',
        },
        temperature: 0.8,
        max_response_output_tokens: 'inf' as const,
      };

      const provider = new OpenAiRealtimeProvider('gpt-4o-realtime-preview', { config });
      const body = await provider.getRealtimeSessionBody();

      expect(body).toEqual({
        type: 'realtime',
        model: 'gpt-4o-realtime-preview',
        output_modalities: ['audio'],
        instructions: 'Test instructions',
        audio: {
          input: {
            format: {
              type: 'audio/pcm',
              rate: 24000,
            },
            transcription: {
              model: 'whisper-1',
              language: 'en',
              prompt: 'Transcribe the following audio',
            },
          },
          output: {
            format: {
              type: 'audio/pcm',
              rate: 24000,
            },
            voice: 'alloy',
          },
        },
        max_output_tokens: 'inf',
      });
    });

    it('should pass through realtime 2 reasoning options and realtime whisper transcription config', async () => {
      const provider = new OpenAiRealtimeProvider('gpt-realtime-2', {
        config: {
          input_audio_transcription: {
            model: 'gpt-realtime-whisper',
            language: 'en',
            delay: 'low',
          },
          parallel_tool_calls: true,
          reasoning: {
            effort: 'high',
          },
        },
      });

      const body = await provider.getRealtimeSessionBody();

      expect(body).toMatchObject({
        type: 'realtime',
        model: 'gpt-realtime-2',
        audio: {
          input: {
            transcription: {
              model: 'gpt-realtime-whisper',
              language: 'en',
              delay: 'low',
            },
          },
        },
        parallel_tool_calls: true,
        reasoning: {
          effort: 'high',
        },
      });
    });

    it('should pass through semantic VAD configuration', async () => {
      const provider = new OpenAiRealtimeProvider('gpt-realtime-2', {
        config: {
          turn_detection: {
            type: 'semantic_vad',
            eagerness: 'high',
            create_response: true,
            interrupt_response: false,
          },
        },
      });

      const body = await provider.getRealtimeSessionBody();

      expect(body).toMatchObject({
        audio: {
          input: {
            turn_detection: {
              type: 'semantic_vad',
              eagerness: 'high',
              create_response: true,
              interrupt_response: false,
            },
          },
        },
      });
    });

    it('should preserve native Realtime tools and tool choices in session payloads', async () => {
      const provider = new OpenAiRealtimeProvider('gpt-realtime', {
        config: {
          tools: [
            {
              type: 'function',
              name: 'get_weather',
              description: 'Get the weather',
              parameters: {
                type: 'object',
                properties: {
                  location: { type: 'string' },
                },
              },
            },
          ],
          tool_choice: {
            type: 'function',
            name: 'get_weather',
          },
        },
      });

      const body = await provider.getRealtimeSessionBody();

      expect(body.tools).toEqual([
        {
          type: 'function',
          name: 'get_weather',
          description: 'Get the weather',
          parameters: {
            type: 'object',
            properties: {
              location: { type: 'string' },
            },
          },
        },
      ]);
      expect(body.tool_choice).toEqual({
        type: 'function',
        name: 'get_weather',
      });
    });

    it('should normalize chat-style tools for Realtime session payloads', async () => {
      const provider = new OpenAiRealtimeProvider('gpt-realtime', {
        config: {
          tools: [
            {
              type: 'function',
              function: {
                name: 'get_weather',
                description: 'Get the weather',
                parameters: {
                  type: 'object',
                  properties: {
                    location: { type: 'string' },
                  },
                },
              },
            },
          ],
          tool_choice: 'auto',
        },
      });

      const body = await provider.getRealtimeSessionBody();

      expect(body.tools).toEqual([
        {
          type: 'function',
          name: 'get_weather',
          description: 'Get the weather',
          parameters: {
            type: 'object',
            properties: {
              location: { type: 'string' },
            },
          },
        },
      ]);
      expect(body.tool_choice).toBe('auto');
    });

    it('should normalize chat-style forced tool choices to the native Realtime shape', async () => {
      const provider = new OpenAiRealtimeProvider('gpt-realtime', {
        config: {
          tools: [
            {
              type: 'function',
              function: {
                name: 'get_weather',
                description: 'Get the weather',
                parameters: { type: 'object', properties: {} },
              },
            },
            {
              type: 'function',
              function: {
                name: 'get_time',
                description: 'Get the time',
                parameters: { type: 'object', properties: {} },
              },
            },
          ],
          tool_choice: {
            type: 'function',
            function: { name: 'get_weather' },
          },
        },
      });

      const body = await provider.getRealtimeSessionBody();

      expect(body.tools).toEqual([
        {
          type: 'function',
          name: 'get_weather',
          description: 'Get the weather',
          parameters: { type: 'object', properties: {} },
        },
        {
          type: 'function',
          name: 'get_time',
          description: 'Get the time',
          parameters: { type: 'object', properties: {} },
        },
      ]);
      expect(body.tool_choice).toEqual({
        type: 'function',
        name: 'get_weather',
      });
    });

    it.each([
      [0, 'inf'],
      [-1, 'inf'],
      [1.5, 'inf'],
      [4097, 'inf'],
      ['0', 'inf'],
      ['1', 'inf'],
      ['inf', 'inf'],
      [1, 1],
      [4096, 4096],
    ] as const)('normalizes max_response_output_tokens=%s to %s in the session body', async (maxResponseOutputTokens, expected) => {
      const provider = new OpenAiRealtimeProvider('gpt-4o-realtime-preview', {
        config: {
          modalities: ['text'],
          max_response_output_tokens: maxResponseOutputTokens as any,
        },
      });

      const body = await provider.getRealtimeSessionBody();

      expect(body.max_output_tokens).toBe(expected);
    });

    it('should handle basic text response with persistent connection', async () => {
      const config = {
        modalities: ['text'],
        instructions: 'Test instructions',
        maintainContext: true,
      };

      const provider = new OpenAiRealtimeProvider('gpt-4o-realtime-preview', { config });

      // Create mock WebSocket connection with proper type
      provider.persistentConnection = {
        on: vi.fn((event: string, handler: Function) => {
          mockHandlers[event].push(handler);
          return provider.persistentConnection;
        }),
        once: vi.fn((event: string, handler: Function) => {
          mockHandlers[event].push(handler);
          return provider.persistentConnection;
        }),
        send: vi.fn(),
        close: vi.fn(),
        removeListener: vi.fn(),
      } as unknown as WebSocket;

      // Create context with conversationId to ensure maintainContext stays true
      const context = {
        test: {
          metadata: { conversationId: 'test-conv-123' },
        },
      } as any;

      // Create a promise for the API call
      const responsePromise = provider.callApi('Hello', context);

      // Wait for microtask to process so handler is registered
      await flushMicrotasks();

      // Get the message handler
      const messageHandlers = mockHandlers.message;
      const lastHandler = messageHandlers[messageHandlers.length - 1];

      // Simulate conversation item created
      lastHandler(
        Buffer.from(
          JSON.stringify({
            type: 'conversation.item.created',
            item: { id: 'msg_123', role: 'user' },
          }),
        ),
      );

      // Simulate response created
      lastHandler(
        Buffer.from(
          JSON.stringify({
            type: 'response.created',
            response: { id: 'resp_123' },
          }),
        ),
      );

      // Simulate text delta
      lastHandler(
        Buffer.from(
          JSON.stringify({
            type: 'response.text.delta',
            delta: 'Hello',
          }),
        ),
      );

      // Simulate text done
      lastHandler(
        Buffer.from(
          JSON.stringify({
            type: 'response.text.done',
            text: 'Hello',
          }),
        ),
      );

      // Simulate response done
      lastHandler(
        Buffer.from(
          JSON.stringify({
            type: 'response.done',
            response: {
              usage: {
                total_tokens: 10,
                input_tokens: 5,
                output_tokens: 5,
              },
            },
          }),
        ),
      );

      const response = await responsePromise;

      // Verify the response
      expect(response.output).toBe('Hello');
      expect(response.metadata?.responseId).toBe('resp_123');
      expect(response.metadata?.messageId).toBe('msg_123');

      // Verify that the connection was not closed (persistent)
      expect(provider.persistentConnection?.close).not.toHaveBeenCalled();

      // Verify that the connection is maintained
      expect(provider.persistentConnection).not.toBeNull();
    });

    it('should maintain conversation context across multiple messages', async () => {
      const config = {
        modalities: ['text'],
        instructions: 'Test instructions',
        maintainContext: true,
      };

      const provider = new OpenAiRealtimeProvider('gpt-4o-realtime-preview', { config });

      // Create mock WebSocket connection with proper type
      provider.persistentConnection = {
        on: vi.fn((event: string, handler: Function) => {
          mockHandlers[event].push(handler);
          return provider.persistentConnection;
        }),
        once: vi.fn((event: string, handler: Function) => {
          mockHandlers[event].push(handler);
          return provider.persistentConnection;
        }),
        send: vi.fn(),
        close: vi.fn(),
        removeListener: vi.fn(),
      } as unknown as WebSocket;

      // Helper function to simulate message sequence
      const simulateMessageSequence = async (
        messageId: string,
        assistantId: string,
        responseId: string,
        responseText: string,
      ) => {
        const messageHandlers = mockHandlers.message;
        const lastHandler = messageHandlers[messageHandlers.length - 1];

        // User message
        await Promise.resolve(
          lastHandler(
            Buffer.from(
              JSON.stringify({
                type: 'conversation.item.created',
                item: { id: messageId, role: 'user' },
              }),
            ),
          ),
        );

        // Assistant message
        await Promise.resolve(
          lastHandler(
            Buffer.from(
              JSON.stringify({
                type: 'conversation.item.created',
                item: { id: assistantId, role: 'assistant' },
              }),
            ),
          ),
        );

        // Manually set the previousItemId since the mock doesn't properly handle this
        provider.previousItemId = assistantId;
        if (!provider.assistantMessageIds.includes(assistantId)) {
          provider.assistantMessageIds.push(assistantId);
        }

        // Response created
        await Promise.resolve(
          lastHandler(
            Buffer.from(
              JSON.stringify({
                type: 'response.created',
                response: { id: responseId },
              }),
            ),
          ),
        );

        // Text delta
        await Promise.resolve(
          lastHandler(
            Buffer.from(
              JSON.stringify({
                type: 'response.text.delta',
                delta: responseText,
              }),
            ),
          ),
        );

        // Text done
        await Promise.resolve(
          lastHandler(
            Buffer.from(
              JSON.stringify({
                type: 'response.text.done',
                text: responseText,
              }),
            ),
          ),
        );

        // Response done
        await Promise.resolve(
          lastHandler(
            Buffer.from(
              JSON.stringify({
                type: 'response.done',
                response: {
                  usage: {
                    total_tokens: 24,
                    prompt_tokens: 12,
                    completion_tokens: 12,
                  },
                },
              }),
            ),
          ),
        );
      };

      // Create context with conversationId to ensure maintainContext stays true
      const context = {
        test: {
          metadata: { conversationId: 'test-conv-multi' },
        },
      } as any;

      // First message
      const firstResponsePromise = provider.callApi('First message', context);
      // Wait for microtask to process so handler is registered
      await flushMicrotasks();
      await simulateMessageSequence('msg_1', 'assistant_1', 'resp_1', 'First response');
      const firstResponse = await firstResponsePromise;

      // Verify first response
      expect(firstResponse.output).toBe('First response');
      expect(provider.previousItemId).toBe('assistant_1');
      expect(provider.assistantMessageIds).toContain('assistant_1');

      // Second message
      const secondResponsePromise = provider.callApi('Second message', context);

      // Wait for microtask to process so handler is registered
      await flushMicrotasks();

      // Skip the WebSocket send assertion as it's not reliable in the test

      await simulateMessageSequence('msg_2', 'assistant_2', 'resp_2', 'Second response');
      const secondResponse = await secondResponsePromise;

      // Verify second response
      expect(secondResponse.output).toBe('Second response');
      expect(provider.previousItemId).toBe('assistant_2');
      expect(provider.assistantMessageIds).toContain('assistant_2');
      expect(provider.assistantMessageIds).toHaveLength(2);

      // Verify connection state
      expect(provider.persistentConnection?.close).not.toHaveBeenCalled();
      expect(provider.persistentConnection).not.toBeNull();
    });

    it('should handle WebSocket errors in persistent connection', async () => {
      const config = {
        modalities: ['text'],
        maintainContext: true,
      };

      const provider = new OpenAiRealtimeProvider('gpt-4o-realtime-preview', { config });

      // Create mock WebSocket connection with proper type
      provider.persistentConnection = {
        on: vi.fn((event: string, handler: Function) => {
          mockHandlers[event].push(handler);
          return provider.persistentConnection;
        }),
        once: vi.fn((event: string, handler: Function) => {
          mockHandlers[event].push(handler);
          return provider.persistentConnection;
        }),
        send: vi.fn(),
        close: vi.fn(),
        removeListener: vi.fn(),
      } as unknown as WebSocket;

      // Create context with conversationId to ensure maintainContext stays true
      const context = {
        test: {
          metadata: { conversationId: 'test-conv-error' },
        },
      } as any;

      const responsePromise = provider.callApi('Hello', context);

      // Wait for microtask to process so handler is registered
      await flushMicrotasks();

      // Get the error handler and simulate a WebSocket error
      const errorHandlers = mockHandlers.error;
      const lastErrorHandler = errorHandlers[errorHandlers.length - 1];
      lastErrorHandler(new Error('Connection failed'));

      // Manually set the persistentConnection to null as the mock doesn't do this
      provider.persistentConnection = null;

      const response = await responsePromise;
      expect(response.error).toBe('WebSocket error: Error: Connection failed');
      expect(response.metadata).toEqual({});
      expect(provider.persistentConnection).toBeNull();
    });

    it('should handle audio response in persistent connection', async () => {
      const config = {
        modalities: ['text', 'audio'],
        maintainContext: true,
        voice: 'alloy' as const,
      };

      const provider = new OpenAiRealtimeProvider('gpt-4o-realtime-preview', { config });

      // Create mock WebSocket connection with proper type
      provider.persistentConnection = {
        on: vi.fn((event: string, handler: Function) => {
          mockHandlers[event].push(handler);
          return provider.persistentConnection;
        }),
        once: vi.fn((event: string, handler: Function) => {
          mockHandlers[event].push(handler);
          return provider.persistentConnection;
        }),
        send: vi.fn(),
        close: vi.fn(),
        removeListener: vi.fn(),
      } as unknown as WebSocket;

      // Create context with conversationId to ensure maintainContext stays true
      const context = {
        test: {
          metadata: { conversationId: 'test-conv-audio' },
        },
      } as any;

      const responsePromise = provider.callApi('Hello', context);

      // Wait for microtask to process so handler is registered
      await flushMicrotasks();

      // Get the message handler
      const messageHandlers = mockHandlers.message;
      const lastHandler = messageHandlers[messageHandlers.length - 1];

      // Simulate conversation item created
      lastHandler(
        Buffer.from(
          JSON.stringify({
            type: 'conversation.item.created',
            item: { id: 'msg_1', role: 'user' },
          }),
        ),
      );

      // Simulate response created
      lastHandler(
        Buffer.from(
          JSON.stringify({
            type: 'response.created',
            response: { id: 'resp_1' },
          }),
        ),
      );

      // Simulate audio response
      const audioData = Buffer.from('fake_audio_data');
      lastHandler(
        Buffer.from(
          JSON.stringify({
            type: 'response.audio.delta',
            item_id: 'audio_1',
            audio: audioData.toString('base64'),
          }),
        ),
      );

      // Simulate audio done
      lastHandler(
        Buffer.from(
          JSON.stringify({
            type: 'response.audio.done',
            format: 'wav',
            item_id: 'audio_1',
          }),
        ),
      );

      // Simulate text response
      lastHandler(
        Buffer.from(
          JSON.stringify({
            type: 'response.text.delta',
            delta: 'Hello there',
          }),
        ),
      );

      // Simulate text done
      lastHandler(
        Buffer.from(
          JSON.stringify({
            type: 'response.text.done',
            text: 'Hello there',
          }),
        ),
      );

      // Simulate response done
      lastHandler(
        Buffer.from(
          JSON.stringify({
            type: 'response.done',
            response: {
              usage: {
                total_tokens: 10,
                prompt_tokens: 5,
                completion_tokens: 5,
              },
            },
          }),
        ),
      );

      const response = await responsePromise;

      // Verify text response
      expect(response.output).toBe('Hello there');

      // First verify audio exists
      expect(response.audio).toBeDefined();
      expect(response.metadata).toBeDefined();
      expect(response.metadata!.audio).toBeDefined();

      // Then verify audio properties
      expect(response.audio!.format).toBe('wav');
      // The audio data should be converted from PCM16 to WAV, so it will be different from the original
      expect(response.audio!.data).toBeDefined();
      expect(response.audio!.data!.length).toBeGreaterThanOrEqual(
        audioData.toString('base64').length,
      ); // WAV has headers
      expect(response.audio!.transcript).toBe('Hello there');

      // Verify metadata
      expect(response.metadata!.audio!.format).toBe('wav');
      expect(response.metadata!.audio!.data).toBe(response.audio!.data); // Should match the audio data
    });

    it('should configure tools and handle function calls in persistent connections', async () => {
      const functionCallHandler = vi.fn().mockResolvedValue('{"call_status":"callback"}');
      const provider = new OpenAiRealtimeProvider('gpt-realtime', {
        config: {
          modalities: ['text'],
          maintainContext: true,
          tools: [
            {
              type: 'function',
              function: {
                name: 'end_of_dialog_tool',
                description: 'End the dialog',
                parameters: {
                  type: 'object',
                  properties: {
                    call_status: { type: 'string' },
                  },
                },
              },
            },
          ],
          tool_choice: 'auto',
          functionCallHandler,
        },
      });

      provider.persistentConnection = {
        on: vi.fn((event: string, handler: Function) => {
          mockHandlers[event].push(handler);
          return provider.persistentConnection;
        }),
        once: vi.fn((event: string, handler: Function) => {
          mockHandlers[event].push(handler);
          return provider.persistentConnection;
        }),
        send: vi.fn(),
        close: vi.fn(),
        removeListener: vi.fn(),
      } as unknown as WebSocket;

      const responsePromise = provider.callApi('Can you call me back later?', {
        test: {
          metadata: { conversationId: 'loan-application-flow' },
        },
      } as any);

      await vi.waitFor(() => {
        expect(provider.persistentConnection?.send).toHaveBeenCalled();
      });

      const sentEvents = vi
        .mocked(provider.persistentConnection.send as Mock)
        .mock.calls.map(([payload]) => JSON.parse(payload as string));

      expect(sentEvents[0]).toMatchObject({
        type: 'session.update',
        session: {
          tools: [
            {
              type: 'function',
              name: 'end_of_dialog_tool',
            },
          ],
          tool_choice: 'auto',
        },
      });

      const lastHandler = mockHandlers.message[mockHandlers.message.length - 1];
      await Promise.resolve(
        lastHandler(
          Buffer.from(
            JSON.stringify({
              type: 'conversation.item.created',
              item: { id: 'msg_1', role: 'user' },
            }),
          ),
        ),
      );

      const responseCreate = vi
        .mocked(provider.persistentConnection.send as Mock)
        .mock.calls.map(([payload]) => JSON.parse(payload as string))
        .find((event) => event.type === 'response.create' && event.response);

      expect(responseCreate.response).toMatchObject({
        tools: [
          {
            type: 'function',
            name: 'end_of_dialog_tool',
          },
        ],
        tool_choice: 'auto',
      });

      await Promise.resolve(
        lastHandler(
          Buffer.from(
            JSON.stringify({
              type: 'response.text.done',
              text: 'Let me check that.',
            }),
          ),
        ),
      );
      await Promise.resolve(
        lastHandler(
          Buffer.from(
            JSON.stringify({
              type: 'response.output_item.added',
              item: {
                type: 'function_call',
                call_id: 'call_1',
                name: 'end_of_dialog_tool',
                arguments: '{}',
              },
            }),
          ),
        ),
      );
      await Promise.resolve(
        lastHandler(
          Buffer.from(
            JSON.stringify({
              type: 'response.function_call_arguments.done',
              call_id: 'call_1',
              arguments: '{"call_status":"callback"}',
            }),
          ),
        ),
      );
      await Promise.resolve(
        lastHandler(
          Buffer.from(
            JSON.stringify({
              type: 'response.done',
              response: {
                usage: {
                  total_tokens: 8,
                  input_tokens: 5,
                  output_tokens: 3,
                },
              },
            }),
          ),
        ),
      );

      expect(functionCallHandler).toHaveBeenCalledWith(
        'end_of_dialog_tool',
        '{"call_status":"callback"}',
      );

      const followupEvents = vi
        .mocked(provider.persistentConnection.send as Mock)
        .mock.calls.map(([payload]) => JSON.parse(payload as string));

      expect(followupEvents).toEqual(
        expect.arrayContaining([
          expect.objectContaining({
            type: 'conversation.item.create',
            item: {
              type: 'function_call_output',
              call_id: 'call_1',
              output: '{"call_status":"callback"}',
            },
          }),
          expect.objectContaining({
            type: 'response.create',
          }),
        ]),
      );

      await Promise.resolve(
        lastHandler(
          Buffer.from(
            JSON.stringify({
              type: 'response.text.done',
              text: 'We will call you back later.',
            }),
          ),
        ),
      );
      await Promise.resolve(
        lastHandler(
          Buffer.from(
            JSON.stringify({
              type: 'response.done',
              response: {
                usage: {
                  total_tokens: 11,
                  input_tokens: 7,
                  output_tokens: 4,
                },
              },
            }),
          ),
        ),
      );

      const response = await responsePromise;

      expect(response.output).toContain('We will call you back later.');
      expect(response.output).not.toContain('Let me check that.');
      expect(response.metadata?.functionCallOccurred).toBe(true);
      expect(response.metadata?.functionCallResults).toEqual(['{"call_status":"callback"}']);
      expect(response.tokenUsage).toEqual({
        total: 11,
        prompt: 7,
        completion: 4,
        cached: 0,
        numRequests: 1,
      });
    });

    it('redacts handler errors before sending function_call_output to the model', async () => {
      const sensitive = new Error('connect ECONNREFUSED /var/run/secret.sock 127.0.0.1:5432');
      const functionCallHandler = vi.fn().mockRejectedValue(sensitive);
      const provider = new OpenAiRealtimeProvider('gpt-realtime', {
        config: {
          modalities: ['text'],
          maintainContext: true,
          tools: [
            {
              type: 'function',
              name: 'lookup',
              parameters: { type: 'object', properties: {} },
            },
          ],
          tool_choice: 'auto',
          functionCallHandler,
        },
      });

      const persistentConnection = {
        on: vi.fn((event: string, h: Function) => {
          mockHandlers[event].push(h);
        }),
        once: vi.fn((event: string, h: Function) => {
          mockHandlers[event].push(h);
        }),
        send: vi.fn(),
        close: vi.fn(),
        removeListener: vi.fn(),
        readyState: 1,
      } as unknown as WebSocket;
      provider.persistentConnection = persistentConnection;

      const responsePromise = provider.callApi('go', {
        test: { metadata: { conversationId: 'redact' } },
      } as any);

      await flushMicrotasks();
      const handler = mockHandlers.message[mockHandlers.message.length - 1];

      handler(
        Buffer.from(
          JSON.stringify({
            type: 'conversation.item.added',
            item: { id: 'u1', role: 'user' },
          }),
        ),
      );
      handler(
        Buffer.from(
          JSON.stringify({
            type: 'response.output_item.added',
            item: { type: 'function_call', call_id: 'c1', name: 'lookup', arguments: '{}' },
          }),
        ),
      );
      handler(
        Buffer.from(
          JSON.stringify({
            type: 'response.function_call_arguments.done',
            call_id: 'c1',
            arguments: '{}',
          }),
        ),
      );
      handler(
        Buffer.from(
          JSON.stringify({
            type: 'response.done',
            response: { usage: { total_tokens: 1, input_tokens: 1, output_tokens: 0 } },
          }),
        ),
      );

      // Let the for-await-loop in response.done resolve.
      await flushMicrotasks();

      const sentPayloads = vi
        .mocked(provider.persistentConnection!.send as Mock)
        .mock.calls.map(([raw]) => JSON.parse(raw as string));
      const toolOutput = sentPayloads.find(
        (e) => e.type === 'conversation.item.create' && e.item?.type === 'function_call_output',
      );
      expect(toolOutput).toBeDefined();
      // Generic redacted payload — no leaked path or error message.
      expect(toolOutput.item.output).toBe('{"error":"Tool execution failed"}');
      expect(toolOutput.item.output).not.toContain('ECONNREFUSED');
      expect(toolOutput.item.output).not.toContain('/var/run/secret.sock');

      // Resolve the follow-up response so the test promise settles cleanly.
      handler(Buffer.from(JSON.stringify({ type: 'response.output_text.delta', delta: 'tried' })));
      handler(Buffer.from(JSON.stringify({ type: 'response.output_text.done', text: 'tried' })));
      handler(
        Buffer.from(
          JSON.stringify({
            type: 'response.done',
            response: { usage: { total_tokens: 2, input_tokens: 1, output_tokens: 1 } },
          }),
        ),
      );

      await responsePromise;
      provider.cleanup();
    });

    it('keeps lifecycle listeners on idle persistent sockets', async () => {
      const provider = new OpenAiRealtimeProvider('gpt-realtime', {
        config: { modalities: ['text'], maintainContext: true },
      });

      const persistentConnection = {
        on: vi.fn((event: string, h: Function) => mockHandlers[event].push(h)),
        once: vi.fn((event: string, h: Function) => mockHandlers[event].push(h)),
        send: vi.fn(),
        close: vi.fn(),
        removeListener: vi.fn((event: string, h: Function) => {
          const handlers = mockHandlers[event];
          const idx = handlers.indexOf(h);
          if (idx !== -1) {
            handlers.splice(idx, 1);
          }
        }),
        readyState: WebSocket.OPEN,
      } as unknown as WebSocket;
      provider.persistentConnection = persistentConnection;

      const responsePromise = provider.callApi('hi', {
        test: { metadata: { conversationId: 'idle-lifecycle' } },
      } as any);
      await flushMicrotasks();
      const h = mockHandlers.message[mockHandlers.message.length - 1];

      h(
        Buffer.from(
          JSON.stringify({
            type: 'response.output_text.done',
            text: 'hello',
          }),
        ),
      );
      h(
        Buffer.from(
          JSON.stringify({
            type: 'response.done',
            response: { usage: { total_tokens: 3, input_tokens: 1, output_tokens: 2 } },
          }),
        ),
      );

      await expect(responsePromise).resolves.toMatchObject({ output: 'hello' });

      // Turn-scoped cleanup removes its listener, but the idle lifecycle guard
      // must remain so a later socket failure is handled instead of becoming an
      // unhandled EventEmitter error.
      expect(mockHandlers.error.length).toBeGreaterThan(0);
      const idleErrorHandler = mockHandlers.error[mockHandlers.error.length - 1];
      idleErrorHandler(new Error('idle disconnect'));

      expect(provider.persistentConnection).toBeNull();
      expect((provider as any).connectionReady).toBeNull();
    });

    it('rejects when handler exceeds toolCallTimeout', async () => {
      vi.useFakeTimers();
      try {
        const functionCallHandler = vi.fn().mockImplementation(() => new Promise(() => {}));
        const provider = new OpenAiRealtimeProvider('gpt-realtime', {
          config: {
            modalities: ['text'],
            maintainContext: true,
            toolCallTimeout: 50,
            tools: [
              { type: 'function', name: 'hang', parameters: { type: 'object', properties: {} } },
            ],
            tool_choice: 'auto',
            functionCallHandler,
          },
        });

        provider.persistentConnection = {
          on: vi.fn((event: string, h: Function) => {
            mockHandlers[event].push(h);
          }),
          once: vi.fn((event: string, h: Function) => {
            mockHandlers[event].push(h);
          }),
          send: vi.fn(),
          close: vi.fn(),
          removeListener: vi.fn(),
          readyState: 1,
        } as unknown as WebSocket;

        const responsePromise = provider.callApi('go', {
          test: { metadata: { conversationId: 'timeout' } },
        } as any);

        // Need to flush in real-timer mode before the handler attaches.
        await vi.advanceTimersByTimeAsync(0);
        await Promise.resolve();
        await Promise.resolve();
        await Promise.resolve();
        const handler = mockHandlers.message[mockHandlers.message.length - 1];

        handler(
          Buffer.from(
            JSON.stringify({
              type: 'conversation.item.added',
              item: { id: 'u1', role: 'user' },
            }),
          ),
        );
        handler(
          Buffer.from(
            JSON.stringify({
              type: 'response.output_item.added',
              item: { type: 'function_call', call_id: 'c1', name: 'hang', arguments: '{}' },
            }),
          ),
        );
        const responseDoneFire = handler(
          Buffer.from(
            JSON.stringify({
              type: 'response.done',
              response: { usage: { total_tokens: 1, input_tokens: 1, output_tokens: 0 } },
            }),
          ),
        );

        // Fire the timeout.
        await vi.advanceTimersByTimeAsync(60);
        // Allow the catch + redacted send to flush.
        await responseDoneFire;

        const sentPayloads = vi
          .mocked(provider.persistentConnection!.send as Mock)
          .mock.calls.map(([raw]) => JSON.parse(raw as string));
        const toolOutput = sentPayloads.find(
          (e) => e.type === 'conversation.item.create' && e.item?.type === 'function_call_output',
        );
        expect(toolOutput).toBeDefined();
        expect(toolOutput.item.output).toBe('{"error":"Tool execution failed"}');
        provider.cleanup();
        // Don't await responsePromise — the follow-up response never completes
        // in this test; relying on cleanup to release.
        responsePromise.catch(() => {});
      } finally {
        vi.useRealTimers();
      }
    });

    it('rejects when tool-call iterations exceed maxToolIterations', async () => {
      const functionCallHandler = vi.fn().mockResolvedValue('{}');
      const provider = new OpenAiRealtimeProvider('gpt-realtime', {
        config: {
          modalities: ['text'],
          maintainContext: true,
          maxToolIterations: 2,
          tools: [
            { type: 'function', name: 'loop', parameters: { type: 'object', properties: {} } },
          ],
          tool_choice: 'auto',
          functionCallHandler,
        },
      });

      const persistentConnection = {
        on: vi.fn((event: string, h: Function) => {
          mockHandlers[event].push(h);
        }),
        once: vi.fn((event: string, h: Function) => {
          mockHandlers[event].push(h);
        }),
        send: vi.fn(),
        close: vi.fn(),
        removeListener: vi.fn(),
        readyState: 1,
      } as unknown as WebSocket;
      provider.persistentConnection = persistentConnection;

      const responsePromise = provider.callApi('go', {
        test: { metadata: { conversationId: 'cap' } },
      } as any);

      await flushMicrotasks();
      const handler = mockHandlers.message[mockHandlers.message.length - 1];

      handler(
        Buffer.from(
          JSON.stringify({
            type: 'conversation.item.added',
            item: { id: 'u1', role: 'user' },
          }),
        ),
      );

      const fireToolRound = async () => {
        handler(
          Buffer.from(
            JSON.stringify({
              type: 'response.output_item.added',
              item: { type: 'function_call', call_id: 'c', name: 'loop', arguments: '{}' },
            }),
          ),
        );
        handler(
          Buffer.from(
            JSON.stringify({
              type: 'response.done',
              response: { usage: { total_tokens: 1, input_tokens: 1, output_tokens: 0 } },
            }),
          ),
        );
        await flushMicrotasks();
      };

      await fireToolRound();
      await fireToolRound();
      await fireToolRound();

      const result = await responsePromise;
      expect(result.error).toMatch(/maxToolIterations=2/);
      provider.cleanup();
    });

    it('detaches the persistent message handler when iteration cap fires and preserves partial tool results', async () => {
      // Regression: hitting maxToolIterations on the persistent path used to
      // reject without removing the message-handler listener. On a shared
      // socket reused across turns, that stale listener would re-fire for
      // every subsequent event, recreating the handler-interleaving race
      // this PR was meant to prevent. Also asserts partial functionCallResults
      // are preserved in the rejected error's metadata so users can audit
      // what the model exchanged before the cap fired.
      const functionCallHandler = vi.fn().mockResolvedValue('{"called":true}');
      const provider = new OpenAiRealtimeProvider('gpt-realtime', {
        config: {
          modalities: ['text'],
          maintainContext: true,
          maxToolIterations: 2,
          tools: [
            { type: 'function', name: 'loop', parameters: { type: 'object', properties: {} } },
          ],
          tool_choice: 'auto',
          functionCallHandler,
        },
      });

      const persistentConnection = {
        on: vi.fn((event: string, h: Function) => {
          mockHandlers[event].push(h);
        }),
        once: vi.fn((event: string, h: Function) => {
          mockHandlers[event].push(h);
        }),
        send: vi.fn(),
        close: vi.fn(),
        removeListener: vi.fn((event: string, h: Function) => {
          const arr = mockHandlers[event];
          const idx = arr.indexOf(h);
          if (idx !== -1) {
            arr.splice(idx, 1);
          }
        }),
        readyState: 1,
      } as unknown as WebSocket;
      provider.persistentConnection = persistentConnection;

      const responsePromise = provider.callApi('go', {
        test: { metadata: { conversationId: 'cap-cleanup' } },
      } as any);

      await flushMicrotasks();
      const handler = mockHandlers.message[mockHandlers.message.length - 1];
      expect(mockHandlers.message).toContain(handler);

      handler(
        Buffer.from(
          JSON.stringify({ type: 'conversation.item.added', item: { id: 'u1', role: 'user' } }),
        ),
      );

      const fireToolRound = async () => {
        handler(
          Buffer.from(
            JSON.stringify({
              type: 'response.output_item.added',
              item: { type: 'function_call', call_id: 'c', name: 'loop', arguments: '{}' },
            }),
          ),
        );
        handler(
          Buffer.from(
            JSON.stringify({
              type: 'response.done',
              response: { usage: { total_tokens: 1, input_tokens: 1, output_tokens: 0 } },
            }),
          ),
        );
        await flushMicrotasks();
      };

      await fireToolRound();
      await fireToolRound();
      await fireToolRound();

      const result = await responsePromise;

      // Cap fired with diagnostic message.
      expect(result.error).toMatch(/maxToolIterations=2/);

      // I2: partial tool exchange is preserved in the error metadata, not
      // dropped on the floor.
      expect(result.metadata?.functionCallOccurred).toBe(true);
      expect(Array.isArray(result.metadata?.functionCallResults)).toBe(true);
      expect((result.metadata?.functionCallResults as string[]).length).toBeGreaterThan(0);

      // C2: the message handler must be detached so subsequent turns on the
      // shared socket don't fire stale listeners.
      expect(mockHandlers.message).not.toContain(handler);
      // The cap leaves an unresolved function_call on the server-side session,
      // so the persistent socket itself must be discarded before the next turn.
      expect(provider.persistentConnection).toBeNull();
      expect(vi.mocked(persistentConnection.close)).toHaveBeenCalledTimes(1);

      provider.cleanup();
    });

    it('should reset the persistent request timeout before a tool follow-up response', async () => {
      vi.useFakeTimers();

      try {
        const functionCallHandler = vi.fn().mockResolvedValue('{"call_status":"callback"}');
        const provider = new OpenAiRealtimeProvider('gpt-realtime', {
          config: {
            modalities: ['text'],
            maintainContext: true,
            websocketTimeout: 1000,
            tools: [
              {
                type: 'function',
                name: 'end_of_dialog_tool',
                parameters: {
                  type: 'object',
                  properties: {
                    call_status: { type: 'string' },
                  },
                },
              },
            ],
            functionCallHandler,
          },
        });

        provider.persistentConnection = {
          on: vi.fn((event: string, handler: Function) => {
            mockHandlers[event].push(handler);
            return provider.persistentConnection;
          }),
          once: vi.fn((event: string, handler: Function) => {
            mockHandlers[event].push(handler);
            return provider.persistentConnection;
          }),
          send: vi.fn(),
          close: vi.fn(),
          removeListener: vi.fn(),
        } as unknown as WebSocket;

        const responsePromise = provider.callApi('Can you call me back later?', {
          test: {
            metadata: { conversationId: 'loan-application-flow' },
          },
        } as any);

        await vi.waitFor(() => {
          expect(mockHandlers.message.length).toBeGreaterThan(0);
        });
        const lastHandler = mockHandlers.message[mockHandlers.message.length - 1];

        await Promise.resolve(
          lastHandler(
            Buffer.from(
              JSON.stringify({
                type: 'conversation.item.created',
                item: { id: 'msg_1', role: 'user' },
              }),
            ),
          ),
        );

        await vi.advanceTimersByTimeAsync(900);

        await Promise.resolve(
          lastHandler(
            Buffer.from(
              JSON.stringify({
                type: 'response.output_item.added',
                item: {
                  type: 'function_call',
                  call_id: 'call_1',
                  name: 'end_of_dialog_tool',
                  arguments: '{}',
                },
              }),
            ),
          ),
        );
        await Promise.resolve(
          lastHandler(
            Buffer.from(
              JSON.stringify({
                type: 'response.function_call_arguments.done',
                call_id: 'call_1',
                arguments: '{"call_status":"callback"}',
              }),
            ),
          ),
        );
        await Promise.resolve(
          lastHandler(
            Buffer.from(
              JSON.stringify({
                type: 'response.done',
                response: {
                  usage: {
                    total_tokens: 8,
                    input_tokens: 5,
                    output_tokens: 3,
                  },
                },
              }),
            ),
          ),
        );

        await vi.advanceTimersByTimeAsync(200);

        await Promise.resolve(
          lastHandler(
            Buffer.from(
              JSON.stringify({
                type: 'response.text.done',
                text: 'We will call you back later.',
              }),
            ),
          ),
        );
        await Promise.resolve(
          lastHandler(
            Buffer.from(
              JSON.stringify({
                type: 'response.done',
                response: {
                  usage: {
                    total_tokens: 11,
                    input_tokens: 7,
                    output_tokens: 4,
                  },
                },
              }),
            ),
          ),
        );

        await expect(responsePromise).resolves.toMatchObject({
          output: expect.stringContaining('We will call you back later.'),
          metadata: {
            functionCallOccurred: true,
            functionCallResults: ['{"call_status":"callback"}'],
          },
        });
      } finally {
        vi.useRealTimers();
      }
    });

    it('should pause the persistent request timeout while a tool handler is running', async () => {
      vi.useFakeTimers();

      try {
        let resolveFunctionCall: ((value: string) => void) | undefined;
        const functionCallHandler = vi.fn(
          () =>
            new Promise<string>((resolve) => {
              resolveFunctionCall = resolve;
            }),
        );
        const provider = new OpenAiRealtimeProvider('gpt-realtime', {
          config: {
            modalities: ['text'],
            maintainContext: true,
            websocketTimeout: 1000,
            tools: [
              {
                type: 'function',
                name: 'end_of_dialog_tool',
                parameters: {
                  type: 'object',
                  properties: {
                    call_status: { type: 'string' },
                  },
                },
              },
            ],
            functionCallHandler,
          },
        });

        provider.persistentConnection = {
          on: vi.fn((event: string, handler: Function) => {
            mockHandlers[event].push(handler);
            return provider.persistentConnection;
          }),
          once: vi.fn((event: string, handler: Function) => {
            mockHandlers[event].push(handler);
            return provider.persistentConnection;
          }),
          send: vi.fn(),
          close: vi.fn(),
          removeListener: vi.fn(),
        } as unknown as WebSocket;

        let responseSettled = false;
        const responsePromise = provider
          .callApi('Can you call me back later?', {
            test: {
              metadata: { conversationId: 'loan-application-flow' },
            },
          } as any)
          .finally(() => {
            responseSettled = true;
          });

        await vi.waitFor(() => {
          expect(mockHandlers.message.length).toBeGreaterThan(0);
        });
        const lastHandler = mockHandlers.message[mockHandlers.message.length - 1];

        await Promise.resolve(
          lastHandler(
            Buffer.from(
              JSON.stringify({
                type: 'conversation.item.created',
                item: { id: 'msg_1', role: 'user' },
              }),
            ),
          ),
        );
        await Promise.resolve(
          lastHandler(
            Buffer.from(
              JSON.stringify({
                type: 'response.output_item.added',
                item: {
                  type: 'function_call',
                  call_id: 'call_1',
                  name: 'end_of_dialog_tool',
                  arguments: '{}',
                },
              }),
            ),
          ),
        );
        await Promise.resolve(
          lastHandler(
            Buffer.from(
              JSON.stringify({
                type: 'response.function_call_arguments.done',
                call_id: 'call_1',
                arguments: '{"call_status":"callback"}',
              }),
            ),
          ),
        );

        const toolTurnPromise = Promise.resolve(
          lastHandler(
            Buffer.from(
              JSON.stringify({
                type: 'response.done',
                response: {
                  usage: {
                    total_tokens: 8,
                    input_tokens: 5,
                    output_tokens: 3,
                  },
                },
              }),
            ),
          ),
        );

        await vi.waitFor(() => {
          expect(functionCallHandler).toHaveBeenCalledWith(
            'end_of_dialog_tool',
            '{"call_status":"callback"}',
          );
        });

        await vi.advanceTimersByTimeAsync(1200);
        expect(responseSettled).toBe(false);

        resolveFunctionCall?.('{"call_status":"callback"}');
        await toolTurnPromise;

        await Promise.resolve(
          lastHandler(
            Buffer.from(
              JSON.stringify({
                type: 'response.text.done',
                text: 'We will call you back later.',
              }),
            ),
          ),
        );
        await Promise.resolve(
          lastHandler(
            Buffer.from(
              JSON.stringify({
                type: 'response.done',
                response: {
                  usage: {
                    total_tokens: 11,
                    input_tokens: 7,
                    output_tokens: 4,
                  },
                },
              }),
            ),
          ),
        );

        await expect(responsePromise).resolves.toMatchObject({
          output: expect.stringContaining('We will call you back later.'),
          metadata: {
            functionCallOccurred: true,
          },
        });
      } finally {
        vi.useRealTimers();
      }
    });

    it('should start the persistent request timeout after tool config loads', async () => {
      vi.useFakeTimers();

      try {
        let resolveTools: ((value: unknown[]) => void) | undefined;
        const toolsPromise = new Promise<unknown[]>((resolve) => {
          resolveTools = resolve;
        });
        const loadToolsSpy = vi
          .spyOn(util, 'maybeLoadToolsFromExternalFile')
          .mockReturnValue(toolsPromise as any);
        const provider = new OpenAiRealtimeProvider('gpt-realtime', {
          config: {
            modalities: ['text'],
            maintainContext: true,
            websocketTimeout: 2000,
            tools: [
              {
                type: 'function',
                name: 'end_of_dialog_tool',
                parameters: { type: 'object', properties: {} },
              },
            ],
          },
        });

        provider.persistentConnection = {
          on: vi.fn((event: string, handler: Function) => {
            mockHandlers[event].push(handler);
            return provider.persistentConnection;
          }),
          once: vi.fn((event: string, handler: Function) => {
            mockHandlers[event].push(handler);
            return provider.persistentConnection;
          }),
          send: vi.fn(),
          close: vi.fn(),
          removeListener: vi.fn(),
        } as unknown as WebSocket;

        const responsePromise = provider.callApi('Can you call me back later?', {
          test: {
            metadata: { conversationId: 'loan-application-flow' },
          },
        } as any);
        let settled = false;
        void responsePromise.finally(() => {
          settled = true;
        });

        await vi.advanceTimersByTimeAsync(1500);

        expect(settled).toBe(false);
        expect(provider.persistentConnection!.send).not.toHaveBeenCalled();

        resolveTools?.([]);
        await vi.waitFor(() => {
          expect(provider.persistentConnection!.send).toHaveBeenCalled();
        });

        const lastHandler = mockHandlers.message[mockHandlers.message.length - 1];
        await Promise.resolve(
          lastHandler(
            Buffer.from(
              JSON.stringify({
                type: 'conversation.item.created',
                item: { id: 'msg_1', role: 'user' },
              }),
            ),
          ),
        );
        await Promise.resolve(
          lastHandler(
            Buffer.from(
              JSON.stringify({
                type: 'response.text.done',
                text: 'We will call you back later.',
              }),
            ),
          ),
        );
        await Promise.resolve(
          lastHandler(
            Buffer.from(
              JSON.stringify({
                type: 'response.done',
                response: {
                  usage: {
                    total_tokens: 11,
                    input_tokens: 7,
                    output_tokens: 4,
                  },
                },
              }),
            ),
          ),
        );

        await expect(responsePromise).resolves.toMatchObject({
          output: expect.stringContaining('We will call you back later.'),
        });
        loadToolsSpy.mockRestore();
      } finally {
        vi.useRealTimers();
      }
    });

    it('should not leak a persistent request timeout when tool config loading fails', async () => {
      vi.useFakeTimers();

      try {
        const loadToolsSpy = vi
          .spyOn(util, 'maybeLoadToolsFromExternalFile')
          .mockRejectedValue(new Error('tools failed'));
        const provider = new OpenAiRealtimeProvider('gpt-realtime', {
          config: {
            modalities: ['text'],
            maintainContext: true,
            websocketTimeout: 1000,
            tools: [
              {
                type: 'function',
                name: 'end_of_dialog_tool',
                parameters: { type: 'object', properties: {} },
              },
            ],
          },
        });

        provider.persistentConnection = {
          on: vi.fn((event: string, handler: Function) => {
            mockHandlers[event].push(handler);
            return provider.persistentConnection;
          }),
          once: vi.fn((event: string, handler: Function) => {
            mockHandlers[event].push(handler);
            return provider.persistentConnection;
          }),
          send: vi.fn(),
          close: vi.fn(),
          removeListener: vi.fn(),
        } as unknown as WebSocket;

        const responsePromise = provider.callApi('Can you call me back later?', {
          test: {
            metadata: { conversationId: 'loan-application-flow' },
          },
        } as any);

        await expect(responsePromise).resolves.toMatchObject({
          error: expect.stringContaining('tools failed'),
        });
        await vi.advanceTimersByTimeAsync(1500);

        expect(vi.getTimerCount()).toBe(0);
        loadToolsSpy.mockRestore();
      } finally {
        vi.useRealTimers();
      }
    });

    it('should reject persistent requests when tool config loading times out', async () => {
      vi.useFakeTimers();

      try {
        const loadToolsSpy = vi
          .spyOn(util, 'maybeLoadToolsFromExternalFile')
          .mockReturnValue(new Promise<unknown[]>(() => undefined) as any);
        const provider = new OpenAiRealtimeProvider('gpt-realtime', {
          config: {
            modalities: ['text'],
            maintainContext: true,
            websocketTimeout: 1000,
            tools: [
              {
                type: 'function',
                name: 'end_of_dialog_tool',
                parameters: { type: 'object', properties: {} },
              },
            ],
          },
        });

        provider.persistentConnection = {
          on: vi.fn((event: string, handler: Function) => {
            mockHandlers[event].push(handler);
            return provider.persistentConnection;
          }),
          once: vi.fn((event: string, handler: Function) => {
            mockHandlers[event].push(handler);
            return provider.persistentConnection;
          }),
          send: vi.fn(),
          close: vi.fn(),
          removeListener: vi.fn(),
        } as unknown as WebSocket;

        const responsePromise = provider.callApi('Can you call me back later?', {
          test: {
            metadata: { conversationId: 'loan-application-flow' },
          },
        } as any);

        await vi.advanceTimersByTimeAsync(1000);

        await expect(responsePromise).resolves.toMatchObject({
          error: expect.stringContaining('Realtime tool configuration timed out after 1000ms'),
        });
        expect(provider.persistentConnection!.send).not.toHaveBeenCalled();
        expect(vi.getTimerCount()).toBe(0);
        loadToolsSpy.mockRestore();
      } finally {
        vi.useRealTimers();
      }
    });

    it('should reuse existing connection for subsequent requests', async () => {
      const provider = new OpenAiRealtimeProvider('gpt-4o-realtime-preview', {
        config: { modalities: ['text'], maintainContext: true },
      });
      const persistentConnection = createPersistentMockWebSocket(provider);
      provider.persistentConnection = persistentConnection;
      const context = { test: { metadata: { conversationId: 'reuse-connection' } } } as any;

      const completeTurn = async (output: string) => {
        await flushMicrotasks();
        const handler = lastMessageHandler();
        emitUserItem(handler, `item-${output}`);
        emitOutputTextDone(handler, output);
        emitResponseDone(handler);
      };

      (MockWebSocket as any).mockClear();

      const firstTurn = provider.callApi('first', context);
      await completeTurn('first');
      await expect(firstTurn).resolves.toMatchObject({ output: 'first' });

      const secondTurn = provider.callApi('second', context);
      await completeTurn('second');
      await expect(secondTurn).resolves.toMatchObject({ output: 'second' });

      expect(MockWebSocket).not.toHaveBeenCalled();
      expect(provider.persistentConnection).toBe(persistentConnection);
      expect(
        sentWebSocketEvents(persistentConnection).filter(
          (event) => event.type === 'conversation.item.create',
        ),
      ).toHaveLength(2);

      provider.cleanup();
    });

    it('serializes concurrent persistent-path turns and waits for OPEN before sending', async () => {
      // Regression test for the concurrency race that caused the shipped
      // promptfooconfig-conversation.yaml example to fail 100% at default
      // concurrency=4 with "WebSocket is not open: readyState 0 (CONNECTING)".
      const provider = new OpenAiRealtimeProvider('gpt-4o-realtime-preview', {
        config: { modalities: ['text'], maintainContext: true },
      });

      // Pre-set persistentConnection so openPersistentConnection() takes the
      // pre-existing-connection fast path and returns Promise.resolve().
      provider.persistentConnection = {
        on: vi.fn((event: string, handler: Function) => {
          mockHandlers[event].push(handler);
        }),
        once: vi.fn((event: string, handler: Function) => {
          mockHandlers[event].push(handler);
        }),
        send: vi.fn(),
        close: vi.fn(),
        removeListener: vi.fn(),
        readyState: 1, // OPEN
      } as unknown as WebSocket;

      const ctx = (id: string) => ({ test: { metadata: { conversationId: id } } }) as any;

      const drainTurn = async (text: string) => {
        await flushMicrotasks();
        const handler = mockHandlers.message[mockHandlers.message.length - 1];
        if (typeof handler !== 'function') {
          throw new Error('expected message handler attached');
        }
        handler(
          Buffer.from(
            JSON.stringify({
              type: 'conversation.item.added',
              item: { id: `item_${text}`, role: 'user' },
            }),
          ),
        );
        handler(Buffer.from(JSON.stringify({ type: 'response.created', response: { id: text } })));
        handler(Buffer.from(JSON.stringify({ type: 'response.output_text.delta', delta: text })));
        handler(Buffer.from(JSON.stringify({ type: 'response.output_text.done', text })));
        handler(
          Buffer.from(
            JSON.stringify({
              type: 'response.done',
              response: {
                usage: { total_tokens: 1, input_tokens: 1, output_tokens: 0 },
              },
            }),
          ),
        );
      };

      // Kick off two concurrent turns. Without serialization, both would
      // attach message handlers and interleave their session.update /
      // conversation.item.create events on the same socket.
      const p1 = provider.callApi('first', ctx('t1'));
      const p2 = provider.callApi('second', ctx('t1'));

      await drainTurn('first');
      const r1 = await p1;
      expect(r1.output).toBe('first');

      await drainTurn('second');
      const r2 = await p2;
      expect(r2.output).toBe('second');

      provider.cleanup();
    });

    it("reopens the persistent socket after cleanup() so a re-used provider isn't poisoned", async () => {
      // Regression: after a teardown, connectionReady stayed cached as a
      // resolved promise even though persistentConnection was nulled. Subsequent
      // turns then took the openPersistentConnection() fast path and failed in
      // sendEvent with "persistent WebSocket is not set", making the provider
      // permanently unusable for maintain-context flows after the first drop.
      const provider = new OpenAiRealtimeProvider('gpt-realtime', {
        config: { modalities: ['text'], maintainContext: true },
      });

      // Pre-set persistentConnection so the first openPersistentConnection()
      // takes the pre-existing-connection branch and caches connectionReady.
      provider.persistentConnection = createPersistentMockWebSocket();

      const ctx = { test: { metadata: { conversationId: 'reopen' } } } as any;
      const p1 = provider.callApi('first', ctx);
      await flushMicrotasks();

      // Drive a successful first turn so connectionReady is now cached.
      const lastHandler = () => mockHandlers.message[mockHandlers.message.length - 1];
      lastHandler()(
        Buffer.from(
          JSON.stringify({ type: 'conversation.item.added', item: { id: 'u1', role: 'user' } }),
        ),
      );
      lastHandler()(
        Buffer.from(JSON.stringify({ type: 'response.output_text.done', text: 'first' })),
      );
      lastHandler()(
        Buffer.from(
          JSON.stringify({
            type: 'response.done',
            response: { usage: { total_tokens: 1, input_tokens: 1, output_tokens: 0 } },
          }),
        ),
      );
      const r1 = await p1;
      expect(r1.output).toBe('first');

      // Simulate a server-initiated disconnect mid-life — fire a close event
      // on the live socket. The mid-turn close listener tearing down state is
      // covered by the existing turn-error pathway; here we just need the
      // cleanup helper to clear connectionReady so the next call reconnects.
      provider.cleanup();
      expect(provider.persistentConnection).toBeNull();
      expect((provider as any).connectionReady).toBeNull();

      // The next persistent turn must construct a new WebSocket. The mock
      // constructor was registered in the outer beforeEach; reset its call
      // count and prove a fresh socket is opened.
      (MockWebSocket as any).mockClear();
      mockHandlers.open = [];
      mockHandlers.error = [];
      mockHandlers.close = [];
      mockHandlers.message = [];

      const p2 = provider.callApi('second', ctx);
      await flushMicrotasks();
      // openPersistentConnection() must have called the WebSocket constructor.
      expect(MockWebSocket).toHaveBeenCalled();
      // Settle the new socket so the test doesn't leak.
      mockHandlers.open.forEach((h) => h());
      await flushMicrotasks();
      const h2 = mockHandlers.message[mockHandlers.message.length - 1];
      h2(
        Buffer.from(
          JSON.stringify({ type: 'conversation.item.added', item: { id: 'u2', role: 'user' } }),
        ),
      );
      h2(Buffer.from(JSON.stringify({ type: 'response.output_text.done', text: 'second' })));
      h2(
        Buffer.from(
          JSON.stringify({
            type: 'response.done',
            response: { usage: { total_tokens: 1, input_tokens: 1, output_tokens: 0 } },
          }),
        ),
      );
      const r2 = await p2;
      expect(r2.output).toBe('second');

      provider.cleanup();
    });

    it('rejects the in-flight turn and tears down state when the socket closes mid-turn', async () => {
      // Regression: setupMessageHandlers had no 'close' listener, so an
      // unexpected disconnect during a turn would let the caller's promise
      // dangle until the request timeout fired and would also leave
      // connectionReady cached so subsequent turns skipped reconnection.
      const provider = new OpenAiRealtimeProvider('gpt-realtime', {
        config: { modalities: ['text'], maintainContext: true },
      });

      const firstWs = {
        on: vi.fn((event: string, h: Function) => mockHandlers[event].push(h)),
        once: vi.fn((event: string, h: Function) => mockHandlers[event].push(h)),
        send: vi.fn(),
        close: vi.fn(),
        removeListener: vi.fn(),
        readyState: 1,
      } as unknown as WebSocket;
      provider.persistentConnection = firstWs;

      const ctx = { test: { metadata: { conversationId: 'mid-turn-close' } } } as any;
      const turnPromise = provider.callApi('mid-turn close please', ctx);
      await flushMicrotasks();

      // Send a partial-turn signal so the handler is mid-stream when close fires.
      emitUserItem(lastMessageHandler());

      // Fire the mid-turn 'close' on the same socket. The setupMessageHandlers
      // close listener must (a) reject the in-flight turn, (b) clear cached
      // lifecycle state. Pre-fix this listener didn't exist.
      const closeHandlers = mockHandlers.close;
      expect(closeHandlers.length).toBeGreaterThan(0);
      // The setupMessageHandlers listener is registered AFTER any
      // before-OPEN listeners; pick the last one to fire it.
      const midTurnClose = closeHandlers[closeHandlers.length - 1];
      midTurnClose(1011, Buffer.from('server policy violation'));

      const result = await turnPromise;
      expect(result.error).toMatch(/closed mid-turn|code=1011/i);
      // tearDown must run so the next turn reconnects.
      expect(provider.persistentConnection).toBeNull();
      expect((provider as any).connectionReady).toBeNull();
    });

    it('rejects the in-flight turn and tears down state when the response times out', async () => {
      // Regression: the websocketTimeout callback rejected the turn but — unlike
      // every other abort path — never detached the message handler or tore
      // down the persistent connection. The stale handler kept firing for later
      // turns' events (recreating the event-interleaving bug the provider was
      // hardened against) and connectionReady stayed cached so the next turn
      // skipped reconnection.
      vi.useFakeTimers();
      try {
        const provider = new OpenAiRealtimeProvider('gpt-realtime', {
          config: { modalities: ['text'], maintainContext: true, websocketTimeout: 50 },
        });

        const ws = {
          on: vi.fn((event: string, h: Function) => mockHandlers[event].push(h)),
          once: vi.fn((event: string, h: Function) => mockHandlers[event].push(h)),
          send: vi.fn(),
          close: vi.fn(),
          removeListener: vi.fn(),
          readyState: 1,
        } as unknown as WebSocket;
        provider.persistentConnection = ws;

        const ctx = { test: { metadata: { conversationId: 'timeout-turn' } } } as any;
        const turnPromise = provider.callApi('no response please', ctx);

        // Let setup register the request timeout, then advance past it without
        // ever delivering a server response.
        await vi.advanceTimersByTimeAsync(0);
        emitRealtimeEvent(lastMessageHandler(), {
          type: 'conversation.item.added',
          item: { id: 'assistant-timeout', role: 'assistant' },
        });
        expect(provider.previousItemId).toBe('assistant-timeout');
        expect(provider.assistantMessageIds).toEqual(['assistant-timeout']);
        await vi.advanceTimersByTimeAsync(100);

        const result = await turnPromise;
        expect(result.error).toMatch(/timed out/i);
        // The fix: the timeout path must detach the stale message handler and
        // drop the socket so the next turn reconnects cleanly.
        expect(ws.removeListener).toHaveBeenCalledWith('message', expect.any(Function));
        expect(ws.close).toHaveBeenCalled();
        expect(provider.persistentConnection).toBeNull();
        expect((provider as any).connectionReady).toBeNull();
        expect(provider.previousItemId).toBeNull();
        expect(provider.assistantMessageIds).toEqual([]);
      } finally {
        vi.useRealTimers();
      }
    });

    it('clears a failed setup timeout before a later persistent turn', async () => {
      vi.useFakeTimers();
      try {
        const provider = new OpenAiRealtimeProvider('gpt-realtime', {
          config: { modalities: ['text'], maintainContext: true, websocketTimeout: 50 },
        });
        const ctx = { test: { metadata: { conversationId: 'failed-setup' } } } as any;

        const firstWs = createPersistentMockWebSocket();
        vi.mocked(firstWs.send as Mock).mockImplementationOnce(() => {
          throw new Error('session update send failed');
        });
        provider.persistentConnection = firstWs;

        const failedTurn = await provider.callApi('fail during setup', ctx);
        expect(failedTurn.error).toMatch(/session update send failed/i);
        provider.cleanup();

        const nextWs = createPersistentMockWebSocket();
        provider.persistentConnection = nextWs;

        const laterTurn = provider.callApi('succeed after setup failure', ctx);
        await flushMicrotasks();
        const handler = lastMessageHandler();
        emitUserItem(handler, 'u-later');
        emitOutputTextDone(handler, 'later turn succeeded');
        emitResponseDone(handler);

        const result = await laterTurn;
        expect(result.output).toBe('later turn succeeded');

        await vi.advanceTimersByTimeAsync(100);
        expect(nextWs.close).not.toHaveBeenCalled();
        provider.cleanup();
      } finally {
        vi.useRealTimers();
      }
    });

    it('rejects connectionReady and clears state when the socket closes before OPEN', async () => {
      // Regression: openPersistentConnection() only listened for 'error', so
      // a handshake rejection that surfaces as 'close' (without a preceding
      // 'error') would leave connectionReady pending forever and freeze every
      // concurrent caller awaiting OPEN.
      const provider = new OpenAiRealtimeProvider('gpt-realtime', {
        config: { modalities: ['text'], maintainContext: true },
      });

      const ctx = { test: { metadata: { conversationId: 'close-before-open' } } } as any;
      const responsePromise = provider.callApi('hi', ctx);

      // Allow the WebSocket constructor + listener registration to settle.
      await flushMicrotasks();
      expect(mockHandlers.close.length).toBeGreaterThan(0);

      // Fire close without ever firing open.
      const closeHandler = mockHandlers.close[0];
      closeHandler(1006, Buffer.from('handshake failed'));

      const result = await responsePromise;
      expect(result.error).toMatch(/closed before OPEN/i);
      expect(provider.persistentConnection).toBeNull();
      expect((provider as any).connectionReady).toBeNull();
    });

    it('records redacted output in functionCallResults when the handler throws', async () => {
      // Regression: runToolCallRound() only pushed successful tool outputs,
      // so a handler that throws produced functionCallResults=[] even though
      // a redacted function_call_output had been sent to the model. callApi()
      // would then surface "no functionCallHandler was configured" as the
      // empty-response hint — wrong, since one was configured and ran.
      const functionCallHandler = vi
        .fn()
        .mockRejectedValue(new Error('connect ECONNREFUSED /var/secret/db.sock'));
      const provider = new OpenAiRealtimeProvider('gpt-realtime', {
        config: {
          modalities: ['text'],
          maintainContext: true,
          tools: [
            {
              type: 'function',
              name: 'lookup',
              parameters: { type: 'object', properties: {} },
            },
          ],
          tool_choice: 'auto',
          functionCallHandler,
        },
      });

      const persistentConnection = createPersistentMockWebSocket();
      provider.persistentConnection = persistentConnection;

      const responsePromise = provider.callApi('lookup something', {
        test: { metadata: { conversationId: 'handler-throw' } },
      } as any);
      await flushMicrotasks();
      const h = lastMessageHandler();

      emitUserItem(h);
      emitFunctionCall(h, 'lookup');
      emitFunctionCallArgumentsDone(h);
      await Promise.resolve(
        emitResponseDone(h, { total_tokens: 1, input_tokens: 1, output_tokens: 0 }),
      );
      // Drain microtasks until the handler-throw branch has executed.
      await flushMicrotasks();

      // The model's follow-up turn produces a normal text response after
      // seeing the redacted error.
      emitOutputTextDone(h, 'I had trouble looking that up.');
      emitResponseDone(h, { total_tokens: 4, input_tokens: 2, output_tokens: 2 });

      const result = await responsePromise;
      // The output must still resolve normally (not be wrapped in the
      // empty-response error) and the redacted error must be reported in
      // functionCallResults so users can audit what the model saw.
      expect(result.error).toBeUndefined();
      expect(result.metadata?.functionCallOccurred).toBe(true);
      expect(result.metadata?.functionCallResults).toEqual(['{"error":"Tool execution failed"}']);
      // Critical: the redacted output must NOT contain the leaky details.
      const sentPayloads = sentWebSocketEvents(persistentConnection);
      const toolOutput = sentPayloads.find(
        (e) => e.type === 'conversation.item.create' && e.item?.type === 'function_call_output',
      );
      expect(toolOutput.item.output).toBe('{"error":"Tool execution failed"}');
      expect(JSON.stringify(toolOutput)).not.toMatch(/ECONNREFUSED|var\/secret/);

      provider.cleanup();
    });

    it('pauses the direct-path request timeout while a tool handler is running', async () => {
      // Regression: in webSocketRequest / directWebSocketRequest the outer
      // websocketTimeout kept ticking while runToolCallRound() awaited user
      // code. A slow handler could be killed by the outer timeout before the
      // redacted function_call_output reached the model, masking the real
      // cause and making toolCallTimeout > websocketTimeout configurations
      // silently broken. We use fake timers (per test/AGENTS.md) and advance
      // past websocketTimeout while the handler promise is still pending —
      // if the production code didn't pause the outer timer, the request
      // would have rejected by now.
      vi.useFakeTimers();
      try {
        const deferredToolResult = createDeferredString();
        const functionCallHandler = vi.fn(() => deferredToolResult.promise);
        const provider = new OpenAiRealtimeProvider('gpt-realtime', {
          config: {
            modalities: ['text'],
            maintainContext: false,
            websocketTimeout: 50,
            toolCallTimeout: 5000,
            tools: [
              {
                type: 'function',
                name: 'slow',
                parameters: { type: 'object', properties: {} },
              },
            ],
            tool_choice: 'auto',
            functionCallHandler,
          },
        });

        let settled = false;
        const responsePromise = provider.callApi('slow tool please').finally(() => {
          settled = true;
        });

        await vi.advanceTimersByTimeAsync(0);
        mockHandlers.open.forEach((h) => h());
        await vi.advanceTimersByTimeAsync(0);

        const h = lastMessageHandler();
        emitUserItem(h);
        emitFunctionCall(h, 'slow');
        emitResponseDone(h, { total_tokens: 1, input_tokens: 1, output_tokens: 0 });
        await vi.advanceTimersByTimeAsync(0);
        expect(functionCallHandler).toHaveBeenCalledWith('slow', '{}');

        // Advance past websocketTimeout while the handler is still pending.
        // Without the pause/restart, the outer timeout would fire and reject
        // the request mid-tool execution.
        await vi.advanceTimersByTimeAsync(150);
        expect(settled).toBe(false);

        // Resolve the handler and complete the follow-up turn.
        deferredToolResult.resolve('{"ok":true}');
        await vi.advanceTimersByTimeAsync(0);
        emitOutputTextDone(h, 'tool done');
        emitResponseDone(h, { total_tokens: 2, input_tokens: 1, output_tokens: 1 });

        const result = await responsePromise;
        expect(result.error).toBeUndefined();
        expect(result.output).toContain('tool done');
      } finally {
        vi.useRealTimers();
      }
    });

    it('pauses the client-secret request timeout while a tool handler is running', async () => {
      vi.useFakeTimers();
      try {
        const deferredToolResult = createDeferredString();
        const functionCallHandler = vi.fn(() => deferredToolResult.promise);
        const provider = new OpenAiRealtimeProvider('gpt-realtime', {
          config: {
            modalities: ['text'],
            websocketTimeout: 50,
            toolCallTimeout: 5000,
            tools: [
              {
                type: 'function',
                name: 'slow',
                parameters: { type: 'object', properties: {} },
              },
            ],
            tool_choice: 'auto',
            functionCallHandler,
          },
        });

        let settled = false;
        const responsePromise = provider
          .webSocketRequest('secret123', 'slow tool please')
          .finally(() => {
            settled = true;
          });

        await vi.advanceTimersByTimeAsync(0);
        mockHandlers.open.forEach((h) => h());
        await vi.advanceTimersByTimeAsync(0);

        const h = lastMessageHandler();
        emitUserItem(h);
        emitFunctionCall(h, 'slow');
        emitResponseDone(h, { total_tokens: 1, input_tokens: 1, output_tokens: 0 });
        await vi.advanceTimersByTimeAsync(0);
        expect(functionCallHandler).toHaveBeenCalledWith('slow', '{}');

        await vi.advanceTimersByTimeAsync(150);
        expect(settled).toBe(false);

        deferredToolResult.resolve('{"ok":true}');
        await vi.advanceTimersByTimeAsync(0);
        emitOutputTextDone(h, 'tool done');
        emitResponseDone(h, { total_tokens: 2, input_tokens: 1, output_tokens: 1 });

        const result = await responsePromise;
        expect(result.output).toContain('tool done');
      } finally {
        vi.useRealTimers();
      }
    });
  });

  describe('Cleanup', () => {
    it('should properly clean up resources', () => {
      const provider = new OpenAiRealtimeProvider('gpt-4o-realtime-preview');

      // Create a properly typed mock
      const cleanupMockWs = {
        close: vi.fn(),
      } as unknown as WebSocket & {
        close: Mock;
      };

      provider.persistentConnection = cleanupMockWs;

      provider.cleanup();

      expect(cleanupMockWs.close).toHaveBeenCalledWith();
      expect(provider.persistentConnection).toBeNull();
    });
  });

  describe('WebSocket URL configuration', () => {
    beforeEach(() => {
      (MockWebSocket as any).mockClear();
    });

    const simulateMinimalFlow = () => {
      const messageHandlers = mockHandlers.message;
      const lastHandler = messageHandlers[messageHandlers.length - 1];

      // Simulate server creating user item so client will proceed
      lastHandler(
        Buffer.from(
          JSON.stringify({
            type: 'conversation.item.created',
            item: { id: 'msg_x', role: 'user' },
          }),
        ),
      );

      // Simulate response created and text events to resolve promises
      lastHandler(
        Buffer.from(JSON.stringify({ type: 'response.created', response: { id: 'r1' } })),
      );
      lastHandler(Buffer.from(JSON.stringify({ type: 'response.text.delta', delta: 'ok' })));
      lastHandler(Buffer.from(JSON.stringify({ type: 'response.text.done', text: 'ok' })));
      lastHandler(
        Buffer.from(
          JSON.stringify({
            type: 'response.done',
            response: { usage: { total_tokens: 1, input_tokens: 1, output_tokens: 0 } },
          }),
        ),
      );
    };

    const simulateGaFlow = () => {
      const messageHandlers = mockHandlers.message;
      const lastHandler = messageHandlers[messageHandlers.length - 1];

      lastHandler(
        Buffer.from(
          JSON.stringify({
            type: 'conversation.item.added',
            item: { id: 'msg_ga', role: 'user' },
          }),
        ),
      );
      lastHandler(
        Buffer.from(JSON.stringify({ type: 'response.created', response: { id: 'r_ga' } })),
      );
      lastHandler(Buffer.from(JSON.stringify({ type: 'response.output_text.delta', delta: 'ok' })));
      lastHandler(Buffer.from(JSON.stringify({ type: 'response.output_text.done', text: 'ok' })));
      lastHandler(
        Buffer.from(
          JSON.stringify({
            type: 'response.done',
            response: { usage: { total_tokens: 1, input_tokens: 1, output_tokens: 0 } },
          }),
        ),
      );
    };

    it('uses default OpenAI base for direct WebSocket', async () => {
      const provider = new OpenAiRealtimeProvider('gpt-4o-realtime-preview');
      const promise = provider.directWebSocketRequest('hi');

      // Trigger open to allow client to send
      mockHandlers.open.forEach((h) => h());
      simulateMinimalFlow();

      await promise;

      const constructedUrl = (MockWebSocket as any).mock.calls[0][0];
      const wsOptions = (MockWebSocket as any).mock.calls[0][1];
      expect(constructedUrl).toBe(
        'wss://api.openai.com/v1/realtime?model=' + encodeURIComponent('gpt-4o-realtime-preview'),
      );
      expect(wsOptions.headers['X-OpenAI-Originator']).toBe('promptfoo');
    });

    it('uses the GA realtime wire shape without the beta header', async () => {
      const provider = new OpenAiRealtimeProvider('gpt-realtime', {
        config: {
          modalities: ['text'],
          tools: [
            {
              type: 'function',
              name: 'get_weather',
              parameters: { type: 'object', properties: {} },
            },
          ],
        },
      });
      const promise = provider.directWebSocketRequest('hi');

      mockHandlers.open.forEach((h) => h());

      await vi.waitFor(() => {
        expect(mockWs.send).toHaveBeenCalled();
      });

      const sessionUpdate = JSON.parse(mockWs.send.mock.calls[0][0]);
      expect(sessionUpdate.session).toMatchObject({
        type: 'realtime',
        model: 'gpt-realtime',
        output_modalities: ['text'],
        tools: [{ type: 'function', name: 'get_weather' }],
      });

      const wsOptions = (MockWebSocket as any).mock.calls[0][1];
      expect(wsOptions.headers).not.toHaveProperty('OpenAI-Beta');

      simulateGaFlow();
      await expect(promise).resolves.toMatchObject({
        output: 'ok',
      });

      const responseCreate = JSON.parse(mockWs.send.mock.calls[2][0]);
      expect(responseCreate.response).toMatchObject({
        output_modalities: ['text'],
        tools: [{ type: 'function', name: 'get_weather' }],
      });
    });

    it('rejects direct WebSocket requests if the socket closes before a tool follow-up completes', async () => {
      const provider = new OpenAiRealtimeProvider('gpt-realtime', {
        config: {
          modalities: ['text'],
          functionCallHandler: vi.fn().mockResolvedValue('{"ok":true}'),
        },
      });
      const promise = provider.directWebSocketRequest('hi');

      mockHandlers.open.forEach((handler) => handler());

      const lastMessageHandler = mockHandlers.message[mockHandlers.message.length - 1];
      await Promise.resolve(
        lastMessageHandler(
          Buffer.from(
            JSON.stringify({
              type: 'conversation.item.created',
              item: { id: 'msg_tool', role: 'user' },
            }),
          ),
        ),
      );
      await Promise.resolve(
        lastMessageHandler(
          Buffer.from(
            JSON.stringify({
              type: 'response.output_item.added',
              item: {
                type: 'function_call',
                call_id: 'call_tool',
                name: 'get_weather',
                arguments: '{}',
              },
            }),
          ),
        ),
      );
      await Promise.resolve(
        lastMessageHandler(
          Buffer.from(
            JSON.stringify({
              type: 'response.function_call_arguments.done',
              call_id: 'call_tool',
              arguments: '{}',
            }),
          ),
        ),
      );
      await Promise.resolve(
        lastMessageHandler(
          Buffer.from(
            JSON.stringify({
              type: 'response.done',
              response: { usage: { total_tokens: 1, input_tokens: 1, output_tokens: 0 } },
            }),
          ),
        ),
      );

      mockHandlers.close[mockHandlers.close.length - 1](1006, Buffer.from('aborted'));

      await expect(promise).rejects.toThrow('WebSocket closed unexpectedly with code 1006');
    });

    it('preserves structured multimodal user content for direct WebSocket requests', async () => {
      const provider = new OpenAiRealtimeProvider('gpt-realtime-2', {
        config: {
          modalities: ['text'],
        },
      });
      const promise = provider.directWebSocketRequest(
        JSON.stringify([
          {
            role: 'user',
            content: [
              { type: 'input_text', text: 'Describe these inputs.' },
              { type: 'input_audio', audio: 'ZmFrZS1hdWRpbw==' },
              { type: 'input_image', image_url: 'data:image/jpeg;base64,ZmFrZS1pbWFnZQ==' },
            ],
          },
        ]),
      );

      mockHandlers.open.forEach((handler) => handler());

      await vi.waitFor(() => {
        expect(mockWs.send).toHaveBeenCalledTimes(2);
      });

      expect(JSON.parse(mockWs.send.mock.calls[1][0])).toMatchObject({
        type: 'conversation.item.create',
        item: {
          role: 'user',
          content: [
            { type: 'input_text', text: 'Describe these inputs.' },
            { type: 'input_audio', audio: 'ZmFrZS1hdWRpbw==' },
            { type: 'input_image', image_url: 'data:image/jpeg;base64,ZmFrZS1pbWFnZQ==' },
          ],
        },
      });

      simulateGaFlow();
      await expect(promise).resolves.toMatchObject({ output: 'ok' });
    });

    it('normalizes invalid max_response_output_tokens in session.update', async () => {
      const provider = new OpenAiRealtimeProvider('gpt-4o-realtime-preview', {
        config: { max_response_output_tokens: -1 },
      });
      const promise = provider.directWebSocketRequest('hi');

      mockHandlers.open.forEach((h) => h());

      await vi.waitFor(() => {
        expect(mockWs.send).toHaveBeenCalled();
      });

      const sessionUpdate = JSON.parse(mockWs.send.mock.calls[0][0]);
      expect(sessionUpdate.session.max_output_tokens).toBe('inf');

      const messageHandlers = mockHandlers.message;
      const lastHandler = messageHandlers[messageHandlers.length - 1];
      lastHandler(
        Buffer.from(
          JSON.stringify({
            type: 'conversation.item.created',
            item: { id: 'msg_zero', role: 'user' },
          }),
        ),
      );
      lastHandler(
        Buffer.from(JSON.stringify({ type: 'response.created', response: { id: 'resp_zero' } })),
      );
      lastHandler(Buffer.from(JSON.stringify({ type: 'response.text.done', text: 'ok' })));
      lastHandler(
        Buffer.from(
          JSON.stringify({
            type: 'response.done',
            response: { usage: { total_tokens: 1, input_tokens: 1, output_tokens: 0 } },
          }),
        ),
      );

      await promise;
    });

    it('rejects direct WebSocket requests when loading tools fails during open', async () => {
      const provider = new OpenAiRealtimeProvider('gpt-4o-realtime-preview', {
        config: { tools: 'file://missing-tools.yaml' as any },
      });
      const promise = provider.directWebSocketRequest('hi');

      mockHandlers.open.forEach((handler) => handler());

      await expect(promise).rejects.toThrow('File does not exist');
      expect(mockWs.close).toHaveBeenCalled();
    });

    describe('direct WebSocket tool config preload rejection handling', () => {
      const unhandledRejections: unknown[] = [];
      const onUnhandledRejection = (reason: unknown) => {
        unhandledRejections.push(reason);
      };

      beforeEach(() => {
        unhandledRejections.length = 0;
        process.on('unhandledRejection', onUnhandledRejection);
      });

      afterEach(() => {
        process.off('unhandledRejection', onUnhandledRejection);
      });

      it('handles direct WebSocket tool config preload rejections before open', async () => {
        const provider = new OpenAiRealtimeProvider('gpt-4o-realtime-preview', {
          config: { tools: 'file://missing-tools.yaml' as any },
        });
        const promise = provider.directWebSocketRequest('hi');

        await flushMicrotasks();
        await flushMicrotasks();
        expect(unhandledRejections).toEqual([]);

        mockHandlers.open.forEach((handler) => handler());
        await expect(promise).rejects.toThrow('File does not exist');
      });
    });

    it('converts custom https apiBaseUrl to wss for direct WebSocket', async () => {
      const provider = new OpenAiRealtimeProvider('gpt-4o-realtime-preview', {
        config: { apiBaseUrl: 'https://my-custom-api.com/v1' },
      });
      const promise = provider.directWebSocketRequest('hi');

      mockHandlers.open.forEach((h) => h());
      simulateMinimalFlow();

      await promise;

      const constructedUrl = (MockWebSocket as any).mock.calls[0][0];
      const wsOptions = (MockWebSocket as any).mock.calls[0][1];
      expect(constructedUrl).toBe(
        'wss://my-custom-api.com/v1/realtime?model=' +
          encodeURIComponent('gpt-4o-realtime-preview'),
      );
      expect(wsOptions.headers.Origin).toBe('https://my-custom-api.com');
    });

    it('converts custom http apiBaseUrl to ws for direct WebSocket', async () => {
      const provider = new OpenAiRealtimeProvider('gpt-4o-realtime-preview', {
        config: { apiBaseUrl: 'http://localhost:8080/v1' },
      });
      const promise = provider.directWebSocketRequest('hi');

      mockHandlers.open.forEach((h) => h());
      simulateMinimalFlow();

      await promise;

      const constructedUrl = (MockWebSocket as any).mock.calls[0][0];
      const wsOptions = (MockWebSocket as any).mock.calls[0][1];
      expect(constructedUrl).toBe(
        'ws://localhost:8080/v1/realtime?model=' + encodeURIComponent('gpt-4o-realtime-preview'),
      );
      expect(wsOptions.headers.Origin).toBe('http://localhost:8080');
    });

    it('uses apiBaseUrl for client-secret socket URL', async () => {
      const provider = new OpenAiRealtimeProvider('gpt-4o-realtime-preview', {
        config: { apiBaseUrl: 'https://my-custom-api.com/v1' },
      });
      const promise = provider.webSocketRequest('secret123', 'hi');

      mockHandlers.open.forEach((h) => h());
      simulateMinimalFlow();

      await promise;

      const constructedUrl = (MockWebSocket as any).mock.calls[0][0];
      const wsOptions = (MockWebSocket as any).mock.calls[0][1];
      expect(constructedUrl).toBe(
        'wss://my-custom-api.com/v1/realtime/socket?client_secret=' +
          encodeURIComponent('secret123'),
      );
      expect(wsOptions.headers.Origin).toBe('https://my-custom-api.com');
    });

    it('normalizes response-level tools for client-secret requests', async () => {
      const provider = new OpenAiRealtimeProvider('gpt-4o-realtime-preview', {
        config: {
          tools: [
            {
              type: 'function',
              function: {
                name: 'get_weather',
                description: 'Get the weather',
                parameters: { type: 'object', properties: {} },
              },
            },
          ],
          tool_choice: {
            type: 'function',
            function: { name: 'get_weather' },
          },
        },
      });
      const promise = provider.webSocketRequest('secret123', 'hi');

      mockHandlers.open.forEach((h) => h());
      simulateMinimalFlow();

      await promise;
      await vi.waitFor(() => {
        expect(
          mockWs.send.mock.calls
            .map(
              ([payload]: [string]) =>
                JSON.parse(payload) as {
                  type: string;
                  response?: { tools?: Record<string, unknown>[] };
                },
            )
            .some((event: { type: string }) => event.type === 'response.create'),
        ).toBe(true);
      });

      const sentEvents = mockWs.send.mock.calls.map(
        ([payload]: [string]) =>
          JSON.parse(payload) as { type: string; response?: { tools?: Record<string, unknown>[] } },
      );
      const responseCreate = sentEvents.find(
        (event: { type: string }) => event.type === 'response.create',
      );

      expect(responseCreate.response.tools).toEqual([
        {
          type: 'function',
          name: 'get_weather',
          description: 'Get the weather',
          parameters: { type: 'object', properties: {} },
        },
      ]);
      expect(responseCreate.response.tool_choice).toEqual({
        type: 'function',
        name: 'get_weather',
      });
    });
  });
});