diff --git a/backend.tests/DeviceRegistryTests.cs b/backend.tests/DeviceRegistryTests.cs new file mode 100644 index 0000000..327783e --- /dev/null +++ b/backend.tests/DeviceRegistryTests.cs @@ -0,0 +1,25 @@ +using System.Net.WebSockets; +using backend.DeviceHub; +using FluentAssertions; +using Xunit; + +namespace backend.tests; + +public class DeviceRegistryTests +{ + [Fact] + public void Register_unregister_round_trip() + { + var reg = new DeviceRegistry(); + var id = Guid.NewGuid(); + var dev = new ActiveDevice(id, new ClientWebSocket()); + reg.TryRegister(dev).Should().BeTrue(); + reg.TryRegister(dev).Should().BeFalse(); + reg.IsOnline(id).Should().BeTrue(); + reg.OnlineCount.Should().Be(1); + + reg.Unregister(id).Should().BeTrue(); + reg.IsOnline(id).Should().BeFalse(); + reg.OnlineCount.Should().Be(0); + } +} diff --git a/backend.tests/FakeDeviceChannel.cs b/backend.tests/FakeDeviceChannel.cs index e972ea4..9f92d27 100644 --- a/backend.tests/FakeDeviceChannel.cs +++ b/backend.tests/FakeDeviceChannel.cs @@ -1,14 +1,28 @@ using System.Collections.Concurrent; using System.Text.Json; +using backend.DeviceHub; using backend.Tools; namespace backend.tests; -// Slim version for Tasks 9-15. Task 16 reopens this file to add IDeviceClient + envelope sending. -public class FakeDeviceChannel : IDeviceChannel +public class FakeDeviceChannel : IDeviceChannel, IDeviceClient { + public List SentEnvelopes { get; } = new(); + public List SentBinary { get; } = new(); public ConcurrentQueue PiToolReplies { get; } = new(); + public Task SendEnvelopeAsync(T envelope, CancellationToken ct) where T : HubEnvelope + { + SentEnvelopes.Add(envelope); + return Task.CompletedTask; + } + + public Task SendBinaryAsync(ReadOnlyMemory bytes, CancellationToken ct) + { + SentBinary.Add(bytes.ToArray()); + return Task.CompletedTask; + } + public Task CallPiToolAsync(string name, JsonElement args, CancellationToken ct) { if (PiToolReplies.TryDequeue(out var reply)) diff --git a/backend/DeviceHub/ActiveDevice.cs b/backend/DeviceHub/ActiveDevice.cs new file mode 100644 index 0000000..e0e791f --- /dev/null +++ b/backend/DeviceHub/ActiveDevice.cs @@ -0,0 +1,112 @@ +using System.Collections.Concurrent; +using System.Net.WebSockets; +using System.Text; +using System.Text.Json; +using System.Text.Json.Nodes; +using backend.Realtime; +using backend.Tools; + +namespace backend.DeviceHub; + +public interface IDeviceClient +{ + Task SendEnvelopeAsync(T envelope, CancellationToken ct) where T : HubEnvelope; + Task SendBinaryAsync(ReadOnlyMemory bytes, CancellationToken ct); +} + +public class ActiveDevice : IDeviceChannel, IDeviceClient, ISessionOutput, IAsyncDisposable +{ + public Guid DeviceId { get; } + public WebSocket Ws => _ws; + public RealtimeSession? CurrentSession { get; set; } + public CancellationTokenSource? SessionCts { get; set; } + + private readonly WebSocket _ws; + private readonly SemaphoreSlim _sendLock = new(1, 1); + private readonly ConcurrentDictionary> _pendingTools = new(); + + public ActiveDevice(Guid deviceId, WebSocket ws) + { + DeviceId = deviceId; + _ws = ws; + } + + public async Task SendEnvelopeAsync(T envelope, CancellationToken ct) where T : HubEnvelope + { + var json = JsonSerializer.Serialize(envelope, envelope.GetType()); + await _sendLock.WaitAsync(ct); + try + { + await _ws.SendAsync(Encoding.UTF8.GetBytes(json), + WebSocketMessageType.Text, endOfMessage: true, ct); + } + finally { _sendLock.Release(); } + } + + public async Task SendBinaryAsync(ReadOnlyMemory bytes, CancellationToken ct) + { + await _sendLock.WaitAsync(ct); + try + { + await _ws.SendAsync(bytes, WebSocketMessageType.Binary, endOfMessage: true, ct); + } + finally { _sendLock.Release(); } + } + + public Task WriteEnvelopeAsync(JsonObject envelope, CancellationToken ct) + { + var json = envelope.ToJsonString(); + return WriteRawTextAsync(json, ct); + } + + private async Task WriteRawTextAsync(string json, CancellationToken ct) + { + await _sendLock.WaitAsync(ct); + try + { + await _ws.SendAsync(Encoding.UTF8.GetBytes(json), + WebSocketMessageType.Text, endOfMessage: true, ct); + } + finally { _sendLock.Release(); } + } + + public Task WriteBinaryAsync(ReadOnlyMemory bytes, CancellationToken ct) + => SendBinaryAsync(bytes, ct); + + public Task CallPiToolAsync(string name, JsonElement args, CancellationToken ct) + { + var callId = "srv_" + Guid.NewGuid().ToString("N")[..12]; + var tcs = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + _pendingTools[callId] = tcs; + + var envelope = new ToolCallEnvelope("tool_call", callId, name, args); + _ = SendEnvelopeAsync(envelope, ct); + + ct.Register(() => tcs.TrySetCanceled()); + return tcs.Task; + } + + public void CompleteToolResult(ToolResultEnvelope env) + { + if (!_pendingTools.TryRemove(env.CallId, out var tcs)) return; + if (env.Ok && env.Result is JsonElement r) tcs.TrySetResult(r); + else + { + var msg = (env.Error ?? "unknown").Replace("\\", "\\\\").Replace("\"", "\\\""); + var doc = JsonDocument.Parse($$"""{"ok":false,"error":"{{msg}}"}"""); + tcs.TrySetResult(doc.RootElement.Clone()); + } + } + + public async ValueTask DisposeAsync() + { + try + { + if (_ws.State == WebSocketState.Open) + await _ws.CloseAsync(WebSocketCloseStatus.NormalClosure, "bye", CancellationToken.None); + } + catch { } + _ws.Dispose(); + _sendLock.Dispose(); + } +} diff --git a/backend/DeviceHub/DeviceRegistry.cs b/backend/DeviceHub/DeviceRegistry.cs new file mode 100644 index 0000000..c7db098 --- /dev/null +++ b/backend/DeviceHub/DeviceRegistry.cs @@ -0,0 +1,21 @@ +using System.Collections.Concurrent; + +namespace backend.DeviceHub; + +public class DeviceRegistry +{ + private readonly ConcurrentDictionary _online = new(); + + public bool TryRegister(ActiveDevice dev) + => _online.TryAdd(dev.DeviceId, dev); + + public bool Unregister(Guid deviceId) + => _online.TryRemove(deviceId, out _); + + public ActiveDevice? Get(Guid deviceId) + => _online.TryGetValue(deviceId, out var d) ? d : null; + + public bool IsOnline(Guid deviceId) => _online.ContainsKey(deviceId); + + public int OnlineCount => _online.Count; +} diff --git a/backend/DeviceHub/HubEnvelopes.cs b/backend/DeviceHub/HubEnvelopes.cs new file mode 100644 index 0000000..b7da0c6 --- /dev/null +++ b/backend/DeviceHub/HubEnvelopes.cs @@ -0,0 +1,42 @@ +using System.Text.Json; +using System.Text.Json.Serialization; + +namespace backend.DeviceHub; + +public record HubEnvelope([property: JsonPropertyName("type")] string Type); + +public record HelloEnvelope(string Type, + [property: JsonPropertyName("device_id")] Guid DeviceId, + [property: JsonPropertyName("client_version")] string ClientVersion) : HubEnvelope(Type); + +public record HelloAckEnvelope(string Type, + [property: JsonPropertyName("config")] JsonElement Config) : HubEnvelope(Type); + +public record PongEnvelope(string Type) : HubEnvelope(Type); + +public record WakeEnvelope(string Type, + [property: JsonPropertyName("at")] long? At) : HubEnvelope(Type); + +public record SessionStartedEnvelope(string Type, + [property: JsonPropertyName("conversation_id")] Guid ConversationId) : HubEnvelope(Type); + +public record SessionEndedEnvelope(string Type, + [property: JsonPropertyName("reason")] string Reason) : HubEnvelope(Type); + +public record AssistantDoneEnvelope(string Type) : HubEnvelope(Type); + +public record ToolCallEnvelope(string Type, + [property: JsonPropertyName("call_id")] string CallId, + [property: JsonPropertyName("name")] string Name, + [property: JsonPropertyName("arguments")] JsonElement Arguments) : HubEnvelope(Type); + +public record ToolResultEnvelope(string Type, + [property: JsonPropertyName("call_id")] string CallId, + [property: JsonPropertyName("ok")] bool Ok, + [property: JsonPropertyName("result")] JsonElement? Result, + [property: JsonPropertyName("error")] string? Error) : HubEnvelope(Type); + +public record ErrorEnvelope(string Type, + [property: JsonPropertyName("code")] string Code, + [property: JsonPropertyName("message")] string Message, + [property: JsonPropertyName("fatal")] bool? Fatal) : HubEnvelope(Type); diff --git a/backend/Program.cs b/backend/Program.cs index 3af97c9..02fa7ca 100644 --- a/backend/Program.cs +++ b/backend/Program.cs @@ -40,6 +40,7 @@ builder.Services.AddSingleton builder.Services.AddSingleton(); builder.Services.AddSingleton(); builder.Services.AddScoped(); +builder.Services.AddSingleton(); var app = builder.Build();