Feature/rule engine parallel execution

src/Infrastructure/BotSharp.Abstraction/Rules/IRuleEngine.cs

@@ -1,3 +1,5 @@
+using System.Threading;
+
 namespace BotSharp.Abstraction.Rules;
 
 public interface IRuleEngine
@@ -9,19 +11,9 @@ public interface IRuleEngine
     /// <param name="text"></param>
     /// <param name="states"></param>
     /// <param name="options"></param>
+    /// <param name="cancellationToken">Stops dispatching further rules and cancels the pause between them.</param>
     /// <returns></returns>
     /// <exception cref="NotImplementedException"></exception>
-    Task<IEnumerable<string>> Triggered(IRuleTrigger trigger, string text, IEnumerable<MessageState>? states = null, RuleTriggerOptions? options = null)
+    Task<IEnumerable<string>> Triggered(IRuleTrigger trigger, string text, IEnumerable<MessageState>? states = null, RuleTriggerOptions? options = null, CancellationToken cancellationToken = default)
         => throw new NotImplementedException();
-
-    ///// <summary>
-    ///// Execute rule graph node
-    ///// </summary>
-    ///// <param name="node"></param>
-    ///// <param name="graph"></param>
-    ///// <param name="agentId"></param>
-    ///// <param name="trigger"></param>
-    ///// <param name="options"></param>
-    ///// <returns></returns>
-    //Task ExecuteGraphNode(FlowNode node, FlowGraph graph, string agentId, IRuleTrigger trigger, RuleNodeExecutionOptions options);
 }

src/Infrastructure/BotSharp.Abstraction/Rules/Options/RuleTriggerOptions.cs

@@ -1,4 +1,4 @@
-using BotSharp.Abstraction.Repositories.Filters;
+using BotSharp.Abstraction.Repositories.Filters;
 using BotSharp.Abstraction.Rules.Constants;
 using System.Text.Json;
 
@@ -15,6 +15,15 @@ public class RuleTriggerOptions
     /// Criteria
     /// </summary>
     public CriteriaOptions? Criteria { get; set; }
+
+    /// <summary>
+    /// How long to pause after sending a message to a triggered agent, before moving on
+    /// to the next rule. Keeps a burst of triggered rules from hammering the LLM provider
+    /// all at once. Set to zero to disable the pause.
+    /// </summary>
+    public int SendMessageDelayMs { get; set; } = DefaultSendMessageDelayMs;
+
+    public const int DefaultSendMessageDelayMs = 200;
 }
 
 public class CriteriaOptions

src/Infrastructure/BotSharp.Abstraction/Rules/RuleTriggerCanceledException.cs

@@ -0,0 +1,25 @@
+using System.Threading;
+
+namespace BotSharp.Abstraction.Rules;
+
+/// <summary>
+/// Thrown when a rule trigger is cancelled part way through. Rules that already started a
+/// conversation cannot be undone, so their ids are carried on the exception and the caller
+/// can still act on them (or ignore them) while seeing the run as cancelled.
+/// </summary>
+public class RuleTriggerCanceledException : OperationCanceledException
+{
+    /// <summary>
+    /// Conversations that were created before the run was cancelled. Never null.
+    /// </summary>
+    public IReadOnlyList<string> ConversationIds { get; }
+
+    public RuleTriggerCanceledException(
+        IReadOnlyList<string> conversationIds,
+        CancellationToken cancellationToken,
+        Exception? innerException = null)
+        : base($"Rule trigger was cancelled after starting {conversationIds.Count} conversation(s).", innerException, cancellationToken)
+    {
+        ConversationIds = conversationIds;
+    }
+}

src/Infrastructure/BotSharp.Core.Rules/Controllers/RuleController.cs

@@ -1,4 +1,4 @@
-using BotSharp.Core.Rules.Models;
+using BotSharp.Core.Rules.Models;
 using Microsoft.AspNetCore.Authorization;
 using Microsoft.AspNetCore.Mvc;
 

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

@@ -1,3 +1,5 @@
+using BotSharp.Abstraction.MessageHub.Models;
+using BotSharp.Abstraction.MessageHub.Services;
 using BotSharp.Abstraction.Templating;
 
 namespace BotSharp.Core.Rules.Engines;
@@ -15,7 +17,7 @@ public RuleEngine(
         _logger = logger;
     }
 
-    public async Task<IEnumerable<string>> Triggered(IRuleTrigger trigger, string text, IEnumerable<MessageState>? states = null, RuleTriggerOptions? options = null)
+    public async Task<IEnumerable<string>> Triggered(IRuleTrigger trigger, string text, IEnumerable<MessageState>? states = null, RuleTriggerOptions? options = null, CancellationToken cancellationToken = default)
     {
         var newConversationIds = new List<string>();
 
@@ -29,70 +31,111 @@ public async Task<IEnumerable<string>> Triggered(IRuleTrigger trigger, string te
             }
         });
 
-        // Resolve the criteria evaluator
-        IRuleCriteriaEvaluator? criteriaEvaluator = null;
-        if (options?.Criteria != null)
+        // Flatten the agent/rule pairs so the two loops read as one sequence of rules.
+        var pendingRules = agents.Items
+            .Where(x => !x.Disabled)
+            .SelectMany(x => x.Rules
+                .Where(r => r != null && r.TriggerName.IsEqualTo(trigger.Name) && !r.Disabled)
+                .Select(r => (Agent: x, Rule: r)))
+            .ToList();
+
+        foreach (var item in pendingRules)
         {
-            criteriaEvaluator = ResolveCriteriaEvaluator(options.Criteria.Mode);
-            if (criteriaEvaluator == null)
+            try
+            {
+                var convId = await RunRule(item.Agent, item.Rule, trigger, text, states, options, cancellationToken);
+                if (!string.IsNullOrEmpty(convId))
+                {
+                    newConversationIds.Add(convId);
+                }
+            }
+            catch (OperationCanceledException ex)
+            {
+                // Cancellation still surfaces to the caller, but 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 (Exception ex)
             {
-                _logger.LogWarning($"Unable to find rule criteria evaluator for type ({options.Criteria.Mode}).");
+                // One misbehaving rule should not take down the rules that follow it.
+                _logger.LogError(ex, $"Error when running rule ({item.Rule.TriggerName}) for agent ({item.Agent.Name}).");
             }
         }
 
-        // Trigger agents
-        var filteredAgents = agents.Items.Where(x => x.Rules.Exists(r => r.TriggerName.IsEqualTo(trigger.Name) && !x.Disabled)).ToList();
-        foreach (var agent in filteredAgents)
+        return newConversationIds;
+    }
+
+    /// <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.
+    /// </summary>
+    private async Task<string?> RunRule(
+        Agent agent,
+        AgentRule rule,
+        IRuleTrigger trigger,
+        string text,
+        IEnumerable<MessageState>? states,
+        RuleTriggerOptions? options,
+        CancellationToken cancellationToken)
+    {
+        // Every rule runs in its own scope so the scoped conversation, state and routing
+        // services start clean per run, concurrent rules cannot bleed into each other, and
+        // the caller's own scope is left untouched.
+        using var scope = _services.CreateScope();
+        var sp = scope.ServiceProvider;
+
+        // The rule's own mode wins over the mode carried on the trigger options, so an agent can
+        // pick how its criteria is judged without the caller knowing.
+        var evaluator = ResolveCriteriaEvaluator(sp, rule.CriteriaConfig?.Mode)
+            ?? ResolveCriteriaEvaluator(sp, options?.Criteria?.Mode);
+
+        if (evaluator == null && !string.IsNullOrWhiteSpace(options?.Criteria?.Mode))
+        {
+            _logger.LogWarning($"Unable to find rule criteria evaluator for type ({options.Criteria.Mode}).");
+        }
+
+        if (evaluator != null && options?.Criteria != null)
         {
-            var rules = agent.Rules.Where(x => x.TriggerName.IsEqualTo(trigger.Name) && !x.Disabled).ToList();
-            if (rules.IsNullOrEmpty())
+            var criteriaContext = new RuleCriteriaContext
             {
-                continue;
-            }
+                Options = options.Criteria,
+                States = states
+            };
 
-            foreach (var rule in rules)
+            var isTriggered = await EvaluateCriteria(sp, evaluator, agent, rule, trigger, criteriaContext);
+            if (!isTriggered)
             {
-                if (rule == null)
-                {
-                    continue;
-                }
+                return null;
+            }
+        }
 
-                // The rule's own mode wins over the mode carried on the trigger options, so an agent can
-                // pick how its criteria is judged without the caller knowing.
-                var evaluator = ResolveCriteriaEvaluator(rule.CriteriaConfig?.Mode) ?? criteriaEvaluator;
-                if (evaluator != null && options?.Criteria != null)
-                {
-                    var criteriaContext = new RuleCriteriaContext
-                    {
-                        Options = options.Criteria,
-                        States = states
-                    };
-
-                    var isTriggered = await EvaluateCriteria(evaluator, agent, rule, trigger, criteriaContext);
-                    if (!isTriggered)
-                    {
-                        continue;
-                    }
-                }
+        // Criteria evaluation can be slow (the llm evaluator calls out), so re-check before
+        // starting a conversation that nobody is waiting on any more.
+        cancellationToken.ThrowIfCancellationRequested();
 
-                var msg = !string.IsNullOrWhiteSpace(rule.Message) ? rule.Message : text;
-                var convId = await SendMessageToAgent(agent, trigger, text, msg, states);
-                newConversationIds.Add(convId);
-            }
+        var msg = !string.IsNullOrWhiteSpace(rule.Message) ? rule.Message : text;
+        var convId = await SendMessageToAgent(sp, agent, trigger, text, msg, states);
+
+        // Pause before the next rule, so a large batch does not hammer the downstream provider.
+        var delay = options?.SendMessageDelayMs ?? RuleTriggerOptions.DefaultSendMessageDelayMs;
+        if (delay > 0)
+        {
+            await Task.Delay(delay, cancellationToken);
         }
 
-        return newConversationIds;
+        return convId;
     }
 
     #region Criteria
-    private IRuleCriteriaEvaluator? ResolveCriteriaEvaluator(string? mode)
+    private IRuleCriteriaEvaluator? ResolveCriteriaEvaluator(IServiceProvider sp, string? mode)
     {
         if (string.IsNullOrWhiteSpace(mode))
         {
             return null;
         }
 
-        return _services.GetServices<IRuleCriteriaEvaluator>().FirstOrDefault(x => x.Type.IsEqualTo(mode));
+        return sp.GetServices<IRuleCriteriaEvaluator>().FirstOrDefault(x => x.Type.IsEqualTo(mode));
     }
 
     /// <summary>
@@ -102,6 +145,7 @@ public async Task<IEnumerable<string>> Triggered(IRuleTrigger trigger, string te
     /// so a null from it means "not triggered".
     /// </summary>
     private async Task<bool> EvaluateCriteria(
+        IServiceProvider sp,
         IRuleCriteriaEvaluator evaluator,
         Agent agent,
         AgentRule agentRule,
@@ -119,7 +163,7 @@ private async Task<bool> EvaluateCriteria(
             return false;
         }
 
-        var llmEvaluator = ResolveCriteriaEvaluator(BuiltInRuleCriteria.Llm);
+        var llmEvaluator = ResolveCriteriaEvaluator(sp, BuiltInRuleCriteria.Llm);
         if (llmEvaluator == null)
         {
             _logger.LogWarning($"Unable to find llm rule criteria evaluator to fall back to from ({evaluator.Type}).");
@@ -132,9 +176,9 @@ private async Task<bool> EvaluateCriteria(
     #endregion
 
     #region Send message to agent
-    private async Task<string> SendMessageToAgent(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)
     {
-        var convService = _services.GetRequiredService<IConversationService>();
+        var convService = sp.GetRequiredService<IConversationService>();
         var conv = await convService.NewConversation(new Conversation
         {
             Channel = trigger.Channel,
@@ -152,7 +196,12 @@ private async Task<string> SendMessageToAgent(Agent agent, IRuleTrigger trigger,
             allStates.AddRange(states!);
         }
 
-        var message = new RoleDialogModel(AgentRole.User, RenderMessage(msg, allStates));
+        var message = new RoleDialogModel(AgentRole.User, RenderMessage(sp, msg, allStates));
+
+        // Subscribe the message hub observers so the rule-triggered conversation emits the same
+        // events (streaming, indications, etc.) as a user-initiated one.
+        var observer = sp.GetRequiredService<IObserverService>();
+        using var container = observer.SubscribeObservers<HubObserveData<RoleDialogModel>>(conv.Id);
 
         await convService.SetConversationId(conv.Id, allStates);
         await convService.SendMessage(agent.Id,
@@ -161,11 +210,10 @@ await convService.SendMessage(agent.Id,
             msg => Task.CompletedTask);
 
         await convService.SaveStates();
-
         return conv.Id;
     }
 
-    private string RenderMessage(string msg, IEnumerable<MessageState> states)
+    private string RenderMessage(IServiceProvider sp, string msg, IEnumerable<MessageState> states)
     {
         if (string.IsNullOrWhiteSpace(msg))
         {
@@ -185,7 +233,7 @@ private string RenderMessage(string msg, IEnumerable<MessageState> states)
                 data[state.Key] = state.Value;
             }
 
-            var render = _services.GetRequiredService<ITemplateRender>();
+            var render = sp.GetRequiredService<ITemplateRender>();
             return render.Render(msg, data);
         }
         catch (Exception ex)

src/Infrastructure/BotSharp.Core.Rules/RulesPlugin.cs

@@ -1,4 +1,4 @@
-using BotSharp.Core.Rules.Actions;
+using BotSharp.Core.Rules.Actions;
 using BotSharp.Core.Rules.Conditions;
 using BotSharp.Core.Rules.Criteria.Code;
 using BotSharp.Core.Rules.Criteria.Llm;

src/Infrastructure/BotSharp.Core.Rules/Using.cs

@@ -1,4 +1,4 @@
-global using Microsoft.Extensions.Configuration;
+global using Microsoft.Extensions.Configuration;
 global using Microsoft.Extensions.DependencyInjection;
 global using Microsoft.Extensions.Logging;
 global using System.Text;

src/Infrastructure/BotSharp.Core.Rules/data/agents/201e49a2-40b3-4ccd-b8cc-2476565a1b40/agent.json

@@ -13,7 +13,7 @@
   "llmConfig": {
     "is_inherit": false,
     "provider": "openai",
-    "model": "gpt-5.4-mini",
+    "model": "gpt-5.6-luna",
     "max_recursion_depth": 3
   }
 }