fix(twincat): replay native ADS notifications after reconnect (archreview STAB-2 / Critical 4)

Native ADS notifications (the default subscribe mode) were stored as opaque handles with no record of the symbol/type/interval/handler needed to replay. On a client swap (EnsureConnectedAsync building a fresh client after a drop) the notifications were silently orphaned — no Bad status, no error, pushes just stopped until redeploy. Compounding: the IsConnected fast-path keys on AMS-port state, not wire liveness, and a probe failure only transitioned state without recycling the dead client. Fix: - Store REPLAYABLE INTENT: NativeRegistration (symbol/type/bit/interval/onChange + swappable live handle) hung off DeviceState.NativeRegistrations, populated by SubscribeAsync via RegisterNotificationAsync (under ConnectGate). - Split EnsureConnectedAsync into a gate wrapper + EnsureConnectedUnderGateAsync core; when the core installs a NEW client it replays every stored intent onto it and swaps the live handle (disposing the dead one). Register + replay both run under ConnectGate so they can't race. - Probe loop: on a wire-probe failure (false or throw) RecycleClientAsync disposes+nulls the client so the next tick rebuilds + replays — closes the fast-path-keys-on-port-state compounding bug. No TwinCAT docker fixture exists (integration needs a real TC3 XAR), so the fake-client unit tests are the authoritative coverage: - 4 new guards in TwinCATReconnectReplayTests (replay-onto-fresh-client + push reaches OnDataChange + old handle disposed; replay-all-tags; unsubscribe-after-reconnect stops replaying; probe-failure recycles+rebuilds). - Full TwinCAT unit suite 174/174 green; full solution builds 0 errors.
2026-07-08 17:20:10 -04:00
parent 9cad9ed0fc
commit af318fb442
3 changed files with 413 additions and 39 deletions
@@ -440,12 +440,14 @@ public sealed class TwinCATDriver : IDriver, IReadable, IWritable, ITagDiscovery

    private readonly ConcurrentDictionary<long, NativeSubscription> _nativeSubs = new();
    private long _nextNativeSubId;
+    private long _nextRegId;

    /// <inheritdoc />
    // Subscribe via native ADS notifications when TwinCATDriverOptions.UseNativeNotifications is
    // true, otherwise fall through to the shared PollGroupEngine. Native path registers one
-    // ITwinCATNotificationHandle per tag against the target's PLC runtime — the PLC pushes
-    // changes on its own cycle so we skip the poll loop entirely. Unsub path disposes the handles.
+    // notification per tag against the target's PLC runtime — the PLC pushes changes on its own
+    // cycle so we skip the poll loop entirely. Each registration stores its REPLAYABLE INTENT on
+    // the owning device so EnsureConnectedAsync can re-register it after a reconnect (STAB-2).
    public async Task<ISubscriptionHandle> SubscribeAsync(
        IReadOnlyList<string> fullReferences, TimeSpan publishingInterval, CancellationToken cancellationToken)
    {
@@ -454,8 +456,7 @@ public sealed class TwinCATDriver : IDriver, IReadable, IWritable, ITagDiscovery

        var id = Interlocked.Increment(ref _nextNativeSubId);
        var handle = new NativeSubscriptionHandle(id);
-        var registrations = new List<ITwinCATNotificationHandle>(fullReferences.Count);
-        var now = DateTime.UtcNow;
+        var registrations = new List<NativeRegistration>(fullReferences.Count);

        try
        {
@@ -464,18 +465,20 @@ public sealed class TwinCATDriver : IDriver, IReadable, IWritable, ITagDiscovery
                if (!_resolver.TryResolve(reference, out var def)) continue;
                if (!_devices.TryGetValue(def.DeviceHostAddress, out var device)) continue;

-                var client = await EnsureConnectedAsync(device, cancellationToken).ConfigureAwait(false);
                var parsed = TwinCATSymbolPath.TryParse(def.SymbolPath);
                var symbolName = parsed?.ToAdsSymbolName() ?? def.SymbolPath;
                var bitIndex = parsed?.BitIndex;

-                var reg = await client.AddNotificationAsync(
-                    symbolName, def.DataType, bitIndex, publishingInterval,
-                    _options.NotificationMaxDelayMs,
+                // The callback closes over the stable (handle, reference) pair, so it survives a
+                // client swap unchanged — replay re-registers the SAME intent onto the new client.
+                var reg = new NativeRegistration(
+                    Interlocked.Increment(ref _nextRegId), device, reference, symbolName,
+                    def.DataType, bitIndex, publishingInterval,
                    (_, value) => OnDataChange?.Invoke(this,
                        new DataChangeEventArgs(handle, reference, new DataValueSnapshot(
-                            value, TwinCATStatusMapper.Good, DateTime.UtcNow, DateTime.UtcNow))),
-                    cancellationToken).ConfigureAwait(false);
+                            value, TwinCATStatusMapper.Good, DateTime.UtcNow, DateTime.UtcNow))));
+
+                await RegisterNotificationAsync(device, reg, cancellationToken).ConfigureAwait(false);
                registrations.Add(reg);
            }
        }
@@ -496,11 +499,38 @@ public sealed class TwinCATDriver : IDriver, IReadable, IWritable, ITagDiscovery
        return handle;
    }

+    /// <summary>
+    ///     Registers one native notification and records its replayable intent on the device — both
+    ///     under the device <see cref="DeviceState.ConnectGate"/> so the register can't race a
+    ///     concurrent reconnect-replay (which also runs under that gate). The intent is added to the
+    ///     device registry only AFTER the handle is live, so a replay triggered by this same
+    ///     <c>EnsureConnected</c> can't double-register the not-yet-stored tag.
+    /// </summary>
+    private async Task RegisterNotificationAsync(DeviceState device, NativeRegistration reg, CancellationToken ct)
+    {
+        await device.ConnectGate.WaitAsync(ct).ConfigureAwait(false);
+        try
+        {
+            var client = await EnsureConnectedUnderGateAsync(device, ct, timeoutOverride: null).ConfigureAwait(false);
+            var newHandle = await client.AddNotificationAsync(
+                reg.SymbolName, reg.DataType, reg.BitIndex, reg.Interval,
+                _options.NotificationMaxDelayMs, reg.OnChange, ct).ConfigureAwait(false);
+            reg.SwapHandle(newHandle); // initial set — previous handle is null
+            device.NativeRegistrations[reg.Id] = reg;
+        }
+        finally
+        {
+            device.ConnectGate.Release();
+        }
+    }
+
    /// <inheritdoc />
    public Task UnsubscribeAsync(ISubscriptionHandle handle, CancellationToken cancellationToken)
    {
        if (handle is NativeSubscriptionHandle native && _nativeSubs.TryRemove(native.Id, out var sub))
        {
+            // NativeRegistration.Dispose removes itself from its device registry (so a later replay
+            // won't touch it) and disposes its live handle.
            foreach (var r in sub.Registrations) { try { r.Dispose(); } catch { } }
            return Task.CompletedTask;
        }
@@ -516,7 +546,64 @@ public sealed class TwinCATDriver : IDriver, IReadable, IWritable, ITagDiscovery

    private sealed record NativeSubscription(
        NativeSubscriptionHandle Handle,
-        IReadOnlyList<ITwinCATNotificationHandle> Registrations);
+        IReadOnlyList<NativeRegistration> Registrations);
+
+    /// <summary>
+    ///     One live native-notification registration: its replayable intent plus the current live
+    ///     <see cref="ITwinCATNotificationHandle"/>. The handle is swapped (not re-created) when the
+    ///     device reconnects, so unsubscribe / teardown always dispose the latest one. The handle
+    ///     field is exchanged atomically so a reconnect-replay swap and an unsubscribe dispose can't
+    ///     corrupt each other on the common path.
+    /// </summary>
+    internal sealed class NativeRegistration : IDisposable
+    {
+        /// <summary>Globally-unique registration id (registry key on the device).</summary>
+        public long Id { get; }
+        /// <summary>The device this notification is registered against.</summary>
+        public DeviceState Device { get; }
+        /// <summary>Driver-side full reference the pushed value is reported under.</summary>
+        public string Reference { get; }
+        /// <summary>ADS symbol name (resolved from the symbol path).</summary>
+        public string SymbolName { get; }
+        /// <summary>Declared data type.</summary>
+        public TwinCATDataType DataType { get; }
+        /// <summary>Bit index for a BOOL-within-word symbol; null otherwise.</summary>
+        public int? BitIndex { get; }
+        /// <summary>Requested notification interval.</summary>
+        public TimeSpan Interval { get; }
+        /// <summary>Value-change callback (closes over the owning subscription handle + reference).</summary>
+        public Action<string, object?> OnChange { get; }
+
+        private ITwinCATNotificationHandle? _handle;
+
+        /// <summary>Initializes a new instance of the <see cref="NativeRegistration"/> class.</summary>
+        public NativeRegistration(long id, DeviceState device, string reference, string symbolName,
+            TwinCATDataType dataType, int? bitIndex, TimeSpan interval, Action<string, object?> onChange)
+        {
+            Id = id;
+            Device = device;
+            Reference = reference;
+            SymbolName = symbolName;
+            DataType = dataType;
+            BitIndex = bitIndex;
+            Interval = interval;
+            OnChange = onChange;
+        }
+
+        /// <summary>Installs a freshly-registered handle and returns the previous one (null on first set).</summary>
+        /// <param name="newHandle">The newly-registered notification handle.</param>
+        /// <returns>The previous handle, or null.</returns>
+        public ITwinCATNotificationHandle? SwapHandle(ITwinCATNotificationHandle newHandle) =>
+            Interlocked.Exchange(ref _handle, newHandle);
+
+        /// <summary>Removes this registration from its device and disposes its live handle.</summary>
+        public void Dispose()
+        {
+            Device.NativeRegistrations.TryRemove(Id, out _);
+            var h = Interlocked.Exchange(ref _handle, null);
+            try { h?.Dispose(); } catch { /* best-effort */ }
+        }
+    }

    // ---- IHostConnectivityProbe ----

@@ -536,12 +623,21 @@ public sealed class TwinCATDriver : IDriver, IReadable, IWritable, ITagDiscovery
                var client = await EnsureConnectedAsync(state, ct, _options.Probe.Timeout)
                    .ConfigureAwait(false);
                success = await client.ProbeAsync(ct).ConfigureAwait(false);
+
+                // A locally-connected client whose wire probe FAILS is wire-dead but port-open:
+                // force a recycle so the next tick rebuilds + replays native notifications. Without
+                // this the IsConnected fast-path would keep handing back the dead client forever
+                // (the STAB-2 compounding bug — native pushes would never recover).
+                if (!success)
+                    await RecycleClientAsync(state).ConfigureAwait(false);
            }
            catch (OperationCanceledException) when (ct.IsCancellationRequested) { break; }
            catch
            {
-                // Probe failure — EnsureConnectedAsync's connect-failure path already disposed
-                // + cleared the client, so next tick will reconnect.
+                // Probe threw — the connect-failure path clears the client only when the CONNECT
+                // failed; a throw from ProbeAsync on an already-connected client would leave it in
+                // place, so force a recycle to guarantee the next tick rebuilds + replays.
+                await RecycleClientAsync(state).ConfigureAwait(false);
            }

            TransitionDeviceState(state, success ? HostState.Running : HostState.Stopped);
@@ -608,41 +704,126 @@ public sealed class TwinCATDriver : IDriver, IReadable, IWritable, ITagDiscovery
        await device.ConnectGate.WaitAsync(ct).ConfigureAwait(false);
        try
        {
-            // Re-check under the gate: another caller may have connected while we waited.
-            if (device.Client is { IsConnected: true } c) return c;
+            return await EnsureConnectedUnderGateAsync(device, ct, timeoutOverride).ConfigureAwait(false);
+        }
+        finally
+        {
+            device.ConnectGate.Release();
+        }
+    }

-            // Discard a stale (created-but-disconnected) client before making a fresh one.
-            if (device.Client is { IsConnected: false } stale)
-            {
-                try { stale.Dispose(); } catch { /* best-effort */ }
-                device.Client = null;
-            }
+    /// <summary>
+    ///     The connect-or-reconnect core, assuming the caller already holds
+    ///     <see cref="DeviceState.ConnectGate"/>. When it installs a NEW client (a genuine
+    ///     reconnect, not a fast-path reuse) it replays every stored native-notification intent
+    ///     onto it — otherwise native ADS pushes would silently stop after a drop (STAB-2). Both
+    ///     the initial subscribe (<see cref="RegisterNotificationAsync"/>) and the probe loop call
+    ///     through here under the same gate, so a replay can never race a register.
+    /// </summary>
+    private async Task<ITwinCATClient> EnsureConnectedUnderGateAsync(
+        DeviceState device, CancellationToken ct, TimeSpan? timeoutOverride)
+    {
+        // Re-check under the gate: another caller may have connected while we waited.
+        if (device.Client is { IsConnected: true } c) return c;

-            var client = _clientFactory.Create();
-            client.OnSymbolVersionChanged += HandleSymbolVersionChanged;
-            // timeoutOverride lets the probe loop use TwinCATProbeOptions.Timeout for probe-
-            // initiated connects rather than the driver-level _options.Timeout. Reads / writes
-            // pass null and get the driver default.
-            var effectiveTimeout = timeoutOverride ?? _options.Timeout;
+        // Discard a stale (created-but-disconnected) client before making a fresh one.
+        if (device.Client is { IsConnected: false } stale)
+        {
+            stale.OnSymbolVersionChanged -= HandleSymbolVersionChanged;
+            try { stale.Dispose(); } catch { /* best-effort */ }
+            device.Client = null;
+        }
+
+        var client = _clientFactory.Create();
+        client.OnSymbolVersionChanged += HandleSymbolVersionChanged;
+        // timeoutOverride lets the probe loop use TwinCATProbeOptions.Timeout for probe-
+        // initiated connects rather than the driver-level _options.Timeout. Reads / writes
+        // pass null and get the driver default.
+        var effectiveTimeout = timeoutOverride ?? _options.Timeout;
+        try
+        {
+            await client.ConnectAsync(device.ParsedAddress, effectiveTimeout, ct)
+                .ConfigureAwait(false);
+            _logger.LogInformation(
+                "TwinCAT driver '{DriverInstanceId}' connected to {HostAddress}",
+                _driverInstanceId, device.Options.HostAddress);
+        }
+        catch (Exception ex)
+        {
+            _logger.LogWarning(ex,
+                "TwinCAT driver '{DriverInstanceId}' failed to connect to {HostAddress}",
+                _driverInstanceId, device.Options.HostAddress);
+            client.OnSymbolVersionChanged -= HandleSymbolVersionChanged;
+            client.Dispose();
+            throw;
+        }
+        device.Client = client;
+
+        // Re-register every native notification onto the fresh client. Native ADS is push, so
+        // (unlike the poll fallback) it cannot self-heal by re-reading — without this the
+        // subscription is silently dead after a reconnect.
+        await ReplayNativeRegistrationsAsync(device, client, ct).ConfigureAwait(false);
+        return client;
+    }
+
+    /// <summary>
+    ///     Re-registers every stored <see cref="NativeRegistration"/> for a device onto the given
+    ///     freshly-connected client and swaps in the new live handle (disposing the dead one from
+    ///     the previous client). Runs under <see cref="DeviceState.ConnectGate"/>. A single
+    ///     registration that fails to replay is logged and skipped — one bad symbol must not abort
+    ///     recovery of the rest.
+    /// </summary>
+    private async Task ReplayNativeRegistrationsAsync(DeviceState device, ITwinCATClient client, CancellationToken ct)
+    {
+        var intents = device.NativeRegistrations.Values.ToArray();
+        if (intents.Length == 0) return;
+
+        var replayed = 0;
+        foreach (var reg in intents)
+        {
            try
            {
-                await client.ConnectAsync(device.ParsedAddress, effectiveTimeout, ct)
-                    .ConfigureAwait(false);
-                _logger.LogInformation(
-                    "TwinCAT driver '{DriverInstanceId}' connected to {HostAddress}",
-                    _driverInstanceId, device.Options.HostAddress);
+                var newHandle = await client.AddNotificationAsync(
+                    reg.SymbolName, reg.DataType, reg.BitIndex, reg.Interval,
+                    _options.NotificationMaxDelayMs, reg.OnChange, ct).ConfigureAwait(false);
+                var old = reg.SwapHandle(newHandle);
+                try { old?.Dispose(); } catch { /* dead handle from the disposed client */ }
+                replayed++;
            }
            catch (Exception ex)
            {
                _logger.LogWarning(ex,
-                    "TwinCAT driver '{DriverInstanceId}' failed to connect to {HostAddress}",
-                    _driverInstanceId, device.Options.HostAddress);
-                client.OnSymbolVersionChanged -= HandleSymbolVersionChanged;
-                client.Dispose();
-                throw;
+                    "TwinCAT driver '{DriverInstanceId}' failed to replay native notification " +
+                    "'{Symbol}' on device '{HostAddress}' after reconnect",
+                    _driverInstanceId, reg.SymbolName, device.Options.HostAddress);
+            }
+        }
+
+        _logger.LogInformation(
+            "TwinCAT driver '{DriverInstanceId}' re-registered {Replayed}/{Total} native notifications " +
+            "on device '{HostAddress}' after reconnect",
+            _driverInstanceId, replayed, intents.Length, device.Options.HostAddress);
+    }
+
+    /// <summary>
+    ///     Disposes + nulls a device's client under <see cref="DeviceState.ConnectGate"/> so the
+    ///     next <see cref="EnsureConnectedAsync"/> rebuilds it (and replays notifications). Called
+    ///     by the probe loop when a wire-level probe fails on a locally-"connected" client — the
+    ///     <see cref="ITwinCATClient.IsConnected"/> fast-path keys on the AMS-port state, not wire
+    ///     liveness, so without this forced recycle a wire-dead-but-port-open client would keep
+    ///     being reused and native pushes would never recover.
+    /// </summary>
+    private async Task RecycleClientAsync(DeviceState device)
+    {
+        await device.ConnectGate.WaitAsync(CancellationToken.None).ConfigureAwait(false);
+        try
+        {
+            if (device.Client is { } c)
+            {
+                c.OnSymbolVersionChanged -= HandleSymbolVersionChanged;
+                try { c.Dispose(); } catch { /* best-effort */ }
+                device.Client = null;
            }
-            device.Client = client;
-            return client;
        }
        finally
        {
@@ -722,6 +903,15 @@ public sealed class TwinCATDriver : IDriver, IReadable, IWritable, ITagDiscovery
        ///     create-or-dispose for this device.</summary>
        public SemaphoreSlim ConnectGate { get; } = new(1, 1);

+        /// <summary>
+        ///     Live native-notification registrations against this device, keyed by registration id.
+        ///     Holds the <b>replayable intent</b> (symbol / type / bit / interval / callback), not just
+        ///     an opaque handle, so that when <see cref="TwinCATDriver.EnsureConnectedAsync"/> swaps in
+        ///     a fresh client after a drop it can re-register every notification onto it (STAB-2).
+        ///     Populated by <c>SubscribeAsync</c>, drained by <c>UnsubscribeAsync</c> / teardown.
+        /// </summary>
+        public ConcurrentDictionary<long, NativeRegistration> NativeRegistrations { get; } = new();
+
        /// <summary>Gets the lock object for synchronizing host state transitions.</summary>
        public object ProbeLock { get; } = new();
        /// <summary>Gets or sets the current host connectivity state.</summary>
@@ -42,6 +42,11 @@ internal class FakeTwinCATClient : ITwinCATClient
    /// <summary>Test hook — fire the symbol-version-changed signal as the real client would.</summary>
    public void FireSymbolVersionChanged() => OnSymbolVersionChanged?.Invoke(this, EventArgs.Empty);

+    /// <summary>Test hook — simulate a wire-level drop that leaves the AMS port "connected" but dead:
+    ///     flips <see cref="IsConnected"/> to false without disposing, so the driver's IsConnected
+    ///     fast-path treats the client as stale and rebuilds (exercises the STAB-2 reconnect+replay).</summary>
+    public void SimulateWireDrop() => IsConnected = false;
+
    /// <summary>Simulates connecting to the TwinCAT system.</summary>
    /// <param name="address">The AMS address to connect to.</param>
    /// <param name="timeout">The connection timeout.</param>
@@ -0,0 +1,179 @@
+using System.Collections.Concurrent;
+using Shouldly;
+using Xunit;
+using ZB.MOM.WW.OtOpcUa.Core.Abstractions;
+using ZB.MOM.WW.OtOpcUa.Driver.TwinCAT;
+
+namespace ZB.MOM.WW.OtOpcUa.Driver.TwinCAT.Tests;
+
+/// <summary>
+///     Unit tests for the STAB-2 fix: native ADS notifications are re-registered onto a fresh
+///     client after a reconnect, instead of being silently orphaned. There is <b>no TwinCAT docker
+///     fixture</b> (the 13 integration tests need a real TC3 XAR target), so these fake-client tests
+///     are the <i>authoritative</i> automated coverage for the reconnect path.
+/// </summary>
+[Trait("Category", "Unit")]
+public sealed class TwinCATReconnectReplayTests
+{
+    private const string Host = "ads://5.23.91.23.1.1:851";
+
+    private static (TwinCATDriver drv, FakeTwinCATClientFactory factory) NewNativeDriver(
+        bool probe = false, TimeSpan? probeInterval = null, params TwinCATTagDefinition[] tags)
+    {
+        var factory = new FakeTwinCATClientFactory();
+        var drv = new TwinCATDriver(new TwinCATDriverOptions
+        {
+            Devices = [new TwinCATDeviceOptions(Host)],
+            Tags = tags,
+            Probe = new TwinCATProbeOptions
+            {
+                Enabled = probe,
+                Interval = probeInterval ?? TimeSpan.FromMilliseconds(30),
+                Timeout = TimeSpan.FromMilliseconds(100),
+            },
+            UseNativeNotifications = true,
+        }, "drv-1", factory);
+        return (drv, factory);
+    }
+
+    /// <summary>
+    ///     After a client drop, the next connection (driven here through a read) re-registers the
+    ///     native notification with the same symbol / type / interval, a push on the fresh client
+    ///     reaches OnDataChange, and the old handle is disposed.
+    /// </summary>
+    [Fact]
+    public async Task Reconnect_replays_native_notifications_onto_the_fresh_client()
+    {
+        var (drv, factory) = NewNativeDriver(
+            tags: new TwinCATTagDefinition("Speed", Host, "MAIN.Speed", TwinCATDataType.DInt));
+        await drv.InitializeAsync("{}", TestContext.Current.CancellationToken);
+
+        var events = new ConcurrentQueue<DataChangeEventArgs>();
+        drv.OnDataChange += (_, e) => events.Enqueue(e);
+
+        var interval = TimeSpan.FromMilliseconds(250);
+        _ = await drv.SubscribeAsync(["Speed"], interval, TestContext.Current.CancellationToken);
+
+        var client1 = factory.Clients[0];
+        client1.Notifications.Count.ShouldBe(1);
+        var oldHandle = client1.Notifications[0];
+
+        // Simulate a wire-level drop, then trigger a reconnect via a read.
+        client1.SimulateWireDrop();
+        _ = await drv.ReadAsync(["Speed"], TestContext.Current.CancellationToken);
+
+        // A brand-new client was built and the notification replayed onto it — same intent.
+        factory.Clients.Count.ShouldBe(2);
+        var client2 = factory.Clients[1];
+        client2.Notifications.Count.ShouldBe(1);
+        var replayed = client2.Notifications[0];
+        replayed.SymbolPath.ShouldBe("MAIN.Speed");
+        replayed.Type.ShouldBe(TwinCATDataType.DInt);
+
+        // The old handle was disposed; a push on the FRESH client reaches OnDataChange.
+        oldHandle.Disposed.ShouldBeTrue();
+        client2.FireNotification("MAIN.Speed", 7777);
+        events.Count.ShouldBe(1);
+        events.Last().Snapshot.Value.ShouldBe(7777);
+        events.Last().FullReference.ShouldBe("Speed");
+    }
+
+    /// <summary>Every subscribed tag on the device is replayed, not just the first.</summary>
+    [Fact]
+    public async Task Reconnect_replays_all_registered_tags()
+    {
+        var (drv, factory) = NewNativeDriver(
+            tags:
+            [
+                new TwinCATTagDefinition("A", Host, "MAIN.A", TwinCATDataType.DInt),
+                new TwinCATTagDefinition("B", Host, "MAIN.B", TwinCATDataType.Real),
+            ]);
+        await drv.InitializeAsync("{}", TestContext.Current.CancellationToken);
+        _ = await drv.SubscribeAsync(["A", "B"], TimeSpan.FromMilliseconds(100), TestContext.Current.CancellationToken);
+
+        factory.Clients[0].SimulateWireDrop();
+        _ = await drv.ReadAsync(["A"], TestContext.Current.CancellationToken);
+
+        factory.Clients.Count.ShouldBe(2);
+        factory.Clients[1].Notifications.Select(n => n.SymbolPath)
+            .ShouldBe(["MAIN.A", "MAIN.B"], ignoreOrder: true);
+    }
+
+    /// <summary>
+    ///     After a reconnect + replay, unsubscribing disposes the CURRENT (replayed) handle and the
+    ///     registration no longer replays on a subsequent reconnect.
+    /// </summary>
+    [Fact]
+    public async Task Unsubscribe_after_reconnect_disposes_the_current_handle_and_stops_replaying()
+    {
+        var (drv, factory) = NewNativeDriver(
+            tags: new TwinCATTagDefinition("X", Host, "MAIN.X", TwinCATDataType.DInt));
+        await drv.InitializeAsync("{}", TestContext.Current.CancellationToken);
+        var handle = await drv.SubscribeAsync(["X"], TimeSpan.FromMilliseconds(100), TestContext.Current.CancellationToken);
+
+        // First reconnect → replayed onto client2.
+        factory.Clients[0].SimulateWireDrop();
+        _ = await drv.ReadAsync(["X"], TestContext.Current.CancellationToken);
+        var client2 = factory.Clients[1];
+        client2.Notifications.Count.ShouldBe(1);
+
+        // Unsubscribe disposes the current (client2) handle.
+        await drv.UnsubscribeAsync(handle, TestContext.Current.CancellationToken);
+        client2.Notifications.ShouldBeEmpty();
+
+        // A second reconnect must NOT resurrect the unsubscribed notification.
+        client2.SimulateWireDrop();
+        _ = await drv.ReadAsync(["X"], TestContext.Current.CancellationToken);
+        factory.Clients.Count.ShouldBe(3);
+        factory.Clients[2].Notifications.ShouldBeEmpty();
+    }
+
+    /// <summary>
+    ///     A wire-level probe failure on a locally-"connected" client forces a recycle so the next
+    ///     tick rebuilds the client — the compounding half of STAB-2 (the IsConnected fast-path
+    ///     keys on port state, not wire liveness).
+    /// </summary>
+    [Fact]
+    public async Task Probe_failure_recycles_the_client_and_rebuilds()
+    {
+        // First client's wire probe fails; the recycle nulls it and the next tick builds a second
+        // client whose probe succeeds.
+        var buildCount = 0;
+        var factory = new FakeTwinCATClientFactory
+        {
+            Customise = () =>
+            {
+                buildCount++;
+                // Client #1 reports a failed wire probe; #2+ succeed.
+                return new FakeTwinCATClient { ProbeResult = buildCount > 1 };
+            },
+        };
+        var drv = new TwinCATDriver(new TwinCATDriverOptions
+        {
+            Devices = [new TwinCATDeviceOptions(Host)],
+            Tags = [new TwinCATTagDefinition("X", Host, "MAIN.X", TwinCATDataType.DInt)],
+            Probe = new TwinCATProbeOptions
+            {
+                Enabled = true,
+                Interval = TimeSpan.FromMilliseconds(30),
+                Timeout = TimeSpan.FromMilliseconds(100),
+            },
+            UseNativeNotifications = true,
+        }, "drv-probe", factory);
+        await drv.InitializeAsync("{}", TestContext.Current.CancellationToken);
+
+        // The probe loop should recycle the wire-dead client and rebuild a healthy one.
+        await WaitForAsync(() => factory.Clients.Count >= 2, TimeSpan.FromSeconds(3));
+        factory.Clients.Count.ShouldBeGreaterThanOrEqualTo(2);
+        factory.Clients[0].DisposeCount.ShouldBeGreaterThanOrEqualTo(1);
+
+        await drv.ShutdownAsync(TestContext.Current.CancellationToken);
+    }
+
+    private static async Task WaitForAsync(Func<bool> condition, TimeSpan timeout)
+    {
+        var deadline = DateTime.UtcNow + timeout;
+        while (!condition() && DateTime.UtcNow < deadline)
+            await Task.Delay(20);
+    }
+}