From f4aad5114046c04e1e1631ef175268f8768a8900 Mon Sep 17 00:00:00 2001 From: "Harol A. Reina H." Date: Sun, 4 Oct 2026 06:26:34 -0500 Subject: [PATCH 1/2] build(deps): move every Verbara.Sdk.Pro pin to 2.17.1-pro All 22 Verbara.Sdk.Pro.* pins move from 2.17.0-pro together: Pro packages depend on their siblings at the same version, so a partial bump risks NU1605. 2.17.1-pro is a patch with no API change; its queue_members upsert keeps an existing row's paused value on conflict. --- Directory.Packages.props | 44 ++++++++++++++++++++-------------------- 1 file changed, 22 insertions(+), 22 deletions(-) diff --git a/Directory.Packages.props b/Directory.Packages.props index 0ef6bc72..0158ce47 100644 --- a/Directory.Packages.props +++ b/Directory.Packages.props @@ -57,28 +57,28 @@ - - - - - - - - - - - - - - - - - - - - - - + + + + + + + + + + + + + + + + + + + + + + From 91b7b369a3ea25f48d4336912f6781089ea6c2b9 Mon Sep 17 00:00:00 2001 From: "Harol A. Reina H." Date: Sun, 4 Oct 2026 06:26:34 -0500 Subject: [PATCH 2/2] fix(realtime): keep an Available agent unpaused across reconcile and membership saves Every 60 s RealtimeReconciliationService, and every membership save through RealtimeSyncingQueueMembershipStore, re-issued AddQueueMemberAsync, which wrote queue_members.paused = 1. app_queue then skipped an idle Available agent until the agent changed state by hand. Extract AgentPauseCoordinator: the pause rule (!AgentStateMachine.IsRoutable(agent.State) || agent.PendingState is not null) and the per-agent lock, moved out of RealtimeStateBridge. After each voice upsert, the reconciler (once per agent per pass, after its last upsert) and the membership decorator call ConvergeAsync, which re-reads the agent from the undecorated store under that lock and writes SyncAgentPausedAsync (DB only, no AMI). The bridge keeps its event mapping and AMI QueuePause, under the same lock. The coordinator is registered in AddRealtimeSyncingStores, so Program.cs does not grow. The lock is per process: with several API replicas a reconcile on one pod can still race a state change on another (accepted, design R2). Change: queue-members-stay-unpaused-across-reconcile (Verbara.Platform/ADR-0026). --- CHANGELOG.md | 20 ++ .../RealtimeSyncingStoresExtensions.cs | 6 + .../Services/AgentPauseCoordinator.cs | 113 +++++++++ .../Services/RealtimeReconciliationService.cs | 55 ++++- .../Services/RealtimeStateBridge.cs | 21 +- .../RealtimeSyncingQueueMembershipStore.cs | 15 ++ .../AgentPauseCoordinatorTests.cs | 148 ++++++++++++ .../RealtimeReconciliationServiceTests.cs | 218 +++++++++++++++++- .../RealtimeStateBridgeTests.cs | 49 ++++ .../RealtimeSyncCallLog.cs | 23 ++ .../RealtimeSyncingStoreDecoratorTests.cs | 75 +++++- .../RealtimeSyncingStoresRegistrarTests.cs | 42 ++++ 12 files changed, 763 insertions(+), 22 deletions(-) create mode 100644 src/Verbara.Platform.Api/Services/AgentPauseCoordinator.cs create mode 100644 tests/Verbara.Platform.Api.Tests/AgentPauseCoordinatorTests.cs create mode 100644 tests/Verbara.Platform.Api.Tests/RealtimeSyncCallLog.cs diff --git a/CHANGELOG.md b/CHANGELOG.md index 8345674c..7f9c610e 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -45,6 +45,26 @@ Versioning: [Semantic Versioning](https://semver.org/spec/v2.0.0.html). - **Deliberately out of scope** (`Verbara.Sdk/ADR-0040` D4): `Microsoft.Extensions.TimeProvider.Testing` stays at `10.10.0`, and neither `OpenTelemetry` itself nor `NATS.Client.Core` gains a direct pin. The second hop, Sdk `2.8.0` (H142), is a separate change. decision_ref `Verbara.Sdk/ADR-0040`. +- **Verbara.Sdk.Pro `2.17.0-pro` → `2.17.1-pro`** — all 22 `Verbara.Sdk.Pro.*` pins move together + (Pro packages depend on their siblings at the same version, so a partial bump risks `NU1605`). + A patch with no API change: the `queue_members` upsert stops overwriting `paused` on conflict (see + *Fixed*). (#N) + +### Fixed + +- **An idle `Available` agent no longer stops receiving queue calls after a minute.** Every 60 s the + realtime reconciliation (and every queue-membership save) re-wrote the agent's Asterisk Realtime + `queue_members` rows with `paused = 1`, so `app_queue` skipped an agent the console still showed as + `Available` until the agent changed state by hand, and callers waited while a free agent sat idle. + Verbara.Sdk.Pro `2.17.1-pro` now keeps an existing row's `paused` value on re-upsert, and Platform + re-asserts the agent's real pause state after every member upsert it issues (reconcile pass and + membership save): it re-reads the agent under the same per-agent lock the state bridge uses and + pauses only a non-routable agent (`Available` and `Busy` are routable) or one with a pending pause. + This also unpauses a member created, or re-created after orphan cleanup, for an agent who is + already `Available`. Rows already stuck at `paused = 1` are corrected by the first reconcile after + the upgrade. **Known limitation:** the per-agent lock is per process. With several API replicas, + one pod's reconcile can still briefly race another pod's state change; the next state change or + reconcile re-asserts the correct value. A cross-pod lock is a separate change. (#N) --- diff --git a/src/Verbara.Platform.Api/DependencyInjection/RealtimeSyncingStoresExtensions.cs b/src/Verbara.Platform.Api/DependencyInjection/RealtimeSyncingStoresExtensions.cs index 7824150a..ed1ea467 100644 --- a/src/Verbara.Platform.Api/DependencyInjection/RealtimeSyncingStoresExtensions.cs +++ b/src/Verbara.Platform.Api/DependencyInjection/RealtimeSyncingStoresExtensions.cs @@ -1,4 +1,5 @@ using Microsoft.Extensions.DependencyInjection; +using Microsoft.Extensions.DependencyInjection.Extensions; using Microsoft.Extensions.Logging; using Verbara.Platform.Api.Services; using Verbara.Platform.Queues; @@ -46,6 +47,10 @@ public static IServiceCollection AddRealtimeSyncingStores(this IServiceCollectio { ArgumentNullException.ThrowIfNull(services); + // The per-agent pause lock + rule shared by RealtimeStateBridge, the reconciler and the + // membership decorator (queue-members-stay-unpaused-across-reconcile). No external seam. + services.TryAddSingleton(); + // Re-key each concrete store as the inner, drop the unkeyed alias, then re-register // the unkeyed service as the decorator (which falls back to the inner when realtime // is not registered). @@ -85,6 +90,7 @@ public static IServiceCollection AddRealtimeSyncingStores(this IServiceCollectio sp.GetRequiredKeyedService(QueueStoreInner), sp.GetRequiredKeyedService(AgentStoreInner), sync, + sp.GetRequiredService(), sp.GetRequiredService>()); }); diff --git a/src/Verbara.Platform.Api/Services/AgentPauseCoordinator.cs b/src/Verbara.Platform.Api/Services/AgentPauseCoordinator.cs new file mode 100644 index 00000000..36b419ba --- /dev/null +++ b/src/Verbara.Platform.Api/Services/AgentPauseCoordinator.cs @@ -0,0 +1,113 @@ +using System.Collections.Concurrent; +using Verbara.Platform.Core; +using Verbara.Platform.Queues; +using Verbara.Sdk.Pro.Realtime; + +namespace Verbara.Platform.Api.Services; + +/// +/// The one owner of an agent's Asterisk Realtime pause decision: the pause rule and the per-agent +/// lock that every queue_members.paused writer in this process takes +/// (queue-members-stay-unpaused-across-reconcile, design D2). +/// +/// +/// +/// inserts a new row with paused = 1 +/// and, from Sdk.Pro 2.17.1-pro, leaves an existing row's paused untouched. So after every +/// upsert Platform issues, re-asserts the agent's real state with +/// , which sets every row of the agent's +/// interface. It writes the database only, never AMI (design D3): app_queue re-reads the +/// realtime row when a caller joins. keeps sending AMI +/// QueuePause on real state changes, under the same lock. +/// +/// +/// Per-process lock (accepted risk R2). The semaphores live in this singleton, so they +/// serialize writers inside one API process only. With several API replicas, one pod's reconcile +/// can still race another pod's bridge; the fresh read under the lock narrows that window to the +/// read-to-write gap, and the next state event re-asserts the truth. +/// +/// +/// The coordinator has no external dependency: the agent store and the sync service are passed +/// per call, so it is safe to construct at host start without any realtime configuration. +/// +/// +internal sealed class AgentPauseCoordinator +{ + private readonly ConcurrentDictionary _agentLocks = new(StringComparer.Ordinal); + + /// + /// The pause rule shared with : an agent is paused in Asterisk + /// when its state is not routable ( and + /// are) or when a deferred pause is pending. + /// + public static bool ShouldPause(Agent agent) + { + ArgumentNullException.ThrowIfNull(agent); + return !AgentStateMachine.IsRoutable(agent.State) || agent.PendingState is not null; + } + + /// + /// Mirrors the Sdk.Pro voice gate of + /// (ADR-0026 Phase B): a null channel list, or one that contains "voice" + /// (case-insensitive), upserts a queue_members row. Any other list is digital-only and + /// removes the row instead, so there is nothing to converge. + /// + public static bool CreatesVoiceRow(IReadOnlyList? allowedChannels) + { + if (allowedChannels is null) + return true; + foreach (var channel in allowedChannels) + { + if (string.Equals(channel, "voice", StringComparison.OrdinalIgnoreCase)) + return true; + } + return false; + } + + /// + /// Waits for the per-agent lock and returns a handle that releases it on dispose. Every + /// queue_members.paused write for in this process runs under it. + /// + public async Task AcquireAsync(string agentId, CancellationToken ct) + { + ArgumentException.ThrowIfNullOrEmpty(agentId); + var semaphore = _agentLocks.GetOrAdd(agentId, static _ => new SemaphoreSlim(1, 1)); + await semaphore.WaitAsync(ct).ConfigureAwait(false); + return new Releaser(semaphore); + } + + /// + /// Re-reads the agent under its lock and writes its real pause state to every + /// queue_members row of its interface. Returns the value written, or null when + /// the agent no longer exists (nothing is written). + /// + /// The agent's tenant. + /// The agent to converge. + /// An agent store that reads the current row (no tick-level cache). + /// The realtime sync service that owns the write. + /// Cancellation. + public async Task ConvergeAsync( + TenantId tenantId, EntityId agentId, IAgentStore agents, IRealtimeSyncService sync, CancellationToken ct) + { + ArgumentNullException.ThrowIfNull(agents); + ArgumentNullException.ThrowIfNull(sync); + + using (await AcquireAsync(agentId.Value, ct).ConfigureAwait(false)) + { + var agent = await agents.GetByIdAsync(tenantId, agentId, ct).ConfigureAwait(false); + if (agent is null) + return null; + + var shouldPause = ShouldPause(agent); + await sync.SyncAgentPausedAsync(tenantId.Value, agentId.Value, shouldPause, ct).ConfigureAwait(false); + return shouldPause; + } + } + + private sealed class Releaser(SemaphoreSlim semaphore) : IDisposable + { + private SemaphoreSlim? _semaphore = semaphore; + + public void Dispose() => Interlocked.Exchange(ref _semaphore, null)?.Release(); + } +} diff --git a/src/Verbara.Platform.Api/Services/RealtimeReconciliationService.cs b/src/Verbara.Platform.Api/Services/RealtimeReconciliationService.cs index c093a8f1..086e28aa 100644 --- a/src/Verbara.Platform.Api/Services/RealtimeReconciliationService.cs +++ b/src/Verbara.Platform.Api/Services/RealtimeReconciliationService.cs @@ -1,4 +1,5 @@ using System.Diagnostics.Metrics; +using Verbara.Platform.Api.DependencyInjection; using Verbara.Platform.Api.Health; using Verbara.Platform.Core; using Verbara.Platform.Queues; @@ -26,10 +27,20 @@ namespace Verbara.Platform.Api.Services; /// gate handles uniformly: /// /// -/// null or includes "voice" → idempotent upsert (no-op when already present). +/// null or includes "voice" → upsert. A new row is inserted with +/// paused = 1; an existing row keeps its paused value (Sdk.Pro 2.17.1-pro) while its +/// name and penalty are refreshed. So a freshly inserted row starts paused whatever the agent's +/// state, which the paused convergence below corrects. /// populated and excludes "voice" → short-circuits to (idempotent when row absent). /// /// +/// Paused convergence. After an agent's last voice upsert of the pass, the reconciler calls +/// once for that agent: under the per-agent lock +/// shared with it re-reads the agent (not the tick cache) and +/// writes its real pause state with , so an +/// idle Available agent is never left paused by a tick (queue-members-stay-unpaused-across-reconcile). +/// +/// /// Out of scope. Orphan rows in Asterisk Realtime with no corresponding /// Verbara membership are NOT detected here — they are handled by the /// scripts/infer-memberships-from-skills.sh one-shot for legacy upgrades @@ -61,6 +72,7 @@ internal sealed partial class RealtimeReconciliationService : BackgroundService private readonly IServiceHeartbeat _heartbeat; private readonly TimeSpan _interval; private readonly ResiliencePolicy _policy; + private readonly AgentPauseCoordinator _pauseCoordinator; private readonly ILogger _logger; private readonly Meter _meter; @@ -72,12 +84,14 @@ public RealtimeReconciliationService( IServiceProvider services, IServiceHeartbeat heartbeat, Microsoft.Extensions.Options.IOptions options, + AgentPauseCoordinator pauseCoordinator, ILogger logger, [FromKeyedServices(ResiliencePolicyKey)] ResiliencePolicy? policy = null) { ArgumentNullException.ThrowIfNull(services); ArgumentNullException.ThrowIfNull(heartbeat); ArgumentNullException.ThrowIfNull(options); + ArgumentNullException.ThrowIfNull(pauseCoordinator); ArgumentNullException.ThrowIfNull(logger); _services = services; @@ -85,6 +99,7 @@ public RealtimeReconciliationService( var intervalSeconds = Math.Max(5, options.Value.ReconcilerIntervalSeconds); _interval = TimeSpan.FromSeconds(intervalSeconds); _policy = policy ?? ResiliencePolicy.NoOp; + _pauseCoordinator = pauseCoordinator; _logger = logger; _meter = new Meter(MeterName); @@ -93,7 +108,7 @@ public RealtimeReconciliationService( _membershipsReconciled = _meter.CreateCounter("realtime.reconciliation.memberships", description: "queue_memberships rows re-issued to IRealtimeSyncService per tick."); _syncFailures = _meter.CreateCounter("realtime.reconciliation.sync_failures", - description: "AddQueueMemberAsync calls that threw during reconciliation (counted, not retried inline)."); + description: "AddQueueMemberAsync or paused-convergence calls that threw during reconciliation (counted, not retried inline)."); } protected override async Task ExecuteAsync(CancellationToken stoppingToken) @@ -168,6 +183,10 @@ internal async Task ReconcileAsync(CancellationToken ct) var membershipStore = sp.GetRequiredService(); var queueStore = sp.GetRequiredService(); var agentStore = sp.GetRequiredService(); + // The paused convergence re-reads the agent from the UNDECORATED store when the + // realtime-syncing decorators are wired (never the tick cache below). + var freshAgentStore = + sp.GetKeyedService(RealtimeSyncingStoresExtensions.AgentStoreInner) ?? agentStore; var tenants = await tenantStore.GetAllActiveAsync(ct); @@ -177,7 +196,7 @@ internal async Task ReconcileAsync(CancellationToken ct) var tid = new TenantId(tenant.TenantId); await ReconcileTenantAsync( - tid, tenant.TenantId, membershipStore, queueStore, agentStore, syncService, ct); + tid, tenant.TenantId, membershipStore, queueStore, agentStore, freshAgentStore, syncService, ct); _tenantsProcessed.Add(1, new KeyValuePair("tenant", tenant.TenantId)); } @@ -189,6 +208,7 @@ private async Task ReconcileTenantAsync( IQueueMembershipStore membershipStore, IQueueStore queueStore, IAgentStore agentStore, + IAgentStore freshAgentStore, IRealtimeSyncService syncService, CancellationToken ct) { @@ -200,6 +220,10 @@ private async Task ReconcileTenantAsync( // for every membership row. var queueCache = new Dictionary(StringComparer.Ordinal); var agentCache = new Dictionary(StringComparer.Ordinal); + // Agents with at least one voice row upserted this pass, in first-seen order. Each one is + // converged once, after the loop — i.e. after that agent's last AddQueueMemberAsync. + var toConverge = new List(); + var queuedForConverge = new HashSet(StringComparer.Ordinal); foreach (var m in memberships) { @@ -231,6 +255,8 @@ await syncService.AddQueueMemberAsync( Math.Clamp(m.Penalty, 0, 10), allowedChannels: m.AllowedChannels, ct); _membershipsReconciled.Add(1, new KeyValuePair("tenant", tenantId)); + if (AgentPauseCoordinator.CreatesVoiceRow(m.AllowedChannels) && queuedForConverge.Add(agentIdValue)) + toConverge.Add(agent.AgentId); } catch (OperationCanceledException) when (ct.IsCancellationRequested) { @@ -244,6 +270,25 @@ await syncService.AddQueueMemberAsync( LogSyncFailure(tenantId, queue.Name, agent.AgentId.Value, ex); } } + + foreach (var agentId in toConverge) + { + try + { + await _pauseCoordinator.ConvergeAsync(tid, agentId, freshAgentStore, syncService, ct); + } + catch (OperationCanceledException) when (ct.IsCancellationRequested) + { + throw; + } + catch (Exception ex) + { + _syncFailures.Add(1, + new KeyValuePair("tenant", tenantId), + new KeyValuePair("operation", "converge_paused")); + LogConvergeFailure(tenantId, agentId.Value, ex); + } + } } [LoggerMessage(Level = LogLevel.Debug, @@ -254,6 +299,10 @@ await syncService.AddQueueMemberAsync( Message = "Realtime reconciliation failed for tenant {TenantId} queue {QueueName} agent {AgentId}.")] private partial void LogSyncFailure(string tenantId, string queueName, string agentId, Exception ex); + [LoggerMessage(Level = LogLevel.Warning, + Message = "Realtime reconciliation could not converge the paused state of agent {AgentId} in tenant {TenantId}.")] + private partial void LogConvergeFailure(string tenantId, string agentId, Exception ex); + [LoggerMessage(Level = LogLevel.Error, Message = "Realtime reconciliation cycle failed.")] private partial void LogReconcileCycleError(Exception ex); diff --git a/src/Verbara.Platform.Api/Services/RealtimeStateBridge.cs b/src/Verbara.Platform.Api/Services/RealtimeStateBridge.cs index 61b7def7..e9e6ca22 100644 --- a/src/Verbara.Platform.Api/Services/RealtimeStateBridge.cs +++ b/src/Verbara.Platform.Api/Services/RealtimeStateBridge.cs @@ -1,4 +1,3 @@ -using System.Collections.Concurrent; using Verbara.Platform.Core; using Verbara.Platform.Queues; using Verbara.Sdk.Ami.Actions; @@ -17,6 +16,8 @@ namespace Verbara.Platform.Api.Services; /// Asterisk AMI via a QueuePause action through the . /// /// Both operations are best-effort; failures are logged without interrupting the event stream. +/// Each write runs under the per-agent lock owned by , the same +/// lock the reconcile and membership-save convergence paths take. /// internal sealed partial class RealtimeStateBridge : IHostedService, IDisposable { @@ -32,19 +33,21 @@ internal sealed partial class RealtimeStateBridge : IHostedService, IDisposable private readonly VerbaraServerPool _serverPool; private readonly ResiliencePolicy _policy; private readonly ILogger _logger; - private readonly ConcurrentDictionary _agentLocks = new(); + private readonly AgentPauseCoordinator _pauseCoordinator; private IDisposable? _subscription; public RealtimeStateBridge( PlatformEventBus eventBus, IRealtimeSyncService syncService, VerbaraServerPool serverPool, + AgentPauseCoordinator pauseCoordinator, ILogger logger, [FromKeyedServices(ResiliencePolicyKey)] ResiliencePolicy? policy = null) { _eventBus = eventBus; _syncService = syncService; _serverPool = serverPool; + _pauseCoordinator = pauseCoordinator; _logger = logger; _policy = policy ?? ResiliencePolicy.NoOp; } @@ -85,18 +88,16 @@ internal Task HandleEventAsync(PlatformEvent evt) /// /// Propagates a paused/unpaused decision to the Asterisk Realtime DB and AMI - /// under a per-agent semaphore. Both side-effects are independent and best-effort: - /// a DB failure does NOT prevent the AMI QueuePause from being attempted (and vice - /// versa). Both share the resilience policy key so circuit state is aggregated at + /// under the per-agent lock from . Both side-effects + /// are independent and best-effort: a DB failure does NOT prevent the AMI QueuePause + /// from being attempted (and vice versa). Both share the resilience policy key so circuit state is aggregated at /// the bridge level. Shared by (drain/normal /// state flips) and (W4 deferred-pause /// set/cancel, where State stays routable). /// private async Task ApplyPauseAsync(string tenantId, string agentId, bool shouldPause, string reason, CancellationToken ct) { - var semaphore = _agentLocks.GetOrAdd(agentId, _ => new SemaphoreSlim(1, 1)); - await semaphore.WaitAsync(ct); - try + using (await _pauseCoordinator.AcquireAsync(agentId, ct)) { var iface = $"PJSIP/{tenantId}-agent-{agentId}"; @@ -151,10 +152,6 @@ await server.Connection.SendActionAsync( } } } - finally - { - semaphore.Release(); - } } public Task StopAsync(CancellationToken ct) diff --git a/src/Verbara.Platform.Api/Services/RealtimeSyncingQueueMembershipStore.cs b/src/Verbara.Platform.Api/Services/RealtimeSyncingQueueMembershipStore.cs index 39e682d4..8ecd3861 100644 --- a/src/Verbara.Platform.Api/Services/RealtimeSyncingQueueMembershipStore.cs +++ b/src/Verbara.Platform.Api/Services/RealtimeSyncingQueueMembershipStore.cs @@ -16,6 +16,10 @@ namespace Verbara.Platform.Api.Services; /// ) do NOT per-member sync — the queue/agent removal path already /// removes the parent Asterisk rows, and the reconciler re-converges, matching the pre-migration /// endpoints. Every sync is best-effort: a throw is swallowed + logged (EventId 4130). +/// After a voice upsert, converges the agent's paused value through +/// (fresh read of the undecorated agent store under the +/// per-agent lock shared with ), because the upsert inserts a new row +/// paused (queue-members-stay-unpaused-across-reconcile). /// /// /// R3 (decoration-cycle avoidance). needs @@ -30,6 +34,7 @@ internal sealed class RealtimeSyncingQueueMembershipStore : IQueueMembershipStor private readonly IQueueStore _queues; private readonly IAgentStore _agents; private readonly IRealtimeSyncService _sync; + private readonly AgentPauseCoordinator _pauseCoordinator; private readonly ILogger _logger; public RealtimeSyncingQueueMembershipStore( @@ -37,18 +42,21 @@ public RealtimeSyncingQueueMembershipStore( IQueueStore queues, IAgentStore agents, IRealtimeSyncService sync, + AgentPauseCoordinator pauseCoordinator, ILogger logger) { ArgumentNullException.ThrowIfNull(inner); ArgumentNullException.ThrowIfNull(queues); ArgumentNullException.ThrowIfNull(agents); ArgumentNullException.ThrowIfNull(sync); + ArgumentNullException.ThrowIfNull(pauseCoordinator); ArgumentNullException.ThrowIfNull(logger); _inner = inner; _queues = queues; _agents = agents; _sync = sync; + _pauseCoordinator = pauseCoordinator; _logger = logger; } @@ -86,6 +94,13 @@ await _sync.AddQueueMemberAsync( membership.TenantId.Value, queue.Name, agent.AgentId.Value, agent.DisplayName, Math.Clamp(membership.Penalty, 0, 10), allowedChannels: membership.AllowedChannels, ct).ConfigureAwait(false); + + // The upsert inserts a new row with paused = 1: re-assert the agent's real state. + if (AgentPauseCoordinator.CreatesVoiceRow(membership.AllowedChannels)) + { + await _pauseCoordinator.ConvergeAsync( + membership.TenantId, membership.AgentId, _agents, _sync, ct).ConfigureAwait(false); + } } catch (Exception ex) { diff --git a/tests/Verbara.Platform.Api.Tests/AgentPauseCoordinatorTests.cs b/tests/Verbara.Platform.Api.Tests/AgentPauseCoordinatorTests.cs new file mode 100644 index 00000000..d7fa4bd5 --- /dev/null +++ b/tests/Verbara.Platform.Api.Tests/AgentPauseCoordinatorTests.cs @@ -0,0 +1,148 @@ +using Verbara.Platform.Api.Services; +using Verbara.Platform.Core; +using Verbara.Platform.Queues; +using Verbara.Platform.Storage.InMemory; +using Verbara.Sdk.Pro.Realtime; +using NSubstitute; + +namespace Verbara.Platform.Api.Tests; + +/// +/// queue-members-stay-unpaused-across-reconcile — the shared pause rule, voice gate and per-agent +/// lock of (design D2). +/// +public sealed class AgentPauseCoordinatorTests +{ + private const string Tenant = "t-coord"; + + [Theory] + [InlineData(AgentState.Available, false)] + [InlineData(AgentState.Busy, false)] + [InlineData(AgentState.Break, true)] + [InlineData(AgentState.Lunch, true)] + [InlineData(AgentState.Training, true)] + [InlineData(AgentState.ACW, true)] + [InlineData(AgentState.DND, true)] + [InlineData(AgentState.Offline, true)] + public async Task ConvergeAsync_ShouldWriteRulePausedValue_ForEachState(AgentState state, bool expectedPaused) + { + var (agents, agent) = await SeedAsync(state); + var sync = Substitute.For(); + + var written = await new AgentPauseCoordinator().ConvergeAsync( + agent.TenantId, agent.AgentId, agents, sync, CancellationToken.None); + + written.Should().Be(expectedPaused); + await sync.Received(1).SyncAgentPausedAsync(Tenant, agent.AgentId.Value, expectedPaused, Arg.Any()); + } + + [Fact] + public async Task ConvergeAsync_ShouldPause_WhenAvailableAgentHasPendingPause() + { + var (agents, agent) = await SeedAsync(AgentState.Available, pending: AgentState.Break); + var sync = Substitute.For(); + + var written = await new AgentPauseCoordinator().ConvergeAsync( + agent.TenantId, agent.AgentId, agents, sync, CancellationToken.None); + + written.Should().BeTrue(); + await sync.Received(1).SyncAgentPausedAsync(Tenant, agent.AgentId.Value, true, Arg.Any()); + } + + [Fact] + public async Task ConvergeAsync_ShouldWriteNothing_WhenAgentIsMissing() + { + var sync = Substitute.For(); + + var written = await new AgentPauseCoordinator().ConvergeAsync( + new TenantId(Tenant), EntityId.New(), new InMemoryAgentStore(), sync, CancellationToken.None); + + written.Should().BeNull(); + await sync.DidNotReceiveWithAnyArgs().SyncAgentPausedAsync(default!, default!, default, default); + } + + [Fact] + public async Task ConvergeAsync_ShouldReadTheAgentOnlyAfterTheLockIsFree() + { + // The read happens UNDER the lock: a writer holding it (the bridge) finishes first, so the + // convergence sees the state that writer published, never a value read before it. + var (agents, agent) = await SeedAsync(AgentState.Break); + var sync = Substitute.For(); + var sut = new AgentPauseCoordinator(); + + var held = await sut.AcquireAsync(agent.AgentId.Value, CancellationToken.None); + var converge = sut.ConvergeAsync(agent.TenantId, agent.AgentId, agents, sync, CancellationToken.None); + // No wall-clock wait: a contended SemaphoreSlim.WaitAsync returns an incomplete task synchronously. + converge.IsCompleted.Should().BeFalse("the convergence must wait for the agent's lock"); + + // The holder's state change lands before release, as a NEW row (a read taken before the + // lock would still hold the old Break instance). + await agents.SaveAsync(CopyWithState(agent, AgentState.Available), CancellationToken.None); + held.Dispose(); + + (await converge.WaitAsync(TimeSpan.FromSeconds(5))).Should().BeFalse(); + await sync.Received(1).SyncAgentPausedAsync(Tenant, agent.AgentId.Value, false, Arg.Any()); + } + + [Fact] + public async Task AcquireAsync_ShouldNotBlockOtherAgents() + { + var sut = new AgentPauseCoordinator(); + + using var held = await sut.AcquireAsync("a1", CancellationToken.None); + var other = sut.AcquireAsync("a2", CancellationToken.None); + + (await other.WaitAsync(TimeSpan.FromSeconds(5))).Dispose(); + } + + [Fact] + public async Task AcquireAsync_ShouldReleaseOnce_WhenHandleIsDisposedTwice() + { + var sut = new AgentPauseCoordinator(); + var first = await sut.AcquireAsync("a1", CancellationToken.None); + first.Dispose(); + first.Dispose(); + + using var second = await sut.AcquireAsync("a1", CancellationToken.None); + var third = sut.AcquireAsync("a1", CancellationToken.None); + // No wall-clock wait: a contended SemaphoreSlim.WaitAsync returns an incomplete task synchronously. + third.IsCompleted.Should().BeFalse("a double dispose must not over-release the semaphore"); + } + + [Theory] + [InlineData(null, true)] + [InlineData(new[] { "voice" }, true)] + [InlineData(new[] { "WebChat", "Voice" }, true)] + [InlineData(new[] { "WebChat" }, false)] + [InlineData(new string[0], false)] + public void CreatesVoiceRow_ShouldMirrorTheSdkVoiceGate(string[]? channels, bool expected) + => AgentPauseCoordinator.CreatesVoiceRow(channels).Should().Be(expected); + + private static Agent CopyWithState(Agent agent, AgentState state) => new() + { + AgentId = agent.AgentId, + TenantId = agent.TenantId, + UserId = agent.UserId, + DisplayName = agent.DisplayName, + State = state, + CreatedAt = agent.CreatedAt, + }; + + private static async Task<(InMemoryAgentStore Agents, Agent Agent)> SeedAsync( + AgentState state, AgentState? pending = null) + { + var agents = new InMemoryAgentStore(); + var agent = new Agent + { + AgentId = EntityId.New(), + TenantId = new TenantId(Tenant), + UserId = EntityId.New(), + DisplayName = "Coordinator Agent", + State = state, + PendingState = pending, + CreatedAt = DateTimeOffset.UtcNow, + }; + await agents.SaveAsync(agent, CancellationToken.None); + return (agents, agent); + } +} diff --git a/tests/Verbara.Platform.Api.Tests/RealtimeReconciliationServiceTests.cs b/tests/Verbara.Platform.Api.Tests/RealtimeReconciliationServiceTests.cs index 65b1b296..40b6f245 100644 --- a/tests/Verbara.Platform.Api.Tests/RealtimeReconciliationServiceTests.cs +++ b/tests/Verbara.Platform.Api.Tests/RealtimeReconciliationServiceTests.cs @@ -100,20 +100,216 @@ public async Task ReconcileAsync_ShouldNotThrow_WhenSyncServiceUnavailable() var sp = services.BuildServiceProvider(); var sut = new RealtimeReconciliationService( - sp, heartbeat, options, + sp, heartbeat, options, new AgentPauseCoordinator(), NullLogger.Instance); var act = async () => await sut.ReconcileAsync(CancellationToken.None); await act.Should().NotThrowAsync(); } + [Fact] + public async Task ReconcileTenantAsync_ShouldSyncAgentUnpaused_WhenAgentIsAvailable() + { + // Regression (queue-members-stay-unpaused-across-reconcile): AddQueueMemberAsync + // writes paused=1 on insert, so after an agent's upserts the tick must re-assert + // the agent's real pause state. An Available agent with no pending pause ends unpaused. + var harness = BuildHarness(out var sync); + var agent = MakeAgent("agent-available"); + var queue = MakeQueue(EntityId.From("q-1")); + harness.SeedAgent(agent); + harness.SeedQueue(queue); + harness.SeedMembership(MakeMembership(agent.AgentId, queue.QueueId)); + + await harness.Sut.ReconcileAsync(CancellationToken.None); + + // The Add (paused=1 on insert) must be followed by the convergence write. + RealtimeSyncCallLog.Of(sync).Should().Equal("AddQueueMemberAsync", "SyncAgentPausedAsync:False"); + await sync.Received(1).SyncAgentPausedAsync(TestTenantId, "agent-available", false, Arg.Any()); + } + + [Theory] + [InlineData(AgentState.Break, true)] + [InlineData(AgentState.ACW, true)] + [InlineData(AgentState.Offline, true)] + [InlineData(AgentState.Busy, false)] + public async Task ReconcileTenantAsync_ShouldSyncRulePausedValue_ForAgentState(AgentState state, bool expectedPaused) + { + var harness = BuildHarness(out var sync); + var agent = MakeAgent("agent-state"); + agent.State = state; + var queue = MakeQueue(EntityId.From("q-1")); + harness.SeedAgent(agent); + harness.SeedQueue(queue); + harness.SeedMembership(MakeMembership(agent.AgentId, queue.QueueId)); + + await harness.Sut.ReconcileAsync(CancellationToken.None); + + RealtimeSyncCallLog.Of(sync).Should().Equal("AddQueueMemberAsync", $"SyncAgentPausedAsync:{expectedPaused}"); + } + + [Fact] + public async Task ReconcileTenantAsync_ShouldSyncAgentPaused_WhenAvailableAgentHasPendingPause() + { + var harness = BuildHarness(out var sync); + var agent = MakeAgent("agent-pending"); + agent.PendingState = AgentState.Break; + var queue = MakeQueue(EntityId.From("q-1")); + harness.SeedAgent(agent); + harness.SeedQueue(queue); + harness.SeedMembership(MakeMembership(agent.AgentId, queue.QueueId)); + + await harness.Sut.ReconcileAsync(CancellationToken.None); + + RealtimeSyncCallLog.Of(sync).Should().Equal("AddQueueMemberAsync", "SyncAgentPausedAsync:True"); + } + + [Fact] + public async Task ReconcileTenantAsync_ShouldConvergeOncePerAgent_AfterItsLastUpsert() + { + // SyncAgentPausedAsync sets every row of the agent's interface, so one write per agent per + // pass, after the agent's last AddQueueMemberAsync, covers all its queues. + var harness = BuildHarness(out var sync); + var agent = MakeAgent("agent-two-queues"); + var q1 = MakeQueue(EntityId.From("q-1")); + var q2 = MakeQueue(EntityId.From("q-2")); + harness.SeedAgent(agent); + harness.SeedQueue(q1); + harness.SeedQueue(q2); + harness.SeedMembership(MakeMembership(agent.AgentId, q1.QueueId)); + harness.SeedMembership(MakeMembership(agent.AgentId, q2.QueueId)); + + await harness.Sut.ReconcileAsync(CancellationToken.None); + + RealtimeSyncCallLog.Of(sync).Should().Equal( + "AddQueueMemberAsync", "AddQueueMemberAsync", "SyncAgentPausedAsync:False"); + } + + [Fact] + public async Task ReconcileTenantAsync_ShouldNotConverge_WhenAgentOnlyHasExcludedMemberships() + { + var harness = BuildHarness(out var sync); + var agent = MakeAgent("agent-excluded-only"); + var queue = MakeQueue(EntityId.From("q-1")); + harness.SeedAgent(agent); + harness.SeedQueue(queue); + harness.SeedMembership(MakeMembership(agent.AgentId, queue.QueueId, isExcluded: true)); + + await harness.Sut.ReconcileAsync(CancellationToken.None); + + await sync.DidNotReceiveWithAnyArgs().SyncAgentPausedAsync(default!, default!, default, default); + } + + [Fact] + public async Task ReconcileTenantAsync_ShouldNotConverge_WhenMembershipIsDigitalOnly() + { + // Voice gate: a digital-only membership has no queue_members row (the SDK removes it). + var harness = BuildHarness(out var sync); + var agent = MakeAgent("agent-digital"); + var queue = MakeQueue(EntityId.From("q-1")); + harness.SeedAgent(agent); + harness.SeedQueue(queue); + harness.SeedMembership(MakeMembership(agent.AgentId, queue.QueueId, allowedChannels: ["WebChat"])); + + await harness.Sut.ReconcileAsync(CancellationToken.None); + + await sync.DidNotReceiveWithAnyArgs().SyncAgentPausedAsync(default!, default!, default, default); + } + + [Fact] + public async Task ReconcileTenantAsync_ShouldNotConverge_WhenUpsertThrows() + { + var harness = BuildHarness(out var sync); +#pragma warning disable CA2012 // ValueTask used in NSubstitute mock setup + sync.AddQueueMemberAsync(default!, default!, default!, default!, default, default, default) + .ReturnsForAnyArgs(_ => new ValueTask(Task.FromException(new InvalidOperationException("boom")))); +#pragma warning restore CA2012 + var agent = MakeAgent("agent-add-fails"); + var queue = MakeQueue(EntityId.From("q-1")); + harness.SeedAgent(agent); + harness.SeedQueue(queue); + harness.SeedMembership(MakeMembership(agent.AgentId, queue.QueueId)); + + await harness.Sut.ReconcileAsync(CancellationToken.None); + + await sync.DidNotReceiveWithAnyArgs().SyncAgentPausedAsync(default!, default!, default, default); + } + + [Fact] + public async Task ReconcileTenantAsync_ShouldWriteFreshState_WhenAgentChangesStateDuringThePass() + { + // A pass that started while the agent was Break; the agent becomes Available before the + // convergence. The convergence re-reads the agent, so the newer state wins over the tick cache. + var harness = BuildHarness(out var sync); + var agent = MakeAgent("agent-racing"); + agent.State = AgentState.Break; +#pragma warning disable CA2012 // ValueTask used in NSubstitute mock setup + sync.AddQueueMemberAsync(default!, default!, default!, default!, default, default, default) + .ReturnsForAnyArgs(_ => + { + harness.ReplaceAgent(MakeAgent("agent-racing")); // fresh row, State = Available + return ValueTask.CompletedTask; + }); +#pragma warning restore CA2012 + var queue = MakeQueue(EntityId.From("q-1")); + harness.SeedAgent(agent); + harness.SeedQueue(queue); + harness.SeedMembership(MakeMembership(agent.AgentId, queue.QueueId)); + + await harness.Sut.ReconcileAsync(CancellationToken.None); + + RealtimeSyncCallLog.Of(sync).Should().Equal("AddQueueMemberAsync", "SyncAgentPausedAsync:False"); + } + + [Fact] + public async Task ReconcileTenantAsync_ShouldReadTheUndecoratedAgentStore_WhenItIsRegistered() + { + var undecorated = Substitute.For(); + var harness = BuildHarness(out var sync, undecorated); + var agent = MakeAgent("agent-inner"); + var innerView = MakeAgent("agent-inner"); + innerView.State = AgentState.DND; + undecorated.GetByIdAsync(Arg.Any(), agent.AgentId, Arg.Any()) + .Returns(Task.FromResult(innerView)); + var queue = MakeQueue(EntityId.From("q-1")); + harness.SeedAgent(agent); + harness.SeedQueue(queue); + harness.SeedMembership(MakeMembership(agent.AgentId, queue.QueueId)); + + await harness.Sut.ReconcileAsync(CancellationToken.None); + + RealtimeSyncCallLog.Of(sync).Should().Equal("AddQueueMemberAsync", "SyncAgentPausedAsync:True"); + } + + [Fact] + public async Task ReconcileTenantAsync_ShouldKeepConvergingOtherAgents_WhenOneConvergenceThrows() + { + var harness = BuildHarness(out var sync); +#pragma warning disable CA2012 // ValueTask used in NSubstitute mock setup + sync.SyncAgentPausedAsync(TestTenantId, "agent-a", Arg.Any(), Arg.Any()) + .Returns(_ => new ValueTask(Task.FromException(new InvalidOperationException("db down")))); +#pragma warning restore CA2012 + var a = MakeAgent("agent-a"); + var b = MakeAgent("agent-b"); + var queue = MakeQueue(EntityId.From("q-1")); + harness.SeedAgent(a); + harness.SeedAgent(b); + harness.SeedQueue(queue); + harness.SeedMembership(MakeMembership(a.AgentId, queue.QueueId)); + harness.SeedMembership(MakeMembership(b.AgentId, queue.QueueId)); + + var act = async () => await harness.Sut.ReconcileAsync(CancellationToken.None); + + await act.Should().NotThrowAsync(); + await sync.Received(1).SyncAgentPausedAsync(TestTenantId, "agent-b", false, Arg.Any()); + } + // ── Test harness ────────────────────────────────────────────────────── - private static Harness BuildHarness(out IRealtimeSyncService sync) + private static Harness BuildHarness(out IRealtimeSyncService sync, IAgentStore? undecoratedAgentStore = null) { sync = Substitute.For(); sync.Events.Returns(System.Reactive.Linq.Observable.Empty()); - return new Harness(sync); + return new Harness(sync, undecoratedAgentStore); } private sealed class Harness @@ -128,7 +324,9 @@ private sealed class Harness public RealtimeReconciliationService Sut { get; } - public Harness(IRealtimeSyncService sync) + public AgentPauseCoordinator Coordinator { get; } = new(); + + public Harness(IRealtimeSyncService sync, IAgentStore? undecoratedAgentStore = null) { _tenantStore.GetAllActiveAsync(Arg.Any()) .Returns(new[] { new Tenant { TenantId = TestTenantId, Name = "Recon", Status = TenantStatus.Active } }); @@ -154,15 +352,27 @@ public Harness(IRealtimeSyncService sync) services.AddSingleton(_queueStore); services.AddSingleton(_agentStore); services.AddSingleton(sync); + if (undecoratedAgentStore is not null) + { + services.AddKeyedSingleton( + Verbara.Platform.Api.DependencyInjection.RealtimeSyncingStoresExtensions.AgentStoreInner, + undecoratedAgentStore); + } Sut = new RealtimeReconciliationService( services.BuildServiceProvider(), new ServiceHeartbeat(), Options.Create(new RealtimeOptions { ReconcilerIntervalSeconds = 60 }), + Coordinator, NullLogger.Instance); } public void SeedAgent(Agent a) => _agents.Add(a); + public void ReplaceAgent(Agent a) + { + _agents.RemoveAll(x => x.AgentId == a.AgentId); + _agents.Add(a); + } public void SeedQueue(Queue q) => _queues.Add(q); public void SeedMembership(QueueMembership m) => _memberships.Add(m); } diff --git a/tests/Verbara.Platform.Api.Tests/RealtimeStateBridgeTests.cs b/tests/Verbara.Platform.Api.Tests/RealtimeStateBridgeTests.cs index a20f4a3f..a3bba49e 100644 --- a/tests/Verbara.Platform.Api.Tests/RealtimeStateBridgeTests.cs +++ b/tests/Verbara.Platform.Api.Tests/RealtimeStateBridgeTests.cs @@ -25,6 +25,7 @@ public sealed class RealtimeStateBridgeTests : IDisposable private readonly IRealtimeSyncService _syncService; private readonly IAmiConnection _ami; private readonly VerbaraServerPool _serverPool; + private readonly AgentPauseCoordinator _pauseCoordinator = new(); public RealtimeStateBridgeTests() { @@ -45,6 +46,7 @@ private RealtimeStateBridge CreateBridge() _eventBus, _syncService, _serverPool, + _pauseCoordinator, NullLogger.Instance); } @@ -267,6 +269,53 @@ await _ami.Received(1).SendActionAsync( Arg.Is(a => a != null && a.Paused == false && a.Reason == "Available")); } + // ─── Shared per-agent lock (queue-members-stay-unpaused-across-reconcile) ── + + [Fact] + public async Task HandleEventAsync_ShouldWaitForConvergence_WhenCoordinatorLockIsHeldForSameAgent() + { + // The bridge and AgentPauseCoordinator.ConvergeAsync (reconciler + membership save) take + // ONE per-agent lock, so a bridge write never interleaves with a convergence read-then-write. + var agents = new Verbara.Platform.Storage.InMemory.InMemoryAgentStore(); + var agent = new Agent + { + AgentId = EntityId.From("a1"), + TenantId = new TenantId("t1"), + UserId = EntityId.New(), + DisplayName = "Agent One", + State = AgentState.Break, + CreatedAt = DateTimeOffset.UtcNow, + }; + await agents.SaveAsync(agent, CancellationToken.None); + var convergeEntered = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + var releaseConverge = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); +#pragma warning disable CA2012 // ValueTask used in NSubstitute mock setup + _syncService.SyncAgentPausedAsync("t1", "a1", true, Arg.Any()) + .Returns(_ => + { + convergeEntered.TrySetResult(); + return new ValueTask(releaseConverge.Task); + }); +#pragma warning restore CA2012 + AddPrimaryServer(); + var bridge = CreateBridge(); + + var converge = _pauseCoordinator.ConvergeAsync( + new TenantId("t1"), EntityId.From("a1"), agents, _syncService, CancellationToken.None); + await convergeEntered.Task.WaitAsync(TimeSpan.FromSeconds(5)); + var bridgeWrite = bridge.HandleEventAsync(MakeEvent("Available")); + // No wall-clock wait: a contended SemaphoreSlim.WaitAsync returns an incomplete task synchronously. + bridgeWrite.IsCompleted.Should().BeFalse("the bridge must wait for the convergence holding the lock"); + await _syncService.DidNotReceive().SyncAgentPausedAsync("t1", "a1", false, Arg.Any()); + + releaseConverge.SetResult(); + await converge.WaitAsync(TimeSpan.FromSeconds(5)); + await bridgeWrite.WaitAsync(TimeSpan.FromSeconds(5)); + + RealtimeSyncCallLog.Of(_syncService).Should().Equal( + "SyncAgentPausedAsync:True", "SyncAgentPausedAsync:False"); + } + public void Dispose() { _subject.Dispose(); diff --git a/tests/Verbara.Platform.Api.Tests/RealtimeSyncCallLog.cs b/tests/Verbara.Platform.Api.Tests/RealtimeSyncCallLog.cs new file mode 100644 index 00000000..e7e703bf --- /dev/null +++ b/tests/Verbara.Platform.Api.Tests/RealtimeSyncCallLog.cs @@ -0,0 +1,23 @@ +using NSubstitute; +using Verbara.Sdk.Pro.Realtime; + +namespace Verbara.Platform.Api.Tests; + +/// +/// Ordered log of the queue-member writes a substituted received: +/// AddQueueMemberAsync and SyncAgentPausedAsync:{paused}. Lets a test assert that the +/// paused convergence write came after the upsert without Received.InOrder, which CA2012 rejects +/// for ValueTask-returning members. +/// +internal static class RealtimeSyncCallLog +{ + public static IReadOnlyList Of(IRealtimeSyncService sync) => + sync.ReceivedCalls() + .Select(c => (Name: c.GetMethodInfo().Name, Args: c.GetArguments())) + .Where(c => c.Name is nameof(IRealtimeSyncService.AddQueueMemberAsync) + or nameof(IRealtimeSyncService.SyncAgentPausedAsync)) + .Select(c => c.Name == nameof(IRealtimeSyncService.SyncAgentPausedAsync) + ? $"{c.Name}:{c.Args[2]}" + : c.Name) + .ToList(); +} diff --git a/tests/Verbara.Platform.Api.Tests/RealtimeSyncingStoreDecoratorTests.cs b/tests/Verbara.Platform.Api.Tests/RealtimeSyncingStoreDecoratorTests.cs index 4d6270bd..4a5aa383 100644 --- a/tests/Verbara.Platform.Api.Tests/RealtimeSyncingStoreDecoratorTests.cs +++ b/tests/Verbara.Platform.Api.Tests/RealtimeSyncingStoreDecoratorTests.cs @@ -264,6 +264,74 @@ public async Task MembershipBulkDelete_ShouldNotPerMemberSync() await sync.DidNotReceiveWithAnyArgs().RemoveQueueMemberAsync(default!, default!, default!, default); } + [Fact] + public async Task SaveAsync_ShouldSyncAgentUnpaused_WhenAgentIsAvailable() + { + // Regression (queue-members-stay-unpaused-across-reconcile): AddQueueMemberAsync + // inserts paused=1, so a save must converge the Available agent back to unpaused. + var (sut, _, sync, queue, agent) = BuildMembershipSut(out _); + + await sut.SaveAsync(MakeMembership(agent.AgentId, queue.QueueId), CancellationToken.None); + + // The Add (paused=1 on insert) must be followed by the convergence write. + RealtimeSyncCallLog.Of(sync).Should().Equal("AddQueueMemberAsync", "SyncAgentPausedAsync:False"); + await sync.Received(1).SyncAgentPausedAsync(Tenant, agent.AgentId.Value, false, Arg.Any()); + } + + [Fact] + public async Task SaveAsync_ShouldSyncAgentPaused_WhenAgentIsOnBreak() + { + var (sut, _, sync, queue, agent) = BuildMembershipSut(out _); + agent.State = AgentState.Break; + + await sut.SaveAsync(MakeMembership(agent.AgentId, queue.QueueId), CancellationToken.None); + + RealtimeSyncCallLog.Of(sync).Should().Equal("AddQueueMemberAsync", "SyncAgentPausedAsync:True"); + } + + [Fact] + public async Task SaveAsync_ShouldNotConverge_WhenMembershipIsDigitalOnly() + { + var (sut, _, sync, queue, agent) = BuildMembershipSut(out _); + + await sut.SaveAsync( + MakeMembership(agent.AgentId, queue.QueueId, allowedChannels: ["WebChat"]), CancellationToken.None); + + await sync.DidNotReceiveWithAnyArgs().SyncAgentPausedAsync(default!, default!, default, default); + } + + [Fact] + public async Task SaveAsync_ShouldNotConverge_WhenUpsertThrows() + { + var (sut, _, sync, queue, agent) = BuildMembershipSut(out _); +#pragma warning disable CA2012 // ValueTask used in NSubstitute mock setup + sync.AddQueueMemberAsync(default!, default!, default!, default!, default, default, default) + .ReturnsForAnyArgs(_ => Faulted()); +#pragma warning restore CA2012 + + await sut.SaveAsync(MakeMembership(agent.AgentId, queue.QueueId), CancellationToken.None); + + await sync.DidNotReceiveWithAnyArgs().SyncAgentPausedAsync(default!, default!, default, default); + } + + [Fact] + public async Task SaveAsync_ShouldSwallowAndLog_WhenConvergenceThrows() + { + var (sut, membershipInner, sync, queue, agent) = BuildMembershipSut(out var logger); +#pragma warning disable CA2012 // ValueTask used in NSubstitute mock setup + sync.SyncAgentPausedAsync(default!, default!, default, default) + .ReturnsForAnyArgs(_ => Faulted()); +#pragma warning restore CA2012 + var membership = MakeMembership(agent.AgentId, queue.QueueId); + + var act = async () => await sut.SaveAsync(membership, CancellationToken.None); + + await act.Should().NotThrowAsync("the convergence is best-effort like the upsert"); + (await membershipInner.GetAsync(membership.TenantId, queue.QueueId, agent.AgentId, CancellationToken.None)) + .Should().NotBeNull(); + logger.Warned(DeferralEventId).Should().BeTrue(); + } + // ─── Helpers ───────────────────────────────────────────────────────────── /// A faulted that throws when awaited — models a fully @@ -287,7 +355,7 @@ private static (RealtimeSyncingQueueMembershipStore Sut, IQueueMembershipStore M agentInner.SaveAsync(agent, CancellationToken.None).GetAwaiter().GetResult(); var sut = new RealtimeSyncingQueueMembershipStore( - membershipInner, queueInner, agentInner, sync, logger); + membershipInner, queueInner, agentInner, sync, new AgentPauseCoordinator(), logger); return (sut, membershipInner, sync, queue, agent); } @@ -313,7 +381,8 @@ private static (RealtimeSyncingQueueMembershipStore Sut, IQueueMembershipStore M CreatedAt = DateTimeOffset.UtcNow, }; - private static QueueMembership MakeMembership(EntityId agentId, EntityId queueId, int penalty = 0) => new() + private static QueueMembership MakeMembership( + EntityId agentId, EntityId queueId, int penalty = 0, IReadOnlyList? allowedChannels = null) => new() { TenantId = new TenantId(Tenant), QueueId = queueId, @@ -322,7 +391,7 @@ private static (RealtimeSyncingQueueMembershipStore Sut, IQueueMembershipStore M Source = MembershipSource.Manual, IsExcluded = false, CreatedAt = DateTimeOffset.UtcNow, - AllowedChannels = null, + AllowedChannels = allowedChannels, }; /// Minimal capturing logger — records the (level, eventId) of every log call so a test diff --git a/tests/Verbara.Platform.Api.Tests/RealtimeSyncingStoresRegistrarTests.cs b/tests/Verbara.Platform.Api.Tests/RealtimeSyncingStoresRegistrarTests.cs index 11ce0f5d..fb5e6268 100644 --- a/tests/Verbara.Platform.Api.Tests/RealtimeSyncingStoresRegistrarTests.cs +++ b/tests/Verbara.Platform.Api.Tests/RealtimeSyncingStoresRegistrarTests.cs @@ -54,6 +54,48 @@ public void Resolve_ShouldExposeUndecoratedInner_ViaKey() .Should().BeOfType(); } + [Fact] + public async Task MembershipDecorator_ShouldConvergeUnderTheSingletonPauseCoordinator() + { + // queue-members-stay-unpaused-across-reconcile — the membership decorator must take the SAME + // per-agent lock as RealtimeStateBridge and the reconciler, i.e. the DI singleton. + var provider = BuildProvider(withRealtime: true); + var coordinator = provider.GetRequiredService(); + provider.GetRequiredService().Should().BeSameAs(coordinator); + var tenant = new Verbara.Platform.Core.TenantId("t-reg"); + var queue = new Queue + { + QueueId = Verbara.Platform.Core.EntityId.New(), TenantId = tenant, Name = "q-reg", + IsActive = true, CreatedAt = DateTimeOffset.UtcNow, + }; + var agent = new Agent + { + AgentId = Verbara.Platform.Core.EntityId.New(), TenantId = tenant, UserId = Verbara.Platform.Core.EntityId.New(), + DisplayName = "Registrar Agent", State = AgentState.Available, CreatedAt = DateTimeOffset.UtcNow, + }; + await provider.GetRequiredKeyedService(RealtimeSyncingStoresExtensions.QueueStoreInner) + .SaveAsync(queue, CancellationToken.None); + await provider.GetRequiredKeyedService(RealtimeSyncingStoresExtensions.AgentStoreInner) + .SaveAsync(agent, CancellationToken.None); + var membership = new QueueMembership + { + TenantId = tenant, QueueId = queue.QueueId, AgentId = agent.AgentId, + Source = MembershipSource.Manual, CreatedAt = DateTimeOffset.UtcNow, + }; + + Task save; + using (await coordinator.AcquireAsync(agent.AgentId.Value, CancellationToken.None)) + { + save = provider.GetRequiredService().SaveAsync(membership, CancellationToken.None); + // No wall-clock wait: a contended SemaphoreSlim.WaitAsync returns an incomplete task synchronously. + save.IsCompleted.Should().BeFalse("the decorator's convergence must wait on the singleton's lock"); + } + + await save.WaitAsync(TimeSpan.FromSeconds(5)); + await provider.GetRequiredService().Received(1) + .SyncAgentPausedAsync("t-reg", agent.AgentId.Value, false, Arg.Any()); + } + [Fact] public void AddRealtimeSyncingStores_ShouldThrow_WhenStorageNotRegisteredFirst() {