From bf039def7867682df0809b061f5e518484d35138 Mon Sep 17 00:00:00 2001 From: "mars.yu" Date: Tue, 20 Oct 2026 14:47:35 +0800 Subject: [PATCH 01/10] fix(openai): bound text-to-speech requests with a per-attempt timeout GenerateSpeechAsync used the SDK defaults: 100s per attempt and 3 retries. In production some /v1/audio/speech requests never return, so a failed call blocked the caller for about 7 minutes ("Retry failed after 4 tries"). AudioSynthesisProvider now waits at most 30s per attempt and keeps 3 retries. Both can be set per model under Audio.Synthesis (TimeoutSeconds, MaxRetries). ProviderHelper.GetClient takes the timeout and retry count as optional arguments; other callers keep the SDK defaults. Verified against a local server that never answers: 4 attempts about 30s apart, failing after 125.7s (was ~400s); a hang followed by a good response recovers on the retry. Co-Authored-By: Claude Opus 5.5 --- .../MLTasks/Settings/LlmModelSetting.cs | 17 ++++++++++++++++ .../Providers/Audio/AudioSynthesisProvider.cs | 13 +++++++++++- .../Providers/ProviderHelper.cs | 20 ++++++++++++++++--- 3 files changed, 46 insertions(+), 4 deletions(-) diff --git a/src/Infrastructure/BotSharp.Abstraction/MLTasks/Settings/LlmModelSetting.cs b/src/Infrastructure/BotSharp.Abstraction/MLTasks/Settings/LlmModelSetting.cs index 111442367..dccd20377 100644 --- a/src/Infrastructure/BotSharp.Abstraction/MLTasks/Settings/LlmModelSetting.cs +++ b/src/Infrastructure/BotSharp.Abstraction/MLTasks/Settings/LlmModelSetting.cs @@ -154,6 +154,23 @@ public class ImageVariationSetting public class AudioSetting { public AudioTranscriptionSetting? Transcription { get; set; } + public AudioSynthesisSetting? Synthesis { get; set; } +} + +/// +/// Audio synthesis (text-to-speech) setting +/// +public class AudioSynthesisSetting +{ + /// + /// Timeout of a single request attempt, in seconds + /// + public int? TimeoutSeconds { get; set; } + + /// + /// Number of retries after the first attempt fails or times out + /// + public int? MaxRetries { get; set; } } /// diff --git a/src/Plugins/BotSharp.Plugin.OpenAI/Providers/Audio/AudioSynthesisProvider.cs b/src/Plugins/BotSharp.Plugin.OpenAI/Providers/Audio/AudioSynthesisProvider.cs index 10341abe3..6770fd037 100644 --- a/src/Plugins/BotSharp.Plugin.OpenAI/Providers/Audio/AudioSynthesisProvider.cs +++ b/src/Plugins/BotSharp.Plugin.OpenAI/Providers/Audio/AudioSynthesisProvider.cs @@ -21,9 +21,20 @@ public void SetModelName(string model) _model = model; } + // A short phrase normally comes back in a few seconds, but some requests never finish; + // with the SDK default of 100s per attempt, four stalled attempts block the caller for ~7 minutes. + private const int DefaultTimeoutSeconds = 30; + private const int DefaultMaxRetries = 3; + public async Task GenerateAudioAsync(string text, string? voice = "alloy", string? format = "mp3", string? instructions = null) { - var audioClient = ProviderHelper.GetClient(Provider, _model, apiKey: null, _services) + var settingsService = _services.GetRequiredService(); + var synthesis = settingsService.GetSetting(Provider, _model)?.Audio?.Synthesis; + var timeout = TimeSpan.FromSeconds(synthesis?.TimeoutSeconds ?? DefaultTimeoutSeconds); + var maxRetries = synthesis?.MaxRetries ?? DefaultMaxRetries; + + var audioClient = ProviderHelper.GetClient(Provider, _model, apiKey: null, _services, + networkTimeout: timeout, maxRetries: maxRetries) .GetAudioClient(_model); var (speechVoice, options) = PrepareGenerationOptions(voice: voice, format: format); diff --git a/src/Plugins/BotSharp.Plugin.OpenAI/Providers/ProviderHelper.cs b/src/Plugins/BotSharp.Plugin.OpenAI/Providers/ProviderHelper.cs index df937390d..7900afd2d 100644 --- a/src/Plugins/BotSharp.Plugin.OpenAI/Providers/ProviderHelper.cs +++ b/src/Plugins/BotSharp.Plugin.OpenAI/Providers/ProviderHelper.cs @@ -1,16 +1,30 @@ using OpenAI; using System.ClientModel; +using System.ClientModel.Primitives; namespace BotSharp.Plugin.OpenAI.Providers; public class ProviderHelper { - public static OpenAIClient GetClient(string provider, string model, string? apiKey, IServiceProvider services) + public static OpenAIClient GetClient(string provider, string model, string? apiKey, IServiceProvider services, + TimeSpan? networkTimeout = null, int? maxRetries = null) { var settingsService = services.GetRequiredService(); var settings = settingsService.GetSetting(provider, model); - var options = !string.IsNullOrEmpty(settings?.Endpoint) ? - new OpenAIClientOptions { Endpoint = new Uri(settings.Endpoint) } : null; + var options = new OpenAIClientOptions(); + if (!string.IsNullOrEmpty(settings?.Endpoint)) + { + options.Endpoint = new Uri(settings.Endpoint); + } + // Left unset, the SDK defaults apply: 100s per attempt and 3 retries. + if (networkTimeout.HasValue) + { + options.NetworkTimeout = networkTimeout.Value; + } + if (maxRetries.HasValue) + { + options.RetryPolicy = new ClientRetryPolicy(maxRetries.Value); + } return new OpenAIClient(new ApiKeyCredential(apiKey ?? settings!.ApiKey), options); } From 8beb5f76d69994fe05ff76108074b8c73e9c80b2 Mon Sep 17 00:00:00 2001 From: "mars.yu" Date: Tue, 20 Oct 2026 14:47:36 +0800 Subject: [PATCH 02/10] feat(twilio): speak with TwiML Say and add Twilio ConversationRelay Say mode (Twilio:SayGeneratedSpeech, off by default) - Generated speech is saved as a text file and spoken with instead of pre-generating an mp3 with the audio synthesizer: the opening line (no longer synthesized before dialing), replies, hang-up, transfer, voicemail and tool-call indications. - Prerecorded prompts (twilio/{name}.mp3) are spoken from Twilio:PromptTexts so the whole call uses one voice; prompts without text keep playing. - TwilioService.SaveSpeechAsync / AppendSpeech are the single places that write and play speech. With the switch off, behaviour is unchanged. ConversationRelay (Twilio:ConversationRelay, off by default) - Outbound calls for enabled agents (AgentIds, empty = all) are answered with : Twilio transcribes and speaks, and exchanges text with TwilioConversationRelayMiddleware at /twilio/relay/{agent}/{conv}. - ConversationRelaySession runs one agent turn at a time (later words are merged into the next turn), speaks tool-call indications, re-prompts on caller silence, hangs up after MaxIdlePrompts or AiResponseTimeoutSeconds, and ends with handoffData hangup/transfer. - twilio/relay/action dials the CSR on transfer, hangs up otherwise, and reconnects a failed session up to MaxReconnects times. - The WebSocket handshake is checked against X-Twilio-Signature. - The caller-message handling used by the webhook queue moved to TwilioConversationProcessor so both flows share it. Tested with in-process harnesses: 24 TwiML checks for Say mode and 28 checks driving the relay WebSocket (turns, merging, DTMF, indications, silence, timeouts, signature, TwiML, action callback). Co-Authored-By: Claude Opus 5.5 --- .../TwilioConversationRelayController.cs | 78 +++ .../Controllers/TwilioInboundController.cs | 4 +- .../Controllers/TwilioOutboundController.cs | 15 +- .../Controllers/TwilioReconnectController.cs | 3 +- .../Relay/ConversationRelayActionRequest.cs | 33 ++ .../Relay/ConversationRelayParameters.cs | 18 + .../Functions/HangupPhoneCallFn.cs | 9 +- .../Functions/LeaveVoicemailFn.cs | 9 +- .../Functions/OutboundPhoneCallFn.cs | 24 +- .../Functions/TransferPhoneCallFn.cs | 8 +- .../Services/ConversationRelaySession.cs | 471 ++++++++++++++++++ .../Services/TwilioConversationProcessor.cs | 103 ++++ .../Services/TwilioMessageQueueService.cs | 94 +--- .../Services/TwilioService.cs | 199 +++++++- .../Settings/ConversationRelaySetting.cs | 70 +++ .../Settings/TwilioSetting.cs | 18 + .../TwilioConversationRelayMiddleware.cs | 85 ++++ 17 files changed, 1098 insertions(+), 143 deletions(-) create mode 100644 src/Plugins/BotSharp.Plugin.Twilio/Controllers/TwilioConversationRelayController.cs create mode 100644 src/Plugins/BotSharp.Plugin.Twilio/Models/Relay/ConversationRelayActionRequest.cs create mode 100644 src/Plugins/BotSharp.Plugin.Twilio/Models/Relay/ConversationRelayParameters.cs create mode 100644 src/Plugins/BotSharp.Plugin.Twilio/Services/ConversationRelaySession.cs create mode 100644 src/Plugins/BotSharp.Plugin.Twilio/Services/TwilioConversationProcessor.cs create mode 100644 src/Plugins/BotSharp.Plugin.Twilio/Settings/ConversationRelaySetting.cs create mode 100644 src/Plugins/BotSharp.Plugin.Twilio/TwilioConversationRelayMiddleware.cs diff --git a/src/Plugins/BotSharp.Plugin.Twilio/Controllers/TwilioConversationRelayController.cs b/src/Plugins/BotSharp.Plugin.Twilio/Controllers/TwilioConversationRelayController.cs new file mode 100644 index 000000000..92cab7f4f --- /dev/null +++ b/src/Plugins/BotSharp.Plugin.Twilio/Controllers/TwilioConversationRelayController.cs @@ -0,0 +1,78 @@ +using BotSharp.Abstraction.Infrastructures; +using BotSharp.Core.Infrastructures; +using BotSharp.Plugin.Twilio.Interfaces; +using BotSharp.Plugin.Twilio.Models.Relay; +using BotSharp.Plugin.Twilio.Services; +using Microsoft.AspNetCore.Mvc; + +namespace BotSharp.Plugin.Twilio.Controllers; + +public class TwilioConversationRelayController : TwilioController +{ + private readonly TwilioSetting _settings; + private readonly IServiceProvider _services; + private readonly ILogger _logger; + + public TwilioConversationRelayController(TwilioSetting settings, IServiceProvider services, ILogger logger) + { + _settings = settings; + _services = services; + _logger = logger; + } + + /// + /// Twilio calls this when a ConversationRelay session ends: we sent "end" (handoff), the caller hung up, or the session failed. + /// + [ValidateRequest] + [HttpPost("twilio/relay/action")] + public async Task SessionEnded(ConversationRelayActionRequest request) + { + _logger.LogInformation($"ConversationRelay session for conversation {request.ConversationId} ended: {request.SessionStatus} {request.ErrorCode} {request.ErrorMessage}"); + + // Authenticate (synchronous, so identity writes survive into the async session hooks) + HookEmitter.Emit(_services, hook => hook.OnAuthenticate(request), request.AgentId); + + var twilio = _services.GetRequiredService(); + var action = GetHandoffAction(request.HandoffData); + + if (action == ConversationRelayHandoff.Transfer && !string.IsNullOrEmpty(_settings.CsrAgentNumber)) + { + await HookEmitter.Emit(_services, async hook => await hook.OnAgentTransferring(request, _settings), request.AgentId); + return TwiML(twilio.DialCsrAgent(string.Empty)); + } + + var callStillUp = !string.Equals(request.CallStatus, "completed", StringComparison.OrdinalIgnoreCase); + if (request.SessionStatus == "failed" && callStillUp && request.Reconnects < _settings.ConversationRelay.MaxReconnects) + { + _logger.LogWarning($"Reconnecting ConversationRelay for conversation {request.ConversationId} (attempt {request.Reconnects + 1})."); + return TwiML(twilio.ReturnConversationRelay(request.AgentId, request.ConversationId, welcomeGreeting: null, request.Reconnects + 1)); + } + + if (action == ConversationRelayHandoff.HangUp) + { + await HookEmitter.Emit(_services, async hook => await hook.OnAgentHangUp(request), request.AgentId); + } + + var response = new VoiceResponse(); + response.Hangup(); + return TwiML(response); + } + + private static string? GetHandoffAction(string? handoffData) + { + if (string.IsNullOrWhiteSpace(handoffData)) + { + return null; + } + + try + { + using var doc = JsonDocument.Parse(handoffData); + return doc.RootElement.TryGetProperty("action", out var action) ? action.GetString() : null; + } + catch (JsonException) + { + return handoffData; + } + } +} diff --git a/src/Plugins/BotSharp.Plugin.Twilio/Controllers/TwilioInboundController.cs b/src/Plugins/BotSharp.Plugin.Twilio/Controllers/TwilioInboundController.cs index a63e776d7..7c1f083e4 100644 --- a/src/Plugins/BotSharp.Plugin.Twilio/Controllers/TwilioInboundController.cs +++ b/src/Plugins/BotSharp.Plugin.Twilio/Controllers/TwilioInboundController.cs @@ -84,8 +84,8 @@ await HookEmitter.Emit(_services, async hook => await HookEmitter.Emit(_services, async hook => await hook.OnVoicemailStarting(request), request.AgentId); - var url = twilio.GetSpeechPath(request.ConversationId, "voicemail.mp3"); - response.Play(new Uri(url)); + var voicemail = twilio.GetSavedSpeechFile(request.ConversationId, "voicemail"); + twilio.AppendSpeech(response, request.ConversationId, voicemail); } else { diff --git a/src/Plugins/BotSharp.Plugin.Twilio/Controllers/TwilioOutboundController.cs b/src/Plugins/BotSharp.Plugin.Twilio/Controllers/TwilioOutboundController.cs index f355d77ce..96ca2788e 100644 --- a/src/Plugins/BotSharp.Plugin.Twilio/Controllers/TwilioOutboundController.cs +++ b/src/Plugins/BotSharp.Plugin.Twilio/Controllers/TwilioOutboundController.cs @@ -38,8 +38,19 @@ public async Task InitiateOutboundCall(ConversationalVoiceRequest r await HookEmitter.Emit(_services, async hook => await hook.OnVoicemailStarting(request), request.AgentId); - var url = twilio.GetSpeechPath(request.ConversationId, "voicemail.mp3"); - response.Play(new Uri(url)); + var voicemail = twilio.GetSavedSpeechFile(request.ConversationId, "voicemail"); + twilio.AppendSpeech(response, request.ConversationId, voicemail); + } + else if (_settings.ConversationRelay.IsEnabledFor(request.AgentId)) + { + // OutboundPhoneCallFn stored the initial message as reply 0 + var sessionManager = _services.GetRequiredService(); + var initial = await sessionManager.GetAssistantReplyAsync(request.ConversationId, 0); + if (string.IsNullOrWhiteSpace(initial?.Content)) + { + _logger.LogWarning($"No initial message for conversation {request.ConversationId}; starting ConversationRelay without a greeting."); + } + response = twilio.ReturnConversationRelay(request.AgentId, request.ConversationId, initial?.Content); } else { diff --git a/src/Plugins/BotSharp.Plugin.Twilio/Controllers/TwilioReconnectController.cs b/src/Plugins/BotSharp.Plugin.Twilio/Controllers/TwilioReconnectController.cs index 6daf19589..b56e5c7f3 100644 --- a/src/Plugins/BotSharp.Plugin.Twilio/Controllers/TwilioReconnectController.cs +++ b/src/Plugins/BotSharp.Plugin.Twilio/Controllers/TwilioReconnectController.cs @@ -28,8 +28,7 @@ public async Task Reconnect(ConversationalVoiceRequest request) if (!string.IsNullOrEmpty(request.InitAudioFile)) { var twilio = _services.GetRequiredService(); - var audioUrl = twilio.GetSpeechPath(request.ConversationId, request.InitAudioFile); - response.Play(new Uri(audioUrl)); + twilio.AppendSpeech(response, request.ConversationId, request.InitAudioFile); } else { diff --git a/src/Plugins/BotSharp.Plugin.Twilio/Models/Relay/ConversationRelayActionRequest.cs b/src/Plugins/BotSharp.Plugin.Twilio/Models/Relay/ConversationRelayActionRequest.cs new file mode 100644 index 000000000..10228123b --- /dev/null +++ b/src/Plugins/BotSharp.Plugin.Twilio/Models/Relay/ConversationRelayActionRequest.cs @@ -0,0 +1,33 @@ +using Microsoft.AspNetCore.Mvc; + +namespace BotSharp.Plugin.Twilio.Models.Relay; + +/// +/// What Twilio posts to the Connect action URL when a ConversationRelay session ends +/// +public class ConversationRelayActionRequest : ConversationalVoiceRequest +{ + [FromForm] + public string? SessionId { get; set; } + + /// + /// ended (we sent "end"), completed (caller hung up) or failed + /// + [FromForm] + public string? SessionStatus { get; set; } + + [FromForm] + public string? SessionDuration { get; set; } + + [FromForm] + public string? HandoffData { get; set; } + + [FromForm] + public string? ErrorCode { get; set; } + + [FromForm] + public string? ErrorMessage { get; set; } + + [FromQuery(Name = "reconnects")] + public int Reconnects { get; set; } +} diff --git a/src/Plugins/BotSharp.Plugin.Twilio/Models/Relay/ConversationRelayParameters.cs b/src/Plugins/BotSharp.Plugin.Twilio/Models/Relay/ConversationRelayParameters.cs new file mode 100644 index 000000000..f8d7bb95f --- /dev/null +++ b/src/Plugins/BotSharp.Plugin.Twilio/Models/Relay/ConversationRelayParameters.cs @@ -0,0 +1,18 @@ +namespace BotSharp.Plugin.Twilio.Models.Relay; + +/// +/// Custom TwiML Parameter names passed into the ConversationRelay "setup" message +/// +public static class ConversationRelayParameters +{ + public const string WelcomeGreetingLength = "welcome-greeting-length"; +} + +/// +/// Values we put in the "end" message's handoffData, read back by the Connect action callback +/// +public static class ConversationRelayHandoff +{ + public const string HangUp = "hangup"; + public const string Transfer = "transfer"; +} diff --git a/src/Plugins/BotSharp.Plugin.Twilio/OutboundPhoneCallHandler/Functions/HangupPhoneCallFn.cs b/src/Plugins/BotSharp.Plugin.Twilio/OutboundPhoneCallHandler/Functions/HangupPhoneCallFn.cs index 3bf6fb906..aea4d2df2 100644 --- a/src/Plugins/BotSharp.Plugin.Twilio/OutboundPhoneCallHandler/Functions/HangupPhoneCallFn.cs +++ b/src/Plugins/BotSharp.Plugin.Twilio/OutboundPhoneCallHandler/Functions/HangupPhoneCallFn.cs @@ -2,6 +2,7 @@ using BotSharp.Abstraction.Routing; using BotSharp.Core.Infrastructures; using BotSharp.Plugin.Twilio.OutboundPhoneCallHandler.LlmContexts; +using BotSharp.Plugin.Twilio.Services; using Twilio.Rest.Api.V2010.Account; namespace BotSharp.Plugin.Twilio.OutboundPhoneCallHandler.Functions; @@ -28,7 +29,6 @@ public async Task Execute(RoleDialogModel message) { var args = JsonSerializer.Deserialize(message.FunctionArgs); - var fileStorage = _services.GetRequiredService(); var convService = _services.GetRequiredService(); var conversationId = convService.ConversationId; var states = _services.GetRequiredService(); @@ -44,13 +44,10 @@ public async Task Execute(RoleDialogModel message) var processUrl = $"{_twilioSetting.CallbackHost}/twilio/voice/hang-up?agent-id={message.CurrentAgentId}&conversation-id={conversationId}"; // Generate initial assistant audio - string initAudioFile = null; if (!string.IsNullOrEmpty(args.ResponseContent) && _twilioSetting.GenerateEndingAudio) { - var completion = CompletionProvider.GetAudioSynthesizer(_services); - var data = await completion.GenerateAudioAsync(args.ResponseContent); - initAudioFile = "ending.mp3"; - fileStorage.SaveSpeechFile(conversationId, initAudioFile, data); + var twilio = _services.GetRequiredService(); + var initAudioFile = await twilio.SaveSpeechAsync(conversationId, "ending", args.ResponseContent); processUrl += $"&init-audio-file={initAudioFile}"; } diff --git a/src/Plugins/BotSharp.Plugin.Twilio/OutboundPhoneCallHandler/Functions/LeaveVoicemailFn.cs b/src/Plugins/BotSharp.Plugin.Twilio/OutboundPhoneCallHandler/Functions/LeaveVoicemailFn.cs index 96aa55809..77a02b777 100644 --- a/src/Plugins/BotSharp.Plugin.Twilio/OutboundPhoneCallHandler/Functions/LeaveVoicemailFn.cs +++ b/src/Plugins/BotSharp.Plugin.Twilio/OutboundPhoneCallHandler/Functions/LeaveVoicemailFn.cs @@ -2,6 +2,7 @@ using BotSharp.Abstraction.Routing; using BotSharp.Core.Infrastructures; using BotSharp.Plugin.Twilio.OutboundPhoneCallHandler.LlmContexts; +using BotSharp.Plugin.Twilio.Services; namespace BotSharp.Plugin.Twilio.OutboundPhoneCallHandler.Functions; @@ -28,7 +29,6 @@ public async Task Execute(RoleDialogModel message) { var args = JsonSerializer.Deserialize(message.FunctionArgs); - var fileStorage = _services.GetRequiredService(); var routing = _services.GetRequiredService(); var conversationId = routing.Context.ConversationId; var states = _services.GetRequiredService(); @@ -42,13 +42,10 @@ public async Task Execute(RoleDialogModel message) } // Generate voice message audio - string initAudioFile = null; if (!string.IsNullOrEmpty(args.VoicemailMessage)) { - var completion = CompletionProvider.GetAudioSynthesizer(_services); - var data = await completion.GenerateAudioAsync(args.VoicemailMessage); - initAudioFile = "voicemail.mp3"; - fileStorage.SaveSpeechFile(conversationId, initAudioFile, data); + var twilio = _services.GetRequiredService(); + await twilio.SaveSpeechAsync(conversationId, "voicemail", args.VoicemailMessage); } message.Content = args.VoicemailMessage; diff --git a/src/Plugins/BotSharp.Plugin.Twilio/OutboundPhoneCallHandler/Functions/OutboundPhoneCallFn.cs b/src/Plugins/BotSharp.Plugin.Twilio/OutboundPhoneCallHandler/Functions/OutboundPhoneCallFn.cs index c0003b785..b49a10b16 100644 --- a/src/Plugins/BotSharp.Plugin.Twilio/OutboundPhoneCallHandler/Functions/OutboundPhoneCallFn.cs +++ b/src/Plugins/BotSharp.Plugin.Twilio/OutboundPhoneCallHandler/Functions/OutboundPhoneCallFn.cs @@ -9,6 +9,7 @@ using BotSharp.Plugin.Twilio.Interfaces; using BotSharp.Plugin.Twilio.Models; using BotSharp.Plugin.Twilio.OutboundPhoneCallHandler.LlmContexts; +using BotSharp.Plugin.Twilio.Services; using Twilio.Exceptions; using Twilio.Rest.Api.V2010.Account; using Twilio.TwiML.Messaging; @@ -62,7 +63,6 @@ public async Task Execute(RoleDialogModel message) return false; } - var fileStorage = _services.GetRequiredService(); var states = _services.GetRequiredService(); // Fork conversation @@ -73,24 +73,24 @@ public async Task Execute(RoleDialogModel message) var statusUrl = $"{_twilioSetting.CallbackHost}/twilio/voice/status?agent-id={message.CurrentAgentId}&conversation-id={newConversationId}"; var recordingStatusUrl = $"{_twilioSetting.CallbackHost}/twilio/recording/status?agent-id={message.CurrentAgentId}&conversation-id={newConversationId}"; - // Generate initial assistant audio + // load agent profile + var agentService = _services.GetRequiredService(); + var agent = await agentService.GetAgent(message.CurrentAgentId); + var realtime = agent.Labels.Contains("realtime"); + + // Generate initial assistant audio; on ConversationRelay Twilio speaks the initial message as the welcome greeting string initAudioFile = null; - if (!string.IsNullOrEmpty(args.InitialMessage)) + var useConversationRelay = !realtime && _twilioSetting.ConversationRelay.IsEnabledFor(message.CurrentAgentId); + if (!string.IsNullOrEmpty(args.InitialMessage) && !useConversationRelay) { - var completion = CompletionProvider.GetAudioSynthesizer(_services); - var data = await completion.GenerateAudioAsync(args.InitialMessage); - initAudioFile = "intial.mp3"; - fileStorage.SaveSpeechFile(newConversationId, initAudioFile, data); + var twilio = _services.GetRequiredService(); + initAudioFile = await twilio.SaveSpeechAsync(newConversationId, "intial", args.InitialMessage); statusUrl += $"&init-audio-file={initAudioFile}"; } - // load agent profile - var agentService = _services.GetRequiredService(); - var agent = await agentService.GetAgent(message.CurrentAgentId); - // Set up process URL streaming or synchronous - if (agent.Labels.Contains("realtime")) + if (realtime) { processUrl += "/inbound"; } diff --git a/src/Plugins/BotSharp.Plugin.Twilio/OutboundPhoneCallHandler/Functions/TransferPhoneCallFn.cs b/src/Plugins/BotSharp.Plugin.Twilio/OutboundPhoneCallHandler/Functions/TransferPhoneCallFn.cs index a5c644234..71aa9c5a5 100644 --- a/src/Plugins/BotSharp.Plugin.Twilio/OutboundPhoneCallHandler/Functions/TransferPhoneCallFn.cs +++ b/src/Plugins/BotSharp.Plugin.Twilio/OutboundPhoneCallHandler/Functions/TransferPhoneCallFn.cs @@ -3,6 +3,7 @@ using BotSharp.Abstraction.Routing; using BotSharp.Core.Infrastructures; using BotSharp.Plugin.Twilio.OutboundPhoneCallHandler.LlmContexts; +using BotSharp.Plugin.Twilio.Services; using Twilio.Rest.Api.V2010.Account; namespace BotSharp.Plugin.Twilio.OutboundPhoneCallHandler.Functions; @@ -33,7 +34,6 @@ public async Task Execute(RoleDialogModel message) { var args = JsonSerializer.Deserialize(message.FunctionArgs, _options.JsonSerializerOptions); - var fileStorage = _services.GetRequiredService(); var states = _services.GetRequiredService(); var sid = states.GetState("twilio_call_sid"); if (string.IsNullOrEmpty(sid)) @@ -50,10 +50,8 @@ public async Task Execute(RoleDialogModel message) // Generate initial assistant audio if (!string.IsNullOrEmpty(args.TransitionMessage)) { - var completion = CompletionProvider.GetAudioSynthesizer(_services); - var data = await completion.GenerateAudioAsync(args.TransitionMessage); - var initAudioFile = "transfer.mp3"; - fileStorage.SaveSpeechFile(conversationId, initAudioFile, data); + var twilio = _services.GetRequiredService(); + var initAudioFile = await twilio.SaveSpeechAsync(conversationId, "transfer", args.TransitionMessage); processUrl += $"&init-audio-file={initAudioFile}"; } diff --git a/src/Plugins/BotSharp.Plugin.Twilio/Services/ConversationRelaySession.cs b/src/Plugins/BotSharp.Plugin.Twilio/Services/ConversationRelaySession.cs new file mode 100644 index 000000000..8a15d4950 --- /dev/null +++ b/src/Plugins/BotSharp.Plugin.Twilio/Services/ConversationRelaySession.cs @@ -0,0 +1,471 @@ +using BotSharp.Plugin.Twilio.Models; +using BotSharp.Plugin.Twilio.Models.Relay; +using System.IO; +using System.Net.WebSockets; +using System.Threading; +using Task = System.Threading.Tasks.Task; + +namespace BotSharp.Plugin.Twilio.Services; + +/// +/// One phone call on Twilio ConversationRelay: Twilio sends what the caller said as text and speaks the text we send back. +/// https://www.twilio.com/docs/voice/conversationrelay/websocket-messages +/// +public class ConversationRelaySession +{ + private const string DefaultSayItAgainText = "Sorry, can you say it again?"; + private const string DefaultErrorText = "Sorry, something went wrong. Please call back later."; + private const double SpokenCharsPerSecond = 15; + + private readonly WebSocket _socket; + private readonly string _agentId; + private readonly string _conversationId; + private readonly TwilioSetting _settings; + private readonly Func, Task> _sendCallerMessage; + private readonly ILogger _logger; + + private readonly SemaphoreSlim _sendLock = new(1, 1); + private readonly object _sync = new(); + private readonly List _pendingInputs = []; + private readonly StringBuilder _digits = new(); + private readonly StringBuilder _partialPrompt = new(); + private string? _pendingDigits; + private bool _isProcessing; + private Task _processing = Task.CompletedTask; + private int _seqNum; + private int _idlePrompts; + private bool _ended; + + // Caller silence is measured from this moment: the estimated end of what we last said, or the caller's last input + private DateTime _quietSinceUtc = DateTime.MaxValue; + + private string? _callSid; + private string? _from; + private string? _to; + private string? _direction; + + /// Sends the caller's words to the agent and returns its reply; the second argument receives tool-call indications + public ConversationRelaySession( + WebSocket socket, + string agentId, + string conversationId, + TwilioSetting settings, + Func, Task> sendCallerMessage, + ILogger logger) + { + _socket = socket; + _agentId = agentId; + _conversationId = conversationId; + _settings = settings; + _sendCallerMessage = sendCallerMessage; + _logger = logger; + } + + public async Task RunAsync(CancellationToken cancellationToken) + { + using var cts = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken); + var idleWatch = WatchIdleAsync(cts.Token); + try + { + await ReceiveLoopAsync(cts.Token); + } + catch (Exception ex) when (ex is WebSocketException or OperationCanceledException) + { + _logger.LogInformation($"ConversationRelay socket closed for conversation {_conversationId}: {ex.Message}"); + } + finally + { + cts.Cancel(); + await SwallowAsync(idleWatch); + } + + // Let a turn already running finish so the conversation is saved + await SwallowAsync(_processing); + } + + private async Task ReceiveLoopAsync(CancellationToken cancellationToken) + { + var buffer = new byte[16 * 1024]; + using var message = new MemoryStream(); + while (_socket.State == WebSocketState.Open && !cancellationToken.IsCancellationRequested) + { + var result = await _socket.ReceiveAsync(new ArraySegment(buffer), cancellationToken); + if (result.MessageType == WebSocketMessageType.Close) + { + if (_socket.State == WebSocketState.CloseReceived) + { + await _socket.CloseOutputAsync(WebSocketCloseStatus.NormalClosure, string.Empty, CancellationToken.None); + } + break; + } + + message.Write(buffer, 0, result.Count); + if (!result.EndOfMessage) + { + continue; + } + + var json = Encoding.UTF8.GetString(message.ToArray()); + message.SetLength(0); + try + { + await HandleMessageAsync(json); + } + catch (JsonException ex) + { + _logger.LogWarning($"Unreadable ConversationRelay message for conversation {_conversationId}: {ex.Message}"); + } + } + } + + private async Task HandleMessageAsync(string json) + { + using var doc = JsonDocument.Parse(json); + var root = doc.RootElement; + switch (GetString(root, "type")) + { + case "setup": + _callSid = GetString(root, "callSid"); + _from = GetString(root, "from"); + _to = GetString(root, "to"); + _direction = GetString(root, "direction"); + var greetingLength = 0; + if (root.TryGetProperty("customParameters", out var parameters) + && GetString(parameters, ConversationRelayParameters.WelcomeGreetingLength) is string length) + { + int.TryParse(length, out greetingLength); + } + lock (_sync) + { + _quietSinceUtc = DateTime.UtcNow + EstimateSpeech(greetingLength); + } + _logger.LogInformation($"ConversationRelay session started for conversation {_conversationId} (call {_callSid})."); + break; + + case "prompt": + var text = GetString(root, "voicePrompt") ?? string.Empty; + var last = !root.TryGetProperty("last", out var lastElement) || lastElement.ValueKind != JsonValueKind.False; + _partialPrompt.Append(text); + if (last) + { + var prompt = _partialPrompt.ToString().Trim(); + _partialPrompt.Clear(); + if (prompt.Length > 0) + { + Enqueue(prompt, digits: null); + } + } + break; + + case "dtmf": + var digit = GetString(root, "digit"); + if (digit == "#") + { + var digits = _digits.ToString(); + _digits.Clear(); + if (digits.Length > 0) + { + Enqueue(digits, digits); + } + } + else if (!string.IsNullOrEmpty(digit)) + { + _digits.Append(digit); + lock (_sync) + { + _quietSinceUtc = DateTime.UtcNow; + } + } + break; + + case "interrupt": + // Playback stopped where the caller cut in; the caller's words follow as a prompt + lock (_sync) + { + _quietSinceUtc = DateTime.UtcNow; + } + _logger.LogInformation($"Caller interrupted the agent in conversation {_conversationId} after {GetString(root, "durationUntilInterruptMs")} ms."); + break; + + case "error": + _logger.LogWarning($"ConversationRelay error in conversation {_conversationId}: {GetString(root, "description")}"); + break; + } + + await Task.CompletedTask; + } + + private void Enqueue(string content, string? digits) + { + lock (_sync) + { + if (_ended) + { + return; + } + + _pendingInputs.Add(content); + if (digits != null) + { + _pendingDigits = digits; + } + _quietSinceUtc = DateTime.UtcNow; + _idlePrompts = 0; + + if (!_isProcessing) + { + _isProcessing = true; + _processing = Task.Run(ProcessPendingAsync); + } + } + } + + /// + /// Runs one agent turn at a time; what the caller says meanwhile is merged into the next turn. + /// + private async Task ProcessPendingAsync() + { + while (true) + { + CallerMessage message; + lock (_sync) + { + if (_ended || _pendingInputs.Count == 0) + { + _isProcessing = false; + return; + } + + message = new CallerMessage + { + AgentId = _agentId, + ConversationId = _conversationId, + SeqNumber = ++_seqNum, + Content = string.Join("\r\n", _pendingInputs), + Digits = _pendingDigits, + From = (string.Equals(_direction, "inbound", StringComparison.OrdinalIgnoreCase) ? _from : _to) ?? string.Empty + }; + _pendingInputs.Clear(); + _pendingDigits = null; + } + + AssistantMessage? reply; + try + { + // Run the turn outside the WebSocket request's execution context; it sets up its own HttpContext + Task turn; + using (ExecutionContext.SuppressFlow()) + { + turn = Task.Run(() => _sendCallerMessage(message, SpeakIndicationAsync)); + } + + var timeout = Task.Delay(TimeSpan.FromSeconds(_settings.ConversationRelay.AiResponseTimeoutSeconds)); + if (await Task.WhenAny(turn, timeout) != turn) + { + _logger.LogWarning($"Agent did not reply within {_settings.ConversationRelay.AiResponseTimeoutSeconds}s in conversation {_conversationId}."); + await FailAsync(); + return; + } + reply = await turn; + } + catch (Exception ex) + { + _logger.LogError(ex, $"Agent turn failed in conversation {_conversationId}."); + await FailAsync(); + return; + } + + if (reply == null || string.IsNullOrWhiteSpace(reply.Content) && !reply.ConversationEnd && !reply.HumanIntervationNeeded) + { + continue; + } + + await SpeakAsync(reply.Content); + if (reply.HumanIntervationNeeded) + { + await EndAsync(ConversationRelayHandoff.Transfer); + return; + } + if (reply.ConversationEnd) + { + await EndAsync(ConversationRelayHandoff.HangUp); + return; + } + } + } + + private async Task WatchIdleAsync(CancellationToken cancellationToken) + { + var relay = _settings.ConversationRelay; + var idleTimeout = TimeSpan.FromSeconds(relay.IdleTimeoutSeconds); + while (!cancellationToken.IsCancellationRequested) + { + await Task.Delay(TimeSpan.FromMilliseconds(500), cancellationToken); + + var askAgain = false; + var hangUp = false; + lock (_sync) + { + if (_ended || _isProcessing || _quietSinceUtc == DateTime.MaxValue || DateTime.UtcNow - _quietSinceUtc < idleTimeout) + { + continue; + } + + if (_idlePrompts < relay.MaxIdlePrompts) + { + _idlePrompts++; + askAgain = true; + } + else + { + hangUp = true; + } + _quietSinceUtc = DateTime.UtcNow; + } + + if (askAgain) + { + await SpeakPromptAsync($"say-it-again-{Random.Shared.Next(1, 5)}", DefaultSayItAgainText); + } + else if (hangUp) + { + _logger.LogInformation($"Caller stayed silent, hanging up conversation {_conversationId}."); + await EndAsync(ConversationRelayHandoff.HangUp); + } + } + } + + private async Task FailAsync() + { + await SpeakPromptAsync("error", DefaultErrorText); + await EndAsync(ConversationRelayHandoff.HangUp); + } + + /// + /// Speaks a tool-call indication: "#name" is a prerecorded prompt, anything else is spoken as is; segments are separated by "|". + /// + private async Task SpeakIndicationAsync(string indication) + { + foreach (var segment in indication.Split('|')) + { + var text = segment.Trim(); + if (text.StartsWith('#')) + { + await SpeakPromptAsync(text.Substring(1), fallbackText: null); + } + else + { + await SpeakAsync(text); + } + } + } + + /// + /// Speaks a prerecorded prompt's configured text so the voice matches; without one, plays the recording. + /// + private async Task SpeakPromptAsync(string name, string? fallbackText) + { + if (_settings.PromptTexts.TryGetValue(name, out var text) && !string.IsNullOrWhiteSpace(text)) + { + await SpeakAsync(text); + } + else if (fallbackText != null) + { + await SpeakAsync(fallbackText); + } + else + { + await SendAsync(new { type = "play", source = $"{_settings.CallbackHost}/twilio/{name}.mp3" }); + MarkSpeaking(TimeSpan.FromSeconds(3)); + } + } + + private async Task SpeakAsync(string? text) + { + if (string.IsNullOrWhiteSpace(text)) + { + return; + } + + lock (_sync) + { + if (_ended) + { + return; + } + } + + await SendAsync(new { type = "text", token = text, last = true }); + MarkSpeaking(EstimateSpeech(text.Length)); + } + + private async Task EndAsync(string action) + { + lock (_sync) + { + if (_ended) + { + return; + } + _ended = true; + } + + await SendAsync(new { type = "end", handoffData = JsonSerializer.Serialize(new { action }) }); + } + + private async Task SendAsync(object payload) + { + var bytes = Encoding.UTF8.GetBytes(JsonSerializer.Serialize(payload)); + await _sendLock.WaitAsync(); + try + { + if (_socket.State == WebSocketState.Open) + { + await _socket.SendAsync(new ArraySegment(bytes), WebSocketMessageType.Text, true, CancellationToken.None); + } + } + catch (Exception ex) when (ex is WebSocketException or ObjectDisposedException) + { + // The call already moved on, e.g. a hang-up or transfer redirected it + _logger.LogInformation($"ConversationRelay socket for conversation {_conversationId} is gone: {ex.Message}"); + } + finally + { + _sendLock.Release(); + } + } + + private void MarkSpeaking(TimeSpan duration) + { + lock (_sync) + { + var start = _quietSinceUtc == DateTime.MaxValue || _quietSinceUtc < DateTime.UtcNow ? DateTime.UtcNow : _quietSinceUtc; + _quietSinceUtc = start + duration; + } + } + + private static TimeSpan EstimateSpeech(int characters) + { + return TimeSpan.FromSeconds(characters / SpokenCharsPerSecond + 1); + } + + private static string? GetString(JsonElement element, string name) + { + if (element.ValueKind != JsonValueKind.Object || !element.TryGetProperty(name, out var value)) + { + return null; + } + return value.ValueKind == JsonValueKind.String ? value.GetString() : value.ToString(); + } + + private static async Task SwallowAsync(Task task) + { + try + { + await task; + } + catch (OperationCanceledException) + { + } + } +} diff --git a/src/Plugins/BotSharp.Plugin.Twilio/Services/TwilioConversationProcessor.cs b/src/Plugins/BotSharp.Plugin.Twilio/Services/TwilioConversationProcessor.cs new file mode 100644 index 000000000..e3902d897 --- /dev/null +++ b/src/Plugins/BotSharp.Plugin.Twilio/Services/TwilioConversationProcessor.cs @@ -0,0 +1,103 @@ +using BotSharp.Abstraction.Infrastructures.Enums; +using BotSharp.Abstraction.MessageHub.Models; +using BotSharp.Abstraction.MessageHub.Services; +using BotSharp.Abstraction.Routing; +using BotSharp.Plugin.Twilio.Models; +using Microsoft.AspNetCore.Http; +using System.Security.Claims; +using Task = System.Threading.Tasks.Task; + +namespace BotSharp.Plugin.Twilio.Services; + +/// +/// Sends what the caller said to the agent and returns the agent's reply. +/// Shared by the webhook flow (TwilioMessageQueueService) and the ConversationRelay WebSocket flow. +/// +public static class TwilioConversationProcessor +{ + /// A service scope created for this one caller message + /// Receives tool-call indications while the agent is working + public static async Task SendCallerMessageAsync( + IServiceProvider sp, + CallerMessage message, + Func? onIndication = null) + { + // Clean static HttpContext + var httpContext = sp.GetRequiredService(); + httpContext.HttpContext = new DefaultHttpContext(); + httpContext.HttpContext.User = new ClaimsPrincipal(new ClaimsIdentity()); + foreach (var header in message.RequestHeaders ?? []) + { + httpContext.HttpContext.Request.Headers[header.Key] = header.Value; + } + httpContext.HttpContext.Request.Headers["X-Twilio-BotSharp"] = "LOST"; + + AssistantMessage? reply = null; + + var inputMsg = new RoleDialogModel(AgentRole.User, message.Content); + var conv = sp.GetRequiredService(); + var routing = sp.GetRequiredService(); + var observer = sp.GetRequiredService(); + + using var container = observer.SubscribeObservers>(message.ConversationId, listeners: new() + { + { + ChatEvent.OnIndicationReceived, async data => + { + if (onIndication != null && !string.IsNullOrEmpty(data.Data.Indication)) + { + await onIndication(data.Data.Indication); + } + } + } + }); + await InitConversation(message, inputMsg, conv, routing); + + // Need to consider Inbound and Outbound call + var conversation = await conv.GetConversation(message.ConversationId); + + await conv.SendMessage(message.AgentId, + inputMsg, + replyMessage: await BuildPostbackMessageModel(conv, message), + async msg => + { + reply = new AssistantMessage() + { + ConversationEnd = msg.Instruction?.ConversationEnd ?? false, + HumanIntervationNeeded = string.Equals("human_intervention_needed", msg.FunctionName), + Content = msg.Content, + MessageId = msg.MessageId + }; + } + ); + + return reply; + } + + private static async Task BuildPostbackMessageModel(IConversationService conv, CallerMessage message) + { + var messages = await conv.GetDialogHistory(1); + if (!messages.Any()) return null; + var lastMessage = messages[0]; + if (string.IsNullOrEmpty(lastMessage.PostbackFunctionName)) return null; + return new PostbackMessageModel + { + FunctionName = lastMessage.PostbackFunctionName, + ParentId = lastMessage.MessageId, + Payload = message.Digits + }; + } + + private static async Task InitConversation(CallerMessage message, RoleDialogModel inputMsg, IConversationService conv, IRoutingService routing) + { + routing.Context.SetMessageId(message.ConversationId, inputMsg.MessageId); + var states = new List + { + new(StateConst.CHANNEL, ConversationChannel.Phone), + new("channel_id", message.From), + new("calling_phone", message.From) + }; + states.AddRange(message.States.Select(kvp => new MessageState(kvp.Key, kvp.Value))); + await conv.SetConversationId(message.ConversationId, states); + } +} diff --git a/src/Plugins/BotSharp.Plugin.Twilio/Services/TwilioMessageQueueService.cs b/src/Plugins/BotSharp.Plugin.Twilio/Services/TwilioMessageQueueService.cs index c7b7783b5..1afe12fdd 100644 --- a/src/Plugins/BotSharp.Plugin.Twilio/Services/TwilioMessageQueueService.cs +++ b/src/Plugins/BotSharp.Plugin.Twilio/Services/TwilioMessageQueueService.cs @@ -67,96 +67,30 @@ private async Task ProcessUserMessageAsync(CallerMessage message) { using var scope = _serviceProvider.CreateScope(); var sp = scope.ServiceProvider; - - // Clean static HttpContext - var httpContext = sp.GetRequiredService(); - httpContext.HttpContext = new DefaultHttpContext(); - httpContext.HttpContext.User = new ClaimsPrincipal(new ClaimsIdentity()); - foreach (var header in message.RequestHeaders ?? []) - { - httpContext.HttpContext.Request.Headers[header.Key] = header.Value; - } - httpContext.HttpContext.Request.Headers["X-Twilio-BotSharp"] = "LOST"; - - AssistantMessage reply = default!; - - var inputMsg = new RoleDialogModel(AgentRole.User, message.Content); - var conv = sp.GetRequiredService(); - var routing = sp.GetRequiredService(); - var config = sp.GetRequiredService(); var sessionManager = sp.GetRequiredService(); - var observer = sp.GetRequiredService(); - using var container = observer.SubscribeObservers>(message.ConversationId, listeners: new() + var reply = await TwilioConversationProcessor.SendCallerMessageAsync(sp, message, + async indication => await sessionManager.SetReplyIndicationAsync(message.ConversationId, message.SeqNumber, indication)); + if (reply == null) { - { ChatEvent.OnIndicationReceived, async data => await OnReceiveToolCallIndication(data.Data, message, sessionManager) } - }); - await InitConversation(message, inputMsg, conv, routing); - - // Need to consider Inbound and Outbound call - var conversation = await conv.GetConversation(message.ConversationId); - var agentId = message.AgentId; - - var result = await conv.SendMessage(agentId, - inputMsg, - replyMessage: await BuildPostbackMessageModel(conv, message), - async msg => - { - reply = new AssistantMessage() - { - ConversationEnd = msg.Instruction?.ConversationEnd ?? false, - HumanIntervationNeeded = string.Equals("human_intervention_needed", msg.FunctionName), - Content = msg.Content, - MessageId = msg.MessageId - }; - } - ); + _logger.LogWarning($"No reply for {message}."); + return; + } var settings = sp.GetRequiredService(); if (settings.GenerateReplyAudio) { reply.SpeechFileName = await GetReplySpeechFileName(message.ConversationId, reply, sp); } - - reply.Hints = await GetHints(agentId, reply, sp); - await sessionManager.SetAssistantReplyAsync(message.ConversationId, message.SeqNumber, reply); - } - private async Task BuildPostbackMessageModel(IConversationService conv, CallerMessage message) - { - var messages = await conv.GetDialogHistory(1); - if (!messages.Any()) return null; - var lastMessage = messages[0]; - if (string.IsNullOrEmpty(lastMessage.PostbackFunctionName)) return null; - return new PostbackMessageModel - { - FunctionName = lastMessage.PostbackFunctionName, - ParentId = lastMessage.MessageId, - Payload = message.Digits - }; - } - - private static async Task InitConversation(CallerMessage message, RoleDialogModel inputMsg, IConversationService conv, IRoutingService routing) - { - routing.Context.SetMessageId(message.ConversationId, inputMsg.MessageId); - var states = new List - { - new(StateConst.CHANNEL, ConversationChannel.Phone), - new("channel_id", message.From), - new("calling_phone", message.From) - }; - states.AddRange(message.States.Select(kvp => new MessageState(kvp.Key, kvp.Value))); - await conv.SetConversationId(message.ConversationId, states); + reply.Hints = await GetHints(message.AgentId, reply, sp); + await sessionManager.SetAssistantReplyAsync(message.ConversationId, message.SeqNumber, reply); } private static async Task GetReplySpeechFileName(string conversationId, AssistantMessage reply, IServiceProvider sp) { - var completion = CompletionProvider.GetAudioSynthesizer(sp); - var fileStorage = sp.GetRequiredService(); - var data = await completion.GenerateAudioAsync(reply.Content); - var fileName = $"reply_{reply.MessageId}.mp3"; - fileStorage.SaveSpeechFile(conversationId, fileName, data); - return fileName; + var twilio = sp.GetRequiredService(); + return await twilio.SaveSpeechAsync(conversationId, $"reply_{reply.MessageId}", reply.Content); } private static async Task GetHints(string agentId, AssistantMessage reply, IServiceProvider sp) @@ -191,12 +125,4 @@ private static async Task GetHints(string agentId, AssistantMessage repl hints.AddRange(extraWords); return string.Join(", ", hints.Select(x => x.ToLower()).Distinct().Reverse()); } - - private static async Task OnReceiveToolCallIndication(RoleDialogModel msg, CallerMessage message, ITwilioSessionManager sessionManager) - { - if (!string.IsNullOrEmpty(msg.Indication)) - { - await sessionManager.SetReplyIndicationAsync(message.ConversationId, message.SeqNumber, msg.Indication); - } - } } diff --git a/src/Plugins/BotSharp.Plugin.Twilio/Services/TwilioService.cs b/src/Plugins/BotSharp.Plugin.Twilio/Services/TwilioService.cs index 1720d95d5..cec309d72 100644 --- a/src/Plugins/BotSharp.Plugin.Twilio/Services/TwilioService.cs +++ b/src/Plugins/BotSharp.Plugin.Twilio/Services/TwilioService.cs @@ -5,6 +5,7 @@ using BotSharp.Core.Infrastructures; using BotSharp.Plugin.Twilio.Interfaces; using BotSharp.Plugin.Twilio.Models; +using BotSharp.Plugin.Twilio.Models.Relay; using Twilio.Jwt.AccessToken; using Twilio.Rest.Api.V2010.Account.Call; using Task = System.Threading.Tasks.Task; @@ -17,6 +18,9 @@ namespace BotSharp.Plugin.Twilio.Services; /// public class TwilioService { + private const string TextSpeechExtension = ".txt"; + private const string AudioSpeechExtension = ".mp3"; + private readonly TwilioSetting _settings; private readonly IServiceProvider _services; public readonly ILogger _logger; @@ -79,13 +83,13 @@ public VoiceResponse ReturnInstructions(ConversationalVoiceResponse conversation { foreach (var speechPath in conversationalVoiceResponse.SpeechPaths) { - gather.Play(new Uri($"{_settings.CallbackHost}/{speechPath}")); + AppendSpeech(gather, conversationalVoiceResponse.ConversationId, speechPath); } } if (!string.IsNullOrEmpty(conversationalVoiceResponse.Text)) { - gather.Say(conversationalVoiceResponse.Text); + gather.Append(CreateSay(conversationalVoiceResponse.Text)); } response.Append(gather); @@ -100,8 +104,7 @@ public VoiceResponse ReturnNoninterruptedInstructions(ConversationalVoiceRespons { foreach (var speechPath in voiceResponse.SpeechPaths) { - var uri = GetSpeechPath(conversationId, speechPath); - response.Play(new Uri(uri)); + AppendSpeech(response, conversationId, speechPath); } } @@ -145,7 +148,7 @@ public VoiceResponse HangUp(string speechPath) var response = new VoiceResponse(); if (!string.IsNullOrEmpty(speechPath)) { - response.Play(new Uri($"{_settings.CallbackHost}/{speechPath}")); + AppendSpeech(response, conversationId: null, speechPath); } response.Hangup(); return response; @@ -159,8 +162,7 @@ public VoiceResponse HangUp(ConversationalVoiceResponse voiceResponse) { foreach (var speechPath in voiceResponse.SpeechPaths) { - var uri = GetSpeechPath(conversationId, speechPath); - response.Play(new Uri(uri)); + AppendSpeech(response, conversationId, speechPath); } } @@ -173,7 +175,7 @@ public VoiceResponse DialCsrAgent(string speechPath) var response = new VoiceResponse(); if (!string.IsNullOrEmpty(speechPath)) { - response.Play(new Uri($"{_settings.CallbackHost}/{speechPath}")); + AppendSpeech(response, conversationId: null, speechPath); } response.Dial(_settings.CsrAgentNumber); return response; @@ -187,8 +189,7 @@ public VoiceResponse TransferCall(ConversationalVoiceResponse conversationalVoic { foreach (var speechPath in conversationalVoiceResponse.SpeechPaths) { - var uri = GetSpeechPath(conversationId, speechPath); - response.Play(new Uri(uri)); + AppendSpeech(response, conversationId, speechPath); } } response.Dial(conversationalVoiceResponse.TransferTo, answerOnBridge: true); @@ -214,7 +215,7 @@ public VoiceResponse HoldOn(int interval, string message = null) if (!string.IsNullOrEmpty(message)) { - gather.Say(message); + gather.Append(CreateSay(message)); } gather.Pause(interval); response.Append(gather); @@ -252,8 +253,7 @@ public VoiceResponse ReturnBidirectionalMediaStreamsInstructions(ConversationalV { foreach (var speechPath in conversationalVoiceResponse.SpeechPaths) { - var uri = GetSpeechPath(conversationId, speechPath); - response.Play(new Uri(uri)); + AppendSpeech(response, conversationId, speechPath); } } @@ -265,11 +265,54 @@ public VoiceResponse ReturnBidirectionalMediaStreamsInstructions(ConversationalV return response; } + /// + /// Hands the call to Twilio ConversationRelay. Twilio opens a WebSocket to twilio/relay/{agentId}/{conversationId} + /// and posts to twilio/relay/action when the session ends. + /// + public VoiceResponse ReturnConversationRelay(string agentId, string conversationId, string? welcomeGreeting, int reconnects = 0) + { + var relay = _settings.ConversationRelay; + var connect = new Connect(action: new Uri($"{_settings.CallbackHost}/twilio/relay/action?agent-id={agentId}&conversation-id={conversationId}&reconnects={reconnects}")); + + var element = new ConversationRelay( + url: GetConversationRelayUrl(agentId, conversationId), + language: relay.Language, + ttsProvider: relay.TtsProvider, + voice: relay.Voice, + transcriptionProvider: relay.TranscriptionProvider, + speechModel: relay.SpeechModel, + dtmfDetection: relay.DtmfDetection, + welcomeGreeting: string.IsNullOrWhiteSpace(welcomeGreeting) ? null : welcomeGreeting); + + // Attributes that the SDK's typed parameters don't cover (or only cover as booleans) + element.SetOption("interruptible", relay.Interruptible); + element.SetOption("welcomeGreetingInterruptible", relay.WelcomeGreetingInterruptible); + element.SetOption("interruptSensitivity", relay.InterruptSensitivity); + element.SetOption("ignoreBackchannel", relay.IgnoreBackchannel ? "true" : "false"); + if (!string.IsNullOrWhiteSpace(relay.Hints)) + { + element.SetOption("hints", relay.Hints); + } + + // Lets the session time its idle check from the end of the greeting + element.Parameter(name: ConversationRelayParameters.WelcomeGreetingLength, value: (welcomeGreeting?.Length ?? 0).ToString()); + + connect.Append(element); + var response = new VoiceResponse(); + response.Append(connect); + return response; + } + + public string GetConversationRelayUrl(string agentId, string conversationId) + { + var host = _settings.CallbackHost.Split("://").Last(); + return $"wss://{host}/twilio/relay/{agentId}/{conversationId}"; + } + public async Task WaitingForAiResponse(ConversationalVoiceRequest request) { VoiceResponse response; var sessionManager = _services.GetRequiredService(); - var fileStorage = _services.GetRequiredService(); var indication = await sessionManager.GetReplyIndicationAsync(request.ConversationId, request.SeqNum); if (indication != null) @@ -286,16 +329,7 @@ public async Task WaitingForAiResponse(ConversationalVoiceRequest else { var hash = Utilities.HashTextMd5(seg); - var fileName = $"indication_{hash}.mp3"; - - var existing = fileStorage.GetSpeechFile(request.ConversationId, fileName); - if (existing == BinaryData.Empty) - { - var completion = CompletionProvider.GetAudioSynthesizer(_services); - var data = await completion.GenerateAudioAsync(seg); - fileStorage.SaveSpeechFile(request.ConversationId, fileName, data); - } - + var fileName = await SaveSpeechAsync(request.ConversationId, $"indication_{hash}", seg, reuseExisting: true); speechPaths.Add($"twilio/voice/speeches/{request.ConversationId}/{fileName}"); } } @@ -363,6 +397,123 @@ public string GetSpeechPath(string conversationId, string speechPath) } } + /// + /// Saves the speech of a text in the conversation and returns its file name: a text file spoken later with Say + /// when SayGeneratedSpeech is on, otherwise an mp3 from the audio synthesizer. + /// + public async Task SaveSpeechAsync(string conversationId, string name, string text, bool reuseExisting = false) + { + var fileStorage = _services.GetRequiredService(); + var fileName = name + (_settings.SayGeneratedSpeech ? TextSpeechExtension : AudioSpeechExtension); + if (reuseExisting && !fileStorage.GetSpeechFile(conversationId, fileName).ToMemory().IsEmpty) + { + return fileName; + } + + var data = _settings.SayGeneratedSpeech + ? BinaryData.FromString(text) + : await CompletionProvider.GetAudioSynthesizer(_services).GenerateAudioAsync(text); + fileStorage.SaveSpeechFile(conversationId, fileName, data); + return fileName; + } + + /// + /// Returns the file name of a speech saved by SaveSpeechAsync, whichever mode it was saved in. + /// + public string GetSavedSpeechFile(string conversationId, string name) + { + var fileStorage = _services.GetRequiredService(); + var textFile = name + TextSpeechExtension; + return fileStorage.GetSpeechFile(conversationId, textFile).ToMemory().IsEmpty + ? name + AudioSpeechExtension + : textFile; + } + + /// + /// Appends a speech to the TwiML: Say for a text file saved by SaveSpeechAsync, Play for an audio file. + /// + public void AppendSpeech(TwiML parent, string? conversationId, string speechPath) + { + if (_settings.SayGeneratedSpeech && TryGetPromptText(speechPath, out var promptText)) + { + parent.Append(CreateSay(promptText)); + return; + } + + if (!speechPath.EndsWith(TextSpeechExtension, StringComparison.OrdinalIgnoreCase)) + { + parent.Append(new Play(new Uri(GetSpeechPath(conversationId, speechPath)))); + return; + } + + var (convId, fileName) = SplitSpeechPath(conversationId, speechPath); + var fileStorage = _services.GetRequiredService(); + var data = fileStorage.GetSpeechFile(convId, fileName); + if (data.ToMemory().IsEmpty) + { + _logger.LogWarning($"Speech text {fileName} is missing (conversation {convId})."); + return; + } + + parent.Append(CreateSay(data.ToString())); + } + + /// + /// Looks up the configured text of a prerecorded prompt such as "twilio/say-it-again-1.mp3". + /// Files generated per conversation (twilio/voice/speeches/...) never match. + /// + private bool TryGetPromptText(string speechPath, out string text) + { + text = string.Empty; + const string promptRoute = "twilio/"; + var path = StripCallbackHost(speechPath); + if (!path.StartsWith(promptRoute)) + { + return false; + } + + var fileName = path.Substring(promptRoute.Length); + if (fileName.Contains('/')) + { + return false; + } + + return _settings.PromptTexts.TryGetValue(System.IO.Path.GetFileNameWithoutExtension(fileName), out text!) + && !string.IsNullOrWhiteSpace(text); + } + + private string StripCallbackHost(string speechPath) + { + return speechPath.StartsWith(_settings.CallbackHost) + ? speechPath.Substring(_settings.CallbackHost.Length).TrimStart('/') + : speechPath; + } + + private (string? ConversationId, string FileName) SplitSpeechPath(string? conversationId, string speechPath) + { + var path = StripCallbackHost(speechPath); + + // Accepts "twilio/voice/speeches/{conversationId}/{fileName}" or a bare file name in the given conversation + const string speechRoute = "twilio/voice/speeches/"; + if (path.StartsWith(speechRoute)) + { + var parts = path.Substring(speechRoute.Length).Split('/'); + if (parts.Length == 2) + { + return (parts[0], parts[1]); + } + } + + return (conversationId, path); + } + + private Say CreateSay(string text) + { + return string.IsNullOrEmpty(_settings.SayVoice) + ? new Say(text) + : new Say(text, voice: _settings.SayVoice); + } + public string GenerateStatesParameter(List states) { if (states is null || states.Count == 0) diff --git a/src/Plugins/BotSharp.Plugin.Twilio/Settings/ConversationRelaySetting.cs b/src/Plugins/BotSharp.Plugin.Twilio/Settings/ConversationRelaySetting.cs new file mode 100644 index 000000000..d32df70e3 --- /dev/null +++ b/src/Plugins/BotSharp.Plugin.Twilio/Settings/ConversationRelaySetting.cs @@ -0,0 +1,70 @@ +namespace BotSharp.Plugin.Twilio.Settings; + +/// +/// Twilio ConversationRelay: Twilio transcribes the caller and speaks our replies, exchanging text with us over a WebSocket. +/// https://www.twilio.com/docs/voice/twiml/connect/conversationrelay +/// +public class ConversationRelaySetting +{ + public bool Enabled { get; set; } + + /// + /// Agents that use ConversationRelay for outbound calls. Empty means every agent. + /// + public List AgentIds { get; set; } = []; + + public string Language { get; set; } = "en-US"; + public string? TtsProvider { get; set; } + public string? Voice { get; set; } + public string? TranscriptionProvider { get; set; } + public string? SpeechModel { get; set; } + public string? Hints { get; set; } + + /// + /// none, dtmf, speech or any + /// + public string WelcomeGreetingInterruptible { get; set; } = "none"; + + /// + /// none, dtmf, speech or any + /// + public string Interruptible { get; set; } = "any"; + + /// + /// low, medium or high + /// + public string InterruptSensitivity { get; set; } = "medium"; + + public bool IgnoreBackchannel { get; set; } = true; + public bool DtmfDetection { get; set; } = true; + + /// + /// Seconds of caller silence, after the reply is expected to finish playing, before asking the caller to repeat + /// + public int IdleTimeoutSeconds { get; set; } = 8; + + /// + /// How many times to ask the caller to repeat before hanging up + /// + public int MaxIdlePrompts { get; set; } = 2; + + /// + /// How long to wait for the agent's reply before apologizing and hanging up + /// + public int AiResponseTimeoutSeconds { get; set; } = 30; + + /// + /// How many times to reconnect a session that failed (e.g. the pod holding the WebSocket went away) + /// + public int MaxReconnects { get; set; } = 2; + + /// + /// Verify the X-Twilio-Signature header of the WebSocket handshake + /// + public bool ValidateSignature { get; set; } = true; + + public bool IsEnabledFor(string? agentId) + { + return Enabled && (AgentIds.Count == 0 || (agentId != null && AgentIds.Contains(agentId))); + } +} diff --git a/src/Plugins/BotSharp.Plugin.Twilio/Settings/TwilioSetting.cs b/src/Plugins/BotSharp.Plugin.Twilio/Settings/TwilioSetting.cs index 94132241d..af884284e 100644 --- a/src/Plugins/BotSharp.Plugin.Twilio/Settings/TwilioSetting.cs +++ b/src/Plugins/BotSharp.Plugin.Twilio/Settings/TwilioSetting.cs @@ -33,4 +33,22 @@ public class TwilioSetting public bool GenerateReplyAudio { get; set; } = true; public bool GenerateEndingAudio { get; set; } = true; + + /// + /// Speak generated text with TwiML Say instead of pre-generating audio with the audio synthesizer + /// + public bool SayGeneratedSpeech { get; set; } + + /// + /// Voice for TwiML Say, e.g. "Polly.Joanna-Neural" or "Google.en-US-Neural2-F". Twilio's default voice when empty. + /// + public string? SayVoice { get; set; } + + /// + /// Text of the prerecorded prompts under twilio/, keyed by file name without extension (e.g. "say-it-again-1"). + /// Spoken with Say when SayGeneratedSpeech is on; prompts not listed keep playing their audio file. + /// + public Dictionary PromptTexts { get; set; } = new(); + + public ConversationRelaySetting ConversationRelay { get; set; } = new(); } diff --git a/src/Plugins/BotSharp.Plugin.Twilio/TwilioConversationRelayMiddleware.cs b/src/Plugins/BotSharp.Plugin.Twilio/TwilioConversationRelayMiddleware.cs new file mode 100644 index 000000000..b887a3c16 --- /dev/null +++ b/src/Plugins/BotSharp.Plugin.Twilio/TwilioConversationRelayMiddleware.cs @@ -0,0 +1,85 @@ +using BotSharp.Plugin.Twilio.Models; +using BotSharp.Plugin.Twilio.Services; +using Microsoft.AspNetCore.Http; +using Twilio.Security; +using Task = System.Threading.Tasks.Task; + +namespace BotSharp.Plugin.Twilio; + +/// +/// Accepts Twilio ConversationRelay WebSockets at /twilio/relay/{agentId}/{conversationId} +/// +public class TwilioConversationRelayMiddleware +{ + private readonly RequestDelegate _next; + private readonly ILogger _logger; + + public TwilioConversationRelayMiddleware(RequestDelegate next, ILogger logger) + { + _next = next; + _logger = logger; + } + + public async Task Invoke(HttpContext httpContext) + { + if (!httpContext.Request.Path.StartsWithSegments("/twilio/relay", out var rest) + || !httpContext.WebSockets.IsWebSocketRequest) + { + await _next(httpContext); + return; + } + + var parts = (rest.Value ?? string.Empty).Trim('/').Split('/'); + if (parts.Length != 2 || parts.Any(string.IsNullOrWhiteSpace)) + { + httpContext.Response.StatusCode = StatusCodes.Status404NotFound; + return; + } + + var (agentId, conversationId) = (parts[0], parts[1]); + var services = httpContext.RequestServices; + var settings = services.GetRequiredService(); + if (settings.ConversationRelay.ValidateSignature && !IsSignatureValid(httpContext, agentId, conversationId)) + { + httpContext.Response.StatusCode = StatusCodes.Status403Forbidden; + return; + } + + using var socket = await httpContext.WebSockets.AcceptWebSocketAsync(); + var scopeFactory = services.GetRequiredService(); + var session = new ConversationRelaySession(socket, agentId, conversationId, settings, + (message, onIndication) => SendCallerMessageAsync(scopeFactory, message, onIndication), + _logger); + await session.RunAsync(httpContext.RequestAborted); + } + + private static async Task SendCallerMessageAsync(IServiceScopeFactory scopeFactory, CallerMessage message, Func onIndication) + { + using var scope = scopeFactory.CreateScope(); + return await TwilioConversationProcessor.SendCallerMessageAsync(scope.ServiceProvider, message, onIndication); + } + + /// + /// Twilio signs the WebSocket handshake with X-Twilio-Signature over the wss URL from the TwiML. + /// + private bool IsSignatureValid(HttpContext httpContext, string agentId, string conversationId) + { + var services = httpContext.RequestServices; + var authToken = services.GetRequiredService()["Twilio:RequestValidation:AuthToken"]; + if (string.IsNullOrEmpty(authToken)) + { + _logger.LogWarning("Twilio auth token is not configured; skipping ConversationRelay signature check."); + return true; + } + + var signature = httpContext.Request.Headers["X-Twilio-Signature"].ToString(); + var url = services.GetRequiredService().GetConversationRelayUrl(agentId, conversationId); + var valid = !string.IsNullOrEmpty(signature) + && new RequestValidator(authToken).Validate(url, new Dictionary(), signature); + if (!valid) + { + _logger.LogWarning($"Rejected ConversationRelay connection for conversation {conversationId}: invalid or missing X-Twilio-Signature."); + } + return valid; + } +} From 4e10b8149bc6856cca8b1e61b3707c9ab0e22b5c Mon Sep 17 00:00:00 2001 From: "mars.yu" Date: Wed, 21 Oct 2026 11:02:38 +0800 Subject: [PATCH 03/10] feat(twilio): stream ConversationRelay replies while the model writes them With Twilio:ConversationRelay:StreamReplies on (off by default), a relay turn asks BotSharp to stream (use_stream_message, and language English when unset) and forwards each OnReceiveLlmStreamMessage piece to Twilio as a text token with last=false, so Twilio starts speaking with the first words instead of after the whole reply. - Each model response is closed with last=true when the next one starts, before a tool-call indication is spoken, and at the end of the turn. - The final reply is not spoken again; only the part not streamed is (text set by a tool, or a reply rewritten after the model wrote it). - When the caller cuts in, the rest of the streamed reply is dropped; when the turn times out, nothing streamed afterwards reaches the call. - TwilioConversationProcessor takes AgentTurnCallbacks (indication, stream start, stream text); the webhook queue passes only the indication, so that flow does not stream. Needs a chat provider that streams (OpenAI, Azure OpenAI, Anthropic, Google); non-English conversations get the whole reply as before. Tested with in-process harnesses: relay harness 42/42 (13 new: pieces and close, first words before the model finishes, tool-call boundary, remainder, rewritten reply, tool-set reply, interrupt, timeout, and the processor's state and event wiring against stubbed BotSharp services); Say harness passes. Co-Authored-By: Claude Opus 5.5 --- .../Models/AgentTurnCallbacks.cs | 24 ++++ .../Services/ConversationRelaySession.cs | 121 +++++++++++++++++- .../Services/TwilioConversationProcessor.cs | 43 ++++++- .../Services/TwilioMessageQueueService.cs | 6 +- .../Settings/ConversationRelaySetting.cs | 6 + .../TwilioConversationRelayMiddleware.cs | 6 +- 6 files changed, 190 insertions(+), 16 deletions(-) create mode 100644 src/Plugins/BotSharp.Plugin.Twilio/Models/AgentTurnCallbacks.cs diff --git a/src/Plugins/BotSharp.Plugin.Twilio/Models/AgentTurnCallbacks.cs b/src/Plugins/BotSharp.Plugin.Twilio/Models/AgentTurnCallbacks.cs new file mode 100644 index 000000000..f5735f168 --- /dev/null +++ b/src/Plugins/BotSharp.Plugin.Twilio/Models/AgentTurnCallbacks.cs @@ -0,0 +1,24 @@ +using Task = System.Threading.Tasks.Task; + +namespace BotSharp.Plugin.Twilio.Models; + +/// +/// What a phone flow wants to hear about while the agent works on one caller message. +/// +public class AgentTurnCallbacks +{ + /// + /// A tool-call indication, e.g. "#hold-on-1" or a sentence to say while the tool runs + /// + public Func? OnIndication { get; init; } + + /// + /// When set, the turn streams: called for each piece of reply text as the model writes it + /// + public Func? OnStreamText { get; init; } + + /// + /// Called when the model starts writing a new response, e.g. after a tool call + /// + public Func? OnStreamStarted { get; init; } +} diff --git a/src/Plugins/BotSharp.Plugin.Twilio/Services/ConversationRelaySession.cs b/src/Plugins/BotSharp.Plugin.Twilio/Services/ConversationRelaySession.cs index 8a15d4950..6ecd4db8e 100644 --- a/src/Plugins/BotSharp.Plugin.Twilio/Services/ConversationRelaySession.cs +++ b/src/Plugins/BotSharp.Plugin.Twilio/Services/ConversationRelaySession.cs @@ -21,7 +21,7 @@ public class ConversationRelaySession private readonly string _agentId; private readonly string _conversationId; private readonly TwilioSetting _settings; - private readonly Func, Task> _sendCallerMessage; + private readonly Func> _sendCallerMessage; private readonly ILogger _logger; private readonly SemaphoreSlim _sendLock = new(1, 1); @@ -39,18 +39,25 @@ public class ConversationRelaySession // Caller silence is measured from this moment: the estimated end of what we last said, or the caller's last input private DateTime _quietSinceUtc = DateTime.MaxValue; + // Streaming of the current turn's reply (ConversationRelay.StreamReplies) + private readonly StringBuilder _streamSegment = new(); + private string _lastStreamSegment = string.Empty; + private bool _utteranceOpen; + private bool _streamedThisTurn; + private bool _dropStream; + private string? _callSid; private string? _from; private string? _to; private string? _direction; - /// Sends the caller's words to the agent and returns its reply; the second argument receives tool-call indications + /// Sends the caller's words to the agent and returns its reply; the callbacks receive tool-call indications and the streamed reply public ConversationRelaySession( WebSocket socket, string agentId, string conversationId, TwilioSetting settings, - Func, Task> sendCallerMessage, + Func> sendCallerMessage, ILogger logger) { _socket = socket; @@ -183,6 +190,12 @@ private async Task HandleMessageAsync(string json) lock (_sync) { _quietSinceUtc = DateTime.UtcNow; + + // The caller talked over a reply still being streamed: don't send the rest of it + if (_isProcessing && _streamedThisTurn) + { + _dropStream = true; + } } _logger.LogInformation($"Caller interrupted the agent in conversation {_conversationId} after {GetString(root, "durationUntilInterruptMs")} ms."); break; @@ -247,8 +260,22 @@ private async Task ProcessPendingAsync() }; _pendingInputs.Clear(); _pendingDigits = null; + + _streamSegment.Clear(); + _lastStreamSegment = string.Empty; + _utteranceOpen = false; + _streamedThisTurn = false; + _dropStream = false; } + var stream = _settings.ConversationRelay.StreamReplies; + var callbacks = new AgentTurnCallbacks + { + OnIndication = SpeakIndicationAsync, + OnStreamText = stream ? StreamTextAsync : null, + OnStreamStarted = stream ? CloseUtteranceAsync : null + }; + AssistantMessage? reply; try { @@ -256,7 +283,7 @@ private async Task ProcessPendingAsync() Task turn; using (ExecutionContext.SuppressFlow()) { - turn = Task.Run(() => _sendCallerMessage(message, SpeakIndicationAsync)); + turn = Task.Run(() => _sendCallerMessage(message, callbacks)); } var timeout = Task.Delay(TimeSpan.FromSeconds(_settings.ConversationRelay.AiResponseTimeoutSeconds)); @@ -275,12 +302,13 @@ private async Task ProcessPendingAsync() return; } + await CloseUtteranceAsync(); if (reply == null || string.IsNullOrWhiteSpace(reply.Content) && !reply.ConversationEnd && !reply.HumanIntervationNeeded) { continue; } - await SpeakAsync(reply.Content); + await SpeakAsync(GetUnstreamedText(reply.Content)); if (reply.HumanIntervationNeeded) { await EndAsync(ConversationRelayHandoff.Transfer); @@ -337,15 +365,98 @@ private async Task WatchIdleAsync(CancellationToken cancellationToken) private async Task FailAsync() { + // A turn that timed out may still be streaming; keep its words out of the apology + lock (_sync) + { + _dropStream = true; + } + await CloseUtteranceAsync(); await SpeakPromptAsync("error", DefaultErrorText); await EndAsync(ConversationRelayHandoff.HangUp); } + /// + /// Sends a piece of the reply as the model writes it; Twilio speaks it as it arrives. + /// + private async Task StreamTextAsync(string text) + { + lock (_sync) + { + if (_ended || _dropStream) + { + return; + } + _streamSegment.Append(text); + _utteranceOpen = true; + _streamedThisTurn = true; + } + + await SendAsync(new { type = "text", token = text, last = false }); + } + + /// + /// Ends the utterance being streamed, so what follows (an indication, the next response) is spoken on its own. + /// + private async Task CloseUtteranceAsync() + { + int length; + lock (_sync) + { + if (!_utteranceOpen) + { + return; + } + _utteranceOpen = false; + _lastStreamSegment = _streamSegment.ToString(); + _streamSegment.Clear(); + length = _lastStreamSegment.Length; + } + + await SendAsync(new { type = "text", token = string.Empty, last = true }); + MarkSpeaking(EstimateSpeech(length)); + } + + /// + /// The part of the final reply that was not streamed. Normally nothing; the whole reply when it is not + /// what the model streamed last, e.g. text set by a tool, or a reply rewritten after the model wrote it. + /// + private string GetUnstreamedText(string? reply) + { + var final = reply?.Trim() ?? string.Empty; + string streamed; + lock (_sync) + { + if (!_streamedThisTurn) + { + return final; + } + if (_dropStream) + { + // The caller cut in; their words come next as a new turn + return string.Empty; + } + streamed = _lastStreamSegment.Trim(); + } + + if (final == streamed) + { + return string.Empty; + } + if (streamed.Length > 0 && final.StartsWith(streamed, StringComparison.Ordinal)) + { + return final.Substring(streamed.Length); + } + + _logger.LogInformation($"Reply in conversation {_conversationId} differs from what was streamed; speaking it in full."); + return final; + } + /// /// Speaks a tool-call indication: "#name" is a prerecorded prompt, anything else is spoken as is; segments are separated by "|". /// private async Task SpeakIndicationAsync(string indication) { + await CloseUtteranceAsync(); foreach (var segment in indication.Split('|')) { var text = segment.Trim(); diff --git a/src/Plugins/BotSharp.Plugin.Twilio/Services/TwilioConversationProcessor.cs b/src/Plugins/BotSharp.Plugin.Twilio/Services/TwilioConversationProcessor.cs index e3902d897..29cf0f57e 100644 --- a/src/Plugins/BotSharp.Plugin.Twilio/Services/TwilioConversationProcessor.cs +++ b/src/Plugins/BotSharp.Plugin.Twilio/Services/TwilioConversationProcessor.cs @@ -16,11 +16,11 @@ namespace BotSharp.Plugin.Twilio.Services; public static class TwilioConversationProcessor { /// A service scope created for this one caller message - /// Receives tool-call indications while the agent is working + /// Receives tool-call indications, and the reply as it is written when OnStreamText is set public static async Task SendCallerMessageAsync( IServiceProvider sp, CallerMessage message, - Func? onIndication = null) + AgentTurnCallbacks? callbacks = null) { // Clean static HttpContext var httpContext = sp.GetRequiredService(); @@ -39,20 +39,51 @@ public static class TwilioConversationProcessor var routing = sp.GetRequiredService(); var observer = sp.GetRequiredService(); - using var container = observer.SubscribeObservers>(message.ConversationId, listeners: new() + var streaming = callbacks?.OnStreamText != null; + var listeners = new Dictionary, Task>> { { ChatEvent.OnIndicationReceived, async data => { - if (onIndication != null && !string.IsNullOrEmpty(data.Data.Indication)) + if (callbacks?.OnIndication != null && !string.IsNullOrEmpty(data.Data.Indication)) { - await onIndication(data.Data.Indication); + await callbacks.OnIndication(data.Data.Indication); } } } - }); + }; + if (streaming) + { + // Listeners run in push order on the model's stream, so the pieces arrive in order + listeners[ChatEvent.BeforeReceiveLlmStreamMessage] = async _ => + { + if (callbacks!.OnStreamStarted != null) + { + await callbacks.OnStreamStarted(); + } + }; + listeners[ChatEvent.OnReceiveLlmStreamMessage] = async data => + { + if (!string.IsNullOrEmpty(data.Data.Content)) + { + await callbacks!.OnStreamText!(data.Data.Content); + } + }; + } + + using var container = observer.SubscribeObservers>(message.ConversationId, listeners: listeners); await InitConversation(message, inputMsg, conv, routing); + if (streaming) + { + // Routing streams only when both are set (RoutingService, InstructExecutor) + conv.States.SetState(StateConst.USE_STREAM_MESSAGE, true, source: StateSource.Application); + if (string.IsNullOrEmpty(conv.States.GetState(StateConst.LANGUAGE))) + { + conv.States.SetState(StateConst.LANGUAGE, LanguageType.ENGLISH, source: StateSource.Application); + } + } + // Need to consider Inbound and Outbound call var conversation = await conv.GetConversation(message.ConversationId); diff --git a/src/Plugins/BotSharp.Plugin.Twilio/Services/TwilioMessageQueueService.cs b/src/Plugins/BotSharp.Plugin.Twilio/Services/TwilioMessageQueueService.cs index 1afe12fdd..d1ebed39c 100644 --- a/src/Plugins/BotSharp.Plugin.Twilio/Services/TwilioMessageQueueService.cs +++ b/src/Plugins/BotSharp.Plugin.Twilio/Services/TwilioMessageQueueService.cs @@ -69,8 +69,10 @@ private async Task ProcessUserMessageAsync(CallerMessage message) var sp = scope.ServiceProvider; var sessionManager = sp.GetRequiredService(); - var reply = await TwilioConversationProcessor.SendCallerMessageAsync(sp, message, - async indication => await sessionManager.SetReplyIndicationAsync(message.ConversationId, message.SeqNumber, indication)); + var reply = await TwilioConversationProcessor.SendCallerMessageAsync(sp, message, new AgentTurnCallbacks + { + OnIndication = async indication => await sessionManager.SetReplyIndicationAsync(message.ConversationId, message.SeqNumber, indication) + }); if (reply == null) { _logger.LogWarning($"No reply for {message}."); diff --git a/src/Plugins/BotSharp.Plugin.Twilio/Settings/ConversationRelaySetting.cs b/src/Plugins/BotSharp.Plugin.Twilio/Settings/ConversationRelaySetting.cs index d32df70e3..0fbc419d8 100644 --- a/src/Plugins/BotSharp.Plugin.Twilio/Settings/ConversationRelaySetting.cs +++ b/src/Plugins/BotSharp.Plugin.Twilio/Settings/ConversationRelaySetting.cs @@ -53,6 +53,12 @@ public class ConversationRelaySetting /// public int AiResponseTimeoutSeconds { get; set; } = 30; + /// + /// Send the reply to Twilio while the model is still writing it, so speaking starts with the first words. + /// Needs a chat provider that supports streaming, and only applies to English conversations. + /// + public bool StreamReplies { get; set; } + /// /// How many times to reconnect a session that failed (e.g. the pod holding the WebSocket went away) /// diff --git a/src/Plugins/BotSharp.Plugin.Twilio/TwilioConversationRelayMiddleware.cs b/src/Plugins/BotSharp.Plugin.Twilio/TwilioConversationRelayMiddleware.cs index b887a3c16..899084dbd 100644 --- a/src/Plugins/BotSharp.Plugin.Twilio/TwilioConversationRelayMiddleware.cs +++ b/src/Plugins/BotSharp.Plugin.Twilio/TwilioConversationRelayMiddleware.cs @@ -48,15 +48,15 @@ public async Task Invoke(HttpContext httpContext) using var socket = await httpContext.WebSockets.AcceptWebSocketAsync(); var scopeFactory = services.GetRequiredService(); var session = new ConversationRelaySession(socket, agentId, conversationId, settings, - (message, onIndication) => SendCallerMessageAsync(scopeFactory, message, onIndication), + (message, callbacks) => SendCallerMessageAsync(scopeFactory, message, callbacks), _logger); await session.RunAsync(httpContext.RequestAborted); } - private static async Task SendCallerMessageAsync(IServiceScopeFactory scopeFactory, CallerMessage message, Func onIndication) + private static async Task SendCallerMessageAsync(IServiceScopeFactory scopeFactory, CallerMessage message, AgentTurnCallbacks callbacks) { using var scope = scopeFactory.CreateScope(); - return await TwilioConversationProcessor.SendCallerMessageAsync(scope.ServiceProvider, message, onIndication); + return await TwilioConversationProcessor.SendCallerMessageAsync(scope.ServiceProvider, message, callbacks); } /// From bb0bafd391f2506bfd58a667fc6dfb5bfc9519dd Mon Sep 17 00:00:00 2001 From: "mars.yu" Date: Thu, 22 Oct 2026 16:44:19 +0800 Subject: [PATCH 04/10] fix(twilio): address review findings in ConversationRelay - Stop dequeuing caller input once the call is gone: when the relay socket closes, a turn already running finishes (so the conversation is saved) but words queued during it no longer start another agent turn. - Keep each keypad entry's Digits: inputs are queued with their digits, and a keypad entry ends the merge of words said while the agent was busy, so two entries during one turn become two turns instead of losing the first. - Decide and end the silence hang-up atomically, so caller input arriving at that moment cannot start a turn on a call being hung up; skip the "say it again" prompt when the caller spoke since it was decided. - Drop the unused conversation lookup on every caller message (moved from the webhook queue, never read). - Escape agent and conversation ids in the relay action query and as single path segments of the WebSocket URL; the signature check rebuilds the same escaped URL. - Move ConversationRelayHandoff to its own file and wrap lines over 125 characters. Tested with the relay harness, 49/49: seven new checks for these fixes, six of which fail on the previous code (the hang-up race itself cannot be forced, its check guards the outcome). Say harness passes. Downstream app: build clean, unit and integration suites at their known baselines. Co-Authored-By: Claude Opus 5.5 --- .../TwilioConversationRelayController.cs | 27 ++-- .../Controllers/TwilioOutboundController.cs | 3 +- .../Models/Relay/ConversationRelayHandoff.cs | 10 ++ .../Relay/ConversationRelayParameters.cs | 9 -- .../Services/ConversationRelaySession.cs | 115 ++++++++++++++---- .../Services/TwilioConversationProcessor.cs | 16 +-- .../Services/TwilioMessageQueueService.cs | 3 +- .../Services/TwilioService.cs | 16 ++- .../TwilioConversationRelayMiddleware.cs | 6 +- 9 files changed, 146 insertions(+), 59 deletions(-) create mode 100644 src/Plugins/BotSharp.Plugin.Twilio/Models/Relay/ConversationRelayHandoff.cs diff --git a/src/Plugins/BotSharp.Plugin.Twilio/Controllers/TwilioConversationRelayController.cs b/src/Plugins/BotSharp.Plugin.Twilio/Controllers/TwilioConversationRelayController.cs index 92cab7f4f..46a4bd17a 100644 --- a/src/Plugins/BotSharp.Plugin.Twilio/Controllers/TwilioConversationRelayController.cs +++ b/src/Plugins/BotSharp.Plugin.Twilio/Controllers/TwilioConversationRelayController.cs @@ -13,7 +13,10 @@ public class TwilioConversationRelayController : TwilioController private readonly IServiceProvider _services; private readonly ILogger _logger; - public TwilioConversationRelayController(TwilioSetting settings, IServiceProvider services, ILogger logger) + public TwilioConversationRelayController( + TwilioSetting settings, + IServiceProvider services, + ILogger logger) { _settings = settings; _services = services; @@ -21,13 +24,15 @@ public TwilioConversationRelayController(TwilioSetting settings, IServiceProvide } /// - /// Twilio calls this when a ConversationRelay session ends: we sent "end" (handoff), the caller hung up, or the session failed. + /// Twilio calls this when a ConversationRelay session ends: we sent "end" (handoff), the caller hung up, + /// or the session failed. /// [ValidateRequest] [HttpPost("twilio/relay/action")] public async Task SessionEnded(ConversationRelayActionRequest request) { - _logger.LogInformation($"ConversationRelay session for conversation {request.ConversationId} ended: {request.SessionStatus} {request.ErrorCode} {request.ErrorMessage}"); + _logger.LogInformation($"ConversationRelay session for conversation {request.ConversationId} ended: " + + $"{request.SessionStatus} {request.ErrorCode} {request.ErrorMessage}"); // Authenticate (synchronous, so identity writes survive into the async session hooks) HookEmitter.Emit(_services, hook => hook.OnAuthenticate(request), request.AgentId); @@ -37,20 +42,26 @@ public async Task SessionEnded(ConversationRelayActionRequest reque if (action == ConversationRelayHandoff.Transfer && !string.IsNullOrEmpty(_settings.CsrAgentNumber)) { - await HookEmitter.Emit(_services, async hook => await hook.OnAgentTransferring(request, _settings), request.AgentId); + await HookEmitter.Emit(_services, + async hook => await hook.OnAgentTransferring(request, _settings), request.AgentId); return TwiML(twilio.DialCsrAgent(string.Empty)); } var callStillUp = !string.Equals(request.CallStatus, "completed", StringComparison.OrdinalIgnoreCase); - if (request.SessionStatus == "failed" && callStillUp && request.Reconnects < _settings.ConversationRelay.MaxReconnects) + var canReconnect = request.Reconnects < _settings.ConversationRelay.MaxReconnects; + if (request.SessionStatus == "failed" && callStillUp && canReconnect) { - _logger.LogWarning($"Reconnecting ConversationRelay for conversation {request.ConversationId} (attempt {request.Reconnects + 1})."); - return TwiML(twilio.ReturnConversationRelay(request.AgentId, request.ConversationId, welcomeGreeting: null, request.Reconnects + 1)); + _logger.LogWarning($"Reconnecting ConversationRelay for conversation {request.ConversationId} " + + $"(attempt {request.Reconnects + 1})."); + var relay = twilio.ReturnConversationRelay(request.AgentId, request.ConversationId, + welcomeGreeting: null, request.Reconnects + 1); + return TwiML(relay); } if (action == ConversationRelayHandoff.HangUp) { - await HookEmitter.Emit(_services, async hook => await hook.OnAgentHangUp(request), request.AgentId); + await HookEmitter.Emit(_services, + async hook => await hook.OnAgentHangUp(request), request.AgentId); } var response = new VoiceResponse(); diff --git a/src/Plugins/BotSharp.Plugin.Twilio/Controllers/TwilioOutboundController.cs b/src/Plugins/BotSharp.Plugin.Twilio/Controllers/TwilioOutboundController.cs index 96ca2788e..7049aeeb7 100644 --- a/src/Plugins/BotSharp.Plugin.Twilio/Controllers/TwilioOutboundController.cs +++ b/src/Plugins/BotSharp.Plugin.Twilio/Controllers/TwilioOutboundController.cs @@ -48,7 +48,8 @@ await HookEmitter.Emit(_services, var initial = await sessionManager.GetAssistantReplyAsync(request.ConversationId, 0); if (string.IsNullOrWhiteSpace(initial?.Content)) { - _logger.LogWarning($"No initial message for conversation {request.ConversationId}; starting ConversationRelay without a greeting."); + _logger.LogWarning($"No initial message for conversation {request.ConversationId}; " + + "starting ConversationRelay without a greeting."); } response = twilio.ReturnConversationRelay(request.AgentId, request.ConversationId, initial?.Content); } diff --git a/src/Plugins/BotSharp.Plugin.Twilio/Models/Relay/ConversationRelayHandoff.cs b/src/Plugins/BotSharp.Plugin.Twilio/Models/Relay/ConversationRelayHandoff.cs new file mode 100644 index 000000000..1338b10df --- /dev/null +++ b/src/Plugins/BotSharp.Plugin.Twilio/Models/Relay/ConversationRelayHandoff.cs @@ -0,0 +1,10 @@ +namespace BotSharp.Plugin.Twilio.Models.Relay; + +/// +/// Values we put in the "end" message's handoffData, read back by the Connect action callback +/// +public static class ConversationRelayHandoff +{ + public const string HangUp = "hangup"; + public const string Transfer = "transfer"; +} diff --git a/src/Plugins/BotSharp.Plugin.Twilio/Models/Relay/ConversationRelayParameters.cs b/src/Plugins/BotSharp.Plugin.Twilio/Models/Relay/ConversationRelayParameters.cs index f8d7bb95f..b8b78f40c 100644 --- a/src/Plugins/BotSharp.Plugin.Twilio/Models/Relay/ConversationRelayParameters.cs +++ b/src/Plugins/BotSharp.Plugin.Twilio/Models/Relay/ConversationRelayParameters.cs @@ -7,12 +7,3 @@ public static class ConversationRelayParameters { public const string WelcomeGreetingLength = "welcome-greeting-length"; } - -/// -/// Values we put in the "end" message's handoffData, read back by the Connect action callback -/// -public static class ConversationRelayHandoff -{ - public const string HangUp = "hangup"; - public const string Transfer = "transfer"; -} diff --git a/src/Plugins/BotSharp.Plugin.Twilio/Services/ConversationRelaySession.cs b/src/Plugins/BotSharp.Plugin.Twilio/Services/ConversationRelaySession.cs index 6ecd4db8e..80ae82ec8 100644 --- a/src/Plugins/BotSharp.Plugin.Twilio/Services/ConversationRelaySession.cs +++ b/src/Plugins/BotSharp.Plugin.Twilio/Services/ConversationRelaySession.cs @@ -26,16 +26,18 @@ public class ConversationRelaySession private readonly SemaphoreSlim _sendLock = new(1, 1); private readonly object _sync = new(); - private readonly List _pendingInputs = []; + private readonly List _pendingInputs = []; private readonly StringBuilder _digits = new(); private readonly StringBuilder _partialPrompt = new(); - private string? _pendingDigits; private bool _isProcessing; private Task _processing = Task.CompletedTask; private int _seqNum; private int _idlePrompts; private bool _ended; + // Bumped on every caller input, so a decision taken during silence can tell the caller spoke since + private int _activity; + // Caller silence is measured from this moment: the estimated end of what we last said, or the caller's last input private DateTime _quietSinceUtc = DateTime.MaxValue; @@ -51,7 +53,10 @@ public class ConversationRelaySession private string? _to; private string? _direction; - /// Sends the caller's words to the agent and returns its reply; the callbacks receive tool-call indications and the streamed reply + /// + /// Sends the caller's words to the agent and returns its reply; the callbacks receive tool-call indications + /// and the streamed reply + /// public ConversationRelaySession( WebSocket socket, string agentId, @@ -86,7 +91,19 @@ public async Task RunAsync(CancellationToken cancellationToken) await SwallowAsync(idleWatch); } - // Let a turn already running finish so the conversation is saved + // The call is gone: start no new turns, but let a turn already running finish so the conversation is saved + int dropped; + lock (_sync) + { + _ended = true; + dropped = _pendingInputs.Count; + _pendingInputs.Clear(); + } + if (dropped > 0) + { + _logger.LogInformation($"Dropped {dropped} caller input(s) received after the agent's last turn started; " + + $"conversation {_conversationId} has ended."); + } await SwallowAsync(_processing); } @@ -146,7 +163,8 @@ private async Task HandleMessageAsync(string json) { _quietSinceUtc = DateTime.UtcNow + EstimateSpeech(greetingLength); } - _logger.LogInformation($"ConversationRelay session started for conversation {_conversationId} (call {_callSid})."); + _logger.LogInformation( + $"ConversationRelay session started for conversation {_conversationId} (call {_callSid})."); break; case "prompt": @@ -181,6 +199,7 @@ private async Task HandleMessageAsync(string json) lock (_sync) { _quietSinceUtc = DateTime.UtcNow; + _activity++; } } break; @@ -197,11 +216,14 @@ private async Task HandleMessageAsync(string json) _dropStream = true; } } - _logger.LogInformation($"Caller interrupted the agent in conversation {_conversationId} after {GetString(root, "durationUntilInterruptMs")} ms."); + var interruptedAfter = GetString(root, "durationUntilInterruptMs"); + _logger.LogInformation( + $"Caller interrupted the agent in conversation {_conversationId} after {interruptedAfter} ms."); break; case "error": - _logger.LogWarning($"ConversationRelay error in conversation {_conversationId}: {GetString(root, "description")}"); + _logger.LogWarning( + $"ConversationRelay error in conversation {_conversationId}: {GetString(root, "description")}"); break; } @@ -217,13 +239,10 @@ private void Enqueue(string content, string? digits) return; } - _pendingInputs.Add(content); - if (digits != null) - { - _pendingDigits = digits; - } + _pendingInputs.Add(new PendingInput(content, digits)); _quietSinceUtc = DateTime.UtcNow; _idlePrompts = 0; + _activity++; if (!_isProcessing) { @@ -235,6 +254,7 @@ private void Enqueue(string content, string? digits) /// /// Runs one agent turn at a time; what the caller says meanwhile is merged into the next turn. + /// A keypad entry ends the merge, so each entry keeps its own Digits. /// private async Task ProcessPendingAsync() { @@ -249,17 +269,27 @@ private async Task ProcessPendingAsync() return; } + var batch = new List(); + foreach (var input in _pendingInputs) + { + batch.Add(input); + if (input.Digits != null) + { + break; + } + } + _pendingInputs.RemoveRange(0, batch.Count); + + var inbound = string.Equals(_direction, "inbound", StringComparison.OrdinalIgnoreCase); message = new CallerMessage { AgentId = _agentId, ConversationId = _conversationId, SeqNumber = ++_seqNum, - Content = string.Join("\r\n", _pendingInputs), - Digits = _pendingDigits, - From = (string.Equals(_direction, "inbound", StringComparison.OrdinalIgnoreCase) ? _from : _to) ?? string.Empty + Content = string.Join("\r\n", batch.Select(x => x.Content)), + Digits = batch[^1].Digits, + From = (inbound ? _from : _to) ?? string.Empty }; - _pendingInputs.Clear(); - _pendingDigits = null; _streamSegment.Clear(); _lastStreamSegment = string.Empty; @@ -289,7 +319,8 @@ private async Task ProcessPendingAsync() var timeout = Task.Delay(TimeSpan.FromSeconds(_settings.ConversationRelay.AiResponseTimeoutSeconds)); if (await Task.WhenAny(turn, timeout) != turn) { - _logger.LogWarning($"Agent did not reply within {_settings.ConversationRelay.AiResponseTimeoutSeconds}s in conversation {_conversationId}."); + var timeoutSeconds = _settings.ConversationRelay.AiResponseTimeoutSeconds; + _logger.LogWarning($"Agent did not reply within {timeoutSeconds}s in conversation {_conversationId}."); await FailAsync(); return; } @@ -303,7 +334,8 @@ private async Task ProcessPendingAsync() } await CloseUtteranceAsync(); - if (reply == null || string.IsNullOrWhiteSpace(reply.Content) && !reply.ConversationEnd && !reply.HumanIntervationNeeded) + if (reply == null + || string.IsNullOrWhiteSpace(reply.Content) && !reply.ConversationEnd && !reply.HumanIntervationNeeded) { continue; } @@ -332,9 +364,11 @@ private async Task WatchIdleAsync(CancellationToken cancellationToken) var askAgain = false; var hangUp = false; + var activity = 0; lock (_sync) { - if (_ended || _isProcessing || _quietSinceUtc == DateTime.MaxValue || DateTime.UtcNow - _quietSinceUtc < idleTimeout) + var quietFor = DateTime.UtcNow - _quietSinceUtc; + if (_ended || _isProcessing || _quietSinceUtc == DateTime.MaxValue || quietFor < idleTimeout) { continue; } @@ -343,22 +377,25 @@ private async Task WatchIdleAsync(CancellationToken cancellationToken) { _idlePrompts++; askAgain = true; + activity = _activity; } else { + // Decide and end in one step, so caller input arriving now cannot start a turn on a call being hung up + _ended = true; hangUp = true; } _quietSinceUtc = DateTime.UtcNow; } - if (askAgain) + if (askAgain && IsStillQuiet(activity)) { await SpeakPromptAsync($"say-it-again-{Random.Shared.Next(1, 5)}", DefaultSayItAgainText); } else if (hangUp) { _logger.LogInformation($"Caller stayed silent, hanging up conversation {_conversationId}."); - await EndAsync(ConversationRelayHandoff.HangUp); + await SendEndAsync(ConversationRelayHandoff.HangUp); } } } @@ -447,12 +484,14 @@ private string GetUnstreamedText(string? reply) return final.Substring(streamed.Length); } - _logger.LogInformation($"Reply in conversation {_conversationId} differs from what was streamed; speaking it in full."); + _logger.LogInformation( + $"Reply in conversation {_conversationId} differs from what was streamed; speaking it in full."); return final; } /// - /// Speaks a tool-call indication: "#name" is a prerecorded prompt, anything else is spoken as is; segments are separated by "|". + /// Speaks a tool-call indication: "#name" is a prerecorded prompt, anything else is spoken as is; + /// segments are separated by "|". /// private async Task SpeakIndicationAsync(string indication) { @@ -521,9 +560,31 @@ private async Task EndAsync(string action) _ended = true; } + await SendEndAsync(action); + } + + /// + /// Tells Twilio to end the session; the caller has already marked the session ended. + /// + private async Task SendEndAsync(string action) + { await SendAsync(new { type = "end", handoffData = JsonSerializer.Serialize(new { action }) }); } + /// + /// True when the caller has not spoken or pressed a key since was read + /// and nothing else took over. + /// + private bool IsStillQuiet(int activity) + { + lock (_sync) + { + return !_ended && !_isProcessing && _activity == activity; + } + } + + private record PendingInput(string Content, string? Digits); + private async Task SendAsync(object payload) { var bytes = Encoding.UTF8.GetBytes(JsonSerializer.Serialize(payload)); @@ -532,7 +593,8 @@ private async Task SendAsync(object payload) { if (_socket.State == WebSocketState.Open) { - await _socket.SendAsync(new ArraySegment(bytes), WebSocketMessageType.Text, true, CancellationToken.None); + var segment = new ArraySegment(bytes); + await _socket.SendAsync(segment, WebSocketMessageType.Text, true, CancellationToken.None); } } catch (Exception ex) when (ex is WebSocketException or ObjectDisposedException) @@ -550,7 +612,8 @@ private void MarkSpeaking(TimeSpan duration) { lock (_sync) { - var start = _quietSinceUtc == DateTime.MaxValue || _quietSinceUtc < DateTime.UtcNow ? DateTime.UtcNow : _quietSinceUtc; + var now = DateTime.UtcNow; + var start = _quietSinceUtc == DateTime.MaxValue || _quietSinceUtc < now ? now : _quietSinceUtc; _quietSinceUtc = start + duration; } } diff --git a/src/Plugins/BotSharp.Plugin.Twilio/Services/TwilioConversationProcessor.cs b/src/Plugins/BotSharp.Plugin.Twilio/Services/TwilioConversationProcessor.cs index 29cf0f57e..213c024c0 100644 --- a/src/Plugins/BotSharp.Plugin.Twilio/Services/TwilioConversationProcessor.cs +++ b/src/Plugins/BotSharp.Plugin.Twilio/Services/TwilioConversationProcessor.cs @@ -16,7 +16,9 @@ namespace BotSharp.Plugin.Twilio.Services; public static class TwilioConversationProcessor { /// A service scope created for this one caller message - /// Receives tool-call indications, and the reply as it is written when OnStreamText is set + /// + /// Receives tool-call indications, and the reply as it is written when OnStreamText is set + /// public static async Task SendCallerMessageAsync( IServiceProvider sp, CallerMessage message, @@ -71,7 +73,8 @@ public static class TwilioConversationProcessor }; } - using var container = observer.SubscribeObservers>(message.ConversationId, listeners: listeners); + using var container = observer.SubscribeObservers>( + message.ConversationId, listeners: listeners); await InitConversation(message, inputMsg, conv, routing); if (streaming) @@ -84,9 +87,6 @@ public static class TwilioConversationProcessor } } - // Need to consider Inbound and Outbound call - var conversation = await conv.GetConversation(message.ConversationId); - await conv.SendMessage(message.AgentId, inputMsg, replyMessage: await BuildPostbackMessageModel(conv, message), @@ -105,7 +105,8 @@ await conv.SendMessage(message.AgentId, return reply; } - private static async Task BuildPostbackMessageModel(IConversationService conv, CallerMessage message) + private static async Task BuildPostbackMessageModel( + IConversationService conv, CallerMessage message) { var messages = await conv.GetDialogHistory(1); if (!messages.Any()) return null; @@ -119,7 +120,8 @@ await conv.SendMessage(message.AgentId, }; } - private static async Task InitConversation(CallerMessage message, RoleDialogModel inputMsg, IConversationService conv, IRoutingService routing) + private static async Task InitConversation( + CallerMessage message, RoleDialogModel inputMsg, IConversationService conv, IRoutingService routing) { routing.Context.SetMessageId(message.ConversationId, inputMsg.MessageId); var states = new List diff --git a/src/Plugins/BotSharp.Plugin.Twilio/Services/TwilioMessageQueueService.cs b/src/Plugins/BotSharp.Plugin.Twilio/Services/TwilioMessageQueueService.cs index d1ebed39c..10c3376dc 100644 --- a/src/Plugins/BotSharp.Plugin.Twilio/Services/TwilioMessageQueueService.cs +++ b/src/Plugins/BotSharp.Plugin.Twilio/Services/TwilioMessageQueueService.cs @@ -71,7 +71,8 @@ private async Task ProcessUserMessageAsync(CallerMessage message) var reply = await TwilioConversationProcessor.SendCallerMessageAsync(sp, message, new AgentTurnCallbacks { - OnIndication = async indication => await sessionManager.SetReplyIndicationAsync(message.ConversationId, message.SeqNumber, indication) + OnIndication = async indication => + await sessionManager.SetReplyIndicationAsync(message.ConversationId, message.SeqNumber, indication) }); if (reply == null) { diff --git a/src/Plugins/BotSharp.Plugin.Twilio/Services/TwilioService.cs b/src/Plugins/BotSharp.Plugin.Twilio/Services/TwilioService.cs index cec309d72..07c5caeac 100644 --- a/src/Plugins/BotSharp.Plugin.Twilio/Services/TwilioService.cs +++ b/src/Plugins/BotSharp.Plugin.Twilio/Services/TwilioService.cs @@ -269,10 +269,13 @@ public VoiceResponse ReturnBidirectionalMediaStreamsInstructions(ConversationalV /// Hands the call to Twilio ConversationRelay. Twilio opens a WebSocket to twilio/relay/{agentId}/{conversationId} /// and posts to twilio/relay/action when the session ends. /// - public VoiceResponse ReturnConversationRelay(string agentId, string conversationId, string? welcomeGreeting, int reconnects = 0) + public VoiceResponse ReturnConversationRelay( + string agentId, string conversationId, string? welcomeGreeting, int reconnects = 0) { var relay = _settings.ConversationRelay; - var connect = new Connect(action: new Uri($"{_settings.CallbackHost}/twilio/relay/action?agent-id={agentId}&conversation-id={conversationId}&reconnects={reconnects}")); + var query = $"agent-id={Uri.EscapeDataString(agentId)}" + + $"&conversation-id={Uri.EscapeDataString(conversationId)}&reconnects={reconnects}"; + var connect = new Connect(action: new Uri($"{_settings.CallbackHost}/twilio/relay/action?{query}")); var element = new ConversationRelay( url: GetConversationRelayUrl(agentId, conversationId), @@ -295,7 +298,8 @@ public VoiceResponse ReturnConversationRelay(string agentId, string conversation } // Lets the session time its idle check from the end of the greeting - element.Parameter(name: ConversationRelayParameters.WelcomeGreetingLength, value: (welcomeGreeting?.Length ?? 0).ToString()); + var greetingLength = (welcomeGreeting?.Length ?? 0).ToString(); + element.Parameter(name: ConversationRelayParameters.WelcomeGreetingLength, value: greetingLength); connect.Append(element); var response = new VoiceResponse(); @@ -305,8 +309,9 @@ public VoiceResponse ReturnConversationRelay(string agentId, string conversation public string GetConversationRelayUrl(string agentId, string conversationId) { + // Each id is one path segment, so a "/" or "?" in it cannot change the route var host = _settings.CallbackHost.Split("://").Last(); - return $"wss://{host}/twilio/relay/{agentId}/{conversationId}"; + return $"wss://{host}/twilio/relay/{Uri.EscapeDataString(agentId)}/{Uri.EscapeDataString(conversationId)}"; } public async Task WaitingForAiResponse(ConversationalVoiceRequest request) @@ -329,7 +334,8 @@ public async Task WaitingForAiResponse(ConversationalVoiceRequest else { var hash = Utilities.HashTextMd5(seg); - var fileName = await SaveSpeechAsync(request.ConversationId, $"indication_{hash}", seg, reuseExisting: true); + var fileName = await SaveSpeechAsync(request.ConversationId, $"indication_{hash}", seg, + reuseExisting: true); speechPaths.Add($"twilio/voice/speeches/{request.ConversationId}/{fileName}"); } } diff --git a/src/Plugins/BotSharp.Plugin.Twilio/TwilioConversationRelayMiddleware.cs b/src/Plugins/BotSharp.Plugin.Twilio/TwilioConversationRelayMiddleware.cs index 899084dbd..eb2e863a9 100644 --- a/src/Plugins/BotSharp.Plugin.Twilio/TwilioConversationRelayMiddleware.cs +++ b/src/Plugins/BotSharp.Plugin.Twilio/TwilioConversationRelayMiddleware.cs @@ -53,7 +53,8 @@ public async Task Invoke(HttpContext httpContext) await session.RunAsync(httpContext.RequestAborted); } - private static async Task SendCallerMessageAsync(IServiceScopeFactory scopeFactory, CallerMessage message, AgentTurnCallbacks callbacks) + private static async Task SendCallerMessageAsync( + IServiceScopeFactory scopeFactory, CallerMessage message, AgentTurnCallbacks callbacks) { using var scope = scopeFactory.CreateScope(); return await TwilioConversationProcessor.SendCallerMessageAsync(scope.ServiceProvider, message, callbacks); @@ -78,7 +79,8 @@ private bool IsSignatureValid(HttpContext httpContext, string agentId, string co && new RequestValidator(authToken).Validate(url, new Dictionary(), signature); if (!valid) { - _logger.LogWarning($"Rejected ConversationRelay connection for conversation {conversationId}: invalid or missing X-Twilio-Signature."); + _logger.LogWarning($"Rejected ConversationRelay connection for conversation {conversationId}: " + + "invalid or missing X-Twilio-Signature."); } return valid; } From 3e7973b0a25a90167cd63c57ac3fd784464c703b Mon Sep 17 00:00:00 2001 From: "mars.yu" Date: Thu, 22 Oct 2026 22:36:58 +0800 Subject: [PATCH 05/10] feat(twilio): inbound ConversationRelay and a response watchdog Inbound calls for agents listed in ConversationRelay.AgentIds now connect to ConversationRelay (streamed replies) instead of the Gather flow. The welcome prompts queued by session hooks become the relay greeting through their Twilio:PromptTexts text. Calls that arrive with an intent keep the Gather flow, which already has their first message queued. ConversationRelay.ResponseWatchdogSeconds (off by default) says one short filler from ResponseWatchdogPrompts (default hold-on-1..5) when the agent has said nothing that long after the caller finished, at most once per turn. ResponseWatchdogAgentIds limits it to some agents. The filler is decided while holding the send lock, so it never lands inside a streamed reply, and a turn whose reply is already in never gets one. Co-Authored-By: Claude Opus 5.5 --- .../Controllers/TwilioInboundController.cs | 14 +++ .../Services/ConversationRelaySession.cs | 85 +++++++++++++++++-- .../Services/TwilioService.cs | 9 ++ .../Settings/ConversationRelaySetting.cs | 24 +++++- 4 files changed, 126 insertions(+), 6 deletions(-) diff --git a/src/Plugins/BotSharp.Plugin.Twilio/Controllers/TwilioInboundController.cs b/src/Plugins/BotSharp.Plugin.Twilio/Controllers/TwilioInboundController.cs index 7c1f083e4..8fab8aee1 100644 --- a/src/Plugins/BotSharp.Plugin.Twilio/Controllers/TwilioInboundController.cs +++ b/src/Plugins/BotSharp.Plugin.Twilio/Controllers/TwilioInboundController.cs @@ -93,6 +93,20 @@ await HookEmitter.Emit(_services, { response = twilio.ReturnBidirectionalMediaStreamsInstructions(instruction, agent); } + // A call that arrives with an intent already has its first message queued for the Gather flow + else if (string.IsNullOrWhiteSpace(request.Intent) && _settings.ConversationRelay.IsEnabledFor(request.AgentId)) + { + // ConversationRelay speaks text only: the prompts the session hooks queued become the greeting + var greeting = string.Join(" ", instruction.SpeechPaths + .Select(twilio.GetPromptText) + .Where(text => !string.IsNullOrWhiteSpace(text))); + if (greeting.Length == 0 && instruction.SpeechPaths.Count > 0) + { + _logger.LogWarning($"No prompt text for the welcome speech of agent {request.AgentId}; " + + "starting ConversationRelay without a greeting."); + } + response = twilio.ReturnConversationRelay(request.AgentId, request.ConversationId, greeting); + } else { if (string.IsNullOrWhiteSpace(request.Intent)) diff --git a/src/Plugins/BotSharp.Plugin.Twilio/Services/ConversationRelaySession.cs b/src/Plugins/BotSharp.Plugin.Twilio/Services/ConversationRelaySession.cs index 80ae82ec8..47c222ccd 100644 --- a/src/Plugins/BotSharp.Plugin.Twilio/Services/ConversationRelaySession.cs +++ b/src/Plugins/BotSharp.Plugin.Twilio/Services/ConversationRelaySession.cs @@ -15,7 +15,9 @@ public class ConversationRelaySession { private const string DefaultSayItAgainText = "Sorry, can you say it again?"; private const string DefaultErrorText = "Sorry, something went wrong. Please call back later."; + private const string DefaultHoldOnText = "One moment, please."; private const double SpokenCharsPerSecond = 15; + private static readonly List DefaultWatchdogPrompts = ["hold-on-1", "hold-on-2", "hold-on-3", "hold-on-4", "hold-on-5"]; private readonly WebSocket _socket; private readonly string _agentId; @@ -48,6 +50,9 @@ public class ConversationRelaySession private bool _streamedThisTurn; private bool _dropStream; + // Anything was said to the caller during the current turn; the response watchdog stays quiet once it is set + private bool _spokeThisTurn; + private string? _callSid; private string? _from; private string? _to; @@ -296,8 +301,14 @@ private async Task ProcessPendingAsync() _utteranceOpen = false; _streamedThisTurn = false; _dropStream = false; + _spokeThisTurn = false; } + using var watchdogCts = new CancellationTokenSource(); + var watchdog = _settings.ConversationRelay.IsResponseWatchdogEnabledFor(_agentId) + ? WatchResponseAsync(message.SeqNumber, watchdogCts.Token) + : Task.CompletedTask; + var stream = _settings.ConversationRelay.StreamReplies; var callbacks = new AgentTurnCallbacks { @@ -332,6 +343,16 @@ private async Task ProcessPendingAsync() await FailAsync(); return; } + finally + { + // The reply is in: a filler that already passed its delay must not slip in ahead of it + lock (_sync) + { + _spokeThisTurn = true; + } + watchdogCts.Cancel(); + await SwallowAsync(watchdog); + } await CloseUtteranceAsync(); if (reply == null @@ -354,6 +375,47 @@ private async Task ProcessPendingAsync() } } + /// + /// Says a short filler when the agent has said nothing this turn after ResponseWatchdogSeconds, + /// so the caller does not sit in silence while the model thinks or a tool runs. + /// + private async Task WatchResponseAsync(int seqNumber, CancellationToken cancellationToken) + { + await Task.Delay(TimeSpan.FromSeconds(_settings.ConversationRelay.ResponseWatchdogSeconds), cancellationToken); + + var text = PickFillerText(); + await _sendLock.WaitAsync(cancellationToken); + try + { + // Decided while holding the send lock, so a reply cannot start streaming between the check and the + // filler; a stream that already began has set _utteranceOpen and is waiting for this lock + lock (_sync) + { + if (_ended || _seqNum != seqNumber || _spokeThisTurn || _utteranceOpen) + { + return; + } + _spokeThisTurn = true; + } + await SendLockedAsync(new { type = "text", token = text, last = true }); + } + finally + { + _sendLock.Release(); + } + MarkSpeaking(EstimateSpeech(text.Length)); + } + + private string PickFillerText() + { + var names = _settings.ConversationRelay.ResponseWatchdogPrompts; + var texts = (names.Count > 0 ? names : DefaultWatchdogPrompts) + .Select(name => _settings.PromptTexts.TryGetValue(name, out var text) ? text : null) + .Where(text => !string.IsNullOrWhiteSpace(text)) + .ToList(); + return texts.Count == 0 ? DefaultHoldOnText : texts[Random.Shared.Next(texts.Count)]!; + } + private async Task WatchIdleAsync(CancellationToken cancellationToken) { var relay = _settings.ConversationRelay; @@ -426,6 +488,7 @@ private async Task StreamTextAsync(string text) _streamSegment.Append(text); _utteranceOpen = true; _streamedThisTurn = true; + _spokeThisTurn = true; } await SendAsync(new { type = "text", token = text, last = false }); @@ -543,6 +606,7 @@ private async Task SpeakAsync(string? text) { return; } + _spokeThisTurn = true; } await SendAsync(new { type = "text", token = text, last = true }); @@ -587,9 +651,24 @@ private record PendingInput(string Content, string? Digits); private async Task SendAsync(object payload) { - var bytes = Encoding.UTF8.GetBytes(JsonSerializer.Serialize(payload)); await _sendLock.WaitAsync(); try + { + await SendLockedAsync(payload); + } + finally + { + _sendLock.Release(); + } + } + + /// + /// Sends with already held by the caller. + /// + private async Task SendLockedAsync(object payload) + { + var bytes = Encoding.UTF8.GetBytes(JsonSerializer.Serialize(payload)); + try { if (_socket.State == WebSocketState.Open) { @@ -602,10 +681,6 @@ private async Task SendAsync(object payload) // The call already moved on, e.g. a hang-up or transfer redirected it _logger.LogInformation($"ConversationRelay socket for conversation {_conversationId} is gone: {ex.Message}"); } - finally - { - _sendLock.Release(); - } } private void MarkSpeaking(TimeSpan duration) diff --git a/src/Plugins/BotSharp.Plugin.Twilio/Services/TwilioService.cs b/src/Plugins/BotSharp.Plugin.Twilio/Services/TwilioService.cs index 07c5caeac..9c1ef3e3b 100644 --- a/src/Plugins/BotSharp.Plugin.Twilio/Services/TwilioService.cs +++ b/src/Plugins/BotSharp.Plugin.Twilio/Services/TwilioService.cs @@ -464,6 +464,15 @@ public void AppendSpeech(TwiML parent, string? conversationId, string speechPath parent.Append(CreateSay(data.ToString())); } + /// + /// The configured text of a prerecorded prompt, or null when the path has none (e.g. a per-conversation file). + /// ConversationRelay can only speak text, so this is how its greeting reuses the prompts a hook queued. + /// + public string? GetPromptText(string speechPath) + { + return TryGetPromptText(speechPath, out var text) ? text : null; + } + /// /// Looks up the configured text of a prerecorded prompt such as "twilio/say-it-again-1.mp3". /// Files generated per conversation (twilio/voice/speeches/...) never match. diff --git a/src/Plugins/BotSharp.Plugin.Twilio/Settings/ConversationRelaySetting.cs b/src/Plugins/BotSharp.Plugin.Twilio/Settings/ConversationRelaySetting.cs index 0fbc419d8..f9797f1d6 100644 --- a/src/Plugins/BotSharp.Plugin.Twilio/Settings/ConversationRelaySetting.cs +++ b/src/Plugins/BotSharp.Plugin.Twilio/Settings/ConversationRelaySetting.cs @@ -9,7 +9,7 @@ public class ConversationRelaySetting public bool Enabled { get; set; } /// - /// Agents that use ConversationRelay for outbound calls. Empty means every agent. + /// Agents that use ConversationRelay for their calls, outbound and inbound. Empty means every agent. /// public List AgentIds { get; set; } = []; @@ -69,8 +69,30 @@ public class ConversationRelaySetting /// public bool ValidateSignature { get; set; } = true; + /// + /// Seconds to wait for the agent's first words after the caller finishes speaking before saying a short + /// filler (e.g. "One moment, please."), at most once per turn. 0 turns it off. + /// + public double ResponseWatchdogSeconds { get; set; } + + /// + /// Agents the response watchdog applies to. Empty means every ConversationRelay agent. + /// + public List ResponseWatchdogAgentIds { get; set; } = []; + + /// + /// Names of Twilio:PromptTexts entries the filler is picked from at random. Empty uses hold-on-1 to hold-on-5. + /// + public List ResponseWatchdogPrompts { get; set; } = []; + public bool IsEnabledFor(string? agentId) { return Enabled && (AgentIds.Count == 0 || (agentId != null && AgentIds.Contains(agentId))); } + + public bool IsResponseWatchdogEnabledFor(string? agentId) + { + return ResponseWatchdogSeconds > 0 + && (ResponseWatchdogAgentIds.Count == 0 || (agentId != null && ResponseWatchdogAgentIds.Contains(agentId))); + } } From 644ef6944554db2a975cacaea016ec420c2fbf4f Mon Sep 17 00:00:00 2001 From: "mars.yu" Date: Fri, 23 Oct 2026 07:16:44 +0800 Subject: [PATCH 06/10] feat(twilio): repeat the response watchdog filler at intervals, never the same twice ResponseWatchdogSeconds becomes ResponseWatchdogIntervals: seconds of silence before each filler while the agent works on a reply. The first counts from when the caller finished, each next one from when whatever was said last (a filler, an indication, a streamed reply) ends; the last value repeats until the reply comes. The first filler of a turn comes from ResponseWatchdogPrompts (default hold-on-1..5), later ones from ResponseWatchdogLongPrompts (default hold-on-long-1..10), and no sentence is repeated in a turn while unused ones remain. Still nothing while a reply is streaming, and nothing once the turn's reply is in. Co-Authored-By: Claude Opus 5.5 --- .../Services/ConversationRelaySession.cs | 83 ++++++++++++++----- .../Settings/ConversationRelaySetting.cs | 16 ++-- 2 files changed, 75 insertions(+), 24 deletions(-) diff --git a/src/Plugins/BotSharp.Plugin.Twilio/Services/ConversationRelaySession.cs b/src/Plugins/BotSharp.Plugin.Twilio/Services/ConversationRelaySession.cs index 47c222ccd..7e7c37092 100644 --- a/src/Plugins/BotSharp.Plugin.Twilio/Services/ConversationRelaySession.cs +++ b/src/Plugins/BotSharp.Plugin.Twilio/Services/ConversationRelaySession.cs @@ -1,4 +1,4 @@ -using BotSharp.Plugin.Twilio.Models; +using BotSharp.Plugin.Twilio.Models; using BotSharp.Plugin.Twilio.Models.Relay; using System.IO; using System.Net.WebSockets; @@ -18,6 +18,8 @@ public class ConversationRelaySession private const string DefaultHoldOnText = "One moment, please."; private const double SpokenCharsPerSecond = 15; private static readonly List DefaultWatchdogPrompts = ["hold-on-1", "hold-on-2", "hold-on-3", "hold-on-4", "hold-on-5"]; + private static readonly List DefaultWatchdogLongPrompts = Enumerable.Range(1, 10).Select(i => $"hold-on-long-{i}").ToList(); + private static readonly TimeSpan WatchdogPollInterval = TimeSpan.FromMilliseconds(100); private readonly WebSocket _socket; private readonly string _agentId; @@ -50,8 +52,9 @@ public class ConversationRelaySession private bool _streamedThisTurn; private bool _dropStream; - // Anything was said to the caller during the current turn; the response watchdog stays quiet once it is set - private bool _spokeThisTurn; + // The current turn's reply is in; the response watchdog says nothing more once it is set + private bool _turnReplied; + private DateTime _turnStartedUtc; private string? _callSid; private string? _from; @@ -301,7 +304,8 @@ private async Task ProcessPendingAsync() _utteranceOpen = false; _streamedThisTurn = false; _dropStream = false; - _spokeThisTurn = false; + _turnReplied = false; + _turnStartedUtc = DateTime.UtcNow; } using var watchdogCts = new CancellationTokenSource(); @@ -348,7 +352,7 @@ private async Task ProcessPendingAsync() // The reply is in: a filler that already passed its delay must not slip in ahead of it lock (_sync) { - _spokeThisTurn = true; + _turnReplied = true; } watchdogCts.Cancel(); await SwallowAsync(watchdog); @@ -376,26 +380,59 @@ private async Task ProcessPendingAsync() } /// - /// Says a short filler when the agent has said nothing this turn after ResponseWatchdogSeconds, - /// so the caller does not sit in silence while the model thinks or a tool runs. + /// While the agent works on a reply, says a filler each time the line has been silent for the next + /// ResponseWatchdogIntervals value, so the caller never sits in a long silence while the model thinks or a tool runs. + /// Silence counts from the end of whatever was said last (the caller, a filler, an indication), and no filler is + /// said twice in a turn while unused ones remain. /// private async Task WatchResponseAsync(int seqNumber, CancellationToken cancellationToken) { - await Task.Delay(TimeSpan.FromSeconds(_settings.ConversationRelay.ResponseWatchdogSeconds), cancellationToken); + var relay = _settings.ConversationRelay; + var said = new HashSet(); + for (var step = 0; ; step++) + { + var interval = TimeSpan.FromSeconds(relay.ResponseWatchdogIntervals[Math.Min(step, relay.ResponseWatchdogIntervals.Count - 1)]); + while (true) + { + await Task.Delay(WatchdogPollInterval, cancellationToken); + lock (_sync) + { + if (_ended || _seqNum != seqNumber || _turnReplied) + { + return; + } + // _quietSinceUtc is the estimated end of what was last said, or the caller's last input + var quietSince = _quietSinceUtc == DateTime.MaxValue || _quietSinceUtc < _turnStartedUtc ? _turnStartedUtc : _quietSinceUtc; + if (!_utteranceOpen && DateTime.UtcNow - quietSince >= interval) + { + break; + } + } + } - var text = PickFillerText(); - await _sendLock.WaitAsync(cancellationToken); + var text = PickFillerText(step == 0 ? relay.ResponseWatchdogPrompts : relay.ResponseWatchdogLongPrompts, + step == 0 ? DefaultWatchdogPrompts : DefaultWatchdogLongPrompts, said); + if (!await SpeakFillerAsync(seqNumber, text)) + { + return; + } + said.Add(text); + } + } + + private async Task SpeakFillerAsync(int seqNumber, string text) + { + await _sendLock.WaitAsync(); try { // Decided while holding the send lock, so a reply cannot start streaming between the check and the // filler; a stream that already began has set _utteranceOpen and is waiting for this lock lock (_sync) { - if (_ended || _seqNum != seqNumber || _spokeThisTurn || _utteranceOpen) + if (_ended || _seqNum != seqNumber || _turnReplied || _utteranceOpen) { - return; + return false; } - _spokeThisTurn = true; } await SendLockedAsync(new { type = "text", token = text, last = true }); } @@ -404,16 +441,26 @@ private async Task WatchResponseAsync(int seqNumber, CancellationToken cancellat _sendLock.Release(); } MarkSpeaking(EstimateSpeech(text.Length)); + return true; } - private string PickFillerText() + private string PickFillerText(List names, List defaultNames, HashSet said) { - var names = _settings.ConversationRelay.ResponseWatchdogPrompts; - var texts = (names.Count > 0 ? names : DefaultWatchdogPrompts) + var texts = (names.Count > 0 ? names : defaultNames) .Select(name => _settings.PromptTexts.TryGetValue(name, out var text) ? text : null) .Where(text => !string.IsNullOrWhiteSpace(text)) + .Select(text => text!) + .Distinct() .ToList(); - return texts.Count == 0 ? DefaultHoldOnText : texts[Random.Shared.Next(texts.Count)]!; + if (texts.Count == 0) + { + return DefaultHoldOnText; + } + + // A different sentence each time; repeat only once every one has been said + var unsaid = texts.Where(text => !said.Contains(text)).ToList(); + var pool = unsaid.Count > 0 ? unsaid : texts; + return pool[Random.Shared.Next(pool.Count)]; } private async Task WatchIdleAsync(CancellationToken cancellationToken) @@ -488,7 +535,6 @@ private async Task StreamTextAsync(string text) _streamSegment.Append(text); _utteranceOpen = true; _streamedThisTurn = true; - _spokeThisTurn = true; } await SendAsync(new { type = "text", token = text, last = false }); @@ -606,7 +652,6 @@ private async Task SpeakAsync(string? text) { return; } - _spokeThisTurn = true; } await SendAsync(new { type = "text", token = text, last = true }); diff --git a/src/Plugins/BotSharp.Plugin.Twilio/Settings/ConversationRelaySetting.cs b/src/Plugins/BotSharp.Plugin.Twilio/Settings/ConversationRelaySetting.cs index f9797f1d6..cb6cbbfe9 100644 --- a/src/Plugins/BotSharp.Plugin.Twilio/Settings/ConversationRelaySetting.cs +++ b/src/Plugins/BotSharp.Plugin.Twilio/Settings/ConversationRelaySetting.cs @@ -70,10 +70,11 @@ public class ConversationRelaySetting public bool ValidateSignature { get; set; } = true; /// - /// Seconds to wait for the agent's first words after the caller finishes speaking before saying a short - /// filler (e.g. "One moment, please."), at most once per turn. 0 turns it off. + /// Seconds of silence on the line, while the agent works on its reply, before each filler such as "One moment, please.": + /// the first counts from when the caller finished, each next one from when the previous filler (or anything else + /// we said) finished. The last value repeats until the reply comes. Empty turns the watchdog off. /// - public double ResponseWatchdogSeconds { get; set; } + public List ResponseWatchdogIntervals { get; set; } = []; /// /// Agents the response watchdog applies to. Empty means every ConversationRelay agent. @@ -81,10 +82,15 @@ public class ConversationRelaySetting public List ResponseWatchdogAgentIds { get; set; } = []; /// - /// Names of Twilio:PromptTexts entries the filler is picked from at random. Empty uses hold-on-1 to hold-on-5. + /// Names of Twilio:PromptTexts entries the first filler of a turn is picked from. Empty uses hold-on-1 to hold-on-5. /// public List ResponseWatchdogPrompts { get; set; } = []; + /// + /// Names of Twilio:PromptTexts entries the later fillers of a long wait are picked from. Empty uses hold-on-long-1 to hold-on-long-10. + /// + public List ResponseWatchdogLongPrompts { get; set; } = []; + public bool IsEnabledFor(string? agentId) { return Enabled && (AgentIds.Count == 0 || (agentId != null && AgentIds.Contains(agentId))); @@ -92,7 +98,7 @@ public bool IsEnabledFor(string? agentId) public bool IsResponseWatchdogEnabledFor(string? agentId) { - return ResponseWatchdogSeconds > 0 + return ResponseWatchdogIntervals.Count > 0 && ResponseWatchdogIntervals.All(seconds => seconds > 0) && (ResponseWatchdogAgentIds.Count == 0 || (agentId != null && ResponseWatchdogAgentIds.Contains(agentId))); } } From d15a2c4d3e764aceadd9929390307d037537c1ba Mon Sep 17 00:00:00 2001 From: "mars.yu" Date: Fri, 23 Oct 2026 14:32:15 +0800 Subject: [PATCH 07/10] feat(twilio): close the streamed utterance when a response ends in a tool call A line the model says before a slow tool ("Okay, that's everything I need") stayed an open utterance until the next response started streaming, so the response watchdog never filled the silence while the tool ran. The relay now closes it as soon as the response turns out to be a tool call (AgentTurnCallbacks.OnStreamEndedWithToolCall), and the watchdog counts from the end of that line. Co-Authored-By: Claude Opus 5.5 --- .../BotSharp.Plugin.Twilio/Models/AgentTurnCallbacks.cs | 6 ++++++ .../Services/ConversationRelaySession.cs | 5 ++++- .../Services/TwilioConversationProcessor.cs | 7 +++++++ 3 files changed, 17 insertions(+), 1 deletion(-) diff --git a/src/Plugins/BotSharp.Plugin.Twilio/Models/AgentTurnCallbacks.cs b/src/Plugins/BotSharp.Plugin.Twilio/Models/AgentTurnCallbacks.cs index f5735f168..fcbb782b7 100644 --- a/src/Plugins/BotSharp.Plugin.Twilio/Models/AgentTurnCallbacks.cs +++ b/src/Plugins/BotSharp.Plugin.Twilio/Models/AgentTurnCallbacks.cs @@ -21,4 +21,10 @@ public class AgentTurnCallbacks /// Called when the model starts writing a new response, e.g. after a tool call /// public Func? OnStreamStarted { get; init; } + + /// + /// Called when a response the model wrote ends in a tool call: what it said before the call is complete, + /// and the tool may take a while before anything else is said + /// + public Func? OnStreamEndedWithToolCall { get; init; } } diff --git a/src/Plugins/BotSharp.Plugin.Twilio/Services/ConversationRelaySession.cs b/src/Plugins/BotSharp.Plugin.Twilio/Services/ConversationRelaySession.cs index 7e7c37092..3f54fea6c 100644 --- a/src/Plugins/BotSharp.Plugin.Twilio/Services/ConversationRelaySession.cs +++ b/src/Plugins/BotSharp.Plugin.Twilio/Services/ConversationRelaySession.cs @@ -318,7 +318,10 @@ private async Task ProcessPendingAsync() { OnIndication = SpeakIndicationAsync, OnStreamText = stream ? StreamTextAsync : null, - OnStreamStarted = stream ? CloseUtteranceAsync : null + OnStreamStarted = stream ? CloseUtteranceAsync : null, + // A line said before a slow tool ("Okay, that's everything I need") is spoken now, and the watchdog + // counts the silence after it instead of waiting for the next response to close it + OnStreamEndedWithToolCall = stream ? CloseUtteranceAsync : null }; AssistantMessage? reply; diff --git a/src/Plugins/BotSharp.Plugin.Twilio/Services/TwilioConversationProcessor.cs b/src/Plugins/BotSharp.Plugin.Twilio/Services/TwilioConversationProcessor.cs index 213c024c0..6ae5ce0b9 100644 --- a/src/Plugins/BotSharp.Plugin.Twilio/Services/TwilioConversationProcessor.cs +++ b/src/Plugins/BotSharp.Plugin.Twilio/Services/TwilioConversationProcessor.cs @@ -71,6 +71,13 @@ public static class TwilioConversationProcessor await callbacks!.OnStreamText!(data.Data.Content); } }; + listeners[ChatEvent.AfterReceiveLlmStreamMessage] = async data => + { + if (data.Data.Role == AgentRole.Function && callbacks!.OnStreamEndedWithToolCall != null) + { + await callbacks.OnStreamEndedWithToolCall(); + } + }; } using var container = observer.SubscribeObservers>( From 7364c1c903b340914ab59f2bcdf4955b34f47067 Mon Sep 17 00:00:00 2001 From: "mars.yu" Date: Fri, 23 Oct 2026 15:05:55 +0800 Subject: [PATCH 08/10] Log what the model wrote when an OpenAI chat reply hits the output limit A reply cut off by max_output_tokens is replaced by a fixed notice, so the log is the only record of what the model wrote. The streaming path logged nothing, and the non-streaming paths left out the tool calls, which is where a cut-off usually happens. All three now log the text and the tool calls with their arguments as far as the model got. Co-Authored-By: Claude Opus 5.5 --- .../Chat/ChatCompletionProvider.Chat.cs | 17 +++++++++++++++-- 1 file changed, 15 insertions(+), 2 deletions(-) diff --git a/src/Plugins/BotSharp.Plugin.OpenAI/Providers/Chat/ChatCompletionProvider.Chat.cs b/src/Plugins/BotSharp.Plugin.OpenAI/Providers/Chat/ChatCompletionProvider.Chat.cs index f731a9842..7f91dbbe8 100644 --- a/src/Plugins/BotSharp.Plugin.OpenAI/Providers/Chat/ChatCompletionProvider.Chat.cs +++ b/src/Plugins/BotSharp.Plugin.OpenAI/Providers/Chat/ChatCompletionProvider.Chat.cs @@ -56,7 +56,7 @@ private async Task InnerGetChatCompletions(Agent agent, List InnerGetChatCompletionsAsync(Agent agent, } else if (reason == ChatFinishReason.Length) { - _logger.LogWarning($"Action: {nameof(InnerGetChatCompletionsAsync)}, Reason: {reason}, Agent: {agent.Name}, MaxOutputTokens: {options.MaxOutputTokenCount}, Content:{text}"); + _logger.LogWarning($"Action: {nameof(InnerGetChatCompletionsAsync)}, Reason: {reason}, Agent: {agent.Name}, MaxOutputTokens: {options.MaxOutputTokenCount}, Content:{text}, ToolCalls: {DescribeToolCalls(ToLlmToolCalls(value.ToolCalls))}"); msg = new RoleDialogModel(AgentRole.Assistant, $"AI response exceeded max output length") { @@ -321,6 +321,13 @@ private async Task InnerGetChatCompletionsStreamingAsync(Agent } else if (choice.FinishReason.HasValue) { + if (choice.FinishReason == ChatFinishReason.Length) + { + // What the model wrote before the limit is replaced by the notice below, so this is the only record of it + _logger.LogWarning($"Action: {nameof(InnerGetChatCompletionsStreamingAsync)}, Reason: {choice.FinishReason}, Agent: {agent.Name}, " + + $"MaxOutputTokens: {options.MaxOutputTokenCount}, Content:{textStream.GetText()}, ToolCalls: {DescribeToolCalls(ReconstructToolCalls(toolCalls))}"); + } + var text = choice.FinishReason == ChatFinishReason.Length ? "Model reached the maximum number of tokens allowed." : choice.FinishReason == ChatFinishReason.ContentFilter ? "Content is omitted due to content filter rule." : choice.FinishReason.Value.ToString(); @@ -759,6 +766,12 @@ private static List ToLlmToolCalls(IEnumerable? toolC .Select(x => new LlmToolCall(x.Id, x.FunctionName, x.FunctionArguments?.ToString())) .ToList(); + /// + /// The tool calls of a reply cut off by the output limit, arguments as far as the model got + /// + private static string DescribeToolCalls(List calls) + => string.Join(", ", calls.Select(x => $"{x.FunctionName}({x.FunctionArgs})")); + /// /// Rebuilds the tool calls a streaming reply asked for. /// From 1fb0499edd29ed94be1cb8ad0370c9822be650c4 Mon Sep 17 00:00:00 2001 From: "mars.yu" Date: Fri, 23 Oct 2026 15:47:47 +0800 Subject: [PATCH 09/10] Let a Twilio session hook supply the ConversationRelay greeting text The inbound ConversationRelay greeting could only come from the prompt texts behind the speech paths the session hooks queue, which are fixed. A hook that writes ConversationalVoiceResponse.Text (e.g. a greeting with the caller's name) now has that said instead; without it the greeting is built as before. Co-Authored-By: Claude Opus 5.5 --- .../Controllers/TwilioInboundController.cs | 11 +++++++---- 1 file changed, 7 insertions(+), 4 deletions(-) diff --git a/src/Plugins/BotSharp.Plugin.Twilio/Controllers/TwilioInboundController.cs b/src/Plugins/BotSharp.Plugin.Twilio/Controllers/TwilioInboundController.cs index 8fab8aee1..d1ee441a1 100644 --- a/src/Plugins/BotSharp.Plugin.Twilio/Controllers/TwilioInboundController.cs +++ b/src/Plugins/BotSharp.Plugin.Twilio/Controllers/TwilioInboundController.cs @@ -96,10 +96,13 @@ await HookEmitter.Emit(_services, // A call that arrives with an intent already has its first message queued for the Gather flow else if (string.IsNullOrWhiteSpace(request.Intent) && _settings.ConversationRelay.IsEnabledFor(request.AgentId)) { - // ConversationRelay speaks text only: the prompts the session hooks queued become the greeting - var greeting = string.Join(" ", instruction.SpeechPaths - .Select(twilio.GetPromptText) - .Where(text => !string.IsNullOrWhiteSpace(text))); + // ConversationRelay speaks text only: a greeting a session hook wrote (e.g. with the caller's name) is + // said as it is, otherwise the prompts the session hooks queued become the greeting + var greeting = !string.IsNullOrWhiteSpace(instruction.Text) + ? instruction.Text + : string.Join(" ", instruction.SpeechPaths + .Select(twilio.GetPromptText) + .Where(text => !string.IsNullOrWhiteSpace(text))); if (greeting.Length == 0 && instruction.SpeechPaths.Count > 0) { _logger.LogWarning($"No prompt text for the welcome speech of agent {request.AgentId}; " + From 3747588b036ee819f67281195883338054bbc0da Mon Sep 17 00:00:00 2001 From: "mars.yu" Date: Fri, 23 Oct 2026 15:48:07 +0800 Subject: [PATCH 10/10] fix(twilio): recheck relay send decisions once the send lock is held A streamed piece, an idle "say it again" prompt or a spoken line was checked first and sent after waiting for the send lock, so the decision could go stale meanwhile: - a streamed piece queued behind a filler could go out after the caller cut in or the session ended, leaving one fragment of the reply; - an idle prompt could be sent after caller input had started a turn, landing before or over that turn's reply. SendIfAsync now takes the send lock first and evaluates the condition under _sync right before sending; streaming bookkeeping changes only when the piece is actually sent. Every nonempty partial prompt (partialPrompts) now counts as caller activity, so a caller mid-sentence is never asked to repeat or treated as silent. Tests (onebrain host): a reply queued behind a filler goes out whole, a cut-in on a streamed reply stops the rest, and a caller sending partial prompts past the idle timeout is not prompted. The first and last fail on the previous code. Co-Authored-By: Claude Opus 5.5 --- .../Services/ConversationRelaySession.cs | 110 ++++++++++-------- 1 file changed, 61 insertions(+), 49 deletions(-) diff --git a/src/Plugins/BotSharp.Plugin.Twilio/Services/ConversationRelaySession.cs b/src/Plugins/BotSharp.Plugin.Twilio/Services/ConversationRelaySession.cs index 3f54fea6c..c41b96359 100644 --- a/src/Plugins/BotSharp.Plugin.Twilio/Services/ConversationRelaySession.cs +++ b/src/Plugins/BotSharp.Plugin.Twilio/Services/ConversationRelaySession.cs @@ -179,6 +179,15 @@ private async Task HandleMessageAsync(string json) var text = GetString(root, "voicePrompt") ?? string.Empty; var last = !root.TryGetProperty("last", out var lastElement) || lastElement.ValueKind != JsonValueKind.False; _partialPrompt.Append(text); + if (!last && text.Length > 0) + { + // With partialPrompts on, the caller is mid-sentence: not silent, and not to be asked to repeat + lock (_sync) + { + _quietSinceUtc = DateTime.UtcNow; + _activity++; + } + } if (last) { var prompt = _partialPrompt.ToString().Trim(); @@ -425,26 +434,15 @@ private async Task WatchResponseAsync(int seqNumber, CancellationToken cancellat private async Task SpeakFillerAsync(int seqNumber, string text) { - await _sendLock.WaitAsync(); - try + // A stream that already began has set _utteranceOpen and may be waiting for the send lock + var sent = await SendIfAsync( + () => !_ended && _seqNum == seqNumber && !_turnReplied && !_utteranceOpen, + new { type = "text", token = text, last = true }); + if (sent) { - // Decided while holding the send lock, so a reply cannot start streaming between the check and the - // filler; a stream that already began has set _utteranceOpen and is waiting for this lock - lock (_sync) - { - if (_ended || _seqNum != seqNumber || _turnReplied || _utteranceOpen) - { - return false; - } - } - await SendLockedAsync(new { type = "text", token = text, last = true }); + MarkSpeaking(EstimateSpeech(text.Length)); } - finally - { - _sendLock.Release(); - } - MarkSpeaking(EstimateSpeech(text.Length)); - return true; + return sent; } private string PickFillerText(List names, List defaultNames, HashSet said) @@ -500,9 +498,12 @@ private async Task WatchIdleAsync(CancellationToken cancellationToken) _quietSinceUtc = DateTime.UtcNow; } - if (askAgain && IsStillQuiet(activity)) + if (askAgain) { - await SpeakPromptAsync($"say-it-again-{Random.Shared.Next(1, 5)}", DefaultSayItAgainText); + // Rechecked once the send lock is held: caller input arriving meanwhile starts a turn, and the + // prompt must not then be said over or after that turn's reply + await SpeakPromptAsync($"say-it-again-{Random.Shared.Next(1, 5)}", DefaultSayItAgainText, + canSpeak: () => !_isProcessing && _activity == activity); } else if (hangUp) { @@ -529,18 +530,19 @@ private async Task FailAsync() /// private async Task StreamTextAsync(string text) { - lock (_sync) + // Decided once the send lock is held: a piece queued behind another send must not go out after the + // caller cut in (_dropStream) or the session ended meanwhile + await SendIfAsync(() => { if (_ended || _dropStream) { - return; + return false; } _streamSegment.Append(text); _utteranceOpen = true; _streamedThisTurn = true; - } - - await SendAsync(new { type = "text", token = text, last = false }); + return true; + }, new { type = "text", token = text, last = false }); } /// @@ -625,40 +627,36 @@ private async Task SpeakIndicationAsync(string indication) /// /// Speaks a prerecorded prompt's configured text so the voice matches; without one, plays the recording. /// - private async Task SpeakPromptAsync(string name, string? fallbackText) + /// Extra condition, checked under once the send lock is held + private async Task SpeakPromptAsync(string name, string? fallbackText, Func? canSpeak = null) { if (_settings.PromptTexts.TryGetValue(name, out var text) && !string.IsNullOrWhiteSpace(text)) { - await SpeakAsync(text); + await SpeakAsync(text, canSpeak); } else if (fallbackText != null) { - await SpeakAsync(fallbackText); + await SpeakAsync(fallbackText, canSpeak); } - else + else if (await SendIfAsync(() => !_ended && (canSpeak?.Invoke() ?? true), + new { type = "play", source = $"{_settings.CallbackHost}/twilio/{name}.mp3" })) { - await SendAsync(new { type = "play", source = $"{_settings.CallbackHost}/twilio/{name}.mp3" }); MarkSpeaking(TimeSpan.FromSeconds(3)); } } - private async Task SpeakAsync(string? text) + /// Extra condition, checked under once the send lock is held + private async Task SpeakAsync(string? text, Func? canSpeak = null) { if (string.IsNullOrWhiteSpace(text)) { return; } - lock (_sync) + if (await SendIfAsync(() => !_ended && (canSpeak?.Invoke() ?? true), new { type = "text", token = text, last = true })) { - if (_ended) - { - return; - } + MarkSpeaking(EstimateSpeech(text.Length)); } - - await SendAsync(new { type = "text", token = text, last = true }); - MarkSpeaking(EstimateSpeech(text.Length)); } private async Task EndAsync(string action) @@ -683,26 +681,40 @@ private async Task SendEndAsync(string action) await SendAsync(new { type = "end", handoffData = JsonSerializer.Serialize(new { action }) }); } - /// - /// True when the caller has not spoken or pressed a key since was read - /// and nothing else took over. - /// - private bool IsStillQuiet(int activity) + private record PendingInput(string Content, string? Digits); + + private async Task SendAsync(object payload) { - lock (_sync) + await _sendLock.WaitAsync(); + try + { + await SendLockedAsync(payload); + } + finally { - return !_ended && !_isProcessing && _activity == activity; + _sendLock.Release(); } } - private record PendingInput(string Content, string? Digits); - - private async Task SendAsync(object payload) + /// + /// Sends only if still holds once the send lock is held, so a decision taken + /// before waiting for the lock cannot go stale. It runs under and may update session + /// state that must change only when the send goes ahead. + /// + private async Task SendIfAsync(Func canSend, object payload) { await _sendLock.WaitAsync(); try { + lock (_sync) + { + if (!canSend()) + { + return false; + } + } await SendLockedAsync(payload); + return true; } finally {