refine rule engine

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;
+
+    /// <summary>
+    /// 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.
+    /// </summary>
+    /// <remarks>
+    /// 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 <c>Timeout.InfiniteTimeSpan</c> says the same thing as null.
+    /// </remarks>
+    public TimeSpan? RuleTimeout { get; set; }
+
+    /// <summary>
+    /// Called with a conversation id the moment that conversation is created, before its message is sent.
+    /// </summary>
+    /// <remarks>
+    /// What <c>Triggered</c> 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.
+    /// </remarks>
+    [JsonIgnore]
+    public Func<string, Task>? OnConversationCreated { get; set; }
 }
 
 public class CriteriaOptions

src/Infrastructure/BotSharp.Core.Rules/Engines/RuleEngine.cs

@@ -41,21 +41,35 @@ public async Task<IEnumerable<string>> 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<IEnumerable<string>> Triggered(IRuleTrigger trigger, string te
         return newConversationIds;
     }
 
+    /// <summary>
+    /// The cancellation source a single rule runs under.
+    /// </summary>
+    /// <returns>
+    /// 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 <see cref="Timeout.InfiniteTimeSpan"/> and null say the same thing.
+    /// </returns>
+    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;
+    }
+
     /// <summary>
     /// 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<IEnumerable<string>> 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<bool> EvaluateCriteria(
     #endregion
 
     #region Send message to agent
-    private async Task<string> SendMessageToAgent(IServiceProvider sp, Agent agent, IRuleTrigger trigger, string title, string msg, IEnumerable<MessageState>? states = null)
+    private async Task<string> SendMessageToAgent(IServiceProvider sp, Agent agent, IRuleTrigger trigger, string title, string msg, IEnumerable<MessageState>? states = null, RuleTriggerOptions? options = null)
     {
         var convService = sp.GetRequiredService<IConversationService>();
         var conv = await convService.NewConversation(new Conversation
@@ -186,6 +224,12 @@ private async Task<string> 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<MessageState>
         {
             new("channel", trigger.Channel)
@@ -213,6 +257,30 @@ await convService.SendMessage(agent.Id,
         return conv.Id;
     }
 
+    /// <summary>
+    /// Tells the caller a conversation was created, if it asked to be told.
+    /// </summary>
+    /// <remarks>
+    /// 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.
+    /// </remarks>
+    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<MessageState> states)
     {
         if (string.IsNullOrWhiteSpace(msg))

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;
+
+/// <summary>
+/// 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.
+/// </summary>
+public class RuleEngineCancellationTests
+{
+    private const string TriggerName = "stub_trigger";
+
+    private static List<string> _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";
+    }
+
+    /// <summary>
+    /// Two agents, each subscribed to the trigger, dispatched in the order given.
+    /// </summary>
+    /// <param name="sendDuration">
+    /// 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.
+    /// </param>
+    /// <summary>
+    /// 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.
+    /// </summary>
+    private sealed class CapturingLogger<T> : ILogger<T>
+    {
+        public List<string> Messages { get; } = [];
+        public IDisposable? BeginScope<TState>(TState state) where TState : notnull => null;
+        public bool IsEnabled(LogLevel logLevel) => true;
+        public void Log<TState>(LogLevel logLevel, EventId eventId, TState state, Exception? exception, Func<TState, Exception?, string> formatter)
+            => Messages.Add($"{logLevel}: {formatter(state, exception)}{(exception == null ? "" : " | " + exception)}");
+    }
+
+    private static (RuleEngine engine, List<string> sentToAgents) Build(
+        Dictionary<string, TimeSpan>? sendDuration = null,
+        Action<string>? onSendStarted = null)
+    {
+        var durations = sendDuration ?? [];
+        var sentToAgents = new List<string>();
+
+        var agents = new PagedItems<Agent>
+        {
+            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<IAgentService>();
+        agentService.Setup(x => x.GetAgents(It.IsAny<AgentFilter>())).ReturnsAsync(agents);
+
+        var convService = new Mock<IConversationService>();
+        convService.Setup(x => x.NewConversation(It.IsAny<Conversation>()))
+            .ReturnsAsync((Conversation c) => new Conversation { Id = $"conv-{c.AgentId}", AgentId = c.AgentId });
+        convService.Setup(x => x.SetConversationId(It.IsAny<string>(), It.IsAny<List<MessageState>>(), It.IsAny<bool>()))
+            .Returns(Task.CompletedTask);
+        convService.Setup(x => x.SaveStates()).Returns(Task.CompletedTask);
+        convService.Setup(x => x.SendMessage(
+                It.IsAny<string>(),
+                It.IsAny<RoleDialogModel>(),
+                It.IsAny<PostbackMessageModel?>(),
+                It.IsAny<Func<RoleDialogModel, Task>>()))
+            .Returns(async (string agentId, RoleDialogModel _, PostbackMessageModel? _, Func<RoleDialogModel, Task> _) =>
+            {
+                sentToAgents.Add(agentId);
+                onSendStarted?.Invoke(agentId);
+
+                if (durations.TryGetValue(agentId, out var duration))
+                {
+                    await Task.Delay(duration);
+                }
+
+                return true;
+            });
+
+        var observer = new Mock<IObserverService>();
+        observer.Setup(x => x.SubscribeObservers<HubObserveData<RoleDialogModel>>(
+                It.IsAny<string>(), It.IsAny<IEnumerable<string>?>(), It.IsAny<Dictionary<string, Func<HubObserveData<RoleDialogModel>, Task>>?>()))
+            .Returns(Mock.Of<IDisposable>());
+
+        var services = new ServiceCollection();
+        services.AddSingleton(agentService.Object);
+        services.AddSingleton(convService.Object);
+        services.AddSingleton(observer.Object);
+
+        var logger = new CapturingLogger<RuleEngine>();
+        _log = logger.Messages;
+        return (new RuleEngine(services.BuildServiceProvider(), logger), sentToAgents);
+    }
+
+    private static Task<IEnumerable<string>> Run(RuleEngine engine, RuleTriggerOptions options, CancellationToken token = default)
+        => engine.Triggered(new StubTrigger(), "text", states: null, options, token);
+
+    /// <summary>
+    /// The default. Nothing about the loop changes when no per-rule limit is set: a run nobody cancels
+    /// dispatches every rule.
+    /// </summary>
+    [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);
+    }
+
+    /// <summary>
+    /// 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.
+    /// </summary>
+    [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<string, TimeSpan> { ["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);
+    }
+
+    /// <summary>
+    /// 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.
+    /// </summary>
+    [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<RuleTriggerCanceledException>(() => 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);
+    }
+}