Assistant/backend/Realtime/OpenAIRealtimeUpstream.cs

using System.Net.WebSockets;
using System.Text;
using System.Text.Json.Nodes;

namespace backend.Realtime;

public class OpenAIRealtimeUpstream : IRealtimeUpstream
{
    private readonly ClientWebSocket _ws = new();
    private string? _closeReason;

    public bool IsClosed => _ws.State != WebSocketState.Open && _ws.State != WebSocketState.Connecting;
    public string? CloseReason => _closeReason;

    public async Task ConnectAsync(string apiKey, string model, CancellationToken ct)
    {
        _ws.Options.SetRequestHeader("Authorization", $"Bearer {apiKey}");
        // The GA Realtime API does NOT take the OpenAI-Beta: realtime=v1 header.
        // Sending it returns upstream error code=beta_api_shape_disabled.
        var uri = new Uri($"wss://api.openai.com/v1/realtime?model={Uri.EscapeDataString(model)}");
        await _ws.ConnectAsync(uri, ct);
    }

    public async Task SendJsonAsync(JsonObject envelope, CancellationToken ct)
    {
        var json = envelope.ToJsonString();
        var bytes = Encoding.UTF8.GetBytes(json);
        await _ws.SendAsync(bytes, WebSocketMessageType.Text, endOfMessage: true, ct);
    }

    public async Task<JsonObject?> ReceiveJsonAsync(CancellationToken ct)
    {
        var buffer = new byte[8192];
        using var stream = new MemoryStream();
        while (true)
        {
            WebSocketReceiveResult result;
            try
            {
                result = await _ws.ReceiveAsync(buffer, ct);
            }
            catch (WebSocketException exc)
            {
                _closeReason ??= $"WebSocketException:{exc.WebSocketErrorCode}:{exc.Message}";
                return null;
            }
            catch (OperationCanceledException)
            {
                return null;
            }

            if (result.MessageType == WebSocketMessageType.Close)
            {
                var status = _ws.CloseStatus?.ToString() ?? "<none>";
                var desc = _ws.CloseStatusDescription ?? "<no description>";
                _closeReason ??= $"close status={status} desc={desc}";
                return null;
            }
            stream.Write(buffer, 0, result.Count);
            if (result.EndOfMessage)
            {
                var json = Encoding.UTF8.GetString(stream.ToArray());
                return JsonNode.Parse(json)?.AsObject();
            }
        }
    }

    public async ValueTask DisposeAsync()
    {
        if (_ws.State == WebSocketState.Open)
        {
            try { await _ws.CloseAsync(WebSocketCloseStatus.NormalClosure, "bye", CancellationToken.None); }
            catch { }
        }
        _ws.Dispose();
    }
}