diff --git a/backend.tests/ApiFactory.cs b/backend.tests/ApiFactory.cs index 9d0d274..dd2cd34 100644 --- a/backend.tests/ApiFactory.cs +++ b/backend.tests/ApiFactory.cs @@ -1,7 +1,11 @@ +using backend.Realtime; using Microsoft.AspNetCore.Hosting; using Microsoft.AspNetCore.Mvc.Testing; +using Microsoft.AspNetCore.TestHost; using Microsoft.Data.Sqlite; using Microsoft.Extensions.Configuration; +using Microsoft.Extensions.DependencyInjection; +using Microsoft.Extensions.Logging; namespace backend.tests; @@ -17,8 +21,14 @@ public class ApiFactory : WebApplicationFactory cfg.AddInMemoryCollection(new Dictionary { ["ConnectionStrings:Default"] = $"Data Source={_dbPath}", + ["OPENAI_API_KEY"] = "test-key", }); }); + + builder.ConfigureTestServices(services => + { + services.AddScoped(); + }); } protected override void Dispose(bool disposing) @@ -32,3 +42,24 @@ public class ApiFactory : WebApplicationFactory } } } + +public class FakeRealtimeSessionFactory( + OpenAIKeyProvider keys, + backend.Tools.ToolRegistry registry, + backend.Conversations.ConversationLog log, + ILoggerFactory loggerFactory) + : RealtimeSessionFactory(keys, registry, log, loggerFactory) +{ + public static ScriptedRealtimeUpstream Upstream { get; set; } = new(); + + public override Task CreateAsync( + Guid deviceId, RealtimeSettings cfg, ISessionOutput output, + backend.Tools.IDeviceChannel channel, CancellationToken ct) + { + Upstream.Push(new System.Text.Json.Nodes.JsonObject { ["type"] = "session.updated" }); + var s = new RealtimeSession( + deviceId, Upstream, output, log, registry, cfg, channel, + loggerFactory.CreateLogger()); + return Task.FromResult(s); + } +} diff --git a/backend.tests/DeviceHubWakeFlowTests.cs b/backend.tests/DeviceHubWakeFlowTests.cs new file mode 100644 index 0000000..8d013fc --- /dev/null +++ b/backend.tests/DeviceHubWakeFlowTests.cs @@ -0,0 +1,117 @@ +using System.Net.Http.Json; +using System.Net.WebSockets; +using System.Text; +using System.Text.Json; +using System.Text.Json.Nodes; +using FluentAssertions; +using Xunit; + +namespace backend.tests; + +[Collection("WakeFlow")] // tests touch FakeRealtimeSessionFactory.Upstream static — serialize +public class DeviceHubWakeFlowTests(ApiFactory factory) : IClassFixture +{ + private readonly ApiFactory _factory = factory; + + private async Task PairAndGetTokenAsync(string email) + { + var http = _factory.CreateClient(); + await http.PostAsJsonAsync("/api/auth/register", + new { email, password = "Passw0rd!" }); + await http.PostAsJsonAsync("/api/auth/login", + new { email, password = "Passw0rd!" }); + var codeRes = await http.PostAsJsonAsync("/api/pair-code", new { name = "wake-test" }); + var code = (await codeRes.Content.ReadFromJsonAsync>())!["code"].ToString()!; + var pairRes = await _factory.CreateClient().PostAsJsonAsync("/api/pair", + new { code, hostname = "smoke", client_version = "0.1" }); + var pairBody = await pairRes.Content.ReadFromJsonAsync>(); + return pairBody!["device_token"].ToString()!; + } + + [Fact] + public async Task Wake_opens_session_started_then_assistant_done_then_session_ended_on_cancel() + { + FakeRealtimeSessionFactory.Upstream = new ScriptedRealtimeUpstream(); + var token = await PairAndGetTokenAsync("wake1@example.com"); + var client = _factory.Server.CreateWebSocketClient(); + client.ConfigureRequest = req => req.Headers["Authorization"] = $"Bearer {token}"; + var uri = new Uri(_factory.Server.BaseAddress, "device"); + var ws = await client.ConnectAsync(uri, CancellationToken.None); + + await SendJson(ws, new { type = "hello", device_id = Guid.NewGuid(), client_version = "0.1" }); + await ReceiveJson(ws); // ack + + await SendJson(ws, new { type = "wake", at = 0L }); + var started = await ReceiveJson(ws); + started.GetProperty("type").GetString().Should().Be("session_started"); + + FakeRealtimeSessionFactory.Upstream.Push(new JsonObject + { + ["type"] = "response.audio_transcript.done", + ["transcript"] = "hello!", + }); + FakeRealtimeSessionFactory.Upstream.Push(new JsonObject { ["type"] = "response.done" }); + + var done = await ReceiveJson(ws); + done.GetProperty("type").GetString().Should().Be("assistant_done"); + + await SendJson(ws, new { type = "cancel" }); + var ended = await ReceiveJson(ws); + ended.GetProperty("type").GetString().Should().Be("session_ended"); + } + + [Fact] + public async Task Binary_uplink_after_wake_reaches_upstream_as_input_audio_buffer_append() + { + FakeRealtimeSessionFactory.Upstream = new ScriptedRealtimeUpstream(); + var token = await PairAndGetTokenAsync("wake2@example.com"); + var client = _factory.Server.CreateWebSocketClient(); + client.ConfigureRequest = req => req.Headers["Authorization"] = $"Bearer {token}"; + var uri = new Uri(_factory.Server.BaseAddress, "device"); + var ws = await client.ConnectAsync(uri, CancellationToken.None); + + await SendJson(ws, new { type = "hello", device_id = Guid.NewGuid(), client_version = "0.1" }); + await ReceiveJson(ws); + await SendJson(ws, new { type = "wake", at = 0L }); + await ReceiveJson(ws); // session_started + + var frame = new byte[3840]; + for (int i = 0; i < frame.Length; i++) frame[i] = 0x55; + await ws.SendAsync(frame, WebSocketMessageType.Binary, true, CancellationToken.None); + + // Allow time for the pump's 250 ms poll + send. + var deadline = DateTime.UtcNow + TimeSpan.FromSeconds(2); + while (DateTime.UtcNow < deadline) + { + if (FakeRealtimeSessionFactory.Upstream.Sent.Any(e => + (string?)e["type"] == "input_audio_buffer.append")) break; + await Task.Delay(50); + } + FakeRealtimeSessionFactory.Upstream.Sent + .Any(e => (string?)e["type"] == "input_audio_buffer.append") + .Should().BeTrue(); + + await SendJson(ws, new { type = "cancel" }); + await ReceiveJson(ws); + } + + private static async Task SendJson(WebSocket ws, object payload) + { + var bytes = Encoding.UTF8.GetBytes(JsonSerializer.Serialize(payload)); + await ws.SendAsync(bytes, WebSocketMessageType.Text, true, CancellationToken.None); + } + + private static async Task ReceiveJson(WebSocket ws) + { + var buf = new byte[16384]; + using var stream = new MemoryStream(); + while (true) + { + var r = await ws.ReceiveAsync(buf, CancellationToken.None); + if (r.MessageType == WebSocketMessageType.Binary) { stream.Write(buf, 0, r.Count); continue; } + stream.Write(buf, 0, r.Count); + if (r.EndOfMessage) break; + } + return JsonDocument.Parse(stream.ToArray()).RootElement.Clone(); + } +} diff --git a/backend/DeviceHub/DeviceHubEndpoint.cs b/backend/DeviceHub/DeviceHubEndpoint.cs index 98f4c9d..62e1b54 100644 --- a/backend/DeviceHub/DeviceHubEndpoint.cs +++ b/backend/DeviceHub/DeviceHubEndpoint.cs @@ -2,9 +2,13 @@ using System.Net.WebSockets; using System.Text; using System.Text.Json; using System.Text.Json.Nodes; +using backend.Conversations; using backend.Data; using backend.Devices; +using backend.Realtime; +using backend.Tools; using Microsoft.EntityFrameworkCore; +using Microsoft.Extensions.DependencyInjection; namespace backend.DeviceHub; @@ -12,7 +16,13 @@ public static class DeviceHubEndpoint { public static IEndpointRouteBuilder MapDeviceHub(this IEndpointRouteBuilder app) { - app.Map("/device", async (HttpContext ctx, DeviceAuth auth, DeviceRegistry reg, AppDbContext db) => + app.Map("/device", async ( + HttpContext ctx, + DeviceAuth auth, + DeviceRegistry reg, + AppDbContext db, + RealtimeSessionFactory sessions, + IServiceScopeFactory scopes) => { if (!ctx.WebSockets.IsWebSocketRequest) { @@ -39,7 +49,7 @@ public static class DeviceHubEndpoint try { - await HandleAsync(active, device, db, ctx.RequestAborted); + await HandleAsync(active, device, db, sessions, scopes, ctx.RequestAborted); } finally { @@ -49,7 +59,10 @@ public static class DeviceHubEndpoint return app; } - private static async Task HandleAsync(ActiveDevice active, Device device, AppDbContext db, CancellationToken ct) + private static async Task HandleAsync( + ActiveDevice active, Device device, AppDbContext db, + RealtimeSessionFactory sessions, IServiceScopeFactory scopes, + CancellationToken ct) { var buffer = new byte[8192]; var ws = active.Ws; @@ -68,8 +81,10 @@ public static class DeviceHubEndpoint if (res.MessageType == WebSocketMessageType.Binary) { - if (msg.Length != 3840) continue; - // Binary routing into a session lives in Task 18. + if (msg.Length == 3840 && active.CurrentSession is not null) + { + await active.CurrentSession.WriteUplinkAsync(msg.ToArray(), ct); + } continue; } @@ -87,10 +102,75 @@ public static class DeviceHubEndpoint await db.SaveChangesAsync(ct); await active.SendEnvelopeAsync(new PongEnvelope("pong"), ct); break; + + case "wake": + if (active.CurrentSession is null) + await StartSessionAsync(active, device, sessions, scopes, ct); + break; + + case "cancel": + active.SessionCts?.Cancel(); + break; + + case "playback_done": + break; // v1: client coordination only + + case "tool_result": + var env = JsonSerializer.Deserialize(text)!; + active.CompleteToolResult(env); + break; } } } + private static async Task StartSessionAsync( + ActiveDevice active, Device device, + RealtimeSessionFactory sessions, IServiceScopeFactory scopes, + CancellationToken parentCt) + { + var enabledTools = JsonSerializer.Deserialize(device.Config?.EnabledToolsJson ?? "[]") + ?? Array.Empty(); + var cfg = new RealtimeSettings( + Model: device.Config?.Model ?? "gpt-4o-realtime-preview", + Voice: device.Config?.Voice ?? "alloy", + SystemPrompt: device.Config?.SystemPrompt ?? "", + IdleTimeoutSeconds: device.Config?.IdleTimeoutSeconds ?? 30, + EnabledTools: new HashSet(enabledTools)); + + active.SessionCts = CancellationTokenSource.CreateLinkedTokenSource(parentCt); + var ct = active.SessionCts.Token; + + var sessionScope = scopes.CreateAsyncScope(); + var scopedSessions = sessionScope.ServiceProvider.GetRequiredService(); + + RealtimeSession session; + try + { + session = await scopedSessions.CreateAsync( + device.Id, cfg, active, active, ct); + } + catch (Exception ex) + { + await active.SendEnvelopeAsync( + new ErrorEnvelope("error", "upstream", ex.Message, true), parentCt); + await sessionScope.DisposeAsync(); + return; + } + + active.CurrentSession = session; + _ = Task.Run(async () => + { + try { await session.RunAsync(ct); } + finally + { + active.CurrentSession = null; + active.SessionCts?.Dispose(); + active.SessionCts = null; + await sessionScope.DisposeAsync(); + } + }, ct); + } + private static async Task SendHelloAckAsync(ActiveDevice active, Device device, CancellationToken ct) { var enabled = device.Config?.EnabledToolsJson ?? "[]";