Merge branch 'fix/gwc-26-27-alarm-attach'

# Conflicts:
#	archreview/2026-07-12/remediation/00-tracking.md
#	archreview/2026-07-12/remediation/10-gateway-core.md

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

@@ -1,5 +1,4 @@
 using System.Diagnostics.CodeAnalysis;
-using System.Runtime.CompilerServices;
 using System.Threading.Channels;
 using Google.Protobuf.WellKnownTypes;
 using Microsoft.Extensions.Logging.Abstractions;
@@ -8,6 +7,7 @@ using ZB.MOM.WW.MxGateway.Server.Alarms;
 using ZB.MOM.WW.MxGateway.Server.Configuration;
 using ZB.MOM.WW.MxGateway.Server.Metrics;
 using ZB.MOM.WW.MxGateway.Server.Sessions;
+using ZB.MOM.WW.MxGateway.Tests.TestSupport;
 
 namespace ZB.MOM.WW.MxGateway.Tests.Alarms;
 
@@ -420,6 +420,9 @@ public sealed class AlarmFailoverEndToEndTests
             string? ownerKeyId,
             CancellationToken cancellationToken)
         {
+            // The monitor attaches its internal subscriber directly on this session, so the
+            // session has to be a genuinely Ready one with a worker client feeding the
+            // distributor pump — EmitEvent writes into that worker's event stream.
             GatewaySession session = new(
                 Guid.NewGuid().ToString("N"),
                 "Galaxy",
@@ -432,6 +435,8 @@ public sealed class AlarmFailoverEndToEndTests
                 TimeSpan.FromSeconds(30),
                 TimeSpan.FromSeconds(30),
                 DateTimeOffset.UtcNow);
+            session.AttachWorkerClient(new ChannelWorkerClient(session.SessionId, _events.Reader));
+            session.MarkReady();
             return Task.FromResult(session);
         }
 
@@ -460,29 +465,9 @@ public sealed class AlarmFailoverEndToEndTests
         }
 
         /// <inheritdoc />
-        public async IAsyncEnumerable<WorkerEvent> ReadEventsAsync(
+        public IAsyncEnumerable<WorkerEvent> ReadEventsAsync(
             string sessionId,
-            [EnumeratorCancellation] CancellationToken cancellationToken)
-        {
-            await foreach (WorkerEvent workerEvent in _events.Reader.ReadAllAsync(cancellationToken))
-            {
-                yield return workerEvent;
-            }
-        }
-
-        /// <inheritdoc />
-        public async IAsyncEnumerable<MxEvent> ReadAlarmEventsAsync(
-            string sessionId,
-            [EnumeratorCancellation] CancellationToken cancellationToken)
-        {
-            await foreach (WorkerEvent workerEvent in _events.Reader.ReadAllAsync(cancellationToken))
-            {
-                if (workerEvent.Event is not null)
-                {
-                    yield return workerEvent.Event;
-                }
-            }
-        }
+            CancellationToken cancellationToken) => throw new NotSupportedException();
 
         /// <inheritdoc />
         public bool TryGetSession(string sessionId, [MaybeNullWhen(false)] out GatewaySession session)

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

@@ -0,0 +1,468 @@
+using System.Diagnostics.CodeAnalysis;
+using System.Threading.Channels;
+using Google.Protobuf.WellKnownTypes;
+using Microsoft.Extensions.Logging.Abstractions;
+using ZB.MOM.WW.MxGateway.Contracts.Proto;
+using ZB.MOM.WW.MxGateway.Server.Alarms;
+using ZB.MOM.WW.MxGateway.Server.Configuration;
+using ZB.MOM.WW.MxGateway.Server.Grpc;
+using ZB.MOM.WW.MxGateway.Server.Metrics;
+using ZB.MOM.WW.MxGateway.Server.Sessions;
+using ZB.MOM.WW.MxGateway.Tests.TestSupport;
+
+namespace ZB.MOM.WW.MxGateway.Tests.Alarms;
+
+/// <summary>
+///     GWC-26 regression tests for the alarm monitor's startup ordering. Unlike the
+///     sibling alarm-monitor tests, the session manager here hands the monitor a REAL
+///     <see cref="GatewaySession"/> that is driven to Ready with a dashboard mirror, so the
+///     distributor pump is already running when the monitor attaches — the production
+///     condition under which a late internal subscriber silently misses everything the pump
+///     already fanned.
+/// </summary>
+public sealed class GatewayAlarmMonitorAttachOrderTests
+{
+    private const string AlarmReference = "Galaxy!Area.Tank01.Hi";
+    private static readonly TimeSpan WaitTimeout = TimeSpan.FromSeconds(30);
+
+    /// <summary>
+    ///     The monitor must take its internal distributor lease BEFORE issuing
+    ///     <c>SubscribeAlarms</c>, so transitions the worker emits during the
+    ///     subscribe + first-reconcile window buffer in the lease instead of being fanned to a
+    ///     subscriber set the monitor has not joined yet. The test parks the monitor inside its
+    ///     <c>SubscribeAlarms</c> round trip, emits a Raise and an Acknowledge, waits until the
+    ///     dashboard mirror proves the pump has already fanned both, and only then releases the
+    ///     monitor: with a late attach both transitions are lost to the alarm feed forever.
+    /// </summary>
+    /// <returns>A task that represents the asynchronous operation.</returns>
+    [Fact]
+    public async Task TransitionsDuringSubscribeWindow_StillReachTheAlarmFeed()
+    {
+        using GatewayMetrics metrics = new();
+        await using FakeSessionManager sessions = new();
+        sessions.HoldSubscribeUntilReleased();
+        using GatewayAlarmMonitor monitor = CreateMonitor(sessions, metrics);
+
+        using CancellationTokenSource cts = new();
+        await monitor.StartAsync(cts.Token);
+        await sessions.WaitForSubscribeStartAsync(WaitTimeout);
+
+        // A live feed subscriber, drained past its baseline ProviderStatus so it is registered
+        // before any window transition is broadcast.
+        List<AlarmFeedMessage> received = [];
+        TaskCompletionSource baselineReceived = new(TaskCreationOptions.RunContinuationsAsynchronously);
+        using CancellationTokenSource streamCts = new();
+        Task reader = ReadFeedAsync(monitor, received, baselineReceived, streamCts.Token);
+        await baselineReceived.Task.WaitAsync(WaitTimeout);
+
+        // The window: the worker reports a raise and an acknowledge while the monitor is still
+        // waiting for its SubscribeAlarms reply.
+        sessions.EmitEvent(Transition(1, AlarmTransitionKind.Raise));
+        sessions.EmitEvent(Transition(2, AlarmTransitionKind.Acknowledge));
+
+        // The dashboard mirror is an independent distributor subscriber: once it has both
+        // events, the pump has provably already fanned them, so a subscriber that registers
+        // after this point can never receive them.
+        await WaitUntilAsync(() => sessions.Broadcaster.Captures.Count == 2, WaitTimeout);
+
+        sessions.ReleaseSubscribe();
+
+        AlarmFeedMessage raise = await WaitForAsync(
+            received,
+            m => m.PayloadCase == AlarmFeedMessage.PayloadOneofCase.Transition
+                && m.Transition.TransitionKind == AlarmTransitionKind.Raise,
+            WaitTimeout);
+        AlarmFeedMessage acknowledge = await WaitForAsync(
+            received,
+            m => m.PayloadCase == AlarmFeedMessage.PayloadOneofCase.Transition
+                && m.Transition.TransitionKind == AlarmTransitionKind.Acknowledge,
+            WaitTimeout);
+
+        Assert.Equal(AlarmReference, raise.Transition.AlarmFullReference);
+        Assert.Equal(AlarmReference, acknowledge.Transition.AlarmFullReference);
+
+        await streamCts.CancelAsync();
+        await reader;
+        await cts.CancelAsync();
+        await monitor.StopAsync(CancellationToken.None);
+    }
+
+    /// <summary>
+    ///     Defense in depth for any window the attach reorder cannot cover (worker restart,
+    ///     internal-subscriber overflow disconnect): when a reconcile snapshot reports an alarm
+    ///     the cache already holds but with the state advanced to
+    ///     <see cref="AlarmConditionState.ActiveAcked"/>, the monitor broadcasts an
+    ///     <see cref="AlarmTransitionKind.Acknowledge"/> feed transition. Before the fix the
+    ///     acked state was absorbed silently by the snapshot replace, leaving live subscribers
+    ///     showing the alarm unacked until it cleared.
+    /// </summary>
+    /// <returns>A task that represents the asynchronous operation.</returns>
+    [Fact]
+    public async Task ApplyReconcileBroadcastsAcknowledgeDelta()
+    {
+        using GatewayMetrics metrics = new();
+        await using FakeSessionManager sessions = new();
+        using GatewayAlarmMonitor monitor = CreateMonitor(sessions, metrics);
+
+        using CancellationTokenSource cts = new();
+        await monitor.StartAsync(cts.Token);
+        await sessions.WaitForSubscribeStartAsync(WaitTimeout);
+
+        // Seed the cache with an active (unacked) alarm through a reconcile rather than a live
+        // transition: a buffered live transition could still be in flight when the cache first
+        // shows the alarm, and would then broadcast to the reader below and pollute the
+        // exactly-one-transition assertion. A provider-mode event forces the reconcile
+        // immediately, so the test never waits on the periodic timer.
+        sessions.SetReconcileSnapshot(Snapshot(AlarmConditionState.Active));
+        sessions.EmitEvent(ProviderModeProbe(1));
+        await WaitUntilAsync(
+            () => monitor.CurrentAlarms.Any(alarm => alarm.AlarmFullReference == AlarmReference
+                && alarm.CurrentState == AlarmConditionState.Active),
+            WaitTimeout);
+
+        // Subscribe AFTER the seed: this reader's snapshot carries the alarm, so every
+        // transition it observes from here on is a reconcile-derived broadcast.
+        List<AlarmFeedMessage> received = [];
+        TaskCompletionSource snapshotComplete = new(TaskCreationOptions.RunContinuationsAsynchronously);
+        using CancellationTokenSource streamCts = new();
+        Task reader = ReadFeedAsync(monitor, received, snapshotComplete, streamCts.Token, untilSnapshotComplete: true);
+        await snapshotComplete.Task.WaitAsync(WaitTimeout);
+
+        // The worker now reports the same alarm acknowledged. A provider-mode event forces an
+        // immediate reconcile pass so the test does not wait on the periodic timer.
+        sessions.SetReconcileSnapshot(Snapshot(AlarmConditionState.ActiveAcked));
+        sessions.EmitEvent(ProviderModeProbe(2));
+
+        AlarmFeedMessage acknowledge = await WaitForAsync(
+            received,
+            m => m.PayloadCase == AlarmFeedMessage.PayloadOneofCase.Transition
+                && m.Transition.TransitionKind == AlarmTransitionKind.Acknowledge,
+            WaitTimeout);
+        Assert.Equal(AlarmReference, acknowledge.Transition.AlarmFullReference);
+
+        lock (received)
+        {
+            AlarmFeedMessage[] transitions = received
+                .Where(m => m.PayloadCase == AlarmFeedMessage.PayloadOneofCase.Transition)
+                .ToArray();
+            AlarmFeedMessage single = Assert.Single(transitions);
+            Assert.Equal(AlarmTransitionKind.Acknowledge, single.Transition.TransitionKind);
+        }
+
+        await streamCts.CancelAsync();
+        await reader;
+        await cts.CancelAsync();
+        await monitor.StopAsync(CancellationToken.None);
+    }
+
+    private static GatewayAlarmMonitor CreateMonitor(FakeSessionManager sessions, GatewayMetrics metrics)
+    {
+        AlarmsOptions options = new()
+        {
+            Enabled = true,
+            SubscriptionExpression = @"\\NODE\Galaxy!Area",
+        };
+        return new GatewayAlarmMonitor(
+            sessions,
+            new StubWatchListResolver(),
+            metrics,
+            Microsoft.Extensions.Options.Options.Create(new GatewayOptions { Alarms = options }),
+            NullLogger<GatewayAlarmMonitor>.Instance);
+    }
+
+    // Drains the monitor's feed into received, signalling gate on the first message (the
+    // baseline ProviderStatus) or, when untilSnapshotComplete is set, on SnapshotComplete.
+    private static Task ReadFeedAsync(
+        GatewayAlarmMonitor monitor,
+        List<AlarmFeedMessage> received,
+        TaskCompletionSource gate,
+        CancellationToken cancellationToken,
+        bool untilSnapshotComplete = false)
+    {
+        return Task.Run(
+            async () =>
+            {
+                try
+                {
+                    await foreach (AlarmFeedMessage message in monitor.StreamAsync(null, cancellationToken))
+                    {
+                        bool opensGate = !untilSnapshotComplete
+                            || message.PayloadCase == AlarmFeedMessage.PayloadOneofCase.SnapshotComplete;
+
+                        // Record only what arrives AFTER the gate opened: everything up to and
+                        // including the gate message is this subscriber's snapshot preamble, not
+                        // a live broadcast, so assertions stay about broadcasts alone.
+                        lock (received)
+                        {
+                            if (gate.Task.IsCompleted)
+                            {
+                                received.Add(message);
+                            }
+                        }
+
+                        if (opensGate)
+                        {
+                            gate.TrySetResult();
+                        }
+                    }
+                }
+                catch (OperationCanceledException)
+                {
+                    // Expected when the test cancels the stream.
+                }
+            },
+            CancellationToken.None);
+    }
+
+    private static MxEvent Transition(ulong sequence, AlarmTransitionKind kind) => new()
+    {
+        Family = MxEventFamily.OnAlarmTransition,
+        WorkerSequence = sequence,
+        OnAlarmTransition = new OnAlarmTransitionEvent
+        {
+            AlarmFullReference = AlarmReference,
+            SourceObjectReference = "Tank01",
+            AlarmTypeName = "AnalogLimitAlarm.Hi",
+            TransitionKind = kind,
+            Severity = 500,
+            SourceProvider = AlarmProviderMode.Alarmmgr,
+            TransitionTimestamp = Timestamp.FromDateTimeOffset(DateTimeOffset.UtcNow),
+        },
+    };
+
+    // A no-op provider-mode event. The monitor forces an immediate reconcile after every one,
+    // which is how these tests drive a reconcile pass without waiting on the periodic timer.
+    private static MxEvent ProviderModeProbe(ulong sequence) => new()
+    {
+        Family = MxEventFamily.OnAlarmProviderModeChanged,
+        WorkerSequence = sequence,
+        OnAlarmProviderModeChanged = new OnAlarmProviderModeChangedEvent
+        {
+            Mode = AlarmProviderMode.Alarmmgr,
+            Reason = "probe",
+            At = Timestamp.FromDateTimeOffset(DateTimeOffset.UtcNow),
+        },
+    };
+
+    private static ActiveAlarmSnapshot Snapshot(AlarmConditionState state) => new()
+    {
+        AlarmFullReference = AlarmReference,
+        SourceObjectReference = "Tank01",
+        AlarmTypeName = "AnalogLimitAlarm.Hi",
+        CurrentState = state,
+        Severity = 500,
+        SourceProvider = AlarmProviderMode.Alarmmgr,
+    };
+
+    private static async Task<AlarmFeedMessage> WaitForAsync(
+        List<AlarmFeedMessage> received,
+        Func<AlarmFeedMessage, bool> predicate,
+        TimeSpan timeout)
+    {
+        DateTime deadline = DateTime.UtcNow + timeout;
+        while (DateTime.UtcNow < deadline)
+        {
+            lock (received)
+            {
+                AlarmFeedMessage? match = received.FirstOrDefault(predicate);
+                if (match is not null)
+                {
+                    return match;
+                }
+            }
+
+            await Task.Delay(25);
+        }
+
+        throw new TimeoutException("No matching AlarmFeedMessage was received in time.");
+    }
+
+    private static async Task WaitUntilAsync(Func<bool> condition, TimeSpan timeout)
+    {
+        DateTime deadline = DateTime.UtcNow + timeout;
+        while (DateTime.UtcNow < deadline)
+        {
+            if (condition())
+            {
+                return;
+            }
+
+            await Task.Delay(25);
+        }
+
+        throw new TimeoutException("Condition was not met in time.");
+    }
+
+    /// <summary><see cref="IAlarmWatchListResolver"/> that resolves an empty watch-list.</summary>
+    private sealed class StubWatchListResolver : IAlarmWatchListResolver
+    {
+        /// <inheritdoc />
+        public Task<IReadOnlyList<AlarmSubtagTarget>> ResolveAsync(
+            AlarmsOptions options,
+            CancellationToken cancellationToken = default) =>
+            Task.FromResult<IReadOnlyList<AlarmSubtagTarget>>([]);
+    }
+
+    /// <summary>
+    ///     Session manager that hands the monitor a real <see cref="GatewaySession"/> driven to
+    ///     Ready with a dashboard mirror, so the distributor pump is running before the monitor
+    ///     attaches. <see cref="EmitEvent"/> pushes worker events through the fake worker client
+    ///     into that pump, exactly as a live worker would.
+    /// </summary>
+    private sealed class FakeSessionManager : ISessionManager, IAsyncDisposable
+    {
+        private readonly Channel<WorkerEvent> _events = Channel.CreateUnbounded<WorkerEvent>();
+        private readonly TaskCompletionSource _subscribeStarted =
+            new(TaskCreationOptions.RunContinuationsAsynchronously);
+        private readonly object _sync = new();
+        private TaskCompletionSource _subscribeGate = CreateReleasedGate();
+        private ActiveAlarmSnapshot[] _reconcileSnapshot = [];
+        private GatewaySession? _session;
+
+        /// <summary>Dashboard mirror attached to the session; proves what the pump has fanned.</summary>
+        public RecordingDashboardEventBroadcaster Broadcaster { get; } = new();
+
+        /// <summary>Re-arms the gate so <c>SubscribeAlarms</c> parks until <see cref="ReleaseSubscribe"/>.</summary>
+        public void HoldSubscribeUntilReleased() =>
+            _subscribeGate = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously);
+
+        /// <summary>Releases a gate armed by <see cref="HoldSubscribeUntilReleased"/>.</summary>
+        public void ReleaseSubscribe() => _subscribeGate.TrySetResult();
+
+        /// <summary>Completes once the monitor's <c>SubscribeAlarms</c> command has arrived.</summary>
+        /// <param name="timeout">The maximum time to wait.</param>
+        /// <returns>A task that completes when the command arrives.</returns>
+        public Task WaitForSubscribeStartAsync(TimeSpan timeout) => _subscribeStarted.Task.WaitAsync(timeout);
+
+        /// <summary>Sets the active-alarm snapshot every <c>QueryActiveAlarms</c> reconcile returns.</summary>
+        /// <param name="snapshots">The snapshots to report.</param>
+        public void SetReconcileSnapshot(params ActiveAlarmSnapshot[] snapshots)
+        {
+            lock (_sync)
+            {
+                _reconcileSnapshot = snapshots;
+            }
+        }
+
+        /// <summary>Pushes a worker event into the session's distributor pump.</summary>
+        /// <param name="mxEvent">The event to push.</param>
+        public void EmitEvent(MxEvent mxEvent) =>
+            _events.Writer.TryWrite(new WorkerEvent { Event = mxEvent });
+
+        /// <inheritdoc />
+        public Task<GatewaySession> OpenSessionAsync(
+            SessionOpenRequest request,
+            string? clientIdentity,
+            string? ownerKeyId,
+            CancellationToken cancellationToken)
+        {
+            GatewaySession session = new(
+                sessionId: "session-alarm-attach-order",
+                backendName: "Galaxy",
+                pipeName: "mxaccess-gateway-1-session-alarm-attach-order",
+                nonce: "nonce",
+                clientIdentity: clientIdentity,
+                ownerKeyId: ownerKeyId,
+                clientSessionName: request.ClientSessionName,
+                clientCorrelationId: request.ClientCorrelationId,
+                commandTimeout: TimeSpan.FromSeconds(30),
+                startupTimeout: TimeSpan.FromSeconds(30),
+                shutdownTimeout: TimeSpan.FromSeconds(30),
+                leaseDuration: TimeSpan.FromMinutes(30),
+                openedAt: DateTimeOffset.UtcNow,
+                eventStreaming: new SessionEventStreaming(
+                    new MxAccessGrpcMapper(),
+                    new EventOptions { QueueCapacity = 64 },
+                    NullLogger<SessionEventDistributor>.Instance,
+                    TimeProvider.System,
+                    new GatewayMetrics(),
+                    Broadcaster));
+            session.AttachWorkerClient(new ChannelWorkerClient(session.SessionId, _events.Reader));
+
+            // MarkReady starts the dashboard mirror, and with it the distributor pump — the
+            // production precondition this regression depends on.
+            session.MarkReady();
+            _session = session;
+            return Task.FromResult(session);
+        }
+
+        /// <inheritdoc />
+        public async Task<WorkerCommandReply> InvokeAsync(
+            string sessionId,
+            WorkerCommand command,
+            CancellationToken cancellationToken)
+        {
+            MxCommandReply reply = new()
+            {
+                ProtocolStatus = new ProtocolStatus { Code = ProtocolStatusCode.Ok },
+            };
+
+            switch (command.Command?.Kind)
+            {
+                case MxCommandKind.SubscribeAlarms:
+                    _subscribeStarted.TrySetResult();
+                    await _subscribeGate.Task.WaitAsync(cancellationToken).ConfigureAwait(false);
+                    break;
+                case MxCommandKind.QueryActiveAlarms:
+                    QueryActiveAlarmsReplyPayload payload = new();
+                    lock (_sync)
+                    {
+                        payload.Snapshots.AddRange(_reconcileSnapshot.Select(snapshot => snapshot.Clone()));
+                    }
+
+                    reply.QueryActiveAlarms = payload;
+                    break;
+            }
+
+            return new WorkerCommandReply { Reply = reply };
+        }
+
+        /// <inheritdoc />
+        public IAsyncEnumerable<WorkerEvent> ReadEventsAsync(
+            string sessionId,
+            CancellationToken cancellationToken) => throw new NotSupportedException();
+
+        /// <inheritdoc />
+        public bool TryGetSession(string sessionId, [MaybeNullWhen(false)] out GatewaySession session)
+        {
+            session = _session;
+            return session is not null;
+        }
+
+        /// <inheritdoc />
+        public Task<SessionCloseResult> CloseSessionAsync(string sessionId, CancellationToken cancellationToken)
+        {
+            _events.Writer.TryComplete();
+            return Task.FromResult(new SessionCloseResult(sessionId, SessionState.Closed, AlreadyClosed: false));
+        }
+
+        /// <inheritdoc />
+        public Task<SessionCloseResult> KillWorkerAsync(string sessionId, string reason, CancellationToken cancellationToken) =>
+            Task.FromResult(new SessionCloseResult(sessionId, SessionState.Closed, AlreadyClosed: false));
+
+        /// <inheritdoc />
+        public Task<int> CloseExpiredLeasesAsync(DateTimeOffset now, CancellationToken cancellationToken) =>
+            Task.FromResult(0);
+
+        /// <inheritdoc />
+        public Task ShutdownAsync(CancellationToken cancellationToken) => Task.CompletedTask;
+
+        /// <summary>Disposes the session the fake handed out.</summary>
+        /// <returns>A task that represents the asynchronous operation.</returns>
+        public async ValueTask DisposeAsync()
+        {
+            _events.Writer.TryComplete();
+            if (_session is not null)
+            {
+                await _session.DisposeAsync().ConfigureAwait(false);
+            }
+        }
+
+        private static TaskCompletionSource CreateReleasedGate()
+        {
+            TaskCompletionSource gate = new(TaskCreationOptions.RunContinuationsAsynchronously);
+            gate.SetResult();
+            return gate;
+        }
+    }
+}

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

@@ -1,6 +1,5 @@
 using System.Diagnostics.CodeAnalysis;
 using System.Diagnostics.Metrics;
-using System.Runtime.CompilerServices;
 using System.Threading.Channels;
 using Google.Protobuf.WellKnownTypes;
 using Microsoft.Extensions.Logging.Abstractions;
@@ -10,6 +9,7 @@ using ZB.MOM.WW.MxGateway.Server.Alarms;
 using ZB.MOM.WW.MxGateway.Server.Configuration;
 using ZB.MOM.WW.MxGateway.Server.Metrics;
 using ZB.MOM.WW.MxGateway.Server.Sessions;
+using ZB.MOM.WW.MxGateway.Tests.TestSupport;
 
 namespace ZB.MOM.WW.MxGateway.Tests.Alarms;
 
@@ -733,6 +733,9 @@ public sealed class GatewayAlarmMonitorProviderModeTests
             string? ownerKeyId,
             CancellationToken cancellationToken)
         {
+            // The monitor attaches its internal subscriber directly on this session, so the
+            // session has to be a genuinely Ready one with a worker client feeding the
+            // distributor pump — EmitEvent writes into that worker's event stream.
             GatewaySession session = new(
                 Guid.NewGuid().ToString("N"),
                 "Galaxy",
@@ -745,6 +748,8 @@ public sealed class GatewayAlarmMonitorProviderModeTests
                 TimeSpan.FromSeconds(30),
                 TimeSpan.FromSeconds(30),
                 DateTimeOffset.UtcNow);
+            session.AttachWorkerClient(new ChannelWorkerClient(session.SessionId, _events.Reader));
+            session.MarkReady();
             return Task.FromResult(session);
         }
 
@@ -773,29 +778,9 @@ public sealed class GatewayAlarmMonitorProviderModeTests
         }
 
         /// <inheritdoc />
-        public async IAsyncEnumerable<WorkerEvent> ReadEventsAsync(
+        public IAsyncEnumerable<WorkerEvent> ReadEventsAsync(
             string sessionId,
-            [EnumeratorCancellation] CancellationToken cancellationToken)
-        {
-            await foreach (WorkerEvent workerEvent in _events.Reader.ReadAllAsync(cancellationToken))
-            {
-                yield return workerEvent;
-            }
-        }
-
-        /// <inheritdoc />
-        public async IAsyncEnumerable<MxEvent> ReadAlarmEventsAsync(
-            string sessionId,
-            [EnumeratorCancellation] CancellationToken cancellationToken)
-        {
-            await foreach (WorkerEvent workerEvent in _events.Reader.ReadAllAsync(cancellationToken))
-            {
-                if (workerEvent.Event is not null)
-                {
-                    yield return workerEvent.Event;
-                }
-            }
-        }
+            CancellationToken cancellationToken) => throw new NotSupportedException();
 
         /// <inheritdoc />
         public bool TryGetSession(string sessionId, [MaybeNullWhen(false)] out GatewaySession session)

src/ZB.MOM.WW.MxGateway.Tests/Gateway/Dashboard/DashboardSessionAdminServiceTests.cs

@@ -378,14 +378,6 @@ public sealed class DashboardSessionAdminServiceTests
             throw new NotSupportedException();
         }
 
-        /// <inheritdoc />
-        public IAsyncEnumerable<MxEvent> ReadAlarmEventsAsync(
-            string sessionId,
-            CancellationToken cancellationToken)
-        {
-            throw new NotSupportedException();
-        }
-
         /// <inheritdoc />
         public Task<SessionCloseResult> CloseSessionAsync(
             string sessionId,

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

@@ -755,14 +755,6 @@ public sealed class EventStreamServiceTests
             return _sessions[sessionId].ReadEventsAsync(cancellationToken);
         }
 
-        /// <inheritdoc />
-        public IAsyncEnumerable<MxEvent> ReadAlarmEventsAsync(
-            string sessionId,
-            CancellationToken cancellationToken)
-        {
-            throw new NotSupportedException();
-        }
-
         /// <inheritdoc />
         public Task<SessionCloseResult> CloseSessionAsync(
             string sessionId,

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

@@ -948,14 +948,6 @@ public sealed class MxAccessGatewayServiceConstraintTests
             }
         }
 
-        /// <inheritdoc />
-        public IAsyncEnumerable<MxEvent> ReadAlarmEventsAsync(
-            string sessionId,
-            CancellationToken cancellationToken)
-        {
-            throw new NotSupportedException();
-        }
-
         /// <inheritdoc />
         public Task<SessionCloseResult> CloseSessionAsync(
             string sessionId,

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

@@ -616,14 +616,6 @@ public sealed class MxAccessGatewayServiceTests
             }
         }
 
-        /// <inheritdoc />
-        public IAsyncEnumerable<MxEvent> ReadAlarmEventsAsync(
-            string sessionId,
-            CancellationToken cancellationToken)
-        {
-            throw new NotSupportedException();
-        }
-
         /// <inheritdoc />
         public Task<SessionCloseResult> CloseSessionAsync(
             string sessionId,

src/ZB.MOM.WW.MxGateway.Tests/Gateway/Sessions/GatewaySessionDashboardMirrorTests.cs

@@ -352,11 +352,6 @@ public sealed class GatewaySessionDashboardMirrorTests
             string sessionId,
             CancellationToken cancellationToken) => session.ReadEventsAsync(cancellationToken);
 
-        /// <inheritdoc />
-        public IAsyncEnumerable<MxEvent> ReadAlarmEventsAsync(
-            string sessionId,
-            CancellationToken cancellationToken) => throw new NotSupportedException();
-
         /// <inheritdoc />
         public Task<SessionCloseResult> CloseSessionAsync(
             string sessionId,

src/ZB.MOM.WW.MxGateway.Tests/Gateway/Sessions/GatewaySessionTests.cs

@@ -668,6 +668,81 @@ public sealed class GatewaySessionTests
         Assert.Equal(SessionState.Ready, session.State);
     }
 
+    /// <summary>
+    ///     GWC-27: <see cref="GatewaySession.AttachInternalEventSubscriber"/> must refuse to
+    ///     attach before the session is Ready. Without the gate the attach would construct and
+    ///     start the distributor against a not-yet-Ready worker; the pump source throws
+    ///     <c>SessionNotReady</c>, every subscriber is completed with that error, and the
+    ///     distributor latches — leaving a session that reaches Ready with permanently dead
+    ///     event streaming. The second half of the test is the load-bearing one: after the
+    ///     failed attach the session still streams live events, proving the distributor was
+    ///     never created or started.
+    /// </summary>
+    /// <returns>A task that represents the asynchronous operation.</returns>
+    [Fact]
+    public async Task AttachInternalEventSubscriberBeforeReadyThrowsAndDoesNotPoisonDistributor()
+    {
+        FakeWorkerClient workerClient = new();
+        workerClient.Events.Add(new WorkerEvent
+        {
+            Event = new MxEvent { Family = MxEventFamily.OnDataChange, WorkerSequence = 1, OnDataChange = new OnDataChangeEvent() },
+        });
+        workerClient.Events.Add(new WorkerEvent
+        {
+            Event = new MxEvent { Family = MxEventFamily.OnDataChange, WorkerSequence = 2, OnDataChange = new OnDataChangeEvent() },
+        });
+
+        // Constructed but neither worker-attached nor Ready — the premature-attach case.
+        await using GatewaySession session = CreateSession();
+
+        SessionManagerException exception = Assert.Throws<SessionManagerException>(
+            () => session.AttachInternalEventSubscriber());
+        Assert.Equal(SessionManagerErrorCode.SessionNotReady, exception.ErrorCode);
+
+        // Drive the session to Ready and stream: the failed attach must not have poisoned
+        // (or even created) the distributor, so a normal subscriber still receives events.
+        session.AttachWorkerClient(workerClient);
+        session.MarkReady();
+
+        using IEventSubscriberLease lease = session.AttachEventSubscriber(maxSubscribers: 1);
+        List<MxEvent> received = [];
+        using CancellationTokenSource readCts = new(TimeSpan.FromSeconds(5));
+        await foreach (MxEvent mxEvent in lease.Reader.ReadAllAsync(readCts.Token))
+        {
+            received.Add(mxEvent);
+            if (received.Count == 2)
+            {
+                break;
+            }
+        }
+
+        Assert.Equal([1UL, 2UL], received.Select(mxEvent => mxEvent.WorkerSequence).ToArray());
+    }
+
+    private static GatewaySession CreateSession()
+    {
+        return new GatewaySession(
+            sessionId: "session-test-internal-attach",
+            backendName: "mxaccess",
+            pipeName: "mxaccess-gateway-1-session-test-internal-attach",
+            nonce: "nonce",
+            clientIdentity: "client-1",
+            ownerKeyId: null,
+            clientSessionName: "test-session",
+            clientCorrelationId: "client-correlation-1",
+            commandTimeout: TimeSpan.FromSeconds(5),
+            startupTimeout: TimeSpan.FromSeconds(5),
+            shutdownTimeout: TimeSpan.FromSeconds(5),
+            leaseDuration: TimeSpan.FromMinutes(30),
+            openedAt: DateTimeOffset.UtcNow,
+            eventStreaming: new SessionEventStreaming(
+                new MxAccessGrpcMapper(),
+                new EventOptions { QueueCapacity = 8 },
+                NullLogger<SessionEventDistributor>.Instance,
+                TimeProvider.System,
+                new GatewayMetrics()));
+    }
+
     private static GatewaySession CreateReadySessionWithDetachGrace(
         IWorkerClient workerClient,
         TimeProvider timeProvider,
@@ -855,6 +930,9 @@ public sealed class GatewaySessionTests
         /// <summary>Gets the count of dispose invocations.</summary>
         public int DisposeCount { get; private set; }
 
+        /// <summary>Events <see cref="ReadEventsAsync"/> yields, in order, before completing. Empty by default.</summary>
+        public List<WorkerEvent> Events { get; } = [];
+
         /// <inheritdoc />
         public Task StartAsync(CancellationToken cancellationToken) => Task.CompletedTask;
 
@@ -869,7 +947,11 @@ public sealed class GatewaySessionTests
             [EnumeratorCancellation] CancellationToken cancellationToken)
         {
             await Task.CompletedTask.ConfigureAwait(false);
-            yield break;
+            foreach (WorkerEvent workerEvent in Events)
+            {
+                cancellationToken.ThrowIfCancellationRequested();
+                yield return workerEvent;
+            }
         }
 
         /// <inheritdoc />

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

@@ -579,14 +579,6 @@ public sealed class GatewayGrpcAuthorizationInterceptorTests
             return AsyncEnumerable.Empty<WorkerEvent>();
         }
 
-        /// <inheritdoc />
-        public IAsyncEnumerable<MxEvent> ReadAlarmEventsAsync(
-            string sessionId,
-            CancellationToken cancellationToken)
-        {
-            return AsyncEnumerable.Empty<MxEvent>();
-        }
-
         /// <inheritdoc />
         public Task<SessionCloseResult> CloseSessionAsync(
             string sessionId,

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

@@ -0,0 +1,66 @@
+using System.Runtime.CompilerServices;
+using System.Threading.Channels;
+using ZB.MOM.WW.MxGateway.Contracts.Proto;
+using ZB.MOM.WW.MxGateway.Server.Workers;
+
+namespace ZB.MOM.WW.MxGateway.Tests.TestSupport;
+
+/// <summary>
+///     Always-<see cref="WorkerClientState.Ready"/> <see cref="IWorkerClient"/> whose event
+///     stream is a channel the test writes to. Lets a test attach a real
+///     <c>GatewaySession</c> to a worker it drives by hand — the session can be marked Ready,
+///     its <c>SessionEventDistributor</c> pump then drains this channel, and the test controls
+///     exactly when each event is fanned.
+/// </summary>
+/// <remarks>
+///     Commands are not scripted here: <see cref="InvokeAsync"/> returns an empty reply, because
+///     the consumers of this fake route commands through their own <c>ISessionManager</c> double
+///     rather than through the worker client. Use a purpose-built worker client instead when a
+///     test needs command behavior.
+/// </remarks>
+/// <param name="sessionId">Session identifier the client reports.</param>
+/// <param name="events">Channel whose events <see cref="ReadEventsAsync"/> yields, in order.</param>
+public sealed class ChannelWorkerClient(string sessionId, ChannelReader<WorkerEvent> events) : IWorkerClient
+{
+    /// <inheritdoc />
+    public string SessionId { get; } = sessionId;
+
+    /// <inheritdoc />
+    public int? ProcessId { get; } = 4321;
+
+    /// <inheritdoc />
+    public WorkerClientState State { get; } = WorkerClientState.Ready;
+
+    /// <inheritdoc />
+    public DateTimeOffset LastHeartbeatAt { get; } = DateTimeOffset.UtcNow;
+
+    /// <inheritdoc />
+    public Task StartAsync(CancellationToken cancellationToken) => Task.CompletedTask;
+
+    /// <inheritdoc />
+    public Task<WorkerCommandReply> InvokeAsync(
+        WorkerCommand command,
+        TimeSpan timeout,
+        CancellationToken cancellationToken) => Task.FromResult(new WorkerCommandReply());
+
+    /// <inheritdoc />
+    public async IAsyncEnumerable<WorkerEvent> ReadEventsAsync(
+        [EnumeratorCancellation] CancellationToken cancellationToken)
+    {
+        await foreach (WorkerEvent workerEvent in events.ReadAllAsync(cancellationToken).ConfigureAwait(false))
+        {
+            yield return workerEvent;
+        }
+    }
+
+    /// <inheritdoc />
+    public Task ShutdownAsync(TimeSpan timeout, CancellationToken cancellationToken) => Task.CompletedTask;
+
+    /// <inheritdoc />
+    public void Kill(string reason)
+    {
+    }
+
+    /// <inheritdoc />
+    public ValueTask DisposeAsync() => ValueTask.CompletedTask;
+}