diff --git a/backend.tests/RealtimeSessionLifecycleTests.cs b/backend.tests/RealtimeSessionLifecycleTests.cs new file mode 100644 index 0000000..42882f3 --- /dev/null +++ b/backend.tests/RealtimeSessionLifecycleTests.cs @@ -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 +{ + 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(); + 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 { "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 state) where TState : notnull => null; + public bool IsEnabled(Microsoft.Extensions.Logging.LogLevel logLevel) => false; + public void Log(Microsoft.Extensions.Logging.LogLevel l, Microsoft.Extensions.Logging.EventId e, + TState s, Exception? ex, Func f) { } +} + +internal class RecordingSink : ISessionOutput +{ + public List Envelopes { get; } = new(); + public List Binary { get; } = new(); + public Task WriteEnvelopeAsync(JsonObject e, CancellationToken ct) + { + lock (Envelopes) Envelopes.Add(e); + _gate.Release(); + return Task.CompletedTask; + } + public Task WriteBinaryAsync(ReadOnlyMemory 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}"); + } +} diff --git a/backend/Realtime/RealtimeSession.cs b/backend/Realtime/RealtimeSession.cs new file mode 100644 index 0000000..f2c7edc --- /dev/null +++ b/backend/Realtime/RealtimeSession.cs @@ -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 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 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; + } + } +}