Assistant/tests/04a-realtime-oneshot-cs/Program.cs

using System.Net.WebSockets;
using System.Runtime.InteropServices;
using System.Text;
using System.Text.Json;
using System.Threading.Channels;
using PortAudioSharp;
using Stream = PortAudioSharp.Stream;

// .NET-side mirror for any readers that go through Environment.GetEnvironmentVariable,
// AND libc setenv so PortAudio (which uses getenv()) sees the value.
Environment.SetEnvironmentVariable("PA_ALSA_PLUGHW", "1");
Libc.setenv("PA_ALSA_PLUGHW", "1", 1);

string apiKey = LoadApiKey();

PortAudio.Initialize();

using var cts = new CancellationTokenSource();
Console.CancelKeyPress += (_, e) => { e.Cancel = true; cts.Cancel(); };

var upstreamChannel = Channel.CreateBounded<short[]>(new BoundedChannelOptions(C.UpstreamChannelCapacity)
{
    FullMode = BoundedChannelFullMode.Wait,
    SingleWriter = true,
    SingleReader = true,
});
var downstreamChannel = Channel.CreateBounded<short[]>(new BoundedChannelOptions(C.DownstreamChannelCapacity)
{
    FullMode = BoundedChannelFullMode.Wait,
    SingleWriter = false,
    SingleReader = true,
});

var doneFlag = new ManualResetEventSlim(false);
var state = new SharedState();

var ws = new ClientWebSocket();
ws.Options.SetRequestHeader("Authorization", $"Bearer {apiKey}");
// No OpenAI-Beta header — the GA Realtime API rejects beta-shaped requests with
// "beta_api_shape_disabled". Documented at developers.openai.com/api/docs/guides/realtime-websocket.

Stream? inputStream = null;
Stream? outputStream = null;
Task? receiveTask = null;
Task? sendTask = null;

try
{
    var uri = new Uri($"wss://api.openai.com/v1/realtime?model={C.ModelName}");
    Console.WriteLine($"[ws] connecting to {uri}");
    try
    {
        await ws.ConnectAsync(uri, cts.Token);
    }
    catch (Exception ex)
    {
        Console.Error.WriteLine($"[error] WS connect failed: {ex.Message}");
        if (ex.InnerException is HttpRequestException httpEx)
            Console.Error.WriteLine($"[error] inner: HTTP {(int?)httpEx.StatusCode}");
        Environment.ExitCode = 3;
        return;
    }
    Console.WriteLine("[ws] connected");

    receiveTask = Task.Run(() => ReceiveLoop(ws, downstreamChannel, cts, state));

    await SendSessionUpdate(ws, cts.Token);

    int device = FindUsbDevice();
    Console.WriteLine($"Using device {device} ('{PortAudio.GetDeviceInfo(device).name}')");

    inputStream = OpenInputStream(device, upstreamChannel, state);
    try { inputStream.Start(); }
    catch (Exception ex)
    {
        Console.Error.WriteLine($"[error] input stream start failed: {ex.Message}");
        Environment.ExitCode = 5;
        return;
    }

    outputStream = OpenOutputStream(device, downstreamChannel, state, doneFlag);
    try { outputStream.Start(); }
    catch (Exception ex)
    {
        Console.Error.WriteLine($"[error] output stream start failed: {ex.Message}");
        Environment.ExitCode = 6;
        return;
    }

    sendTask = Task.Run(() => UpstreamSendLoop(ws, upstreamChannel, cts, state));

    // Startup beep — queued onto downstreamChannel before the receive loop produces any audio.
    short[] beep = MakeBeep(C.BeepHz, C.BeepSeconds, C.SampleRate);
    Console.WriteLine("[beep]");
    await downstreamChannel.Writer.WriteAsync(beep, cts.Token);

    // Wait for beep to drain (200 ms beep + 100 ms margin) so it doesn't get captured
    // by the mic and shipped to OpenAI.
    Thread.Sleep((int)(C.BeepSeconds * 1000) + C.BeepDrainMs);
    state.MicArmed = true;
    Console.WriteLine("Speak now.");

    try
    {
        doneFlag.Wait(cts.Token);
    }
    catch (OperationCanceledException) { /* Ctrl-C path */ }

    Console.WriteLine("[ws] closing");
}
catch (OperationCanceledException)
{
    // Mid-startup cancellation (typically a WS error event fired cts.Cancel while
    // the main thread was awaiting SendSessionUpdate / a channel write).
    // The receive loop has already logged the underlying [error]; we just route through cleanup.
}
catch (Exception ex)
{
    Console.Error.WriteLine($"[error] unexpected: {ex.GetType().Name}: {ex.Message}");
    Environment.ExitCode = 7;
}
finally
{
    // Cancel any background work first so receive/send loops exit cleanly.
    cts.Cancel();

    if (ws.State == WebSocketState.Open || ws.State == WebSocketState.CloseReceived)
    {
        try
        {
            using var closeTimeout = new CancellationTokenSource(TimeSpan.FromSeconds(2));
            await ws.CloseAsync(WebSocketCloseStatus.NormalClosure, "done", closeTimeout.Token);
        }
        catch (Exception ex)
        {
            Console.Error.WriteLine($"[ws] close failed: {ex.Message}");
        }
    }

    try { inputStream?.Stop(); } catch { }
    try { outputStream?.Stop(); } catch { }
    try { inputStream?.Dispose(); } catch { }
    try { outputStream?.Dispose(); } catch { }
    try { ws.Dispose(); } catch { }
    try { PortAudio.Terminate(); } catch { }

    Console.WriteLine("bye");
}

static string LoadApiKey()
{
    string home = Environment.GetEnvironmentVariable("HOME") ?? "/root";
    string path = Path.Combine(home, ".openai_key");
    if (!File.Exists(path))
    {
        Console.Error.WriteLine($"[error] API key file not found at {path}");
        Environment.Exit(2);
    }

    // Non-fatal mode-600 check.
    try
    {
        var mode = File.GetUnixFileMode(path);
        var leaky = mode & (UnixFileMode.GroupRead | UnixFileMode.GroupWrite | UnixFileMode.GroupExecute
                          | UnixFileMode.OtherRead | UnixFileMode.OtherWrite | UnixFileMode.OtherExecute);
        if (leaky != UnixFileMode.None)
            Console.Error.WriteLine($"[warn] {path} mode includes group/other perms ({mode}); expected 600");
    }
    catch (PlatformNotSupportedException) { /* not Unix */ }
    catch (Exception ex)
    {
        Console.Error.WriteLine($"[warn] could not check {path} mode: {ex.Message}");
    }

    string key = File.ReadAllText(path).Trim();
    if (string.IsNullOrEmpty(key))
    {
        Console.Error.WriteLine($"[error] API key file empty");
        Environment.Exit(2);
    }
    if (!key.StartsWith("sk-"))
    {
        Console.Error.WriteLine($"[error] API key doesn't start with sk-");
        Environment.Exit(2);
    }
    return key;
}

static int FindUsbDevice()
{
    int count = PortAudio.DeviceCount;
    for (int i = 0; i < count; i++)
    {
        var info = PortAudio.GetDeviceInfo(i);
        if (info.name.ToLowerInvariant().Contains("usb") && info.maxInputChannels >= 1)
            return i;
    }
    Console.Error.WriteLine("USB audio device not found. Devices:");
    for (int i = 0; i < count; i++)
    {
        var info = PortAudio.GetDeviceInfo(i);
        Console.Error.WriteLine(
            $"  [{i}] {info.name} in={info.maxInputChannels} out={info.maxOutputChannels}");
    }
    Environment.Exit(1);
    return -1; // unreachable
}

static short[] MakeBeep(double freqHz, double durationS, int sampleRate)
{
    int n = (int)(sampleRate * durationS);
    var buf = new short[n];
    for (int i = 0; i < n; i++)
    {
        double t = i / (double)sampleRate;
        double v = 0.3 * Math.Sin(2.0 * Math.PI * freqHz * t);
        buf[i] = (short)(v * short.MaxValue);
    }
    return buf;
}

static async Task SendSessionUpdate(ClientWebSocket ws, CancellationToken ct)
{
    // GA Realtime API shape: nested session.audio.{input,output} with format + voice + turn_detection
    // moved under audio.input. session.type = "realtime" and session.model inline.
    // (Beta-era flat shape with session.input_audio_format/turn_detection at the top is rejected.)
    var msg = new
    {
        type = "session.update",
        session = new
        {
            type = "realtime",
            model = C.ModelName,
            instructions = C.Instructions,
            // No 'modalities' field — the beta name. GA defaults are fine; if we ever need to
            // constrain to audio-only we'd use 'output_modalities' (the documented GA name in
            // newer OpenAI APIs).
            audio = new
            {
                input = new
                {
                    // GA shape: format is now an object with a "type" field, not a bare string.
                    format = new { type = "audio/pcm", rate = 24_000 },
                    turn_detection = new
                    {
                        type = "server_vad",
                        threshold = 0.5,
                        prefix_padding_ms = 300,
                        silence_duration_ms = 500,
                        create_response = true,
                    },
                },
                output = new
                {
                    format = new { type = "audio/pcm", rate = 24_000 },
                    voice = C.Voice,
                },
            },
        },
    };
    byte[] bytes = JsonSerializer.SerializeToUtf8Bytes(msg);
    await ws.SendAsync(new ArraySegment<byte>(bytes), WebSocketMessageType.Text, endOfMessage: true, ct);
}

static Stream OpenInputStream(int device, Channel<short[]> upstream, SharedState state)
{
    var inParams = new StreamParameters
    {
        device = device,
        channelCount = C.Channels,
        sampleFormat = SampleFormat.Int16,
        suggestedLatency = PortAudio.GetDeviceInfo(device).defaultLowInputLatency,
        hostApiSpecificStreamInfo = IntPtr.Zero,
    };

    Stream.Callback callback = (IntPtr input, IntPtr _, uint frameCount,
        ref StreamCallbackTimeInfo _2, StreamCallbackFlags status, IntPtr _3) =>
    {
        if (status.HasFlag(StreamCallbackFlags.InputOverflow))
            Console.Error.WriteLine("[status] input overflow");

        if (!state.MicArmed || state.StopSending)
            return StreamCallbackResult.Continue;

        if (frameCount != C.MicBlockFrames)
        {
            Console.Error.WriteLine($"[status] unexpected input frameCount={frameCount} (want {C.MicBlockFrames})");
            return StreamCallbackResult.Continue;
        }

        var frame = new short[C.MicBlockFrames];
        unsafe
        {
            short* src = (short*)input.ToPointer();
            fixed (short* dst = &frame[0])
                Buffer.MemoryCopy(src, dst, C.MicBlockFrames * sizeof(short), C.MicBlockFrames * sizeof(short));
        }
        if (!upstream.Writer.TryWrite(frame))
            Console.Error.WriteLine("[status] upstream behind, dropping mic frame");
        return StreamCallbackResult.Continue;
    };

    return new Stream(inParams, null, C.SampleRate, C.MicBlockFrames, StreamFlags.NoFlag, callback, IntPtr.Zero);
}

static Stream OpenOutputStream(int device, Channel<short[]> downstream, SharedState state, ManualResetEventSlim doneFlag)
{
    var outParams = new StreamParameters
    {
        device = device,
        channelCount = C.Channels,
        sampleFormat = SampleFormat.Int16,
        suggestedLatency = PortAudio.GetDeviceInfo(device).defaultLowOutputLatency,
        hostApiSpecificStreamInfo = IntPtr.Zero,
    };

    Stream.Callback callback = (IntPtr _, IntPtr output, uint frameCount,
        ref StreamCallbackTimeInfo _2, StreamCallbackFlags _3, IntPtr _4) =>
    {
        int fill = 0;

        while (fill < frameCount)
        {
            // Pull next chunk if we have no working one.
            if (state.CurrentChunk == null)
            {
                if (!downstream.Reader.TryRead(out var nextChunk))
                    break; // ring empty
                state.CurrentChunk = nextChunk;
                state.ChunkOffset = 0;
            }
            // Local non-null alias so the compiler doesn't warn on dereference inside `fixed`.
            short[] chunk = state.CurrentChunk!;
            int slack = (int)frameCount - fill;
            int remaining = chunk.Length - state.ChunkOffset;
            int take = Math.Min(remaining, slack);
            unsafe
            {
                short* dst = (short*)output.ToPointer();
                fixed (short* src = &chunk[state.ChunkOffset])
                    Buffer.MemoryCopy(src, dst + fill, take * sizeof(short), take * sizeof(short));
            }
            state.ChunkOffset += take;
            fill += take;
            if (state.ChunkOffset >= chunk.Length)
                state.CurrentChunk = null;
        }

        unsafe
        {
            short* dst = (short*)output.ToPointer();
            for (int i = fill; i < (int)frameCount; i++) dst[i] = 0;
        }

        // Completion = working chunk drained AND channel empty AND no more deltas expected.
        // _noMoreDeltas is the spec's "redundant fast path" — set immediately on response.done,
        // avoids touching downstream.Reader.Completion (a Task allocation per callback).
        if (state.CurrentChunk == null
            && downstream.Reader.Count == 0
            && state.NoMoreDeltas)
        {
            doneFlag.Set();
            return StreamCallbackResult.Complete;
        }
        return StreamCallbackResult.Continue;
    };

    return new Stream(null, outParams, C.SampleRate, C.OutBlockFrames, StreamFlags.NoFlag, callback, IntPtr.Zero);
}

static async Task UpstreamSendLoop(ClientWebSocket ws, Channel<short[]> upstream, CancellationTokenSource cts, SharedState state)
{
    try
    {
        await foreach (var frame in upstream.Reader.ReadAllAsync(cts.Token))
        {
            if (state.StopSending)
            {
                // After server VAD said the user is done, drop in-flight mic frames and exit
                // once the channel is drained. Matches the spec's "_stopSending && empty" exit.
                if (upstream.Reader.Count == 0) break;
                continue;
            }
            ReadOnlySpan<byte> bytes = MemoryMarshal.AsBytes(frame.AsSpan());
            string base64 = Convert.ToBase64String(bytes);
            var msg = new { type = "input_audio_buffer.append", audio = base64 };
            byte[] json = JsonSerializer.SerializeToUtf8Bytes(msg);
            await ws.SendAsync(new ArraySegment<byte>(json), WebSocketMessageType.Text, endOfMessage: true, cts.Token);
        }
    }
    catch (OperationCanceledException) { /* expected on Cancel */ }
    catch (Exception ex)
    {
        Console.Error.WriteLine($"[error] upstream send: {ex.Message}");
        cts.Cancel();
    }
}

static async Task ReceiveLoop(ClientWebSocket ws, Channel<short[]> downstream, CancellationTokenSource cts, SharedState state)
{
    var transcript = new StringBuilder();
    var buffer = new byte[4096];
    bool firstDeltaSeen = false;
    DateTime speechStoppedTime = DateTime.MinValue;
    using var msgBuffer = new MemoryStream();

    try
    {
        while (!cts.IsCancellationRequested)
        {
            msgBuffer.SetLength(0);
            WebSocketReceiveResult result;
            do
            {
                result = await ws.ReceiveAsync(new ArraySegment<byte>(buffer), cts.Token);
                if (result.MessageType == WebSocketMessageType.Close) return;
                msgBuffer.Write(buffer, 0, result.Count);
            } while (!result.EndOfMessage);

            msgBuffer.Position = 0;
            using var doc = JsonDocument.Parse(msgBuffer);
            string type = doc.RootElement.GetProperty("type").GetString() ?? "";

            switch (type)
            {
                case "session.created":
                {
                    string? sid = doc.RootElement.TryGetProperty("session", out var sess)
                        && sess.TryGetProperty("id", out var idEl) ? idEl.GetString() : null;
                    Console.WriteLine($"[ws] session.created {sid}");
                    break;
                }
                case "session.updated":
                    Console.WriteLine("[ws] session.updated — session pinned");
                    break;
                case "input_audio_buffer.speech_started":
                    Console.WriteLine("[vad] speech started");
                    break;
                case "input_audio_buffer.speech_stopped":
                    Console.WriteLine("[vad] speech stopped");
                    speechStoppedTime = DateTime.UtcNow;
                    state.StopSending = true;
                    break;
                case "input_audio_buffer.committed":
                    Console.WriteLine("[ws] input_audio_buffer.committed");
                    break;
                case "response.created":
                    Console.WriteLine("[ws] response.created");
                    break;
                // GA event names are response.output_audio* (beta was response.audio*).
                case "response.output_audio_transcript.delta":
                    if (doc.RootElement.TryGetProperty("delta", out var tEl))
                        transcript.Append(tEl.GetString());
                    break;
                case "response.output_audio.delta":
                {
                    string b64 = doc.RootElement.GetProperty("delta").GetString() ?? "";
                    byte[] audioBytes = Convert.FromBase64String(b64);
                    short[] samples = new short[audioBytes.Length / 2];
                    Buffer.BlockCopy(audioBytes, 0, samples, 0, audioBytes.Length);
                    if (!firstDeltaSeen)
                    {
                        firstDeltaSeen = true;
                        double elapsed = speechStoppedTime == DateTime.MinValue
                            ? -1
                            : (DateTime.UtcNow - speechStoppedTime).TotalSeconds;
                        Console.WriteLine($"[audio] first delta received (t = {elapsed:0.00} s since speech_stopped)");
                    }
                    await downstream.Writer.WriteAsync(samples, cts.Token);
                    break;
                }
                case "response.output_audio.done":
                    Console.WriteLine("[ws] response.output_audio.done");
                    break;
                case "response.output_audio_transcript.done":
                    // transcript already accumulated; nothing to do here, server signals end of stream.
                    break;
                // GA emits a stream of conversation/response framing events we don't need for the
                // probe — quiet them so they don't bury real signal in the log.
                case "conversation.item.added":
                case "conversation.item.done":
                case "response.output_item.added":
                case "response.output_item.done":
                case "response.content_part.added":
                case "response.content_part.done":
                case "rate_limits.updated":
                    break;
                case "response.done":
                    state.NoMoreDeltas = true;
                    downstream.Writer.Complete();
                    Console.WriteLine("[ws] response.done");
                    Console.WriteLine($"[transcript] {transcript}");
                    break;
                case "error":
                    Console.Error.WriteLine($"[error] WS error event: {doc.RootElement.GetRawText()}");
                    cts.Cancel();
                    break;
                default:
                    Console.Error.WriteLine($"[ws] ignored type={type}");
                    break;
            }
        }
    }
    catch (OperationCanceledException) { /* expected on Cancel */ }
    catch (Exception ex)
    {
        Console.Error.WriteLine($"[error] receive loop: {ex.Message}");
        cts.Cancel();
    }
}

sealed class SharedState
{
    public volatile bool MicArmed;
    public volatile bool StopSending;
    public volatile bool NoMoreDeltas;
    // Output-callback-thread-local; do NOT touch from anywhere else.
    public short[]? CurrentChunk;
    public int ChunkOffset;
}

static class C
{
    public const int SampleRate = 24_000;
    public const int Channels = 1;
    public const uint MicBlockFrames = 1920;        // 40 ms @ 24 kHz
    public const uint OutBlockFrames = 1024;        // ~43 ms @ 24 kHz
    public const string ModelName = "gpt-realtime";
    public const string Voice = "alloy";
    public const string Instructions = "Reply in one short sentence.";
    public const double BeepHz = 880.0;
    public const double BeepSeconds = 0.2;
    public const int BeepDrainMs = 100;
    public const int UpstreamChannelCapacity = 64;
    public const int DownstreamChannelCapacity = 64;
}

static class Libc
{
    [DllImport("libc", EntryPoint = "setenv")]
    public static extern int setenv(string name, string value, int overwrite);
}