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..47c2b0ae6f1 --- /dev/null +++ b/dotnet/src/Microsoft.Agents.AI/Harness/AsyncRequestBroker.cs @@ -0,0 +1,209 @@ +// Copyright (c) Microsoft. All rights reserved. + +using System; +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; + +/// +/// 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 Dictionary> _pending = new(StringComparer.Ordinal); + private readonly Queue _settledQueue = new(); + private readonly HashSet _settledSet = new(StringComparer.Ordinal); + private readonly int _settledQueueSize; + + // 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. + /// + /// 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 + { + get + { + lock (this._lock) + { + return this._pending.Count; + } + } + } + + /// Suspend until the request is resolved, rejected, or cancelled. + /// Unique identifier for this request. + /// + /// 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._lock) + { + if (this._settledSet.Contains(requestId)) + { + throw new InvalidOperationException($"Request '{requestId}' was already settled."); + } + + if (this._pending.ContainsKey(requestId)) + { + throw new InvalidOperationException($"Request '{requestId}' is already pending."); + } + + this._pending.Add(requestId, tcs); + } + + using var cts = hasTimeout + ? CancellationTokenSource.CreateLinkedTokenSource(cancellationToken) + : null; + + // The timeout was validated above, so CancelAfter cannot throw here. + cts?.CancelAfter(timeout!.Value); + + var effectiveToken = cts?.Token ?? cancellationToken; + using var registration = effectiveToken.Register(() => + { + 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) + { + 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, 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.TrySettle(requestId, out var tcs)) + { + return false; + } + + tcs?.TrySetResult(result); + return true; + } + + /// Deliver an exception to a waiting call. + /// The request identifier. + /// The exception to deliver. + /// + /// 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.TrySettle(requestId, out var tcs)) + { + return false; + } + + tcs?.TrySetException(exception); + return true; + } + + /// + /// 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)) + { + this._pending.Remove(requestId); + this.TrackSettledLocked(requestId); + return true; + } + + tcs = null; + return this._settledSet.Contains(requestId); + } + } + + // 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/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 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..9823262c73f --- /dev/null +++ b/dotnet/tests/Microsoft.Agents.AI.UnitTests/Harness/AsyncRequestBrokerTests.cs @@ -0,0 +1,392 @@ +// 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 a recently resolved ID returns true, is ignored, and does not affect state. + /// + [Fact] + public async Task ResolveAndReject_LateResponseForSettledId_ReturnTrueAsync() + { + // 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.True(lateResolve); + Assert.True(lateReject); + Assert.Equal(1, await task); + 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.True(broker.Resolve("req-1", 1)); + Assert.True(broker.Reject("req-1", new InvalidOperationException())); + 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.True(broker.Resolve("req-1", 1)); + await Assert.ThrowsAnyAsync(() => task); + 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 + 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")); + 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 determines the outcome, late responses are accepted, 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 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(() => 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}")))))); + + // Assert + Assert.Equal(0, broker.PendingCount); + for (int i = 0; i < Count; 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]); + + // 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 (request.IsCanceled) + { + await Assert.ThrowsAnyAsync(() => request); + } + else + { + await Assert.ThrowsAsync(() => 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 +}