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() {