diff --git a/src/Plugins/BotSharp.Plugin.ChatHub/ChatHubPlugin.cs b/src/Plugins/BotSharp.Plugin.ChatHub/ChatHubPlugin.cs index 4ba2e30c0..9708612e9 100644 --- a/src/Plugins/BotSharp.Plugin.ChatHub/ChatHubPlugin.cs +++ b/src/Plugins/BotSharp.Plugin.ChatHub/ChatHubPlugin.cs @@ -1,4 +1,4 @@ -using BotSharp.Abstraction.MessageHub.Models; +using BotSharp.Abstraction.MessageHub.Models; using BotSharp.Abstraction.MessageHub.Observers; using BotSharp.Plugin.ChatHub.Hooks; using BotSharp.Plugin.ChatHub.Observers; @@ -22,6 +22,11 @@ public void RegisterDI(IServiceCollection services, IConfiguration config) config.Bind("ChatHub", settings); services.AddSingleton(x => settings); + // One per process: it holds a delivery chain per target, so it has to outlive any scope. + // Everything that reaches SignalR goes through it, which is what keeps a stalled client + // from stopping the thread that produced the event. See ChatEventDispatcher. + services.AddSingleton(); + services.AddScoped>, ChatHubObserver>(); // Register hooks diff --git a/src/Plugins/BotSharp.Plugin.ChatHub/Helpers/ChatEventDispatcher.cs b/src/Plugins/BotSharp.Plugin.ChatHub/Helpers/ChatEventDispatcher.cs new file mode 100644 index 000000000..c9be1c6a3 --- /dev/null +++ b/src/Plugins/BotSharp.Plugin.ChatHub/Helpers/ChatEventDispatcher.cs @@ -0,0 +1,161 @@ +using System.Collections.Concurrent; +using System.Threading; + +namespace BotSharp.Plugin.ChatHub.Helpers; + +/// +/// Hands chat events to SignalR without ever making the caller wait for the network. +/// +/// +/// +/// WHY THIS EXISTS. Sending an event used to be a plain await on +/// IClientProxy.SendAsync, and reached it through +/// GetAwaiter().GetResult() — a synchronous, unbounded, uncancellable wait on a socket. +/// That observer runs on whatever thread pushed to the MessageHub, and for a tool's +/// notifications/progress that thread is the MCP client's own read loop (McpToolExecutor +/// supplies the progress reporter, and the MCP SDK invokes it inline on the loop). So one client +/// that had stopped reading — a dead browser tab, a socket a proxy had dropped without a FIN — +/// stopped the read loop, and the tool call whose progress was being relayed never saw its own +/// response: it neither returned nor threw, not even at the HttpClient timeout, because that +/// continuation needs the loop that is stalled. A ten-minute browser task finished perfectly and +/// the agent's turn was already dead (OneFlow, 2026-09-15). +/// +/// +/// Since Subject.Synchronize serialises every push, one such send also held up chat events +/// for every other conversation in the process. +/// +/// +/// WHAT THIS DOES. is O(1) and returns at once; delivery happens on the +/// thread pool. Three properties matter and each answers a way the naive fix goes wrong: +/// +/// +/// ORDER IS KEPT, per target. Indications are a narration — "Now on plan step 2.1", then +/// "Clicking SEND" — and the one that arrives last is the one left on screen, so delivering them +/// concurrently would show them out of order. Each target has its own chain and nothing overtakes +/// within it. Two targets never wait for each other, which is the property the old code lacked. +/// A WEDGED SEND CANNOT OWN ITS CHAIN FOREVER. Each delivery carries +/// , so a socket that will never drain costs that conversation one +/// timeout rather than everything queued behind it for the life of the process. Nothing blocks +/// while that timer runs — it only bounds a task. +/// THE QUEUE IS BOUNDED. A client that stops reading entirely would otherwise accumulate one +/// entry per event for as long as the conversation lives. Past the event +/// is dropped and said so in the log; by then that browser is not showing anything anyway, and the +/// conversation itself is persisted elsewhere — SignalR is the live view, not the record. +/// +/// +/// Nothing here may throw into a caller: a send that fails is logged and the chain moves on. The +/// lanes clean themselves up — the last delivery for a target removes it — so an idle process +/// holds no entry per conversation it has ever seen. +/// +/// +public sealed class ChatEventDispatcher +{ + /// How long one delivery may take before it is abandoned. See the remarks. + public static readonly TimeSpan DefaultSendTimeout = TimeSpan.FromSeconds(30); + + /// How many events may be waiting for one target before further ones are dropped. + public const int DefaultMaxQueued = 256; + + private readonly ConcurrentDictionary _lanes = new(); + private readonly TimeSpan _sendTimeout; + private readonly int _maxQueued; + + public ChatEventDispatcher() : this(DefaultSendTimeout, DefaultMaxQueued) + { + } + + /// Overridable for tests, which cannot wait out the real timeout. + public ChatEventDispatcher(TimeSpan sendTimeout, int maxQueued) + { + _sendTimeout = sendTimeout; + _maxQueued = maxQueued; + } + + /// + /// Queue one delivery for and return. Never blocks, never throws. + /// runs on the thread pool, after everything already queued for the + /// same target and never concurrently with it. + /// + /// + /// What the event is addressed to — the group or the user, prefixed with which — so the two + /// dispatch modes cannot share a chain. + /// + /// Named in the log if the delivery fails or is dropped. + public void Enqueue(string target, Func send, ILogger logger, string description) + { + while (true) + { + var lane = _lanes.GetOrAdd(target, _ => new Lane()); + + lock (lane.Gate) + { + // The lane retired between GetOrAdd and this lock — its last delivery finished. + // Drop it (a no-op if the retiring side already did) and take a fresh one. + if (lane.Retired) + { + _lanes.TryRemove(new KeyValuePair(target, lane)); + continue; + } + + if (lane.Pending >= _maxQueued) + { + logger.LogWarning( + $"Dropped chat event '{description}' for {target}: {lane.Pending} already waiting, so the client is not reading."); + return; + } + + lane.Pending++; + lane.Tail = Deliver(lane.Tail, target, lane, send, logger, description); + } + + return; + } + } + + /// + /// The number of targets currently holding a queue. For tests and diagnostics: a healthy + /// process settles back to zero. + /// + public int ActiveTargets => _lanes.Count; + + private Task Deliver(Task tail, string target, Lane lane, Func send, ILogger logger, string description) + { + // TaskScheduler.Default, and no ExecuteSynchronously: a tail that is already complete must + // still be continued on the thread pool. Inlining it here would run the send on the caller + // — which is the thread this class exists to keep free. + return tail.ContinueWith(async _ => + { + try + { + using var cts = new CancellationTokenSource(_sendTimeout); + await send(cts.Token); + } + catch (Exception ex) + { + logger.LogWarning(ex, $"Failed to send chat event '{description}' to {target}"); + } + finally + { + lock (lane.Gate) + { + if (--lane.Pending == 0) + { + // Nothing left for this target. Retire under the same lock an enqueue takes, + // so a writer either gets in before this and keeps the lane alive, or sees + // Retired and starts a new one. It cannot land on a lane nothing will drain. + lane.Retired = true; + _lanes.TryRemove(new KeyValuePair(target, lane)); + } + } + } + }, CancellationToken.None, TaskContinuationOptions.DenyChildAttach, TaskScheduler.Default).Unwrap(); + } + + private sealed class Lane + { + public readonly object Gate = new(); + public Task Tail = Task.CompletedTask; + public int Pending; + public bool Retired; + } +} diff --git a/src/Plugins/BotSharp.Plugin.ChatHub/Helpers/EventEmitter.cs b/src/Plugins/BotSharp.Plugin.ChatHub/Helpers/EventEmitter.cs index 821b173e3..3686cbb7e 100644 --- a/src/Plugins/BotSharp.Plugin.ChatHub/Helpers/EventEmitter.cs +++ b/src/Plugins/BotSharp.Plugin.ChatHub/Helpers/EventEmitter.cs @@ -1,11 +1,29 @@ using Microsoft.AspNetCore.SignalR; using System.Runtime.CompilerServices; +using System.Threading; namespace BotSharp.Plugin.ChatHub.Helpers; internal class EventEmitter { - internal static async Task SendChatEvent( + /// + /// Address one chat event and hand it to . + /// + /// + /// + /// Returns as soon as the event is QUEUED, not when the client has it. The awaits at the call + /// sites are therefore free, and the one caller that cannot await — ChatHubObserver, + /// which is an IObserver and runs on whatever thread pushed to the MessageHub — no + /// longer stops that thread on a socket. See the dispatcher for what a stalled client used to + /// cost. + /// + /// + /// Everything that needs the request's services is resolved HERE, on the caller's thread, and + /// captured: the proxy is taken from the singleton hub context, so a delivery running after the + /// scope has been disposed still has what it needs. + /// + /// + internal static Task SendChatEvent( IServiceProvider services, ILogger logger, string @event, @@ -20,20 +38,40 @@ internal static async Task SendChatEvent( { var settings = services.GetRequiredService(); var chatHub = services.GetRequiredService>(); + var dispatcher = services.GetRequiredService(); + + IClientProxy? clients = null; + string? target = null; switch (settings.EventDispatchBy) { case EventDispatchType.Group when !string.IsNullOrEmpty(conversationId): - await chatHub.Clients.Group(conversationId).SendAsync(@event, data); + clients = chatHub.Clients.Group(conversationId); + // Prefixed so the two dispatch modes cannot share one delivery chain. + target = $"{EventDispatchType.Group}:{conversationId}"; break; case EventDispatchType.User when !string.IsNullOrEmpty(userId): - await chatHub.Clients.User(userId).SendAsync(@event, data); + clients = chatHub.Clients.User(userId); + target = $"{EventDispatchType.User}:{userId}"; break; } + + if (clients == null || target == null) + { + return Task.CompletedTask; + } + + dispatcher.Enqueue( + target, + ct => clients.SendAsync(@event, data, ct), + logger, + $"{@event} ({callerClass}-{callerMethod}) (conversation id: {conversationId})"); } catch (Exception ex) { logger.Log(logLevel, ex, $"Failed to send event '{@event}' in ({callerClass}-{callerMethod}) (conversation id: {conversationId})"); } + + return Task.CompletedTask; } } diff --git a/src/Plugins/BotSharp.Plugin.ChatHub/Observers/ChatHubObserver.cs b/src/Plugins/BotSharp.Plugin.ChatHub/Observers/ChatHubObserver.cs index 614340cfd..f15d27a8b 100644 --- a/src/Plugins/BotSharp.Plugin.ChatHub/Observers/ChatHubObserver.cs +++ b/src/Plugins/BotSharp.Plugin.ChatHub/Observers/ChatHubObserver.cs @@ -1,4 +1,4 @@ -using BotSharp.Abstraction.Conversations.Dtos; +using BotSharp.Abstraction.Conversations.Dtos; using BotSharp.Abstraction.Conversations.Enums; using BotSharp.Abstraction.MessageHub.Models; using BotSharp.Abstraction.MessageHub.Observers; @@ -159,8 +159,12 @@ private void SendEvent(string @event, string conversationId, T data, [CallerM { var user = _services.GetRequiredService(); var json = JsonSerializer.Serialize(data, _options.JsonSerializerOptions); - EventEmitter.SendChatEvent(_services, _logger, @event, conversationId, user?.Id, json, nameof(ChatHubObserver), callerName) - .ConfigureAwait(false).GetAwaiter().GetResult(); + + // NOT awaited, and it does not need to be: SendChatEvent queues the event and returns. + // This method is called from OnNext, on whatever thread pushed to the MessageHub — for a + // tool's progress notifications, the MCP client's own read loop — so anything that waited + // here for a socket could stop that thread, and did. See ChatEventDispatcher. + _ = EventEmitter.SendChatEvent(_services, _logger, @event, conversationId, user?.Id, json, nameof(ChatHubObserver), callerName); } #endregion } diff --git a/tests/BotSharp.Core.UnitTests/BotSharp.Core.UnitTests.csproj b/tests/BotSharp.Core.UnitTests/BotSharp.Core.UnitTests.csproj index 030ffeaa7..a56498e7e 100644 --- a/tests/BotSharp.Core.UnitTests/BotSharp.Core.UnitTests.csproj +++ b/tests/BotSharp.Core.UnitTests/BotSharp.Core.UnitTests.csproj @@ -32,6 +32,8 @@ + + diff --git a/tests/BotSharp.Core.UnitTests/ChatHub/ChatEventDispatcherTests.cs b/tests/BotSharp.Core.UnitTests/ChatHub/ChatEventDispatcherTests.cs new file mode 100644 index 000000000..b72eef5eb --- /dev/null +++ b/tests/BotSharp.Core.UnitTests/ChatHub/ChatEventDispatcherTests.cs @@ -0,0 +1,166 @@ +using System.Diagnostics; +using BotSharp.Plugin.ChatHub.Helpers; +using Microsoft.Extensions.Logging.Abstractions; +using Xunit; + +namespace BotSharp.Core.UnitTests.ChatHub; + +/// +/// Chat events must never make their producer wait for a client. +/// +/// The producer is often not a request thread: ChatHubObserver runs on whatever pushed to the +/// MessageHub, and for a tool's progress notifications that is the MCP client's own read loop. A +/// send that waited there stopped the loop, so the tool call being narrated never saw its own +/// response — it neither returned nor threw. These pin the properties that close that off. +/// +public class ChatEventDispatcherTests +{ + private static readonly TimeSpan ShortTimeout = TimeSpan.FromMilliseconds(300); + + private static ChatEventDispatcher Dispatcher(int maxQueued = 256) + => new(ShortTimeout, maxQueued); + + private static void Enqueue(ChatEventDispatcher dispatcher, string target, Func send) + => dispatcher.Enqueue(target, send, NullLogger.Instance, "test event"); + + [Fact] + public async Task EnqueueDoesNotWaitForTheSend() + { + var dispatcher = Dispatcher(); + var stuck = new TaskCompletionSource(); + var entered = new TaskCompletionSource(); + + var watch = Stopwatch.StartNew(); + Enqueue(dispatcher, "group:a", _ => + { + entered.TrySetResult(); + return stuck.Task; + }); + watch.Stop(); + + Assert.True(watch.ElapsedMilliseconds < 200, $"Enqueue took {watch.ElapsedMilliseconds}ms — it waited for the send."); + + // And the send really did start, so the call above returned while it was in flight rather + // than because nothing happened. + await entered.Task.WaitAsync(TimeSpan.FromSeconds(5)); + stuck.TrySetResult(); + } + + [Fact] + public async Task OneStalledTargetDoesNotHoldUpAnother() + { + var dispatcher = Dispatcher(); + var stuck = new TaskCompletionSource(); + var delivered = new TaskCompletionSource(); + + Enqueue(dispatcher, "group:stalled", _ => stuck.Task); + Enqueue(dispatcher, "group:healthy", _ => + { + delivered.TrySetResult(); + return Task.CompletedTask; + }); + + await delivered.Task.WaitAsync(TimeSpan.FromSeconds(5)); + stuck.TrySetResult(); + } + + [Fact] + public async Task DeliveriesToOneTargetKeepTheirOrder() + { + var dispatcher = Dispatcher(); + var order = new List(); + var all = new TaskCompletionSource(); + const int count = 50; + + for (var i = 0; i < count; i++) + { + var n = i; + Enqueue(dispatcher, "group:a", async _ => + { + // Uneven work: concurrent delivery would reorder these, a serial chain cannot. + await Task.Delay(n % 3); + order.Add(n); + if (order.Count == count) all.TrySetResult(); + }); + } + + await all.Task.WaitAsync(TimeSpan.FromSeconds(30)); + Assert.Equal(Enumerable.Range(0, count), order); + } + + [Fact] + public async Task AWedgedSendIsAbandonedSoTheNextOneRuns() + { + var dispatcher = Dispatcher(); + var cancelled = new TaskCompletionSource(); + var next = new TaskCompletionSource(); + + // Never completes on its own; only the dispatcher's own timeout can end it. + Enqueue(dispatcher, "group:a", async ct => + { + try + { + await Task.Delay(Timeout.Infinite, ct); + } + catch (OperationCanceledException) + { + cancelled.TrySetResult(); + throw; + } + }); + Enqueue(dispatcher, "group:a", _ => + { + next.TrySetResult(); + return Task.CompletedTask; + }); + + await cancelled.Task.WaitAsync(TimeSpan.FromSeconds(10)); + await next.Task.WaitAsync(TimeSpan.FromSeconds(10)); + } + + [Fact] + public async Task AClientThatNeverReadsCannotQueueWithoutBound() + { + var dispatcher = Dispatcher(maxQueued: 4); + var stuck = new TaskCompletionSource(); + var started = 0; + + for (var i = 0; i < 100; i++) + { + Enqueue(dispatcher, "group:a", _ => + { + Interlocked.Increment(ref started); + return stuck.Task; + }); + } + + stuck.TrySetResult(); + await WaitUntil(() => dispatcher.ActiveTargets == 0, TimeSpan.FromSeconds(10)); + + // The first is in flight and at most `maxQueued` are ever waiting, so nothing near 100 + // was accepted. The rest were dropped with a warning rather than held. + Assert.InRange(Volatile.Read(ref started), 1, 8); + } + + [Fact] + public async Task AnIdleDispatcherHoldsNothing() + { + var dispatcher = Dispatcher(); + for (var i = 0; i < 20; i++) + { + Enqueue(dispatcher, $"group:{i}", _ => Task.CompletedTask); + } + + await WaitUntil(() => dispatcher.ActiveTargets == 0, TimeSpan.FromSeconds(10)); + Assert.Equal(0, dispatcher.ActiveTargets); + } + + private static async Task WaitUntil(Func condition, TimeSpan limit) + { + var watch = Stopwatch.StartNew(); + while (!condition() && watch.Elapsed < limit) + { + await Task.Delay(10); + } + } +}