From 79879f8b10ddbcb24654fadfbcb34eedd8c03de5 Mon Sep 17 00:00:00 2001 From: "mars.yu" <18671933+yuyixg@users.noreply.github.com> Date: Fri, 13 Nov 2026 10:47:23 +0800 Subject: [PATCH 1/3] fix(twilio): keep the lines a relay session says on its own in the history A dev call's last turn failed, the relay apologized and hung up, but the conversation ends with the caller's "Bye" and shows nothing the caller heard after it. What the session says on its own went out without being stored: - The apology of a failed or timed-out turn, and the fillers said before it: fillers were stored only with a reply, so a turn that threw or ran past AiResponseTimeoutSeconds lost them - Idle prompts (say-it-again-question, still-there, say-it-again-N) and the idle goodbye - A line said again after a cut-in nothing followed (ResumeAfterCutInSeconds) The session now takes a keepSaidLine callback, and the middleware gives it one that appends an assistant dialog out of the agent's context, at the time it was said, in a scope of its own, logging storage errors. Only a line that went out is kept, with its text rather than its prompt name, and a hang-up's lines only after the end frame, so the storage never holds up the hang-up. Fillers are noted on the turn's callbacks as they go out and taken once: by the turn with its reply, or by the session for a turn that failed, timed out or stored no reply. Tool-call indications stay the flow's to keep. The postback a keypad entry answers is read from the last dialog in the agent's context, as a kept line may now come after the reply that asked for it. Co-Authored-By: Claude Opus 5.5 --- .../Models/AgentTurnCallbacks.cs | 31 +- .../Services/ConversationRelaySession.cs | 115 +++++- .../Services/TwilioConversationProcessor.cs | 56 ++- .../TwilioConversationRelayMiddleware.cs | 14 +- .../BotSharp.Core.UnitTests.csproj | 2 + .../ConversationRelayKeptLinesTests.cs | 381 ++++++++++++++++++ 6 files changed, 543 insertions(+), 56 deletions(-) create mode 100644 tests/BotSharp.Core.UnitTests/Plugins/ConversationRelayKeptLinesTests.cs 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/Services/ConversationRelaySession.cs b/src/Plugins/BotSharp.Plugin.Twilio/Services/ConversationRelaySession.cs index 73a5ce4fb..3db2d3b90 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 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) @@ -604,7 +613,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 +621,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 +633,7 @@ private async Task ProcessPendingAsync() } watchdogCts.Cancel(); await SwallowAsync(watchdog); + await KeepFillersLeftAsync(callbacks); } await CloseUtteranceAsync(); @@ -673,7 +683,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 +709,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 +728,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 +853,34 @@ 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), StillSilent); } else if (askAgain && relay.IdleAsksAgain && stage > 0) { - await SpeakPromptAsync("still-there", DefaultStillThereText, StillSilent); + await SayOwnLineAsync(PromptText("still-there", DefaultStillThereText), 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), 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)); + goodbye = PromptText("idle-goodbye", DefaultIdleGoodbyeText); + await SendAsync(TextFrame(goodbye, last: true, keepUninterrupted: false)); + 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, goodbyeUtc); + } } } } @@ -922,8 +944,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 +959,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, 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 +976,63 @@ private async Task FailAsync() _dropStream = true; } await CloseUtteranceAsync(); - await SpeakPromptAsync("error", DefaultErrorText); + var apology = PromptText("error", DefaultErrorText); + var apologized = await SpeakAsync(apology); + 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, 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, saidUtc); + } + } + + /// + /// Says a line of the session's own, not the agent's, and keeps it in the history once it went out + /// + /// Extra condition, checked under once the send lock is held + private async Task SayOwnLineAsync(string text, Func? canSpeak = null) + { + if (await SpeakAsync(text, canSpeak)) + { + await KeepSaidAsync(text, DateTime.UtcNow); + } + } + + /// + /// Keeps a line the session said on its own in the conversation history, so the history shows what the caller heard + /// + private async Task KeepSaidAsync(string text, DateTime saidUtc) + { + if (_keepSaidLine == null || string.IsNullOrWhiteSpace(text)) + { + return; + } + + try + { + await _keepSaidLine(text.Trim(), 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}."); + } } /// @@ -1203,14 +1279,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..8f64d1504 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, + callbacks?.TakeFillers() ?? []); // What the caller is told, which a rewrite below must not change var spoken = msg.Content; RewriteForHistory(msg, callbacks); @@ -255,48 +248,45 @@ 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 + public static async Task KeepSaidLines(IServiceProvider sp, string agentId, string conversationId, string? messageId, + 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(), + 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/TwilioConversationRelayMiddleware.cs b/src/Plugins/BotSharp.Plugin.Twilio/TwilioConversationRelayMiddleware.cs index 9b8374085..05db1deb4 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, saidUtc) => KeepSaidLineAsync(scopeFactory, agentId, conversationId, line, 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, DateTime saidUtc) + { + using var scope = scopeFactory.CreateScope(); + await TwilioConversationProcessor.KeepSaidLines(scope.ServiceProvider, agentId, conversationId, messageId: null, + [(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..431bd4b12 --- /dev/null +++ b/tests/BotSharp.Core.UnitTests/Plugins/ConversationRelayKeptLinesTests.cs @@ -0,0 +1,381 @@ +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, 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, and a turn's fillers are kept only once. +/// +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.True(call.Kept.First().SaidUtc <= call.Kept.Last().SaidUtc); + Assert.Equal(new[] { Filler, Apology }, call.Socket.SentTexts()); + Assert.True(call.Socket.SentEnd); + } + + [Fact] + public async Task A_timed_out_turn_leaves_no_filler_to_keep_again_when_it_finally_replies() + { + var lateTake = new TaskCompletionSource>(); + 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 as it stores the reply + 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()); + } + + [Fact] + public async Task Fillers_a_turn_kept_with_its_reply_are_not_kept_again() + { + List<(string Text, DateTime SaidUtc)>? taken = null; + var call = Call.Start(Settings(), async (self, callbacks) => + { + await WaitUntil(() => self.Socket.SentTexts().Contains(Filler)); + taken = callbacks.TakeFillers(); + 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); + } + + [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 }, call.Kept.Select(line => line.Text)); + } + + [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[] { "Which unit?", "Pardon? Which unit?", "Hello?", "Bye now." }, call.Socket.SentTexts()); + } + + [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), prompt: null); + + await WaitUntil(() => call.Socket.SentEnd); + await call.HangUpAsync(); + + // Without IdleAsksAgain the hang-up says nothing + Assert.Equal(new[] { "Say that again?" }, call.Kept.Select(line => line.Text)); + Assert.Equal(new[] { "Say that again?" }, call.Socket.SentTexts()); + } + + [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 }, call.Kept.Select(kept => kept.Text)); + Assert.Equal(new[] { line, line }, call.Socket.SentTexts()); + } + + [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, + [(Apology, said), ("Bye now.", said.AddSeconds(2))]); + await TwilioConversationProcessor.KeepSaidLines(sp, AgentId, "conv-1", messageId: "msg-1", [(Filler, said)]); + + Assert.Equal(new[] { Apology, "Bye now.", Filler }, stored.Select(d => d.Content)); + 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, [(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, 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, saidUtc) => + { + call.Kept.Enqueue((text, 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 Frames() + .Where(frame => frame.GetProperty("type").GetString() == "text") + .Select(frame => frame.GetProperty("token").GetString() ?? string.Empty) + .Where(token => token.Length > 0) + .ToList(); + } + + 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() { } + } +} From d64dd5d95550701766cf37e8606c6d2087e100f8 Mon Sep 17 00:00:00 2001 From: "mars.yu" <18671933+yuyixg@users.noreply.github.com> Date: Fri, 13 Nov 2026 11:22:29 +0800 Subject: [PATCH 2/3] fix(twilio): label the lines a relay session keeps, and let its hang-up lines play to the end - Each line kept only as a record of what the caller heard carries what it was in its MessageLabel (ConversationRelayLabels: filler, idle prompt, idle goodbye, apology, said again), so a flow can tell an idle prompt from a line of the agent's: asked "what?" after "Are you still there?", it should say its question again, not the prompt - The apology of a failed turn and the idle goodbye go out uninterruptible: the call ends after them and nothing the caller says over them is answered, so a cut-in only cost the caller the end of a line the history keeps whole. Idle prompts and a line said again stay interruptible, as they wait for the caller to speak Co-Authored-By: Claude Opus 5.5 --- .../Models/ConversationRelayLabels.cs | 35 +++++++++++ .../Services/ConversationRelaySession.cs | 46 ++++++++------ .../Services/TwilioConversationProcessor.cs | 6 +- .../TwilioConversationRelayMiddleware.cs | 6 +- .../ConversationRelayKeptLinesTests.cs | 62 +++++++++++++------ 5 files changed, 112 insertions(+), 43 deletions(-) create mode 100644 src/Plugins/BotSharp.Plugin.Twilio/Models/ConversationRelayLabels.cs 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 3db2d3b90..89dc49be5 100644 --- a/src/Plugins/BotSharp.Plugin.Twilio/Services/ConversationRelaySession.cs +++ b/src/Plugins/BotSharp.Plugin.Twilio/Services/ConversationRelaySession.cs @@ -39,7 +39,7 @@ public class ConversationRelaySession private readonly ConversationRelaySetting _relay; private readonly Func> _sendCallerMessage; private readonly ILogger _logger; - private readonly Func? _keepSaidLine; + private readonly Func? _keepSaidLine; private readonly SemaphoreSlim _sendLock = new(1, 1); private readonly object _sync = new(); @@ -142,10 +142,10 @@ public class ConversationRelaySession /// and the streamed reply /// /// - /// Keeps a line the session said on its own in the conversation history, with 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. + /// 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, @@ -154,7 +154,7 @@ public ConversationRelaySession( TwilioSetting settings, Func> sendCallerMessage, ILogger logger, - Func? keepSaidLine = null) + Func? keepSaidLine = null) { _socket = socket; _agentId = agentId; @@ -853,15 +853,17 @@ private async Task WatchIdleAsync(CancellationToken cancellationToken) bool StillSilent() => !_isProcessing && _activity == activity; if (askAgain && question != null) { - await SayOwnLineAsync(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) { - await SayOwnLineAsync(PromptText("still-there", DefaultStillThereText), StillSilent); + await SayOwnLineAsync(PromptText("still-there", DefaultStillThereText), ConversationRelayLabels.IdlePrompt, StillSilent); } else if (askAgain) { - await SayOwnLineAsync(PromptText($"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) { @@ -870,16 +872,17 @@ private async Task WatchIdleAsync(CancellationToken cancellationToken) var goodbyeUtc = DateTime.MinValue; if (relay.IdleAsksAgain) { - // The session is marked ended already, so nothing else can be said after it + // 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: false)); + 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, goodbyeUtc); + await KeepSaidAsync(goodbye, ConversationRelayLabels.IdleGoodbye, goodbyeUtc); } } } @@ -964,7 +967,7 @@ private async Task ResumeIfSilentAsync() if (said) { _logger.LogInformation($"Nothing followed the caller's cut-in in conversation {_conversationId}; said the line again."); - await KeepSaidAsync(line, DateTime.UtcNow); + await KeepSaidAsync(line, ConversationRelayLabels.SaidAgain, DateTime.UtcNow); } } @@ -977,7 +980,8 @@ private async Task FailAsync(AgentTurnCallbacks callbacks) } await CloseUtteranceAsync(); var apology = PromptText("error", DefaultErrorText); - var apologized = await SpeakAsync(apology); + // 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); @@ -986,7 +990,7 @@ private async Task FailAsync(AgentTurnCallbacks callbacks) await KeepFillersLeftAsync(callbacks); if (apologized) { - await KeepSaidAsync(apology, apologyUtc); + await KeepSaidAsync(apology, ConversationRelayLabels.Apology, apologyUtc); } } @@ -998,26 +1002,28 @@ private async Task KeepFillersLeftAsync(AgentTurnCallbacks callbacks) { foreach (var (text, saidUtc) in callbacks.TakeFillers()) { - await KeepSaidAsync(text, saidUtc); + 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, Func? canSpeak = null) + private async Task SayOwnLineAsync(string text, string label, Func? canSpeak = null) { if (await SpeakAsync(text, canSpeak)) { - await KeepSaidAsync(text, DateTime.UtcNow); + 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 /// - private async Task KeepSaidAsync(string text, DateTime saidUtc) + /// What the line is () + private async Task KeepSaidAsync(string text, string label, DateTime saidUtc) { if (_keepSaidLine == null || string.IsNullOrWhiteSpace(text)) { @@ -1026,7 +1032,7 @@ private async Task KeepSaidAsync(string text, DateTime saidUtc) try { - await _keepSaidLine(text.Trim(), saidUtc); + await _keepSaidLine(text.Trim(), label, saidUtc); } catch (Exception ex) { diff --git a/src/Plugins/BotSharp.Plugin.Twilio/Services/TwilioConversationProcessor.cs b/src/Plugins/BotSharp.Plugin.Twilio/Services/TwilioConversationProcessor.cs index 8f64d1504..3681ee0ae 100644 --- a/src/Plugins/BotSharp.Plugin.Twilio/Services/TwilioConversationProcessor.cs +++ b/src/Plugins/BotSharp.Plugin.Twilio/Services/TwilioConversationProcessor.cs @@ -157,7 +157,7 @@ await conv.SendMessage(message.AgentId, // 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, - callbacks?.TakeFillers() ?? []); + ConversationRelayLabels.Filler, callbacks?.TakeFillers() ?? []); // What the caller is told, which a rewrite below must not change var spoken = msg.Content; RewriteForHistory(msg, callbacks); @@ -253,8 +253,9 @@ await sp.GetRequiredService().UpdateConversationMessage(co /// ConversationRelay session says on its own (an apology, an idle prompt, a line said again) /// /// 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, - IReadOnlyCollection<(string Text, DateTime SaidUtc)> lines) + string label, IReadOnlyCollection<(string Text, DateTime SaidUtc)> lines) { if (lines.Count == 0) { @@ -265,6 +266,7 @@ public static async Task KeepSaidLines(IServiceProvider sp, string agentId, stri { CurrentAgentId = agentId, MessageId = messageId ?? Guid.NewGuid().ToString(), + MessageLabel = label, CreatedAt = line.SaidUtc, ExcludeFromContext = true }).ToList(); diff --git a/src/Plugins/BotSharp.Plugin.Twilio/TwilioConversationRelayMiddleware.cs b/src/Plugins/BotSharp.Plugin.Twilio/TwilioConversationRelayMiddleware.cs index 05db1deb4..baf0442a0 100644 --- a/src/Plugins/BotSharp.Plugin.Twilio/TwilioConversationRelayMiddleware.cs +++ b/src/Plugins/BotSharp.Plugin.Twilio/TwilioConversationRelayMiddleware.cs @@ -50,7 +50,7 @@ public async Task Invoke(HttpContext httpContext) var session = new ConversationRelaySession(socket, agentId, conversationId, settings, (message, callbacks) => SendCallerMessageAsync(scopeFactory, message, callbacks), _logger, - (line, saidUtc) => KeepSaidLineAsync(scopeFactory, agentId, conversationId, line, saidUtc)); + (line, label, saidUtc) => KeepSaidLineAsync(scopeFactory, agentId, conversationId, line, label, saidUtc)); await services.GetRequiredService() .RunAsync(conversationId, () => session.RunAsync(httpContext.RequestAborted)); } @@ -66,11 +66,11 @@ await services.GetRequiredService() /// 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, DateTime saidUtc) + 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, - [(line, saidUtc)]); + label, [(line, saidUtc)]); } /// diff --git a/tests/BotSharp.Core.UnitTests/Plugins/ConversationRelayKeptLinesTests.cs b/tests/BotSharp.Core.UnitTests/Plugins/ConversationRelayKeptLinesTests.cs index 431bd4b12..e1282df19 100644 --- a/tests/BotSharp.Core.UnitTests/Plugins/ConversationRelayKeptLinesTests.cs +++ b/tests/BotSharp.Core.UnitTests/Plugins/ConversationRelayKeptLinesTests.cs @@ -17,9 +17,10 @@ namespace BotSharp.Core.UnitTests.Plugins; /// -/// A ConversationRelay session keeps every line it says on its own in the conversation history, 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, and a turn's fillers are kept only once. +/// 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 { @@ -41,9 +42,12 @@ public async Task A_failed_turn_keeps_its_fillers_then_the_apology() 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] @@ -97,7 +101,7 @@ public async Task A_turn_that_stored_no_reply_leaves_its_fillers_to_the_session( await WaitUntil(() => call.Kept.Count == 1); await call.HangUpAsync(); - Assert.Equal(new[] { Filler }, call.Kept.Select(line => line.Text)); + Assert.Equal(new[] { (Filler, ConversationRelayLabels.Filler) }, call.Kept.Select(line => (line.Text, line.Label))); } [Fact] @@ -126,7 +130,7 @@ public async Task A_turn_failing_after_the_caller_hung_up_keeps_its_fillers_but_ 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.")); + keep: (_, _, _) => throw new InvalidOperationException("The storage is down.")); await WaitUntil(() => call.Socket.SentEnd); await call.HangUpAsync(); @@ -152,7 +156,11 @@ public async Task Idle_prompts_and_the_goodbye_are_kept_as_said() // 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] @@ -170,7 +178,7 @@ public async Task A_say_it_again_prompt_is_kept_with_its_text() await call.HangUpAsync(); // Without IdleAsksAgain the hang-up says nothing - Assert.Equal(new[] { "Say that again?" }, call.Kept.Select(line => line.Text)); + 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()); } @@ -191,8 +199,9 @@ public async Task A_line_said_again_after_a_cut_in_is_kept() await WaitUntil(() => call.Kept.Count == 1); await call.HangUpAsync(); - Assert.Equal(new[] { line }, call.Kept.Select(kept => kept.Text)); + 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] @@ -207,10 +216,13 @@ public async Task Kept_lines_are_the_agent_s_out_of_its_context_at_the_time_they var said = new DateTime(2026, 11, 13, 1, 31, 0, DateTimeKind.Utc); await TwilioConversationProcessor.KeepSaidLines(sp, AgentId, "conv-1", messageId: null, - [(Apology, said), ("Bye now.", said.AddSeconds(2))]); - await TwilioConversationProcessor.KeepSaidLines(sp, AgentId, "conv-1", messageId: "msg-1", [(Filler, said)]); + 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[] { Apology, "Bye now.", Filler }, stored.Select(d => d.Content)); + 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); @@ -231,7 +243,8 @@ public async Task A_storage_failure_loses_only_the_record() .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, [(Apology, DateTime.UtcNow)]); + 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); } @@ -283,19 +296,19 @@ private static async Task WaitUntil(Func condition) private sealed class Call { public FakeWebSocket Socket { get; } = new(); - public ConcurrentQueue<(string Text, DateTime SaidUtc)> Kept { 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") + 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, saidUtc) => + keep ?? ((text, label, saidUtc) => { - call.Kept.Enqueue((text, saidUtc)); + call.Kept.Enqueue((text, label, saidUtc)); return Task.CompletedTask; })); call.Run = session.RunAsync(CancellationToken.None); @@ -328,12 +341,25 @@ private sealed class FakeWebSocket : WebSocket /// 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") - .Select(frame => frame.GetProperty("token").GetString() ?? string.Empty) - .Where(token => token.Length > 0) - .ToList(); + .Where(frame => !string.IsNullOrEmpty(frame.GetProperty("token").GetString())); } public bool SentEnd => Frames().Any(frame => frame.GetProperty("type").GetString() == "end"); From c39a619e3f06b0c534f00417988311171031a9cd Mon Sep 17 00:00:00 2001 From: "mars.yu" <18671933+yuyixg@users.noreply.github.com> Date: Fri, 13 Nov 2026 12:54:40 +0800 Subject: [PATCH 3/3] fix(twilio): a reply the call ended before is stored as never said, and a caller who said nothing isn't asked to say it again Found running calls against a local onebrain: - A turn's reply that comes back after the call ended (the caller hung up while the agent worked, or the turn ran past AiResponseTimeoutSeconds and the session apologized and hung up) was stored as a line the agent said, after the apology. With nothing of it sent, it is now stored as "[the call ended before this was said; the caller did not hear: "..."]", as a reply dropped for the caller going on talking is. RewriteReply is set for every relay turn, not only streamed ones; the other marks only ever apply to a streamed reply - Silence after the greeting got "Sorry, can you say it again?" though the caller had said nothing. Until they say or key something, an idle prompt only asks if they are still there Co-Authored-By: Claude Opus 5.5 --- .../Services/ConversationRelaySession.cs | 16 ++++- .../Settings/ConversationRelaySetting.cs | 3 +- .../ConversationRelayKeptLinesTests.cs | 60 ++++++++++++++++++- 3 files changed, 72 insertions(+), 7 deletions(-) diff --git a/src/Plugins/BotSharp.Plugin.Twilio/Services/ConversationRelaySession.cs b/src/Plugins/BotSharp.Plugin.Twilio/Services/ConversationRelaySession.cs index 89dc49be5..a323d7abc 100644 --- a/src/Plugins/BotSharp.Plugin.Twilio/Services/ConversationRelaySession.cs +++ b/src/Plugins/BotSharp.Plugin.Twilio/Services/ConversationRelaySession.cs @@ -531,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; @@ -856,7 +857,9 @@ private async Task WatchIdleAsync(CancellationToken cancellationToken) 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 SayOwnLineAsync(PromptText("still-there", DefaultStillThereText), ConversationRelayLabels.IdlePrompt, StillSilent); } @@ -1129,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) { @@ -1144,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; 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/tests/BotSharp.Core.UnitTests/Plugins/ConversationRelayKeptLinesTests.cs b/tests/BotSharp.Core.UnitTests/Plugins/ConversationRelayKeptLinesTests.cs index e1282df19..75521cd5f 100644 --- a/tests/BotSharp.Core.UnitTests/Plugins/ConversationRelayKeptLinesTests.cs +++ b/tests/BotSharp.Core.UnitTests/Plugins/ConversationRelayKeptLinesTests.cs @@ -51,14 +51,16 @@ public async Task A_failed_turn_keeps_its_fillers_then_the_apology() } [Fact] - public async Task A_timed_out_turn_leaves_no_filler_to_keep_again_when_it_finally_replies() + 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 as it stores the reply + // 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." }; }); @@ -69,16 +71,45 @@ public async Task A_timed_out_turn_leaves_no_filler_to_keep_again_when_it_finall 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." }; }); @@ -87,6 +118,8 @@ public async Task Fillers_a_turn_kept_with_its_reply_are_not_kept_again() 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] @@ -172,7 +205,7 @@ public async Task A_say_it_again_prompt_is_kept_with_its_text() relay.IdleTimeouts = [0.1]; relay.MaxIdlePrompts = 1; }), - (_, _) => Task.FromResult(null), prompt: null); + (_, _) => Task.FromResult(null)); await WaitUntil(() => call.Socket.SentEnd); await call.HangUpAsync(); @@ -182,6 +215,27 @@ public async Task A_say_it_again_prompt_is_kept_with_its_text() 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() {