From 8b85a9d412ef4149f7589406b978e726035b597d Mon Sep 17 00:00:00 2001 From: "mars.yu" Date: Wed, 28 Oct 2026 15:09:53 +0800 Subject: [PATCH] feat(agent-testing): run a suite's cases several at a time A run executed its cases one after another, so a suite cost the sum of every case's model wait: about 25 minutes for the 39-case suite whose cases each take 30 to 50 s, nearly all of it spent waiting. Cases are independent by construction. Each one opens its own conversation and its own DI scope, and the mock seam, the model override and the tool recorder all find their case by conversation id in a ConcurrentDictionary, never by an ambient "current case" -- so two cases in flight cannot see each other. The executor now runs up to MaxParallelCases work items at a time, taken from the run's own value, else the suite's (default 4, 1 to 16, rejected rather than clamped at the API). What must not overlap is this layer's own bookkeeping: the run document's counts are a read-modify-write of one document, and two of those interleaved lose whichever increment landed first. Each case's tally is therefore taken under a semaphore while only the case executions themselves overlap, which costs one round trip against a case's tens of seconds. Cancelling now distinguishes "stop starting new work" from "the run ended early because of it". A cancel that lands while the last case is executing holds nothing back, so that run still ends in its real status rather than being retroactively Cancelled. Cases already in flight cannot be aborted -- BotSharp's SendMessage takes no CancellationToken -- so they finish and record instead of losing work already paid for. The run records the concurrency it ran at, because a latency percentile from a four-at-a-time run is not the same measurement as one from a serial run, and the executor logs one line per finished case: interleaved cases make progress unreadable from a case's own log lines. Co-Authored-By: Claude Opus 5 --- .../Controllers/AgentTestController.cs | 33 ++- .../Models/AgentTestDtos.cs | 17 ++ .../Models/AgentTestRun.cs | 13 ++ .../Models/AgentTestSuite.cs | 48 ++++ .../Services/AgentTestRunExecutor.cs | 173 ++++++++++---- .../AgentTesting/AgentTestControllerTests.cs | 100 +++++++- .../AgentTesting/AgentTestRunExecutorTests.cs | 219 +++++++++++++++++- 7 files changed, 544 insertions(+), 59 deletions(-) diff --git a/src/Plugins/BotSharp.Plugin.AgentTesting/Controllers/AgentTestController.cs b/src/Plugins/BotSharp.Plugin.AgentTesting/Controllers/AgentTestController.cs index 770c24c0c..d9d097d2c 100644 --- a/src/Plugins/BotSharp.Plugin.AgentTesting/Controllers/AgentTestController.cs +++ b/src/Plugins/BotSharp.Plugin.AgentTesting/Controllers/AgentTestController.cs @@ -52,8 +52,13 @@ public async Task> ListSuites([FromQuery] string? agentId) => await _repo.ListSuitesAsync(agentId); [HttpPost("suites")] - public async Task CreateSuite([FromBody] AgentTestSuiteUpsertRequest request) + public async Task> CreateSuite([FromBody] AgentTestSuiteUpsertRequest request) { + if (ValidateParallelism(request.MaxParallelCases) is { } parallelError) + { + return BadRequest(parallelError); + } + var suite = new AgentTestSuite(); ApplySuite(suite, request); @@ -95,6 +100,11 @@ public async Task> UpdateSuite(string id, [FromBody request.Name = suite.Name; } + if (ValidateParallelism(request.MaxParallelCases) is { } parallelError) + { + return BadRequest(parallelError); + } + ApplySuite(suite, request); await _repo.UpsertSuiteAsync(suite); @@ -577,6 +587,11 @@ public async Task> TriggerRun( return BadRequest(modelError); } + if (ValidateParallelism(request?.MaxParallelCases) is { } parallelError) + { + return BadRequest(parallelError); + } + if (!suite.Enabled) { // The suite's own Enabled flag is advertised in the design as a way to turn a whole @@ -598,6 +613,9 @@ public async Task> TriggerRun( // Empty list and null mean the same thing downstream ("agent's own LlmConfig"), so // normalise to null rather than persisting an empty array that reads like a choice. Models = request?.Models is { Count: > 0 } ? request.Models : null, + // Zero means "the suite decides" -- AgentTestRunExecutor resolves it against the suite + // as the run starts and writes the figure it settled on back onto this document. + MaxParallelCases = request?.MaxParallelCases ?? 0, // This controller action is the only place in the whole trigger-to-execute path that // ever runs inside a real HTTP request -- AgentTestRunQueue's background loop has no // HttpContext at all, so if the triggering identity isn't captured here it can never @@ -712,6 +730,17 @@ public async Task>> GetMockTargets([FromQuery] string? return duplicate == null ? null : $"model '{duplicate.Key}' is listed more than once"; } + /// + /// Rejected rather than clamped, for a suite and for a run alike: a stored 100 that every run + /// silently reads as 16 is a setting that lies about itself every time someone opens it, and a + /// caller asking for 100 cases at a time has misunderstood what the harness will do to the + /// provider's rate limit. + /// + private static string? ValidateParallelism(int? maxParallelCases) + => maxParallelCases is { } value && (value < 1 || value > AgentTestConcurrency.Max) + ? $"'maxParallelCases' has to be between 1 and {AgentTestConcurrency.Max}" + : null; + private static void ApplySuite(AgentTestSuite suite, AgentTestSuiteUpsertRequest request) { suite.AgentId = request.AgentId; @@ -728,6 +757,8 @@ private static void ApplySuite(AgentTestSuite suite, AgentTestSuiteUpsertRequest suite.ExtraAllowedFunctions = request.ExtraAllowedFunctions ?? []; suite.ForceBlockedFunctions = request.ForceBlockedFunctions ?? []; suite.CaseTimeoutSeconds = request.CaseTimeoutSeconds; + // Null means the body omitted it -- keep what the suite has, like Enabled above. + suite.MaxParallelCases = request.MaxParallelCases ?? suite.MaxParallelCases; } private static void ApplyCase(AgentTestCase testCase, AgentTestCaseUpsertRequest request) diff --git a/src/Plugins/BotSharp.Plugin.AgentTesting/Models/AgentTestDtos.cs b/src/Plugins/BotSharp.Plugin.AgentTesting/Models/AgentTestDtos.cs index ddac1975f..bede63e89 100644 --- a/src/Plugins/BotSharp.Plugin.AgentTesting/Models/AgentTestDtos.cs +++ b/src/Plugins/BotSharp.Plugin.AgentTesting/Models/AgentTestDtos.cs @@ -26,6 +26,13 @@ public class AgentTestSuiteUpsertRequest public List ExtraAllowedFunctions { get; set; } = []; public List ForceBlockedFunctions { get; set; } = []; public int CaseTimeoutSeconds { get; set; } = 120; + + /// + /// Cases to execute at the same time, 1 to . Nullable for + /// the same reason is: a partial PUT that omits it must leave the suite's + /// value alone rather than resetting it to the default. + /// + public int? MaxParallelCases { get; set; } } /// @@ -211,6 +218,16 @@ public class AgentTestRunTriggerRequest /// and every AgentTestCaseResult records which model produced it. /// public List? Models { get; set; } + + /// + /// Cases to execute at the same time for this run only, 1 to + /// ; null falls back to the suite's own setting. + /// + /// Per run as well as per suite because the reason to change it is usually about the moment + /// rather than the suite: turning it down to 1 to read a flaky case's logs without three other + /// cases interleaved in them, or because the provider is already busy with something else. + /// + public int? MaxParallelCases { get; set; } } /// diff --git a/src/Plugins/BotSharp.Plugin.AgentTesting/Models/AgentTestRun.cs b/src/Plugins/BotSharp.Plugin.AgentTesting/Models/AgentTestRun.cs index 2111649d4..25116fa0b 100644 --- a/src/Plugins/BotSharp.Plugin.AgentTesting/Models/AgentTestRun.cs +++ b/src/Plugins/BotSharp.Plugin.AgentTesting/Models/AgentTestRun.cs @@ -41,6 +41,19 @@ public class AgentTestRun : MongoBase /// public List? Models { get; set; } + /// + /// Cases this run executed at the same time. Written by the executor as the run starts, from + /// the trigger request's value or the suite's -- a record of what actually happened, not a + /// request. + /// + /// It belongs on the run because the performance figures below are only comparable between two + /// runs that were produced under the same concurrency: a case's own latency includes whatever + /// queueing the other in-flight cases caused at the provider, so a P95 from a four-at-a-time run + /// and one from a serial run are different measurements. Zero on a run document written before + /// this field existed, which reads as "unknown, it ran serially". + /// + public int MaxParallelCases { get; set; } + public int TotalCount { get; set; } public int PassedCount { get; set; } public int FailedCount { get; set; } diff --git a/src/Plugins/BotSharp.Plugin.AgentTesting/Models/AgentTestSuite.cs b/src/Plugins/BotSharp.Plugin.AgentTesting/Models/AgentTestSuite.cs index ce511d220..72723eecb 100644 --- a/src/Plugins/BotSharp.Plugin.AgentTesting/Models/AgentTestSuite.cs +++ b/src/Plugins/BotSharp.Plugin.AgentTesting/Models/AgentTestSuite.cs @@ -22,6 +22,54 @@ public class AgentTestSuite : MongoBase public int CaseTimeoutSeconds { get; set; } = 120; + /// + /// How many cases of this suite execute at the same time. One run's cases are independent by + /// construction -- each one opens its own conversation and gets its own DI scope, and the mock + /// seam is keyed by conversation id -- so the only thing serial execution ever bought was a + /// gentler load on the model provider. + /// + /// Wall clock is what this costs: a case spends nearly all of its time waiting on the model, so + /// a suite of forty 40-second cases takes half an hour serially and a few minutes at four at a + /// time. Four rather than "as many as there are cases" because the limit that bites first is the + /// provider's rate limit, and a 429 storm turns a slow run into a failed one. + /// + /// A stored suite written before this field existed has no value for it, which deserialises to + /// this default -- so existing suites speed up without being touched. Set it to 1 for a suite + /// whose cases must not overlap. + /// + public int MaxParallelCases { get; set; } = AgentTestConcurrency.Default; + public DateTime CreateDate { get; set; } = DateTime.UtcNow; public DateTime UpdateDate { get; set; } = DateTime.UtcNow; } + +/// +/// The one place that decides how many cases of a run execute at the same time, so the suite +/// default, the per-run override and the API's validation message can never drift apart. +/// +public static class AgentTestConcurrency +{ + /// Cases at a time when neither the run nor the suite says otherwise. + public const int Default = 4; + + /// + /// The highest value accepted. Not a technical ceiling -- the harness would happily open + /// thirty conversations -- but past a handful the provider's rate limit, not the host, decides + /// what happens, and a run that fails on 429s is worse than a slow one. Raise it deliberately. + /// + public const int Max = 16; + + /// + /// The effective degree of parallelism for a run: the run's own value when it has one, else the + /// suite's, else , clamped to [1, ]. + /// + /// Zero and negative mean "not set" rather than "no parallelism": that is what a run document + /// written before this field existed deserialises to, and treating it as 0 would stop such a + /// run from executing anything at all. + /// + public static int Resolve(int runValue, int suiteValue) + { + var chosen = runValue > 0 ? runValue : suiteValue > 0 ? suiteValue : Default; + return Math.Clamp(chosen, 1, Max); + } +} diff --git a/src/Plugins/BotSharp.Plugin.AgentTesting/Services/AgentTestRunExecutor.cs b/src/Plugins/BotSharp.Plugin.AgentTesting/Services/AgentTestRunExecutor.cs index 6a293cee1..62802326a 100644 --- a/src/Plugins/BotSharp.Plugin.AgentTesting/Services/AgentTestRunExecutor.cs +++ b/src/Plugins/BotSharp.Plugin.AgentTesting/Services/AgentTestRunExecutor.cs @@ -5,11 +5,20 @@ namespace BotSharp.Plugin.AgentTesting.Services; /// -/// Run-level orchestration: turns an already-created AgentTestRun into serial execution of every -/// enabled case in its suite. +/// Run-level orchestration: turns an already-created AgentTestRun into execution of every enabled +/// case in its suite, up to AgentTestConcurrency.Resolve(run, suite) of them at a time. +/// +/// Concurrency lives here and nowhere else. Two things make it safe at this layer rather than +/// needing changes further down: every case already gets its own DI scope (see ScopedCaseRunner +/// below) and its own conversation id, and the mock seam, the model override and the tool recorder +/// all find their case by that conversation id in a ConcurrentDictionary (AgentTestRunRegistry) -- +/// never by an ambient or static "current case". So two cases in flight cannot see each other. +/// What is NOT safe to overlap is this class's own bookkeeping: the run document's counts are a +/// read-modify-write of one document, so each case's tally is taken under `bookkeeping` while only +/// the case executions themselves overlap. /// /// This class knows nothing about DI scopes -- it calls the single ICaseRunner instance handed to -/// its constructor, and the loop below decides how often and when. "A fresh DI scope per case" is +/// its constructor, and the loop below decides how many, how often and when. "A fresh DI scope per case" is /// not done here; AgentTestRunQueue achieves it by injecting an ICaseRunner decorator that wraps /// IServiceScopeFactory/IServiceProvider (see AgentTestRunQueue.ScopedCaseRunner). That decorator's /// RunAsync opens a new scope on every call -- that is, per case -- resolves the real @@ -23,7 +32,7 @@ namespace BotSharp.Plugin.AgentTesting.Services; /// makes it the only field a whole-document ReplaceOneAsync could revert to a stale value -- /// TotalCount/PassedCount/... are written by this class alone. Hence the re-read AFTER each case /// rather than before: the object it returns becomes the `run` this method then mutates and -/// persists, so both "should the next case run" and "will this write clobber an external one" are +/// persists, so both "should the next case start" and "will this write clobber an external one" are /// decided from the state as of the moment that case finished, not from the copy read at the very /// top and never refreshed. One read covers both questions, so no separate check is needed before /// it. @@ -124,6 +133,13 @@ public async Task ExecuteAsync(string runId, CancellationToken ct) } } + // The run's own value wins, then the suite's, then the default -- resolved once here and + // written onto the run, so a reader of the finished document can tell what the latency + // figures below were produced under rather than having to guess from the suite's value as + // it stands today. + var parallelism = AgentTestConcurrency.Resolve(run.MaxParallelCases, suite.MaxParallelCases); + + run.MaxParallelCases = parallelism; run.Status = AgentTestStatus.Running; run.StartedAt = DateTime.UtcNow; run.TotalCount = 0; @@ -132,7 +148,12 @@ public async Task ExecuteAsync(string runId, CancellationToken ct) run.ErrorCount = 0; await _repo.UpdateRunAsync(run); - var cancelled = false; + // "Stop starting new work" (cancelRequested) is a different fact from "the run ended + // early because of it" (cancelSkipped). A cancel that lands while the LAST case is + // executing stops nothing -- the run finished everything it was ever going to run -- so + // only a case that was actually held back turns the terminal status into Cancelled. + var cancelRequested = false; + var cancelSkipped = 0; // The model dimension. An absent/empty Models list means "one pass, using each agent's own // LlmConfig" -- byte-for-byte the behavior from before multi-model existed, which is what @@ -150,8 +171,34 @@ public async Task ExecuteAsync(string runId, CancellationToken ct) .SelectMany(testCase => models.Select(model => (Case: testCase, Model: model))) .ToList(); - foreach (var (testCase, model) in workItems) + // Serialises everything that touches the run document or the result collection, so the + // only thing that actually overlaps is the case executions. Each pass below is a + // read-modify-write of one document (GetRunAsync, then UpdateRunAsync -- a whole-document + // replace), and two of those interleaved lose whichever increment landed first. It costs + // nothing worth having: a case takes tens of seconds, this takes one round trip. + var bookkeeping = new SemaphoreSlim(1, 1); + + _logger.LogInformation( + "Agent test run {RunId} is executing {WorkItemCount} case(s) {Parallelism} at a time.", + runId, workItems.Count, parallelism); + + // The host is shutting down. Set by whichever worker notices first and read by the others, + // so the remaining work items return without starting rather than each being half-run + // through a disappearing process. + var hostStopping = false; + + // MaxDegreeOfParallelism, not the CancellationToken: passing `ct` here would surface host + // shutdown as an OperationCanceledException out of this method, and this run's row has to + // be left exactly as it is on that path (see the hostStopping check inside). At + // parallelism 1 this runs the work items in order, one at a time, which is what a suite + // that sets MaxParallelCases = 1 is asking for. + await Parallel.ForEachAsync( + workItems, + new ParallelOptions { MaxDegreeOfParallelism = parallelism }, + async (item, _) => { + var (testCase, model) = item; + if (ct.IsCancellationRequested) { // The host itself is shutting down (BackgroundService's stoppingToken), not a @@ -159,17 +206,19 @@ public async Task ExecuteAsync(string runId, CancellationToken ct) // the shutdown grace period to persist a different terminal status here -- the // queue's own startup reconciliation step sweeps any Running row a killed process // left behind into Error the next time it boots. + Volatile.Write(ref hostStopping, true); return; } - // `run` here is either the initial load (for the first case) or the post-case read - // from the PREVIOUS iteration (see below) -- either way it is the freshest state this - // method has seen, so this check can never miss a cancel that arrived any time up to - // "the moment the previous case finished." - if (run.CancelRequested) + // Set under `bookkeeping` by whichever case first read CancelRequested back as true + // (see below), so this can never miss a cancel that arrived up to "the moment the last + // case to finish finished." The cases already in flight are not abortable -- BotSharp's + // SendMessage takes no CancellationToken, see AgentTestCaseRunner -- so a cancel stops + // the ones that have not started and lets the running ones finish and record. + if (Volatile.Read(ref cancelRequested)) { - cancelled = true; - break; + Interlocked.Increment(ref cancelSkipped); + return; } AgentTestCaseResult result; @@ -200,45 +249,75 @@ public async Task ExecuteAsync(string runId, CancellationToken ct) }; } - await _repo.AddCaseResultAsync(result); - - // Re-read AFTER the case ran, not before. This is what picks up a concurrent - // POST /runs/{id}/cancel that landed WHILE the case was executing -- both for the - // NEXT iteration's check above, and so the persist a few lines down (a whole-document - // ReplaceOneAsync) never overwrites that external write with a stale - // CancelRequested=false. `run` becomes this fresh object for the remainder of the - // method (including the terminal write after the loop, if this was the last case) -- - // that's what keeps a cancel that arrives during the very last case's own execution - // from being silently erased, with no extra read needed after the loop. - run = await _repo.GetRunAsync(runId) ?? run; - - run.TotalCount++; - TallyRoutingAccuracy(run, result); - switch (result.Status) + // CancellationToken.None: the tally for a case that already ran has to be recorded + // even as the host goes down, and it is a couple of writes. + await bookkeeping.WaitAsync(CancellationToken.None); + try { - case AgentTestStatus.Passed: - run.PassedCount++; - break; - case AgentTestStatus.Failed: - run.FailedCount++; - break; - default: - // Error, Cancelled (e.g. a per-case timeout inside the real case runner), or - // any other non-Passed/Failed status all count against ErrorCount -- AgentTestRun - // has no separate CancelledCount field, and a per-case timeout is an - // infrastructure failure for THAT case, not a graceful run-level cancellation. - run.ErrorCount++; - break; + await _repo.AddCaseResultAsync(result); + + // Re-read AFTER the case ran, not before. This is what picks up a concurrent + // POST /runs/{id}/cancel that landed WHILE the case was executing -- both for the + // cancel check the work items that have not started yet make above, and so the + // persist a few lines down (a whole-document ReplaceOneAsync) never overwrites that + // external write with a stale CancelRequested=false. `run` becomes this fresh + // object for the remainder of the method (including the terminal write after the + // loop, if this was the last case) -- that's what keeps a cancel that arrives + // during the very last case's own execution from being silently erased, with no + // extra read needed after the loop. Safe to assign across workers because every + // reader and writer of it holds `bookkeeping`, and the loop has joined before the + // terminal write reads it. + run = await _repo.GetRunAsync(runId) ?? run; + + run.TotalCount++; + TallyRoutingAccuracy(run, result); + switch (result.Status) + { + case AgentTestStatus.Passed: + run.PassedCount++; + break; + case AgentTestStatus.Failed: + run.FailedCount++; + break; + default: + // Error, Cancelled (e.g. a per-case timeout inside the real case runner), or + // any other non-Passed/Failed status all count against ErrorCount -- AgentTestRun + // has no separate CancelledCount field, and a per-case timeout is an + // infrastructure failure for THAT case, not a graceful run-level cancellation. + run.ErrorCount++; + break; + } + + if (run.CancelRequested) + { + Volatile.Write(ref cancelRequested, true); + } + + // Progress, one line per finished case. Serial execution used to make progress + // readable from the order of a case's own log lines; several cases interleaved do + // not, so the run says where it is itself. + _logger.LogInformation( + "Agent test run {RunId}: {Completed}/{Total} -- case {CaseName} {Status}.", + runId, run.TotalCount, workItems.Count, testCase.Name, result.Status); + + // Real-time accumulation, per case -- not just at the very end. Without this, GET + // /agent-test/runs/{id} shows 0/0/0/0 for the run's entire duration, and a process + // death mid-run leaves the startup sweep's Error row claiming zero cases ran even + // though N AgentTestCaseResult rows already exist for it. + await _repo.UpdateRunAsync(run); + } + finally + { + bookkeeping.Release(); } + }); - // Real-time accumulation, per case -- not just at the very end. Without this, GET - // /agent-test/runs/{id} shows 0/0/0/0 for the run's entire duration, and a process - // death mid-run leaves the startup sweep's Error row claiming zero cases ran even - // though N AgentTestCaseResult rows already exist for it. - await _repo.UpdateRunAsync(run); + if (hostStopping) + { + return; } - run.Status = cancelled + run.Status = cancelSkipped > 0 ? AgentTestStatus.Cancelled : run.FailedCount == 0 && run.ErrorCount == 0 ? AgentTestStatus.Passed diff --git a/tests/BotSharp.Core.UnitTests/AgentTesting/AgentTestControllerTests.cs b/tests/BotSharp.Core.UnitTests/AgentTesting/AgentTestControllerTests.cs index ffb6ef79d..a1060a33b 100644 --- a/tests/BotSharp.Core.UnitTests/AgentTesting/AgentTestControllerTests.cs +++ b/tests/BotSharp.Core.UnitTests/AgentTesting/AgentTestControllerTests.cs @@ -600,7 +600,7 @@ public async Task CreateSuite_defaults_to_enabled_when_the_request_omits_it() var result = await controller.CreateSuite(new AgentTestSuiteUpsertRequest { AgentId = "agent-1", Name = "s" }); - Assert.True(result.Enabled); + Assert.True(result.Value!.Enabled); } [Fact] @@ -612,7 +612,103 @@ public async Task CreateSuite_honors_an_explicit_disabled_flag() var result = await controller.CreateSuite( new AgentTestSuiteUpsertRequest { AgentId = "agent-1", Name = "s", Enabled = false }); - Assert.False(result.Enabled); + Assert.False(result.Value!.Enabled); + } + + // ---- How many cases a run executes at once ---------------------------------------------- + + [Fact] + public async Task CreateSuite_defaults_to_running_several_cases_at_a_time() + { + var repo = new InMemoryRepo(); + var controller = BuildController(repo, out _); + + var result = await controller.CreateSuite(new AgentTestSuiteUpsertRequest { AgentId = "agent-1", Name = "s" }); + + Assert.Equal(AgentTestConcurrency.Default, result.Value!.MaxParallelCases); + } + + [Fact] + public async Task UpdateSuite_keeps_the_stored_parallelism_when_the_request_omits_it() + { + // Same hazard as Enabled: a partial PUT that only meant to change the timeout must not + // quietly put a suite deliberately pinned to one case at a time back to the default. + var repo = new InMemoryRepo(); + repo.Suites["suite-1"] = new AgentTestSuite + { + Id = "suite-1", AgentId = "agent-1", Name = "s", MaxParallelCases = 1 + }; + var controller = BuildController(repo, out _); + + await controller.UpdateSuite("suite-1", new AgentTestSuiteUpsertRequest { CaseTimeoutSeconds = 300 }); + + Assert.Equal(1, repo.Suites["suite-1"].MaxParallelCases); + } + + [Theory] + [InlineData(0)] + [InlineData(-1)] + [InlineData(100)] + public async Task CreateSuite_rejects_a_parallelism_outside_the_allowed_range(int value) + { + // Rejected rather than clamped: a stored 100 that every run silently reads as 16 is a + // setting that lies about itself every time someone opens the suite. + var repo = new InMemoryRepo(); + var controller = BuildController(repo, out _); + + var result = await controller.CreateSuite(new AgentTestSuiteUpsertRequest + { + AgentId = "agent-1", Name = "s", MaxParallelCases = value + }); + + Assert.IsType(result.Result); + Assert.Empty(repo.Suites); + } + + [Fact] + public async Task TriggerRun_persists_the_requested_parallelism_on_the_run() + { + var repo = new InMemoryRepo(); + repo.Suites["suite-1"] = EnabledSuite(); + var controller = BuildController(repo, out _); + + var result = await controller.TriggerRun( + "suite-1", new AgentTestRunTriggerRequest { MaxParallelCases = 2 }); + + Assert.Null(result.Result); + Assert.Equal(2, result.Value!.MaxParallelCases); + } + + [Fact] + public async Task TriggerRun_leaves_the_parallelism_to_the_suite_when_the_body_omits_it() + { + // Zero is "the suite decides" -- AgentTestRunExecutor resolves it as the run starts and + // writes back the figure it settled on. + var repo = new InMemoryRepo(); + repo.Suites["suite-1"] = EnabledSuite(); + var controller = BuildController(repo, out _); + + var result = await controller.TriggerRun("suite-1", new AgentTestRunTriggerRequest()); + + Assert.Equal(0, result.Value!.MaxParallelCases); + } + + [Theory] + [InlineData(0)] + [InlineData(-3)] + [InlineData(64)] + public async Task TriggerRun_rejects_a_parallelism_outside_the_allowed_range(int value) + { + var repo = new InMemoryRepo(); + repo.Suites["suite-1"] = EnabledSuite(); + var controller = BuildController(repo, out var queue); + + var result = await controller.TriggerRun( + "suite-1", new AgentTestRunTriggerRequest { MaxParallelCases = value }); + + Assert.IsType(result.Result); + Assert.Empty(queue.Enqueued); + Assert.Empty(repo.Runs); } // ---- Item 8c: cancelling an already-terminal run is rejected ---------------------------- diff --git a/tests/BotSharp.Core.UnitTests/AgentTesting/AgentTestRunExecutorTests.cs b/tests/BotSharp.Core.UnitTests/AgentTesting/AgentTestRunExecutorTests.cs index b2527f583..be85a6e82 100644 --- a/tests/BotSharp.Core.UnitTests/AgentTesting/AgentTestRunExecutorTests.cs +++ b/tests/BotSharp.Core.UnitTests/AgentTesting/AgentTestRunExecutorTests.cs @@ -15,10 +15,13 @@ namespace BotSharp.Core.UnitTests.AgentTesting; /// -/// Run-level orchestration: cases run serially, one failing case does not affect the rest, the counts -/// are right, and cancellation takes effect promptly. Serial execution is a safety choice rather than -/// a performance one -- cases share external dependencies, and running them concurrently would let -/// them pollute each other's state. +/// Run-level orchestration: cases run up to MaxParallelCases at a time, one failing case does not +/// affect the rest, the counts are right whether or not cases overlapped, and cancellation takes +/// effect promptly. +/// +/// Both fake repos below hand out a run pinned to MaxParallelCases = 1, so every test that asserts +/// an ORDER (which case ran first, which result landed first) states what it depends on instead of +/// relying on a default that is now 4. The concurrency tests set it themselves. /// public class AgentTestRunExecutorTests { @@ -26,7 +29,9 @@ private sealed class InMemoryRepo : IAgentTestRepository { public AgentTestSuite Suite = new() { Id = "suite-1", AgentId = "agent-1", Name = "s" }; public List Cases = []; - public AgentTestRun Run = new() { Id = "run-1", SuiteId = "suite-1" }; + // Serial unless a test says otherwise: the tests below that assert an execution or + // result ORDER only mean anything one case at a time, and the default is 4. + public AgentTestRun Run = new() { Id = "run-1", SuiteId = "suite-1", MaxParallelCases = 1 }; public List Results = []; public Task GetSuiteAsync(string id) => Task.FromResult(Suite); @@ -68,7 +73,7 @@ private sealed class CloningRunRepo : IAgentTestRepository public List Cases = []; public List Results = []; - private AgentTestRun _stored = new() { Id = "run-1", SuiteId = "suite-1" }; + private AgentTestRun _stored = new() { Id = "run-1", SuiteId = "suite-1", MaxParallelCases = 1 }; /// The current backing state, for assertions -- a clone, so tests can't cheat by mutating it directly. public AgentTestRun Stored => Clone(_stored); @@ -105,6 +110,7 @@ public Task DeleteRunAsync(string id) Status = source.Status, TriggeredBy = source.TriggeredBy, CaseIds = source.CaseIds, + MaxParallelCases = source.MaxParallelCases, TotalCount = source.TotalCount, PassedCount = source.PassedCount, FailedCount = source.FailedCount, @@ -154,8 +160,17 @@ private static ILlmProviderService PricingFor(string provider, string model, flo private sealed class DelegatingCaseRunner(Func run) : ICaseRunner { - /// Every (case, model) pair the executor asked for, in the order it asked. - public List<(string CaseId, string? Model)> Invocations { get; } = []; + private readonly List<(string CaseId, string? Model)> _invocations = []; + + /// + /// Every (case, model) pair the executor asked for, in the order it asked. Locked because + /// the executor can call RunAsync from several threads at once -- only meaningful as an + /// ORDER for a run pinned to MaxParallelCases = 1. + /// + public List<(string CaseId, string? Model)> Invocations + { + get { lock (_invocations) return [.. _invocations]; } + } /// /// Set instead of relying on the constructor delegate when a test's outcome has to depend on @@ -167,7 +182,7 @@ private sealed class DelegatingCaseRunner(Func RunAsync( AgentTestSuite suite, AgentTestCase testCase, string runId, TestModel? model, CancellationToken ct) { - Invocations.Add((testCase.Id, model?.Model)); + lock (_invocations) _invocations.Add((testCase.Id, model?.Model)); var result = RunWithModel != null ? RunWithModel(testCase, model) : run(testCase); // The real runner stamps these onto the result; mirror it so tests can assert that a // result can be attributed back to the model that produced it. @@ -330,6 +345,192 @@ public async Task Stops_between_cases_once_cancellation_is_requested() Assert.Equal(AgentTestStatus.Cancelled, repo.Run.Status); } + /// + /// An ICaseRunner that records how many cases were in flight at once, and holds each one until + /// it is released -- the only way to observe overlap rather than infer it from timing. + /// + private sealed class ConcurrencyProbeRunner : ICaseRunner + { + private readonly TaskCompletionSource _release = new(TaskCreationOptions.RunContinuationsAsynchronously); + private readonly object _gate = new(); + private int _inFlight; + + /// The most cases that were ever executing at the same moment. + public int PeakInFlight { get; private set; } + + public int Started { get; private set; } + + /// + /// Completes once cases have started and are waiting. Gives up + /// after ten seconds rather than hanging the test run: "they never overlapped" is the + /// failure this probe exists to catch, and a hung suite reports it as nothing at all. + /// + public async Task WaitForStartedAsync(int count) + { + for (var attempt = 0; attempt < 2000; attempt++) + { + lock (_gate) + { + if (Started >= count) return; + } + await Task.Delay(5); + } + + throw new TimeoutException($"only {Started} case(s) ever started at once; expected {count}"); + } + + public void ReleaseAll() => _release.TrySetResult(); + + public async Task RunAsync( + AgentTestSuite suite, AgentTestCase testCase, string runId, TestModel? model, CancellationToken ct) + { + lock (_gate) + { + Started++; + _inFlight++; + PeakInFlight = Math.Max(PeakInFlight, _inFlight); + } + + await _release.Task; + + lock (_gate) + { + _inFlight--; + } + + return new AgentTestCaseResult { CaseId = testCase.Id, Status = AgentTestStatus.Passed }; + } + } + + [Fact] + public async Task Cases_run_several_at_a_time_up_to_the_runs_limit() + { + // The point of the feature: a suite of cases that each spend ~40s waiting on the model used + // to take the sum of those waits. This is the only test that proves cases actually overlap + // -- every other one here would pass just as well against the old serial loop. + var repo = new InMemoryRepo { Cases = [CaseNamed("a"), CaseNamed("b"), CaseNamed("c"), CaseNamed("d"), CaseNamed("e")] }; + repo.Run.MaxParallelCases = 3; + var runner = new ConcurrencyProbeRunner(); + var executor = new AgentTestRunExecutor(repo, runner, NullLogger.Instance); + + var execution = executor.ExecuteAsync("run-1", CancellationToken.None); + + // Three cases hanging in the runner at once is only reachable if the executor started them + // without waiting for the first to finish. + await runner.WaitForStartedAsync(3); + runner.ReleaseAll(); + await execution; + + Assert.Equal(3, runner.PeakInFlight); + // ... and the limit is a limit: the remaining two waited for a slot rather than all five + // going at once. + Assert.Equal(5, repo.Run.TotalCount); + Assert.Equal(5, repo.Run.PassedCount); + Assert.Equal(5, repo.Results.Count); + Assert.Equal(5, repo.Results.Select(r => r.CaseId).Distinct().Count()); + } + + [Fact] + public async Task A_run_that_asks_for_no_parallelism_takes_the_suites() + { + var repo = new InMemoryRepo { Cases = [CaseNamed("a"), CaseNamed("b"), CaseNamed("c")] }; + // What POST .../run writes when its body omits maxParallelCases. + repo.Run.MaxParallelCases = 0; + repo.Suite.MaxParallelCases = 2; + var runner = new ConcurrencyProbeRunner(); + var executor = new AgentTestRunExecutor(repo, runner, NullLogger.Instance); + + var execution = executor.ExecuteAsync("run-1", CancellationToken.None); + await runner.WaitForStartedAsync(2); + runner.ReleaseAll(); + await execution; + + Assert.Equal(2, runner.PeakInFlight); + // Written back onto the run: a latency percentile from a two-at-a-time run is not the same + // measurement as one from a serial run, and the suite's setting today says nothing about + // what this run did. + Assert.Equal(2, repo.Run.MaxParallelCases); + } + + [Fact] + public async Task A_run_document_written_before_parallelism_existed_runs_at_the_default() + { + // Neither the run nor the suite carries a value -- which is exactly what deserialising an + // old Mongo document gives, since AgentTestSuite's own default only applies to objects this + // process constructs. Zero must not read as "no cases at a time." + var repo = new InMemoryRepo { Cases = [CaseNamed("a")] }; + repo.Run.MaxParallelCases = 0; + repo.Suite.MaxParallelCases = 0; + var executor = Build(repo, c => new AgentTestCaseResult { CaseId = c.Id, Status = AgentTestStatus.Passed }); + + await executor.ExecuteAsync("run-1", CancellationToken.None); + + Assert.Single(repo.Results); + Assert.Equal(AgentTestConcurrency.Default, repo.Run.MaxParallelCases); + } + + [Fact] + public async Task A_parallelism_above_the_ceiling_is_clamped_rather_than_honoured() + { + // The API rejects such a value, so this is only reachable from a hand-edited document -- + // but "the document said 500" must not mean 500 conversations at once. + var repo = new InMemoryRepo { Cases = [CaseNamed("a")] }; + repo.Run.MaxParallelCases = 500; + var executor = Build(repo, c => new AgentTestCaseResult { CaseId = c.Id, Status = AgentTestStatus.Passed }); + + await executor.ExecuteAsync("run-1", CancellationToken.None); + + Assert.Equal(AgentTestConcurrency.Max, repo.Run.MaxParallelCases); + } + + [Fact] + public async Task Every_case_is_counted_exactly_once_when_cases_overlap() + { + // The run document's counts are a read-modify-write of one document. Without the + // executor's bookkeeping lock, concurrent tallies lose increments -- and a run that + // reports 7 of 10 cases, all passed, looks like a short suite rather than a broken count. + var repo = new InMemoryRepo { Cases = [.. Enumerable.Range(0, 24).Select(i => CaseNamed($"c{i}"))] }; + repo.Run.MaxParallelCases = AgentTestConcurrency.Max; + var executor = Build(repo, c => new AgentTestCaseResult + { + CaseId = c.Id, + Status = int.Parse(c.Id[1..]) % 5 == 0 ? AgentTestStatus.Failed : AgentTestStatus.Passed + }); + + await executor.ExecuteAsync("run-1", CancellationToken.None); + + Assert.Equal(24, repo.Run.TotalCount); + Assert.Equal(5, repo.Run.FailedCount); + Assert.Equal(19, repo.Run.PassedCount); + Assert.Equal(0, repo.Run.ErrorCount); + Assert.Equal(24, repo.Results.Select(r => r.CaseId).Distinct().Count()); + Assert.Equal(AgentTestStatus.Failed, repo.Run.Status); + } + + [Fact] + public async Task A_cancel_stops_the_cases_that_have_not_started_yet() + { + // Cases already in flight cannot be aborted (BotSharp's SendMessage takes no + // CancellationToken -- see AgentTestCaseRunner), so cancelling a parallel run means the + // ones running finish and record, and nothing new starts. The alternative -- dropping the + // results of work already paid for -- would make cancel lose data. + var repo = new InMemoryRepo { Cases = [.. Enumerable.Range(0, 8).Select(i => CaseNamed($"c{i}"))] }; + repo.Run.MaxParallelCases = 2; + var executor = Build(repo, c => + { + repo.Run.CancelRequested = true; + return new AgentTestCaseResult { CaseId = c.Id, Status = AgentTestStatus.Passed }; + }); + + await executor.ExecuteAsync("run-1", CancellationToken.None); + + Assert.Equal(AgentTestStatus.Cancelled, repo.Run.Status); + // The two that were already in flight when the cancel was seen may both record; the other + // six must not have run at all. + Assert.InRange(repo.Results.Count, 1, 2); + Assert.Equal(repo.Results.Count, repo.Run.TotalCount); + } + [Fact] public async Task Runs_only_the_cases_named_by_CaseIds_when_specified() {