diff --git a/src/Volcanic.AgenticCore.A2A.Server/Agent/A2AAgentHandler.cs b/src/Volcanic.AgenticCore.A2A.Server/Agent/A2AAgentHandler.cs index 9afe0a4..8b61e8d 100644 --- a/src/Volcanic.AgenticCore.A2A.Server/Agent/A2AAgentHandler.cs +++ b/src/Volcanic.AgenticCore.A2A.Server/Agent/A2AAgentHandler.cs @@ -3,6 +3,7 @@ using A2A; using Microsoft.Extensions.Logging; using Microsoft.Extensions.Options; using Volcanic.AgenticCore.A2A.Server.Callers; +using Volcanic.AgenticCore.A2A.Server.Tasks; using Volcanic.AgenticCore.Abstractions.Events; using Volcanic.AgenticCore.Abstractions.Processes; @@ -13,9 +14,10 @@ namespace Volcanic.AgenticCore.A2A.Server.Agent; /// input-required (the agent called request_input), failed (model error, hop limit, bad file, busy context, any /// unexpected exception) or canceled (the request token or a CancelTask). The task is registered as live in the pool /// before its first event and stays live until its final status is written, so restart recovery never touches it. +/// A second run of a task whose run is unfinished (a racing continuation) throws before writing anything. /// internal sealed class A2AAgentHandler( - A2ACaller caller, A2ASessionPool pool, IOptions options, ISecretRedactor redactor, A2AFileDownloader downloader, + A2ACaller caller, WorkspaceTaskStore store, A2ASessionPool pool, IOptions options, ISecretRedactor redactor, A2AFileDownloader downloader, TimeProvider time, ILogger logger) : IAgentHandler { /// Bound on the final status write: after a client disconnect the bounded event queue may never be read. @@ -128,20 +130,27 @@ internal sealed class A2AAgentHandler( } /// - /// The continued task as the store holds it after the SDK appended the caller's message, moved to working: the - /// question (status message) goes to history, as a status update would put it. + /// The continued task, read fresh from the store (the SDK has just appended the caller's message), moved to working: + /// history gets the question (the status message) and then the caller's answer, and the previous run's + /// response artifact is dropped so this run's answer starts empty. /// private AgentTask Resumed(RequestContext context) { - var json = System.Text.Json.JsonSerializer.Serialize(context.Task!, A2AJsonUtilities.DefaultOptions); - var task = System.Text.Json.JsonSerializer.Deserialize(json, A2AJsonUtilities.DefaultOptions)!; + var task = store.Peek(context.TaskId) ?? Clone(context.Task!); var history = task.History ??= []; - if (!history.Any(m => m.MessageId == context.Message.MessageId)) history.Add(context.Message); + history.RemoveAll(m => m.MessageId == context.Message.MessageId); if (task.Status.Message is { } question) history.Add(question); - task.Status = new global::A2A.TaskStatus { State = TaskState.Working, Timestamp = time.GetUtcNow() }; + history.Add(context.Message); + task.Artifacts?.RemoveAll(a => a.ArtifactId == A2ARunMapper.ResponseArtifactId); + // Same clock as the SDK's TaskUpdater, so the task's timestamps stay in order. + task.Status = new global::A2A.TaskStatus { State = TaskState.Working, Timestamp = DateTimeOffset.UtcNow }; return task; } + private static AgentTask Clone(AgentTask task) => + System.Text.Json.JsonSerializer.Deserialize( + System.Text.Json.JsonSerializer.Serialize(task, A2AJsonUtilities.DefaultOptions), A2AJsonUtilities.DefaultOptions)!; + /// Write the final status (and answer artifact) with a bounded token; returns the label to log. private async Task WriteFinalAsync(RequestContext context, TaskUpdater updater, Result result) { diff --git a/src/Volcanic.AgenticCore.A2A.Server/Agent/A2ARunMapper.cs b/src/Volcanic.AgenticCore.A2A.Server/Agent/A2ARunMapper.cs index 0c8ed8c..60f93b1 100644 --- a/src/Volcanic.AgenticCore.A2A.Server/Agent/A2ARunMapper.cs +++ b/src/Volcanic.AgenticCore.A2A.Server/Agent/A2ARunMapper.cs @@ -28,6 +28,7 @@ internal sealed class A2ARunMapper : IAsyncDisposable private readonly CancellationTokenSource _cts; private readonly CancellationToken _ct; private readonly Task _pump; + private bool _textStarted; public A2ARunMapper(TaskUpdater updater, ISecretRedactor redactor, ILogger logger, CancellationToken cancellationToken) { @@ -48,8 +49,11 @@ internal sealed class A2ARunMapper : IAsyncDisposable switch (e) { case AgentSessionEvent.MessageUpdate { AssistantMessageEvent: AssistantMessageEvent.TextDelta delta }: + // The run's first chunk replaces whatever response artifact the task already has. + var append = _textStarted; + _textStarted = true; Post(() => _updater.AddArtifactAsync([Part.FromText(delta.Delta)], ResponseArtifactId, "response", null, - lastChunk: false, append: true, _ct)); + lastChunk: false, append: append, _ct)); break; case AgentSessionEvent.MessageEnd { Message: AssistantMessage message }: if (message.StopReason == StopReason.Aborted) diff --git a/src/Volcanic.AgenticCore.A2A.Server/Agent/A2ASessionPool.cs b/src/Volcanic.AgenticCore.A2A.Server/Agent/A2ASessionPool.cs index afd83b6..8e0ca54 100644 --- a/src/Volcanic.AgenticCore.A2A.Server/Agent/A2ASessionPool.cs +++ b/src/Volcanic.AgenticCore.A2A.Server/Agent/A2ASessionPool.cs @@ -87,19 +87,28 @@ internal sealed class A2ASessionPool : IA2ALiveRuns, IAsyncDisposable /// /// Mark the task live (so restart recovery leaves it alone) until after the returned handle /// is disposed. Call before the task's first event is queued and dispose after its final status is written. + /// Throws UnsupportedOperation while another run of the same task is unfinished (two concurrent continuations). /// public IDisposable Track(string taskId, out LiveRun run) { foreach (var (id, old) in _live) if (IsExpired(old)) _live.TryRemove(new KeyValuePair(id, old)); var created = new LiveRun(); - // A continuation can start while the previous run of the same task is still in its grace period: the newest run - // replaces it. - _live[taskId] = created; + lock (_live) + { + if (_live.TryGetValue(taskId, out var current) && current.FinishedAt is null) + throw new A2AException($"Task '{taskId}' is already running.", A2AErrorCode.UnsupportedOperation); + // A continuation can start while the previous run of the same task is still in its grace period: the newest + // run replaces it. + _live[taskId] = created; + } run = created; return new Tracking(this, created); } + /// Whether a run of the task is in progress (tracked and not finished). + public bool IsRunning(string taskId) => _live.TryGetValue(taskId, out var run) && run.FinishedAt is null; + /// Take the context's session for one task. Throws UnsupportedOperation while another task runs in it. public async Task AcquireAsync(A2ACaller caller, string contextId, string taskId, CancellationToken cancellationToken) { diff --git a/src/Volcanic.AgenticCore.A2A.Server/Hosting/A2AServerCache.cs b/src/Volcanic.AgenticCore.A2A.Server/Hosting/A2AServerCache.cs index b05b14c..b9fae7a 100644 --- a/src/Volcanic.AgenticCore.A2A.Server/Hosting/A2AServerCache.cs +++ b/src/Volcanic.AgenticCore.A2A.Server/Hosting/A2AServerCache.cs @@ -23,7 +23,7 @@ internal sealed class A2AServerCache(IServiceProvider services) : IAsyncDisposab private VolcanicA2AServer Create(A2ACaller caller) { var store = ActivatorUtilities.CreateInstance(services, caller, (IA2ALiveRuns)_pool); - var handler = ActivatorUtilities.CreateInstance(services, caller); + var handler = ActivatorUtilities.CreateInstance(services, caller, store); return ActivatorUtilities.CreateInstance(services, caller, handler, store); } diff --git a/src/Volcanic.AgenticCore.A2A.Server/Hosting/VolcanicA2AServer.cs b/src/Volcanic.AgenticCore.A2A.Server/Hosting/VolcanicA2AServer.cs index d6488a4..11b05cb 100644 --- a/src/Volcanic.AgenticCore.A2A.Server/Hosting/VolcanicA2AServer.cs +++ b/src/Volcanic.AgenticCore.A2A.Server/Hosting/VolcanicA2AServer.cs @@ -44,7 +44,7 @@ internal sealed class VolcanicA2AServer( { if (message.ContextId is { Length: > 0 } given && given != task.ContextId) throw new A2AException("The task belongs to another context.", A2AErrorCode.InvalidParams); - if (task.Status.State != TaskState.InputRequired && !task.Status.State.IsTerminal()) + if ((task.Status.State != TaskState.InputRequired && !task.Status.State.IsTerminal()) || pool.IsRunning(task.Id)) throw new A2AException($"Task '{task.Id}' is not waiting for input.", A2AErrorCode.UnsupportedOperation); } var contextId = message.ContextId is { Length: > 0 } c ? c : task?.ContextId; diff --git a/tests/Volcanic.AgenticCore.A2A.Server.Tests/ServerPushTests.cs b/tests/Volcanic.AgenticCore.A2A.Server.Tests/ServerPushTests.cs index 58e985c..98211df 100644 --- a/tests/Volcanic.AgenticCore.A2A.Server.Tests/ServerPushTests.cs +++ b/tests/Volcanic.AgenticCore.A2A.Server.Tests/ServerPushTests.cs @@ -21,10 +21,9 @@ public class ServerPushTests Config = new PushNotificationConfig { Url = "https://hooks.example/cb", Token = "tok-1" }, }, CancellationToken.None); await client.SendMessageAsync(Requests.Text("now", contextId: task.ContextId, taskId: task.Id), CancellationToken.None); - var deadline = DateTime.UtcNow.AddSeconds(5); - while (!host.Push.Requests.Any(r => r.Body.Contains("TASK_STATE_COMPLETED") || r.Body.Contains("completed", StringComparison.OrdinalIgnoreCase)) && DateTime.UtcNow < deadline) - await Task.Delay(20); - var push = host.Push.Requests.Last(); + // The completed push: StreamResponse { task } with the state's proto JSON name. + await Requests.WaitUntilAsync(() => host.Push.Requests.Any(r => r.Body.Contains("\"TASK_STATE_COMPLETED\"")), "the completed push"); + var push = host.Push.Requests.First(r => r.Body.Contains("\"TASK_STATE_COMPLETED\"")); Assert.Equal("https://hooks.example/cb", push.Url.ToString()); Assert.Equal("tok-1", push.Token); Assert.Contains(task.Id, push.Body); diff --git a/tests/Volcanic.AgenticCore.A2A.Server.Tests/ServerTaskTests.cs b/tests/Volcanic.AgenticCore.A2A.Server.Tests/ServerTaskTests.cs index 223ee13..e038e8b 100644 --- a/tests/Volcanic.AgenticCore.A2A.Server.Tests/ServerTaskTests.cs +++ b/tests/Volcanic.AgenticCore.A2A.Server.Tests/ServerTaskTests.cs @@ -77,6 +77,73 @@ public class ServerTaskTests Assert.Equal(first.Id, second.Id); Assert.Equal(TaskState.Completed, second.Status.State); Assert.Equal("Booked for Friday", Requests.ResponseText(second)); + var stored = await client.GetTaskAsync(new GetTaskRequest { Id = first.Id }, CancellationToken.None); + var history = stored.History!.Select(m => string.Concat(m.Parts.Select(p => p.Text))).ToList(); + Assert.Single(history, t => t == "Friday"); + var question = history.IndexOf("Which date?"); + Assert.True(question >= 0 && question < history.IndexOf("Friday"), string.Join(" | ", history)); + } + + [Fact] + public async Task RequestInput_StreamingContinuation_StartsWithWorkingTask() + { + await using var host = await A2ATestServer.StartAsync(); + host.Faux.Enqueue(FauxTurn.Tool("request_input", new { question = "Which date?" }), FauxTurn.Say("Booked for Friday")); + var client = host.Client(); + var first = (await client.SendMessageAsync(Requests.Text("book a table"), CancellationToken.None)).Task!; + var events = await CollectAsync(client.SendStreamingMessageAsync( + Requests.Text("Friday", contextId: first.ContextId, taskId: first.Id), CancellationToken.None)); + var resumed = events[0].Task; + Assert.NotNull(resumed); + Assert.Equal(first.Id, resumed.Id); + Assert.Equal(TaskState.Working, resumed.Status.State); + Assert.Equal(TaskState.Completed, events.Last(e => e.StatusUpdate is not null).StatusUpdate!.Status.State); + } + + [Fact] + public async Task RequestInput_FailedContinuation_DropsEarlierResponse() + { + await using var host = await A2ATestServer.StartAsync(); + host.Faux.Enqueue(FauxTurn.Say("Let me check.").WithTool("request_input", new { question = "Which date?" }), + FauxTurn.Error("model exploded")); + var client = host.Client(); + var first = (await client.SendMessageAsync(Requests.Text("book a table"), CancellationToken.None)).Task!; + Assert.Equal(TaskState.InputRequired, first.Status.State); + Assert.Equal("Let me check.", Requests.ResponseText(first)); + var second = (await client.SendMessageAsync(Requests.Text("Friday", contextId: first.ContextId, taskId: first.Id), CancellationToken.None)).Task!; + Assert.Equal(TaskState.Failed, second.Status.State); + Assert.DoesNotContain("Let me check.", Requests.ResponseText(second)); + } + + [Fact] + public async Task RequestInput_ContinuationAnswer_ReplacesEarlierResponse() + { + await using var host = await A2ATestServer.StartAsync(); + host.Faux.Enqueue(FauxTurn.Say("Let me check.").WithTool("request_input", new { question = "Which date?" }), + FauxTurn.Say("Booked for Friday")); + var client = host.Client(); + var first = (await client.SendMessageAsync(Requests.Text("book a table"), CancellationToken.None)).Task!; + var events = await CollectAsync(client.SendStreamingMessageAsync( + Requests.Text("Friday", contextId: first.ContextId, taskId: first.Id), CancellationToken.None)); + var chunks = events.Where(e => e.ArtifactUpdate is { Artifact.ArtifactId: "response" }).Select(e => e.ArtifactUpdate!).ToList(); + Assert.False(chunks[0].Append); // the run's first chunk starts a fresh answer + var stored = await client.GetTaskAsync(new GetTaskRequest { Id = first.Id }, CancellationToken.None); + Assert.Equal("Booked for Friday", Requests.ResponseText(stored)); + } + + [Fact] + public async Task RequestInput_SecondContinuationWhileRunning_Rejected() + { + await using var host = await A2ATestServer.StartAsync(); + host.Faux.Enqueue(FauxTurn.Tool("request_input", new { question = "Which date?" }), FauxTurn.Say("slow").WithDelay(TimeSpan.FromSeconds(5))); + var client = host.Client(); + var first = (await client.SendMessageAsync(Requests.Text("book a table"), CancellationToken.None)).Task!; + await client.SendMessageAsync(Requests.Text("Friday", contextId: first.ContextId, taskId: first.Id, returnImmediately: true), CancellationToken.None); + await Requests.WaitUntilAsync(() => host.Faux.Requests.Count == 2, "the continuation's model call"); + var ex = await Assert.ThrowsAsync(() => + client.SendMessageAsync(Requests.Text("Saturday", contextId: first.ContextId, taskId: first.Id), CancellationToken.None)); + Assert.Equal(A2AErrorCode.UnsupportedOperation, ex.ErrorCode); + await client.CancelTaskAsync(new CancelTaskRequest { Id = first.Id }, CancellationToken.None); } [Fact] @@ -88,6 +155,8 @@ public class ServerTaskTests var first = (await client.SendMessageAsync(Requests.Text("one"), CancellationToken.None)).Task!; var slow = (await client.SendMessageAsync(Requests.Text("two", contextId: first.ContextId, returnImmediately: true), CancellationToken.None)).Task!; await Requests.WaitForStateAsync(client, slow.Id, s => s == TaskState.Working); + // Working is written before the run holds the context: wait until its model call has started. + await Requests.WaitUntilAsync(() => host.Faux.Requests.Count == 2, "the second model call"); var ex = await Assert.ThrowsAsync(() => client.SendMessageAsync(Requests.Text("three", contextId: first.ContextId), CancellationToken.None)); Assert.Equal(A2AErrorCode.UnsupportedOperation, ex.ErrorCode); @@ -111,7 +180,7 @@ public class ServerTaskTests public async Task Subscribe_LiveTask_GetsEvents() { await using var host = await A2ATestServer.StartAsync(); - host.Faux.Enqueue(FauxTurn.Say("eventually").WithDelay(TimeSpan.FromMilliseconds(500))); + host.Faux.Enqueue(FauxTurn.Say("eventually").WithDelay(TimeSpan.FromSeconds(2))); var client = host.Client(); var task = (await client.SendMessageAsync(Requests.Text("go", returnImmediately: true), CancellationToken.None)).Task!; var events = await CollectAsync(client.SubscribeToTaskAsync(new SubscribeToTaskRequest { Id = task.Id }, CancellationToken.None)); diff --git a/tests/Volcanic.AgenticCore.A2A.Server.Tests/SessionPoolTests.cs b/tests/Volcanic.AgenticCore.A2A.Server.Tests/SessionPoolTests.cs index f587b6d..bc6d042 100644 --- a/tests/Volcanic.AgenticCore.A2A.Server.Tests/SessionPoolTests.cs +++ b/tests/Volcanic.AgenticCore.A2A.Server.Tests/SessionPoolTests.cs @@ -5,6 +5,8 @@ using Microsoft.Extensions.Options; using Microsoft.Extensions.Time.Testing; using Volcanic.AgenticCore.A2A.Server.Agent; using Volcanic.AgenticCore.A2A.Server.Callers; +using Volcanic.AgenticCore.A2A.Server.Push; +using Volcanic.AgenticCore.A2A.Server.Tasks; using Volcanic.AgenticCore.Features.Agent; using Volcanic.AgenticCore.Features.Extensions; using Volcanic.AgenticCore.Features.Processes; @@ -48,7 +50,8 @@ public sealed class SessionPoolTests : IAsyncDisposable } private A2AAgentHandler Handler() => - new(_caller, _pool, Options.Create(_options), SecretRedactor.Empty, new A2AFileDownloader(new HttpClient()), TimeProvider.System, + new(_caller, new WorkspaceTaskStore(_caller, _pool, A2APushSender.Disabled, _time, NullLogger.Instance), + _pool, Options.Create(_options), SecretRedactor.Empty, new A2AFileDownloader(new HttpClient()), TimeProvider.System, NullLogger.Instance); private static RequestContext Context(string text, string taskId = "t1", string contextId = "c1") => new() @@ -82,6 +85,15 @@ public sealed class SessionPoolTests : IAsyncDisposable Assert.Null(run.FinishedAt); } + [Fact] + public void Pool_Track_UnfinishedRunOfSameTask_Refused() + { + using var first = _pool.Track("t1", out _); + var ex = Assert.Throws(() => _pool.Track("t1", out _)); + Assert.Equal(A2AErrorCode.UnsupportedOperation, ex.ErrorCode); + Assert.True(_pool.IsRunning("t1")); + } + [Fact] public async Task Pool_AbortAsync_FinishedRun_IsNoOp() { diff --git a/tests/Volcanic.AgenticCore.A2A.Server.Tests/Support/A2ATestServer.cs b/tests/Volcanic.AgenticCore.A2A.Server.Tests/Support/A2ATestServer.cs index 7634327..67cffd4 100644 --- a/tests/Volcanic.AgenticCore.A2A.Server.Tests/Support/A2ATestServer.cs +++ b/tests/Volcanic.AgenticCore.A2A.Server.Tests/Support/A2ATestServer.cs @@ -33,36 +33,44 @@ public sealed class A2ATestServer : IAsyncDisposable public static async Task StartAsync(Action? configure = null, bool withNotifier = true) { var host = new A2ATestServer(); - Directory.CreateDirectory(host.Workspace); - Directory.CreateDirectory(host.OtherWorkspace); - host.WriteCallers(host.Workspace, $$"""[ { "name": "alpha", "key": "{{Key}}" }, { "name": "gamma", "key": "key-gamma" } ]"""); - host.WriteCallers(host.OtherWorkspace, $$"""[ { "name": "beta", "key": "{{OtherKey}}" } ]"""); + try + { + Directory.CreateDirectory(host.Workspace); + Directory.CreateDirectory(host.OtherWorkspace); + host.WriteCallers(host.Workspace, $$"""[ { "name": "alpha", "key": "{{Key}}" }, { "name": "gamma", "key": "key-gamma" } ]"""); + host.WriteCallers(host.OtherWorkspace, $$"""[ { "name": "beta", "key": "{{OtherKey}}" } ]"""); - var builder = WebApplication.CreateBuilder(); - builder.WebHost.UseTestServer(); - builder.Services.AddAgenticCore(o => + var builder = WebApplication.CreateBuilder(); + builder.WebHost.UseTestServer(); + builder.Services.AddAgenticCore(o => + { + o.AgentDir = Path.Combine(host._root, "agent"); + o.DiscoverExtensionAssemblies = false; + o.DefaultModel = FauxProvider.ModelKey; + o.Extensions.Add(ExtensionRegistration.FromInstance(host.Faux.Extension)); + o.ProcessEnvironment.HardenHostProcess = false; + o.Retry = new RetrySettings { Enabled = false }; // FauxTurn.Error fails the task at once + }); + builder.Services.AddSingleton(sp => new HttpA2APushSender(new HttpClient(host.Push), TimeProvider.System, NullLogger.Instance)); + if (withNotifier) builder.Services.AddSingleton(host.Notifier); + builder.Services.AddSingleton(new WorkspaceCallerDirectory( + () => [new A2AWorkspace(host.Workspace, "test-ws1", "ws1"), new A2AWorkspace(host.OtherWorkspace, "test-ws2", "ws2")], + NullLogger.Instance)); + builder.Services.AddA2AServer(o => + { + o.AllowPrivatePushUrls = true; + configure?.Invoke(o); + }); + host.App = builder.Build(); + host.App.MapA2AServer(); + await host.App.StartAsync(); + return host; + } + catch { - o.AgentDir = Path.Combine(host._root, "agent"); - o.DiscoverExtensionAssemblies = false; - o.DefaultModel = FauxProvider.ModelKey; - o.Extensions.Add(ExtensionRegistration.FromInstance(host.Faux.Extension)); - o.ProcessEnvironment.HardenHostProcess = false; - o.Retry = new RetrySettings { Enabled = false }; // FauxTurn.Error fails the task at once - }); - builder.Services.AddSingleton(sp => new HttpA2APushSender(new HttpClient(host.Push), TimeProvider.System, NullLogger.Instance)); - if (withNotifier) builder.Services.AddSingleton(host.Notifier); - builder.Services.AddSingleton(new WorkspaceCallerDirectory( - () => [new A2AWorkspace(host.Workspace, "test-ws1", "ws1"), new A2AWorkspace(host.OtherWorkspace, "test-ws2", "ws2")], - NullLogger.Instance)); - builder.Services.AddA2AServer(o => - { - o.AllowPrivatePushUrls = true; - configure?.Invoke(o); - }); - host.App = builder.Build(); - host.App.MapA2AServer(); - await host.App.StartAsync(); - return host; + await host.DisposeAsync(); + throw; + } } public void WriteCallers(string workspace, string json) @@ -82,9 +90,12 @@ public sealed class A2ATestServer : IAsyncDisposable public async ValueTask DisposeAsync() { - await App.StopAsync(); - await App.DisposeAsync(); - try { Directory.Delete(_root, true); } catch (IOException) { } + if (App is not null) + { + await App.StopAsync(); + await App.DisposeAsync(); + } + try { Directory.Delete(_root, true); } catch (Exception e) when (e is IOException or UnauthorizedAccessException) { } } public sealed class RecordingNotifier : IA2AOwnerNotifier diff --git a/tests/Volcanic.AgenticCore.A2A.Server.Tests/Support/Requests.cs b/tests/Volcanic.AgenticCore.A2A.Server.Tests/Support/Requests.cs index 71353a8..2069979 100644 --- a/tests/Volcanic.AgenticCore.A2A.Server.Tests/Support/Requests.cs +++ b/tests/Volcanic.AgenticCore.A2A.Server.Tests/Support/Requests.cs @@ -26,7 +26,20 @@ public static class Requests while (true) { var task = await client.GetTaskAsync(new GetTaskRequest { Id = taskId }, CancellationToken.None); - if (done(task.Status.State) || DateTime.UtcNow > deadline) return task; + if (done(task.Status.State)) return task; + if (DateTime.UtcNow > deadline) + throw new TimeoutException($"Task {taskId} did not reach the expected state in {timeoutMs} ms; last state {task.Status.State}."); + await Task.Delay(20); + } + } + + /// Poll until holds; throws TimeoutException naming . + public static async Task WaitUntilAsync(Func condition, string what, int timeoutMs = 10_000) + { + var deadline = DateTime.UtcNow.AddMilliseconds(timeoutMs); + while (!condition()) + { + if (DateTime.UtcNow > deadline) throw new TimeoutException($"Timed out after {timeoutMs} ms waiting for {what}."); await Task.Delay(20); } }