Assistant/backend/Realtime/RealtimeSession.cs

using System.Text.Json.Nodes;
using backend.Conversations;
using backend.Tools;
using Microsoft.Extensions.Logging;

namespace backend.Realtime;

public interface ISessionOutput
{
    Task WriteEnvelopeAsync(JsonObject envelope, CancellationToken ct);
    Task WriteBinaryAsync(ReadOnlyMemory<byte> bytes, CancellationToken ct);
}

public class RealtimeSession
{
    private readonly Guid _deviceId;
    private readonly IRealtimeUpstream _upstream;
    private readonly ISessionOutput _output;
    private readonly ConversationLog _log;
    private readonly ToolRegistry _registry;
    private readonly RealtimeSettings _cfg;
    private readonly IDeviceChannel _channel;
    private readonly ILogger _logger;

    private Conversation? _conversation;
    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,
        ISessionOutput output,
        ConversationLog log,
        ToolRegistry registry,
        RealtimeSettings cfg,
        IDeviceChannel channel,
        ILogger logger)
    {
        _deviceId = deviceId;
        _upstream = upstream;
        _output = output;
        _log = log;
        _registry = registry;
        _cfg = cfg;
        _channel = channel;
        _logger = logger;
    }

    public Guid? ConversationId => _conversation?.Id;

    public async Task RunAsync(CancellationToken ct)
    {
        try
        {
            var enabled = _registry.EnabledFor(_cfg.EnabledTools).ToList();
            await _upstream.SendJsonAsync(RealtimeEvents.SessionUpdate(_cfg, enabled), ct);

            var ack = await WaitForUpstreamTypeAsync("session.updated", ct);
            if (ack is null)
            {
                _endReason = EndReason.Error;
                return;
            }

            _conversation = await _log.StartAsync(_deviceId, ct);
            await _output.WriteEnvelopeAsync(new JsonObject
            {
                ["type"] = "session_started",
                ["conversation_id"] = _conversation.Id.ToString(),
            }, ct);

            var uplinkTask = Task.Run(() => PumpUplinkAsync(ct), ct);
            await PumpAsync(ct);
            _uplink.Writer.TryComplete();
            await uplinkTask;
        }
        catch (OperationCanceledException)
        {
            if (_endReason == EndReason.Unknown) _endReason = EndReason.Error;
        }
        catch (Exception ex)
        {
            _logger.LogError(ex, "RealtimeSession crashed");
            _endReason = EndReason.Error;
        }
        finally
        {
            await _upstream.DisposeAsync();
            if (_conversation is not null)
            {
                await _log.EndAsync(_conversation.Id, _endReason, CancellationToken.None);
            }
            var endEnv = new JsonObject
            {
                ["type"] = "session_ended",
                ["reason"] = _endReason.ToString().ToLowerInvariant(),
            };
            await _output.WriteEnvelopeAsync(endEnv, CancellationToken.None);
        }
    }

    private async Task PumpAsync(CancellationToken ct)
    {
        while (!ct.IsCancellationRequested && !_endRequested)
        {
            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;
                case "conversation.item.input_audio_transcription.completed":
                {
                    var text = (string?)evt["transcript"];
                    if (_conversation is not null && !string.IsNullOrEmpty(text))
                        await _log.AppendUserAsync(_conversation.Id, text!, ct);
                    break;
                }
                case "response.audio_transcript.done":
                {
                    var text = (string?)evt["transcript"];
                    if (_conversation is not null && !string.IsNullOrEmpty(text))
                        await _log.AppendAssistantAsync(_conversation.Id, text!, ct);
                    break;
                }
                case "response.done":
                    await _output.WriteEnvelopeAsync(
                        new JsonObject { ["type"] = "assistant_done" }, 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)
    {
        while (true)
        {
            var evt = await _upstream.ReceiveJsonAsync(ct);
            if (evt is null) return null;
            if ((string?)evt["type"] == type) return evt;
        }
    }
}