Support DTMF in realtime

src/Infrastructure/BotSharp.Abstraction/Realtime/Models/RealtimeHubConnection.cs

@@ -9,6 +9,7 @@ public class RealtimeHubConnection
     public string? LastAssistantItem { get; set; } = null!;
     public long LatestMediaTimestamp { get; set; }
     public long? ResponseStartTimestamp { get; set; }
+    public string KeypadInputBuffer { get; set; } = string.Empty;
     public ConcurrentQueue<string> MarkQueue { get; set; } = new();
     public string CurrentAgentId { get; set; } = null!;
     public string ConversationId { get; set; } = null!;

src/Infrastructure/BotSharp.Core/Realtime/RealtimeHub.cs

@@ -4,6 +4,10 @@ using BotSharp.Abstraction.Realtime.Models;
 using BotSharp.Abstraction.MLTasks;
 using BotSharp.Abstraction.Conversations.Enums;
 using BotSharp.Abstraction.Routing.Models;
+using NetTopologySuite.Index.HPRtree;
+using BotSharp.Abstraction.Agents.Models;
+using Microsoft.Identity.Client.Extensions.Msal;
+using Microsoft.AspNetCore.Cors.Infrastructure;
 
 namespace BotSharp.Core.Realtime;
 
@@ -46,9 +50,14 @@ public class RealtimeHub : IRealtimeHub
             {
                 await completer.AppenAudioBuffer(conn.Data);
             }
+            else if (conn.Event == "user_dtmf_received")
+            {
+                await HandleUserDtmfReceived(completer, conn);
+            }
             else if (conn.Event == "user_disconnected")
             {
                 await completer.Disconnect();
+                await HandleUserDisconnected(conn);
             }
         } while (!result.CloseStatus.HasValue);
 
@@ -58,8 +67,6 @@ public class RealtimeHub : IRealtimeHub
     private async Task ConnectToModel(IRealTimeCompletion completer, WebSocket userWebSocket, RealtimeHubConnection conn)
     {
         var hookProvider = _services.GetRequiredService<ConversationHookProvider>();
-        var storage = _services.GetRequiredService<IConversationStorage>();
-
         var convService = _services.GetRequiredService<IConversationService>();
         convService.SetConversationId(conn.ConversationId, []);
         var conversation = await convService.GetConversation(conn.ConversationId);
@@ -183,7 +190,6 @@ public class RealtimeHub : IRealtimeHub
                     else
                     {
                         // append output audio transcript to conversation
-                        storage.Append(conn.ConversationId, message);
                         dialogs.Add(message);
 
                         foreach (var hook in hookProvider.HooksOrderByPriority)
@@ -203,7 +209,6 @@ public class RealtimeHub : IRealtimeHub
             onInputAudioTranscriptionCompleted: async message =>
             {
                 // append input audio transcript to conversation
-                storage.Append(conn.ConversationId, message);
                 dialogs.Add(message);
 
                 foreach (var hook in hookProvider.HooksOrderByPriority)
@@ -226,6 +231,46 @@ public class RealtimeHub : IRealtimeHub
             });
     }
 
+    private async Task HandleUserDtmfReceived(IRealTimeCompletion completer, RealtimeHubConnection conn)
+    {
+        var routing = _services.GetRequiredService<IRoutingService>();
+        var hookProvider = _services.GetRequiredService<ConversationHookProvider>();
+        var agentService = _services.GetRequiredService<IAgentService>();
+        var agent = await agentService.LoadAgent(conn.CurrentAgentId);
+        var dialogs = routing.Context.GetDialogs();
+        var convService = _services.GetRequiredService<IConversationService>();
+        var conversation = await convService.GetConversation(conn.ConversationId);
+
+        var message = new RoleDialogModel(AgentRole.User, conn.Data)
+        {
+            CurrentAgentId = routing.Context.GetCurrentAgentId()
+        };
+        dialogs.Add(message);
+
+        foreach (var hook in hookProvider.HooksOrderByPriority)
+        {
+            hook.SetAgent(agent)
+                .SetConversation(conversation);
+
+            await hook.OnMessageReceived(message);
+        }
+
+        await completer.InsertConversationItem(message);
+        await completer.TriggerModelInference("Reply based on the user input");
+    }
+
+    private async Task HandleUserDisconnected(RealtimeHubConnection conn)
+    {
+        // Save dialog history
+        var routing = _services.GetRequiredService<IRoutingService>();
+        var storage = _services.GetRequiredService<IConversationStorage>();
+        var dialogs = routing.Context.GetDialogs();
+        foreach (var item in dialogs)
+        {
+            storage.Append(conn.ConversationId, item);
+        }
+    }
+
     private async Task SendEventToUser(WebSocket webSocket, object message)
     {
         var data = JsonSerializer.Serialize(message);

src/Plugins/BotSharp.Plugin.OpenAI/Providers/Realtime/RealTimeCompletionProvider.cs

@@ -75,7 +75,10 @@ public class RealTimeCompletionProvider : IRealTimeCompletion
 
     public async Task Disconnect()
     {
-        await _webSocket.CloseAsync(WebSocketCloseStatus.Empty, null, CancellationToken.None);
+        if (_webSocket.State == WebSocketState.Open)
+        {
+            await _webSocket.CloseAsync(WebSocketCloseStatus.Empty, null, CancellationToken.None);
+        }
     }
 
     public async Task AppenAudioBuffer(string message)

src/Plugins/BotSharp.Plugin.Twilio/Models/Stream/StreamEventDtmfResponse.cs

@@ -0,0 +1,17 @@
+using BotSharp.Plugin.Twilio.Models.Stream;
+using System.Text.Json.Serialization;
+
+public class StreamEventDtmfResponse : StreamEventResponse
+{
+    [JsonPropertyName("dtmf")]
+    public StreamEventDtmfBody Body { get; set; }
+}
+
+public class StreamEventDtmfBody
+{
+    [JsonPropertyName("track")]
+    public string Track { get; set; }
+
+    [JsonPropertyName("digit")]
+    public string Digit { get; set; }
+}

src/Plugins/BotSharp.Plugin.Twilio/Services/Stream/TwilioStreamMiddleware.cs

@@ -122,6 +122,20 @@ public class TwilioStreamMiddleware
                 conn.LatestMediaTimestamp = long.Parse(mediaResponse.Body.Timestamp);
                 conn.Data = mediaResponse.Body.Payload;
             }
+            else if (response.Event == "dtmf")
+            {
+                var dtmfResponse = JsonSerializer.Deserialize<StreamEventDtmfResponse>(receivedText);
+                if (dtmfResponse.Body.Digit == "#")
+                {
+                    conn.Event = "user_dtmf_received";
+                    conn.Data = conn.KeypadInputBuffer;
+                    conn.KeypadInputBuffer = string.Empty;
+                }
+                else
+                {
+                    conn.KeypadInputBuffer += dtmfResponse.Body.Digit;
+                }
+            }
 
             return conn;
         });