fix(realtime): keep an Available agent unpaused across reconcile and membership saves

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

Directory.Packages.props

@@ -57,28 +57,28 @@
   </ItemGroup>
 
   <ItemGroup>
-    <PackageVersion Include="Verbara.Sdk.Pro.EventStore" Version="2.17.0-pro" />
-    <PackageVersion Include="Verbara.Sdk.Pro.Analytics" Version="2.17.0-pro" />
-    <PackageVersion Include="Verbara.Sdk.Pro.CallAnalytics" Version="2.17.0-pro" />
-    <PackageVersion Include="Verbara.Sdk.Pro.Dialer" Version="2.17.0-pro" />
-    <PackageVersion Include="Verbara.Sdk.Pro.Dialer.Storage.Postgres" Version="2.17.0-pro" />
-    <PackageVersion Include="Verbara.Sdk.Pro.EventStore.Postgres" Version="2.17.0-pro" />
-    <PackageVersion Include="Verbara.Sdk.Pro.CallAnalytics.Storage.Postgres" Version="2.17.0-pro" />
-    <PackageVersion Include="Verbara.Sdk.Pro.Analytics.Storage.Postgres" Version="2.17.0-pro" />
-    <PackageVersion Include="Verbara.Sdk.Pro.Licensing" Version="2.17.0-pro" />
-    <PackageVersion Include="Verbara.Sdk.Pro.CsatRunner" Version="2.17.0-pro" />
-    <PackageVersion Include="Verbara.Sdk.Pro.AgentAssist" Version="2.17.0-pro" />
-    <PackageVersion Include="Verbara.Sdk.Pro.AgentAssist.Storage.Postgres" Version="2.17.0-pro" />
-    <PackageVersion Include="Verbara.Sdk.Pro.Routing" Version="2.17.0-pro" />
-    <PackageVersion Include="Verbara.Sdk.Pro.Realtime" Version="2.17.0-pro" />
-    <PackageVersion Include="Verbara.Sdk.Pro.Realtime.Storage.Postgres" Version="2.17.0-pro" />
-    <PackageVersion Include="Verbara.Sdk.Pro.Cluster" Version="2.17.0-pro" />
-    <PackageVersion Include="Verbara.Sdk.Pro.Cluster.Storage.Postgres" Version="2.17.0-pro" />
-    <PackageVersion Include="Verbara.Sdk.Pro.MultiTenant" Version="2.17.0-pro" />
-    <PackageVersion Include="Verbara.Sdk.Pro.Push" Version="2.17.0-pro" />
-    <PackageVersion Include="Verbara.Sdk.Pro.Push.SignalR" Version="2.17.0-pro" />
-    <PackageVersion Include="Verbara.Sdk.Pro.Storage.Common" Version="2.17.0-pro" />
-    <PackageVersion Include="Verbara.Sdk.Pro.OpenTelemetry" Version="2.17.0-pro" />
+    <PackageVersion Include="Verbara.Sdk.Pro.EventStore" Version="2.17.1-pro" />
+    <PackageVersion Include="Verbara.Sdk.Pro.Analytics" Version="2.17.1-pro" />
+    <PackageVersion Include="Verbara.Sdk.Pro.CallAnalytics" Version="2.17.1-pro" />
+    <PackageVersion Include="Verbara.Sdk.Pro.Dialer" Version="2.17.1-pro" />
+    <PackageVersion Include="Verbara.Sdk.Pro.Dialer.Storage.Postgres" Version="2.17.1-pro" />
+    <PackageVersion Include="Verbara.Sdk.Pro.EventStore.Postgres" Version="2.17.1-pro" />
+    <PackageVersion Include="Verbara.Sdk.Pro.CallAnalytics.Storage.Postgres" Version="2.17.1-pro" />
+    <PackageVersion Include="Verbara.Sdk.Pro.Analytics.Storage.Postgres" Version="2.17.1-pro" />
+    <PackageVersion Include="Verbara.Sdk.Pro.Licensing" Version="2.17.1-pro" />
+    <PackageVersion Include="Verbara.Sdk.Pro.CsatRunner" Version="2.17.1-pro" />
+    <PackageVersion Include="Verbara.Sdk.Pro.AgentAssist" Version="2.17.1-pro" />
+    <PackageVersion Include="Verbara.Sdk.Pro.AgentAssist.Storage.Postgres" Version="2.17.1-pro" />
+    <PackageVersion Include="Verbara.Sdk.Pro.Routing" Version="2.17.1-pro" />
+    <PackageVersion Include="Verbara.Sdk.Pro.Realtime" Version="2.17.1-pro" />
+    <PackageVersion Include="Verbara.Sdk.Pro.Realtime.Storage.Postgres" Version="2.17.1-pro" />
+    <PackageVersion Include="Verbara.Sdk.Pro.Cluster" Version="2.17.1-pro" />
+    <PackageVersion Include="Verbara.Sdk.Pro.Cluster.Storage.Postgres" Version="2.17.1-pro" />
+    <PackageVersion Include="Verbara.Sdk.Pro.MultiTenant" Version="2.17.1-pro" />
+    <PackageVersion Include="Verbara.Sdk.Pro.Push" Version="2.17.1-pro" />
+    <PackageVersion Include="Verbara.Sdk.Pro.Push.SignalR" Version="2.17.1-pro" />
+    <PackageVersion Include="Verbara.Sdk.Pro.Storage.Common" Version="2.17.1-pro" />
+    <PackageVersion Include="Verbara.Sdk.Pro.OpenTelemetry" Version="2.17.1-pro" />
   </ItemGroup>
 
   <ItemGroup>

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<AgentPauseCoordinator>();
+
         // 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<IQueueStore>(QueueStoreInner),
                     sp.GetRequiredKeyedService<IAgentStore>(AgentStoreInner),
                     sync,
+                    sp.GetRequiredService<AgentPauseCoordinator>(),
                     sp.GetRequiredService<ILogger<RealtimeSyncingQueueMembershipStore>>());
         });
 

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;
+
+/// <summary>
+/// The one owner of an agent's Asterisk Realtime pause decision: the pause rule and the per-agent
+/// lock that every <c>queue_members.paused</c> writer in this process takes
+/// (queue-members-stay-unpaused-across-reconcile, design D2).
+/// </summary>
+/// <remarks>
+/// <para>
+/// <see cref="IRealtimeSyncService.AddQueueMemberAsync"/> inserts a new row with <c>paused</c> = 1
+/// and, from Sdk.Pro 2.17.1-pro, leaves an existing row's <c>paused</c> untouched. So after every
+/// upsert Platform issues, <see cref="ConvergeAsync"/> re-asserts the agent's real state with
+/// <see cref="IRealtimeSyncService.SyncAgentPausedAsync"/>, which sets every row of the agent's
+/// interface. It writes the database only, never AMI (design D3): <c>app_queue</c> re-reads the
+/// realtime row when a caller joins. <see cref="RealtimeStateBridge"/> keeps sending AMI
+/// <c>QueuePause</c> on real state changes, under the same lock.
+/// </para>
+/// <para>
+/// <b>Per-process lock (accepted risk R2).</b> 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.
+/// </para>
+/// <para>
+/// 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.
+/// </para>
+/// </remarks>
+internal sealed class AgentPauseCoordinator
+{
+    private readonly ConcurrentDictionary<string, SemaphoreSlim> _agentLocks = new(StringComparer.Ordinal);
+
+    /// <summary>
+    /// The pause rule shared with <see cref="RealtimeStateBridge"/>: an agent is paused in Asterisk
+    /// when its state is not routable (<see cref="AgentState.Available"/> and <see cref="AgentState.Busy"/>
+    /// are) or when a deferred pause is pending.
+    /// </summary>
+    public static bool ShouldPause(Agent agent)
+    {
+        ArgumentNullException.ThrowIfNull(agent);
+        return !AgentStateMachine.IsRoutable(agent.State) || agent.PendingState is not null;
+    }
+
+    /// <summary>
+    /// Mirrors the Sdk.Pro voice gate of <see cref="IRealtimeSyncService.AddQueueMemberAsync"/>
+    /// (ADR-0026 Phase B): a <c>null</c> channel list, or one that contains <c>"voice"</c>
+    /// (case-insensitive), upserts a <c>queue_members</c> row. Any other list is digital-only and
+    /// removes the row instead, so there is nothing to converge.
+    /// </summary>
+    public static bool CreatesVoiceRow(IReadOnlyList<string>? allowedChannels)
+    {
+        if (allowedChannels is null)
+            return true;
+        foreach (var channel in allowedChannels)
+        {
+            if (string.Equals(channel, "voice", StringComparison.OrdinalIgnoreCase))
+                return true;
+        }
+        return false;
+    }
+
+    /// <summary>
+    /// Waits for the per-agent lock and returns a handle that releases it on dispose. Every
+    /// <c>queue_members.paused</c> write for <paramref name="agentId"/> in this process runs under it.
+    /// </summary>
+    public async Task<IDisposable> 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);
+    }
+
+    /// <summary>
+    /// Re-reads the agent under its lock and writes its real pause state to every
+    /// <c>queue_members</c> row of its interface. Returns the value written, or <c>null</c> when
+    /// the agent no longer exists (nothing is written).
+    /// </summary>
+    /// <param name="tenantId">The agent's tenant.</param>
+    /// <param name="agentId">The agent to converge.</param>
+    /// <param name="agents">An agent store that reads the current row (no tick-level cache).</param>
+    /// <param name="sync">The realtime sync service that owns the write.</param>
+    /// <param name="ct">Cancellation.</param>
+    public async Task<bool?> 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();
+    }
+}