Add RealtimeSession lifecycle (open + session.update + session_started/ended)

2026-06-11 21:30:40 +00:00
parent 8f197d1203
commit a202725c39
2 changed files with 209 additions and 0 deletions
@@ -0,0 +1,92 @@
+using System.Text.Json.Nodes;
+using backend.Conversations;
+using backend.Data;
+using backend.Realtime;
+using backend.Tools;
+using FluentAssertions;
+using Microsoft.Extensions.DependencyInjection;
+using Xunit;
+
+namespace backend.tests;
+
+public class RealtimeSessionLifecycleTests(ApiFactory factory) : IClassFixture<ApiFactory>
+{
+    private readonly ApiFactory _factory = factory;
+
+    [Fact]
+    public async Task Open_emits_session_update_then_session_started_then_session_ended_on_cancel()
+    {
+        using var scope = _factory.Services.CreateScope();
+        var log = scope.ServiceProvider.GetRequiredService<ConversationLog>();
+        var registry = new ToolRegistry(new ITool[] { new GetCurrentTimeTool() });
+
+        var upstream = new ScriptedRealtimeUpstream();
+        var sink = new RecordingSink();
+        var cfg = new RealtimeSettings(
+            Model: "gpt-4o-realtime-preview",
+            Voice: "alloy",
+            SystemPrompt: "test",
+            IdleTimeoutSeconds: 30,
+            EnabledTools: new HashSet<string> { "get_current_time" });
+
+        var session = new RealtimeSession(
+            deviceId: Guid.NewGuid(),
+            upstream, sink, log, registry, cfg,
+            channel: new FakeDeviceChannel(),
+            logger: NullLogger.Instance);
+
+        upstream.Push(new JsonObject { ["type"] = "session.updated" });
+
+        var cts = new CancellationTokenSource();
+        var runTask = session.RunAsync(cts.Token);
+
+        await sink.WaitForAsync("session_started", TimeSpan.FromSeconds(2));
+
+        cts.Cancel();
+        await runTask;
+
+        upstream.Sent.Select(e => (string?)e["type"]).Should().StartWith("session.update");
+        sink.Envelopes.Select(e => (string?)e["type"]).Should().Contain("session_started");
+        sink.Envelopes.Select(e => (string?)e["type"]).Should().Contain("session_ended");
+    }
+}
+
+internal class NullLogger : Microsoft.Extensions.Logging.ILogger
+{
+    public static readonly NullLogger Instance = new();
+    public IDisposable? BeginScope<TState>(TState state) where TState : notnull => null;
+    public bool IsEnabled(Microsoft.Extensions.Logging.LogLevel logLevel) => false;
+    public void Log<TState>(Microsoft.Extensions.Logging.LogLevel l, Microsoft.Extensions.Logging.EventId e,
+        TState s, Exception? ex, Func<TState, Exception?, string> f) { }
+}
+
+internal class RecordingSink : ISessionOutput
+{
+    public List<JsonObject> Envelopes { get; } = new();
+    public List<byte[]> Binary { get; } = new();
+    public Task WriteEnvelopeAsync(JsonObject e, CancellationToken ct)
+    {
+        lock (Envelopes) Envelopes.Add(e);
+        _gate.Release();
+        return Task.CompletedTask;
+    }
+    public Task WriteBinaryAsync(ReadOnlyMemory<byte> b, CancellationToken ct)
+    {
+        Binary.Add(b.ToArray());
+        return Task.CompletedTask;
+    }
+    private readonly SemaphoreSlim _gate = new(0, int.MaxValue);
+    public async Task WaitForAsync(string type, TimeSpan timeout)
+    {
+        var deadline = DateTime.UtcNow + timeout;
+        while (DateTime.UtcNow < deadline)
+        {
+            lock (Envelopes)
+            {
+                if (Envelopes.Any(e => (string?)e["type"] == type)) return;
+            }
+            await _gate.WaitAsync(timeout);
+        }
+        throw new TimeoutException($"never saw {type}");
+    }
+}
@@ -0,0 +1,117 @@
+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;
+
+    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);
+
+            await PumpAsync(ct);
+        }
+        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; }
+        }
+    }
+
+    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;
+        }
+    }
+}