diff --git a/src/Plugins/BotSharp.Plugin.Twilio/Models/AgentTurnCallbacks.cs b/src/Plugins/BotSharp.Plugin.Twilio/Models/AgentTurnCallbacks.cs index ccd5e7740..087ac82c1 100644 --- a/src/Plugins/BotSharp.Plugin.Twilio/Models/AgentTurnCallbacks.cs +++ b/src/Plugins/BotSharp.Plugin.Twilio/Models/AgentTurnCallbacks.cs @@ -61,11 +61,36 @@ public class AgentTurnCallbacks public string? LastReplyLine { get; set; } public string? LastReplyHeard { get; set; } + private readonly List<(string Text, DateTime SaidUtc)> _fillers = []; + + /// + /// Notes a filler said while the agent works on the reply ("One moment, please."), to be kept in the history out of + /// the agent's context, so what the caller heard is all there + /// + public void NoteFillerSaid(string text) + { + lock (_fillers) + { + _fillers.Add((text, DateTime.UtcNow)); + } + } + /// - /// Set by the turn: called with each filler said while the agent works on the reply ("One moment, please."), which - /// the turn keeps in the history out of the agent's context, so what the caller heard is all there + /// The fillers said and not kept yet, oldest first; each is taken once /// - public Action? OnFillerSaid { get; set; } + /// + /// The turn takes them as it stores its reply, to keep them just ahead of it. The flow takes what is left once the + /// turn is over: the fillers of a turn that failed, ran past its time or stored no reply, lost otherwise. + /// + public List<(string Text, DateTime SaidUtc)> TakeFillers() + { + lock (_fillers) + { + var taken = _fillers.ToList(); + _fillers.Clear(); + return taken; + } + } /// /// Called on the turn once it has read the conversation's states ( is set), before the diff --git a/src/Plugins/BotSharp.Plugin.Twilio/Models/ConversationRelayLabels.cs b/src/Plugins/BotSharp.Plugin.Twilio/Models/ConversationRelayLabels.cs new file mode 100644 index 000000000..9214331a1 --- /dev/null +++ b/src/Plugins/BotSharp.Plugin.Twilio/Models/ConversationRelayLabels.cs @@ -0,0 +1,35 @@ +namespace BotSharp.Plugin.Twilio.Models; + +/// +/// What a line kept only as a record of what the caller heard was, kept as its dialog's MessageLabel: such lines look +/// alike otherwise (assistant, out of the agent's context), and a flow may treat them differently +/// +public static class ConversationRelayLabels +{ + /// + /// Said while the agent worked on a reply ("One moment, please.") + /// + public const string Filler = "twilio_relay_filler"; + + /// + /// Said to a caller gone quiet: the last question asked again ("Sorry, I didn't catch that. ..."), "Are you still + /// there?", or "Sorry, can you say it again?". It checks the caller is there, so it is no line of the agent's to say + /// again when they ask what was said. + /// + public const string IdlePrompt = "twilio_relay_idle_prompt"; + + /// + /// Said as the session hangs up on a caller who stayed quiet + /// + public const string IdleGoodbye = "twilio_relay_idle_goodbye"; + + /// + /// Said as the session hangs up on a turn that failed or ran past its time + /// + public const string Apology = "twilio_relay_apology"; + + /// + /// A line the caller cut into, said again once nothing followed the cut-in (ResumeAfterCutInSeconds) + /// + public const string SaidAgain = "twilio_relay_said_again"; +} diff --git a/src/Plugins/BotSharp.Plugin.Twilio/Services/ConversationRelaySession.cs b/src/Plugins/BotSharp.Plugin.Twilio/Services/ConversationRelaySession.cs index 73a5ce4fb..a323d7abc 100644 --- a/src/Plugins/BotSharp.Plugin.Twilio/Services/ConversationRelaySession.cs +++ b/src/Plugins/BotSharp.Plugin.Twilio/Services/ConversationRelaySession.cs @@ -39,6 +39,7 @@ public class ConversationRelaySession private readonly ConversationRelaySetting _relay; private readonly Func> _sendCallerMessage; private readonly ILogger _logger; + private readonly Func? _keepSaidLine; private readonly SemaphoreSlim _sendLock = new(1, 1); private readonly object _sync = new(); @@ -140,13 +141,20 @@ public class ConversationRelaySession /// Sends the caller's words to the agent and returns its reply; the callbacks receive tool-call indications /// and the streamed reply /// + /// + /// Keeps a line the session said on its own in the conversation history, with what it was + /// () and when it was said, so the history shows what the caller heard: the + /// apology of a failed turn, an idle prompt and its goodbye, a line said again after a cut-in, and the fillers of a + /// turn that kept none. A turn keeps its own reply, and a tool-call indication is the flow's to keep. + /// public ConversationRelaySession( WebSocket socket, string agentId, string conversationId, TwilioSetting settings, Func> sendCallerMessage, - ILogger logger) + ILogger logger, + Func? keepSaidLine = null) { _socket = socket; _agentId = agentId; @@ -155,6 +163,7 @@ public ConversationRelaySession( _relay = settings.ConversationRelay.For(agentId); _sendCallerMessage = sendCallerMessage; _logger = logger; + _keepSaidLine = keepSaidLine; } public async Task RunAsync(CancellationToken cancellationToken) @@ -522,7 +531,8 @@ private async Task ProcessPendingAsync() // 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 ? CloseUtteranceBeforeToolAsync : null, - RewriteReply = stream ? MarkWhatTheCallerCutOff : null + // Streamed or not: a reply the call ended before is marked as never said + RewriteReply = MarkWhatTheCallerCutOff }; CallerMessage message; @@ -604,7 +614,7 @@ private async Task ProcessPendingAsync() var timeoutSeconds = _relay.AiResponseTimeoutSeconds; _logger.LogWarning($"Agent did not reply within {timeoutSeconds}s in conversation {_conversationId}."); _timedOutTurn = turn; - await FailAsync(); + await FailAsync(callbacks); return; } reply = await turn; @@ -612,7 +622,7 @@ private async Task ProcessPendingAsync() catch (Exception ex) { _logger.LogError(ex, $"Agent turn failed in conversation {_conversationId}."); - await FailAsync(); + await FailAsync(callbacks); return; } finally @@ -624,6 +634,7 @@ private async Task ProcessPendingAsync() } watchdogCts.Cancel(); await SwallowAsync(watchdog); + await KeepFillersLeftAsync(callbacks); } await CloseUtteranceAsync(); @@ -673,7 +684,8 @@ await SpeakAsync(unstreamed, canSpeak: () => /// 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. The turn is told of each one said (OnFillerSaid). + /// said twice in a turn while unused ones remain. Each one said is noted on the turn's callbacks (NoteFillerSaid), to be + /// kept in the history. /// private async Task WatchResponseAsync(int seqNumber, AgentTurnCallbacks callbacks, CancellationToken cancellationToken) { @@ -698,16 +710,15 @@ private async Task WatchResponseAsync(int seqNumber, AgentTurnCallbacks callback var text = PickFillerText(step == 0 ? relay.ResponseWatchdogPrompts : relay.ResponseWatchdogLongPrompts, step == 0 ? DefaultWatchdogPrompts : DefaultWatchdogLongPrompts, said); - if (await SpeakFillerAsync(seqNumber, text, interval)) + if (await SpeakFillerAsync(seqNumber, text, interval, callbacks)) { said.Add(text); step++; - callbacks.OnFillerSaid?.Invoke(text); } } } - private async Task SpeakFillerAsync(int seqNumber, string text, TimeSpan interval) + private async Task SpeakFillerAsync(int seqNumber, string text, TimeSpan interval, AgentTurnCallbacks callbacks) { // Decided again once the send lock is held: while waiting for it the reply may have come in, or a streamed // line may have started or been closed, and then the silence this filler was meant for is over @@ -718,6 +729,9 @@ private async Task SpeakFillerAsync(int seqNumber, string text, TimeSpan i return false; } RememberUtterance(text, _saidCount + 1); + // Noted as it goes out, under the send lock: a turn that fails meanwhile finds every filler said before + // its apology + callbacks.NoteFillerSaid(text); return true; }, new { type = "text", token = text, last = true }, EstimateSpeech(text.Length)); @@ -840,25 +854,39 @@ private async Task WatchIdleAsync(CancellationToken cancellationToken) bool StillSilent() => !_isProcessing && _activity == activity; if (askAgain && question != null) { - await SpeakAsync(PromptText("say-it-again-question", DefaultAskAgainText).Replace("{0}", question), StillSilent); + await SayOwnLineAsync(PromptText("say-it-again-question", DefaultAskAgainText).Replace("{0}", question), + ConversationRelayLabels.IdlePrompt, StillSilent); } - else if (askAgain && relay.IdleAsksAgain && stage > 0) + // Nothing heard from the caller yet (no words, no keys): there is nothing for them to say again, only whether + // they are there. Why: silence after the greeting got "Sorry, can you say it again?" (local call simulation) + else if (askAgain && (relay.IdleAsksAgain && stage > 0 || activity == 0)) { - await SpeakPromptAsync("still-there", DefaultStillThereText, StillSilent); + await SayOwnLineAsync(PromptText("still-there", DefaultStillThereText), ConversationRelayLabels.IdlePrompt, StillSilent); } else if (askAgain) { - await SpeakPromptAsync($"say-it-again-{Random.Shared.Next(1, 5)}", DefaultSayItAgainText, StillSilent); + await SayOwnLineAsync(PromptText($"say-it-again-{Random.Shared.Next(1, 5)}", DefaultSayItAgainText), + ConversationRelayLabels.IdlePrompt, StillSilent); } else if (hangUp) { _logger.LogInformation($"Caller stayed silent, hanging up conversation {_conversationId}."); + string? goodbye = null; + var goodbyeUtc = DateTime.MinValue; if (relay.IdleAsksAgain) { - // The session is marked ended already, so nothing else can be said after it - await SendAsync(TextFrame(PromptText("idle-goodbye", DefaultIdleGoodbyeText), last: true, keepUninterrupted: false)); + // The session is marked ended already, so nothing else can be said after it, and nothing the caller + // says over it is answered: it plays to the end + goodbye = PromptText("idle-goodbye", DefaultIdleGoodbyeText); + await SendAsync(TextFrame(goodbye, last: true, keepUninterrupted: true)); + goodbyeUtc = DateTime.UtcNow; } await SendEndAsync(ConversationRelayHandoff.HangUp); + // Kept once the call is ended, so the storage doesn't hold up the hang-up + if (goodbye != null) + { + await KeepSaidAsync(goodbye, ConversationRelayLabels.IdleGoodbye, goodbyeUtc); + } } } } @@ -922,8 +950,7 @@ private async Task ResumeIfSilentAsync() _resumeLine = null; } - var said = false; - await SpeakAsync(line, canSpeak: () => + var said = await SpeakAsync(line, canSpeak: () => { if (_isProcessing || _activity != activity) { @@ -938,16 +965,16 @@ await SpeakAsync(line, canSpeak: () => NoteLine(line); // Set as it goes out: a cut-in into it, however soon, doesn't get it said a third time _lastWasResume = true; - said = true; return true; }); if (said) { _logger.LogInformation($"Nothing followed the caller's cut-in in conversation {_conversationId}; said the line again."); + await KeepSaidAsync(line, ConversationRelayLabels.SaidAgain, DateTime.UtcNow); } } - private async Task FailAsync() + private async Task FailAsync(AgentTurnCallbacks callbacks) { // A turn that timed out may still be streaming; keep its words out of the apology lock (_sync) @@ -955,8 +982,66 @@ private async Task FailAsync() _dropStream = true; } await CloseUtteranceAsync(); - await SpeakPromptAsync("error", DefaultErrorText); + var apology = PromptText("error", DefaultErrorText); + // The call ends after it, so nothing the caller says over it is answered: it plays to the end + var apologized = await SpeakAsync(apology, keepUninterrupted: true); + var apologyUtc = DateTime.UtcNow; await EndAsync(ConversationRelayHandoff.HangUp); + + // Kept once the call is ended, so the storage doesn't hold up the hang-up and no filler can come after the + // apology. The turn kept none of the fillers it took to fail: they go first, as they were said first. + await KeepFillersLeftAsync(callbacks); + if (apologized) + { + await KeepSaidAsync(apology, ConversationRelayLabels.Apology, apologyUtc); + } + } + + /// + /// Keeps the fillers of a turn that it didn't keep itself: it failed, ran past its time, or stored no reply. Call once + /// the turn can say nothing more. + /// + private async Task KeepFillersLeftAsync(AgentTurnCallbacks callbacks) + { + foreach (var (text, saidUtc) in callbacks.TakeFillers()) + { + await KeepSaidAsync(text, ConversationRelayLabels.Filler, saidUtc); + } + } + + /// + /// Says a line of the session's own, not the agent's, and keeps it in the history once it went out + /// + /// What the line is () + /// Extra condition, checked under once the send lock is held + private async Task SayOwnLineAsync(string text, string label, Func? canSpeak = null) + { + if (await SpeakAsync(text, canSpeak)) + { + await KeepSaidAsync(text, label, DateTime.UtcNow); + } + } + + /// + /// Keeps a line the session said on its own in the conversation history, so the history shows what the caller heard + /// + /// What the line is () + private async Task KeepSaidAsync(string text, string label, DateTime saidUtc) + { + if (_keepSaidLine == null || string.IsNullOrWhiteSpace(text)) + { + return; + } + + try + { + await _keepSaidLine(text.Trim(), label, saidUtc); + } + catch (Exception ex) + { + // Only the record of the line is lost: the call goes on + _logger.LogWarning(ex, $"Could not keep a line said in conversation {_conversationId}."); + } } /// @@ -1047,7 +1132,9 @@ await SendIfAsync(() => /// Text goes out faster than it is spoken, so the cut is placed where Twilio says playback stopped; when Twilio /// doesn't say, or what it says is not the start of what went out, the cut falls where the sending stopped. A reply /// dropped because the caller went on talking first is marked as never said, so the next turn doesn't take it as - /// heard. + /// heard, and so is one the call ended before: the caller hung up while the agent worked, or the turn ran past + /// AiResponseTimeoutSeconds and the session apologized and hung up (it came back after the apology, and the history + /// showed it as said). /// private string MarkWhatTheCallerCutOff(string content) { @@ -1062,6 +1149,11 @@ private string MarkWhatTheCallerCutOff(string content) ? "[the caller went on talking before anything was said]" : $"[the caller went on talking before this was said; they did not hear: \"{final}\"]"; } + // Nothing of it went out before the call ended, and nothing is said once it has (SpeakAsync) + if (_ended && !_streamedThisTurn && final.Length > 0) + { + return $"[the call ended before this was said; the caller did not hear: \"{final}\"]"; + } if (!_cutInMidReply || final.Length == 0) { return content; @@ -1203,14 +1295,15 @@ await SpeakIfAsync(() => !_ended && (canSpeak?.Invoke() ?? true), /// Extra condition, checked under once the send lock is held /// The caller must hear all of it, so it plays to the end whoever speaks over it - private async Task SpeakAsync(string? text, Func? canSpeak = null, bool keepUninterrupted = false) + /// Whether it went out + private async Task SpeakAsync(string? text, Func? canSpeak = null, bool keepUninterrupted = false) { if (string.IsNullOrWhiteSpace(text)) { - return; + return false; } - await SpeakIfAsync(() => + return await SpeakIfAsync(() => { if (_ended || !(canSpeak?.Invoke() ?? true)) { diff --git a/src/Plugins/BotSharp.Plugin.Twilio/Services/TwilioConversationProcessor.cs b/src/Plugins/BotSharp.Plugin.Twilio/Services/TwilioConversationProcessor.cs index a58a5e652..3681ee0ae 100644 --- a/src/Plugins/BotSharp.Plugin.Twilio/Services/TwilioConversationProcessor.cs +++ b/src/Plugins/BotSharp.Plugin.Twilio/Services/TwilioConversationProcessor.cs @@ -47,8 +47,6 @@ public static class TwilioConversationProcessor HookEmitter.Emit(sp, hook => hook.OnAuthenticate(request), message.AgentId); AssistantMessage? reply = null; - // Said while the agent worked on the reply; kept in the history just ahead of it - var fillers = new List<(string Text, DateTime SaidUtc)>(); var inputMsg = new RoleDialogModel(AgentRole.User, message.Content); var conv = sp.GetRequiredService(); @@ -134,13 +132,6 @@ public static class TwilioConversationProcessor { conv.States.RemoveState(ConversationRelayStates.LastReplyCutOff); } - callbacks.OnFillerSaid = filler => - { - lock (fillers) - { - fillers.Add((filler, DateTime.UtcNow)); - } - }; if (callbacks.OnTurnStarted != null) { await callbacks.OnTurnStarted(); @@ -163,8 +154,10 @@ await conv.SendMessage(message.AgentId, replyMessage: await BuildPostbackMessageModel(conv, message), async msg => { - // The reply is appended once this returns, so the fillers go in ahead of it - await StoreFillers(sp, message, inputMsg.MessageId, fillers); + // Said while the agent worked on the reply: the reply is appended once this returns, so they go in + // ahead of it + await KeepSaidLines(sp, message.AgentId, message.ConversationId, inputMsg.MessageId, + ConversationRelayLabels.Filler, callbacks?.TakeFillers() ?? []); // What the caller is told, which a rewrite below must not change var spoken = msg.Content; RewriteForHistory(msg, callbacks); @@ -255,48 +248,47 @@ await sp.GetRequiredService().UpdateConversationMessage(co } /// - /// Keeps the fillers said while the agent worked on this reply in the history, out of the agent's context: they are - /// what the caller heard, and a reply to one ("Sure.") makes sense only next to it + /// Keeps lines the caller heard in the history, out of the agent's context: the fillers said while the agent worked + /// on a reply, kept just ahead of it (a reply to one, "Sure.", makes sense only next to it), and what a + /// ConversationRelay session says on its own (an apology, an idle prompt, a line said again) /// - private static async Task StoreFillers(IServiceProvider sp, CallerMessage message, string messageId, - List<(string Text, DateTime SaidUtc)> fillers) + /// The caller message the lines go with; null gives each line a message of its own + /// What the lines were (), kept as their MessageLabel + public static async Task KeepSaidLines(IServiceProvider sp, string agentId, string conversationId, string? messageId, + string label, IReadOnlyCollection<(string Text, DateTime SaidUtc)> lines) { - List<(string Text, DateTime SaidUtc)> said; - lock (fillers) - { - said = [.. fillers]; - fillers.Clear(); - } - if (said.Count == 0) + if (lines.Count == 0) { return; } - var dialogs = said.Select(f => new RoleDialogModel(AgentRole.Assistant, f.Text) + var dialogs = lines.Select(line => new RoleDialogModel(AgentRole.Assistant, line.Text) { - CurrentAgentId = message.AgentId, - MessageId = messageId, - CreatedAt = f.SaidUtc, + CurrentAgentId = agentId, + MessageId = messageId ?? Guid.NewGuid().ToString(), + MessageLabel = label, + CreatedAt = line.SaidUtc, ExcludeFromContext = true - }); + }).ToList(); try { - await sp.GetRequiredService().Append(message.ConversationId, dialogs); + await sp.GetRequiredService().Append(conversationId, dialogs); } catch (Exception ex) { - // The reply still goes out and is stored; only the record of the fillers is lost + // The call goes on, and a reply is still stored; only the record of these lines is lost sp.GetRequiredService().CreateLogger(typeof(TwilioConversationProcessor)) - .LogWarning(ex, $"Could not keep the fillers said in conversation {message.ConversationId}."); + .LogWarning(ex, $"Could not keep what was said in conversation {conversationId}."); } } private static async Task BuildPostbackMessageModel( IConversationService conv, CallerMessage message) { - var messages = await conv.GetDialogHistory(1); - if (!messages.Any()) return null; - var lastMessage = messages[0]; + // A line kept only as a record of what the caller heard (KeepSaidLines) may have been said after the reply that + // asked for the postback + var lastMessage = (await conv.GetDialogHistory()).LastOrDefault(x => !x.ExcludeFromContext); + if (lastMessage == null) return null; if (string.IsNullOrEmpty(lastMessage.PostbackFunctionName)) return null; return new PostbackMessageModel { diff --git a/src/Plugins/BotSharp.Plugin.Twilio/Settings/ConversationRelaySetting.cs b/src/Plugins/BotSharp.Plugin.Twilio/Settings/ConversationRelaySetting.cs index ff34e12c2..f4b34420b 100644 --- a/src/Plugins/BotSharp.Plugin.Twilio/Settings/ConversationRelaySetting.cs +++ b/src/Plugins/BotSharp.Plugin.Twilio/Settings/ConversationRelaySetting.cs @@ -120,7 +120,8 @@ public class ConversationRelaySetting /// /// The first idle prompt asks the question the agent's last line asked again ("Sorry, I didn't catch that. ..."), the /// next ones ask if the caller is still there, and the hang-up says goodbye (Twilio:PromptTexts say-it-again-question, - /// still-there and idle-goodbye). Off, every idle prompt is a "say it again" one and the hang-up is silent. + /// still-there and idle-goodbye). Off, every idle prompt is a "say it again" one and the hang-up is silent. Either + /// way, a caller who has said nothing yet is only asked if they are still there. /// public bool IdleAsksAgain { get; set; } diff --git a/src/Plugins/BotSharp.Plugin.Twilio/TwilioConversationRelayMiddleware.cs b/src/Plugins/BotSharp.Plugin.Twilio/TwilioConversationRelayMiddleware.cs index 9b8374085..baf0442a0 100644 --- a/src/Plugins/BotSharp.Plugin.Twilio/TwilioConversationRelayMiddleware.cs +++ b/src/Plugins/BotSharp.Plugin.Twilio/TwilioConversationRelayMiddleware.cs @@ -49,7 +49,8 @@ public async Task Invoke(HttpContext httpContext) var scopeFactory = services.GetRequiredService(); var session = new ConversationRelaySession(socket, agentId, conversationId, settings, (message, callbacks) => SendCallerMessageAsync(scopeFactory, message, callbacks), - _logger); + _logger, + (line, label, saidUtc) => KeepSaidLineAsync(scopeFactory, agentId, conversationId, line, label, saidUtc)); await services.GetRequiredService() .RunAsync(conversationId, () => session.RunAsync(httpContext.RequestAborted)); } @@ -61,6 +62,17 @@ await services.GetRequiredService() return await TwilioConversationProcessor.SendCallerMessageAsync(scope.ServiceProvider, message, callbacks); } + /// + /// Keeps a line the session said on its own in the history, in a scope of its own: it is said outside any turn's + /// + private static async Task KeepSaidLineAsync( + IServiceScopeFactory scopeFactory, string agentId, string conversationId, string line, string label, DateTime saidUtc) + { + using var scope = scopeFactory.CreateScope(); + await TwilioConversationProcessor.KeepSaidLines(scope.ServiceProvider, agentId, conversationId, messageId: null, + label, [(line, saidUtc)]); + } + /// /// Twilio signs the WebSocket handshake with X-Twilio-Signature over the wss URL from the TwiML. /// diff --git a/tests/BotSharp.Core.UnitTests/BotSharp.Core.UnitTests.csproj b/tests/BotSharp.Core.UnitTests/BotSharp.Core.UnitTests.csproj index f8c422235..fc1b8c1c2 100644 --- a/tests/BotSharp.Core.UnitTests/BotSharp.Core.UnitTests.csproj +++ b/tests/BotSharp.Core.UnitTests/BotSharp.Core.UnitTests.csproj @@ -51,6 +51,8 @@ + + diff --git a/tests/BotSharp.Core.UnitTests/Plugins/ConversationRelayKeptLinesTests.cs b/tests/BotSharp.Core.UnitTests/Plugins/ConversationRelayKeptLinesTests.cs new file mode 100644 index 000000000..75521cd5f --- /dev/null +++ b/tests/BotSharp.Core.UnitTests/Plugins/ConversationRelayKeptLinesTests.cs @@ -0,0 +1,461 @@ +using System.Collections.Concurrent; +using System.Net.WebSockets; +using System.Text; +using System.Text.Json; +using System.Threading.Channels; +using BotSharp.Abstraction.Agents.Enums; +using BotSharp.Abstraction.Conversations; +using BotSharp.Abstraction.Conversations.Models; +using BotSharp.Plugin.Twilio.Models; +using BotSharp.Plugin.Twilio.Services; +using BotSharp.Plugin.Twilio.Settings; +using Microsoft.Extensions.DependencyInjection; +using Microsoft.Extensions.Logging.Abstractions; +using Moq; +using Xunit; + +namespace BotSharp.Core.UnitTests.Plugins; + +/// +/// A ConversationRelay session keeps every line it says on its own in the conversation history, labeled with what it +/// was, so the history shows what the caller heard: the apology of a failed turn and the fillers said before it, idle +/// prompts and their goodbye, and a line said again after a cut-in. A line that never went out is not kept, a turn's +/// fillers are kept only once, and the lines said as the session hangs up play to the end. +/// +public class ConversationRelayKeptLinesTests +{ + private const string AgentId = "relay-agent"; + private const string Filler = "One moment."; + private const string Apology = "Sorry, we hit a snag."; + private static readonly TimeSpan Patience = TimeSpan.FromSeconds(15); + + [Fact] + public async Task A_failed_turn_keeps_its_fillers_then_the_apology() + { + var call = Call.Start(Settings(), async (self, callbacks) => + { + await WaitUntil(() => self.Socket.SentTexts().Contains(Filler)); + throw new InvalidOperationException("The model is down."); + }); + + await WaitUntil(() => call.Kept.Count == 2); + await call.HangUpAsync(); + + Assert.Equal(new[] { Filler, Apology }, call.Kept.Select(line => line.Text)); + Assert.Equal(new[] { ConversationRelayLabels.Filler, ConversationRelayLabels.Apology }, call.Kept.Select(line => line.Label)); + Assert.True(call.Kept.First().SaidUtc <= call.Kept.Last().SaidUtc); + Assert.Equal(new[] { Filler, Apology }, call.Socket.SentTexts()); + Assert.True(call.Socket.SentEnd); + // The call ends after the apology: the caller can't cut it off, a filler is left to the call's setting + Assert.Equal(new bool?[] { null, false }, call.Socket.SentInterruptible()); + } + + [Fact] + public async Task A_timed_out_turn_that_finally_replies_keeps_no_filler_again_and_its_reply_as_never_said() + { + var lateTake = new TaskCompletionSource>(); + string? stored = null; + var call = Call.Start(Settings(relay => relay.AiResponseTimeoutSeconds = 1), async (self, callbacks) => + { + await WaitUntil(() => self.Socket.SentTexts().Contains(Filler)); + await WaitUntil(() => self.Kept.Count == 2); + // Where TwilioConversationProcessor takes them, and asks what to store, as it stores the reply + stored = callbacks.RewriteReply!("Here you go."); + lateTake.SetResult(callbacks.TakeFillers()); + return new AssistantMessage { Content = "Here you go." }; + }); + + await WaitUntil(() => call.Kept.Count == 2); + Assert.Empty(await lateTake.Task.WaitAsync(Patience)); + await call.HangUpAsync(); + + Assert.Equal(new[] { Filler, Apology }, call.Kept.Select(line => line.Text)); + Assert.DoesNotContain("Here you go.", call.Socket.SentTexts()); + Assert.Equal("[the call ended before this was said; the caller did not hear: \"Here you go.\"]", stored); + } + + [Fact] + public async Task A_reply_that_comes_back_after_the_caller_hung_up_is_kept_as_never_said() + { + const string reply = "Could you tell me what's broken?"; + var started = new TaskCompletionSource(); + var hungUp = new TaskCompletionSource(); + string? stored = null; + var call = Call.Start(Settings(relay => relay.ResponseWatchdogIntervals = []), async (self, callbacks) => + { + started.SetResult(); + await hungUp.Task; + stored = callbacks.RewriteReply!(reply); + return new AssistantMessage { Content = reply }; + }); + + await started.Task.WaitAsync(Patience); + call.Socket.Close(); + // The session marks the call ended as soon as it reads the close + await Task.Delay(TimeSpan.FromMilliseconds(300)); + hungUp.SetResult(); + await call.Run.WaitAsync(Patience); + + Assert.Equal($"[the call ended before this was said; the caller did not hear: \"{reply}\"]", stored); + Assert.Empty(call.Socket.SentTexts()); + } + + [Fact] + public async Task Fillers_a_turn_kept_with_its_reply_are_not_kept_again() + { + List<(string Text, DateTime SaidUtc)>? taken = null; + string? stored = null; + var call = Call.Start(Settings(), async (self, callbacks) => + { + await WaitUntil(() => self.Socket.SentTexts().Contains(Filler)); + taken = callbacks.TakeFillers(); + stored = callbacks.RewriteReply!("Your request is in."); + return new AssistantMessage { Content = "Your request is in." }; + }); + + await WaitUntil(() => call.Socket.SentTexts().Contains("Your request is in.")); + await call.HangUpAsync(); + + Assert.Equal(new[] { Filler }, taken!.Select(line => line.Text)); + Assert.Empty(call.Kept); + // A reply said while the call is on is stored as it is + Assert.Equal("Your request is in.", stored); + } + + [Fact] + public async Task A_turn_that_stored_no_reply_leaves_its_fillers_to_the_session() + { + var call = Call.Start(Settings(), async (self, callbacks) => + { + await WaitUntil(() => self.Socket.SentTexts().Contains(Filler)); + return null; + }); + + await WaitUntil(() => call.Kept.Count == 1); + await call.HangUpAsync(); + + Assert.Equal(new[] { (Filler, ConversationRelayLabels.Filler) }, call.Kept.Select(line => (line.Text, line.Label))); + } + + [Fact] + public async Task A_turn_failing_after_the_caller_hung_up_keeps_its_fillers_but_no_apology() + { + var fail = new TaskCompletionSource(); + var call = Call.Start(Settings(), async (self, callbacks) => + { + await WaitUntil(() => self.Socket.SentTexts().Contains(Filler)); + await fail.Task; + throw new InvalidOperationException("The model is down."); + }); + + await WaitUntil(() => call.Socket.SentTexts().Contains(Filler)); + call.Socket.Close(); + // The session marks the call ended as soon as it reads the close + await Task.Delay(TimeSpan.FromMilliseconds(300)); + fail.SetResult(); + await call.Run.WaitAsync(Patience); + + Assert.Equal(new[] { Filler }, call.Kept.Select(line => line.Text)); + Assert.DoesNotContain(Apology, call.Socket.SentTexts()); + } + + [Fact] + public async Task A_line_that_cannot_be_kept_does_not_stop_the_hang_up() + { + var call = Call.Start(Settings(), (_, _) => throw new InvalidOperationException("The model is down."), + keep: (_, _, _) => throw new InvalidOperationException("The storage is down.")); + + await WaitUntil(() => call.Socket.SentEnd); + await call.HangUpAsync(); + + Assert.Equal(new[] { Apology }, call.Socket.SentTexts()); + } + + [Fact] + public async Task Idle_prompts_and_the_goodbye_are_kept_as_said() + { + var call = Call.Start(Settings(relay => + { + relay.ResponseWatchdogIntervals = []; + relay.IdleAsksAgain = true; + relay.IdleTimeouts = [0.1]; + relay.MaxIdlePrompts = 2; + }), + (_, _) => Task.FromResult(new AssistantMessage { Content = "Which unit?" })); + + await WaitUntil(() => call.Socket.SentEnd); + await WaitUntil(() => call.Kept.Count == 3); + await call.HangUpAsync(); + + // The reply is the turn's to keep; the prompts are kept with their text, not their names + Assert.Equal(new[] { "Pardon? Which unit?", "Hello?", "Bye now." }, call.Kept.Select(line => line.Text)); + Assert.Equal(new[] { ConversationRelayLabels.IdlePrompt, ConversationRelayLabels.IdlePrompt, ConversationRelayLabels.IdleGoodbye }, + call.Kept.Select(line => line.Label)); + Assert.Equal(new[] { "Which unit?", "Pardon? Which unit?", "Hello?", "Bye now." }, call.Socket.SentTexts()); + // A prompt waits for the caller to speak, so they can cut it off; the goodbye plays to the end + Assert.Equal(new bool?[] { null, null, null, false }, call.Socket.SentInterruptible()); + } + + [Fact] + public async Task A_say_it_again_prompt_is_kept_with_its_text() + { + var call = Call.Start(Settings(relay => + { + relay.ResponseWatchdogIntervals = []; + relay.IdleTimeouts = [0.1]; + relay.MaxIdlePrompts = 1; + }), + (_, _) => Task.FromResult(null)); + + await WaitUntil(() => call.Socket.SentEnd); + await call.HangUpAsync(); + + // Without IdleAsksAgain the hang-up says nothing + Assert.Equal(new[] { ("Say that again?", ConversationRelayLabels.IdlePrompt) }, call.Kept.Select(line => (line.Text, line.Label))); + Assert.Equal(new[] { "Say that again?" }, call.Socket.SentTexts()); + } + + [Theory] + [InlineData(false)] + [InlineData(true)] + public async Task A_caller_who_has_said_nothing_yet_is_asked_if_they_are_there_not_to_say_it_again(bool asksAgain) + { + var call = Call.Start(Settings(relay => + { + relay.ResponseWatchdogIntervals = []; + relay.IdleAsksAgain = asksAgain; + relay.IdleTimeouts = [0.1]; + relay.MaxIdlePrompts = 1; + }), + (_, _) => Task.FromResult(null), prompt: null); + + await WaitUntil(() => call.Socket.SentEnd); + await call.HangUpAsync(); + + Assert.Equal(new[] { ("Hello?", ConversationRelayLabels.IdlePrompt) }, call.Kept.Select(line => (line.Text, line.Label)).Take(1)); + Assert.Equal("Hello?", call.Socket.SentTexts()[0]); + } + + [Fact] + public async Task A_line_said_again_after_a_cut_in_is_kept() + { + const string line = "Is the address right?"; + var call = Call.Start(Settings(relay => + { + relay.ResponseWatchdogIntervals = []; + relay.ResumeAfterCutInSeconds = 0.2; + }), + (_, _) => Task.FromResult(new AssistantMessage { Content = line })); + + await WaitUntil(() => call.Socket.SentTexts().Contains(line)); + call.Socket.Receive(new { type = "interrupt", utteranceUntilInterrupt = "Is the", durationUntilInterruptMs = "400" }); + + await WaitUntil(() => call.Kept.Count == 1); + await call.HangUpAsync(); + + Assert.Equal(new[] { (line, ConversationRelayLabels.SaidAgain) }, call.Kept.Select(kept => (kept.Text, kept.Label))); + Assert.Equal(new[] { line, line }, call.Socket.SentTexts()); + Assert.Equal(new bool?[] { null, null }, call.Socket.SentInterruptible()); + } + + [Fact] + public async Task Kept_lines_are_the_agent_s_out_of_its_context_at_the_time_they_were_said() + { + var stored = new List(); + var storage = new Mock(); + storage.Setup(s => s.Append("conv-1", It.IsAny>())) + .Callback>((_, dialogs) => stored.AddRange(dialogs)) + .Returns(Task.CompletedTask); + var sp = new ServiceCollection().AddLogging().AddSingleton(storage.Object).BuildServiceProvider(); + var said = new DateTime(2026, 11, 13, 1, 31, 0, DateTimeKind.Utc); + + await TwilioConversationProcessor.KeepSaidLines(sp, AgentId, "conv-1", messageId: null, + ConversationRelayLabels.IdlePrompt, [("Hello?", said), ("Say that again?", said.AddSeconds(2))]); + await TwilioConversationProcessor.KeepSaidLines(sp, AgentId, "conv-1", messageId: "msg-1", + ConversationRelayLabels.Filler, [(Filler, said)]); + + Assert.Equal(new[] { "Hello?", "Say that again?", Filler }, stored.Select(d => d.Content)); + Assert.Equal(new[] { ConversationRelayLabels.IdlePrompt, ConversationRelayLabels.IdlePrompt, ConversationRelayLabels.Filler }, + stored.Select(d => d.MessageLabel)); + Assert.All(stored, d => + { + Assert.Equal(AgentRole.Assistant, d.Role); + Assert.True(d.ExcludeFromContext); + Assert.Equal(AgentId, d.CurrentAgentId); + }); + Assert.Equal(new[] { said, said.AddSeconds(2), said }, stored.Select(d => d.CreatedAt)); + // A line of the session's own is a message of its own; a filler goes with the caller message it was said for + Assert.NotEqual(stored[0].MessageId, stored[1].MessageId); + Assert.Equal("msg-1", stored[2].MessageId); + } + + [Fact] + public async Task A_storage_failure_loses_only_the_record() + { + var storage = new Mock(); + storage.Setup(s => s.Append(It.IsAny(), It.IsAny>())) + .ThrowsAsync(new InvalidOperationException("The storage is down.")); + var sp = new ServiceCollection().AddLogging().AddSingleton(storage.Object).BuildServiceProvider(); + + await TwilioConversationProcessor.KeepSaidLines(sp, AgentId, "conv-1", messageId: null, ConversationRelayLabels.Apology, + [(Apology, DateTime.UtcNow)]); + + storage.Verify(s => s.Append("conv-1", It.IsAny>()), Times.Once); + } + + private static TwilioSetting Settings(Action? configure = null) + { + var relay = new ConversationRelaySetting + { + Enabled = true, + ResponseWatchdogIntervals = [0.2], + ResponseWatchdogAgentIds = [AgentId], + ResponseWatchdogPrompts = ["hold-on-1"] + }; + configure?.Invoke(relay); + return new TwilioSetting + { + PromptTexts = new Dictionary + { + ["hold-on-1"] = Filler, + ["error"] = Apology, + ["say-it-again-question"] = "Pardon? {0}", + ["still-there"] = "Hello?", + ["idle-goodbye"] = "Bye now.", + ["say-it-again-1"] = "Say that again?", + ["say-it-again-2"] = "Say that again?", + ["say-it-again-3"] = "Say that again?", + ["say-it-again-4"] = "Say that again?" + }, + ConversationRelay = relay + }; + } + + private static async Task WaitUntil(Func condition) + { + var giveUp = DateTime.UtcNow + Patience; + while (!condition()) + { + if (DateTime.UtcNow > giveUp) + { + throw new TimeoutException("The call never got there."); + } + await Task.Delay(TimeSpan.FromMilliseconds(20)); + } + } + + /// + /// A call on a session, with the lines it kept + /// + private sealed class Call + { + public FakeWebSocket Socket { get; } = new(); + public ConcurrentQueue<(string Text, string Label, DateTime SaidUtc)> Kept { get; } = new(); + public Task Run { get; private set; } = Task.CompletedTask; + + /// What the caller says first; null says nothing + public static Call Start(TwilioSetting settings, Func> turn, + Func? keep = null, string? prompt = "My sink is leaking") + { + var call = new Call(); + var session = new ConversationRelaySession(call.Socket, AgentId, "conv-1", settings, + (_, callbacks) => turn(call, callbacks), NullLogger.Instance, + keep ?? ((text, label, saidUtc) => + { + call.Kept.Enqueue((text, label, saidUtc)); + return Task.CompletedTask; + })); + call.Run = session.RunAsync(CancellationToken.None); + call.Socket.Receive(new { type = "setup", callSid = "CA1", from = "+15550100", to = "+15550199", direction = "inbound" }); + if (prompt != null) + { + call.Socket.Receive(new { type = "prompt", voicePrompt = prompt, last = true }); + } + return call; + } + + public async Task HangUpAsync() + { + Socket.Close(); + await Run.WaitAsync(Patience); + } + } + + private sealed class FakeWebSocket : WebSocket + { + private readonly Channel _incoming = Channel.CreateUnbounded(); + private readonly ConcurrentQueue _sent = new(); + private WebSocketState _state = WebSocketState.Open; + + public void Receive(object message) => _incoming.Writer.TryWrite(JsonSerializer.SerializeToUtf8Bytes(message)); + + public void Close() => _incoming.Writer.TryWrite(null); + + /// + /// The text said, in order: each text frame's token, without the empty ones that close a streamed utterance + /// + public List SentTexts() + { + return TextFrames().Select(frame => frame.GetProperty("token").GetString()!).ToList(); + } + + /// + /// Each text said's interruptible flag, in order: null when the frame leaves it to the call's setting + /// + public List SentInterruptible() + { + return TextFrames() + .Select(frame => frame.TryGetProperty("interruptible", out var flag) ? flag.GetBoolean() : (bool?)null) + .ToList(); + } + + private IEnumerable TextFrames() + { + return Frames() + .Where(frame => frame.GetProperty("type").GetString() == "text") + .Where(frame => !string.IsNullOrEmpty(frame.GetProperty("token").GetString())); + } + + public bool SentEnd => Frames().Any(frame => frame.GetProperty("type").GetString() == "end"); + + private IEnumerable Frames() => _sent.Select(json => JsonDocument.Parse(json).RootElement); + + public override async Task ReceiveAsync(ArraySegment buffer, CancellationToken cancellationToken) + { + var data = await _incoming.Reader.ReadAsync(cancellationToken); + if (data == null) + { + _state = WebSocketState.CloseReceived; + return new WebSocketReceiveResult(0, WebSocketMessageType.Close, true); + } + data.CopyTo(buffer.Array!, buffer.Offset); + return new WebSocketReceiveResult(data.Length, WebSocketMessageType.Text, true); + } + + public override Task SendAsync(ArraySegment buffer, WebSocketMessageType messageType, bool endOfMessage, + CancellationToken cancellationToken) + { + _sent.Enqueue(Encoding.UTF8.GetString(buffer.Array!, buffer.Offset, buffer.Count)); + return Task.CompletedTask; + } + + public override Task CloseOutputAsync(WebSocketCloseStatus closeStatus, string? statusDescription, CancellationToken cancellationToken) + { + _state = WebSocketState.Closed; + return Task.CompletedTask; + } + + public override Task CloseAsync(WebSocketCloseStatus closeStatus, string? statusDescription, CancellationToken cancellationToken) + { + _state = WebSocketState.Closed; + return Task.CompletedTask; + } + + public override WebSocketCloseStatus? CloseStatus => _state == WebSocketState.Closed ? WebSocketCloseStatus.NormalClosure : null; + public override string? CloseStatusDescription => null; + public override WebSocketState State => _state; + public override string? SubProtocol => null; + public override void Abort() => _state = WebSocketState.Aborted; + public override void Dispose() { } + } +}