src/ZB.MOM.WW.OtOpcUa.Driver.AbCip/AbCipDriver.cs

@@ -20,7 +20,8 @@ namespace ZB.MOM.WW.OtOpcUa.Driver.AbCip;
 ///     from native-heap growth that the CLR allocator can't see; it tears down every
 ///     <see cref="PlcTagHandle"/> and reconnects each device.</para>
 /// </remarks>
-public sealed class AbCipDriver : IDriver, IReadable, IWritable, ITagDiscovery, ISubscribable, IDisposable, IAsyncDisposable
+public sealed class AbCipDriver : IDriver, IReadable, IWritable, ITagDiscovery, ISubscribable,
+    IHostConnectivityProbe, IPerCallHostResolver, IDisposable, IAsyncDisposable
 {
     private readonly AbCipDriverOptions _options;
     private readonly string _driverInstanceId;
@@ -33,6 +34,7 @@ public sealed class AbCipDriver : IDriver, IReadable, IWritable, ITagDiscovery,
     private DriverHealth _health = new(DriverState.Unknown, null, null);
 
     public event EventHandler<DataChangeEventArgs>? OnDataChange;
+    public event EventHandler<HostStatusChangedEventArgs>? OnHostStatusChanged;
 
     public AbCipDriver(AbCipDriverOptions options, string driverInstanceId,
         IAbCipTagFactory? tagFactory = null,
@@ -70,9 +72,6 @@ public sealed class AbCipDriver : IDriver, IReadable, IWritable, ITagDiscovery,
             }
             foreach (var tag in _options.Tags)
             {
-                // UDT tags with declared Members fan out into synthetic member-tag entries addressable
-                // by composed full-reference. Parent structure tag also stored so discovery can emit a
-                // folder for it.
                 _tagsByName[tag.Name] = tag;
                 if (tag.DataType == AbCipDataType.Structure && tag.Members is { Count: > 0 })
                 {
@@ -89,6 +88,17 @@ public sealed class AbCipDriver : IDriver, IReadable, IWritable, ITagDiscovery,
                     }
                 }
             }
+
+            // Probe loops — one per device when enabled + a ProbeTagPath is configured.
+            if (_options.Probe.Enabled && !string.IsNullOrWhiteSpace(_options.Probe.ProbeTagPath))
+            {
+                foreach (var state in _devices.Values)
+                {
+                    state.ProbeCts = new CancellationTokenSource();
+                    var ct = state.ProbeCts.Token;
+                    _ = Task.Run(() => ProbeLoopAsync(state, ct), ct);
+                }
+            }
             _health = new DriverHealth(DriverState.Healthy, DateTime.UtcNow, null);
         }
         catch (Exception ex)
@@ -109,7 +119,12 @@ public sealed class AbCipDriver : IDriver, IReadable, IWritable, ITagDiscovery,
     {
         await _poll.DisposeAsync().ConfigureAwait(false);
         foreach (var state in _devices.Values)
+        {
+            try { state.ProbeCts?.Cancel(); } catch { }
+            state.ProbeCts?.Dispose();
+            state.ProbeCts = null;
             state.DisposeHandles();
+        }
         _devices.Clear();
         _tagsByName.Clear();
         _health = new DriverHealth(DriverState.Unknown, _health.LastSuccessfulRead, null);
@@ -127,6 +142,90 @@ public sealed class AbCipDriver : IDriver, IReadable, IWritable, ITagDiscovery,
         return Task.CompletedTask;
     }
 
+    // ---- IHostConnectivityProbe ----
+
+    public IReadOnlyList<HostConnectivityStatus> GetHostStatuses() =>
+        [.. _devices.Values.Select(s => new HostConnectivityStatus(s.Options.HostAddress, s.HostState, s.HostStateChangedUtc))];
+
+    private async Task ProbeLoopAsync(DeviceState state, CancellationToken ct)
+    {
+        var probeParams = new AbCipTagCreateParams(
+            Gateway: state.ParsedAddress.Gateway,
+            Port: state.ParsedAddress.Port,
+            CipPath: state.ParsedAddress.CipPath,
+            LibplctagPlcAttribute: state.Profile.LibplctagPlcAttribute,
+            TagName: _options.Probe.ProbeTagPath!,
+            Timeout: _options.Probe.Timeout);
+
+        IAbCipTagRuntime? probeRuntime = null;
+        while (!ct.IsCancellationRequested)
+        {
+            var success = false;
+            try
+            {
+                probeRuntime ??= _tagFactory.Create(probeParams);
+                // Lazy-init on first attempt; re-init after a transport failure has caused the
+                // native handle to be destroyed.
+                if (!state.ProbeInitialized)
+                {
+                    await probeRuntime.InitializeAsync(ct).ConfigureAwait(false);
+                    state.ProbeInitialized = true;
+                }
+                await probeRuntime.ReadAsync(ct).ConfigureAwait(false);
+                success = probeRuntime.GetStatus() == 0;
+            }
+            catch (OperationCanceledException) when (ct.IsCancellationRequested)
+            {
+                break;
+            }
+            catch
+            {
+                // Wire / init error — tear down the probe runtime so the next tick re-creates it.
+                try { probeRuntime?.Dispose(); } catch { }
+                probeRuntime = null;
+                state.ProbeInitialized = false;
+            }
+
+            TransitionDeviceState(state, success ? HostState.Running : HostState.Stopped);
+
+            try { await Task.Delay(_options.Probe.Interval, ct).ConfigureAwait(false); }
+            catch (OperationCanceledException) { break; }
+        }
+
+        try { probeRuntime?.Dispose(); } catch { }
+    }
+
+    private void TransitionDeviceState(DeviceState state, HostState newState)
+    {
+        HostState old;
+        lock (state.ProbeLock)
+        {
+            old = state.HostState;
+            if (old == newState) return;
+            state.HostState = newState;
+            state.HostStateChangedUtc = DateTime.UtcNow;
+        }
+        OnHostStatusChanged?.Invoke(this,
+            new HostStatusChangedEventArgs(state.Options.HostAddress, old, newState));
+    }
+
+    // ---- IPerCallHostResolver ----
+
+    /// <summary>
+    ///     Resolve the device host address for a given tag full-reference. Per plan decision #144
+    ///     the Phase 6.1 resilience pipeline keys its bulkhead + breaker on
+    ///     <c>(DriverInstanceId, hostName)</c> so multi-PLC drivers get per-device isolation —
+    ///     one dead PLC trips only its own breaker. Unknown references fall back to the
+    ///     first configured device's host address rather than throwing — the invoker handles the
+    ///     mislookup at the capability level when the actual read returns BadNodeIdUnknown.
+    /// </summary>
+    public string ResolveHost(string fullReference)
+    {
+        if (_tagsByName.TryGetValue(fullReference, out var def))
+            return def.DeviceHostAddress;
+        return _options.Devices.FirstOrDefault()?.HostAddress ?? DriverInstanceId;
+    }
+
     // ---- IReadable ----
 
     /// <summary>
@@ -457,6 +556,12 @@ public sealed class AbCipDriver : IDriver, IReadable, IWritable, ITagDiscovery,
         public AbCipDeviceOptions Options { get; } = options;
         public AbCipPlcFamilyProfile Profile { get; } = profile;
 
+        public object ProbeLock { get; } = new();
+        public HostState HostState { get; set; } = HostState.Unknown;
+        public DateTime HostStateChangedUtc { get; set; } = DateTime.UtcNow;
+        public CancellationTokenSource? ProbeCts { get; set; }
+        public bool ProbeInitialized { get; set; }
+
         public Dictionary<string, PlcTagHandle> TagHandles { get; } =
             new(StringComparer.OrdinalIgnoreCase);
 

tests/ZB.MOM.WW.OtOpcUa.Driver.AbCip.Tests/AbCipHostProbeTests.cs

@@ -0,0 +1,227 @@
+using System.Collections.Concurrent;
+using Shouldly;
+using Xunit;
+using ZB.MOM.WW.OtOpcUa.Core.Abstractions;
+using ZB.MOM.WW.OtOpcUa.Driver.AbCip;
+
+namespace ZB.MOM.WW.OtOpcUa.Driver.AbCip.Tests;
+
+[Trait("Category", "Unit")]
+public sealed class AbCipHostProbeTests
+{
+    [Fact]
+    public async Task GetHostStatuses_returns_one_entry_per_device()
+    {
+        var drv = new AbCipDriver(new AbCipDriverOptions
+        {
+            Devices =
+            [
+                new AbCipDeviceOptions("ab://10.0.0.5/1,0"),
+                new AbCipDeviceOptions("ab://10.0.0.6/1,0"),
+            ],
+            Probe = new AbCipProbeOptions { Enabled = false },
+        }, "drv-1");
+        await drv.InitializeAsync("{}", CancellationToken.None);
+
+        var statuses = drv.GetHostStatuses();
+        statuses.Count.ShouldBe(2);
+        statuses.Select(s => s.HostName).ShouldBe(["ab://10.0.0.5/1,0", "ab://10.0.0.6/1,0"], ignoreOrder: true);
+        statuses.ShouldAllBe(s => s.State == HostState.Unknown);
+    }
+
+    [Fact]
+    public async Task Probe_with_successful_read_transitions_to_Running()
+    {
+        var factory = new FakeAbCipTagFactory { Customise = p => new FakeAbCipTag(p) { Status = 0 } };
+        var transitions = new ConcurrentQueue<HostStatusChangedEventArgs>();
+        var drv = new AbCipDriver(new AbCipDriverOptions
+        {
+            Devices = [new AbCipDeviceOptions("ab://10.0.0.5/1,0")],
+            Probe = new AbCipProbeOptions
+            {
+                Enabled = true,
+                Interval = TimeSpan.FromMilliseconds(100),
+                Timeout = TimeSpan.FromMilliseconds(50),
+                ProbeTagPath = "@raw_cpu_type",
+            },
+        }, "drv-1", factory);
+        drv.OnHostStatusChanged += (_, e) => transitions.Enqueue(e);
+
+        await drv.InitializeAsync("{}", CancellationToken.None);
+        await WaitForAsync(() => transitions.Any(t => t.NewState == HostState.Running), TimeSpan.FromSeconds(2));
+
+        transitions.Select(t => t.NewState).ShouldContain(HostState.Running);
+        drv.GetHostStatuses().Single().State.ShouldBe(HostState.Running);
+        await drv.ShutdownAsync(CancellationToken.None);
+    }
+
+    [Fact]
+    public async Task Probe_with_read_failure_transitions_to_Stopped()
+    {
+        var factory = new FakeAbCipTagFactory
+        {
+            Customise = p => new FakeAbCipTag(p) { ThrowOnRead = true },
+        };
+        var transitions = new ConcurrentQueue<HostStatusChangedEventArgs>();
+        var drv = new AbCipDriver(new AbCipDriverOptions
+        {
+            Devices = [new AbCipDeviceOptions("ab://10.0.0.5/1,0")],
+            Probe = new AbCipProbeOptions
+            {
+                Enabled = true,
+                Interval = TimeSpan.FromMilliseconds(100),
+                Timeout = TimeSpan.FromMilliseconds(50),
+                ProbeTagPath = "@raw_cpu_type",
+            },
+        }, "drv-1", factory);
+        drv.OnHostStatusChanged += (_, e) => transitions.Enqueue(e);
+
+        await drv.InitializeAsync("{}", CancellationToken.None);
+        await WaitForAsync(() => transitions.Any(t => t.NewState == HostState.Stopped), TimeSpan.FromSeconds(2));
+
+        drv.GetHostStatuses().Single().State.ShouldBe(HostState.Stopped);
+        await drv.ShutdownAsync(CancellationToken.None);
+    }
+
+    [Fact]
+    public async Task Probe_disabled_when_Enabled_is_false()
+    {
+        var factory = new FakeAbCipTagFactory();
+        var transitions = new ConcurrentQueue<HostStatusChangedEventArgs>();
+        var drv = new AbCipDriver(new AbCipDriverOptions
+        {
+            Devices = [new AbCipDeviceOptions("ab://10.0.0.5/1,0")],
+            Probe = new AbCipProbeOptions { Enabled = false, ProbeTagPath = "@raw_cpu_type" },
+        }, "drv-1", factory);
+        drv.OnHostStatusChanged += (_, e) => transitions.Enqueue(e);
+
+        await drv.InitializeAsync("{}", CancellationToken.None);
+        await Task.Delay(300);
+
+        transitions.ShouldBeEmpty();
+        drv.GetHostStatuses().Single().State.ShouldBe(HostState.Unknown);
+        await drv.ShutdownAsync(CancellationToken.None);
+    }
+
+    [Fact]
+    public async Task Probe_skipped_when_ProbeTagPath_is_null()
+    {
+        var drv = new AbCipDriver(new AbCipDriverOptions
+        {
+            Devices = [new AbCipDeviceOptions("ab://10.0.0.5/1,0")],
+            Probe = new AbCipProbeOptions { Enabled = true, ProbeTagPath = null },
+        }, "drv-1");
+
+        await drv.InitializeAsync("{}", CancellationToken.None);
+        await Task.Delay(200);
+
+        drv.GetHostStatuses().Single().State.ShouldBe(HostState.Unknown);
+        await drv.ShutdownAsync(CancellationToken.None);
+    }
+
+    [Fact]
+    public async Task Probe_loops_across_multiple_devices_independently()
+    {
+        var factory = new FakeAbCipTagFactory
+        {
+            // Device A returns ok, Device B throws on read.
+            Customise = p => p.Gateway == "10.0.0.5"
+                ? new FakeAbCipTag(p)
+                : new FakeAbCipTag(p) { ThrowOnRead = true },
+        };
+        var transitions = new ConcurrentQueue<HostStatusChangedEventArgs>();
+        var drv = new AbCipDriver(new AbCipDriverOptions
+        {
+            Devices =
+            [
+                new AbCipDeviceOptions("ab://10.0.0.5/1,0"),
+                new AbCipDeviceOptions("ab://10.0.0.6/1,0"),
+            ],
+            Probe = new AbCipProbeOptions
+            {
+                Enabled = true, Interval = TimeSpan.FromMilliseconds(100),
+                Timeout = TimeSpan.FromMilliseconds(50), ProbeTagPath = "@raw_cpu_type",
+            },
+        }, "drv-1", factory);
+        drv.OnHostStatusChanged += (_, e) => transitions.Enqueue(e);
+
+        await drv.InitializeAsync("{}", CancellationToken.None);
+        await WaitForAsync(() => transitions.Count >= 2, TimeSpan.FromSeconds(3));
+
+        transitions.ShouldContain(t => t.HostName == "ab://10.0.0.5/1,0" && t.NewState == HostState.Running);
+        transitions.ShouldContain(t => t.HostName == "ab://10.0.0.6/1,0" && t.NewState == HostState.Stopped);
+        await drv.ShutdownAsync(CancellationToken.None);
+    }
+
+    // ---- IPerCallHostResolver ----
+
+    [Fact]
+    public async Task ResolveHost_returns_declared_device_for_known_tag()
+    {
+        var drv = new AbCipDriver(new AbCipDriverOptions
+        {
+            Devices =
+            [
+                new AbCipDeviceOptions("ab://10.0.0.5/1,0"),
+                new AbCipDeviceOptions("ab://10.0.0.6/1,0"),
+            ],
+            Tags =
+            [
+                new AbCipTagDefinition("A", "ab://10.0.0.5/1,0", "A", AbCipDataType.DInt),
+                new AbCipTagDefinition("B", "ab://10.0.0.6/1,0", "B", AbCipDataType.DInt),
+            ],
+            Probe = new AbCipProbeOptions { Enabled = false },
+        }, "drv-1");
+        await drv.InitializeAsync("{}", CancellationToken.None);
+
+        drv.ResolveHost("A").ShouldBe("ab://10.0.0.5/1,0");
+        drv.ResolveHost("B").ShouldBe("ab://10.0.0.6/1,0");
+    }
+
+    [Fact]
+    public async Task ResolveHost_falls_back_to_first_device_for_unknown_reference()
+    {
+        var drv = new AbCipDriver(new AbCipDriverOptions
+        {
+            Devices = [new AbCipDeviceOptions("ab://10.0.0.5/1,0")],
+            Probe = new AbCipProbeOptions { Enabled = false },
+        }, "drv-1");
+        await drv.InitializeAsync("{}", CancellationToken.None);
+
+        drv.ResolveHost("does-not-exist").ShouldBe("ab://10.0.0.5/1,0");
+    }
+
+    [Fact]
+    public async Task ResolveHost_falls_back_to_DriverInstanceId_when_no_devices()
+    {
+        var drv = new AbCipDriver(new AbCipDriverOptions(), "drv-1");
+        await drv.InitializeAsync("{}", CancellationToken.None);
+
+        drv.ResolveHost("anything").ShouldBe("drv-1");
+    }
+
+    [Fact]
+    public async Task ResolveHost_for_UDT_member_walks_to_synthesised_definition()
+    {
+        var drv = new AbCipDriver(new AbCipDriverOptions
+        {
+            Devices = [new AbCipDeviceOptions("ab://10.0.0.7/1,0")],
+            Tags =
+            [
+                new AbCipTagDefinition("Motor1", "ab://10.0.0.7/1,0", "Motor1", AbCipDataType.Structure,
+                    Members: [new AbCipStructureMember("Speed", AbCipDataType.DInt)]),
+            ],
+            Probe = new AbCipProbeOptions { Enabled = false },
+        }, "drv-1");
+        await drv.InitializeAsync("{}", CancellationToken.None);
+
+        drv.ResolveHost("Motor1.Speed").ShouldBe("ab://10.0.0.7/1,0");
+    }
+
+    private static async Task WaitForAsync(Func<bool> condition, TimeSpan timeout)
+    {
+        var deadline = DateTime.UtcNow + timeout;
+        while (!condition() && DateTime.UtcNow < deadline)
+            await Task.Delay(20);
+    }
+}