diff --git a/src/Infrastructure/BotSharp.Abstraction/Rules/IRuleEngine.cs b/src/Infrastructure/BotSharp.Abstraction/Rules/IRuleEngine.cs index d8ac4927d..3251f803d 100644 --- a/src/Infrastructure/BotSharp.Abstraction/Rules/IRuleEngine.cs +++ b/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 /// /// /// + /// Stops dispatching further rules and cancels the pause between them. /// /// - Task> Triggered(IRuleTrigger trigger, string text, IEnumerable? states = null, RuleTriggerOptions? options = null) + Task> Triggered(IRuleTrigger trigger, string text, IEnumerable? states = null, RuleTriggerOptions? options = null, CancellationToken cancellationToken = default) => throw new NotImplementedException(); - - ///// - ///// Execute rule graph node - ///// - ///// - ///// - ///// - ///// - ///// - ///// - //Task ExecuteGraphNode(FlowNode node, FlowGraph graph, string agentId, IRuleTrigger trigger, RuleNodeExecutionOptions options); } diff --git a/src/Infrastructure/BotSharp.Abstraction/Rules/Options/RuleTriggerOptions.cs b/src/Infrastructure/BotSharp.Abstraction/Rules/Options/RuleTriggerOptions.cs index 174274d06..820b260df 100644 --- a/src/Infrastructure/BotSharp.Abstraction/Rules/Options/RuleTriggerOptions.cs +++ b/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 /// public CriteriaOptions? Criteria { get; set; } + + /// + /// 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. + /// + public int SendMessageDelayMs { get; set; } = DefaultSendMessageDelayMs; + + public const int DefaultSendMessageDelayMs = 200; } public class CriteriaOptions diff --git a/src/Infrastructure/BotSharp.Abstraction/Rules/RuleTriggerCanceledException.cs b/src/Infrastructure/BotSharp.Abstraction/Rules/RuleTriggerCanceledException.cs new file mode 100644 index 000000000..d410e20fa --- /dev/null +++ b/src/Infrastructure/BotSharp.Abstraction/Rules/RuleTriggerCanceledException.cs @@ -0,0 +1,25 @@ +using System.Threading; + +namespace BotSharp.Abstraction.Rules; + +/// +/// 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. +/// +public class RuleTriggerCanceledException : OperationCanceledException +{ + /// + /// Conversations that were created before the run was cancelled. Never null. + /// + public IReadOnlyList ConversationIds { get; } + + public RuleTriggerCanceledException( + IReadOnlyList conversationIds, + CancellationToken cancellationToken, + Exception? innerException = null) + : base($"Rule trigger was cancelled after starting {conversationIds.Count} conversation(s).", innerException, cancellationToken) + { + ConversationIds = conversationIds; + } +} diff --git a/src/Infrastructure/BotSharp.Core.Rules/Controllers/RuleController.cs b/src/Infrastructure/BotSharp.Core.Rules/Controllers/RuleController.cs index feb314ef7..a7f524572 100644 --- a/src/Infrastructure/BotSharp.Core.Rules/Controllers/RuleController.cs +++ b/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; diff --git a/src/Infrastructure/BotSharp.Core.Rules/Engines/RuleEngine.cs b/src/Infrastructure/BotSharp.Core.Rules/Engines/RuleEngine.cs index f6b906bc8..6d2083491 100644 --- a/src/Infrastructure/BotSharp.Core.Rules/Engines/RuleEngine.cs +++ b/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> Triggered(IRuleTrigger trigger, string text, IEnumerable? states = null, RuleTriggerOptions? options = null) + public async Task> Triggered(IRuleTrigger trigger, string text, IEnumerable? states = null, RuleTriggerOptions? options = null, CancellationToken cancellationToken = default) { var newConversationIds = new List(); @@ -29,70 +31,111 @@ public async Task> 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; + } + + /// + /// 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. + /// + private async Task RunRule( + Agent agent, + AgentRule rule, + IRuleTrigger trigger, + string text, + IEnumerable? 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().FirstOrDefault(x => x.Type.IsEqualTo(mode)); + return sp.GetServices().FirstOrDefault(x => x.Type.IsEqualTo(mode)); } /// @@ -102,6 +145,7 @@ public async Task> Triggered(IRuleTrigger trigger, string te /// so a null from it means "not triggered". /// private async Task EvaluateCriteria( + IServiceProvider sp, IRuleCriteriaEvaluator evaluator, Agent agent, AgentRule agentRule, @@ -119,7 +163,7 @@ private async Task 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 EvaluateCriteria( #endregion #region Send message to agent - private async Task SendMessageToAgent(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) { - var convService = _services.GetRequiredService(); + var convService = sp.GetRequiredService(); var conv = await convService.NewConversation(new Conversation { Channel = trigger.Channel, @@ -152,7 +196,12 @@ private async Task 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(); + using var container = observer.SubscribeObservers>(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 states) + private string RenderMessage(IServiceProvider sp, string msg, IEnumerable states) { if (string.IsNullOrWhiteSpace(msg)) { @@ -185,7 +233,7 @@ private string RenderMessage(string msg, IEnumerable states) data[state.Key] = state.Value; } - var render = _services.GetRequiredService(); + var render = sp.GetRequiredService(); return render.Render(msg, data); } catch (Exception ex) diff --git a/src/Infrastructure/BotSharp.Core.Rules/RulesPlugin.cs b/src/Infrastructure/BotSharp.Core.Rules/RulesPlugin.cs index eb2a6552b..5ee0d4d5d 100644 --- a/src/Infrastructure/BotSharp.Core.Rules/RulesPlugin.cs +++ b/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; diff --git a/src/Infrastructure/BotSharp.Core.Rules/Using.cs b/src/Infrastructure/BotSharp.Core.Rules/Using.cs index 46092a8cc..54f48a78b 100644 --- a/src/Infrastructure/BotSharp.Core.Rules/Using.cs +++ b/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; diff --git a/src/Infrastructure/BotSharp.Core.Rules/data/agents/201e49a2-40b3-4ccd-b8cc-2476565a1b40/agent.json b/src/Infrastructure/BotSharp.Core.Rules/data/agents/201e49a2-40b3-4ccd-b8cc-2476565a1b40/agent.json index 26cd9c331..7ebb3f30a 100644 --- a/src/Infrastructure/BotSharp.Core.Rules/data/agents/201e49a2-40b3-4ccd-b8cc-2476565a1b40/agent.json +++ b/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 } } \ No newline at end of file