fix(dcl): carry adapter generation on SubscribeCompleted/AlarmSubscribeCompleted; drop stale completions and release captured-adapter handles (arch-review 2026-08-19 P2-1 High)

An in-flight subscribe launched against adapter generation N can take up to
SeedOverallTimeout (120s) to answer. If the adapter is swapped in that window
(failover, or the row-97 in-place reconfigure), the completion was applied
unconditionally: it stored a dead session's handle under the tag key, so the
LIVE completion for the same tag read as a duplicate and had its working
monitored item released (tag resolved forever, no data); it re-stamped the old
endpoint's seed values with the CURRENT generation, bypassing the stale-data
guard; and a connection-level failure row from the dead adapter knocked the
healthy new connection back into Reconnecting. The alarm twin released the new
live alarm feed the same way.

Both completions now carry the generation AND the adapter instance they ran
against, and both handlers drop a completion whose generation no longer
matches - no handle store, no seed application, no per-tag or connection-level
failure processing, no duplicate-release logic. Handles the dropped completion
did create are released best-effort against the CAPTURED adapter (handles are
adapter-scoped; MxGateway's are plain ints that collide across sessions), and
the existing release sites in both handlers were switched to the message's
adapter for the same reason. The in-flight markers are deliberately left alone
(ReSubscribeAll/ReSubscribeAllAlarms clear them wholesale and may already have
re-registered the tag/source), and the requester gets a failure reply so its
retry timer re-sends.

Tests: DataConnectionStaleCompletionTests - stale handle not stored / live item
not released, release lands on the old adapter only, seed values not delivered,
stale connection-level failure does not re-enter Reconnecting, stale alarm
completion does not release the live feed. All five fail against the unfixed
handlers.

src/ZB.MOM.WW.ScadaBridge.DataConnectionLayer/Actors/DataConnectionActor.cs

@@ -983,7 +983,7 @@ public class DataConnectionActor : UntypedActor, IWithStash, IWithTimers
             // dropped, and logs any tag that never yields a value.
             var seedValues = await SeedTagsAsync(adapter, tagsToSeed);
 
-            return new SubscribeCompleted(request, sender, results, seedValues);
+            return new SubscribeCompleted(request, sender, results, seedValues, generation, adapter);
         }).PipeTo(self);
     }
 
@@ -1086,6 +1086,30 @@ public class DataConnectionActor : UntypedActor, IWithStash, IWithTimers
         return Task.WhenAll(subscriptionIds.Select(id => adapter.UnsubscribeAsync(id)));
     }
 
+    /// <summary>
+    /// Best-effort release of handles created by a subscribe whose adapter has since been
+    /// swapped out. <paramref name="release"/> must target the adapter that CREATED the handles
+    /// — they are adapter-scoped (MxGateway's are plain ints that collide across sessions), so
+    /// releasing by handle against the replacement adapter can tear down a live subscription.
+    /// The outgoing adapter is disposed fire-and-forget by <see cref="SwapAdapter"/> and may
+    /// already be gone, so a failure here is expected and logged at Debug rather than surfaced.
+    /// </summary>
+    private void ReleaseStaleAgainstOwnAdapter(string description, Func<Task> release)
+    {
+        try
+        {
+            release().ContinueWith(
+                t => _log.Debug("[{0}] Best-effort release of {1} failed: {2}",
+                    _connectionName, description, t.Exception?.GetBaseException().Message),
+                TaskContinuationOptions.OnlyOnFaulted);
+        }
+        catch (Exception ex)
+        {
+            _log.Debug("[{0}] Best-effort release of {1} threw: {2}",
+                _connectionName, description, ex.Message);
+        }
+    }
+
     /// <summary>
     /// Reads the current value of each tag so the Instance Actor
     /// has an initial value, retrying the still-empty subset a bounded number of times. A
@@ -1256,6 +1280,51 @@ public class DataConnectionActor : UntypedActor, IWithStash, IWithTimers
     private bool HandleSubscribeCompleted(SubscribeCompleted msg)
     {
         var instanceName = msg.Request.InstanceUniqueName;
+
+        // Results produced against an adapter that has since been swapped out (failover or
+        // in-place reconfigure) are dropped WHOLE, mirroring the BatchSubscribeCompleted and
+        // TagValueReceived generation guards. An initial subscribe can take up to
+        // SeedOverallTimeout to complete, so this window is wide. Applying a stale completion
+        // stores a dead session's handle under the tag's key, which then makes the LIVE
+        // completion for the same tag look like a duplicate and get RELEASED (the tag reads
+        // resolved forever and delivers nothing); re-stamps the old endpoint's seed values with
+        // the CURRENT generation, so they bypass the stale-data guard and present as fresh; and
+        // lets a connection-level failure row from the dead adapter knock the healthy new
+        // connection back into Reconnecting.
+        if (msg.AdapterGeneration != _adapterGeneration)
+        {
+            var staleIds = msg.Results
+                .Where(r => r is { Success: true, AlreadySubscribed: false, SubscriptionId: not null })
+                .Select(r => r.SubscriptionId!)
+                .ToList();
+
+            _log.Debug(
+                "[{0}] Dropping stale subscribe completion for instance {1} from adapter " +
+                "generation {2} (current {3}); releasing {4} handle(s) against the adapter that created them",
+                _connectionName, instanceName, msg.AdapterGeneration, _adapterGeneration, staleIds.Count);
+
+            // Release against msg.Adapter, NEVER _adapter: handles are adapter-scoped and
+            // MxGateway's are plain ints that collide across sessions, so releasing by handle
+            // against the replacement adapter can tear down a live subscription.
+            ReleaseStaleAgainstOwnAdapter(
+                $"{staleIds.Count} tag subscription handle(s) from adapter generation {msg.AdapterGeneration}",
+                () => UnsubscribeIdsAsync(msg.Adapter, staleIds));
+
+            // _subscribesInFlight is deliberately NOT cleared here: ReSubscribeAll empties it
+            // wholesale once the replacement adapter connects, and a marker the new generation
+            // has since placed for the same tag must not be removed by this dead message.
+
+            // Reply failure so the requester re-sends. Nothing has registered this request's
+            // tags in _subscriptionsByInstance (registration happens below, on the applied
+            // path), so ReSubscribeAll cannot pick them up — the InstanceActor's per-connection
+            // retry timer, armed until a Success arrives, is what re-establishes them.
+            msg.ReplyTo.Tell(new SubscribeTagsResponse(
+                msg.Request.CorrelationId, instanceName, false,
+                "adapter replaced while the subscribe was in flight — will retry",
+                DateTimeOffset.UtcNow));
+            return false;
+        }
+
         if (!_subscriptionsByInstance.TryGetValue(instanceName, out var instanceTags))
         {
             // The instance was unsubscribed while the
@@ -1292,7 +1361,10 @@ public class DataConnectionActor : UntypedActor, IWithStash, IWithTimers
                 }
             }
 
-            _ = UnsubscribeIdsAsync(_adapter, orphanedIds);
+            // Against the adapter that created them (identical to _adapter here — the
+            // generation guard above already proved the completion is current — but stating
+            // the ownership rule at every release site is what keeps it true).
+            _ = UnsubscribeIdsAsync(msg.Adapter, orphanedIds);
 
             // The original sender is already gone (unsubscribed). Telling a dead
             // ref produces a dead letter, which is the harmless and observable
@@ -1400,8 +1472,9 @@ public class DataConnectionActor : UntypedActor, IWithStash, IWithTimers
             }
         }
 
-        // Fire-and-forget release of every handle this request created redundantly.
+        // Fire-and-forget release of every handle this request created redundantly, against the
-        _ = UnsubscribeIdsAsync(_adapter, redundantIds);
+        // adapter that created them (see the orphaned-release site above).
+        _ = UnsubscribeIdsAsync(msg.Adapter, redundantIds);
 
         // Publish the resolution counts this request just changed. Without this push a
         // connection whose tags ALL resolve on the first attempt reported 0/0 forever:
@@ -2502,14 +2575,46 @@ public class DataConnectionActor : UntypedActor, IWithStash, IWithTimers
         alarmable.SubscribeAlarmsAsync(sourceRef, filter,
                 t => self.Tell(new AlarmTransitionReceived(t, generation)))
             .ContinueWith(task => task.IsCompletedSuccessfully
-                ? new AlarmSubscribeCompleted(sourceRef, true, task.Result, null, subscriber, corr, inst) as object
+                ? new AlarmSubscribeCompleted(
+                    sourceRef, true, task.Result, null, subscriber, corr, inst, generation, alarmable) as object
                 : new AlarmSubscribeCompleted(sourceRef, false, null,
-                    task.Exception?.GetBaseException().Message ?? "Unknown error", subscriber, corr, inst))
+                    task.Exception?.GetBaseException().Message ?? "Unknown error",
+                    subscriber, corr, inst, generation, alarmable))
             .PipeTo(self);
     }
 
     private void HandleAlarmSubscribeCompleted(AlarmSubscribeCompleted msg)
     {
+        // The alarm twin of HandleSubscribeCompleted's generation guard: a completion from an
+        // adapter that has since been swapped out is dropped whole. Applying it would store a
+        // dead feed's id — or, when the new adapter's feed is already stored, take the
+        // duplicate branch and release the LIVE feed, leaving the source mirrored but silent.
+        if (msg.AdapterGeneration != _adapterGeneration)
+        {
+            _log.Debug(
+                "[{0}] Dropping stale alarm subscribe completion for source {1} from adapter " +
+                "generation {2} (current {3})",
+                _connectionName, msg.SourceReference, msg.AdapterGeneration, _adapterGeneration);
+
+            // Release the orphaned feed against the adapter that created it, never the current
+            // one (see ReleaseStaleAgainstOwnAdapter).
+            if (msg is { Success: true, SubscriptionId: not null })
+                ReleaseStaleAgainstOwnAdapter(
+                    $"the alarm feed for {msg.SourceReference} from adapter generation {msg.AdapterGeneration}",
+                    () => msg.Adapter.UnsubscribeAlarmsAsync(msg.SubscriptionId));
+
+            // _alarmSubscribesInFlight is left untouched — ReSubscribeAllAlarms clears it
+            // wholesale and may already have re-registered this source against the new adapter.
+            // The subscriber state (_alarmSourceSubscribers) survives the swap, so the feed is
+            // re-established by ReSubscribeAllAlarms; the failure reply just lets the
+            // NativeAlarmActor retry rather than wait out its response timeout.
+            msg.ReplyTo?.Tell(new SubscribeAlarmsResponse(
+                msg.CorrelationId ?? string.Empty, msg.InstanceUniqueName ?? string.Empty,
+                false, "adapter replaced while the alarm subscribe was in flight — will retry",
+                DateTimeOffset.UtcNow));
+            return;
+        }
+
         _alarmSubscribesInFlight.Remove(msg.SourceReference);
 
         // The last (or only) subscriber may have been
@@ -2521,15 +2626,16 @@ public class DataConnectionActor : UntypedActor, IWithStash, IWithTimers
         // transitions to nobody for the lifetime of the adapter.
         if (!_alarmSourceSubscribers.ContainsKey(msg.SourceReference))
         {
-            if (msg.Success && msg.SubscriptionId != null &&
+            if (msg.Success && msg.SubscriptionId != null)
-                _adapter is IAlarmSubscribableConnection alarmable)
             {
                 _log.Warning(
                     "[{0}] AlarmSubscribeCompleted arrived for source {1} but the last " +
                     "subscriber unsubscribed while the subscribe was in flight; releasing " +
                     "the orphaned adapter alarm feed.",
                     _connectionName, msg.SourceReference);
-                _ = alarmable.UnsubscribeAlarmsAsync(msg.SubscriptionId);
+                // Against the adapter that created the feed (identical to _adapter here — the
+                // generation guard above already proved the completion is current).
+                _ = msg.Adapter.UnsubscribeAlarmsAsync(msg.SubscriptionId);
             }
 
             // No live requester remains to receive a response.
@@ -2546,15 +2652,12 @@ public class DataConnectionActor : UntypedActor, IWithStash, IWithTimers
             // completion is the redundant one — release its feed rather than overwriting
             // the stored id and leaking the already-tracked subscription.
             if (_alarmSubscriptionIds.ContainsKey(msg.SourceReference))
-            {
-                if (_adapter is IAlarmSubscribableConnection alarmable)
             {
                 _log.Warning(
                     "[{0}] Duplicate alarm feed for source {1}; releasing the redundant " +
                     "subscription instead of overwriting the stored one.",
                     _connectionName, msg.SourceReference);
-                    _ = alarmable.UnsubscribeAlarmsAsync(msg.SubscriptionId);
+                _ = msg.Adapter.UnsubscribeAlarmsAsync(msg.SubscriptionId);
-                }
             }
             else
             {
@@ -2755,9 +2858,11 @@ public class DataConnectionActor : UntypedActor, IWithStash, IWithTimers
             alarmable.SubscribeAlarmsAsync(sr, filter,
                     t => self.Tell(new AlarmTransitionReceived(t, generation)))
                 .ContinueWith(task => task.IsCompletedSuccessfully
-                    ? new AlarmSubscribeCompleted(sr, true, task.Result, null, null, null, null) as object
+                    ? new AlarmSubscribeCompleted(
+                        sr, true, task.Result, null, null, null, null, generation, alarmable) as object
                     : new AlarmSubscribeCompleted(sr, false, null,
-                        task.Exception?.GetBaseException().Message ?? "Unknown error", null, null, null))
+                        task.Exception?.GetBaseException().Message ?? "Unknown error",
+                        null, null, null, generation, alarmable))
                 .PipeTo(self);
         }
     }
@@ -2803,16 +2908,30 @@ public class DataConnectionActor : UntypedActor, IWithStash, IWithTimers
     internal record SubscribeTagResult(
         string TagPath, bool AlreadySubscribed, bool Success, string? SubscriptionId, string? Error,
         bool ConnectionLevelFailure = false);
+
+    /// <summary>
+    /// Per-tag results of the initial subscribe issued off the actor thread by
+    /// <see cref="HandleSubscribe"/>. Carries the adapter generation it ran against — and the
+    /// adapter INSTANCE that produced the handles, because a stale completion's handles must be
+    /// released against their own adapter (see <see cref="HandleSubscribeCompleted"/>).
+    /// </summary>
     internal record SubscribeCompleted(
         SubscribeTagsRequest Request, IActorRef ReplyTo, IReadOnlyList<SubscribeTagResult> Results,
-        IReadOnlyList<SeededValue> SeedValues);
+        IReadOnlyList<SeededValue> SeedValues, int AdapterGeneration, IDataConnection Adapter);
 
     /// <summary>An initial-read value captured during subscribe, delivered after the
     /// instance's tags are registered for fan-out.</summary>
     internal record SeededValue(string TagPath, TagValue Value);
     internal record AlarmTransitionReceived(NativeAlarmTransition Transition, int AdapterGeneration);
+
+    /// <summary>
+    /// Result of an alarm-feed subscribe issued off the actor thread. Carries the adapter
+    /// generation and the adapter instance for the same reason
+    /// <see cref="SubscribeCompleted"/> does.
+    /// </summary>
     internal record AlarmSubscribeCompleted(
         string SourceReference, bool Success, string? SubscriptionId, string? Error,
-        IActorRef? ReplyTo, string? CorrelationId, string? InstanceUniqueName);
+        IActorRef? ReplyTo, string? CorrelationId, string? InstanceUniqueName,
+        int AdapterGeneration, IAlarmSubscribableConnection Adapter);
     public record GetHealthReport;
 }

tests/ZB.MOM.WW.ScadaBridge.DataConnectionLayer.Tests/Actors/DataConnectionStaleCompletionTests.cs

@@ -0,0 +1,360 @@
+using Akka.Actor;
+using Akka.TestKit.Xunit2;
+using NSubstitute;
+using ZB.MOM.WW.ScadaBridge.Commons.Interfaces.Protocol;
+using ZB.MOM.WW.ScadaBridge.Commons.Messages.DataConnection;
+using ZB.MOM.WW.ScadaBridge.Commons.Types.Alarms;
+using ZB.MOM.WW.ScadaBridge.Commons.Types.Enums;
+using ZB.MOM.WW.ScadaBridge.DataConnectionLayer.Actors;
+using ZB.MOM.WW.ScadaBridge.HealthMonitoring;
+
+namespace ZB.MOM.WW.ScadaBridge.DataConnectionLayer.Tests.Actors;
+
+/// <summary>
+/// Arch-review 2026-08-19 <b>P2-1</b> — an in-flight subscribe that completes AFTER the adapter
+/// has been swapped out (failover, or the in-place reconfigure of register row 97).
+///
+/// <para>
+/// <b>The gap these pin.</b> <c>SubscribeCompleted</c>/<c>AlarmSubscribeCompleted</c> carried no
+/// adapter generation, and their handlers applied them unconditionally — while an initial
+/// subscribe can stay in flight for up to <c>SeedOverallTimeout</c> (120 s), so the window is
+/// wide. A stale completion stored a DEAD session's handle under the tag's key, which then made
+/// the live completion for the same tag look like a duplicate and get released: the tag reads
+/// resolved forever and delivers nothing. It also re-stamped the old endpoint's seed values with
+/// the CURRENT generation (bypassing the stale-data guard) and let a connection-level failure row
+/// from the dead adapter knock the healthy new connection back into Reconnecting. The alarm twin
+/// released the live alarm feed the same way.
+/// </para>
+///
+/// <para>
+/// Handles are adapter-scoped — MxGateway's are plain ints that collide across sessions — so a
+/// dropped completion's handles are released against the adapter that CREATED them, never the
+/// current one.
+/// </para>
+/// </summary>
+[Collection("DataConnectionManagerActor")]
+public class DataConnectionStaleCompletionTests : TestKit
+{
+    private const string OldEndpoint = "opc.tcp://old-plc:4840";
+    private const string NewEndpoint = "opc.tcp://new-plc:4840";
+    private const string Tag = "sensor/temp";
+    private const string Source = "Tank01";
+
+    private readonly ISiteHealthCollector _health = Substitute.For<ISiteHealthCollector>();
+    private readonly IDataConnectionFactory _factory = Substitute.For<IDataConnectionFactory>();
+
+    private readonly DataConnectionOptions _options = new()
+    {
+        ReconnectInterval = TimeSpan.FromMilliseconds(100),
+        TagResolutionRetryInterval = TimeSpan.FromMilliseconds(200),
+        WriteTimeout = TimeSpan.FromSeconds(5),
+        SeedReadMaxAttempts = 1,
+        SeedReadRetryDelay = TimeSpan.FromMilliseconds(10)
+    };
+
+    private static Dictionary<string, string> Config(string endpoint) =>
+        new() { ["Endpoint"] = endpoint };
+
+    private IActorRef CreateConnectionActor(string name, IDataConnection adapter) =>
+        Sys.ActorOf(Props.Create(() =>
+            new DataConnectionActor(
+                name, adapter, _options, _health, _factory, "OpcUa",
+                Config(OldEndpoint), null, 3)), name);
+
+    private void ExpectNewAdapter(IDataConnection adapter) =>
+        _factory.Create("OpcUa", Arg.Is<IDictionary<string, string>>(d => d["Endpoint"] == NewEndpoint))
+            .Returns(adapter);
+
+    private static IDataConnection HealthyAdapter(string subscriptionId)
+    {
+        var adapter = Substitute.For<IDataConnection>();
+        adapter.ConnectAsync(Arg.Any<IDictionary<string, string>>(), Arg.Any<CancellationToken>())
+            .Returns(Task.CompletedTask);
+        adapter.Status.Returns(ConnectionHealth.Connected);
+        adapter.SubscribeAsync(Arg.Any<string>(), Arg.Any<SubscriptionCallback>(), Arg.Any<CancellationToken>())
+            .Returns(subscriptionId);
+        adapter.ReadAsync(Arg.Any<string>(), Arg.Any<CancellationToken>())
+            .Returns(new ReadResult(false, null, null));
+        return adapter;
+    }
+
+    private static IDataConnection HealthyAlarmAdapter(string alarmSubscriptionId)
+    {
+        var adapter = Substitute.For<IDataConnection, IAlarmSubscribableConnection>();
+        adapter.ConnectAsync(Arg.Any<IDictionary<string, string>>(), Arg.Any<CancellationToken>())
+            .Returns(Task.CompletedTask);
+        adapter.Status.Returns(ConnectionHealth.Connected);
+        ((IAlarmSubscribableConnection)adapter)
+            .SubscribeAlarmsAsync(Arg.Any<string>(), Arg.Any<string?>(),
+                Arg.Any<AlarmTransitionCallback>(), Arg.Any<CancellationToken>())
+            .Returns(Task.FromResult(alarmSubscriptionId));
+        return adapter;
+    }
+
+    private static NativeAlarmTransition Raise(string sourceRef, string sourceObj) =>
+        new(sourceRef, sourceObj, "AnalogLimit.Hi", AlarmTransitionKind.Raise,
+            new AlarmConditionState(true, false, null, AlarmShelveState.Unshelved, false, 500),
+            "Process", "hi", "hi", "", "", null, DateTimeOffset.UtcNow, "92", "90");
+
+    private int HealthTransitions(string connectionName, ConnectionHealth health) =>
+        _health.ReceivedCalls().Count(c =>
+            c.GetMethodInfo().Name == nameof(ISiteHealthCollector.UpdateConnectionHealth) &&
+            c.GetArguments() is [string name, ConnectionHealth reported, ..] &&
+            name == connectionName && reported == health);
+
+    private static bool CalledWith(IDataConnection adapter, string method, string firstArg) =>
+        adapter.ReceivedCalls().Any(c =>
+            c.GetMethodInfo().Name == method && c.GetArguments()[0] as string == firstArg);
+
+    private static bool Called(IDataConnection adapter, string method) =>
+        adapter.ReceivedCalls().Any(c => c.GetMethodInfo().Name == method);
+
+    // ── Tag subscribe ──
+
+    /// <summary>
+    /// The headline data-loss case: the stale completion must not park its dead handle under the
+    /// tag's key, because the live completion for the same tag then reads as a duplicate and has
+    /// its (working) monitored item released — leaving the tag permanently resolved and silent.
+    ///
+    /// <para>
+    /// <b>Fails before the fix:</b> the stale completion stored <c>sub-old</c>, and the live
+    /// handle <c>sub-new</c> was released against the new adapter.
+    /// </para>
+    /// </summary>
+    [Fact]
+    public void StaleSubscribeCompleted_DoesNotStoreItsHandle_AndLeavesTheLiveMonitoredItemAlone()
+    {
+        const string name = "stale-sub-handle";
+        var gate = new TaskCompletionSource<string>();
+
+        var oldAdapter = HealthyAdapter("sub-old");
+        oldAdapter.SubscribeAsync(Tag, Arg.Any<SubscriptionCallback>(), Arg.Any<CancellationToken>())
+            .Returns(_ => gate.Task);
+        var newAdapter = HealthyAdapter("sub-new");
+        ExpectNewAdapter(newAdapter);
+
+        var actor = CreateConnectionActor(name, oldAdapter);
+        AwaitCondition(() => HealthTransitions(name, ConnectionHealth.Connected) >= 1,
+            TimeSpan.FromSeconds(5), TimeSpan.FromMilliseconds(10));
+
+        var probe = CreateTestProbe();
+        probe.Send(actor, new SubscribeTagsRequest("c1", "inst1", name, [Tag], DateTimeOffset.UtcNow));
+        AwaitCondition(() => CalledWith(oldAdapter, nameof(IDataConnection.SubscribeAsync), Tag),
+            TimeSpan.FromSeconds(5));
+
+        actor.Tell(new ReconfigureConnectionCommand(name, "OpcUa", Config(NewEndpoint), null, 3));
+        AwaitCondition(() => HealthTransitions(name, ConnectionHealth.Connected) >= 2,
+            TimeSpan.FromSeconds(5), TimeSpan.FromMilliseconds(10));
+
+        // The Instance Actor's retry — the live subscribe, against the new adapter.
+        probe.Send(actor, new SubscribeTagsRequest("c2", "inst1", name, [Tag], DateTimeOffset.UtcNow));
+        probe.FishForMessage<SubscribeTagsResponse>(
+            m => m.CorrelationId == "c2" && m.Success, TimeSpan.FromSeconds(5));
+
+        // Only now does the generation-N subscribe answer.
+        gate.SetResult("sub-old");
+        AwaitCondition(() => CalledWith(oldAdapter, nameof(IDataConnection.UnsubscribeAsync), "sub-old"),
+            TimeSpan.FromSeconds(5), TimeSpan.FromMilliseconds(10));
+
+        Assert.False(Called(newAdapter, nameof(IDataConnection.UnsubscribeAsync)));
+
+        // The registry still holds the LIVE handle: the teardown releases sub-new, not sub-old.
+        probe.Send(actor, new UnsubscribeTagsRequest("c3", "inst1", name, DateTimeOffset.UtcNow));
+        AwaitCondition(() => CalledWith(newAdapter, nameof(IDataConnection.UnsubscribeAsync), "sub-new"),
+            TimeSpan.FromSeconds(5), TimeSpan.FromMilliseconds(10));
+        Assert.False(CalledWith(newAdapter, nameof(IDataConnection.UnsubscribeAsync), "sub-old"));
+    }
+
+    /// <summary>
+    /// The handles a dropped completion did create are released — against the adapter that
+    /// created them. Releasing them against the CURRENT adapter is not a harmless mismatch:
+    /// MxGateway subscription handles are plain ints reused across sessions, so the release
+    /// would land on whatever the new session happens to have under that number.
+    /// </summary>
+    [Fact]
+    public void StaleSubscribeCompleted_ReleasesItsHandles_AgainstTheAdapterThatCreatedThem()
+    {
+        const string name = "stale-sub-release";
+        var gate = new TaskCompletionSource<string>();
+
+        var oldAdapter = HealthyAdapter("sub-old");
+        oldAdapter.SubscribeAsync(Tag, Arg.Any<SubscriptionCallback>(), Arg.Any<CancellationToken>())
+            .Returns(_ => gate.Task);
+        var newAdapter = HealthyAdapter("sub-new");
+        ExpectNewAdapter(newAdapter);
+
+        var actor = CreateConnectionActor(name, oldAdapter);
+        AwaitCondition(() => HealthTransitions(name, ConnectionHealth.Connected) >= 1,
+            TimeSpan.FromSeconds(5), TimeSpan.FromMilliseconds(10));
+
+        var probe = CreateTestProbe();
+        probe.Send(actor, new SubscribeTagsRequest("c1", "inst1", name, [Tag], DateTimeOffset.UtcNow));
+        AwaitCondition(() => CalledWith(oldAdapter, nameof(IDataConnection.SubscribeAsync), Tag),
+            TimeSpan.FromSeconds(5));
+
+        actor.Tell(new ReconfigureConnectionCommand(name, "OpcUa", Config(NewEndpoint), null, 3));
+        AwaitCondition(() => HealthTransitions(name, ConnectionHealth.Connected) >= 2,
+            TimeSpan.FromSeconds(5), TimeSpan.FromMilliseconds(10));
+
+        gate.SetResult("sub-old");
+
+        AwaitCondition(() => CalledWith(oldAdapter, nameof(IDataConnection.UnsubscribeAsync), "sub-old"),
+            TimeSpan.FromSeconds(5), TimeSpan.FromMilliseconds(10));
+        Assert.False(Called(newAdapter, nameof(IDataConnection.UnsubscribeAsync)));
+
+        // The requester is told the subscribe did not take, so its retry timer re-sends: the
+        // dropped completion registered nothing, so the reconnect re-subscribe cannot cover it.
+        probe.FishForMessage<SubscribeTagsResponse>(
+            m => m.CorrelationId == "c1" && !m.Success, TimeSpan.FromSeconds(5));
+    }
+
+    /// <summary>
+    /// The seed values a dropped completion carries are readings from the endpoint the operator
+    /// has just corrected AWAY from. Applying them re-stamped each one with the CURRENT adapter
+    /// generation, which is precisely what the generation guard on <c>TagValueReceived</c>
+    /// exists to stop — the old endpoint's process values presented as fresh.
+    /// </summary>
+    [Fact]
+    public void StaleSubscribeCompleted_DoesNotDeliverItsSeedValues()
+    {
+        const string name = "stale-sub-seed";
+        var gate = new TaskCompletionSource<string>();
+
+        var oldAdapter = HealthyAdapter("sub-old");
+        oldAdapter.SubscribeAsync(Tag, Arg.Any<SubscriptionCallback>(), Arg.Any<CancellationToken>())
+            .Returns(_ => gate.Task);
+        oldAdapter.ReadAsync(Tag, Arg.Any<CancellationToken>())
+            .Returns(new ReadResult(true, new TagValue(999, QualityCode.Good, DateTimeOffset.UtcNow), null));
+        var newAdapter = HealthyAdapter("sub-new");
+        ExpectNewAdapter(newAdapter);
+
+        var actor = CreateConnectionActor(name, oldAdapter);
+        AwaitCondition(() => HealthTransitions(name, ConnectionHealth.Connected) >= 1,
+            TimeSpan.FromSeconds(5), TimeSpan.FromMilliseconds(10));
+
+        var probe = CreateTestProbe();
+        probe.Send(actor, new SubscribeTagsRequest("c1", "inst1", name, [Tag], DateTimeOffset.UtcNow));
+        AwaitCondition(() => CalledWith(oldAdapter, nameof(IDataConnection.SubscribeAsync), Tag),
+            TimeSpan.FromSeconds(5));
+
+        actor.Tell(new ReconfigureConnectionCommand(name, "OpcUa", Config(NewEndpoint), null, 3));
+        AwaitCondition(() => HealthTransitions(name, ConnectionHealth.Connected) >= 2,
+            TimeSpan.FromSeconds(5), TimeSpan.FromMilliseconds(10));
+
+        gate.SetResult("sub-old");
+
+        var received = probe.ReceiveWhile(TimeSpan.FromSeconds(2), m => m);
+
+        // Positive control: the completion really did arrive and was dropped, rather than the
+        // window simply having been too short to observe anything.
+        Assert.Contains(received, m => m is SubscribeTagsResponse { Success: false });
+        Assert.DoesNotContain(received, m => m is TagValueUpdate u && Equals(u.Value, 999));
+    }
+
+    /// <summary>
+    /// A connection-level failure row belongs to the adapter that produced it. Acting on a stale
+    /// one drives the healthy replacement connection into Reconnecting — dropping every tag to
+    /// bad quality and re-subscribing the whole registry for nothing.
+    /// </summary>
+    [Fact]
+    public void StaleConnectionLevelFailure_DoesNotDriveTheNewConnectionBackToReconnecting()
+    {
+        const string name = "stale-sub-connfail";
+        var gate = new TaskCompletionSource<string>();
+
+        var oldAdapter = HealthyAdapter("sub-old");
+        oldAdapter.SubscribeAsync(Tag, Arg.Any<SubscriptionCallback>(), Arg.Any<CancellationToken>())
+            .Returns(_ => gate.Task);
+        var newAdapter = HealthyAdapter("sub-new");
+        ExpectNewAdapter(newAdapter);
+
+        var actor = CreateConnectionActor(name, oldAdapter);
+        AwaitCondition(() => HealthTransitions(name, ConnectionHealth.Connected) >= 1,
+            TimeSpan.FromSeconds(5), TimeSpan.FromMilliseconds(10));
+
+        var probe = CreateTestProbe();
+        probe.Send(actor, new SubscribeTagsRequest("c1", "inst1", name, [Tag], DateTimeOffset.UtcNow));
+        AwaitCondition(() => CalledWith(oldAdapter, nameof(IDataConnection.SubscribeAsync), Tag),
+            TimeSpan.FromSeconds(5));
+
+        actor.Tell(new ReconfigureConnectionCommand(name, "OpcUa", Config(NewEndpoint), null, 3));
+        AwaitCondition(() => HealthTransitions(name, ConnectionHealth.Connected) >= 2,
+            TimeSpan.FromSeconds(5), TimeSpan.FromMilliseconds(10));
+
+        var disconnectsBefore = HealthTransitions(name, ConnectionHealth.Disconnected);
+
+        // InvalidOperationException is what IsConnectionLevelFailure classifies as "transport
+        // down", i.e. the row that asks the state machine to reconnect.
+        gate.SetException(new InvalidOperationException("adapter not connected"));
+
+        probe.FishForMessage<SubscribeTagsResponse>(
+            m => m.CorrelationId == "c1" && !m.Success, TimeSpan.FromSeconds(5));
+
+        // Entering Reconnecting would push bad quality to every subscriber — the probe is one.
+        probe.ExpectNoMsg(TimeSpan.FromMilliseconds(500));
+        Assert.Equal(disconnectsBefore, HealthTransitions(name, ConnectionHealth.Disconnected));
+        Assert.Equal(2, HealthTransitions(name, ConnectionHealth.Connected));
+    }
+
+    // ── Alarm subscribe ──
+
+    /// <summary>
+    /// The alarm twin. The stale completion lands first and (before the fix) stored its dead feed
+    /// id, so the LIVE feed's completion took the duplicate branch and released the feed the new
+    /// adapter had just established — a source that stays mirrored but never updates again.
+    /// </summary>
+    [Fact]
+    public void StaleAlarmSubscribeCompleted_DoesNotReleaseTheLiveAlarmFeed()
+    {
+        const string name = "stale-alarm-feed";
+        var oldGate = new TaskCompletionSource<string>();
+        var newGate = new TaskCompletionSource<string>();
+        AlarmTransitionCallback? newCallback = null;
+
+        var oldAdapter = HealthyAlarmAdapter("alarm-old");
+        ((IAlarmSubscribableConnection)oldAdapter)
+            .SubscribeAlarmsAsync(Source, Arg.Any<string?>(),
+                Arg.Any<AlarmTransitionCallback>(), Arg.Any<CancellationToken>())
+            .Returns(_ => oldGate.Task);
+
+        var newAdapter = HealthyAlarmAdapter("alarm-new");
+        ((IAlarmSubscribableConnection)newAdapter)
+            .SubscribeAlarmsAsync(Source, Arg.Any<string?>(),
+                Arg.Do<AlarmTransitionCallback>(c => newCallback = c), Arg.Any<CancellationToken>())
+            .Returns(_ => newGate.Task);
+        ExpectNewAdapter(newAdapter);
+
+        var actor = CreateConnectionActor(name, oldAdapter);
+        AwaitCondition(() => HealthTransitions(name, ConnectionHealth.Connected) >= 1,
+            TimeSpan.FromSeconds(5), TimeSpan.FromMilliseconds(10));
+
+        var probe = CreateTestProbe();
+        probe.Send(actor, new SubscribeAlarmsRequest("c1", "inst1", name, Source, null, DateTimeOffset.UtcNow));
+        AwaitCondition(() => CalledWith(oldAdapter, "SubscribeAlarmsAsync", Source), TimeSpan.FromSeconds(5));
+
+        actor.Tell(new ReconfigureConnectionCommand(name, "OpcUa", Config(NewEndpoint), null, 3));
+
+        // The subscriber registry survives the swap, so the reconnect re-subscribe re-opens the
+        // feed against the new adapter without the NativeAlarmActor doing anything.
+        AwaitCondition(() => CalledWith(newAdapter, "SubscribeAlarmsAsync", Source),
+            TimeSpan.FromSeconds(5), TimeSpan.FromMilliseconds(10));
+
+        // The dead feed answers first, the live one second — the ordering that made the live
+        // completion look redundant.
+        oldGate.SetResult("alarm-old");
+        AwaitCondition(() => CalledWith(oldAdapter, "UnsubscribeAlarmsAsync", "alarm-old"),
+            TimeSpan.FromSeconds(5), TimeSpan.FromMilliseconds(10));
+        newGate.SetResult("alarm-new");
+
+        AwaitCondition(() => newCallback != null, TimeSpan.FromSeconds(5));
+        // Settle (and drain the swap's expected NativeAlarmSourceUnavailable) so the live
+        // completion has certainly been applied before asserting nothing released its feed.
+        probe.ReceiveWhile(TimeSpan.FromMilliseconds(300), m => m);
+        Assert.False(Called(newAdapter, "UnsubscribeAlarmsAsync"));
+
+        // And the feed is genuinely live: a transition on it still routes to the subscriber.
+        newCallback!(Raise("Tank01.Hi", Source));
+        probe.FishForMessage<NativeAlarmTransitionUpdate>(
+            u => u.Transition.SourceObjectReference == Source, TimeSpan.FromSeconds(5));
+    }
+}