Merge stream C (gRPC surface: audit aggregation, streaming auth, alarm overflow/snapshot, write cancellation, correlation echo, disconnect reasons, verifier single-flight, opt-in /metrics auth)

src/ZB.MOM.WW.MxGateway.IntegrationTests/WorkerLiveMxAccessSmokeTests.cs

@@ -1704,6 +1704,20 @@ public sealed class WorkerLiveMxAccessSmokeTests(ITestOutputHelper output)
         /// </summary>
         public WriteOptions? WriteOptions { get; set; }
 
+        /// <summary>
+        /// Records the message unless the token is already cancelled.
+        /// </summary>
+        /// <param name="message">The message to write.</param>
+        /// <param name="cancellationToken">Token that cancels the write.</param>
+        /// <returns>A task that represents the asynchronous operation.</returns>
+        public Task WriteAsync(T message, CancellationToken cancellationToken)
+        {
+            // The interface default throws NotSupportedException for a cancelable token; the
+            // gateway passes the call's token on every streamed write.
+            cancellationToken.ThrowIfCancellationRequested();
+            return WriteAsync(message);
+        }
+
         /// <summary>
         /// Records the message and signals any pending waiter.
         /// </summary>
@@ -2112,26 +2126,26 @@ public sealed class WorkerLiveMxAccessSmokeTests(ITestOutputHelper output)
     private sealed class AllowAllConstraintEnforcer : IConstraintEnforcer
     {
         /// <inheritdoc />
-        public Task<ConstraintFailure?> CheckReadTagAsync(
+        public ValueTask<ConstraintFailure?> CheckReadTagAsync(
             ApiKeyIdentity? identity,
             string tagAddress,
-            CancellationToken cancellationToken) => Task.FromResult<ConstraintFailure?>(null);
+            CancellationToken cancellationToken) => ValueTask.FromResult<ConstraintFailure?>(null);
 
         /// <inheritdoc />
-        public Task<ConstraintFailure?> CheckReadHandleAsync(
+        public ValueTask<ConstraintFailure?> CheckReadHandleAsync(
             ApiKeyIdentity? identity,
             GatewaySession session,
             int serverHandle,
             int itemHandle,
-            CancellationToken cancellationToken) => Task.FromResult<ConstraintFailure?>(null);
+            CancellationToken cancellationToken) => ValueTask.FromResult<ConstraintFailure?>(null);
 
         /// <inheritdoc />
-        public Task<ConstraintFailure?> CheckWriteHandleAsync(
+        public ValueTask<ConstraintFailure?> CheckWriteHandleAsync(
             ApiKeyIdentity? identity,
             GatewaySession session,
             int serverHandle,
             int itemHandle,
-            CancellationToken cancellationToken) => Task.FromResult<ConstraintFailure?>(null);
+            CancellationToken cancellationToken) => ValueTask.FromResult<ConstraintFailure?>(null);
 
         /// <inheritdoc />
         public Task RecordDenialAsync(
@@ -2141,6 +2155,14 @@ public sealed class WorkerLiveMxAccessSmokeTests(ITestOutputHelper output)
             ConstraintFailure failure,
             string? correlationId,
             CancellationToken cancellationToken) => Task.CompletedTask;
+
+        /// <inheritdoc />
+        public Task RecordBulkDenialAsync(
+            ApiKeyIdentity? identity,
+            string commandKind,
+            ConstraintDenialSummary summary,
+            string? correlationId,
+            CancellationToken cancellationToken) => Task.CompletedTask;
     }
 
 }

src/ZB.MOM.WW.MxGateway.Server/Alarms/AlarmFeedSubscriberOverflowException.cs

@@ -0,0 +1,39 @@
+namespace ZB.MOM.WW.MxGateway.Server.Alarms;
+
+/// <summary>
+///     Completes an alarm-feed subscriber that fell behind its bounded queue and was dropped.
+/// </summary>
+/// <remarks>
+///     A typed exception rather than a bare <see cref="InvalidOperationException"/> so the gRPC
+///     layer can map it to <c>ResourceExhausted</c> — the status a <c>StreamEvents</c> overflow
+///     carries — without treating a slow client as a gateway failure (an unrecognized exception is
+///     logged as a warning with its stack and surfaced as <c>Unavailable</c>). The remedy is the
+///     same as for an event-stream overflow: reconnect, which re-snapshots the alarm set.
+/// </remarks>
+public sealed class AlarmFeedSubscriberOverflowException : Exception
+{
+    /// <summary>The message every dropped subscriber is completed with.</summary>
+    public const string DefaultMessage =
+        "Alarm feed subscriber fell behind and was dropped; reconnect to re-snapshot.";
+
+    /// <summary>Initializes a new instance of the <see cref="AlarmFeedSubscriberOverflowException"/> class.</summary>
+    public AlarmFeedSubscriberOverflowException()
+        : base(DefaultMessage)
+    {
+    }
+
+    /// <summary>Initializes a new instance of the <see cref="AlarmFeedSubscriberOverflowException"/> class.</summary>
+    /// <param name="message">The error message.</param>
+    public AlarmFeedSubscriberOverflowException(string message)
+        : base(message)
+    {
+    }
+
+    /// <summary>Initializes a new instance of the <see cref="AlarmFeedSubscriberOverflowException"/> class.</summary>
+    /// <param name="message">The error message.</param>
+    /// <param name="innerException">The exception that caused this one.</param>
+    public AlarmFeedSubscriberOverflowException(string message, Exception innerException)
+        : base(message, innerException)
+    {
+    }
+}

src/ZB.MOM.WW.MxGateway.Server/Alarms/GatewayAlarmMonitor.cs

@@ -967,7 +967,9 @@ public sealed class GatewayAlarmMonitor : BackgroundService, IGatewayAlarmServic
     }
 
     // Caller holds _sync. Pushes a feed message to every matching subscriber;
-    // a subscriber that has fallen behind is completed with an error and dropped.
+    // a subscriber that has fallen behind is completed with an AlarmFeedSubscriberOverflowException
+    // and dropped. The typed fault is what lets StreamAlarms answer ResourceExhausted instead of
+    // treating a slow client as a gateway failure.
     private void Broadcast(AlarmFeedMessage message, string reference)
     {
         for (int index = _subscribers.Count - 1; index >= 0; index--)
@@ -980,8 +982,7 @@ public sealed class GatewayAlarmMonitor : BackgroundService, IGatewayAlarmServic
 
             if (!subscriber.Channel.Writer.TryWrite(message))
             {
-                subscriber.Channel.Writer.TryComplete(new InvalidOperationException(
-                    "Alarm feed subscriber fell behind and was dropped; reconnect to re-snapshot."));
+                subscriber.Channel.Writer.TryComplete(new AlarmFeedSubscriberOverflowException());
                 _subscribers.RemoveAt(index);
             }
         }
@@ -997,8 +998,7 @@ public sealed class GatewayAlarmMonitor : BackgroundService, IGatewayAlarmServic
             Subscriber subscriber = _subscribers[index];
             if (!subscriber.Channel.Writer.TryWrite(message))
             {
-                subscriber.Channel.Writer.TryComplete(new InvalidOperationException(
-                    "Alarm feed subscriber fell behind and was dropped; reconnect to re-snapshot."));
+                subscriber.Channel.Writer.TryComplete(new AlarmFeedSubscriberOverflowException());
                 _subscribers.RemoveAt(index);
             }
         }
@@ -1046,9 +1046,14 @@ public sealed class GatewayAlarmMonitor : BackgroundService, IGatewayAlarmServic
             _subscribers.Add(subscriber);
             providerStatus = BuildProviderStatus();
             snapshotTruncated = _snapshotTruncated;
-            snapshot = _alarms.Values
-                .Where(alarm => prefix.Length == 0
-                    || alarm.AlarmFullReference.StartsWith(prefix, StringComparison.Ordinal))
+
+            // Take the memoized projection rather than cloning the whole set per attach: the
+            // clone runs at most once per alarm-set generation (shared with CurrentAlarms), so a
+            // burst of reconnecting subscribers no longer holds the broadcast lock for one full
+            // copy each. The array is never mutated once built — a write to _alarms replaces it
+            // with null — so capturing the reference here pins exactly the generation this
+            // subscriber registered against, which is what keeps snapshot-then-live ordering.
+            snapshot = _currentAlarmsProjection ??= _alarms.Values
                 .Select(alarm => alarm.Clone())
                 .ToArray();
         }
@@ -1068,8 +1073,15 @@ public sealed class GatewayAlarmMonitor : BackgroundService, IGatewayAlarmServic
                 SnapshotStatus = new AlarmSnapshotStatus { Truncated = snapshotTruncated },
             };
 
+            // The prefix filter runs here, outside _sync, over the immutable projection.
             foreach (ActiveAlarmSnapshot alarm in snapshot)
             {
+                if (prefix.Length != 0
+                    && !alarm.AlarmFullReference.StartsWith(prefix, StringComparison.Ordinal))
+                {
+                    continue;
+                }
+
                 yield return new AlarmFeedMessage { ActiveAlarm = alarm };
             }
 

src/ZB.MOM.WW.MxGateway.Server/Configuration/GatewayOptions.cs

@@ -52,4 +52,7 @@ public sealed class GatewayOptions
     /// login / API-key rate limiting).
     /// </summary>
     public SecurityOptions Security { get; init; } = new();
+
+    /// <summary>Gets Prometheus <c>/metrics</c> endpoint options.</summary>
+    public MetricsOptions Metrics { get; init; } = new();
 }

src/ZB.MOM.WW.MxGateway.Server/Configuration/MetricsOptions.cs

@@ -0,0 +1,15 @@
+namespace ZB.MOM.WW.MxGateway.Server.Configuration;
+
+/// <summary>Options for the Prometheus <c>/metrics</c> scrape endpoint (<c>MxGateway:Metrics</c>).</summary>
+public sealed class MetricsOptions
+{
+    /// <summary>
+    /// Gets a value indicating whether <c>/metrics</c> requires an authenticated caller. When
+    /// <see langword="true"/>, a scrape must present either an API key carrying the <c>admin</c>
+    /// scope (<c>Authorization: Bearer mxgw_…</c>) or a dashboard login cookie holding a real
+    /// dashboard role; anonymous localhost access does not qualify. Default <see langword="false"/>
+    /// keeps the endpoint anonymous for existing scrapers. The <c>/health</c> endpoints stay
+    /// anonymous either way.
+    /// </summary>
+    public bool RequireAuthentication { get; init; }
+}

src/ZB.MOM.WW.MxGateway.Server/GatewayApplication.cs

@@ -113,6 +113,7 @@ public static class GatewayApplication
         builder.Services.AddZbSecrets(builder.Configuration, "Secrets");
         builder.Services.AddSqliteAuthStore(builder.Configuration);
         builder.Services.AddGatewayGrpcAuthorization();
+        builder.Services.AddGatewayMetricsAuthorization();
         AddLoginRateLimiter(builder);
         builder.Services.AddHealthChecks()
             .AddTypeActivatedCheck<AuthStoreHealthCheck>(
@@ -350,8 +351,19 @@ public static class GatewayApplication
     {
         endpoints.MapStaticAssets(ResolveStaticAssetsManifestPath());
 
+        // /health stays anonymous unconditionally: an off-repo Overview dashboard polls
+        // /health/active without credentials.
         endpoints.MapZbHealth();
-        endpoints.MapZbMetrics();
+        IEndpointConventionBuilder metricsEndpoint = endpoints.MapZbMetrics();
+        // Read from configuration rather than IOptions<GatewayOptions>: resolving the options here
+        // would run whole-options validation during Build instead of at StartAsync.
+        if (endpoints.ServiceProvider
+            .GetRequiredService<IConfiguration>()
+            .GetValue<bool>("MxGateway:Metrics:RequireAuthentication"))
+        {
+            // Opt-in (SEC-14): an admin-scoped API key or a dashboard login with a real role.
+            metricsEndpoint.RequireAuthorization(MetricsAuthorizationDefaults.Policy);
+        }
 
         endpoints.MapGrpcService<MxAccessGatewayService>();
         endpoints.MapZbGalaxyRepository();

src/ZB.MOM.WW.MxGateway.Server/Grpc/EventStreamDisconnectReasons.cs

@@ -0,0 +1,30 @@
+namespace ZB.MOM.WW.MxGateway.Server.Grpc;
+
+/// <summary>
+///     The bounded set of <c>reason</c> tag values recorded on
+///     <c>mxgateway.grpc.streams.disconnected</c> when a <c>StreamEvents</c> stream ends.
+/// </summary>
+/// <remarks>
+///     Constants rather than exception messages or free text so the metric's tag cardinality stays
+///     fixed: an exporter mints one time series per distinct tag value.
+/// </remarks>
+public static class EventStreamDisconnectReasons
+{
+    /// <summary>The session's event stream completed cleanly (session closed or the worker stream ended).</summary>
+    public const string SessionClosed = "SessionClosed";
+
+    /// <summary>The subscriber fell behind its bounded queue and was disconnected.</summary>
+    public const string Overflow = "Overflow";
+
+    /// <summary>The worker event stream terminated with a fault.</summary>
+    public const string WorkerFault = "WorkerFault";
+
+    /// <summary>The client cancelled the call or disconnected.</summary>
+    public const string ClientCancelled = "ClientCancelled";
+
+    /// <summary>The stream ended on an unclassified error.</summary>
+    public const string Error = "Error";
+
+    /// <summary>The consumer stopped reading without cancelling the call.</summary>
+    public const string Detached = "Detached";
+}

src/ZB.MOM.WW.MxGateway.Server/Grpc/EventStreamService.cs

@@ -129,6 +129,10 @@ public sealed class EventStreamService(
         IDisposable backlogRegistration = metrics.RegisterEventStreamBacklogSource(
             () => subscriber.Reader.CanCount ? subscriber.Reader.Count : 0);
 
+        // Why this stream ended, recorded as a bounded constant on the disconnect metric. Set at
+        // the point the end is observed; a consumer that simply stops iterating keeps Detached.
+        string disconnectReason = EventStreamDisconnectReasons.Detached;
+
         try
         {
             // Emit order for a resume: the ReplayGap sentinel FIRST (only when events were
@@ -173,13 +177,31 @@ public sealed class EventStreamService(
                     // here. Mirror the original ProduceEventsAsync behavior: fault the
                     // session and record the metric, then propagate the terminal fault to the
                     // gRPC client.
+                    disconnectReason = EventStreamDisconnectReasons.WorkerFault;
                     session.MarkFaulted(workerException.Message);
                     metrics.Fault(WorkerClientErrorCode.WorkerFaulted.ToString());
                     throw;
                 }
+                catch (SessionManagerException sessionException)
+                    when (sessionException.ErrorCode == SessionManagerErrorCode.EventQueueOverflow)
+                {
+                    disconnectReason = EventStreamDisconnectReasons.Overflow;
+                    throw;
+                }
+                catch (OperationCanceledException)
+                {
+                    disconnectReason = EventStreamDisconnectReasons.ClientCancelled;
+                    throw;
+                }
+                catch (Exception)
+                {
+                    disconnectReason = EventStreamDisconnectReasons.Error;
+                    throw;
+                }
 
                 if (!hasMore)
                 {
+                    disconnectReason = EventStreamDisconnectReasons.SessionClosed;
                     break;
                 }
 
@@ -214,7 +236,15 @@ public sealed class EventStreamService(
             backlogRegistration.Dispose();
             subscriber.Dispose();
 
-            metrics.StreamDisconnected("Detached");
+            // A consumer that stopped iterating mid-stream lands here with the Detached default;
+            // when the call token has fired, that early stop was the client going away.
+            if (disconnectReason == EventStreamDisconnectReasons.Detached
+                && cancellationToken.IsCancellationRequested)
+            {
+                disconnectReason = EventStreamDisconnectReasons.ClientCancelled;
+            }
+
+            metrics.StreamDisconnected(disconnectReason);
         }
     }
 

src/ZB.MOM.WW.MxGateway.Server/Grpc/MxAccessGatewayService.cs

@@ -144,6 +144,25 @@ public sealed class MxAccessGatewayService(
                 publicReply = bulkConstraintPlan.MergeDeniedInto(publicReply);
             }
 
+            // MxCommandReply.correlation_id always echoes the caller's client_correlation_id —
+            // the same value the constraint-denied replies carry — so a client can match every
+            // reply to its request. The worker reply carries the gateway's internal envelope id,
+            // which is a gateway↔worker routing detail and is overwritten here, empty included.
+            // Record the pairing once, before the envelope id is overwritten, so a client error
+            // report quoting its correlation id can be matched to gateway and worker log lines
+            // (which carry the envelope id). Ids and the command kind only — never values.
+            if (logger.IsEnabled(LogLevel.Debug))
+            {
+                logger.LogDebug(
+                    "Invoke {CommandKind} for session {SessionId}: client correlation id {ClientCorrelationId} maps to worker envelope correlation id {WorkerCorrelationId}.",
+                    command.Kind,
+                    request.SessionId,
+                    request.ClientCorrelationId,
+                    publicReply.CorrelationId);
+            }
+
+            publicReply.CorrelationId = request.ClientCorrelationId;
+
             session.TrackCommandReply(commandToInvoke, publicReply);
             return publicReply;
         }
@@ -172,14 +191,22 @@ public sealed class MxAccessGatewayService(
                 // volume gateway path. Stopwatch.GetTimestamp/GetElapsedTime measure the exact
                 // same wall-clock span the former Stopwatch.StartNew()/.Elapsed did.
                 long sendStartTimestamp = Stopwatch.GetTimestamp();
-                await responseStream.WriteAsync(publicEvent).ConfigureAwait(false);
+
+                // The call token releases a write parked on a slow or vanished client when the
+                // call is cancelled, instead of leaving it to the transport's own timeout.
+                await responseStream.WriteAsync(publicEvent, context.CancellationToken).ConfigureAwait(false);
                 metrics.RecordEventStreamSend(
-                    publicEvent.Family.ToString(),
+                    MxEventFamilyNames.Get(publicEvent.Family),
                     Stopwatch.GetElapsedTime(sendStartTimestamp));
             }
         }
         catch (Exception exception) when (exception is not RpcException)
         {
+            if (IsClientAbort(exception, context))
+            {
+                throw ClientCancelled();
+            }
+
             throw MapException(exception);
         }
     }
@@ -225,11 +252,24 @@ public sealed class MxAccessGatewayService(
                 .WithCancellation(context.CancellationToken)
                 .ConfigureAwait(false))
             {
-                await responseStream.WriteAsync(message).ConfigureAwait(false);
+                await responseStream.WriteAsync(message, context.CancellationToken).ConfigureAwait(false);
             }
         }
+        catch (AlarmFeedSubscriberOverflowException overflow)
+        {
+            // The monitor dropped this subscriber for falling behind its bounded queue. That is a
+            // client-side slowness condition with a client-side remedy (reconnect and
+            // re-snapshot), so it maps to ResourceExhausted like a StreamEvents overflow rather
+            // than reaching MapException's warning-with-stack + Unavailable fallback.
+            throw new RpcException(new Status(StatusCode.ResourceExhausted, overflow.Message));
+        }
         catch (Exception exception) when (exception is not RpcException)
         {
+            if (IsClientAbort(exception, context))
+            {
+                throw ClientCancelled();
+            }
+
             throw MapException(exception);
         }
     }
@@ -264,6 +304,17 @@ public sealed class MxAccessGatewayService(
         }
     }
 
+    // A response write fails with an IOException (or an operation-cancelled exception) when the
+    // client disconnects mid-stream. With the call token already cancelled that is a routine
+    // client abort, not a gateway fault: it maps to Cancelled with no warning. The same exception
+    // on a live call token is a genuine transport failure and keeps the MapException path.
+    private static bool IsClientAbort(Exception exception, ServerCallContext context) =>
+        context.CancellationToken.IsCancellationRequested
+        && exception is IOException or OperationCanceledException;
+
+    private static RpcException ClientCancelled() =>
+        new(new Status(StatusCode.Cancelled, "gRPC request was canceled."));
+
     private string? ResolveClientIdentity()
     {
         return identityAccessor.Current?.DisplayName ?? identityAccessor.Current?.KeyId;
@@ -488,6 +539,7 @@ public sealed class MxAccessGatewayService(
 
         Dictionary<int, SubscribeResult> denied = [];
         List<string> allowed = [];
+        ConstraintDenialSummary? denialSummary = null;
         for (int index = 0; index < tagAddresses.Count; index++)
         {
             string tagAddress = tagAddresses[index];
@@ -500,8 +552,7 @@ public sealed class MxAccessGatewayService(
                 continue;
             }
 
-            await constraintEnforcer.RecordDenialAsync(identity, command.Kind.ToString(), tagAddress, failure, correlationId, cancellationToken)
-                .ConfigureAwait(false);
+            (denialSummary ??= new ConstraintDenialSummary(tagAddresses.Count)).Add(tagAddress, failure);
             denied[index] = new SubscribeResult
             {
                 ServerHandle = serverHandle,
@@ -511,11 +562,17 @@ public sealed class MxAccessGatewayService(
             };
         }
 
-        if (denied.Count == 0)
+        if (denialSummary is null)
         {
             return null;
         }
 
+        // One aggregated audit event per bulk command, not one per denied item (see
+        // ConstraintDenialSummary for why).
+        await constraintEnforcer
+            .RecordBulkDenialAsync(identity, command.Kind.ToString(), denialSummary, correlationId, cancellationToken)
+            .ConfigureAwait(false);
+
         // Build the filtered command directly instead of cloning the original and clearing it:
         // the clone deep-copied every denied address only to drop it. The payload's other fields
         // (server_handle) are copied across explicitly. Nothing aliases the request here — these
@@ -556,6 +613,7 @@ public sealed class MxAccessGatewayService(
 
         Dictionary<int, BulkReadResult> denied = [];
         List<string> allowed = [];
+        ConstraintDenialSummary? denialSummary = null;
         for (int index = 0; index < tagAddresses.Count; index++)
         {
             string tagAddress = tagAddresses[index];
@@ -568,8 +626,7 @@ public sealed class MxAccessGatewayService(
                 continue;
             }
 
-            await constraintEnforcer.RecordDenialAsync(identity, command.Kind.ToString(), tagAddress, failure, correlationId, cancellationToken)
-                .ConfigureAwait(false);
+            (denialSummary ??= new ConstraintDenialSummary(tagAddresses.Count)).Add(tagAddress, failure);
             denied[index] = new BulkReadResult
             {
                 ServerHandle = serverHandle,
@@ -580,11 +637,17 @@ public sealed class MxAccessGatewayService(
             };
         }
 
-        if (denied.Count == 0)
+        if (denialSummary is null)
         {
             return null;
         }
 
+        // One aggregated audit event per bulk command, not one per denied item (see
+        // ConstraintDenialSummary for why).
+        await constraintEnforcer
+            .RecordBulkDenialAsync(identity, command.Kind.ToString(), denialSummary, correlationId, cancellationToken)
+            .ConfigureAwait(false);
+
         MxCommand filtered = new() { Kind = command.Kind };
         ReadBulkCommand payload = new()
         {
@@ -621,6 +684,7 @@ public sealed class MxAccessGatewayService(
 
         Dictionary<int, BulkWriteResult> denied = [];
         List<TEntry> allowed = [];
+        ConstraintDenialSummary? denialSummary = null;
         for (int index = 0; index < entries.Count; index++)
         {
             TEntry entry = entries[index];
@@ -634,14 +698,7 @@ public sealed class MxAccessGatewayService(
                 continue;
             }
 
-            await constraintEnforcer.RecordDenialAsync(
-                    identity,
-                    command.Kind.ToString(),
-                    itemHandle.ToString(System.Globalization.CultureInfo.InvariantCulture),
-                    failure,
-                    correlationId,
-                    cancellationToken)
-                .ConfigureAwait(false);
+            (denialSummary ??= new ConstraintDenialSummary(entries.Count)).Add(itemHandle, failure);
             denied[index] = new BulkWriteResult
             {
                 ServerHandle = serverHandle,
@@ -651,11 +708,17 @@ public sealed class MxAccessGatewayService(
             };
         }
 
-        if (denied.Count == 0)
+        if (denialSummary is null)
         {
             return null;
         }
 
+        // One aggregated audit event per bulk command, not one per denied item (see
+        // ConstraintDenialSummary for why).
+        await constraintEnforcer
+            .RecordBulkDenialAsync(identity, command.Kind.ToString(), denialSummary, correlationId, cancellationToken)
+            .ConfigureAwait(false);
+
         return new WriteBulkConstraintPlan(
             BuildFilteredWriteBulkCommand(command, allowed),
             entries.Count,
@@ -742,6 +805,7 @@ public sealed class MxAccessGatewayService(
 
         Dictionary<int, SubscribeResult> denied = [];
         List<int> allowed = [];
+        ConstraintDenialSummary? denialSummary = null;
         for (int index = 0; index < itemHandles.Count; index++)
         {
             int itemHandle = itemHandles[index];
@@ -754,8 +818,7 @@ public sealed class MxAccessGatewayService(
                 continue;
             }
 
-            await constraintEnforcer.RecordDenialAsync(identity, command.Kind.ToString(), itemHandle.ToString(System.Globalization.CultureInfo.InvariantCulture), failure, correlationId, cancellationToken)
-                .ConfigureAwait(false);
+            (denialSummary ??= new ConstraintDenialSummary(itemHandles.Count)).Add(itemHandle, failure);
             denied[index] = new SubscribeResult
             {
                 ServerHandle = serverHandle,
@@ -765,11 +828,17 @@ public sealed class MxAccessGatewayService(
             };
         }
 
-        if (denied.Count == 0)
+        if (denialSummary is null)
         {
             return null;
         }
 
+        // One aggregated audit event per bulk command, not one per denied item (see
+        // ConstraintDenialSummary for why).
+        await constraintEnforcer
+            .RecordBulkDenialAsync(identity, command.Kind.ToString(), denialSummary, correlationId, cancellationToken)
+            .ConfigureAwait(false);
+
         MxCommand filtered = new() { Kind = command.Kind };
         AdviseItemBulkCommand payload = new() { ServerHandle = command.AdviseItemBulk.ServerHandle };
         payload.ItemHandles.Add(allowed);

src/ZB.MOM.WW.MxGateway.Server/Metrics/MxEventFamilyNames.cs

@@ -0,0 +1,28 @@
+using System.Collections.Frozen;
+using ZB.MOM.WW.MxGateway.Contracts.Proto;
+
+namespace ZB.MOM.WW.MxGateway.Server.Metrics;
+
+/// <summary>
+///     Cached <see cref="MxEventFamily"/> names for the per-event metric tags.
+/// </summary>
+/// <remarks>
+///     <see cref="Enum.ToString()"/> allocates a new string on every call, and the event-received
+///     and stream-send metrics tag every event with its family — once per event at the worker
+///     boundary and once per event per subscriber on the stream. The names are fixed by the
+///     contract, so they are computed once here. A value the contract does not define falls back
+///     to <see cref="Enum.ToString()"/>, which is the same string the uncached call produced.
+/// </remarks>
+public static class MxEventFamilyNames
+{
+    private static readonly FrozenDictionary<MxEventFamily, string> Names =
+        Enum.GetValues<MxEventFamily>()
+            .Distinct()
+            .ToFrozenDictionary(family => family, family => family.ToString());
+
+    /// <summary>Gets the name of an event family.</summary>
+    /// <param name="family">The event family.</param>
+    /// <returns>The family's name, identical to <see cref="Enum.ToString()"/>.</returns>
+    public static string Get(MxEventFamily family) =>
+        Names.TryGetValue(family, out string? name) ? name : family.ToString();
+}

src/ZB.MOM.WW.MxGateway.Server/Security/Authentication/CachingApiKeyVerifier.cs

@@ -58,6 +58,19 @@ public interface IApiKeyCacheInvalidator
 /// window and it is intentionally kept short (<see cref="SecurityOptions.ApiKeyVerificationCacheSeconds"/>,
 /// default 15 s).
 /// </para>
+/// <para>
+/// Misses are single-flight: concurrent misses for the same token and the same key generation share
+/// one inner verification, so a TTL expiry under load costs one SQLite read rather than one per
+/// in-flight RPC. A miss that starts after an <see cref="Invalidate"/> never joins a flight that
+/// started before it (the flight is keyed on the generation too), so the revoke guarantee holds for
+/// joiners as well. The shared inner call is not bound to any one caller's cancellation token; each
+/// caller stops waiting on its own token without cancelling the others.
+/// </para>
+/// <para>
+/// The key-id index is pruned by a post-eviction callback whenever the cache drops an entry on its
+/// own (expiry, compaction), not only by <see cref="Invalidate"/>, so it stays proportional to the
+/// live cache rather than to every token ever verified.
+/// </para>
 /// </remarks>
 public sealed class CachingApiKeyVerifier : IApiKeyVerifier, IApiKeyCacheInvalidator
 {
@@ -69,9 +82,20 @@ public sealed class CachingApiKeyVerifier : IApiKeyVerifier, IApiKeyCacheInvalid
     private readonly TimeSpan _ttl;
 
     // keyId -> set of cache keys currently held for that id, so a revoke/rotate can evict every
-    // secret variant (e.g. an old secret still within TTL after a rotate).
-    private readonly ConcurrentDictionary<string, ConcurrentDictionary<string, byte>> _keyIdIndex =
-        new(StringComparer.Ordinal);
+    // secret variant (e.g. an old secret still within TTL after a rotate). Guarded by _indexSync,
+    // which also covers the cache Set that an index entry describes: the post-eviction callback
+    // runs asynchronously, so Set + index and check-cache + unindex must each be atomic with
+    // respect to the other, or a late callback could unindex a freshly re-cached entry and leave
+    // it out of reach of a later Invalidate. Every operation under the lock is O(1) and the lock is
+    // only taken on misses, evictions and invalidations — never on a cache hit.
+    private readonly object _indexSync = new();
+    private readonly Dictionary<string, HashSet<string>> _keyIdIndex = new(StringComparer.Ordinal);
+
+    // (cache key, key generation) -> the in-flight inner verification for that pair (single-flight).
+    private readonly ConcurrentDictionary<(string CacheKey, long Generation), Lazy<Task<ApiKeyVerification>>> _inflight =
+        new();
+
+    private readonly PostEvictionDelegate _onEvicted;
 
     // keyId -> monotonic generation, bumped by Invalidate before it evicts. A VerifyAsync snapshots
     // the generation before calling the inner verifier and only caches if it is unchanged after the
@@ -103,6 +127,37 @@ public sealed class CachingApiKeyVerifier : IApiKeyVerifier, IApiKeyCacheInvalid
         _inner = inner;
         _cache = cache;
         _ttl = ttl;
+        _onEvicted = OnCacheEntryEvicted;
+    }
+
+    /// <summary>Gets the number of cache keys currently held in the key-id index (test seam).</summary>
+    internal int IndexedCacheKeyCount
+    {
+        get
+        {
+            lock (_indexSync)
+            {
+                int count = 0;
+                foreach (HashSet<string> set in _keyIdIndex.Values)
+                {
+                    count += set.Count;
+                }
+
+                return count;
+            }
+        }
+    }
+
+    /// <summary>Gets the number of key ids currently present in the key-id index (test seam).</summary>
+    internal int IndexedKeyIdCount
+    {
+        get
+        {
+            lock (_indexSync)
+            {
+                return _keyIdIndex.Count;
+            }
+        }
     }
 
     /// <summary>
@@ -126,11 +181,50 @@ public sealed class CachingApiKeyVerifier : IApiKeyVerifier, IApiKeyCacheInvalid
         }
 
         // Snapshot the key's generation BEFORE the inner verify so a concurrent Invalidate that runs
-        // while we are in the library is detected below and its repopulation discarded.
+        // while we are in the library is detected and its repopulation discarded. The snapshot is
+        // also half of the single-flight key, so a miss that starts after an Invalidate cannot join
+        // a flight that started before it.
         string? tokenKeyId = TryParseKeyId(authorizationHeader);
         long generationAtStart = tokenKeyId is null ? 0 : ReadGeneration(tokenKeyId);
+        (string CacheKey, long Generation) flightKey = (cacheKey, generationAtStart);
 
-        ApiKeyVerification result = await _inner.VerifyAsync(authorizationHeader, ct).ConfigureAwait(false);
+        Lazy<Task<ApiKeyVerification>> created = new(
+            () => VerifyMissAsync(authorizationHeader, cacheKey, tokenKeyId, generationAtStart));
+        Lazy<Task<ApiKeyVerification>> flight = _inflight.GetOrAdd(flightKey, created);
+        Task<ApiKeyVerification> verification = flight.Value;
+        if (ReferenceEquals(flight, created))
+        {
+            // The flight's creator retires it once it completes (success, failure or fault), so the
+            // next miss for this token starts fresh. Removing by (key, value) pair never retires a
+            // newer flight that replaced this one.
+            _ = verification.ContinueWith(
+                static (_, state) =>
+                {
+                    var (owner, key, completed) =
+                        ((CachingApiKeyVerifier, (string, long), Lazy<Task<ApiKeyVerification>>))state!;
+                    owner._inflight.TryRemove(
+                        new KeyValuePair<(string CacheKey, long Generation), Lazy<Task<ApiKeyVerification>>>(key, completed));
+                },
+                (this, flightKey, flight),
+                CancellationToken.None,
+                TaskContinuationOptions.ExecuteSynchronously,
+                TaskScheduler.Default);
+        }
+
+        return await verification.WaitAsync(ct).ConfigureAwait(false);
+    }
+
+    // The shared body of one single-flight miss. Not bound to any caller's token: a caller that
+    // gives up must not cancel the verification other callers are waiting on.
+    private async Task<ApiKeyVerification> VerifyMissAsync(
+        string authorizationHeader,
+        string cacheKey,
+        string? tokenKeyId,
+        long generationAtStart)
+    {
+        ApiKeyVerification result = await _inner
+            .VerifyAsync(authorizationHeader, CancellationToken.None)
+            .ConfigureAwait(false);
 
         if (result.Succeeded && result.Identity is not null)
         {
@@ -158,11 +252,21 @@ public sealed class CachingApiKeyVerifier : IApiKeyVerifier, IApiKeyCacheInvalid
         }
 
         string keyId = result.Identity!.KeyId;
-        _cache.Set(cacheKey, result, new MemoryCacheEntryOptions
+        MemoryCacheEntryOptions entryOptions = new()
         {
             AbsoluteExpirationRelativeToNow = _ttl,
+        };
+        entryOptions.PostEvictionCallbacks.Add(new PostEvictionCallbackRegistration
+        {
+            EvictionCallback = _onEvicted,
+            State = keyId,
         });
-        IndexCacheKey(keyId, cacheKey);
+
+        lock (_indexSync)
+        {
+            _cache.Set(cacheKey, result, entryOptions);
+            IndexCacheKey(keyId, cacheKey);
+        }
 
         // Set-then-recheck: an Invalidate that landed between the pre-Set check and IndexCacheKey
         // would have missed this entry (it was not yet indexed). If the generation has moved, evict
@@ -170,13 +274,38 @@ public sealed class CachingApiKeyVerifier : IApiKeyVerifier, IApiKeyCacheInvalid
         if (tokenKeyId is not null && ReadGeneration(tokenKeyId) != generationAtStart)
         {
             _cache.Remove(cacheKey);
-            if (_keyIdIndex.TryGetValue(keyId, out ConcurrentDictionary<string, byte>? set))
+            lock (_indexSync)
             {
-                set.TryRemove(cacheKey, out _);
+                UnindexCacheKey(keyId, cacheKey);
             }
         }
     }
 
+    // Post-eviction callback: drops the evicted entry's key-id index slot so the index tracks the
+    // live cache. MemoryCache raises it asynchronously, so by the time it runs the same cache key
+    // may already hold a newer entry (re-verified after the expiry); that entry owns the slot, and
+    // the check below runs under the lock that also covers Set + index, so it cannot interleave
+    // with a re-cache.
+    private void OnCacheEntryEvicted(object key, object? value, EvictionReason reason, object? state)
+    {
+        if (reason == EvictionReason.Replaced || key is not string cacheKey || state is not string keyId)
+        {
+            // Replaced: a newer entry under the same key was Set in this entry's place and is
+            // indexed under the same slot already.
+            return;
+        }
+
+        lock (_indexSync)
+        {
+            if (_cache.TryGetValue(cacheKey, out _))
+            {
+                return;
+            }
+
+            UnindexCacheKey(keyId, cacheKey);
+        }
+    }
+
     /// <inheritdoc />
     public void Invalidate(string keyId)
     {
@@ -189,12 +318,22 @@ public sealed class CachingApiKeyVerifier : IApiKeyVerifier, IApiKeyCacheInvalid
         // generation refuses to cache (or self-evicts) its now-stale repopulation.
         _generations.AddOrUpdate(keyId, 1, static (_, current) => current + 1);
 
-        if (_keyIdIndex.TryRemove(keyId, out ConcurrentDictionary<string, byte>? cacheKeys))
+        string[] cacheKeys;
+        lock (_indexSync)
         {
-            foreach (string cacheKey in cacheKeys.Keys)
+            if (!_keyIdIndex.Remove(keyId, out HashSet<string>? indexed))
             {
-                _cache.Remove(cacheKey);
+                return;
             }
+
+            cacheKeys = [.. indexed];
+        }
+
+        // Evicted outside the lock: the eviction callbacks these raise take it themselves (and find
+        // the slot already gone).
+        foreach (string cacheKey in cacheKeys)
+        {
+            _cache.Remove(cacheKey);
         }
     }
 
@@ -257,12 +396,28 @@ public sealed class CachingApiKeyVerifier : IApiKeyVerifier, IApiKeyCacheInvalid
         return new string(afterPrefix[..separator]);
     }
 
+    // Caller holds _indexSync.
     private void IndexCacheKey(string keyId, string cacheKey)
     {
-        ConcurrentDictionary<string, byte> set = _keyIdIndex.GetOrAdd(
-            keyId,
-            static _ => new ConcurrentDictionary<string, byte>(StringComparer.Ordinal));
-        set.TryAdd(cacheKey, 0);
+        if (!_keyIdIndex.TryGetValue(keyId, out HashSet<string>? set))
+        {
+            set = new HashSet<string>(StringComparer.Ordinal);
+            _keyIdIndex[keyId] = set;
+        }
+
+        set.Add(cacheKey);
+    }
+
+    // Caller holds _indexSync. Drops the key id's slot entirely once its last cache key goes, so
+    // the index does not accumulate empty sets for keys that are no longer cached.
+    private void UnindexCacheKey(string keyId, string cacheKey)
+    {
+        if (_keyIdIndex.TryGetValue(keyId, out HashSet<string>? set)
+            && set.Remove(cacheKey)
+            && set.Count == 0)
+        {
+            _keyIdIndex.Remove(keyId);
+        }
     }
 
     private static bool TryComputeCacheKey(string? authorizationHeader, out string cacheKey)

src/ZB.MOM.WW.MxGateway.Server/Security/Authorization/ConstraintDenialSummary.cs

@@ -0,0 +1,90 @@
+using System.Globalization;
+
+namespace ZB.MOM.WW.MxGateway.Server.Security.Authorization;
+
+/// <summary>
+///     Accumulates the constraint denials of one bulk command so they are audited as a single
+///     aggregated event rather than one event per denied item.
+/// </summary>
+/// <remarks>
+///     A bulk command can carry thousands of items, and the audit writer is a bounded, drop-on-full
+///     channel. One audit event per denied item let a single low-privilege key flood that channel and
+///     push every other audit record out of it. The summary keeps the forensic signal — how many items
+///     were denied, which constraints denied them, and a bounded sample of the targets — at a fixed
+///     cost of one audit event per command.
+/// </remarks>
+public sealed class ConstraintDenialSummary
+{
+    /// <summary>The maximum number of denied targets retained as a sample.</summary>
+    public const int MaxSampleTargets = 20;
+
+    private readonly List<string> _constraintNames = [];
+    private readonly List<string> _sampleTargets = [];
+
+    /// <summary>Initializes a new instance of the <see cref="ConstraintDenialSummary"/> class.</summary>
+    /// <param name="totalCount">The number of items the bulk command carried.</param>
+    public ConstraintDenialSummary(int totalCount)
+    {
+        TotalCount = totalCount;
+    }
+
+    /// <summary>Gets the number of items the bulk command carried.</summary>
+    public int TotalCount { get; }
+
+    /// <summary>Gets the number of items denied.</summary>
+    public int DeniedCount { get; private set; }
+
+    /// <summary>Gets the distinct constraint names that denied at least one item, in first-seen order.</summary>
+    public IReadOnlyList<string> ConstraintNames => _constraintNames;
+
+    /// <summary>Gets up to <see cref="MaxSampleTargets"/> denied targets, in item order.</summary>
+    public IReadOnlyList<string> SampleTargets => _sampleTargets;
+
+    /// <summary>Gets a value indicating whether more items were denied than the sample retains.</summary>
+    public bool SampleTruncated => DeniedCount > _sampleTargets.Count;
+
+    /// <summary>Records one denied tag-address item.</summary>
+    /// <param name="target">The denied tag address.</param>
+    /// <param name="failure">The constraint failure that denied it.</param>
+    public void Add(string target, ConstraintFailure failure)
+    {
+        ArgumentNullException.ThrowIfNull(failure);
+        AddConstraint(failure.ConstraintName);
+        if (_sampleTargets.Count < MaxSampleTargets)
+        {
+            _sampleTargets.Add(target);
+        }
+
+        DeniedCount++;
+    }
+
+    /// <summary>Records one denied item-handle item.</summary>
+    /// <param name="itemHandle">The denied MXAccess item handle.</param>
+    /// <param name="failure">The constraint failure that denied it.</param>
+    /// <remarks>The handle is formatted only while the sample still has room.</remarks>
+    public void Add(int itemHandle, ConstraintFailure failure)
+    {
+        ArgumentNullException.ThrowIfNull(failure);
+        AddConstraint(failure.ConstraintName);
+        if (_sampleTargets.Count < MaxSampleTargets)
+        {
+            _sampleTargets.Add(itemHandle.ToString(CultureInfo.InvariantCulture));
+        }
+
+        DeniedCount++;
+    }
+
+    private void AddConstraint(string constraintName)
+    {
+        // The constraint vocabulary is a handful of fixed names, so a linear scan beats a set.
+        for (int index = 0; index < _constraintNames.Count; index++)
+        {
+            if (string.Equals(_constraintNames[index], constraintName, StringComparison.Ordinal))
+            {
+                return;
+            }
+        }
+
+        _constraintNames.Add(constraintName);
+    }
+}

src/ZB.MOM.WW.MxGateway.Server/Security/Authorization/ConstraintEnforcer.cs

@@ -25,7 +25,7 @@ public sealed class ConstraintEnforcer(
         identity?.EffectiveConstraints.HasWriteConstraints ?? false;
 
     /// <inheritdoc />
-    public Task<ConstraintFailure?> CheckReadTagAsync(
+    public ValueTask<ConstraintFailure?> CheckReadTagAsync(
         ApiKeyIdentity? identity,
         string tagAddress,
         CancellationToken cancellationToken)
@@ -33,14 +33,14 @@ public sealed class ConstraintEnforcer(
         ApiKeyConstraints constraints = identity?.EffectiveConstraints ?? ApiKeyConstraints.Empty;
         if (!constraints.HasReadConstraints)
         {
-            return Task.FromResult<ConstraintFailure?>(null);
+            return ValueTask.FromResult<ConstraintFailure?>(null);
         }
 
-        return Task.FromResult(CheckReadTarget(constraints, tagAddress));
+        return ValueTask.FromResult(CheckReadTarget(constraints, tagAddress));
     }
 
     /// <inheritdoc />
-    public Task<ConstraintFailure?> CheckReadHandleAsync(
+    public ValueTask<ConstraintFailure?> CheckReadHandleAsync(
         ApiKeyIdentity? identity,
         GatewaySession session,
         int serverHandle,
@@ -50,19 +50,19 @@ public sealed class ConstraintEnforcer(
         ApiKeyConstraints constraints = identity?.EffectiveConstraints ?? ApiKeyConstraints.Empty;
         if (!constraints.HasReadConstraints)
         {
-            return Task.FromResult<ConstraintFailure?>(null);
+            return ValueTask.FromResult<ConstraintFailure?>(null);
         }
 
         if (!session.TryGetItemRegistration(serverHandle, itemHandle, out SessionItemRegistration registration))
         {
-            return Task.FromResult<ConstraintFailure?>(new ConstraintFailure("item_handle", "Item handle is not registered in the constrained session."));
+            return ValueTask.FromResult<ConstraintFailure?>(new ConstraintFailure("item_handle", "Item handle is not registered in the constrained session."));
         }
 
-        return Task.FromResult(CheckReadTarget(constraints, registration.TagAddress));
+        return ValueTask.FromResult(CheckReadTarget(constraints, registration.TagAddress));
     }
 
     /// <inheritdoc />
-    public Task<ConstraintFailure?> CheckWriteHandleAsync(
+    public ValueTask<ConstraintFailure?> CheckWriteHandleAsync(
         ApiKeyIdentity? identity,
         GatewaySession session,
         int serverHandle,
@@ -72,23 +72,23 @@ public sealed class ConstraintEnforcer(
         ApiKeyConstraints constraints = identity?.EffectiveConstraints ?? ApiKeyConstraints.Empty;
         if (!constraints.HasWriteConstraints)
         {
-            return Task.FromResult<ConstraintFailure?>(null);
+            return ValueTask.FromResult<ConstraintFailure?>(null);
         }
 
         if (!session.TryGetItemRegistration(serverHandle, itemHandle, out SessionItemRegistration registration))
         {
-            return Task.FromResult<ConstraintFailure?>(new ConstraintFailure("item_handle", "Item handle is not registered in the constrained session."));
+            return ValueTask.FromResult<ConstraintFailure?>(new ConstraintFailure("item_handle", "Item handle is not registered in the constrained session."));
         }
 
         GalaxyTagLookup? target = ResolveTarget(registration.TagAddress);
         if (target is null)
         {
-            return Task.FromResult<ConstraintFailure?>(new ConstraintFailure("tag_metadata", "Tag metadata is not available in the Galaxy hierarchy cache."));
+            return ValueTask.FromResult<ConstraintFailure?>(new ConstraintFailure("tag_metadata", "Tag metadata is not available in the Galaxy hierarchy cache."));
         }
 
         if (!MatchesPathOrTag(target.ContainedPath, registration.TagAddress, constraints.WriteSubtrees, constraints.WriteTagGlobs))
         {
-            return Task.FromResult<ConstraintFailure?>(new ConstraintFailure("write_scope", "Tag is outside the API key write scope."));
+            return ValueTask.FromResult<ConstraintFailure?>(new ConstraintFailure("write_scope", "Tag is outside the API key write scope."));
         }
 
         if (constraints.MaxWriteClassification is { } maxClassification)
@@ -96,18 +96,18 @@ public sealed class ConstraintEnforcer(
             GalaxyAttribute? attribute = target.Attribute;
             if (attribute is null)
             {
-                return Task.FromResult<ConstraintFailure?>(new ConstraintFailure("max_write_classification", "Attribute security classification is not available."));
+                return ValueTask.FromResult<ConstraintFailure?>(new ConstraintFailure("max_write_classification", "Attribute security classification is not available."));
             }
 
             if (attribute.SecurityClassification > maxClassification)
             {
-                return Task.FromResult<ConstraintFailure?>(new ConstraintFailure(
+                return ValueTask.FromResult<ConstraintFailure?>(new ConstraintFailure(
                     "max_write_classification",
                     $"Attribute security classification {attribute.SecurityClassification} exceeds allowed maximum {maxClassification}."));
             }
         }
 
-        return Task.FromResult<ConstraintFailure?>(null);
+        return ValueTask.FromResult<ConstraintFailure?>(null);
     }
 
     /// <inheritdoc />
@@ -150,6 +150,46 @@ public sealed class ConstraintEnforcer(
         await auditWriter.WriteAsync(auditEvent, cancellationToken).ConfigureAwait(false);
     }
 
+    /// <inheritdoc />
+    public async Task RecordBulkDenialAsync(
+        ApiKeyIdentity? identity,
+        string commandKind,
+        ConstraintDenialSummary summary,
+        string? correlationId,
+        CancellationToken cancellationToken)
+    {
+        ArgumentNullException.ThrowIfNull(summary);
+        if (summary.DeniedCount == 0)
+        {
+            return;
+        }
+
+        // One event for the whole command. Target stays "<commandKind>:<target>"-shaped so audit
+        // queries that split on ':' keep working; the "bulk" target marks the aggregated form.
+        AuditEvent auditEvent = new()
+        {
+            EventId = Guid.NewGuid(),
+            OccurredAtUtc = DateTimeOffset.UtcNow,
+            Actor = identity?.KeyId ?? "anonymous",
+            Action = "constraint-denied",
+            Outcome = AuditOutcome.Denied,
+            Category = CanonicalForwardingApiKeyAuditStore.ApiKeyCategory,
+            Target = $"{commandKind}:bulk",
+            SourceNode = null,
+            CorrelationId = Guid.TryParse(correlationId, out var cid) ? cid : (Guid?)null,
+            DetailsJson = JsonSerializer.Serialize(new BulkDenialDetails(
+                summary.ConstraintNames,
+                commandKind,
+                summary.DeniedCount,
+                summary.TotalCount,
+                summary.SampleTargets,
+                summary.SampleTruncated,
+                correlationId ?? "")),
+        };
+
+        await auditWriter.WriteAsync(auditEvent, cancellationToken).ConfigureAwait(false);
+    }
+
     private ConstraintFailure? CheckReadTarget(
         ApiKeyConstraints constraints,
         string tagAddress)
@@ -240,4 +280,15 @@ public sealed class ConstraintEnforcer(
 
         return false;
     }
+
+    // DetailsJson shape of an aggregated bulk denial. Serialized with camelCase names to match the
+    // single-item envelope's keys ("commandKind", "clientCorrelationId").
+    private sealed record BulkDenialDetails(
+        [property: System.Text.Json.Serialization.JsonPropertyName("constraints")] IReadOnlyList<string> Constraints,
+        [property: System.Text.Json.Serialization.JsonPropertyName("commandKind")] string CommandKind,
+        [property: System.Text.Json.Serialization.JsonPropertyName("deniedCount")] int DeniedCount,
+        [property: System.Text.Json.Serialization.JsonPropertyName("totalCount")] int TotalCount,
+        [property: System.Text.Json.Serialization.JsonPropertyName("sampleTargets")] IReadOnlyList<string> SampleTargets,
+        [property: System.Text.Json.Serialization.JsonPropertyName("sampleTruncated")] bool SampleTruncated,
+        [property: System.Text.Json.Serialization.JsonPropertyName("clientCorrelationId")] string ClientCorrelationId);
 }

src/ZB.MOM.WW.MxGateway.Server/Security/Authorization/GatewayGrpcAuthorizationInterceptor.cs

@@ -30,7 +30,10 @@ public sealed class GatewayGrpcAuthorizationInterceptor(
         ServerCallContext context,
         UnaryServerMethod<TRequest, TResponse> continuation)
     {
-        ApiKeyIdentity? identity = await AuthenticateAndAuthorizeAsync(request, context).ConfigureAwait(false);
+        ApiKeyIdentity? identity = await AuthenticateAndAuthorizeAsync(
+                scopeResolver.ResolveRequiredScope(request),
+                context)
+            .ConfigureAwait(false);
         IDisposable? identityScope = identity is null ? null : identityAccessor.Push(identity);
         using (identityScope)
         {
@@ -45,7 +48,10 @@ public sealed class GatewayGrpcAuthorizationInterceptor(
         ServerCallContext context,
         ServerStreamingServerMethod<TRequest, TResponse> continuation)
     {
-        ApiKeyIdentity? identity = await AuthenticateAndAuthorizeAsync(request, context).ConfigureAwait(false);
+        ApiKeyIdentity? identity = await AuthenticateAndAuthorizeAsync(
+                scopeResolver.ResolveRequiredScope(request),
+                context)
+            .ConfigureAwait(false);
         IDisposable? identityScope = identity is null ? null : identityAccessor.Push(identity);
         using (identityScope)
         {
@@ -53,15 +59,51 @@ public sealed class GatewayGrpcAuthorizationInterceptor(
         }
     }
 
+    // The base Interceptor passes client- and duplex-streaming calls straight to the handler, so
+    // without these overrides any such RPC (none exist today, but a .proto edit could add one, and
+    // the shared Galaxy Repository service is mapped through the same interceptor) would be
+    // reachable with no API key at all. Both run the identical authentication path. There is no
+    // single request message to derive a scope from before the handler reads the stream, so they
+    // fail closed to the scope the resolver assigns an unrecognized request: admin.
+
+    /// <inheritdoc />
+    public override async Task<TResponse> ClientStreamingServerHandler<TRequest, TResponse>(
+        IAsyncStreamReader<TRequest> requestStream,
+        ServerCallContext context,
+        ClientStreamingServerMethod<TRequest, TResponse> continuation)
+    {
+        ApiKeyIdentity? identity = await AuthenticateAndAuthorizeAsync(GatewayScopes.Admin, context)
+            .ConfigureAwait(false);
+        IDisposable? identityScope = identity is null ? null : identityAccessor.Push(identity);
+        using (identityScope)
+        {
+            return await continuation(requestStream, context).ConfigureAwait(false);
+        }
+    }
+
+    /// <inheritdoc />
+    public override async Task DuplexStreamingServerHandler<TRequest, TResponse>(
+        IAsyncStreamReader<TRequest> requestStream,
+        IServerStreamWriter<TResponse> responseStream,
+        ServerCallContext context,
+        DuplexStreamingServerMethod<TRequest, TResponse> continuation)
+    {
+        ApiKeyIdentity? identity = await AuthenticateAndAuthorizeAsync(GatewayScopes.Admin, context)
+            .ConfigureAwait(false);
+        IDisposable? identityScope = identity is null ? null : identityAccessor.Push(identity);
+        using (identityScope)
+        {
+            await continuation(requestStream, responseStream, context).ConfigureAwait(false);
+        }
+    }
+
     /// <summary>Authenticates the API key and authorizes the RPC call by required scope.</summary>
-    /// <typeparam name="TRequest">Request message type.</typeparam>
-    /// <param name="request">Request payload.</param>
+    /// <param name="requiredScope">The scope the call requires, resolved by the caller.</param>
     /// <param name="context">RPC server call context.</param>
     /// <returns>Authenticated API key identity, or null if authentication is disabled.</returns>
-    private async Task<ApiKeyIdentity?> AuthenticateAndAuthorizeAsync<TRequest>(
-        TRequest request,
+    private async Task<ApiKeyIdentity?> AuthenticateAndAuthorizeAsync(
+        string requiredScope,
         ServerCallContext context)
-        where TRequest : class
     {
         if (options.Value.Authentication.Mode == AuthenticationMode.Disabled)
         {
@@ -116,7 +158,6 @@ public sealed class GatewayGrpcAuthorizationInterceptor(
 
         ApiKeyIdentity identity = GatewayApiKeyIdentityMapper.ToGatewayIdentity(verification.Identity);
 
-        string requiredScope = scopeResolver.ResolveRequiredScope(request);
         if (!identity.Scopes.Contains(requiredScope))
         {
             throw new RpcException(new Status(

src/ZB.MOM.WW.MxGateway.Server/Security/Authorization/IConstraintEnforcer.cs

@@ -3,6 +3,13 @@ using ZB.MOM.WW.MxGateway.Server.Sessions;
 
 namespace ZB.MOM.WW.MxGateway.Server.Security.Authorization;
 
+/// <summary>Enforces API-key read/write constraints and records constraint denials for audit.</summary>
+/// <remarks>
+///     The per-item checks return <see cref="ValueTask{TResult}"/> because every production check is
+///     synchronous (an in-memory Galaxy hierarchy lookup) and runs once per item of a bulk command:
+///     a <see cref="Task{TResult}"/> return allocated a task per item for a value that was already
+///     known. An implementation that genuinely needs to await still can.
+/// </remarks>
 public interface IConstraintEnforcer
 {
     /// <summary>
@@ -34,7 +41,7 @@ public interface IConstraintEnforcer
     /// <param name="tagAddress">Tag address to check.</param>
     /// <param name="cancellationToken">Token to observe for cancellation.</param>
     /// <returns>The constraint failure details if denied; otherwise null.</returns>
-    Task<ConstraintFailure?> CheckReadTagAsync(
+    ValueTask<ConstraintFailure?> CheckReadTagAsync(
         ApiKeyIdentity? identity,
         string tagAddress,
         CancellationToken cancellationToken);
@@ -46,7 +53,7 @@ public interface IConstraintEnforcer
     /// <param name="itemHandle">The MXAccess item handle.</param>
     /// <param name="cancellationToken">Token to observe for cancellation.</param>
     /// <returns>The constraint failure details if denied; otherwise null.</returns>
-    Task<ConstraintFailure?> CheckReadHandleAsync(
+    ValueTask<ConstraintFailure?> CheckReadHandleAsync(
         ApiKeyIdentity? identity,
         GatewaySession session,
         int serverHandle,
@@ -60,7 +67,7 @@ public interface IConstraintEnforcer
     /// <param name="itemHandle">The MXAccess item handle.</param>
     /// <param name="cancellationToken">Token to observe for cancellation.</param>
     /// <returns>The constraint failure details if denied; otherwise null.</returns>
-    Task<ConstraintFailure?> CheckWriteHandleAsync(
+    ValueTask<ConstraintFailure?> CheckWriteHandleAsync(
         ApiKeyIdentity? identity,
         GatewaySession session,
         int serverHandle,
@@ -85,4 +92,25 @@ public interface IConstraintEnforcer
         ConstraintFailure failure,
         string? correlationId,
         CancellationToken cancellationToken);
+
+    /// <summary>
+    ///     Records every constraint denial of one bulk command as a single aggregated audit event.
+    /// </summary>
+    /// <param name="identity">The API key identity.</param>
+    /// <param name="commandKind">The kind of bulk command whose items were denied.</param>
+    /// <param name="summary">The accumulated denials: count, constraint names and a bounded target sample.</param>
+    /// <param name="correlationId">The per-request client correlation id, if any (same handling as <see cref="RecordDenialAsync"/>).</param>
+    /// <param name="cancellationToken">Token to observe for cancellation.</param>
+    /// <returns>A task that represents the asynchronous operation.</returns>
+    /// <remarks>
+    ///     Bulk callers use this instead of one <see cref="RecordDenialAsync"/> per item: the audit
+    ///     channel is bounded and drops on full, so per-item events let one bulk command evict every
+    ///     other pending audit record.
+    /// </remarks>
+    Task RecordBulkDenialAsync(
+        ApiKeyIdentity? identity,
+        string commandKind,
+        ConstraintDenialSummary summary,
+        string? correlationId,
+        CancellationToken cancellationToken);
 }

src/ZB.MOM.WW.MxGateway.Server/Security/Authorization/MetricsAuthorization.cs

@@ -0,0 +1,222 @@
+using System.Security.Claims;
+using System.Text.Encodings.Web;
+using Microsoft.AspNetCore.Authentication;
+using Microsoft.AspNetCore.Authorization;
+using Microsoft.Extensions.Options;
+using ZB.MOM.WW.Auth.Abstractions.ApiKeys;
+using ZB.MOM.WW.MxGateway.Server.Configuration;
+using ZB.MOM.WW.MxGateway.Server.Dashboard;
+using ZB.MOM.WW.MxGateway.Server.Metrics;
+
+namespace ZB.MOM.WW.MxGateway.Server.Security.Authorization;
+
+/// <summary>Names shared by the optional <c>/metrics</c> authentication (<c>MxGateway:Metrics:RequireAuthentication</c>).</summary>
+public static class MetricsAuthorizationDefaults
+{
+    /// <summary>Authentication scheme that authenticates a metrics scrape.</summary>
+    public const string AuthenticationScheme = "MxGateway.Metrics";
+
+    /// <summary>Authorization policy attached to <c>/metrics</c> when authentication is required.</summary>
+    public const string Policy = "MxGateway.Metrics";
+
+    /// <summary>Claim type carrying one API-key scope on a principal the metrics scheme authenticated.</summary>
+    public const string ScopeClaimType = "mxgateway:scope";
+}
+
+/// <summary>
+/// Authenticates a <c>/metrics</c> scrape from either credential the gateway already issues: an API
+/// key presented as <c>Authorization: Bearer mxgw_…</c> (verified through the same cached
+/// <see cref="IApiKeyVerifier"/> and per-peer failure limiter as gRPC calls), or, when no
+/// authorization header is sent, the dashboard login cookie.
+/// </summary>
+/// <remarks>
+/// One scheme fronting both credentials keeps the challenge a plain <c>401</c> with
+/// <c>WWW-Authenticate: Bearer</c>, which is what a Prometheus scraper understands; listing the
+/// cookie scheme on the policy directly would turn an anonymous scrape into a 302 redirect to the
+/// login page. A throttled key partition answers <c>429</c>, mirroring the gRPC
+/// <c>ResourceExhausted</c>. Failure reasons are never echoed to the caller and the presented
+/// secret is never logged. The limiter's per-peer partition here is the remote IP, whereas the gRPC
+/// interceptor's is the gRPC peer string (address and port, i.e. per connection), so the two paths
+/// share only the per-key-id aggregate layer.
+/// </remarks>
+public sealed class MetricsAuthenticationHandler(
+    IOptionsMonitor<AuthenticationSchemeOptions> options,
+    ILoggerFactory logger,
+    UrlEncoder encoder,
+    IApiKeyVerifier apiKeyVerifier,
+    ApiKeyFailureLimiter failureLimiter,
+    IAuthenticationSchemeProvider schemeProvider,
+    IOptions<GatewayOptions> gatewayOptions,
+    GatewayMetrics metrics)
+    : AuthenticationHandler<AuthenticationSchemeOptions>(options, logger, encoder)
+{
+    private bool _throttled;
+
+    /// <inheritdoc />
+    protected override async Task<AuthenticateResult> HandleAuthenticateAsync()
+    {
+        string? authorizationHeader = Request.Headers.Authorization;
+        if (string.IsNullOrWhiteSpace(authorizationHeader))
+        {
+            return await AuthenticateDashboardCookieAsync().ConfigureAwait(false);
+        }
+
+        // Same throttle as the gRPC interceptor: checked before the store read so online guessing
+        // against /metrics cannot spend a verification per attempt or bypass the gRPC limiter.
+        ApiKeyThrottlePartition partition = new(
+            Context.Connection.RemoteIpAddress?.ToString() ?? "unknown",
+            GatewayGrpcAuthorizationInterceptor.TryResolveKeyId(authorizationHeader));
+        ApiKeyThrottleDecision decision = failureLimiter.Check(
+            partition,
+            out ApiKeyFailureLimiter.PartitionResolution resolution);
+        if (decision is ApiKeyThrottleDecision.ThrottledByPeer or ApiKeyThrottleDecision.ThrottledByAggregate)
+        {
+            _throttled = true;
+            metrics.RecordAuthThrottled(decision == ApiKeyThrottleDecision.ThrottledByPeer ? "peer" : "aggregate");
+            return AuthenticateResult.Fail("Too many authentication attempts.");
+        }
+
+        ApiKeyVerification verification = await apiKeyVerifier
+            .VerifyAsync(authorizationHeader, Context.RequestAborted)
+            .ConfigureAwait(false);
+        if (!verification.Succeeded || verification.Identity is null)
+        {
+            failureLimiter.RecordFailure(partition);
+            return AuthenticateResult.Fail("Missing or invalid API key.");
+        }
+
+        failureLimiter.Reset(partition, resolution);
+
+        List<Claim> claims = [new Claim(ClaimTypes.Name, verification.Identity.KeyId)];
+        foreach (string scope in verification.Identity.Scopes)
+        {
+            claims.Add(new Claim(MetricsAuthorizationDefaults.ScopeClaimType, scope));
+        }
+
+        ClaimsPrincipal principal = new(new ClaimsIdentity(claims, Scheme.Name));
+        return AuthenticateResult.Success(new AuthenticationTicket(principal, Scheme.Name));
+    }
+
+    /// <inheritdoc />
+    protected override Task HandleChallengeAsync(AuthenticationProperties properties)
+    {
+        if (_throttled)
+        {
+            Response.StatusCode = StatusCodes.Status429TooManyRequests;
+            return Task.CompletedTask;
+        }
+
+        Response.StatusCode = StatusCodes.Status401Unauthorized;
+        Response.Headers.WWWAuthenticate = "Bearer";
+        return Task.CompletedTask;
+    }
+
+    // No authorization header: fall back to the dashboard login cookie, so an operator already
+    // signed in to the dashboard can open /metrics in the same browser.
+    private async Task<AuthenticateResult> AuthenticateDashboardCookieAsync()
+    {
+        // Dashboard:DisableLogin registers an always-succeeding auto-login handler under the cookie
+        // scheme name. Falling back to it would let every anonymous request satisfy /metrics, so
+        // with DisableLogin on, RequireAuthentication=true accepts API keys only.
+        if (gatewayOptions.Value.Dashboard.DisableLogin)
+        {
+            return AuthenticateResult.NoResult();
+        }
+
+        if (await schemeProvider.GetSchemeAsync(DashboardAuthenticationDefaults.AuthenticationScheme).ConfigureAwait(false)
+            is null)
+        {
+            return AuthenticateResult.NoResult();
+        }
+
+        AuthenticateResult cookie = await Context
+            .AuthenticateAsync(DashboardAuthenticationDefaults.AuthenticationScheme)
+            .ConfigureAwait(false);
+        return cookie.Succeeded && cookie.Principal is not null
+            ? AuthenticateResult.Success(new AuthenticationTicket(cookie.Principal, Scheme.Name))
+            : AuthenticateResult.NoResult();
+    }
+}
+
+/// <summary>Requirement for <c>/metrics</c> access; see <see cref="MetricsAccessHandler"/>.</summary>
+public sealed class MetricsAccessRequirement : IAuthorizationRequirement
+{
+}
+
+/// <summary>
+/// Grants <c>/metrics</c> to an API key carrying the <c>admin</c> scope or to a dashboard login
+/// holding a real dashboard role (Administrator or Viewer). It reads no options, so neither
+/// anonymous localhost nor <c>Authentication:Mode=Disabled</c> can satisfy it.
+/// </summary>
+/// <remarks>
+/// A Viewer login qualifies because every dashboard role already sees the same aggregate counters
+/// on the dashboard itself; the metrics carry no per-session or per-tag data. An API key needs
+/// <c>admin</c> because metrics are an operator surface, not one of the data-plane scopes a client
+/// key is normally issued with.
+/// </remarks>
+public sealed class MetricsAccessHandler : AuthorizationHandler<MetricsAccessRequirement>
+{
+    /// <inheritdoc />
+    protected override Task HandleRequirementAsync(
+        AuthorizationHandlerContext context,
+        MetricsAccessRequirement requirement)
+    {
+        foreach (ClaimsIdentity identity in context.User.Identities)
+        {
+            if (!identity.IsAuthenticated)
+            {
+                continue;
+            }
+
+            // Scope claims count only on an identity the metrics scheme minted from a verified key.
+            bool isApiKeyIdentity = string.Equals(
+                identity.AuthenticationType,
+                MetricsAuthorizationDefaults.AuthenticationScheme,
+                StringComparison.Ordinal);
+            if (isApiKeyIdentity
+                && identity.HasClaim(MetricsAuthorizationDefaults.ScopeClaimType, GatewayScopes.Admin))
+            {
+                context.Succeed(requirement);
+                return Task.CompletedTask;
+            }
+
+            if (!isApiKeyIdentity
+                && (identity.HasClaim(identity.RoleClaimType, DashboardRoles.Admin)
+                    || identity.HasClaim(identity.RoleClaimType, DashboardRoles.Viewer)))
+            {
+                context.Succeed(requirement);
+                return Task.CompletedTask;
+            }
+        }
+
+        return Task.CompletedTask;
+    }
+}
+
+/// <summary>Registration for the optional <c>/metrics</c> authentication.</summary>
+public static class MetricsAuthorizationServiceCollectionExtensions
+{
+    /// <summary>
+    /// Registers the metrics authentication scheme, the <see cref="MetricsAuthorizationDefaults.Policy"/>
+    /// policy and its handler. Registration is unconditional and inert: the policy only takes effect
+    /// when <c>MxGateway:Metrics:RequireAuthentication</c> attaches it to the endpoint.
+    /// </summary>
+    /// <param name="services">Service collection to register into.</param>
+    /// <returns>The same service collection, for chaining.</returns>
+    public static IServiceCollection AddGatewayMetricsAuthorization(this IServiceCollection services)
+    {
+        services.AddAuthentication()
+            .AddScheme<AuthenticationSchemeOptions, MetricsAuthenticationHandler>(
+                MetricsAuthorizationDefaults.AuthenticationScheme,
+                _ => { });
+        services.AddAuthorizationBuilder()
+            .AddPolicy(
+                MetricsAuthorizationDefaults.Policy,
+                policy => policy
+                    .AddAuthenticationSchemes(MetricsAuthorizationDefaults.AuthenticationScheme)
+                    .AddRequirements(new MetricsAccessRequirement()));
+        services.AddSingleton<IAuthorizationHandler, MetricsAccessHandler>();
+
+        return services;
+    }
+}

src/ZB.MOM.WW.MxGateway.Server/Workers/WorkerClient.cs

@@ -705,7 +705,7 @@ public sealed class WorkerClient : IWorkerClient
     {
         if (workerEvent.Event is not null)
         {
-            _metrics?.EventReceived(SessionId, workerEvent.Event.Family.ToString());
+            _metrics?.EventReceived(SessionId, MxEventFamilyNames.Get(workerEvent.Event.Family));
         }
 
         if (_eventStaging.Writer.TryWrite(workerEvent))

src/ZB.MOM.WW.MxGateway.Tests/Alarms/AlarmTruncationSignalTests.cs

@@ -372,6 +372,105 @@ public sealed class AlarmTruncationSignalTests
         await monitor.StopAsync(CancellationToken.None);
     }
 
+    /// <summary>
+    ///     A subscriber that stops reading is dropped once its bounded queue fills, and the fault it
+    ///     observes is the typed <see cref="AlarmFeedSubscriberOverflowException"/> — the signal the
+    ///     gRPC layer maps to ResourceExhausted — rather than a bare InvalidOperationException.
+    /// </summary>
+    /// <returns>A task that represents the asynchronous operation.</returns>
+    [Fact]
+    public async Task SubscriberThatStopsReading_IsDroppedWithTypedOverflowFault()
+    {
+        // Comfortably past the monitor's per-subscriber queue capacity (2048).
+        const int ProbeCount = 2300;
+        using GatewayMetrics metrics = new();
+        StubSessionManager sessions = new();
+        using GatewayAlarmMonitor monitor = CreateMonitor(sessions, metrics);
+
+        using CancellationTokenSource cts = new();
+        await monitor.StartAsync(cts.Token);
+        await sessions.WaitForReconcileAsync(WaitTimeout);
+
+        // The slow subscriber: registered (first frame read) and then never read again.
+        using CancellationTokenSource slowCts = new(WaitTimeout);
+        await using IAsyncEnumerator<AlarmFeedMessage> slow =
+            monitor.StreamAsync(null, slowCts.Token).GetAsyncEnumerator(slowCts.Token);
+        Assert.True(await slow.MoveNextAsync());
+
+        // Raises on distinct references: each one is broadcast (no dedup) without forcing a
+        // reconcile, which keeps the burst cheap under a loaded parallel test run.
+        for (ulong sequence = 1; sequence <= ProbeCount; sequence++)
+        {
+            sessions.EmitEvent(UniqueRaise(sequence));
+        }
+
+        // The monitor caches each raise and broadcasts it under the same lock, so once the cache
+        // holds every raise, every broadcast has been offered to the slow subscriber. (A draining
+        // reader is no barrier here: under a loaded test run it can fall 2048 frames behind the
+        // burst and be dropped itself.)
+        await WaitUntilAsync(() => monitor.CurrentAlarms.Count >= ProbeCount, WaitTimeout);
+
+        // Drain what the slow subscriber had buffered; the terminal fault is the typed overflow.
+        await Assert.ThrowsAsync<AlarmFeedSubscriberOverflowException>(async () =>
+        {
+            while (await slow.MoveNextAsync())
+            {
+            }
+        });
+
+        await cts.CancelAsync();
+        await monitor.StopAsync(CancellationToken.None);
+    }
+
+    /// <summary>
+    ///     A StreamAsync attach reuses the memoized alarm-set projection that CurrentAlarms serves
+    ///     (no per-attach clone of the whole set under the broadcast lock) and still applies the
+    ///     subscriber's prefix filter to the snapshot it emits.
+    /// </summary>
+    /// <returns>A task that represents the asynchronous operation.</returns>
+    [Fact]
+    public async Task SubscriberAttach_ReusesMemoizedProjectionAndAppliesPrefixFilter()
+    {
+        using GatewayMetrics metrics = new();
+        StubSessionManager sessions = new()
+        {
+            Snapshots =
+            [
+                NewSnapshot("Galaxy!Area.Tank01.Level.HiHi", fromTruncatedSnapshot: false),
+                NewSnapshot("Galaxy!Area.Pump02.Flow.Lo", fromTruncatedSnapshot: false),
+            ],
+        };
+        using GatewayAlarmMonitor monitor = CreateMonitor(sessions, metrics);
+
+        using CancellationTokenSource cts = new();
+        await monitor.StartAsync(cts.Token);
+        await sessions.WaitForReconcileAsync(WaitTimeout);
+        await WaitUntilAsync(() => monitor.CurrentAlarms.Count == 2, WaitTimeout);
+        IReadOnlyList<ActiveAlarmSnapshot> projection = monitor.CurrentAlarms;
+
+        List<ActiveAlarmSnapshot> streamed = [];
+        using CancellationTokenSource streamCts = new(WaitTimeout);
+        await foreach (AlarmFeedMessage message in monitor.StreamAsync("Galaxy!Area.Tank01", streamCts.Token))
+        {
+            if (message.PayloadCase == AlarmFeedMessage.PayloadOneofCase.ActiveAlarm)
+            {
+                streamed.Add(message.ActiveAlarm);
+            }
+
+            if (message.PayloadCase == AlarmFeedMessage.PayloadOneofCase.SnapshotComplete)
+            {
+                break;
+            }
+        }
+
+        ActiveAlarmSnapshot only = Assert.Single(streamed);
+        Assert.Equal("Galaxy!Area.Tank01.Level.HiHi", only.AlarmFullReference);
+        Assert.Contains(projection, alarm => ReferenceEquals(alarm, only));
+
+        await cts.CancelAsync();
+        await monitor.StopAsync(CancellationToken.None);
+    }
+
     private static int CountOf(List<AlarmFeedMessage> received, AlarmFeedMessage.PayloadOneofCase payloadCase)
     {
         lock (received)
@@ -442,6 +541,23 @@ public sealed class AlarmTruncationSignalTests
         },
     };
 
+    // A Raise on a reference unique to the sequence, so the monitor never dedups it.
+    private static MxEvent UniqueRaise(ulong sequence) => new()
+    {
+        Family = MxEventFamily.OnAlarmTransition,
+        WorkerSequence = sequence,
+        OnAlarmTransition = new OnAlarmTransitionEvent
+        {
+            AlarmFullReference = $"Galaxy!Area.Tank{sequence}.Level.Hi",
+            SourceObjectReference = $"Tank{sequence}.Level",
+            AlarmTypeName = "AnalogLimitAlarm.Hi",
+            TransitionKind = AlarmTransitionKind.Raise,
+            Severity = 500,
+            SourceProvider = AlarmProviderMode.Alarmmgr,
+            TransitionTimestamp = Google.Protobuf.WellKnownTypes.Timestamp.FromDateTimeOffset(DateTimeOffset.UtcNow),
+        },
+    };
+
     private static ActiveAlarmSnapshot NewSnapshot(string reference, bool fromTruncatedSnapshot)
     {
         return new ActiveAlarmSnapshot

src/ZB.MOM.WW.MxGateway.Tests/Gateway/Grpc/EventStreamServiceTests.cs

@@ -263,7 +263,7 @@ public sealed class EventStreamServiceTests
         GatewayMetricsSnapshot snapshot = metrics.GetSnapshot();
         Assert.Equal(1, snapshot.QueueOverflows);
         Assert.Equal(1, snapshot.Faults);
-        // The finally block in StreamEventsAsync calls StreamDisconnected("Detached") on the
+        // The finally block in StreamEventsAsync records StreamDisconnected (reason Overflow) on the
         // overflow+fault path too; pin it here so a regression removing that call is caught.
         Assert.Equal(1, snapshot.StreamDisconnects);
     }
@@ -572,6 +572,142 @@ public sealed class EventStreamServiceTests
         }
     }
 
+    /// <summary>
+    ///     A stream that ends because the session's event stream completed records the
+    ///     <c>SessionClosed</c> disconnect reason rather than a blanket constant.
+    /// </summary>
+    /// <returns>A task that represents the asynchronous operation.</returns>
+    [Fact]
+    public async Task StreamEventsAsync_WhenEventStreamCompletes_RecordsSessionClosedReason()
+    {
+        FakeWorkerClient workerClient = new();
+        using GatewayMetrics metrics = new();
+        using DisconnectReasonRecorder reasons = new(metrics);
+        GatewaySession session = CreateReadySession(workerClient);
+        EventStreamService service = CreateService(new FakeSessionManager(session), metrics);
+        workerClient.Events.Add(CreateWorkerEvent(sequence: 1, MxEventFamily.OnDataChange));
+        workerClient.CompleteAfterConfiguredEvents = true;
+
+        await CollectEventsAsync(service, session.SessionId);
+
+        Assert.Equal([EventStreamDisconnectReasons.SessionClosed], reasons.Reasons);
+    }
+
+    /// <summary>A per-subscriber overflow records the <c>Overflow</c> disconnect reason.</summary>
+    /// <returns>A task that represents the asynchronous operation.</returns>
+    [Fact]
+    public async Task StreamEventsAsync_WhenStreamQueueOverflows_RecordsOverflowReason()
+    {
+        FakeWorkerClient workerClient = new();
+        using GatewayMetrics metrics = new();
+        using DisconnectReasonRecorder reasons = new(metrics);
+        GatewaySession session = CreateReadySession(
+            workerClient,
+            queueCapacity: 1,
+            metrics: metrics,
+            backpressurePolicy: EventBackpressurePolicy.DisconnectSubscriber);
+        EventStreamService service = CreateService(
+            new FakeSessionManager(session),
+            metrics,
+            queueCapacity: 1,
+            backpressurePolicy: EventBackpressurePolicy.DisconnectSubscriber);
+        for (ulong sequence = 1; sequence <= 50; sequence++)
+        {
+            workerClient.Events.Add(CreateWorkerEvent(sequence, MxEventFamily.OnDataChange));
+        }
+
+        workerClient.CompleteAfterConfiguredEvents = true;
+        await using (IAsyncEnumerator<MxEvent> subscriber = service
+            .StreamEventsAsync(CreateRequest(session.SessionId), callerKeyId: null, CancellationToken.None)
+            .GetAsyncEnumerator())
+        {
+            await Assert.ThrowsAsync<SessionManagerException>(
+                async () =>
+                {
+                    while (await subscriber.MoveNextAsync().AsTask().WaitAsync(TestTimeout))
+                    {
+                    }
+                });
+        }
+
+        Assert.Equal([EventStreamDisconnectReasons.Overflow], reasons.Reasons);
+    }
+
+    /// <summary>A terminal worker fault records the <c>WorkerFault</c> disconnect reason.</summary>
+    /// <returns>A task that represents the asynchronous operation.</returns>
+    [Fact]
+    public async Task StreamEventsAsync_WhenWorkerEventStreamFaults_RecordsWorkerFaultReason()
+    {
+        FakeWorkerClient workerClient = new()
+        {
+            TerminalException = new WorkerClientException(
+                WorkerClientErrorCode.WorkerFaulted,
+                "worker terminal fault"),
+        };
+        GatewaySession session = CreateReadySession(workerClient);
+        using GatewayMetrics metrics = new();
+        using DisconnectReasonRecorder reasons = new(metrics);
+        EventStreamService service = CreateService(new FakeSessionManager(session), metrics);
+        await using (IAsyncEnumerator<MxEvent> subscriber = service
+            .StreamEventsAsync(CreateRequest(session.SessionId), callerKeyId: null, CancellationToken.None)
+            .GetAsyncEnumerator())
+        {
+            await Assert.ThrowsAsync<WorkerClientException>(
+                async () => await subscriber.MoveNextAsync().AsTask().WaitAsync(TestTimeout));
+        }
+
+        Assert.Equal([EventStreamDisconnectReasons.WorkerFault], reasons.Reasons);
+    }
+
+    /// <summary>A client cancellation records the <c>ClientCancelled</c> disconnect reason.</summary>
+    /// <returns>A task that represents the asynchronous operation.</returns>
+    [Fact]
+    public async Task StreamEventsAsync_WhenCanceled_RecordsClientCancelledReason()
+    {
+        FakeWorkerClient workerClient = new();
+        GatewaySession session = CreateReadySession(workerClient);
+        using GatewayMetrics metrics = new();
+        using DisconnectReasonRecorder reasons = new(metrics);
+        EventStreamService service = CreateService(new FakeSessionManager(session), metrics);
+        using CancellationTokenSource cancellationTokenSource = new();
+        await using (IAsyncEnumerator<MxEvent> subscriber = service
+            .StreamEventsAsync(CreateRequest(session.SessionId), callerKeyId: null, cancellationTokenSource.Token)
+            .GetAsyncEnumerator(cancellationTokenSource.Token))
+        {
+            Task<bool> moveTask = subscriber.MoveNextAsync().AsTask();
+            await WaitUntilAsync(() => session.ActiveEventSubscriberCount == 1);
+            await cancellationTokenSource.CancelAsync();
+            await Assert.ThrowsAnyAsync<OperationCanceledException>(
+                async () => await moveTask.WaitAsync(TestTimeout));
+        }
+
+        Assert.Equal([EventStreamDisconnectReasons.ClientCancelled], reasons.Reasons);
+    }
+
+    /// <summary>
+    ///     A consumer that stops reading without cancelling (the gRPC writer gave up) keeps the
+    ///     <c>Detached</c> reason.
+    /// </summary>
+    /// <returns>A task that represents the asynchronous operation.</returns>
+    [Fact]
+    public async Task StreamEventsAsync_WhenConsumerStopsEarly_RecordsDetachedReason()
+    {
+        FakeWorkerClient workerClient = new();
+        GatewaySession session = CreateReadySession(workerClient);
+        using GatewayMetrics metrics = new();
+        using DisconnectReasonRecorder reasons = new(metrics);
+        EventStreamService service = CreateService(new FakeSessionManager(session), metrics);
+        workerClient.Events.Add(CreateWorkerEvent(sequence: 1, MxEventFamily.OnDataChange));
+        await using (IAsyncEnumerator<MxEvent> subscriber = service
+            .StreamEventsAsync(CreateRequest(session.SessionId), callerKeyId: null, CancellationToken.None)
+            .GetAsyncEnumerator())
+        {
+            await ReadNextAsync(subscriber);
+        }
+
+        Assert.Equal([EventStreamDisconnectReasons.Detached], reasons.Reasons);
+    }
+
     private static async Task<MxEvent> ReadNextAsync(IAsyncEnumerator<MxEvent> enumerator)
     {
         Assert.True(await enumerator.MoveNextAsync().AsTask().WaitAsync(TestTimeout));
@@ -896,4 +1032,63 @@ public sealed class EventStreamServiceTests
             return ValueTask.CompletedTask;
         }
     }
+
+    /// <summary>
+    ///     Captures the <c>reason</c> tag of every <c>mxgateway.grpc.streams.disconnected</c>
+    ///     measurement recorded on one <see cref="GatewayMetrics"/> instance (filtered by meter
+    ///     reference, so parallel tests sharing the meter name cannot cross-talk).
+    /// </summary>
+    private sealed class DisconnectReasonRecorder : IDisposable
+    {
+        private readonly System.Diagnostics.Metrics.MeterListener _listener = new();
+        private readonly List<string> _reasons = [];
+
+        /// <summary>Initializes a new instance of the <see cref="DisconnectReasonRecorder"/> class.</summary>
+        /// <param name="metrics">The metrics instance whose meter to observe.</param>
+        public DisconnectReasonRecorder(GatewayMetrics metrics)
+        {
+            _listener.InstrumentPublished = (instrument, listener) =>
+            {
+                if (ReferenceEquals(instrument.Meter, metrics.Meter)
+                    && instrument.Name == "mxgateway.grpc.streams.disconnected")
+                {
+                    listener.EnableMeasurementEvents(instrument);
+                }
+            };
+            _listener.SetMeasurementEventCallback<long>((instrument, _, tags, _) =>
+            {
+                if (!ReferenceEquals(instrument.Meter, metrics.Meter))
+                {
+                    return;
+                }
+
+                foreach (KeyValuePair<string, object?> tag in tags)
+                {
+                    if (tag.Key == "reason" && tag.Value is string reason)
+                    {
+                        lock (_reasons)
+                        {
+                            _reasons.Add(reason);
+                        }
+                    }
+                }
+            });
+            _listener.Start();
+        }
+
+        /// <summary>Gets the recorded reasons, in order.</summary>
+        public string[] Reasons
+        {
+            get
+            {
+                lock (_reasons)
+                {
+                    return [.. _reasons];
+                }
+            }
+        }
+
+        /// <inheritdoc />
+        public void Dispose() => _listener.Dispose();
+    }
 }

src/ZB.MOM.WW.MxGateway.Tests/Gateway/Grpc/MxAccessGatewayServiceConstraintTests.cs

@@ -87,8 +87,56 @@ public sealed class MxAccessGatewayServiceConstraintTests
         Assert.True(merged.Results[3].WasSuccessful);
         Assert.Equal("Tank04.Public", merged.Results[3].TagAddress);
         Assert.Equal(104, merged.Results[3].ItemHandle);
-        // Both denied tags recorded.
-        Assert.Equal(2, enforcer.RecordedDenials.Count);
+        // Both denied tags recorded in ONE aggregated bulk denial, not one audit call per item.
+        Assert.Empty(enforcer.RecordedDenials);
+        (string commandKind, ConstraintDenialSummary summary, _) = Assert.Single(enforcer.RecordedBulkDenials);
+        Assert.Equal(nameof(MxCommandKind.AddItemBulk), commandKind);
+        Assert.Equal(2, summary.DeniedCount);
+        Assert.Equal(4, summary.TotalCount);
+        Assert.Equal(["Tank01.Locked", "Tank03.Secret"], summary.SampleTargets);
+    }
+
+    /// <summary>
+    ///     Audit-flood guard: a bulk write whose every entry is denied records exactly one
+    ///     aggregated denial carrying the full denied count, the client correlation id and a sample
+    ///     capped at <see cref="ConstraintDenialSummary.MaxSampleTargets"/> — so one low-privilege
+    ///     key cannot push thousands of events into the bounded, drop-on-full audit channel.
+    /// </summary>
+    /// <returns>A task that represents the asynchronous operation.</returns>
+    [Fact]
+    public async Task Invoke_WriteBulk_WithManyDeniedEntries_RecordsOneAggregatedDenial()
+    {
+        const int EntryCount = 500;
+        PredicateConstraintEnforcer enforcer = new() { DenyWriteHandle = (_, _) => true };
+        FakeSessionManager sessionManager = CreateSessionManagerWithSeed();
+        MxAccessGatewayService service = CreateService(sessionManager, enforcer);
+        WriteBulkCommand payload = new() { ServerHandle = 7 };
+        for (int index = 0; index < EntryCount; index++)
+        {
+            payload.Entries.Add(new WriteBulkEntry { ItemHandle = index + 1 });
+        }
+
+        MxCommandRequest request = new()
+        {
+            SessionId = SessionId,
+            ClientCorrelationId = "bulk-flood-1",
+            Command = new MxCommand { Kind = MxCommandKind.WriteBulk, WriteBulk = payload },
+        };
+
+        MxCommandReply reply = await service.Invoke(request, new TestServerCallContext());
+
+        Assert.Equal(0, sessionManager.InvokeCount);
+        Assert.Equal(EntryCount, reply.WriteBulk.Results.Count);
+        Assert.Empty(enforcer.RecordedDenials);
+        (string commandKind, ConstraintDenialSummary summary, string? correlationId) =
+            Assert.Single(enforcer.RecordedBulkDenials);
+        Assert.Equal(nameof(MxCommandKind.WriteBulk), commandKind);
+        Assert.Equal("bulk-flood-1", correlationId);
+        Assert.Equal(EntryCount, summary.DeniedCount);
+        Assert.Equal(EntryCount, summary.TotalCount);
+        Assert.Equal(ConstraintDenialSummary.MaxSampleTargets, summary.SampleTargets.Count);
+        Assert.True(summary.SampleTruncated);
+        Assert.Equal(["write-handle"], summary.ConstraintNames);
     }
 
     /// <summary>

src/ZB.MOM.WW.MxGateway.Tests/Gateway/Grpc/MxAccessGatewayServiceTests.cs

@@ -2,9 +2,11 @@ using System.Diagnostics.Metrics;
 using System.Runtime.CompilerServices;
 using Google.Protobuf.WellKnownTypes;
 using Grpc.Core;
+using Microsoft.Extensions.Logging;
 using Microsoft.Extensions.Logging.Abstractions;
 using ZB.MOM.WW.MxGateway.Contracts;
 using ZB.MOM.WW.MxGateway.Contracts.Proto;
+using ZB.MOM.WW.MxGateway.Server.Alarms;
 using ZB.MOM.WW.MxGateway.Server.Grpc;
 using ZB.MOM.WW.MxGateway.Server.Metrics;
 using ZB.MOM.WW.MxGateway.Server.Security.Authentication;
@@ -213,6 +215,8 @@ public sealed class MxAccessGatewayServiceTests
         Assert.Equal(42, reply.AddItem.ItemHandle);
         Assert.Equal("status detail", Assert.Single(reply.Statuses).DiagnosticText);
         Assert.Equal("mxaccess diagnostic", reply.DiagnosticMessage);
+        // The reply echoes the caller's client_correlation_id, not the internal worker envelope id.
+        Assert.Equal("client-correlation", reply.CorrelationId);
     }
 
     /// <summary>Verifies that StreamEvents writes only events after the specified worker sequence.</summary>
@@ -463,11 +467,195 @@ public sealed class MxAccessGatewayServiceTests
         Assert.Contains("server-stream-active-alarms", reply.Capabilities);
     }
 
+    /// <summary>
+    ///     Every reply echoes the request's <c>client_correlation_id</c>, including when the caller
+    ///     sent none: the internal worker envelope id never reaches the client.
+    /// </summary>
+    /// <returns>A task that represents the asynchronous operation.</returns>
+    [Fact]
+    public async Task Invoke_WithoutClientCorrelationId_DoesNotLeakWorkerEnvelopeId()
+    {
+        FakeSessionManager sessionManager = new()
+        {
+            InvokeReply = new WorkerCommandReply
+            {
+                Reply = new MxCommandReply
+                {
+                    SessionId = "session-1",
+                    CorrelationId = "4f1d1b0e-internal-envelope-guid",
+                    Kind = MxCommandKind.AddItem,
+                    ProtocolStatus = MxAccessGrpcMapper.Ok(),
+                    AddItem = new AddItemReply { ItemHandle = 42 },
+                },
+            },
+        };
+        MxAccessGatewayService service = CreateService(sessionManager);
+        MxCommandRequest request = new()
+        {
+            SessionId = "session-1",
+            Command = new MxCommand
+            {
+                Kind = MxCommandKind.AddItem,
+                AddItem = new AddItemCommand { ServerHandle = 12, ItemDefinition = "Galaxy.Tag.Value" },
+            },
+        };
+
+        MxCommandReply reply = await service.Invoke(request, new TestServerCallContext());
+
+        Assert.Equal(string.Empty, reply.CorrelationId);
+    }
+
+    /// <summary>
+    ///     Invoke logs the client correlation id alongside the worker envelope id once, at Debug, so
+    ///     a client error report can be matched to gateway and worker logs.
+    /// </summary>
+    /// <returns>A task that represents the asynchronous operation.</returns>
+    [Fact]
+    public async Task Invoke_LogsClientToWorkerCorrelationPairAtDebug()
+    {
+        FakeSessionManager sessionManager = new()
+        {
+            InvokeReply = new WorkerCommandReply
+            {
+                Reply = new MxCommandReply
+                {
+                    SessionId = "session-1",
+                    CorrelationId = "envelope-7f3a",
+                    Kind = MxCommandKind.AddItem,
+                    ProtocolStatus = MxAccessGrpcMapper.Ok(),
+                    AddItem = new AddItemReply { ItemHandle = 42 },
+                },
+            },
+        };
+        RecordingLogger<MxAccessGatewayService> logger = new();
+        MxAccessGatewayService service = CreateService(sessionManager, logger: logger);
+        MxCommandRequest request = new()
+        {
+            SessionId = "session-1",
+            ClientCorrelationId = "client-trace-9",
+            Command = new MxCommand
+            {
+                Kind = MxCommandKind.AddItem,
+                AddItem = new AddItemCommand { ServerHandle = 12, ItemDefinition = "Galaxy.Tag.Value" },
+            },
+        };
+
+        await service.Invoke(request, new TestServerCallContext());
+
+        (LogLevel level, string message, _) = Assert.Single(
+            logger.Entries,
+            entry => entry.Message.Contains("envelope-7f3a", StringComparison.Ordinal));
+        Assert.Equal(LogLevel.Debug, level);
+        Assert.Contains("client-trace-9", message, StringComparison.Ordinal);
+    }
+
+    /// <summary>
+    ///     StreamEvents hands the call's cancellation token to every response write, so a write
+    ///     blocked on a slow or vanished client is released when the call is cancelled.
+    /// </summary>
+    /// <returns>A task that represents the asynchronous operation.</returns>
+    [Fact]
+    public async Task StreamEvents_PassesCallCancellationTokenToResponseWrites()
+    {
+        FakeSessionManager sessionManager = new();
+        sessionManager.Events.Add(CreateWorkerEvent("session-1", workerSequence: 1));
+        MxAccessGatewayService service = CreateService(sessionManager);
+        RecordingServerStreamWriter<MxEvent> writer = new();
+        using CancellationTokenSource callCts = new();
+
+        await service.StreamEvents(
+            new StreamEventsRequest { SessionId = "session-1" },
+            writer,
+            new TestServerCallContext(cancellationToken: callCts.Token));
+
+        Assert.Single(writer.Messages);
+        Assert.Equal(callCts.Token, writer.LastWriteCancellationToken);
+    }
+
+    /// <summary>
+    ///     A response write that fails with an <see cref="IOException"/> because the client went
+    ///     away (the call token is cancelled) maps to Cancelled and logs no warning: a client
+    ///     disconnect is routine, not a gateway failure.
+    /// </summary>
+    /// <returns>A task that represents the asynchronous operation.</returns>
+    [Fact]
+    public async Task StreamEvents_WhenWriteFailsAfterClientCancel_ReturnsCancelledWithoutWarning()
+    {
+        FakeSessionManager sessionManager = new();
+        sessionManager.Events.Add(CreateWorkerEvent("session-1", workerSequence: 1));
+        RecordingLogger<MxAccessGatewayService> logger = new();
+        MxAccessGatewayService service = CreateService(sessionManager, logger: logger);
+        using CancellationTokenSource callCts = new();
+        ThrowingStreamWriter<MxEvent> writer = new(() =>
+        {
+            callCts.Cancel();
+            return new IOException("The client reset the request stream.");
+        });
+
+        RpcException exception = await Assert.ThrowsAsync<RpcException>(
+            () => service.StreamEvents(
+                new StreamEventsRequest { SessionId = "session-1" },
+                writer,
+                new TestServerCallContext(cancellationToken: callCts.Token)));
+
+        Assert.Equal(StatusCode.Cancelled, exception.StatusCode);
+        Assert.DoesNotContain(logger.Entries, entry => entry.Level >= LogLevel.Warning);
+    }
+
+    /// <summary>
+    ///     The same IOException with a live call token is a genuine transport failure and keeps the
+    ///     existing Unavailable + warning mapping.
+    /// </summary>
+    /// <returns>A task that represents the asynchronous operation.</returns>
+    [Fact]
+    public async Task StreamEvents_WhenWriteFailsWithoutCancel_KeepsUnavailableMapping()
+    {
+        FakeSessionManager sessionManager = new();
+        sessionManager.Events.Add(CreateWorkerEvent("session-1", workerSequence: 1));
+        RecordingLogger<MxAccessGatewayService> logger = new();
+        MxAccessGatewayService service = CreateService(sessionManager, logger: logger);
+        ThrowingStreamWriter<MxEvent> writer = new(() => new IOException("Transport failure."));
+
+        RpcException exception = await Assert.ThrowsAsync<RpcException>(
+            () => service.StreamEvents(
+                new StreamEventsRequest { SessionId = "session-1" },
+                writer,
+                new TestServerCallContext()));
+
+        Assert.Equal(StatusCode.Unavailable, exception.StatusCode);
+        Assert.Contains(logger.Entries, entry => entry.Level == LogLevel.Warning);
+    }
+
+    /// <summary>
+    ///     A StreamAlarms subscriber the monitor drops for falling behind receives ResourceExhausted
+    ///     (the same status a StreamEvents overflow carries), and the gateway logs no
+    ///     warning-with-stack for what is a client-side slowness condition.
+    /// </summary>
+    /// <returns>A task that represents the asynchronous operation.</returns>
+    [Fact]
+    public async Task StreamAlarms_WhenSubscriberOverflows_ReturnsResourceExhaustedWithoutWarning()
+    {
+        RecordingLogger<MxAccessGatewayService> logger = new();
+        FakeGatewayAlarmService alarms = new() { StreamFault = new AlarmFeedSubscriberOverflowException() };
+        MxAccessGatewayService service = CreateService(new FakeSessionManager(), alarmService: alarms, logger: logger);
+
+        RpcException exception = await Assert.ThrowsAsync<RpcException>(
+            () => service.StreamAlarms(
+                new StreamAlarmsRequest(),
+                new RecordingServerStreamWriter<AlarmFeedMessage>(),
+                new TestServerCallContext()));
+
+        Assert.Equal(StatusCode.ResourceExhausted, exception.StatusCode);
+        Assert.DoesNotContain(logger.Entries, entry => entry.Level >= LogLevel.Warning);
+    }
+
     private static MxAccessGatewayService CreateService(
         FakeSessionManager sessionManager,
         IGatewayRequestIdentityAccessor? identityAccessor = null,
         GatewayMetrics? metrics = null,
-        IConstraintEnforcer? constraintEnforcer = null)
+        IConstraintEnforcer? constraintEnforcer = null,
+        IGatewayAlarmService? alarmService = null,
+        ILogger<MxAccessGatewayService>? logger = null)
     {
         return new MxAccessGatewayService(
             sessionManager,
@@ -477,8 +665,8 @@ public sealed class MxAccessGatewayServiceTests
             new MxAccessGrpcMapper(),
             new FakeEventStreamService(sessionManager),
             metrics ?? new GatewayMetrics(),
-            NullLogger<MxAccessGatewayService>.Instance,
-            new FakeGatewayAlarmService());
+            logger ?? NullLogger<MxAccessGatewayService>.Instance,
+            alarmService ?? new FakeGatewayAlarmService());
     }
 
     private static ApiKeyIdentity CreateIdentity()
@@ -786,4 +974,18 @@ public sealed class MxAccessGatewayServiceTests
         }
     }
 
+
+    /// <summary>A stream writer whose token-bearing write throws the supplied exception.</summary>
+    /// <typeparam name="T">The streamed message type.</typeparam>
+    private sealed class ThrowingStreamWriter<T>(Func<Exception> createException) : IServerStreamWriter<T>
+    {
+        /// <inheritdoc />
+        public WriteOptions? WriteOptions { get; set; }
+
+        /// <inheritdoc />
+        public Task WriteAsync(T message) => throw createException();
+
+        /// <inheritdoc />
+        public Task WriteAsync(T message, CancellationToken cancellationToken) => throw createException();
+    }
 }

src/ZB.MOM.WW.MxGateway.Tests/Gateway/MetricsEndpointAuthenticationTests.cs

@@ -0,0 +1,182 @@
+using System.Net;
+using System.Net.Http;
+using System.Security.Claims;
+using Microsoft.AspNetCore.Authentication;
+using Microsoft.AspNetCore.Authentication.Cookies;
+using Microsoft.AspNetCore.Builder;
+using Microsoft.Extensions.DependencyInjection;
+using Microsoft.Extensions.Options;
+using ZB.MOM.WW.Auth.ApiKeys.Admin;
+using ZB.MOM.WW.MxGateway.Server;
+using ZB.MOM.WW.MxGateway.Server.Dashboard;
+using ZB.MOM.WW.MxGateway.Server.Security.Authorization;
+
+namespace ZB.MOM.WW.MxGateway.Tests.Gateway;
+
+/// <summary>
+/// Host-level coverage of <c>MxGateway:Metrics:RequireAuthentication=true</c> (SEC-14): each case
+/// builds the real gateway with <see cref="GatewayApplication.Build"/> and drives the full
+/// middleware pipeline (authentication, authorization, the dashboard cookie scheme, the API-key
+/// store) over HTTP on an ephemeral loopback port. Requests therefore arrive from loopback, which
+/// is exactly the anonymous-localhost case the dashboard Viewer policy would admit and /metrics
+/// must not.
+/// </summary>
+public sealed class MetricsEndpointAuthenticationTests
+{
+    // A fixed test pepper: the configuration-backed pepper provider reads MxGateway:ApiKeyPepper,
+    // and the auth store itself is isolated per test process by TestHostEnvironmentInitializer.
+    private const string TestPepper = "metrics-endpoint-tests-pepper-0123456789abcdef";
+
+    /// <summary>
+    /// An anonymous loopback scrape gets a plain 401 Bearer challenge — not a 302 to /login, and not
+    /// the anonymous-localhost admission the dashboard Viewer policy grants.
+    /// </summary>
+    /// <returns>A task that represents the asynchronous operation.</returns>
+    [Fact]
+    public async Task AnonymousLoopbackScrape_Returns401BearerNotLoginRedirect()
+    {
+        await using WebApplication app = await StartAsync();
+        using HttpClient client = CreateClient(app);
+
+        using HttpResponseMessage response = await client.GetAsync("/metrics");
+
+        Assert.Equal(HttpStatusCode.Unauthorized, response.StatusCode);
+        Assert.Null(response.Headers.Location);
+        Assert.Contains(response.Headers.WwwAuthenticate, value => value.Scheme == "Bearer");
+    }
+
+    /// <summary>/health/active stays anonymous: the off-repo Overview dashboard polls it without credentials.</summary>
+    /// <returns>A task that represents the asynchronous operation.</returns>
+    [Fact]
+    public async Task HealthActive_StaysAnonymous()
+    {
+        await using WebApplication app = await StartAsync();
+        using HttpClient client = CreateClient(app);
+
+        using HttpResponseMessage response = await client.GetAsync("/health/active");
+
+        Assert.Equal(HttpStatusCode.OK, response.StatusCode);
+    }
+
+    /// <summary>A dashboard login cookie holding the Viewer role is accepted.</summary>
+    /// <returns>A task that represents the asynchronous operation.</returns>
+    [Fact]
+    public async Task DashboardViewerCookie_Returns200()
+    {
+        await using WebApplication app = await StartAsync();
+        using HttpClient client = CreateClient(app);
+        using HttpRequestMessage request = new(HttpMethod.Get, "/metrics");
+        request.Headers.Add("Cookie", MintDashboardCookie(app, DashboardRoles.Viewer));
+
+        using HttpResponseMessage response = await client.SendAsync(request);
+
+        Assert.Equal(HttpStatusCode.OK, response.StatusCode);
+    }
+
+    /// <summary>A real API key with the admin scope is accepted.</summary>
+    /// <returns>A task that represents the asynchronous operation.</returns>
+    [Fact]
+    public async Task AdminScopedApiKey_Returns200()
+    {
+        await using WebApplication app = await StartAsync();
+        string token = await CreateKeyAsync(app, [GatewayScopes.Admin]);
+        using HttpClient client = CreateClient(app);
+
+        using HttpResponseMessage response = await SendWithBearerAsync(client, token);
+
+        Assert.Equal(HttpStatusCode.OK, response.StatusCode);
+    }
+
+    /// <summary>A real API key without the admin scope authenticates but is forbidden.</summary>
+    /// <returns>A task that represents the asynchronous operation.</returns>
+    [Fact]
+    public async Task NonAdminApiKey_Returns403()
+    {
+        await using WebApplication app = await StartAsync();
+        string token = await CreateKeyAsync(app, [GatewayScopes.EventsRead, GatewayScopes.InvokeRead]);
+        using HttpClient client = CreateClient(app);
+
+        using HttpResponseMessage response = await SendWithBearerAsync(client, token);
+
+        Assert.Equal(HttpStatusCode.Forbidden, response.StatusCode);
+    }
+
+    /// <summary>
+    /// Dev-only <c>Dashboard:DisableLogin</c> registers an always-succeeding auto-login handler under
+    /// the cookie scheme name; /metrics must not fall back to it, or the flag would silently open
+    /// an endpoint the operator explicitly asked to protect.
+    /// </summary>
+    /// <returns>A task that represents the asynchronous operation.</returns>
+    [Fact]
+    public async Task DisableLogin_DoesNotOpenMetrics()
+    {
+        await using WebApplication app = await StartAsync("--MxGateway:Dashboard:DisableLogin=true");
+        using HttpClient client = CreateClient(app);
+
+        using HttpResponseMessage response = await client.GetAsync("/metrics");
+
+        Assert.Equal(HttpStatusCode.Unauthorized, response.StatusCode);
+    }
+
+    private static async Task<WebApplication> StartAsync(params string[] extraArgs)
+    {
+        // Ephemeral port (:0): xUnit runs collections in parallel, so a fixed port would collide.
+        WebApplication app = GatewayApplication.Build(
+        [
+            "--urls=http://127.0.0.1:0",
+            "--MxGateway:Metrics:RequireAuthentication=true",
+            $"--MxGateway:ApiKeyPepper={TestPepper}",
+            .. extraArgs,
+        ]);
+        await app.StartAsync();
+        return app;
+    }
+
+    private static HttpClient CreateClient(WebApplication app) =>
+        new(new HttpClientHandler { AllowAutoRedirect = false, UseCookies = false })
+        {
+            BaseAddress = new Uri(app.Urls.First()),
+        };
+
+    private static async Task<HttpResponseMessage> SendWithBearerAsync(HttpClient client, string token)
+    {
+        using HttpRequestMessage request = new(HttpMethod.Get, "/metrics");
+        request.Headers.TryAddWithoutValidation("Authorization", $"Bearer {token}");
+        return await client.SendAsync(request);
+    }
+
+    // Creates a key through the same admin command set the CLI and dashboard use. Key ids are unique
+    // per call because the isolated auth store is shared by every test in the process.
+    private static async Task<string> CreateKeyAsync(WebApplication app, string[] scopes)
+    {
+        ApiKeyAdminCommands commands = app.Services.GetRequiredService<ApiKeyAdminCommands>();
+        await commands.InitDbAsync(remoteAddress: null, CancellationToken.None);
+        CreateKeyResult created = await commands.CreateKeyAsync(
+            $"metrics{Guid.NewGuid():N}",
+            "Metrics test key",
+            new HashSet<string>(scopes, StringComparer.Ordinal),
+            null,
+            null,
+            remoteAddress: null,
+            CancellationToken.None);
+        return created.Token ?? throw new InvalidOperationException("Key creation returned no token.");
+    }
+
+    // Protects a ticket with the running host's own cookie options (data-protection keys, ticket
+    // format, cookie name) — the cookie a real dashboard login would issue, minus the LDAP bind.
+    private static string MintDashboardCookie(WebApplication app, string role)
+    {
+        CookieAuthenticationOptions cookieOptions = app.Services
+            .GetRequiredService<IOptionsMonitor<CookieAuthenticationOptions>>()
+            .Get(DashboardAuthenticationDefaults.AuthenticationScheme);
+        ClaimsIdentity identity = new(
+            [new Claim(ClaimTypes.Name, "viewer-user"), new Claim(ClaimTypes.Role, role)],
+            DashboardAuthenticationDefaults.AuthenticationScheme,
+            ClaimTypes.Name,
+            ClaimTypes.Role);
+        AuthenticationTicket ticket = new(
+            new ClaimsPrincipal(identity),
+            DashboardAuthenticationDefaults.AuthenticationScheme);
+        return $"{cookieOptions.Cookie.Name}={cookieOptions.TicketDataFormat.Protect(ticket)}";
+    }
+}

src/ZB.MOM.WW.MxGateway.Tests/Metrics/MxEventFamilyNamesTests.cs

@@ -0,0 +1,41 @@
+using ZB.MOM.WW.MxGateway.Contracts.Proto;
+using ZB.MOM.WW.MxGateway.Server.Metrics;
+
+namespace ZB.MOM.WW.MxGateway.Tests.Metrics;
+
+public sealed class MxEventFamilyNamesTests
+{
+    /// <summary>
+    ///     Every defined family maps to exactly the name <see cref="Enum.ToString()"/> produces, so
+    ///     switching the per-event metric tags to the cached lookup changes no exported tag value.
+    /// </summary>
+    [Fact]
+    public void Get_ForEveryDefinedFamily_MatchesEnumToString()
+    {
+        foreach (MxEventFamily family in Enum.GetValues<MxEventFamily>())
+        {
+            Assert.Equal(family.ToString(), MxEventFamilyNames.Get(family));
+        }
+    }
+
+    /// <summary>A defined family returns the same cached string instance on every call (no per-event allocation).</summary>
+    [Fact]
+    public void Get_ForDefinedFamily_ReturnsCachedInstance()
+    {
+        Assert.Same(
+            MxEventFamilyNames.Get(MxEventFamily.OnDataChange),
+            MxEventFamilyNames.Get(MxEventFamily.OnDataChange));
+    }
+
+    /// <summary>
+    ///     A numeric value the contract does not define (a newer worker, or a malformed frame) still
+    ///     produces the <see cref="Enum.ToString()"/> form rather than throwing.
+    /// </summary>
+    [Fact]
+    public void Get_ForUndefinedValue_FallsBackToToString()
+    {
+        MxEventFamily undefined = (MxEventFamily)987654;
+
+        Assert.Equal(undefined.ToString(), MxEventFamilyNames.Get(undefined));
+    }
+}

src/ZB.MOM.WW.MxGateway.Tests/Security/Authentication/CachingApiKeyVerifierTests.cs

@@ -129,6 +129,120 @@ public sealed class CachingApiKeyVerifierTests
         Assert.Equal(2, inner.CallCount);
     }
 
+    /// <summary>
+    /// Single-flight: concurrent misses for the same token (a TTL expiry under load) share one
+    /// inner verification instead of each issuing its own SQLite read.
+    /// </summary>
+    /// <returns>A task that represents the asynchronous operation.</returns>
+    [Fact]
+    public async Task VerifyAsync_ConcurrentMisses_ShareOneInnerVerification()
+    {
+        GatedVerifier inner = new(Success("operator01"));
+        using MemoryCache cache = NewCache();
+        CachingApiKeyVerifier verifier = new(inner, cache, TimeSpan.FromSeconds(30));
+
+        Task<ApiKeyVerification> first = verifier.VerifyAsync(Header, CancellationToken.None);
+        await inner.Entered;
+        Task<ApiKeyVerification>[] joiners = Enumerable
+            .Range(0, 16)
+            .Select(_ => verifier.VerifyAsync(Header, CancellationToken.None))
+            .ToArray();
+
+        inner.Release();
+        ApiKeyVerification[] results = await Task.WhenAll([first, .. joiners]);
+
+        Assert.All(results, result => Assert.True(result.Succeeded));
+        Assert.Equal(1, inner.CallCount);
+    }
+
+    /// <summary>
+    /// A miss that arrives after <see cref="IApiKeyCacheInvalidator.Invalidate"/> must not join a
+    /// verification that started before it: that flight may return the pre-revoke identity, so the
+    /// late caller starts its own flight against the inner verifier.
+    /// </summary>
+    /// <returns>A task that represents the asynchronous operation.</returns>
+    [Fact]
+    public async Task VerifyAsync_MissAfterInvalidate_DoesNotJoinPreInvalidateFlight()
+    {
+        GatedVerifier inner = new(Success("operator01"));
+        using MemoryCache cache = NewCache();
+        CachingApiKeyVerifier verifier = new(inner, cache, TimeSpan.FromSeconds(30));
+
+        Task<ApiKeyVerification> preRevoke = verifier.VerifyAsync(Header, CancellationToken.None);
+        await inner.Entered;
+        ((IApiKeyCacheInvalidator)verifier).Invalidate("operator01");
+
+        // The gated first call is still parked; a joined call would park with it.
+        ApiKeyVerification postRevoke = await verifier
+            .VerifyAsync(Header, CancellationToken.None)
+            .WaitAsync(TimeSpan.FromSeconds(5));
+
+        Assert.True(postRevoke.Succeeded);
+        Assert.Equal(2, inner.CallCount);
+        inner.Release();
+        await preRevoke;
+    }
+
+    /// <summary>
+    /// A caller whose own token is cancelled stops waiting, but the shared verification it joined
+    /// is not cancelled for the other waiters.
+    /// </summary>
+    /// <returns>A task that represents the asynchronous operation.</returns>
+    [Fact]
+    public async Task VerifyAsync_CancelledWaiter_DoesNotCancelSharedVerification()
+    {
+        GatedVerifier inner = new(Success("operator01"));
+        using MemoryCache cache = NewCache();
+        CachingApiKeyVerifier verifier = new(inner, cache, TimeSpan.FromSeconds(30));
+        using CancellationTokenSource cancelled = new();
+
+        Task<ApiKeyVerification> survivor = verifier.VerifyAsync(Header, CancellationToken.None);
+        await inner.Entered;
+        Task<ApiKeyVerification> abandoned = verifier.VerifyAsync(Header, cancelled.Token);
+        await cancelled.CancelAsync();
+
+        await Assert.ThrowsAnyAsync<OperationCanceledException>(() => abandoned);
+        inner.Release();
+        Assert.True((await survivor).Succeeded);
+        Assert.Equal(1, inner.CallCount);
+    }
+
+    /// <summary>
+    /// The key-id index is pruned when the cache evicts an entry on its own (expiry, compaction),
+    /// not only on <see cref="IApiKeyCacheInvalidator.Invalidate"/> — otherwise every token ever
+    /// verified leaves an index entry behind for the life of the process.
+    /// </summary>
+    /// <returns>A task that represents the asynchronous operation.</returns>
+    [Fact]
+    public async Task CacheEviction_PrunesKeyIdIndex()
+    {
+        FakeVerifier inner = new(Success("operator01"));
+        using MemoryCache cache = NewCache();
+        CachingApiKeyVerifier verifier = new(inner, cache, TimeSpan.FromSeconds(30));
+
+        await verifier.VerifyAsync(Header, CancellationToken.None);
+        Assert.Equal(1, verifier.IndexedCacheKeyCount);
+
+        cache.Compact(1.0);
+
+        DateTime deadline = DateTime.UtcNow + TimeSpan.FromSeconds(5);
+        while (verifier.IndexedCacheKeyCount != 0 && DateTime.UtcNow < deadline)
+        {
+            await Task.Delay(10);
+        }
+
+        Assert.Equal(0, verifier.IndexedCacheKeyCount);
+        Assert.Equal(0, verifier.IndexedKeyIdCount);
+
+        // The evicted entry is gone, so the next request re-verifies, and Invalidate still reaches
+        // the repopulated entry through the index.
+        await verifier.VerifyAsync(Header, CancellationToken.None);
+        Assert.Equal(2, inner.CallCount);
+        ((IApiKeyCacheInvalidator)verifier).Invalidate("operator01");
+        await verifier.VerifyAsync(Header, CancellationToken.None);
+        Assert.Equal(3, inner.CallCount);
+    }
+
     /// <summary>
     /// The store decorator coalesces repeated <c>MarkUsed</c> writes for the same key inside the
     /// window down to a single forwarded write — the ≤1/min guarantee for <c>last_used_utc</c>.

src/ZB.MOM.WW.MxGateway.Tests/Security/Authorization/ConstraintEnforcerTests.cs

@@ -154,6 +154,70 @@ public sealed class ConstraintEnforcerTests
         Assert.Equal("Read:Secret.Tag", auditEvent.Target);
     }
 
+    /// <summary>
+    ///     A bulk denial is written as ONE audit event carrying the denied/total counts, the distinct
+    ///     constraint names and a sample capped at <see cref="ConstraintDenialSummary.MaxSampleTargets"/>,
+    ///     however many items were denied.
+    /// </summary>
+    /// <returns>A task that represents the asynchronous operation.</returns>
+    [Fact]
+    public async Task RecordBulkDenialAsync_WritesOneAggregatedEventWithBoundedSample()
+    {
+        ConstraintEnforcer enforcer = CreateEnforcer(out FakeAuditWriter auditWriter);
+        ConstraintDenialSummary summary = new(totalCount: 1000);
+        for (int index = 0; index < 999; index++)
+        {
+            summary.Add(
+                $"Area1.Tag{index}",
+                index % 2 == 0
+                    ? new ConstraintFailure("read_scope", "Tag is outside the API key read scope.")
+                    : new ConstraintFailure("tag_metadata", "Tag metadata is not available."));
+        }
+
+        await enforcer.RecordBulkDenialAsync(
+            CreateIdentity(ApiKeyConstraints.Empty),
+            "ReadBulk",
+            summary,
+            "trace-1",
+            CancellationToken.None);
+
+        AuditEvent auditEvent = Assert.Single(auditWriter.Events);
+        Assert.Equal("operator01", auditEvent.Actor);
+        Assert.Equal("constraint-denied", auditEvent.Action);
+        Assert.Equal(AuditOutcome.Denied, auditEvent.Outcome);
+        Assert.Equal("ReadBulk:bulk", auditEvent.Target);
+        Assert.Null(auditEvent.CorrelationId);
+
+        using JsonDocument details = JsonDocument.Parse(auditEvent.DetailsJson!);
+        JsonElement root = details.RootElement;
+        Assert.Equal(999, root.GetProperty("deniedCount").GetInt32());
+        Assert.Equal(1000, root.GetProperty("totalCount").GetInt32());
+        Assert.True(root.GetProperty("sampleTruncated").GetBoolean());
+        Assert.Equal(ConstraintDenialSummary.MaxSampleTargets, root.GetProperty("sampleTargets").GetArrayLength());
+        Assert.Equal(
+            ["read_scope", "tag_metadata"],
+            root.GetProperty("constraints").EnumerateArray().Select(e => e.GetString() ?? string.Empty).ToArray());
+        Assert.Equal("ReadBulk", root.GetProperty("commandKind").GetString());
+        Assert.Equal("trace-1", root.GetProperty("clientCorrelationId").GetString());
+    }
+
+    /// <summary>An empty summary (nothing denied) writes no audit event at all.</summary>
+    /// <returns>A task that represents the asynchronous operation.</returns>
+    [Fact]
+    public async Task RecordBulkDenialAsync_WithNoDenials_WritesNothing()
+    {
+        ConstraintEnforcer enforcer = CreateEnforcer(out FakeAuditWriter auditWriter);
+
+        await enforcer.RecordBulkDenialAsync(
+            identity: null,
+            "ReadBulk",
+            new ConstraintDenialSummary(totalCount: 3),
+            correlationId: null,
+            CancellationToken.None);
+
+        Assert.Empty(auditWriter.Events);
+    }
+
     /// <summary>Verifies that historized-only constraint requires historized attribute.</summary>
     /// <returns>A task that represents the asynchronous operation.</returns>
     [Fact]

src/ZB.MOM.WW.MxGateway.Tests/Security/Authorization/GatewayGrpcAuthorizationInterceptorTests.cs

@@ -717,6 +717,151 @@ public sealed class GatewayGrpcAuthorizationInterceptorTests
         Assert.Null(GatewayGrpcAuthorizationInterceptor.TryResolveKeyId($"Bearer mxgw_{overCap}_secret"));
     }
 
+    /// <summary>
+    ///     A client-streaming call is authenticated exactly like a unary call: a missing key is
+    ///     rejected with Unauthenticated and the handler never runs. The base Interceptor passes
+    ///     client- and duplex-streaming calls straight through, so without the override any such RPC
+    ///     added to the service would be reachable unauthenticated.
+    /// </summary>
+    /// <returns>A task that represents the asynchronous operation.</returns>
+    [Fact]
+    public async Task ClientStreamingServerHandler_MissingApiKey_ReturnsUnauthenticatedWithoutRunningHandler()
+    {
+        GatewayGrpcAuthorizationInterceptor interceptor = CreateInterceptor(
+            new FakeApiKeyVerifier(Failure(ApiKeyFailure.MissingOrMalformed)),
+            new GatewayRequestIdentityAccessor());
+        bool handlerRan = false;
+
+        RpcException exception = await Assert.ThrowsAsync<RpcException>(
+            () => interceptor.ClientStreamingServerHandler<MxCommandRequest, MxCommandReply>(
+                new EmptyStreamReader<MxCommandRequest>(),
+                new TestServerCallContext([]),
+                (_, _) =>
+                {
+                    handlerRan = true;
+                    return Task.FromResult(new MxCommandReply());
+                }));
+
+        Assert.Equal(StatusCode.Unauthenticated, exception.StatusCode);
+        Assert.False(handlerRan);
+    }
+
+    /// <summary>
+    ///     A client-streaming call has no single request message to derive a scope from, so the
+    ///     interceptor fails closed to the <c>admin</c> scope, the resolver's default for an
+    ///     unrecognized request.
+    /// </summary>
+    /// <returns>A task that represents the asynchronous operation.</returns>
+    [Fact]
+    public async Task ClientStreamingServerHandler_KeyWithoutAdminScope_ReturnsPermissionDenied()
+    {
+        GatewayGrpcAuthorizationInterceptor interceptor = CreateInterceptor(
+            new FakeApiKeyVerifier(SuccessWithScopes(GatewayScopes.InvokeRead, GatewayScopes.InvokeWrite)),
+            new GatewayRequestIdentityAccessor());
+
+        RpcException exception = await Assert.ThrowsAsync<RpcException>(
+            () => interceptor.ClientStreamingServerHandler<MxCommandRequest, MxCommandReply>(
+                new EmptyStreamReader<MxCommandRequest>(),
+                ContextWithAuthorization("Bearer mxgw_operator01_secret"),
+                (_, _) => Task.FromResult(new MxCommandReply())));
+
+        Assert.Equal(StatusCode.PermissionDenied, exception.StatusCode);
+        Assert.Contains(GatewayScopes.Admin, exception.Status.Detail, StringComparison.Ordinal);
+    }
+
+    /// <summary>An admin-scoped key reaches the client-streaming handler with its identity pushed.</summary>
+    /// <returns>A task that represents the asynchronous operation.</returns>
+    [Fact]
+    public async Task ClientStreamingServerHandler_AdminScope_RunsHandlerWithIdentity()
+    {
+        GatewayRequestIdentityAccessor identityAccessor = new();
+        GatewayGrpcAuthorizationInterceptor interceptor = CreateInterceptor(
+            new FakeApiKeyVerifier(SuccessWithScopes(GatewayScopes.Admin)),
+            identityAccessor);
+        string? keyIdSeenByHandler = null;
+
+        await interceptor.ClientStreamingServerHandler<MxCommandRequest, MxCommandReply>(
+            new EmptyStreamReader<MxCommandRequest>(),
+            ContextWithAuthorization("Bearer mxgw_operator01_secret"),
+            (_, _) =>
+            {
+                keyIdSeenByHandler = identityAccessor.Current?.KeyId;
+                return Task.FromResult(new MxCommandReply());
+            });
+
+        Assert.Equal("operator01", keyIdSeenByHandler);
+        Assert.Null(identityAccessor.Current);
+    }
+
+    /// <summary>A duplex-streaming call with no key is rejected before the handler runs.</summary>
+    /// <returns>A task that represents the asynchronous operation.</returns>
+    [Fact]
+    public async Task DuplexStreamingServerHandler_MissingApiKey_ReturnsUnauthenticatedWithoutRunningHandler()
+    {
+        GatewayGrpcAuthorizationInterceptor interceptor = CreateInterceptor(
+            new FakeApiKeyVerifier(Failure(ApiKeyFailure.MissingOrMalformed)),
+            new GatewayRequestIdentityAccessor());
+        bool handlerRan = false;
+
+        RpcException exception = await Assert.ThrowsAsync<RpcException>(
+            () => interceptor.DuplexStreamingServerHandler<MxCommandRequest, MxCommandReply>(
+                new EmptyStreamReader<MxCommandRequest>(),
+                new RecordingServerStreamWriter<MxCommandReply>(),
+                new TestServerCallContext([]),
+                (_, _, _) =>
+                {
+                    handlerRan = true;
+                    return Task.CompletedTask;
+                }));
+
+        Assert.Equal(StatusCode.Unauthenticated, exception.StatusCode);
+        Assert.False(handlerRan);
+    }
+
+    /// <summary>A duplex-streaming call fails closed to the admin scope, like client streaming.</summary>
+    /// <returns>A task that represents the asynchronous operation.</returns>
+    [Fact]
+    public async Task DuplexStreamingServerHandler_KeyWithoutAdminScope_ReturnsPermissionDenied()
+    {
+        GatewayGrpcAuthorizationInterceptor interceptor = CreateInterceptor(
+            new FakeApiKeyVerifier(SuccessWithScopes(GatewayScopes.EventsRead)),
+            new GatewayRequestIdentityAccessor());
+
+        RpcException exception = await Assert.ThrowsAsync<RpcException>(
+            () => interceptor.DuplexStreamingServerHandler<MxCommandRequest, MxCommandReply>(
+                new EmptyStreamReader<MxCommandRequest>(),
+                new RecordingServerStreamWriter<MxCommandReply>(),
+                ContextWithAuthorization("Bearer mxgw_operator01_secret"),
+                (_, _, _) => Task.CompletedTask));
+
+        Assert.Equal(StatusCode.PermissionDenied, exception.StatusCode);
+    }
+
+    /// <summary>An admin-scoped key reaches the duplex handler with its identity pushed.</summary>
+    /// <returns>A task that represents the asynchronous operation.</returns>
+    [Fact]
+    public async Task DuplexStreamingServerHandler_AdminScope_RunsHandlerWithIdentity()
+    {
+        GatewayRequestIdentityAccessor identityAccessor = new();
+        GatewayGrpcAuthorizationInterceptor interceptor = CreateInterceptor(
+            new FakeApiKeyVerifier(SuccessWithScopes(GatewayScopes.Admin)),
+            identityAccessor);
+        RecordingServerStreamWriter<MxCommandReply> writer = new();
+
+        await interceptor.DuplexStreamingServerHandler<MxCommandRequest, MxCommandReply>(
+            new EmptyStreamReader<MxCommandRequest>(),
+            writer,
+            ContextWithAuthorization("Bearer mxgw_operator01_secret"),
+            async (_, responseStream, _) =>
+            {
+                Assert.Equal("operator01", identityAccessor.Current?.KeyId);
+                await responseStream.WriteAsync(new MxCommandReply { SessionId = "duplex" });
+            });
+
+        Assert.Equal("duplex", Assert.Single(writer.Messages).SessionId);
+        Assert.Null(identityAccessor.Current);
+    }
+
     private static MxAccessGatewayService CreateService(
         ISessionManager sessionManager,
         IGatewayRequestIdentityAccessor identityAccessor)
@@ -933,4 +1078,14 @@ public sealed class GatewayGrpcAuthorizationInterceptorTests
         }
     }
 
+    /// <summary>An <see cref="IAsyncStreamReader{T}"/> that yields no messages.</summary>
+    /// <typeparam name="T">The streamed message type.</typeparam>
+    private sealed class EmptyStreamReader<T> : IAsyncStreamReader<T>
+    {
+        /// <inheritdoc />
+        public T Current => throw new InvalidOperationException("The stream is empty.");
+
+        /// <inheritdoc />
+        public Task<bool> MoveNext(CancellationToken cancellationToken) => Task.FromResult(false);
+    }
 }

src/ZB.MOM.WW.MxGateway.Tests/Security/Authorization/MetricsAuthorizationTests.cs

@@ -0,0 +1,251 @@
+using System.Security.Claims;
+using System.Text.Encodings.Web;
+using Microsoft.AspNetCore.Authentication;
+using Microsoft.AspNetCore.Authorization;
+using Microsoft.AspNetCore.Http;
+using Microsoft.Extensions.Logging.Abstractions;
+using Microsoft.Extensions.Options;
+using ZB.MOM.WW.Auth.Abstractions.ApiKeys;
+using ZB.MOM.WW.MxGateway.Server.Configuration;
+using ZB.MOM.WW.MxGateway.Server.Dashboard;
+using ZB.MOM.WW.MxGateway.Server.Metrics;
+using ZB.MOM.WW.MxGateway.Server.Security.Authorization;
+
+using LibApiKeyIdentity = ZB.MOM.WW.Auth.Abstractions.ApiKeys.ApiKeyIdentity;
+
+namespace ZB.MOM.WW.MxGateway.Tests.Security.Authorization;
+
+/// <summary>
+/// The optional <c>/metrics</c> authentication (<c>MxGateway:Metrics:RequireAuthentication</c>,
+/// SEC-14): the scheme that authenticates a scrape and the requirement that authorizes it.
+/// </summary>
+public sealed class MetricsAuthorizationTests
+{
+    private const string AdminHeader = "Bearer mxgw_scraper_secret";
+
+    /// <summary>A verified API key authenticates and carries its scopes as claims.</summary>
+    /// <returns>A task that represents the asynchronous operation.</returns>
+    [Fact]
+    public async Task Authenticate_WithValidApiKey_SucceedsWithScopeClaims()
+    {
+        CountingVerifier verifier = new(Success(GatewayScopes.Admin));
+        DefaultHttpContext httpContext = ContextWithAuthorization(AdminHeader);
+
+        AuthenticateResult result = await AuthenticateAsync(verifier, httpContext);
+
+        Assert.True(result.Succeeded);
+        Assert.Equal("scraper", result.Principal!.Identity!.Name);
+        Assert.True(result.Principal.HasClaim(MetricsAuthorizationDefaults.ScopeClaimType, GatewayScopes.Admin));
+        Assert.Equal(AdminHeader, verifier.LastHeader);
+    }
+
+    /// <summary>An invalid key fails authentication with an opaque reason.</summary>
+    /// <returns>A task that represents the asynchronous operation.</returns>
+    [Fact]
+    public async Task Authenticate_WithInvalidApiKey_Fails()
+    {
+        CountingVerifier verifier = new(new ApiKeyVerification(false, null, ApiKeyFailure.SecretMismatch));
+
+        AuthenticateResult result = await AuthenticateAsync(verifier, ContextWithAuthorization(AdminHeader));
+
+        Assert.False(result.Succeeded);
+        Assert.NotNull(result.Failure);
+        Assert.DoesNotContain("secret", result.Failure!.Message, StringComparison.OrdinalIgnoreCase);
+    }
+
+    /// <summary>
+    /// Repeated bad keys from one peer trip the shared failure limiter, after which the verifier is
+    /// no longer reached and the challenge answers 429 — /metrics is not a bypass of the gRPC throttle.
+    /// </summary>
+    /// <returns>A task that represents the asynchronous operation.</returns>
+    [Fact]
+    public async Task Authenticate_AfterRepeatedFailures_IsThrottledBeforeVerification()
+    {
+        CountingVerifier verifier = new(new ApiKeyVerification(false, null, ApiKeyFailure.SecretMismatch));
+        ApiKeyFailureLimiter limiter = new(
+            limit: 2,
+            TimeSpan.FromMinutes(1),
+            maxPartitions: 16,
+            aggregateLimit: 0,
+            probeInterval: TimeSpan.Zero,
+            TimeProvider.System);
+
+        for (int attempt = 0; attempt < 2; attempt++)
+        {
+            await AuthenticateAsync(verifier, ContextWithAuthorization(AdminHeader), limiter);
+        }
+
+        DefaultHttpContext throttled = ContextWithAuthorization(AdminHeader);
+        MetricsAuthenticationHandler handler = await CreateHandlerAsync(verifier, throttled, limiter);
+        AuthenticateResult result = await handler.AuthenticateAsync();
+        await handler.ChallengeAsync(new AuthenticationProperties());
+
+        Assert.False(result.Succeeded);
+        Assert.Equal(2, verifier.CallCount);
+        Assert.Equal(StatusCodes.Status429TooManyRequests, throttled.Response.StatusCode);
+    }
+
+    /// <summary>An anonymous scrape is challenged with a plain 401 Bearer, not a login redirect.</summary>
+    /// <returns>A task that represents the asynchronous operation.</returns>
+    [Fact]
+    public async Task Challenge_WithoutCredentials_Returns401Bearer()
+    {
+        DefaultHttpContext httpContext = new();
+        MetricsAuthenticationHandler handler = await CreateHandlerAsync(
+            new CountingVerifier(Success(GatewayScopes.Admin)),
+            httpContext);
+
+        AuthenticateResult result = await handler.AuthenticateAsync();
+        await handler.ChallengeAsync(new AuthenticationProperties());
+
+        Assert.True(result.None);
+        Assert.Equal(StatusCodes.Status401Unauthorized, httpContext.Response.StatusCode);
+        Assert.Equal("Bearer", httpContext.Response.Headers.WWWAuthenticate.ToString());
+    }
+
+    /// <summary>An API key holding the admin scope satisfies the metrics requirement.</summary>
+    /// <returns>A task that represents the asynchronous operation.</returns>
+    [Fact]
+    public async Task Requirement_ApiKeyWithAdminScope_Succeeds()
+    {
+        Assert.True(await EvaluateAsync(ApiKeyPrincipal(GatewayScopes.Admin)));
+    }
+
+    /// <summary>An API key without the admin scope does not satisfy the requirement.</summary>
+    /// <returns>A task that represents the asynchronous operation.</returns>
+    [Fact]
+    public async Task Requirement_ApiKeyWithoutAdminScope_Fails()
+    {
+        Assert.False(await EvaluateAsync(ApiKeyPrincipal(GatewayScopes.EventsRead, GatewayScopes.InvokeRead)));
+    }
+
+    /// <summary>A scope claim counts only on an identity the metrics scheme minted from a verified key.</summary>
+    /// <returns>A task that represents the asynchronous operation.</returns>
+    [Fact]
+    public async Task Requirement_ScopeClaimOnForeignIdentity_Fails()
+    {
+        ClaimsPrincipal principal = new(new ClaimsIdentity(
+            [new Claim(MetricsAuthorizationDefaults.ScopeClaimType, GatewayScopes.Admin)],
+            DashboardAuthenticationDefaults.AuthenticationScheme));
+
+        Assert.False(await EvaluateAsync(principal));
+    }
+
+    /// <summary>A dashboard login holding a real dashboard role satisfies the requirement.</summary>
+    /// <param name="role">The dashboard role.</param>
+    /// <returns>A task that represents the asynchronous operation.</returns>
+    [Theory]
+    [InlineData(DashboardRoles.Admin)]
+    [InlineData(DashboardRoles.Viewer)]
+    public async Task Requirement_DashboardLoginWithRole_Succeeds(string role)
+    {
+        ClaimsPrincipal principal = new(new ClaimsIdentity(
+            [new Claim(ClaimTypes.Name, "operator"), new Claim(ClaimTypes.Role, role)],
+            DashboardAuthenticationDefaults.AuthenticationScheme));
+
+        Assert.True(await EvaluateAsync(principal));
+    }
+
+    /// <summary>An unauthenticated principal (anonymous localhost included) never satisfies it.</summary>
+    /// <returns>A task that represents the asynchronous operation.</returns>
+    [Fact]
+    public async Task Requirement_Unauthenticated_Fails()
+    {
+        ClaimsPrincipal principal = new(new ClaimsIdentity([new Claim(ClaimTypes.Role, DashboardRoles.Admin)]));
+
+        Assert.False(await EvaluateAsync(principal));
+    }
+
+    private static async Task<bool> EvaluateAsync(ClaimsPrincipal principal)
+    {
+        MetricsAccessRequirement requirement = new();
+        AuthorizationHandlerContext context = new([requirement], principal, resource: null);
+        await new MetricsAccessHandler().HandleAsync(context);
+        return context.HasSucceeded;
+    }
+
+    private static ClaimsPrincipal ApiKeyPrincipal(params string[] scopes)
+    {
+        List<Claim> claims = [new Claim(ClaimTypes.Name, "scraper")];
+        claims.AddRange(scopes.Select(scope => new Claim(MetricsAuthorizationDefaults.ScopeClaimType, scope)));
+        return new ClaimsPrincipal(new ClaimsIdentity(claims, MetricsAuthorizationDefaults.AuthenticationScheme));
+    }
+
+    private static async Task<AuthenticateResult> AuthenticateAsync(
+        IApiKeyVerifier verifier,
+        HttpContext httpContext,
+        ApiKeyFailureLimiter? limiter = null)
+    {
+        MetricsAuthenticationHandler handler = await CreateHandlerAsync(verifier, httpContext, limiter);
+        return await handler.AuthenticateAsync();
+    }
+
+    private static async Task<MetricsAuthenticationHandler> CreateHandlerAsync(
+        IApiKeyVerifier verifier,
+        HttpContext httpContext,
+        ApiKeyFailureLimiter? limiter = null)
+    {
+        MetricsAuthenticationHandler handler = new(
+            new StaticOptionsMonitor(),
+            NullLoggerFactory.Instance,
+            UrlEncoder.Default,
+            verifier,
+            limiter ?? new ApiKeyFailureLimiter(1000, TimeSpan.FromMinutes(1), 1024, 0, TimeSpan.Zero, TimeProvider.System),
+            new AuthenticationSchemeProvider(Options.Create(new Microsoft.AspNetCore.Authentication.AuthenticationOptions())),
+            Options.Create(new GatewayOptions()),
+            new GatewayMetrics());
+        await handler.InitializeAsync(
+            new AuthenticationScheme(
+                MetricsAuthorizationDefaults.AuthenticationScheme,
+                displayName: null,
+                typeof(MetricsAuthenticationHandler)),
+            httpContext);
+        return handler;
+    }
+
+    private static DefaultHttpContext ContextWithAuthorization(string header)
+    {
+        DefaultHttpContext context = new();
+        context.Request.Headers.Authorization = header;
+        context.Connection.RemoteIpAddress = System.Net.IPAddress.Parse("198.51.100.7");
+        return context;
+    }
+
+    private static ApiKeyVerification Success(params string[] scopes) => new(
+        Succeeded: true,
+        Identity: new LibApiKeyIdentity(
+            KeyId: "scraper",
+            DisplayName: "Scraper",
+            Scopes: new HashSet<string>(scopes, StringComparer.Ordinal),
+            Constraints: null),
+        Failure: null);
+
+    private sealed class CountingVerifier(ApiKeyVerification result) : IApiKeyVerifier
+    {
+        /// <summary>Gets the number of verifications performed.</summary>
+        public int CallCount { get; private set; }
+
+        /// <summary>Gets the last authorization header presented.</summary>
+        public string? LastHeader { get; private set; }
+
+        /// <inheritdoc />
+        public Task<ApiKeyVerification> VerifyAsync(string authorizationHeader, CancellationToken ct)
+        {
+            CallCount++;
+            LastHeader = authorizationHeader;
+            return Task.FromResult(result);
+        }
+    }
+
+    private sealed class StaticOptionsMonitor : IOptionsMonitor<AuthenticationSchemeOptions>
+    {
+        /// <inheritdoc />
+        public AuthenticationSchemeOptions CurrentValue { get; } = new();
+
+        /// <inheritdoc />
+        public AuthenticationSchemeOptions Get(string? name) => CurrentValue;
+
+        /// <inheritdoc />
+        public IDisposable? OnChange(Action<AuthenticationSchemeOptions, string?> listener) => null;
+    }
+}

src/ZB.MOM.WW.MxGateway.Tests/TestSupport/AllowAllConstraintEnforcer.cs

@@ -11,26 +11,26 @@ namespace ZB.MOM.WW.MxGateway.Tests.TestSupport;
 public sealed class AllowAllConstraintEnforcer : IConstraintEnforcer
 {
     /// <inheritdoc />
-    public Task<ConstraintFailure?> CheckReadTagAsync(
+    public ValueTask<ConstraintFailure?> CheckReadTagAsync(
         ApiKeyIdentity? identity,
         string tagAddress,
-        CancellationToken cancellationToken) => Task.FromResult<ConstraintFailure?>(null);
+        CancellationToken cancellationToken) => ValueTask.FromResult<ConstraintFailure?>(null);
 
     /// <inheritdoc />
-    public Task<ConstraintFailure?> CheckReadHandleAsync(
+    public ValueTask<ConstraintFailure?> CheckReadHandleAsync(
         ApiKeyIdentity? identity,
         GatewaySession session,
         int serverHandle,
         int itemHandle,
-        CancellationToken cancellationToken) => Task.FromResult<ConstraintFailure?>(null);
+        CancellationToken cancellationToken) => ValueTask.FromResult<ConstraintFailure?>(null);
 
     /// <inheritdoc />
-    public Task<ConstraintFailure?> CheckWriteHandleAsync(
+    public ValueTask<ConstraintFailure?> CheckWriteHandleAsync(
         ApiKeyIdentity? identity,
         GatewaySession session,
         int serverHandle,
         int itemHandle,
-        CancellationToken cancellationToken) => Task.FromResult<ConstraintFailure?>(null);
+        CancellationToken cancellationToken) => ValueTask.FromResult<ConstraintFailure?>(null);
 
     /// <inheritdoc />
     public Task RecordDenialAsync(
@@ -40,4 +40,12 @@ public sealed class AllowAllConstraintEnforcer : IConstraintEnforcer
         ConstraintFailure failure,
         string? correlationId,
         CancellationToken cancellationToken) => Task.CompletedTask;
+
+    /// <inheritdoc />
+    public Task RecordBulkDenialAsync(
+        ApiKeyIdentity? identity,
+        string commandKind,
+        ConstraintDenialSummary summary,
+        string? correlationId,
+        CancellationToken cancellationToken) => Task.CompletedTask;
 }

src/ZB.MOM.WW.MxGateway.Tests/TestSupport/FakeGatewayAlarmService.cs

@@ -40,8 +40,18 @@ public sealed class FakeGatewayAlarmService : IGatewayAlarmService
 
         yield return new AlarmFeedMessage { SnapshotComplete = true };
         await Task.CompletedTask.ConfigureAwait(false);
+        if (StreamFault is not null)
+        {
+            throw StreamFault;
+        }
     }
 
+    /// <summary>
+    ///     Gets or sets an exception the feed throws after its snapshot, standing in for the monitor
+    ///     completing a subscriber with a fault (for example a slow-subscriber overflow).
+    /// </summary>
+    public Exception? StreamFault { get; set; }
+
     /// <inheritdoc />
     public Task<AcknowledgeAlarmReply> AcknowledgeAsync(
         AcknowledgeAlarmRequest request,

src/ZB.MOM.WW.MxGateway.Tests/TestSupport/PredicateConstraintEnforcer.cs

@@ -26,23 +26,26 @@ public sealed class PredicateConstraintEnforcer : IConstraintEnforcer
     /// <summary>Recorded denial messages — (commandKind, target, correlationId) tuples.</summary>
     public List<(string CommandKind, string Target, string? CorrelationId)> RecordedDenials { get; } = [];
 
+    /// <summary>Recorded aggregated bulk denials — one entry per bulk command that had any item denied.</summary>
+    public List<(string CommandKind, ConstraintDenialSummary Summary, string? CorrelationId)> RecordedBulkDenials { get; } = [];
+
     /// <inheritdoc />
-    public Task<ConstraintFailure?> CheckReadTagAsync(
+    public ValueTask<ConstraintFailure?> CheckReadTagAsync(
         ApiKeyIdentity? identity,
         string tagAddress,
         CancellationToken cancellationToken)
     {
         if (DenyTag(tagAddress))
         {
-            return Task.FromResult<ConstraintFailure?>(
+            return ValueTask.FromResult<ConstraintFailure?>(
                 new ConstraintFailure("read-tag", $"Read denied for tag '{tagAddress}'."));
         }
 
-        return Task.FromResult<ConstraintFailure?>(null);
+        return ValueTask.FromResult<ConstraintFailure?>(null);
     }
 
     /// <inheritdoc />
-    public Task<ConstraintFailure?> CheckReadHandleAsync(
+    public ValueTask<ConstraintFailure?> CheckReadHandleAsync(
         ApiKeyIdentity? identity,
         GatewaySession session,
         int serverHandle,
@@ -51,15 +54,15 @@ public sealed class PredicateConstraintEnforcer : IConstraintEnforcer
     {
         if (DenyReadHandle(serverHandle, itemHandle))
         {
-            return Task.FromResult<ConstraintFailure?>(
+            return ValueTask.FromResult<ConstraintFailure?>(
                 new ConstraintFailure("read-handle", $"Read denied for handle {itemHandle}."));
         }
 
-        return Task.FromResult<ConstraintFailure?>(null);
+        return ValueTask.FromResult<ConstraintFailure?>(null);
     }
 
     /// <inheritdoc />
-    public Task<ConstraintFailure?> CheckWriteHandleAsync(
+    public ValueTask<ConstraintFailure?> CheckWriteHandleAsync(
         ApiKeyIdentity? identity,
         GatewaySession session,
         int serverHandle,
@@ -68,11 +71,11 @@ public sealed class PredicateConstraintEnforcer : IConstraintEnforcer
     {
         if (DenyWriteHandle(serverHandle, itemHandle))
         {
-            return Task.FromResult<ConstraintFailure?>(
+            return ValueTask.FromResult<ConstraintFailure?>(
                 new ConstraintFailure("write-handle", $"Write denied for handle {itemHandle}."));
         }
 
-        return Task.FromResult<ConstraintFailure?>(null);
+        return ValueTask.FromResult<ConstraintFailure?>(null);
     }
 
     /// <inheritdoc />
@@ -87,4 +90,16 @@ public sealed class PredicateConstraintEnforcer : IConstraintEnforcer
         RecordedDenials.Add((commandKind, target, correlationId));
         return Task.CompletedTask;
     }
+
+    /// <inheritdoc />
+    public Task RecordBulkDenialAsync(
+        ApiKeyIdentity? identity,
+        string commandKind,
+        ConstraintDenialSummary summary,
+        string? correlationId,
+        CancellationToken cancellationToken)
+    {
+        RecordedBulkDenials.Add((commandKind, summary, correlationId));
+        return Task.CompletedTask;
+    }
 }

src/ZB.MOM.WW.MxGateway.Tests/TestSupport/RecordingLogger.cs

@@ -0,0 +1,48 @@
+using Microsoft.Extensions.Logging;
+
+namespace ZB.MOM.WW.MxGateway.Tests.TestSupport;
+
+/// <summary>
+///     <see cref="ILogger{TCategoryName}"/> that records every entry's level and formatted message,
+///     so a test can assert that a path logs (or does not log) at a given level.
+/// </summary>
+/// <typeparam name="T">The logger category type.</typeparam>
+public sealed class RecordingLogger<T> : ILogger<T>
+{
+    private readonly object _syncRoot = new();
+    private readonly List<(LogLevel Level, string Message, Exception? Exception)> _entries = [];
+
+    /// <summary>Gets a snapshot of the recorded entries, in order.</summary>
+    public IReadOnlyList<(LogLevel Level, string Message, Exception? Exception)> Entries
+    {
+        get
+        {
+            lock (_syncRoot)
+            {
+                return _entries.ToArray();
+            }
+        }
+    }
+
+    /// <inheritdoc />
+    public IDisposable? BeginScope<TState>(TState state)
+        where TState : notnull => null;
+
+    /// <inheritdoc />
+    public bool IsEnabled(LogLevel logLevel) => true;
+
+    /// <inheritdoc />
+    public void Log<TState>(
+        LogLevel logLevel,
+        EventId eventId,
+        TState state,
+        Exception? exception,
+        Func<TState, Exception?, string> formatter)
+    {
+        ArgumentNullException.ThrowIfNull(formatter);
+        lock (_syncRoot)
+        {
+            _entries.Add((logLevel, formatter(state, exception), exception));
+        }
+    }
+}

src/ZB.MOM.WW.MxGateway.Tests/TestSupport/RecordingServerStreamWriter.cs

@@ -29,6 +29,25 @@ public sealed class RecordingServerStreamWriter<T> : IServerStreamWriter<T>
     /// <summary>Gets or sets options for writing messages to the stream.</summary>
     public WriteOptions? WriteOptions { get; set; }
 
+    /// <summary>Gets the cancellation token passed to the most recent token-bearing write.</summary>
+    public CancellationToken LastWriteCancellationToken { get; private set; }
+
+    /// <summary>Records the supplied message unless the token is already cancelled.</summary>
+    /// <param name="message">The message to record.</param>
+    /// <param name="cancellationToken">Token that cancels the write.</param>
+    /// <returns>A completed task.</returns>
+    /// <remarks>
+    ///     The interface default throws <see cref="NotSupportedException"/> for a cancelable token,
+    ///     and the gateway passes the call's token on every streamed write, so a test writer must
+    ///     override this overload the way the ASP.NET Core writer does.
+    /// </remarks>
+    public Task WriteAsync(T message, CancellationToken cancellationToken)
+    {
+        cancellationToken.ThrowIfCancellationRequested();
+        LastWriteCancellationToken = cancellationToken;
+        return WriteAsync(message);
+    }
+
     /// <summary>Records the supplied message.</summary>
     /// <param name="message">The message to record.</param>
     /// <returns>A completed task.</returns>