From 94438c6b136e63b52d760c375ff68b78e03a190f Mon Sep 17 00:00:00 2001 From: Jicheng Lu <103353@smsassist.com> Date: Tue, 1 Sep 2026 10:31:15 -0500 Subject: [PATCH 1/4] Run triggered rules in parallel with per-rule service scopes Each triggered rule now runs in its own DI scope, so the scoped conversation, state and routing services start clean per run and no longer bleed between rules or into the caller's scope. The message hub observers are subscribed per run so rule-triggered conversations emit the same events as user-initiated ones. The nested agent/rule loops are flattened into a single list and dispatched via Parallel.ForEachAsync, throttled by MaxConcurrency (options -> RuleSettings -> built-in default of 5). Results are written into an indexed array so conversation ids keep rule order without a concurrent-add race. Triggered also takes a CancellationToken: it stops dispatching new rules, interrupts the inter-rule delay, and propagates to the caller. Per-rule failures are logged and isolated so one bad rule does not take down the others. Co-Authored-By: Claude Opus 5 (1M context) --- .../BotSharp.Abstraction/Rules/IRuleEngine.cs | 16 +- .../Rules/Options/RuleTriggerOptions.cs | 21 ++- .../Rules/Settings/RuleSettings.cs | 10 ++ .../Controllers/RuleController.cs | 6 +- .../BotSharp.Core.Rules/Engines/RuleEngine.cs | 165 ++++++++++++------ .../BotSharp.Core.Rules/RulesPlugin.cs | 6 +- .../BotSharp.Core.Rules/Using.cs | 3 +- 7 files changed, 159 insertions(+), 68 deletions(-) create mode 100644 src/Infrastructure/BotSharp.Abstraction/Rules/Settings/RuleSettings.cs 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..cfa0d79a2 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,25 @@ 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; + + /// + /// How many triggered rules may run at the same time. Each one still gets its own + /// service scope, so concurrent runs do not share conversation, state or routing + /// services. Set to one to run them sequentially. Null falls back to + /// RuleSettings.MaxConcurrency, then to . + /// + public int? MaxConcurrency { get; set; } + + public const int DefaultMaxConcurrency = 5; } public class CriteriaOptions diff --git a/src/Infrastructure/BotSharp.Abstraction/Rules/Settings/RuleSettings.cs b/src/Infrastructure/BotSharp.Abstraction/Rules/Settings/RuleSettings.cs new file mode 100644 index 000000000..eea1391d0 --- /dev/null +++ b/src/Infrastructure/BotSharp.Abstraction/Rules/Settings/RuleSettings.cs @@ -0,0 +1,10 @@ +namespace BotSharp.Abstraction.Rules.Settings; + +public class RuleSettings +{ + /// + /// How many triggered rules may run at the same time. Overridden per call by + /// RuleTriggerOptions.MaxConcurrency. Null falls back to the built-in default. + /// + public int? MaxConcurrency { get; set; } +} diff --git a/src/Infrastructure/BotSharp.Core.Rules/Controllers/RuleController.cs b/src/Infrastructure/BotSharp.Core.Rules/Controllers/RuleController.cs index feb314ef7..692e9dcb3 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; @@ -23,7 +23,7 @@ public RuleController( } [HttpPost("/rule/trigger/action")] - public async Task RunAction([FromBody] RuleTriggerActionRequest request) + public async Task RunAction([FromBody] RuleTriggerActionRequest request, CancellationToken cancellationToken) { if (request == null) { @@ -36,7 +36,7 @@ public async Task RunAction([FromBody] RuleTriggerActionRequest r return BadRequest(new { Success = false, Error = "Unable to find rule trigger." }); } - var result = await _ruleEngine.Triggered(trigger, request.Text, request.States, request.Options); + var result = await _ruleEngine.Triggered(trigger, request.Text, request.States, request.Options, cancellationToken); return Ok(new { Success = true }); } } diff --git a/src/Infrastructure/BotSharp.Core.Rules/Engines/RuleEngine.cs b/src/Infrastructure/BotSharp.Core.Rules/Engines/RuleEngine.cs index f6b906bc8..af1f4e518 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,128 @@ 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 they can be throttled as one unit, rather than + // running one agent's rules concurrently but the agents themselves one at a time. + 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(); + + if (pendingRules.IsNullOrEmpty()) + { + return newConversationIds; + } + + // Indexed so the returned conversation ids keep the rule order regardless of + // which run finishes first. + var convIds = new string?[pendingRules.Count]; + + // Per-call options win over the configured setting, which in turn wins over the built-in default. + var settings = _services.GetService(); + var maxConcurrency = options?.MaxConcurrency + ?? settings?.MaxConcurrency + ?? RuleTriggerOptions.DefaultMaxConcurrency; + + var parallelOptions = new ParallelOptions + { + MaxDegreeOfParallelism = Math.Max(1, maxConcurrency), + CancellationToken = cancellationToken + }; + + var indexedRules = pendingRules.Select((item, index) => (item.Agent, item.Rule, Index: index)); + + // Cancellation is not handled here on purpose: it propagates to the caller so they can + // tell a cancelled run apart from one that simply triggered no rules. + await Parallel.ForEachAsync(indexedRules, parallelOptions, async (item, token) => { - criteriaEvaluator = ResolveCriteriaEvaluator(options.Criteria.Mode); - if (criteriaEvaluator == null) + try { - _logger.LogWarning($"Unable to find rule criteria evaluator for type ({options.Criteria.Mode})."); + convIds[item.Index] = await RunRule(item.Agent, item.Rule, trigger, text, states, options, token); } + catch (Exception ex) when (ex is not OperationCanceledException) + { + // One misbehaving rule should not take down the rules that run alongside it. + _logger.LogError(ex, $"Error when running rule ({item.Rule.TriggerName}) for agent ({item.Agent.Name})."); + } + }); + + newConversationIds.AddRange(convIds.Where(x => !string.IsNullOrEmpty(x)).Select(x => x!)); + 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})."); } - // 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) + 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); + + // Hold the concurrency slot a little longer after sending, so a large batch of rules + // does not hammer the downstream provider the moment each slot frees up. + 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 +162,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 +180,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 +193,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 +213,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 +227,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 +250,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..521c74160 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; @@ -21,6 +21,10 @@ public class RulesPlugin : IBotSharpPlugin public void RegisterDI(IServiceCollection services, IConfiguration config) { + var settings = new RuleSettings(); + config.Bind("Rule", settings); + services.AddSingleton(settings); + // Register rule engine services.AddScoped(); diff --git a/src/Infrastructure/BotSharp.Core.Rules/Using.cs b/src/Infrastructure/BotSharp.Core.Rules/Using.cs index 46092a8cc..de7af3130 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; @@ -20,6 +20,7 @@ global using BotSharp.Abstraction.Repositories.Filters; global using BotSharp.Abstraction.Rules; global using BotSharp.Abstraction.Rules.Options; +global using BotSharp.Abstraction.Rules.Settings; global using BotSharp.Abstraction.Rules.Models; global using BotSharp.Abstraction.Rules.Hooks; global using BotSharp.Abstraction.Utilities; From 41cc3ec7f074469df445dcae0113a142a780eb64 Mon Sep 17 00:00:00 2001 From: Jicheng Lu <103353@smsassist.com> Date: Tue, 1 Sep 2026 10:35:20 -0500 Subject: [PATCH 2/4] Carry started conversation ids on rule trigger cancellation A cancelled run cannot both return its conversation ids and surface the cancellation, so the ids now ride along on the exception. Triggered throws RuleTriggerCanceledException, which derives from OperationCanceledException (existing handlers keep working) and exposes the conversations that were already started, so callers can still act on work that cannot be undone. Also drops the CancellationToken from the rule trigger endpoint, so a client disconnect no longer stops rules that are mid-dispatch. The token stays optional on IRuleEngine.Triggered for other callers. Co-Authored-By: Claude Opus 5 (1M context) --- .../Rules/RuleTriggerCanceledException.cs | 25 ++++++++++++ .../Controllers/RuleController.cs | 4 +- .../BotSharp.Core.Rules/Engines/RuleEngine.cs | 38 ++++++++++++------- 3 files changed, 52 insertions(+), 15 deletions(-) create mode 100644 src/Infrastructure/BotSharp.Abstraction/Rules/RuleTriggerCanceledException.cs 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 692e9dcb3..a7f524572 100644 --- a/src/Infrastructure/BotSharp.Core.Rules/Controllers/RuleController.cs +++ b/src/Infrastructure/BotSharp.Core.Rules/Controllers/RuleController.cs @@ -23,7 +23,7 @@ public RuleController( } [HttpPost("/rule/trigger/action")] - public async Task RunAction([FromBody] RuleTriggerActionRequest request, CancellationToken cancellationToken) + public async Task RunAction([FromBody] RuleTriggerActionRequest request) { if (request == null) { @@ -36,7 +36,7 @@ public async Task RunAction([FromBody] RuleTriggerActionRequest r return BadRequest(new { Success = false, Error = "Unable to find rule trigger." }); } - var result = await _ruleEngine.Triggered(trigger, request.Text, request.States, request.Options, cancellationToken); + var result = await _ruleEngine.Triggered(trigger, request.Text, request.States, request.Options); return Ok(new { Success = true }); } } diff --git a/src/Infrastructure/BotSharp.Core.Rules/Engines/RuleEngine.cs b/src/Infrastructure/BotSharp.Core.Rules/Engines/RuleEngine.cs index af1f4e518..332eb722c 100644 --- a/src/Infrastructure/BotSharp.Core.Rules/Engines/RuleEngine.cs +++ b/src/Infrastructure/BotSharp.Core.Rules/Engines/RuleEngine.cs @@ -63,25 +63,37 @@ public async Task> Triggered(IRuleTrigger trigger, string te var indexedRules = pendingRules.Select((item, index) => (item.Agent, item.Rule, Index: index)); - // Cancellation is not handled here on purpose: it propagates to the caller so they can - // tell a cancelled run apart from one that simply triggered no rules. - await Parallel.ForEachAsync(indexedRules, parallelOptions, async (item, token) => + try { - try - { - convIds[item.Index] = await RunRule(item.Agent, item.Rule, trigger, text, states, options, token); - } - catch (Exception ex) when (ex is not OperationCanceledException) + await Parallel.ForEachAsync(indexedRules, parallelOptions, async (item, token) => { - // One misbehaving rule should not take down the rules that run alongside it. - _logger.LogError(ex, $"Error when running rule ({item.Rule.TriggerName}) for agent ({item.Agent.Name})."); - } - }); + try + { + convIds[item.Index] = await RunRule(item.Agent, item.Rule, trigger, text, states, options, token); + } + catch (Exception ex) when (ex is not OperationCanceledException) + { + // One misbehaving rule should not take down the rules that run alongside it. + _logger.LogError(ex, $"Error when running rule ({item.Rule.TriggerName}) for agent ({item.Agent.Name})."); + } + }); + } + 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. + var startedIds = CollectConversationIds(convIds); + _logger.LogWarning($"Rule trigger ({trigger.Name}) was cancelled after starting {startedIds.Count} conversation(s)."); + throw new RuleTriggerCanceledException(startedIds, cancellationToken, ex); + } - newConversationIds.AddRange(convIds.Where(x => !string.IsNullOrEmpty(x)).Select(x => x!)); + newConversationIds.AddRange(CollectConversationIds(convIds)); return newConversationIds; } + private static List CollectConversationIds(string?[] convIds) + => convIds.Where(x => !string.IsNullOrEmpty(x)).Select(x => x!).ToList(); + /// /// 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. From 7074a2e82a33c5ca88c0fc99d4180a0d3f655dab Mon Sep 17 00:00:00 2001 From: Jicheng Lu <103353@smsassist.com> Date: Tue, 1 Sep 2026 10:48:25 -0500 Subject: [PATCH 3/4] change agent llm --- .../data/agents/201e49a2-40b3-4ccd-b8cc-2476565a1b40/agent.json | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) 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 From 6cc200b3beac44f80b715b646b6f8f852f537b4e Mon Sep 17 00:00:00 2001 From: Jicheng Lu <103353@smsassist.com> Date: Tue, 1 Sep 2026 11:55:25 -0500 Subject: [PATCH 4/4] Run triggered rules sequentially again Drops Parallel.ForEachAsync in favour of a plain loop over the rules. Conversation ids are appended as each rule finishes, so the indexed array and its collect helper are no longer needed to keep them in order. MaxConcurrency only existed to throttle the parallel run, so it goes too, along with the RuleSettings class and its "Rule" config binding that were added to configure it. The inter-rule delay falls back to the per-call option and then the built-in default. Per-rule service scopes, cancellation and per-rule error isolation are unchanged. Co-Authored-By: Claude Opus 5 (1M context) --- .../Rules/Options/RuleTriggerOptions.cs | 10 --- .../Rules/Settings/RuleSettings.cs | 10 --- .../BotSharp.Core.Rules/Engines/RuleEngine.cs | 69 ++++++------------- .../BotSharp.Core.Rules/RulesPlugin.cs | 4 -- .../BotSharp.Core.Rules/Using.cs | 1 - 5 files changed, 20 insertions(+), 74 deletions(-) delete mode 100644 src/Infrastructure/BotSharp.Abstraction/Rules/Settings/RuleSettings.cs diff --git a/src/Infrastructure/BotSharp.Abstraction/Rules/Options/RuleTriggerOptions.cs b/src/Infrastructure/BotSharp.Abstraction/Rules/Options/RuleTriggerOptions.cs index cfa0d79a2..820b260df 100644 --- a/src/Infrastructure/BotSharp.Abstraction/Rules/Options/RuleTriggerOptions.cs +++ b/src/Infrastructure/BotSharp.Abstraction/Rules/Options/RuleTriggerOptions.cs @@ -24,16 +24,6 @@ public class RuleTriggerOptions public int SendMessageDelayMs { get; set; } = DefaultSendMessageDelayMs; public const int DefaultSendMessageDelayMs = 200; - - /// - /// How many triggered rules may run at the same time. Each one still gets its own - /// service scope, so concurrent runs do not share conversation, state or routing - /// services. Set to one to run them sequentially. Null falls back to - /// RuleSettings.MaxConcurrency, then to . - /// - public int? MaxConcurrency { get; set; } - - public const int DefaultMaxConcurrency = 5; } public class CriteriaOptions diff --git a/src/Infrastructure/BotSharp.Abstraction/Rules/Settings/RuleSettings.cs b/src/Infrastructure/BotSharp.Abstraction/Rules/Settings/RuleSettings.cs deleted file mode 100644 index eea1391d0..000000000 --- a/src/Infrastructure/BotSharp.Abstraction/Rules/Settings/RuleSettings.cs +++ /dev/null @@ -1,10 +0,0 @@ -namespace BotSharp.Abstraction.Rules.Settings; - -public class RuleSettings -{ - /// - /// How many triggered rules may run at the same time. Overridden per call by - /// RuleTriggerOptions.MaxConcurrency. Null falls back to the built-in default. - /// - public int? MaxConcurrency { get; set; } -} diff --git a/src/Infrastructure/BotSharp.Core.Rules/Engines/RuleEngine.cs b/src/Infrastructure/BotSharp.Core.Rules/Engines/RuleEngine.cs index 332eb722c..6d2083491 100644 --- a/src/Infrastructure/BotSharp.Core.Rules/Engines/RuleEngine.cs +++ b/src/Infrastructure/BotSharp.Core.Rules/Engines/RuleEngine.cs @@ -31,8 +31,7 @@ public async Task> Triggered(IRuleTrigger trigger, string te } }); - // Flatten the agent/rule pairs so they can be throttled as one unit, rather than - // running one agent's rules concurrently but the agents themselves one at a time. + // 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 @@ -40,60 +39,33 @@ public async Task> Triggered(IRuleTrigger trigger, string te .Select(r => (Agent: x, Rule: r))) .ToList(); - if (pendingRules.IsNullOrEmpty()) + foreach (var item in pendingRules) { - return newConversationIds; - } - - // Indexed so the returned conversation ids keep the rule order regardless of - // which run finishes first. - var convIds = new string?[pendingRules.Count]; - - // Per-call options win over the configured setting, which in turn wins over the built-in default. - var settings = _services.GetService(); - var maxConcurrency = options?.MaxConcurrency - ?? settings?.MaxConcurrency - ?? RuleTriggerOptions.DefaultMaxConcurrency; - - var parallelOptions = new ParallelOptions - { - MaxDegreeOfParallelism = Math.Max(1, maxConcurrency), - CancellationToken = cancellationToken - }; - - var indexedRules = pendingRules.Select((item, index) => (item.Agent, item.Rule, Index: index)); - - try - { - await Parallel.ForEachAsync(indexedRules, parallelOptions, async (item, token) => + try { - try + var convId = await RunRule(item.Agent, item.Rule, trigger, text, states, options, cancellationToken); + if (!string.IsNullOrEmpty(convId)) { - convIds[item.Index] = await RunRule(item.Agent, item.Rule, trigger, text, states, options, token); + newConversationIds.Add(convId); } - catch (Exception ex) when (ex is not OperationCanceledException) - { - // One misbehaving rule should not take down the rules that run alongside it. - _logger.LogError(ex, $"Error when running rule ({item.Rule.TriggerName}) for agent ({item.Agent.Name})."); - } - }); - } - 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. - var startedIds = CollectConversationIds(convIds); - _logger.LogWarning($"Rule trigger ({trigger.Name}) was cancelled after starting {startedIds.Count} conversation(s)."); - throw new RuleTriggerCanceledException(startedIds, cancellationToken, ex); + } + 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) + { + // 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})."); + } } - newConversationIds.AddRange(CollectConversationIds(convIds)); return newConversationIds; } - private static List CollectConversationIds(string?[] convIds) - => convIds.Where(x => !string.IsNullOrEmpty(x)).Select(x => x!).ToList(); - /// /// 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. @@ -145,8 +117,7 @@ private static List CollectConversationIds(string?[] convIds) var msg = !string.IsNullOrWhiteSpace(rule.Message) ? rule.Message : text; var convId = await SendMessageToAgent(sp, agent, trigger, text, msg, states); - // Hold the concurrency slot a little longer after sending, so a large batch of rules - // does not hammer the downstream provider the moment each slot frees up. + // Pause before the next rule, so a large batch does not hammer the downstream provider. var delay = options?.SendMessageDelayMs ?? RuleTriggerOptions.DefaultSendMessageDelayMs; if (delay > 0) { diff --git a/src/Infrastructure/BotSharp.Core.Rules/RulesPlugin.cs b/src/Infrastructure/BotSharp.Core.Rules/RulesPlugin.cs index 521c74160..5ee0d4d5d 100644 --- a/src/Infrastructure/BotSharp.Core.Rules/RulesPlugin.cs +++ b/src/Infrastructure/BotSharp.Core.Rules/RulesPlugin.cs @@ -21,10 +21,6 @@ public class RulesPlugin : IBotSharpPlugin public void RegisterDI(IServiceCollection services, IConfiguration config) { - var settings = new RuleSettings(); - config.Bind("Rule", settings); - services.AddSingleton(settings); - // Register rule engine services.AddScoped(); diff --git a/src/Infrastructure/BotSharp.Core.Rules/Using.cs b/src/Infrastructure/BotSharp.Core.Rules/Using.cs index de7af3130..54f48a78b 100644 --- a/src/Infrastructure/BotSharp.Core.Rules/Using.cs +++ b/src/Infrastructure/BotSharp.Core.Rules/Using.cs @@ -20,7 +20,6 @@ global using BotSharp.Abstraction.Repositories.Filters; global using BotSharp.Abstraction.Rules; global using BotSharp.Abstraction.Rules.Options; -global using BotSharp.Abstraction.Rules.Settings; global using BotSharp.Abstraction.Rules.Models; global using BotSharp.Abstraction.Rules.Hooks; global using BotSharp.Abstraction.Utilities;