diff --git a/src/Infrastructure/BotSharp.Abstraction/Rules/Options/RuleTriggerOptions.cs b/src/Infrastructure/BotSharp.Abstraction/Rules/Options/RuleTriggerOptions.cs index 820b260df..f0e7765e8 100644 --- a/src/Infrastructure/BotSharp.Abstraction/Rules/Options/RuleTriggerOptions.cs +++ b/src/Infrastructure/BotSharp.Abstraction/Rules/Options/RuleTriggerOptions.cs @@ -1,6 +1,7 @@ using BotSharp.Abstraction.Repositories.Filters; using BotSharp.Abstraction.Rules.Constants; using System.Text.Json; +using System.Text.Json.Serialization; namespace BotSharp.Abstraction.Rules.Options; @@ -24,6 +25,40 @@ public class RuleTriggerOptions public int SendMessageDelayMs { get; set; } = DefaultSendMessageDelayMs; public const int DefaultSendMessageDelayMs = 200; + + /// + /// How long one rule may take before it is given up on. Null - the default - means no per-rule limit, + /// so only the caller's own cancellation token stops anything, exactly as before this existed. + /// + /// + /// Scoped to a single rule on purpose: the rule that runs out of time is the only one abandoned, and + /// the rules behind it still get their turn. That is what separates it from the caller's token, which + /// stops the whole run. + /// Cancellation is cooperative, so this bounds only the parts of a rule that observe a token. A rule + /// wedged inside its criteria evaluation or its agent turn is not interrupted by it - neither call + /// takes a token - and the loop waits for that rule regardless. A value of zero or less is read as no + /// limit, so Timeout.InfiniteTimeSpan says the same thing as null. + /// + public TimeSpan? RuleTimeout { get; set; } + + /// + /// Called with a conversation id the moment that conversation is created, before its message is sent. + /// + /// + /// What Triggered returns is only the rules that ran to completion, so a rule that throws after + /// its conversation was started - or a run that is cancelled - leaves behind a conversation the caller + /// never hears about. This is how a caller that has to account for every conversation, rather than only + /// the successful ones, is told about them. + /// Awaited before the conversation's message is sent, so a caller that records the id has finished + /// recording it by the time anything that can fail runs. Rules run one after another, so it is never + /// invoked concurrently. A callback that throws is swallowed: reporting the id is not worth costing the + /// rule its run. + /// JsonIgnore because this type is also the body of the rule trigger API request: a delegate has no + /// wire representation, and without this the serializer refuses the whole request model, not just this + /// property. An in-process caller is the only one that can set it, which is the only one it is for. + /// + [JsonIgnore] + public Func? OnConversationCreated { get; set; } } public class CriteriaOptions diff --git a/src/Infrastructure/BotSharp.Core.Rules/Engines/RuleEngine.cs b/src/Infrastructure/BotSharp.Core.Rules/Engines/RuleEngine.cs index 6d2083491..9db33d0f8 100644 --- a/src/Infrastructure/BotSharp.Core.Rules/Engines/RuleEngine.cs +++ b/src/Infrastructure/BotSharp.Core.Rules/Engines/RuleEngine.cs @@ -41,21 +41,35 @@ public async Task> Triggered(IRuleTrigger trigger, string te foreach (var item in pendingRules) { + // A cancellation source of the rule's own, so a rule that runs too long is the only thing given + // up on and the loop still gets to the rules behind it. Linked to the caller's token so a + // cancelled run cuts the rule in flight short the way it did before there was a per-rule limit; + // which of the two fired is what the catch clauses below tell apart. Null when no limit is + // configured, which is the default - then there is nothing to cancel but the caller's token. + using var ruleCts = CreateRuleCancellation(options, cancellationToken); + try { - var convId = await RunRule(item.Agent, item.Rule, trigger, text, states, options, cancellationToken); + var convId = await RunRule(item.Agent, item.Rule, trigger, text, states, options, ruleCts?.Token ?? cancellationToken); if (!string.IsNullOrEmpty(convId)) { newConversationIds.Add(convId); } } - catch (OperationCanceledException ex) + catch (OperationCanceledException ex) when (cancellationToken.IsCancellationRequested) { - // Cancellation still surfaces to the caller, but the conversations that were already - // started ride along on the exception so they are not silently lost. + // The caller cancelled the run, so nothing further is dispatched. The conversations that + // were already started ride along on the exception so they are not silently lost. _logger.LogWarning($"Rule trigger ({trigger.Name}) was cancelled after starting {newConversationIds.Count} conversation(s)."); throw new RuleTriggerCanceledException(newConversationIds.ToList(), cancellationToken, ex); } + catch (OperationCanceledException ex) + { + // The run itself was not cancelled, so this is the rule's own limit running out - or + // something inside it timing out on its own account. Either way it is one rule's problem, + // and the rules that follow still get their turn, each under a source of its own. + _logger.LogError(ex, $"Rule ({item.Rule.TriggerName}) for agent ({item.Agent.Name}) did not finish before it was cancelled, moving on to the next rule."); + } catch (Exception ex) { // One misbehaving rule should not take down the rules that follow it. @@ -66,6 +80,30 @@ public async Task> Triggered(IRuleTrigger trigger, string te return newConversationIds; } + /// + /// The cancellation source a single rule runs under. + /// + /// + /// Null when no per-rule limit is configured, which is the default - the caller's own token is then all + /// there is to run under, and there is no source to dispose. A limit of zero or less is read the same + /// way, so and null say the same thing. + /// + private static CancellationTokenSource? CreateRuleCancellation(RuleTriggerOptions? options, CancellationToken cancellationToken) + { + var timeout = options?.RuleTimeout; + if (timeout == null || timeout <= TimeSpan.Zero) + { + return null; + } + + // Linked rather than standalone: a cancelled run should still reach the rule in flight, and the + // caller's token is checked first when the exception comes back, so a latched one is read as + // "stop the run" rather than being mistaken for this rule's limit. + var cts = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken); + cts.CancelAfter(timeout.Value); + return cts; + } + /// /// Evaluates one rule and, when it is triggered, sends its message to the agent. /// Returns the new conversation id, or null when the rule did not trigger. @@ -115,7 +153,7 @@ public async Task> Triggered(IRuleTrigger trigger, string te cancellationToken.ThrowIfCancellationRequested(); var msg = !string.IsNullOrWhiteSpace(rule.Message) ? rule.Message : text; - var convId = await SendMessageToAgent(sp, agent, trigger, text, msg, states); + var convId = await SendMessageToAgent(sp, agent, trigger, text, msg, states, options); // Pause before the next rule, so a large batch does not hammer the downstream provider. var delay = options?.SendMessageDelayMs ?? RuleTriggerOptions.DefaultSendMessageDelayMs; @@ -176,7 +214,7 @@ private async Task EvaluateCriteria( #endregion #region Send message to agent - private async Task SendMessageToAgent(IServiceProvider sp, Agent agent, IRuleTrigger trigger, string title, string msg, IEnumerable? states = null) + private async Task SendMessageToAgent(IServiceProvider sp, Agent agent, IRuleTrigger trigger, string title, string msg, IEnumerable? states = null, RuleTriggerOptions? options = null) { var convService = sp.GetRequiredService(); var conv = await convService.NewConversation(new Conversation @@ -186,6 +224,12 @@ private async Task SendMessageToAgent(IServiceProvider sp, Agent agent, AgentId = agent.Id }); + // Reported here rather than on the way out: everything below can throw, and the conversation + // already exists by this point, so a caller that only saw the returned ids would be left with a + // conversation nothing points at. Awaited so the caller has finished recording it before the parts + // that can fail run. + await NotifyConversationCreated(options, conv.Id); + var allStates = new List { new("channel", trigger.Channel) @@ -213,6 +257,30 @@ await convService.SendMessage(agent.Id, return conv.Id; } + /// + /// Tells the caller a conversation was created, if it asked to be told. + /// + /// + /// Failures are swallowed on purpose: the callback is a caller's bookkeeping, and letting it cost the + /// rule the run it is in the middle of would be the worse outcome of the two. + /// + private async Task NotifyConversationCreated(RuleTriggerOptions? options, string conversationId) + { + if (options?.OnConversationCreated == null || string.IsNullOrEmpty(conversationId)) + { + return; + } + + try + { + await options.OnConversationCreated(conversationId); + } + catch (Exception ex) + { + _logger.LogWarning(ex, $"Error when reporting the created conversation ({conversationId}) back to the rule trigger caller."); + } + } + private string RenderMessage(IServiceProvider sp, string msg, IEnumerable states) { if (string.IsNullOrWhiteSpace(msg)) diff --git a/tests/BotSharp.Core.UnitTests/Rules/RuleEngineCancellationTests.cs b/tests/BotSharp.Core.UnitTests/Rules/RuleEngineCancellationTests.cs new file mode 100644 index 000000000..917dd60f7 --- /dev/null +++ b/tests/BotSharp.Core.UnitTests/Rules/RuleEngineCancellationTests.cs @@ -0,0 +1,190 @@ +using BotSharp.Abstraction.Agents; +using BotSharp.Abstraction.Models; +using BotSharp.Abstraction.Utilities; +using BotSharp.Abstraction.Agents.Models; +using BotSharp.Abstraction.Conversations; +using BotSharp.Abstraction.Conversations.Models; +using BotSharp.Abstraction.MessageHub.Models; +using BotSharp.Abstraction.MessageHub.Services; +using BotSharp.Abstraction.Repositories.Filters; +using BotSharp.Abstraction.Rules; +using BotSharp.Abstraction.Rules.Options; +using BotSharp.Core.Rules.Engines; +using Microsoft.Extensions.DependencyInjection; +using Microsoft.Extensions.Logging; +using Moq; +using Xunit; + +namespace BotSharp.Core.UnitTests.Rules; + +/// +/// Pins what a cancelled rule costs the rules behind it. The engine runs every rule in one loop, so the +/// two cancellations it can see - the caller giving up on the whole run, and one rule outstaying its own +/// limit - have to be told apart there, or they collapse into each other: a per-rule limit that ends the +/// run, or a cancelled run that carries on dispatching. +/// +public class RuleEngineCancellationTests +{ + private const string TriggerName = "stub_trigger"; + + private static List _log = []; + + private sealed class StubTrigger : IRuleTrigger + { + public string EntityType { get; set; } = "test"; + public string EntityId { get; set; } = "test"; + public string Name => TriggerName; + public string Channel => "test"; + } + + /// + /// Two agents, each subscribed to the trigger, dispatched in the order given. + /// + /// + /// How long that agent's turn takes. The real SendMessage takes no cancellation token, so a slow turn + /// here is uninterruptible in the test for the same reason it is in production - the limit can only be + /// noticed once the turn is over. + /// + /// + /// Captures what the engine logged. The engine swallows a misbehaving rule's exception by design, so + /// without this a broken test setup is indistinguishable from a rule that legitimately did nothing. + /// + private sealed class CapturingLogger : ILogger + { + public List Messages { get; } = []; + public IDisposable? BeginScope(TState state) where TState : notnull => null; + public bool IsEnabled(LogLevel logLevel) => true; + public void Log(LogLevel logLevel, EventId eventId, TState state, Exception? exception, Func formatter) + => Messages.Add($"{logLevel}: {formatter(state, exception)}{(exception == null ? "" : " | " + exception)}"); + } + + private static (RuleEngine engine, List sentToAgents) Build( + Dictionary? sendDuration = null, + Action? onSendStarted = null) + { + var durations = sendDuration ?? []; + var sentToAgents = new List(); + + var agents = new PagedItems + { + Count = 2, + Items = + [ + new Agent { Id = "agent-1", Name = "agent-1", Disabled = false, Rules = [new AgentRule { TriggerName = TriggerName }] }, + new Agent { Id = "agent-2", Name = "agent-2", Disabled = false, Rules = [new AgentRule { TriggerName = TriggerName }] } + ] + }; + + var agentService = new Mock(); + agentService.Setup(x => x.GetAgents(It.IsAny())).ReturnsAsync(agents); + + var convService = new Mock(); + convService.Setup(x => x.NewConversation(It.IsAny())) + .ReturnsAsync((Conversation c) => new Conversation { Id = $"conv-{c.AgentId}", AgentId = c.AgentId }); + convService.Setup(x => x.SetConversationId(It.IsAny(), It.IsAny>(), It.IsAny())) + .Returns(Task.CompletedTask); + convService.Setup(x => x.SaveStates()).Returns(Task.CompletedTask); + convService.Setup(x => x.SendMessage( + It.IsAny(), + It.IsAny(), + It.IsAny(), + It.IsAny>())) + .Returns(async (string agentId, RoleDialogModel _, PostbackMessageModel? _, Func _) => + { + sentToAgents.Add(agentId); + onSendStarted?.Invoke(agentId); + + if (durations.TryGetValue(agentId, out var duration)) + { + await Task.Delay(duration); + } + + return true; + }); + + var observer = new Mock(); + observer.Setup(x => x.SubscribeObservers>( + It.IsAny(), It.IsAny?>(), It.IsAny, Task>>?>())) + .Returns(Mock.Of()); + + var services = new ServiceCollection(); + services.AddSingleton(agentService.Object); + services.AddSingleton(convService.Object); + services.AddSingleton(observer.Object); + + var logger = new CapturingLogger(); + _log = logger.Messages; + return (new RuleEngine(services.BuildServiceProvider(), logger), sentToAgents); + } + + private static Task> Run(RuleEngine engine, RuleTriggerOptions options, CancellationToken token = default) + => engine.Triggered(new StubTrigger(), "text", states: null, options, token); + + /// + /// The default. Nothing about the loop changes when no per-rule limit is set: a run nobody cancels + /// dispatches every rule. + /// + [Fact] + public async Task Runs_every_rule_when_no_per_rule_limit_is_set() + { + var (engine, sentToAgents) = Build(); + + var result = await Run(engine, new RuleTriggerOptions { SendMessageDelayMs = 0 }); + + Assert.Equal(["agent-1", "agent-2"], sentToAgents); + Assert.Equal(["conv-agent-1", "conv-agent-2"], result); + } + + /// + /// The point of the per-rule limit: the rule that outstays it is the only one given up on, and the + /// rule behind it still gets its turn. Before the limit existed there was no way to express this - + /// the only cancellation the loop understood ended the run. + /// + [Fact] + public async Task A_rule_that_outstays_its_limit_does_not_cost_the_next_rule_its_turn() + { + var (engine, sentToAgents) = Build( + sendDuration: new Dictionary { ["agent-1"] = TimeSpan.FromMilliseconds(600) }); + + var result = await Run(engine, new RuleTriggerOptions + { + SendMessageDelayMs = 1, + RuleTimeout = TimeSpan.FromMilliseconds(100) + }); + + // Both were dispatched: the first outstayed its limit, the second still ran. + Assert.Equal(["agent-1", "agent-2"], sentToAgents); + + // Only the second is reported. The first conversation exists - its turn finished - but the rule + // was cancelled before it could hand the id back, which is what OnConversationCreated is for. + Assert.Equal(["conv-agent-2"], result); + } + + /// + /// The other half of the split: a cancelled run still stops dispatching, and still hands back what it + /// started, even though a per-rule limit is now in play. + /// + [Fact] + public async Task A_cancelled_run_still_stops_the_rules_behind_it() + { + using var cts = new CancellationTokenSource(); + var (engine, sentToAgents) = Build(onSendStarted: agentId => + { + if (agentId == "agent-1") + { + cts.Cancel(); + } + }); + + var ex = await Assert.ThrowsAsync(() => Run(engine, new RuleTriggerOptions + { + SendMessageDelayMs = 1, + RuleTimeout = TimeSpan.FromMinutes(5) + }, cts.Token)); + + // The second rule was never dispatched, and the run reports itself cancelled rather than quietly + // treating the caller's token as one rule's limit. + Assert.Equal(["agent-1"], sentToAgents); + Assert.NotNull(ex.ConversationIds); + } +}