RealtimeSession: uplink pump + downlink audio.delta forwarding

backend.tests/RealtimeSessionAudioTests.cs

@@ -0,0 +1,81 @@
+using System.Text.Json.Nodes;
+using backend.Conversations;
+using backend.Realtime;
+using backend.Tools;
+using FluentAssertions;
+using Microsoft.Extensions.DependencyInjection;
+using Xunit;
+
+namespace backend.tests;
+
+public class RealtimeSessionAudioTests(ApiFactory factory) : IClassFixture<ApiFactory>
+{
+    private readonly ApiFactory _factory = factory;
+
+    [Fact]
+    public async Task Uplink_bytes_become_input_audio_buffer_append()
+    {
+        using var scope = _factory.Services.CreateScope();
+        var log = scope.ServiceProvider.GetRequiredService<ConversationLog>();
+        var reg = new ToolRegistry(Array.Empty<ITool>());
+        var upstream = new ScriptedRealtimeUpstream();
+        var sink = new RecordingSink();
+        var cfg = new RealtimeSettings("m", "v", "p", 30, new HashSet<string>());
+        var session = new RealtimeSession(Guid.NewGuid(), upstream, sink, log, reg, cfg,
+            new FakeDeviceChannel(), NullLogger.Instance);
+
+        upstream.Push(new JsonObject { ["type"] = "session.updated" });
+        var cts = new CancellationTokenSource();
+        var run = session.RunAsync(cts.Token);
+        await sink.WaitForAsync("session_started", TimeSpan.FromSeconds(2));
+
+        var frame = new byte[3840];
+        for (int i = 0; i < frame.Length; i++) frame[i] = (byte)(i % 256);
+        await session.WriteUplinkAsync(frame, default);
+
+        await Task.Delay(100);
+
+        var appended = upstream.Sent
+            .FirstOrDefault(e => (string?)e["type"] == "input_audio_buffer.append");
+        appended.Should().NotBeNull();
+        var b64 = (string?)appended!["audio"];
+        Convert.FromBase64String(b64!).Length.Should().Be(3840);
+
+        cts.Cancel();
+        await run;
+    }
+
+    [Fact]
+    public async Task Audio_delta_from_upstream_becomes_binary_to_device()
+    {
+        using var scope = _factory.Services.CreateScope();
+        var log = scope.ServiceProvider.GetRequiredService<ConversationLog>();
+        var reg = new ToolRegistry(Array.Empty<ITool>());
+        var upstream = new ScriptedRealtimeUpstream();
+        var sink = new RecordingSink();
+        var cfg = new RealtimeSettings("m", "v", "p", 30, new HashSet<string>());
+        var session = new RealtimeSession(Guid.NewGuid(), upstream, sink, log, reg, cfg,
+            new FakeDeviceChannel(), NullLogger.Instance);
+
+        upstream.Push(new JsonObject { ["type"] = "session.updated" });
+        var cts = new CancellationTokenSource();
+        var run = session.RunAsync(cts.Token);
+        await sink.WaitForAsync("session_started", TimeSpan.FromSeconds(2));
+
+        var payload = new byte[3840];
+        for (int i = 0; i < payload.Length; i++) payload[i] = (byte)((i + 1) % 256);
+        upstream.Push(new JsonObject
+        {
+            ["type"] = "response.audio.delta",
+            ["delta"] = Convert.ToBase64String(payload),
+        });
+
+        await Task.Delay(100);
+        sink.Binary.Should().NotBeEmpty();
+        sink.Binary[0].Length.Should().Be(3840);
+        sink.Binary[0][0].Should().Be(1);
+
+        cts.Cancel();
+        await run;
+    }
+}

backend/Realtime/RealtimeSession.cs

@@ -26,6 +26,16 @@ public class RealtimeSession
     private EndReason _endReason = EndReason.Unknown;
     private bool _endRequested;
 
+    private readonly System.Threading.Channels.Channel<byte[]> _uplink =
+        System.Threading.Channels.Channel.CreateBounded<byte[]>(
+            new System.Threading.Channels.BoundedChannelOptions(64)
+            {
+                FullMode = System.Threading.Channels.BoundedChannelFullMode.DropOldest,
+            });
+
+    public ValueTask WriteUplinkAsync(byte[] frame, CancellationToken ct)
+        => _uplink.Writer.WriteAsync(frame, ct);
+
     public RealtimeSession(
         Guid deviceId,
         IRealtimeUpstream upstream,
@@ -69,7 +79,10 @@ public class RealtimeSession
                 ["conversation_id"] = _conversation.Id.ToString(),
             }, ct);
 
+            var uplinkTask = Task.Run(() => PumpUplinkAsync(ct), ct);
             await PumpAsync(ct);
+            _uplink.Writer.TryComplete();
+            await uplinkTask;
         }
         catch (OperationCanceledException)
         {
@@ -102,8 +115,36 @@ public class RealtimeSession
         {
             var evt = await _upstream.ReceiveJsonAsync(ct);
             if (evt is null) { _endReason = EndReason.Error; return; }
+            var type = (string?)evt["type"];
+            switch (type)
+            {
+                case "response.audio.delta":
+                    await ForwardAudioDeltaAsync(evt, ct);
+                    break;
             }
         }
+    }
+
+    private async Task ForwardAudioDeltaAsync(JsonObject evt, CancellationToken ct)
+    {
+        var b64 = (string?)evt["delta"];
+        if (string.IsNullOrEmpty(b64)) return;
+        var bytes = Convert.FromBase64String(b64);
+        await _output.WriteBinaryAsync(bytes, ct);
+    }
+
+    private async Task PumpUplinkAsync(CancellationToken ct)
+    {
+        try
+        {
+            await foreach (var frame in _uplink.Reader.ReadAllAsync(ct))
+            {
+                await _upstream.SendJsonAsync(RealtimeEvents.InputAudioBufferAppend(frame), ct);
+            }
+        }
+        catch (OperationCanceledException) { }
+        catch (Exception ex) { _logger.LogWarning(ex, "uplink pump errored"); }
+    }
 
     private async Task<JsonObject?> WaitForUpstreamTypeAsync(string type, CancellationToken ct)
     {