DeviceHub: wake → session lifecycle + uplink relay + tool_result routing

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<Program>
             cfg.AddInMemoryCollection(new Dictionary<string, string?>
             {
                 ["ConnectionStrings:Default"] = $"Data Source={_dbPath}",
+                ["OPENAI_API_KEY"] = "test-key",
             });
         });
+
+        builder.ConfigureTestServices(services =>
+        {
+            services.AddScoped<RealtimeSessionFactory, FakeRealtimeSessionFactory>();
+        });
     }
 
     protected override void Dispose(bool disposing)
@@ -32,3 +42,24 @@ public class ApiFactory : WebApplicationFactory<Program>
         }
     }
 }
+
+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<RealtimeSession> 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<RealtimeSession>());
+        return Task.FromResult(s);
+    }
+}

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<ApiFactory>
+{
+    private readonly ApiFactory _factory = factory;
+
+    private async Task<string> 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<Dictionary<string, object>>())!["code"].ToString()!;
+        var pairRes = await _factory.CreateClient().PostAsJsonAsync("/api/pair",
+            new { code, hostname = "smoke", client_version = "0.1" });
+        var pairBody = await pairRes.Content.ReadFromJsonAsync<Dictionary<string, object>>();
+        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<JsonElement> 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();
+    }
+}

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;
+                if (msg.Length == 3840 && active.CurrentSession is not null)
-                // Binary routing into a session lives in Task 18.
+                {
+                    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<ToolResultEnvelope>(text)!;
+                    active.CompleteToolResult(env);
+                    break;
             }
         }
     }
 
+    private static async Task StartSessionAsync(
+        ActiveDevice active, Device device,
+        RealtimeSessionFactory sessions, IServiceScopeFactory scopes,
+        CancellationToken parentCt)
+    {
+        var enabledTools = JsonSerializer.Deserialize<string[]>(device.Config?.EnabledToolsJson ?? "[]")
+            ?? Array.Empty<string>();
+        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<string>(enabledTools));
+
+        active.SessionCts = CancellationTokenSource.CreateLinkedTokenSource(parentCt);
+        var ct = active.SessionCts.Token;
+
+        var sessionScope = scopes.CreateAsyncScope();
+        var scopedSessions = sessionScope.ServiceProvider.GetRequiredService<RealtimeSessionFactory>();
+
+        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 ?? "[]";