diff --git a/src/ZB.MOM.WW.ScadaBridge.DataConnectionLayer/Adapters/MxGatewayDataConnection.cs b/src/ZB.MOM.WW.ScadaBridge.DataConnectionLayer/Adapters/MxGatewayDataConnection.cs
index 20c22f61..c4d929fe 100644
--- a/src/ZB.MOM.WW.ScadaBridge.DataConnectionLayer/Adapters/MxGatewayDataConnection.cs
+++ b/src/ZB.MOM.WW.ScadaBridge.DataConnectionLayer/Adapters/MxGatewayDataConnection.cs
@@ -133,6 +133,23 @@ public class MxGatewayDataConnection
TaskContinuationOptions.OnlyOnFaulted);
}
+ // Same reasoning as the event loop above, but for the native-alarm feed: without
+ // this reset, the OLD RunAlarmStreamAsync loop keeps retrying against the now-
+ // disposed previous client every 5s forever (it only exits on its own token), and
+ // SubscribeAlarmsAsync's coverage check sees the stale _alarmCts/_alarmStreamPrefix
+ // and short-circuits — so the transparent re-subscribe the actor issues after a
+ // successful reconnect never opens a fresh stream on the NEW client. Mirror the
+ // lock-guarded block already in DisconnectAsync/DisposeAsync.
+ lock (_alarmLock)
+ {
+ _alarmCts?.Cancel();
+ _alarmCts?.Dispose();
+ _alarmCts = null;
+ _alarmSubCount = 0;
+ _alarmSubSources.Clear();
+ _alarmStreamPrefix = string.Empty;
+ }
+
Interlocked.Exchange(ref _disconnectFired, 0); // reset guard on (re)connect, like OPC UA
_client = _clientFactory.Create();
diff --git a/tests/ZB.MOM.WW.ScadaBridge.DataConnectionLayer.Tests/Adapters/MxGatewayDataConnectionAlarmReconnectTests.cs b/tests/ZB.MOM.WW.ScadaBridge.DataConnectionLayer.Tests/Adapters/MxGatewayDataConnectionAlarmReconnectTests.cs
new file mode 100644
index 00000000..c4b33e7f
--- /dev/null
+++ b/tests/ZB.MOM.WW.ScadaBridge.DataConnectionLayer.Tests/Adapters/MxGatewayDataConnectionAlarmReconnectTests.cs
@@ -0,0 +1,186 @@
+using Microsoft.Extensions.Logging.Abstractions;
+using ZB.MOM.WW.ScadaBridge.Commons.Types.Alarms;
+using ZB.MOM.WW.ScadaBridge.Commons.Types.Enums;
+using ZB.MOM.WW.ScadaBridge.DataConnectionLayer.Adapters;
+
+namespace ZB.MOM.WW.ScadaBridge.DataConnectionLayer.Tests.Adapters;
+
+///
+/// P2-2 (arch-review 2026-08-19, High): before the fix, ConnectAsync reset the
+/// event loop and the client but left the alarm-stream state (_alarmCts,
+/// _alarmSubCount, _alarmSubSources, _alarmStreamPrefix) untouched.
+/// After a reconnect, the DataConnectionActor's transparent re-subscribe
+/// (ReSubscribeAllAlarms) called SubscribeAlarmsAsync again, but the stale
+/// non-null _alarmCts plus the coverage check made it short-circuit and return
+/// success WITHOUT ever opening a new stream on the new client — leaving the orphaned
+/// old stream loop retrying the disposed old client forever and the native-alarm mirror
+/// frozen until failover/redeploy.
+///
+[Collection("DataConnectionManagerActor")]
+public class MxGatewayDataConnectionAlarmReconnectTests
+{
+ private static Dictionary Details() => new()
+ {
+ ["Endpoint"] = "http://gw:5000",
+ ["ApiKey"] = "key",
+ };
+
+ [Fact]
+ public async Task Reconnect_ThenSubscribeAlarms_OpensNewStreamOnNewClient()
+ {
+ var clients = new List();
+ var factory = new RecordingFactory(clients);
+ var adapter = new MxGatewayDataConnection(factory, NullLogger.Instance);
+
+ await adapter.ConnectAsync(Details());
+ await adapter.SubscribeAlarmsAsync("Area1.Tank1", null, _ => { });
+ await WaitUntilAsync(() => clients[0].AlarmStreamOpens.Count == 1);
+
+ // Reconnect (same source set as before — the coverage check would previously
+ // treat this as already-covered and never touch the new client at all).
+ await adapter.ConnectAsync(Details());
+ Assert.Equal(2, clients.Count);
+
+ await adapter.SubscribeAlarmsAsync("Area1.Tank1", null, _ => { });
+
+ await WaitUntilAsync(() => clients[1].AlarmStreamOpens.Count == 1);
+ Assert.Equal("Area1.Tank1", clients[1].AlarmStreamOpens[0]);
+ // The OLD client must never see a second alarm-stream open after reconnect.
+ Assert.Single(clients[0].AlarmStreamOpens);
+ }
+
+ [Fact]
+ public async Task Reconnect_CancelsTheOldAlarmStreamToken()
+ {
+ var clients = new List();
+ var factory = new RecordingFactory(clients);
+ var adapter = new MxGatewayDataConnection(factory, NullLogger.Instance);
+
+ await adapter.ConnectAsync(Details());
+ await adapter.SubscribeAlarmsAsync("Area1.Tank1", null, _ => { });
+ await WaitUntilAsync(() => clients[0].AlarmStreamTokens.Count == 1);
+ var oldStreamToken = clients[0].AlarmStreamTokens[0];
+ Assert.False(oldStreamToken.IsCancellationRequested);
+
+ await adapter.ConnectAsync(Details()); // reconnect
+
+ // The token the OLD RunAlarmStreamAsync loop was given must now be cancelled —
+ // that is what stops it retrying against the disposed old client every 5s.
+ Assert.True(oldStreamToken.IsCancellationRequested);
+ }
+
+ [Fact]
+ public async Task Reconnect_AlarmEventsFlowAgainOnTheNewClient()
+ {
+ var clients = new List();
+ var factory = new RecordingFactory(clients);
+ var adapter = new MxGatewayDataConnection(factory, NullLogger.Instance);
+
+ var received = new List();
+ await adapter.ConnectAsync(Details());
+ await adapter.SubscribeAlarmsAsync("Area1.Tank1", null, t => received.Add(t));
+ await WaitUntilAsync(() => clients[0].AlarmStreamOpens.Count == 1);
+
+ // Prove the feed works pre-reconnect.
+ clients[0].EmitAlarm(MakeTransition(AlarmTransitionKind.Snapshot));
+ await WaitUntilAsync(() => received.Count == 1);
+
+ await adapter.ConnectAsync(Details()); // reconnect
+ await adapter.SubscribeAlarmsAsync("Area1.Tank1", null, t => received.Add(t));
+ await WaitUntilAsync(() => clients[1].AlarmStreamOpens.Count == 1);
+
+ // A transition on the OLD (disposed, orphaned) client must never reach the
+ // callback again — only the NEW client's feed is live post-reconnect.
+ clients[0].EmitAlarm(MakeTransition(AlarmTransitionKind.Raise));
+ await Task.Delay(100);
+ Assert.Single(received);
+
+ clients[1].EmitAlarm(MakeTransition(AlarmTransitionKind.Snapshot));
+ await WaitUntilAsync(() => received.Count == 2);
+ }
+
+ private static NativeAlarmTransition MakeTransition(AlarmTransitionKind kind) => new(
+ "Area1.Tank1.HiAlarm", "Area1.Tank1", "HiAlarm", kind,
+ new AlarmConditionState(true, false, null, AlarmShelveState.Unshelved, false, 400),
+ "Area", "", "", "", "", null, DateTimeOffset.UtcNow, "", "");
+
+ private static async Task WaitUntilAsync(Func condition)
+ {
+ for (var i = 0; i < 100 && !condition(); i++)
+ await Task.Delay(50);
+ Assert.True(condition(), "condition not met within timeout");
+ }
+
+ /// Builds a NEW client instance per Create() call, mirroring the real
+ /// factory and 's NSubstitute-based
+ /// setup — needed here (unlike FakeMxGatewayClient) because the test must tell
+ /// the pre- and post-reconnect clients apart.
+ private sealed class RecordingFactory : IMxGatewayClientFactory
+ {
+ private readonly List _clients;
+ public RecordingFactory(List clients) => _clients = clients;
+
+ public IMxGatewayClient Create()
+ {
+ var c = new PerInstanceFakeMxGatewayClient();
+ _clients.Add(c);
+ return c;
+ }
+ }
+
+ /// Minimal per-instance fake — only the members this test exercises are
+ /// implemented meaningfully; the rest are unused no-ops for the connect/alarm-only
+ /// path under test.
+ private sealed class PerInstanceFakeMxGatewayClient : IMxGatewayClient
+ {
+ public readonly List AlarmStreamOpens = new();
+ public readonly List AlarmStreamTokens = new();
+ private Action? _onTransition;
+ private CancellationToken _streamToken;
+
+ public Task ConnectAsync(MxGatewayConnectionOptions o, CancellationToken ct = default) => Task.CompletedTask;
+ public Task DisconnectAsync(CancellationToken ct = default) => Task.CompletedTask;
+ public Task SubscribeAsync(string tagPath, CancellationToken ct = default) => Task.FromResult(Guid.NewGuid().ToString());
+ public Task UnsubscribeAsync(string subscriptionId, CancellationToken ct = default) => Task.CompletedTask;
+ public Task> SubscribeBulkAsync(IReadOnlyList tagPaths, CancellationToken ct = default)
+ => Task.FromResult>(Array.Empty());
+ public Task UnsubscribeBulkAsync(IReadOnlyList subscriptionIds, CancellationToken ct = default) => Task.CompletedTask;
+ public Task> ReadAsync(IReadOnlyList tagPaths, CancellationToken ct = default)
+ => Task.FromResult>(Array.Empty());
+ public Task> WriteAsync(IReadOnlyList<(string TagPath, object? Value)> writes, CancellationToken ct = default)
+ => Task.FromResult>(Array.Empty());
+ public Task<(IReadOnlyList Children, bool Truncated)> BrowseChildrenAsync(string? parentNodeId, CancellationToken ct = default)
+ => Task.FromResult<(IReadOnlyList, bool)>((Array.Empty(), false));
+ public Task RunEventLoopAsync(Action onUpdate, CancellationToken ct = default)
+ => Task.Delay(Timeout.Infinite, ct).ContinueWith(_ => { }, TaskScheduler.Default);
+
+ public Task RunAlarmStreamAsync(string? alarmFilterPrefix, Action onTransition, CancellationToken ct = default)
+ {
+ // Mirrors the real RunAlarmStreamAsync seam: records the open, then just
+ // parks on the token like the live gateway stream would between events —
+ // the adapter-level bug under test is about which CLIENT/TOKEN the loop was
+ // started on, not the internal 5s retry loop inside RealMxGatewayClient.
+ lock (AlarmStreamOpens)
+ {
+ AlarmStreamOpens.Add(alarmFilterPrefix);
+ AlarmStreamTokens.Add(ct);
+ }
+ _onTransition = onTransition;
+ _streamToken = ct;
+ return Task.Delay(Timeout.Infinite, ct).ContinueWith(_ => { }, TaskScheduler.Default);
+ }
+
+ public ValueTask DisposeAsync() => ValueTask.CompletedTask;
+
+ /// Simulates a transition arriving on this client's live stream. Gated on
+ /// the stream token exactly like the real gateway stream would be — once cancelled,
+ /// no more events are delivered — so a test can prove the OLD client's feed is
+ /// genuinely dead post-reconnect rather than just unobserved.
+ public void EmitAlarm(NativeAlarmTransition transition)
+ {
+ if (_streamToken.IsCancellationRequested)
+ return;
+ _onTransition?.Invoke(transition);
+ }
+ }
+}