From 9fcfbf95e5bebe211067dd8cbe2daea6ece3e83e Mon Sep 17 00:00:00 2001 From: Mark Licata Date: Wed, 7 Oct 2026 08:33:32 -0700 Subject: [PATCH 1/3] .NET: add AsyncRequestBroker Co-Authored-By: Claude Opus 5.5 --- .../Harness/AsyncRequestBroker.cs | 142 ++++++++++++++++++ .../PublicAPI/net10.0/PublicAPI.Unshipped.txt | 6 + .../PublicAPI/net472/PublicAPI.Unshipped.txt | 6 + .../PublicAPI/net8.0/PublicAPI.Unshipped.txt | 6 + .../PublicAPI/net9.0/PublicAPI.Unshipped.txt | 6 + .../netstandard2.0/PublicAPI.Unshipped.txt | 6 + 6 files changed, 172 insertions(+) create mode 100644 dotnet/src/Microsoft.Agents.AI/Harness/AsyncRequestBroker.cs diff --git a/dotnet/src/Microsoft.Agents.AI/Harness/AsyncRequestBroker.cs b/dotnet/src/Microsoft.Agents.AI/Harness/AsyncRequestBroker.cs new file mode 100644 index 00000000000..211de77f9b8 --- /dev/null +++ b/dotnet/src/Microsoft.Agents.AI/Harness/AsyncRequestBroker.cs @@ -0,0 +1,142 @@ +// Copyright (c) Microsoft. All rights reserved. + +using System; +using System.Collections.Concurrent; +using System.Collections.Generic; +using System.Diagnostics.CodeAnalysis; +using System.Threading; +using System.Threading.Tasks; +using Microsoft.Shared.DiagnosticIds; + +namespace Microsoft.Agents.AI; + +/// +/// A generic async pause-and-wait broker for request/response patterns. +/// +/// +/// Allows server-side code to suspend on a request ID and await a client response +/// delivered via or , without polling or +/// blocking a thread. +/// +/// The type of the result delivered to the waiter. +[Experimental(DiagnosticIds.Experiments.AgentsAIExperiments)] +public sealed class AsyncRequestBroker +{ + private const int DefaultSettledQueueSize = 256; + + private readonly ConcurrentDictionary> _pending = new(); + private readonly Queue _settledQueue = new(); + private readonly HashSet _settledSet = new(); + private readonly int _settledQueueSize; + private readonly object _settledLock = new(); + + /// + /// Initializes a new instance of the class. + /// + /// Maximum number of recently settled request IDs to track for late-response handling. + public AsyncRequestBroker(int settledQueueSize = DefaultSettledQueueSize) + { + if (settledQueueSize < 1) + { + throw new ArgumentOutOfRangeException(nameof(settledQueueSize), "Settled queue size must be at least 1."); + } + + this._settledQueueSize = settledQueueSize; + } + + /// Gets the number of requests currently awaiting a response. + public int PendingCount => this._pending.Count; + + /// Suspend until the request is resolved, rejected, or cancelled. + /// Unique identifier for this request. + /// Optional timeout. waits indefinitely. + /// Token to cancel the wait. + /// The result passed to . + public async Task RequestAsync(string requestId, TimeSpan? timeout = null, CancellationToken cancellationToken = default) + { + var tcs = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + + lock (this._settledLock) + { + if (this._settledSet.Contains(requestId)) + { + throw new InvalidOperationException($"Request '{requestId}' was already settled."); + } + + if (!this._pending.TryAdd(requestId, tcs)) + { + throw new InvalidOperationException($"Request '{requestId}' is already pending."); + } + } + + using var cts = timeout.HasValue + ? CancellationTokenSource.CreateLinkedTokenSource(cancellationToken) + : null; + + if (cts is not null && timeout.HasValue) + { + cts.CancelAfter(timeout.Value); + } + + var effectiveToken = cts?.Token ?? cancellationToken; + using var registration = effectiveToken.Register(() => + { + // Atomic key+value removal: only cancel our own TCS, never a replacement registered under the same ID. + if (((ICollection>>)this._pending) + .Remove(new KeyValuePair>(requestId, tcs))) + { + this.TrackSettled(requestId); + tcs.TrySetCanceled(effectiveToken); + } + }); + + return await tcs.Task.ConfigureAwait(false); + } + + /// Deliver a successful result to a waiting call. + /// The request identifier. + /// The result to deliver. + /// if a waiter was notified; if the request was not found. + public bool Resolve(string requestId, TResult result) + { + if (!this._pending.TryRemove(requestId, out var tcs)) + { + return false; + } + + this.TrackSettled(requestId); + tcs.TrySetResult(result); + return true; + } + + /// Deliver an exception to a waiting call. + /// The request identifier. + /// The exception to deliver. + /// if a waiter was notified. + public bool Reject(string requestId, Exception exception) + { + if (!this._pending.TryRemove(requestId, out var tcs)) + { + return false; + } + + this.TrackSettled(requestId); + tcs.TrySetException(exception); + return true; + } + + private void TrackSettled(string requestId) + { + lock (this._settledLock) + { + if (this._settledQueue.Count >= this._settledQueueSize) + { + string oldest = this._settledQueue.Dequeue(); + this._settledSet.Remove(oldest); + } + + this._settledQueue.Enqueue(requestId); + this._settledSet.Add(requestId); + } + } +} diff --git a/dotnet/src/Microsoft.Agents.AI/PublicAPI/net10.0/PublicAPI.Unshipped.txt b/dotnet/src/Microsoft.Agents.AI/PublicAPI/net10.0/PublicAPI.Unshipped.txt index fada975c025..d37623ec1b5 100644 --- a/dotnet/src/Microsoft.Agents.AI/PublicAPI/net10.0/PublicAPI.Unshipped.txt +++ b/dotnet/src/Microsoft.Agents.AI/PublicAPI/net10.0/PublicAPI.Unshipped.txt @@ -24,3 +24,9 @@ Microsoft.Agents.AI.AgentModeProviderOptions.DisableModeSetTool.set -> void [MAAI001]virtual Microsoft.Agents.AI.AgentFileStore.SearchAsync(string! directory, string! regexPattern, string? globPattern = null, bool recursive = false, System.Threading.CancellationToken cancellationToken = default(System.Threading.CancellationToken)) -> System.Threading.Tasks.Task!>! [MAAI001]Microsoft.Agents.AI.FunctionInvocationContextExtensions [MAAI001]static Microsoft.Agents.AI.FunctionInvocationContextExtensions.WrapWithPendingMiddleware(this Microsoft.Extensions.AI.FunctionInvocationContext! context, Microsoft.Extensions.AI.AIFunction! function) -> Microsoft.Extensions.AI.AIFunction! +[MAAI001]Microsoft.Agents.AI.AsyncRequestBroker +[MAAI001]Microsoft.Agents.AI.AsyncRequestBroker.AsyncRequestBroker(int settledQueueSize = 256) -> void +[MAAI001]Microsoft.Agents.AI.AsyncRequestBroker.PendingCount.get -> int +[MAAI001]Microsoft.Agents.AI.AsyncRequestBroker.Reject(string! requestId, System.Exception! exception) -> bool +[MAAI001]Microsoft.Agents.AI.AsyncRequestBroker.RequestAsync(string! requestId, System.TimeSpan? timeout = null, System.Threading.CancellationToken cancellationToken = default(System.Threading.CancellationToken)) -> System.Threading.Tasks.Task! +[MAAI001]Microsoft.Agents.AI.AsyncRequestBroker.Resolve(string! requestId, TResult result) -> bool diff --git a/dotnet/src/Microsoft.Agents.AI/PublicAPI/net472/PublicAPI.Unshipped.txt b/dotnet/src/Microsoft.Agents.AI/PublicAPI/net472/PublicAPI.Unshipped.txt index fada975c025..d37623ec1b5 100644 --- a/dotnet/src/Microsoft.Agents.AI/PublicAPI/net472/PublicAPI.Unshipped.txt +++ b/dotnet/src/Microsoft.Agents.AI/PublicAPI/net472/PublicAPI.Unshipped.txt @@ -24,3 +24,9 @@ Microsoft.Agents.AI.AgentModeProviderOptions.DisableModeSetTool.set -> void [MAAI001]virtual Microsoft.Agents.AI.AgentFileStore.SearchAsync(string! directory, string! regexPattern, string? globPattern = null, bool recursive = false, System.Threading.CancellationToken cancellationToken = default(System.Threading.CancellationToken)) -> System.Threading.Tasks.Task!>! [MAAI001]Microsoft.Agents.AI.FunctionInvocationContextExtensions [MAAI001]static Microsoft.Agents.AI.FunctionInvocationContextExtensions.WrapWithPendingMiddleware(this Microsoft.Extensions.AI.FunctionInvocationContext! context, Microsoft.Extensions.AI.AIFunction! function) -> Microsoft.Extensions.AI.AIFunction! +[MAAI001]Microsoft.Agents.AI.AsyncRequestBroker +[MAAI001]Microsoft.Agents.AI.AsyncRequestBroker.AsyncRequestBroker(int settledQueueSize = 256) -> void +[MAAI001]Microsoft.Agents.AI.AsyncRequestBroker.PendingCount.get -> int +[MAAI001]Microsoft.Agents.AI.AsyncRequestBroker.Reject(string! requestId, System.Exception! exception) -> bool +[MAAI001]Microsoft.Agents.AI.AsyncRequestBroker.RequestAsync(string! requestId, System.TimeSpan? timeout = null, System.Threading.CancellationToken cancellationToken = default(System.Threading.CancellationToken)) -> System.Threading.Tasks.Task! +[MAAI001]Microsoft.Agents.AI.AsyncRequestBroker.Resolve(string! requestId, TResult result) -> bool diff --git a/dotnet/src/Microsoft.Agents.AI/PublicAPI/net8.0/PublicAPI.Unshipped.txt b/dotnet/src/Microsoft.Agents.AI/PublicAPI/net8.0/PublicAPI.Unshipped.txt index fada975c025..d37623ec1b5 100644 --- a/dotnet/src/Microsoft.Agents.AI/PublicAPI/net8.0/PublicAPI.Unshipped.txt +++ b/dotnet/src/Microsoft.Agents.AI/PublicAPI/net8.0/PublicAPI.Unshipped.txt @@ -24,3 +24,9 @@ Microsoft.Agents.AI.AgentModeProviderOptions.DisableModeSetTool.set -> void [MAAI001]virtual Microsoft.Agents.AI.AgentFileStore.SearchAsync(string! directory, string! regexPattern, string? globPattern = null, bool recursive = false, System.Threading.CancellationToken cancellationToken = default(System.Threading.CancellationToken)) -> System.Threading.Tasks.Task!>! [MAAI001]Microsoft.Agents.AI.FunctionInvocationContextExtensions [MAAI001]static Microsoft.Agents.AI.FunctionInvocationContextExtensions.WrapWithPendingMiddleware(this Microsoft.Extensions.AI.FunctionInvocationContext! context, Microsoft.Extensions.AI.AIFunction! function) -> Microsoft.Extensions.AI.AIFunction! +[MAAI001]Microsoft.Agents.AI.AsyncRequestBroker +[MAAI001]Microsoft.Agents.AI.AsyncRequestBroker.AsyncRequestBroker(int settledQueueSize = 256) -> void +[MAAI001]Microsoft.Agents.AI.AsyncRequestBroker.PendingCount.get -> int +[MAAI001]Microsoft.Agents.AI.AsyncRequestBroker.Reject(string! requestId, System.Exception! exception) -> bool +[MAAI001]Microsoft.Agents.AI.AsyncRequestBroker.RequestAsync(string! requestId, System.TimeSpan? timeout = null, System.Threading.CancellationToken cancellationToken = default(System.Threading.CancellationToken)) -> System.Threading.Tasks.Task! +[MAAI001]Microsoft.Agents.AI.AsyncRequestBroker.Resolve(string! requestId, TResult result) -> bool diff --git a/dotnet/src/Microsoft.Agents.AI/PublicAPI/net9.0/PublicAPI.Unshipped.txt b/dotnet/src/Microsoft.Agents.AI/PublicAPI/net9.0/PublicAPI.Unshipped.txt index fada975c025..d37623ec1b5 100644 --- a/dotnet/src/Microsoft.Agents.AI/PublicAPI/net9.0/PublicAPI.Unshipped.txt +++ b/dotnet/src/Microsoft.Agents.AI/PublicAPI/net9.0/PublicAPI.Unshipped.txt @@ -24,3 +24,9 @@ Microsoft.Agents.AI.AgentModeProviderOptions.DisableModeSetTool.set -> void [MAAI001]virtual Microsoft.Agents.AI.AgentFileStore.SearchAsync(string! directory, string! regexPattern, string? globPattern = null, bool recursive = false, System.Threading.CancellationToken cancellationToken = default(System.Threading.CancellationToken)) -> System.Threading.Tasks.Task!>! [MAAI001]Microsoft.Agents.AI.FunctionInvocationContextExtensions [MAAI001]static Microsoft.Agents.AI.FunctionInvocationContextExtensions.WrapWithPendingMiddleware(this Microsoft.Extensions.AI.FunctionInvocationContext! context, Microsoft.Extensions.AI.AIFunction! function) -> Microsoft.Extensions.AI.AIFunction! +[MAAI001]Microsoft.Agents.AI.AsyncRequestBroker +[MAAI001]Microsoft.Agents.AI.AsyncRequestBroker.AsyncRequestBroker(int settledQueueSize = 256) -> void +[MAAI001]Microsoft.Agents.AI.AsyncRequestBroker.PendingCount.get -> int +[MAAI001]Microsoft.Agents.AI.AsyncRequestBroker.Reject(string! requestId, System.Exception! exception) -> bool +[MAAI001]Microsoft.Agents.AI.AsyncRequestBroker.RequestAsync(string! requestId, System.TimeSpan? timeout = null, System.Threading.CancellationToken cancellationToken = default(System.Threading.CancellationToken)) -> System.Threading.Tasks.Task! +[MAAI001]Microsoft.Agents.AI.AsyncRequestBroker.Resolve(string! requestId, TResult result) -> bool diff --git a/dotnet/src/Microsoft.Agents.AI/PublicAPI/netstandard2.0/PublicAPI.Unshipped.txt b/dotnet/src/Microsoft.Agents.AI/PublicAPI/netstandard2.0/PublicAPI.Unshipped.txt index fada975c025..d37623ec1b5 100644 --- a/dotnet/src/Microsoft.Agents.AI/PublicAPI/netstandard2.0/PublicAPI.Unshipped.txt +++ b/dotnet/src/Microsoft.Agents.AI/PublicAPI/netstandard2.0/PublicAPI.Unshipped.txt @@ -24,3 +24,9 @@ Microsoft.Agents.AI.AgentModeProviderOptions.DisableModeSetTool.set -> void [MAAI001]virtual Microsoft.Agents.AI.AgentFileStore.SearchAsync(string! directory, string! regexPattern, string? globPattern = null, bool recursive = false, System.Threading.CancellationToken cancellationToken = default(System.Threading.CancellationToken)) -> System.Threading.Tasks.Task!>! [MAAI001]Microsoft.Agents.AI.FunctionInvocationContextExtensions [MAAI001]static Microsoft.Agents.AI.FunctionInvocationContextExtensions.WrapWithPendingMiddleware(this Microsoft.Extensions.AI.FunctionInvocationContext! context, Microsoft.Extensions.AI.AIFunction! function) -> Microsoft.Extensions.AI.AIFunction! +[MAAI001]Microsoft.Agents.AI.AsyncRequestBroker +[MAAI001]Microsoft.Agents.AI.AsyncRequestBroker.AsyncRequestBroker(int settledQueueSize = 256) -> void +[MAAI001]Microsoft.Agents.AI.AsyncRequestBroker.PendingCount.get -> int +[MAAI001]Microsoft.Agents.AI.AsyncRequestBroker.Reject(string! requestId, System.Exception! exception) -> bool +[MAAI001]Microsoft.Agents.AI.AsyncRequestBroker.RequestAsync(string! requestId, System.TimeSpan? timeout = null, System.Threading.CancellationToken cancellationToken = default(System.Threading.CancellationToken)) -> System.Threading.Tasks.Task! +[MAAI001]Microsoft.Agents.AI.AsyncRequestBroker.Resolve(string! requestId, TResult result) -> bool From 0b3a20a7c6dcb8d4de0b21907d7d53d5bfd5b23f Mon Sep 17 00:00:00 2001 From: Mark Licata Date: Wed, 7 Oct 2026 09:04:34 -0700 Subject: [PATCH 2/3] Address review feedback: atomic broker state transitions, timeout validation, unit tests Co-Authored-By: Claude Opus 5.5 --- .../Harness/AsyncRequestBroker.cs | 117 ++++-- .../Harness/AsyncRequestBrokerTests.cs | 378 ++++++++++++++++++ 2 files changed, 465 insertions(+), 30 deletions(-) create mode 100644 dotnet/tests/Microsoft.Agents.AI.UnitTests/Harness/AsyncRequestBrokerTests.cs diff --git a/dotnet/src/Microsoft.Agents.AI/Harness/AsyncRequestBroker.cs b/dotnet/src/Microsoft.Agents.AI/Harness/AsyncRequestBroker.cs index 211de77f9b8..d06d7de5cc6 100644 --- a/dotnet/src/Microsoft.Agents.AI/Harness/AsyncRequestBroker.cs +++ b/dotnet/src/Microsoft.Agents.AI/Harness/AsyncRequestBroker.cs @@ -1,12 +1,12 @@ // Copyright (c) Microsoft. All rights reserved. using System; -using System.Collections.Concurrent; using System.Collections.Generic; using System.Diagnostics.CodeAnalysis; using System.Threading; using System.Threading.Tasks; using Microsoft.Shared.DiagnosticIds; +using Microsoft.Shared.Diagnostics; namespace Microsoft.Agents.AI; @@ -24,11 +24,14 @@ public sealed class AsyncRequestBroker { private const int DefaultSettledQueueSize = 256; - private readonly ConcurrentDictionary> _pending = new(); + private readonly Dictionary> _pending = new(StringComparer.Ordinal); private readonly Queue _settledQueue = new(); - private readonly HashSet _settledSet = new(); + private readonly HashSet _settledSet = new(StringComparer.Ordinal); private readonly int _settledQueueSize; - private readonly object _settledLock = new(); + + // Guards _pending, _settledQueue, and _settledSet so that registration, removal, and + // settled-tracking are a single atomic transition. + private readonly object _lock = new(); /// /// Initializes a new instance of the class. @@ -45,47 +48,80 @@ public AsyncRequestBroker(int settledQueueSize = DefaultSettledQueueSize) } /// Gets the number of requests currently awaiting a response. - public int PendingCount => this._pending.Count; + public int PendingCount + { + get + { + lock (this._lock) + { + return this._pending.Count; + } + } + } /// Suspend until the request is resolved, rejected, or cancelled. /// Unique identifier for this request. - /// Optional timeout. waits indefinitely. + /// + /// Optional timeout. or waits indefinitely. + /// Otherwise must be non-negative and no greater than milliseconds. + /// /// Token to cancel the wait. /// The result passed to . + /// is . + /// is outside the supported range. + /// The request ID is already pending or was recently settled. + /// The timeout elapsed or was cancelled. public async Task RequestAsync(string requestId, TimeSpan? timeout = null, CancellationToken cancellationToken = default) { + _ = Throw.IfNull(requestId); + + bool hasTimeout = timeout.HasValue && timeout.Value != Timeout.InfiniteTimeSpan; + if (hasTimeout && (timeout!.Value < TimeSpan.Zero || timeout.Value.TotalMilliseconds > int.MaxValue)) + { + throw new ArgumentOutOfRangeException(nameof(timeout), timeout, "Timeout must be non-negative and no greater than Int32.MaxValue milliseconds, or Timeout.InfiniteTimeSpan."); + } + var tcs = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); - lock (this._settledLock) + lock (this._lock) { if (this._settledSet.Contains(requestId)) { throw new InvalidOperationException($"Request '{requestId}' was already settled."); } - if (!this._pending.TryAdd(requestId, tcs)) + if (this._pending.ContainsKey(requestId)) { throw new InvalidOperationException($"Request '{requestId}' is already pending."); } + + this._pending.Add(requestId, tcs); } - using var cts = timeout.HasValue + using var cts = hasTimeout ? CancellationTokenSource.CreateLinkedTokenSource(cancellationToken) : null; - if (cts is not null && timeout.HasValue) - { - cts.CancelAfter(timeout.Value); - } + // The timeout was validated above, so CancelAfter cannot throw here. + cts?.CancelAfter(timeout!.Value); var effectiveToken = cts?.Token ?? cancellationToken; using var registration = effectiveToken.Register(() => { - // Atomic key+value removal: only cancel our own TCS, never a replacement registered under the same ID. - if (((ICollection>>)this._pending) - .Remove(new KeyValuePair>(requestId, tcs))) + bool removed; + lock (this._lock) + { + // Only cancel our own TCS, never a replacement registered under the same ID. + removed = this._pending.TryGetValue(requestId, out var current) && ReferenceEquals(current, tcs); + if (removed) + { + this._pending.Remove(requestId); + this.TrackSettledLocked(requestId); + } + } + + if (removed) { - this.TrackSettled(requestId); tcs.TrySetCanceled(effectiveToken); } }); @@ -96,15 +132,17 @@ public async Task RequestAsync(string requestId, TimeSpan? timeout = nu /// Deliver a successful result to a waiting call. /// The request identifier. /// The result to deliver. - /// if a waiter was notified; if the request was not found. + /// + /// if a waiter was notified; if the request was not found + /// (already settled, timed out, cancelled, or never registered). + /// public bool Resolve(string requestId, TResult result) { - if (!this._pending.TryRemove(requestId, out var tcs)) + if (!this.TryRemovePending(requestId, out var tcs)) { return false; } - this.TrackSettled(requestId); tcs.TrySetResult(result); return true; } @@ -112,31 +150,50 @@ public bool Resolve(string requestId, TResult result) /// Deliver an exception to a waiting call. /// The request identifier. /// The exception to deliver. - /// if a waiter was notified. + /// + /// if a waiter was notified; if the request was not found + /// (already settled, timed out, cancelled, or never registered). + /// public bool Reject(string requestId, Exception exception) { - if (!this._pending.TryRemove(requestId, out var tcs)) + _ = Throw.IfNull(exception); + + if (!this.TryRemovePending(requestId, out var tcs)) { return false; } - this.TrackSettled(requestId); tcs.TrySetException(exception); return true; } - private void TrackSettled(string requestId) + private bool TryRemovePending(string requestId, [NotNullWhen(true)] out TaskCompletionSource? tcs) { - lock (this._settledLock) + _ = Throw.IfNull(requestId); + + lock (this._lock) { - if (this._settledQueue.Count >= this._settledQueueSize) + if (!this._pending.TryGetValue(requestId, out tcs)) { - string oldest = this._settledQueue.Dequeue(); - this._settledSet.Remove(oldest); + return false; } - this._settledQueue.Enqueue(requestId); - this._settledSet.Add(requestId); + this._pending.Remove(requestId); + this.TrackSettledLocked(requestId); + return true; } } + + // Caller must hold _lock. + private void TrackSettledLocked(string requestId) + { + if (this._settledQueue.Count >= this._settledQueueSize) + { + string oldest = this._settledQueue.Dequeue(); + this._settledSet.Remove(oldest); + } + + this._settledQueue.Enqueue(requestId); + this._settledSet.Add(requestId); + } } diff --git a/dotnet/tests/Microsoft.Agents.AI.UnitTests/Harness/AsyncRequestBrokerTests.cs b/dotnet/tests/Microsoft.Agents.AI.UnitTests/Harness/AsyncRequestBrokerTests.cs new file mode 100644 index 00000000000..ece1eb9a23e --- /dev/null +++ b/dotnet/tests/Microsoft.Agents.AI.UnitTests/Harness/AsyncRequestBrokerTests.cs @@ -0,0 +1,378 @@ +// Copyright (c) Microsoft. All rights reserved. + +using System; +using System.Linq; +using System.Threading; +using System.Threading.Tasks; + +namespace Microsoft.Agents.AI.UnitTests; + +/// +/// Unit tests for the class. +/// +public class AsyncRequestBrokerTests +{ + #region Constructor Tests + + /// + /// Verify that a non-positive settled queue size is rejected. + /// + [Theory] + [InlineData(0)] + [InlineData(-1)] + public void Constructor_InvalidSettledQueueSize_Throws(int size) + { + // Act & Assert + Assert.Throws(() => new AsyncRequestBroker(size)); + } + + #endregion + + #region Resolve / Reject Tests + + /// + /// Verify that Resolve delivers the result to the waiter and clears the pending entry. + /// + [Fact] + public async Task Resolve_DeliversResultToWaiterAsync() + { + // Arrange + var broker = new AsyncRequestBroker(); + var task = broker.RequestAsync("req-1"); + Assert.Equal(1, broker.PendingCount); + + // Act + bool resolved = broker.Resolve("req-1", "answer"); + + // Assert + Assert.True(resolved); + Assert.Equal("answer", await task); + Assert.Equal(0, broker.PendingCount); + } + + /// + /// Verify that Reject delivers the exception to the waiter and clears the pending entry. + /// + [Fact] + public async Task Reject_DeliversExceptionToWaiterAsync() + { + // Arrange + var broker = new AsyncRequestBroker(); + var task = broker.RequestAsync("req-1"); + var error = new InvalidCastException("boom"); + + // Act + bool rejected = broker.Reject("req-1", error); + + // Assert + Assert.True(rejected); + var thrown = await Assert.ThrowsAsync(() => task); + Assert.Same(error, thrown); + Assert.Equal(0, broker.PendingCount); + } + + /// + /// Verify that resolving or rejecting an unknown ID returns false. + /// + [Fact] + public void ResolveAndReject_UnknownId_ReturnFalse() + { + // Arrange + var broker = new AsyncRequestBroker(); + + // Act & Assert + Assert.False(broker.Resolve("missing", 1)); + Assert.False(broker.Reject("missing", new InvalidOperationException())); + Assert.Equal(0, broker.PendingCount); + } + + /// + /// Verify that a late response for an already-settled ID returns false and does not affect state. + /// + [Fact] + public async Task ResolveAndReject_LateResponseForSettledId_ReturnFalseAsync() + { + // Arrange + var broker = new AsyncRequestBroker(); + var task = broker.RequestAsync("req-1"); + Assert.True(broker.Resolve("req-1", 1)); + Assert.Equal(1, await task); + + // Act + bool lateResolve = broker.Resolve("req-1", 2); + bool lateReject = broker.Reject("req-1", new InvalidOperationException()); + + // Assert + Assert.False(lateResolve); + Assert.False(lateReject); + Assert.Equal(0, broker.PendingCount); + } + + /// + /// Verify that null arguments are rejected. + /// + [Fact] + public async Task NullArguments_ThrowAsync() + { + // Arrange + var broker = new AsyncRequestBroker(); + + // Act & Assert + await Assert.ThrowsAsync(() => broker.RequestAsync(null!)); + Assert.Throws(() => broker.Resolve(null!, 1)); + Assert.Throws(() => broker.Reject(null!, new InvalidOperationException())); + Assert.Throws(() => broker.Reject("req-1", null!)); + Assert.Equal(0, broker.PendingCount); + } + + #endregion + + #region Duplicate ID Tests + + /// + /// Verify that registering an ID that is already pending is rejected without disturbing the original waiter. + /// + [Fact] + public async Task RequestAsync_DuplicatePendingId_ThrowsAsync() + { + // Arrange + var broker = new AsyncRequestBroker(); + var original = broker.RequestAsync("req-1"); + + // Act + var ex = await Assert.ThrowsAsync(() => broker.RequestAsync("req-1")); + + // Assert + Assert.Contains("already pending", ex.Message); + Assert.Equal(1, broker.PendingCount); + Assert.True(broker.Resolve("req-1", 42)); + Assert.Equal(42, await original); + } + + /// + /// Verify that registering a recently settled ID is rejected. + /// + [Fact] + public async Task RequestAsync_SettledId_ThrowsAsync() + { + // Arrange + var broker = new AsyncRequestBroker(); + var task = broker.RequestAsync("req-1"); + broker.Resolve("req-1", 1); + await task; + + // Act + var ex = await Assert.ThrowsAsync(() => broker.RequestAsync("req-1")); + + // Assert + Assert.Contains("already settled", ex.Message); + Assert.Equal(0, broker.PendingCount); + } + + #endregion + + #region Timeout and Cancellation Tests + + /// + /// Verify that an elapsed timeout cancels the waiter, clears the pending entry, and records the ID as settled. + /// + [Fact] + public async Task RequestAsync_Timeout_CleansUpAndTracksSettledAsync() + { + // Arrange + var broker = new AsyncRequestBroker(); + + // Act + await Assert.ThrowsAnyAsync(() => broker.RequestAsync("req-1", TimeSpan.Zero)); + + // Assert + Assert.Equal(0, broker.PendingCount); + Assert.False(broker.Resolve("req-1", 1)); + await Assert.ThrowsAsync(() => broker.RequestAsync("req-1")); + } + + /// + /// Verify that cancelling the token cancels the waiter, clears the pending entry, and records the ID as settled. + /// + [Fact] + public async Task RequestAsync_Cancellation_CleansUpAndTracksSettledAsync() + { + // Arrange + var broker = new AsyncRequestBroker(); + using var cts = new CancellationTokenSource(); + var task = broker.RequestAsync("req-1", cancellationToken: cts.Token); + Assert.Equal(1, broker.PendingCount); + + // Act + cts.Cancel(); + + // Assert + await Assert.ThrowsAnyAsync(() => task); + Assert.Equal(0, broker.PendingCount); + Assert.False(broker.Resolve("req-1", 1)); + await Assert.ThrowsAsync(() => broker.RequestAsync("req-1")); + } + + /// + /// Verify that an already-cancelled token completes the request immediately and leaves no pending entry. + /// + [Fact] + public async Task RequestAsync_PreCancelledToken_CleansUpAsync() + { + // Arrange + var broker = new AsyncRequestBroker(); + using var cts = new CancellationTokenSource(); + cts.Cancel(); + + // Act & Assert + await Assert.ThrowsAnyAsync(() => broker.RequestAsync("req-1", cancellationToken: cts.Token)); + Assert.Equal(0, broker.PendingCount); + } + + /// + /// Verify that resolving before cancellation wins and the later cancellation has no effect. + /// + [Fact] + public async Task RequestAsync_CancelAfterResolve_ReturnsResultAsync() + { + // Arrange + var broker = new AsyncRequestBroker(); + using var cts = new CancellationTokenSource(); + var task = broker.RequestAsync("req-1", cancellationToken: cts.Token); + + // Act + Assert.True(broker.Resolve("req-1", 7)); + cts.Cancel(); + + // Assert + Assert.Equal(7, await task); + Assert.Equal(0, broker.PendingCount); + } + + /// + /// Verify that an unsupported timeout is rejected before the request is registered. + /// + [Fact] + public async Task RequestAsync_InvalidTimeout_ThrowsWithoutRegisteringAsync() + { + // Arrange + var broker = new AsyncRequestBroker(); + + // Act & Assert + await Assert.ThrowsAsync(() => broker.RequestAsync("req-1", TimeSpan.FromMilliseconds(-5))); + await Assert.ThrowsAsync(() => broker.RequestAsync("req-1", TimeSpan.FromMilliseconds((double)int.MaxValue + 1))); + Assert.Equal(0, broker.PendingCount); + + // The ID must remain usable after the rejected attempts. + var task = broker.RequestAsync("req-1"); + Assert.True(broker.Resolve("req-1", 3)); + Assert.Equal(3, await task); + } + + /// + /// Verify that is accepted and waits for a response. + /// + [Fact] + public async Task RequestAsync_InfiniteTimeout_WaitsForResolveAsync() + { + // Arrange + var broker = new AsyncRequestBroker(); + var task = broker.RequestAsync("req-1", Timeout.InfiniteTimeSpan); + + // Act + Assert.True(broker.Resolve("req-1", 9)); + + // Assert + Assert.Equal(9, await task); + } + + #endregion + + #region Settled Queue Eviction Tests + + /// + /// Verify that the oldest settled ID is evicted once the bounded queue is full and can then be reused. + /// + [Fact] + public async Task SettledQueue_EvictsOldestIdAsync() + { + // Arrange + var broker = new AsyncRequestBroker(settledQueueSize: 2); + foreach (string id in new[] { "a", "b", "c" }) + { + var task = broker.RequestAsync(id); + broker.Resolve(id, 0); + await task; + } + + // Act + var reused = broker.RequestAsync("a"); + + // Assert + Assert.Equal(1, broker.PendingCount); + await Assert.ThrowsAsync(() => broker.RequestAsync("b")); + await Assert.ThrowsAsync(() => broker.RequestAsync("c")); + Assert.True(broker.Resolve("a", 5)); + Assert.Equal(5, await reused); + } + + #endregion + + #region Concurrency Tests + + /// + /// Verify invariants under concurrent resolve, reject, cancellation, and duplicate registration: + /// exactly one settlement wins, duplicate registrations are always rejected, and no entries leak. + /// + [Fact] + public async Task ConcurrentSettlementAndRegistration_PreservesInvariantsAsync() + { + // Arrange + const int Count = 500; + var broker = new AsyncRequestBroker(settledQueueSize: Count * 2); + var ctsList = Enumerable.Range(0, Count).Select(_ => new CancellationTokenSource()).ToArray(); + var requests = Enumerable.Range(0, Count) + .Select(i => broker.RequestAsync($"req-{i}", cancellationToken: ctsList[i].Token)) + .ToArray(); + Assert.Equal(Count, broker.PendingCount); + + var resolveWon = new bool[Count]; + var rejectWon = new bool[Count]; + var duplicates = new Task[Count]; + + // Act + await Task.WhenAll(Enumerable.Range(0, Count).Select(i => Task.WhenAll( + Task.Run(() => resolveWon[i] = broker.Resolve($"req-{i}", i)), + Task.Run(() => rejectWon[i] = broker.Reject($"req-{i}", new InvalidOperationException("rejected"))), + Task.Run(() => ctsList[i].Cancel()), + // Cast to Action so Task.Run does not unwrap (and await) the duplicate request task. + Task.Run((Action)(() => duplicates[i] = broker.RequestAsync($"req-{i}")))))); + + // Assert + Assert.Equal(0, broker.PendingCount); + for (int i = 0; i < Count; i++) + { + Assert.False(resolveWon[i] && rejectWon[i]); + var request = requests[i]; + + if (resolveWon[i]) + { + Assert.Equal(i, await request); + } + else if (rejectWon[i]) + { + await Assert.ThrowsAsync(() => request); + } + else + { + await Assert.ThrowsAnyAsync(() => request); + } + + // The ID was always either pending or settled, so the duplicate registration must have been rejected. + await Assert.ThrowsAsync(() => duplicates[i]); + ctsList[i].Dispose(); + } + } + + #endregion +} From ffed2866a6b6b541254805a6d1c0ae3ee1aab777 Mon Sep 17 00:00:00 2001 From: Mark Licata Date: Wed, 7 Oct 2026 09:07:20 -0700 Subject: [PATCH 3/3] Return true from Resolve/Reject for recently settled IDs to match Python Co-Authored-By: Claude Opus 5.5 --- .../Harness/AsyncRequestBroker.cs | 38 +++++++++------ .../Harness/AsyncRequestBrokerTests.cs | 48 ++++++++++++------- 2 files changed, 55 insertions(+), 31 deletions(-) diff --git a/dotnet/src/Microsoft.Agents.AI/Harness/AsyncRequestBroker.cs b/dotnet/src/Microsoft.Agents.AI/Harness/AsyncRequestBroker.cs index d06d7de5cc6..47c2b0ae6f1 100644 --- a/dotnet/src/Microsoft.Agents.AI/Harness/AsyncRequestBroker.cs +++ b/dotnet/src/Microsoft.Agents.AI/Harness/AsyncRequestBroker.cs @@ -133,17 +133,18 @@ public async Task RequestAsync(string requestId, TimeSpan? timeout = nu /// The request identifier. /// The result to deliver. /// - /// if a waiter was notified; if the request was not found - /// (already settled, timed out, cancelled, or never registered). + /// if a waiter was notified, or if the request was recently settled (resolved, rejected, + /// timed out, or cancelled), in which case the late is ignored; + /// if the request ID is unknown or has been evicted from the settled-ID history. /// public bool Resolve(string requestId, TResult result) { - if (!this.TryRemovePending(requestId, out var tcs)) + if (!this.TrySettle(requestId, out var tcs)) { return false; } - tcs.TrySetResult(result); + tcs?.TrySetResult(result); return true; } @@ -151,36 +152,45 @@ public bool Resolve(string requestId, TResult result) /// The request identifier. /// The exception to deliver. /// - /// if a waiter was notified; if the request was not found - /// (already settled, timed out, cancelled, or never registered). + /// if a waiter was notified, or if the request was recently settled (resolved, rejected, + /// timed out, or cancelled), in which case the late is ignored; + /// if the request ID is unknown or has been evicted from the settled-ID history. /// public bool Reject(string requestId, Exception exception) { _ = Throw.IfNull(exception); - if (!this.TryRemovePending(requestId, out var tcs)) + if (!this.TrySettle(requestId, out var tcs)) { return false; } - tcs.TrySetException(exception); + tcs?.TrySetException(exception); return true; } - private bool TryRemovePending(string requestId, [NotNullWhen(true)] out TaskCompletionSource? tcs) + /// + /// Removes a pending request and records it as settled, or detects a late response for a recently settled request. + /// + /// + /// if the ID was pending ( is the waiter to complete) or recently + /// settled ( is ); if the ID is unknown. + /// + private bool TrySettle(string requestId, out TaskCompletionSource? tcs) { _ = Throw.IfNull(requestId); lock (this._lock) { - if (!this._pending.TryGetValue(requestId, out tcs)) + if (this._pending.TryGetValue(requestId, out tcs)) { - return false; + this._pending.Remove(requestId); + this.TrackSettledLocked(requestId); + return true; } - this._pending.Remove(requestId); - this.TrackSettledLocked(requestId); - return true; + tcs = null; + return this._settledSet.Contains(requestId); } } diff --git a/dotnet/tests/Microsoft.Agents.AI.UnitTests/Harness/AsyncRequestBrokerTests.cs b/dotnet/tests/Microsoft.Agents.AI.UnitTests/Harness/AsyncRequestBrokerTests.cs index ece1eb9a23e..9823262c73f 100644 --- a/dotnet/tests/Microsoft.Agents.AI.UnitTests/Harness/AsyncRequestBrokerTests.cs +++ b/dotnet/tests/Microsoft.Agents.AI.UnitTests/Harness/AsyncRequestBrokerTests.cs @@ -87,10 +87,10 @@ public void ResolveAndReject_UnknownId_ReturnFalse() } /// - /// Verify that a late response for an already-settled ID returns false and does not affect state. + /// Verify that a late response for a recently resolved ID returns true, is ignored, and does not affect state. /// [Fact] - public async Task ResolveAndReject_LateResponseForSettledId_ReturnFalseAsync() + public async Task ResolveAndReject_LateResponseForSettledId_ReturnTrueAsync() { // Arrange var broker = new AsyncRequestBroker(); @@ -103,8 +103,9 @@ public async Task ResolveAndReject_LateResponseForSettledId_ReturnFalseAsync() bool lateReject = broker.Reject("req-1", new InvalidOperationException()); // Assert - Assert.False(lateResolve); - Assert.False(lateReject); + Assert.True(lateResolve); + Assert.True(lateReject); + Assert.Equal(1, await task); Assert.Equal(0, broker.PendingCount); } @@ -187,7 +188,8 @@ public async Task RequestAsync_Timeout_CleansUpAndTracksSettledAsync() // Assert Assert.Equal(0, broker.PendingCount); - Assert.False(broker.Resolve("req-1", 1)); + Assert.True(broker.Resolve("req-1", 1)); + Assert.True(broker.Reject("req-1", new InvalidOperationException())); await Assert.ThrowsAsync(() => broker.RequestAsync("req-1")); } @@ -209,7 +211,8 @@ public async Task RequestAsync_Cancellation_CleansUpAndTracksSettledAsync() // Assert await Assert.ThrowsAnyAsync(() => task); Assert.Equal(0, broker.PendingCount); - Assert.False(broker.Resolve("req-1", 1)); + Assert.True(broker.Resolve("req-1", 1)); + await Assert.ThrowsAnyAsync(() => task); await Assert.ThrowsAsync(() => broker.RequestAsync("req-1")); } @@ -306,9 +309,15 @@ public async Task SettledQueue_EvictsOldestIdAsync() } // Act + bool evictedResolve = broker.Resolve("a", 1); + bool evictedReject = broker.Reject("a", new InvalidOperationException()); + bool retainedResolve = broker.Resolve("b", 1); var reused = broker.RequestAsync("a"); // Assert + Assert.False(evictedResolve); + Assert.False(evictedReject); + Assert.True(retainedResolve); Assert.Equal(1, broker.PendingCount); await Assert.ThrowsAsync(() => broker.RequestAsync("b")); await Assert.ThrowsAsync(() => broker.RequestAsync("c")); @@ -322,7 +331,7 @@ public async Task SettledQueue_EvictsOldestIdAsync() /// /// Verify invariants under concurrent resolve, reject, cancellation, and duplicate registration: - /// exactly one settlement wins, duplicate registrations are always rejected, and no entries leak. + /// exactly one settlement determines the outcome, late responses are accepted, duplicate registrations are always rejected, and no entries leak. /// [Fact] public async Task ConcurrentSettlementAndRegistration_PreservesInvariantsAsync() @@ -336,14 +345,14 @@ public async Task ConcurrentSettlementAndRegistration_PreservesInvariantsAsync() .ToArray(); Assert.Equal(Count, broker.PendingCount); - var resolveWon = new bool[Count]; - var rejectWon = new bool[Count]; + var resolveAccepted = new bool[Count]; + var rejectAccepted = new bool[Count]; var duplicates = new Task[Count]; // Act await Task.WhenAll(Enumerable.Range(0, Count).Select(i => Task.WhenAll( - Task.Run(() => resolveWon[i] = broker.Resolve($"req-{i}", i)), - Task.Run(() => rejectWon[i] = broker.Reject($"req-{i}", new InvalidOperationException("rejected"))), + Task.Run(() => resolveAccepted[i] = broker.Resolve($"req-{i}", i)), + Task.Run(() => rejectAccepted[i] = broker.Reject($"req-{i}", new InvalidOperationException("rejected"))), Task.Run(() => ctsList[i].Cancel()), // Cast to Action so Task.Run does not unwrap (and await) the duplicate request task. Task.Run((Action)(() => duplicates[i] = broker.RequestAsync($"req-{i}")))))); @@ -352,20 +361,25 @@ await Task.WhenAll(Enumerable.Range(0, Count).Select(i => Task.WhenAll( Assert.Equal(0, broker.PendingCount); for (int i = 0; i < Count; i++) { - Assert.False(resolveWon[i] && rejectWon[i]); - var request = requests[i]; + // The ID was always either pending or recently settled, so both responses are accepted + // (the one that loses the race is ignored as a late response). + Assert.True(resolveAccepted[i]); + Assert.True(rejectAccepted[i]); - if (resolveWon[i]) + // Exactly one settlement (resolve, reject, or cancel) determines the outcome. + var request = requests[i]; + await Task.WhenAny(request); + if (request.Status == TaskStatus.RanToCompletion) { Assert.Equal(i, await request); } - else if (rejectWon[i]) + else if (request.IsCanceled) { - await Assert.ThrowsAsync(() => request); + await Assert.ThrowsAnyAsync(() => request); } else { - await Assert.ThrowsAnyAsync(() => request); + await Assert.ThrowsAsync(() => request); } // The ID was always either pending or settled, so the duplicate registration must have been rejected.