fix(galaxy): wire ScanState probing, single-sweep discovery, liveness stamp, dispatch-loop fault

Review 2026-10-21 G-4, G-7, G-11, G-10. - G-4: PerPlatformProbeWatcher publishes its probe handles into the driver's SubscriptionRegistry under the reserved ProbeSubscriptionId (-1), so the EventPump's fan-out finally reaches the watcher — every ScanState event used to be dropped as "stale after unsubscribe" and per-platform host status never updated. ResubscribeAllAsync delegates the reserved entry to the watcher's new ResubscribeAsync, so the probe survives a reconnect; StopAsync removes the entry. Probe traffic is routed by the reserved handle and never reaches the public OnDataChange. The watcher is also built on the seam path (subscriber only), which is what makes the wiring testable without a gateway. - G-7: GalaxyDiscoverer.DiscoverAsync returns the hierarchy it fetched and the driver selects its $WinPlatform/$AppEngine probe targets from that same sweep instead of asking the gateway for the whole galaxy twice. - G-11: OnPumpDataChange stamps LastSuccessfulRead from inbound stream data, throttled (1 s default), so the comms-loss watchdog has honest liveness evidence under a Degraded supervisor. - G-10: an EventPump dispatch-loop exit that is not a cancellation now reports a stream fault so the supervisor recreates the pump, instead of leaving a deaf pump reporting Healthy. Refs #735 Claude-Session: https://claude.ai/code/session_01Ub9hGmYLQvn9XETzw7CMsT
2026-10-22 06:35:48 -04:00
parent 2d25c916b6
commit 1b3169273d
8 changed files with 614 additions and 42 deletions
@@ -61,8 +61,12 @@ public sealed class GalaxyDiscoverer
    /// </summary>
    /// <param name="builder">The address space builder to populate with discovery results.</param>
    /// <param name="cancellationToken">The cancellation token for the operation.</param>
-    /// <returns>A task that represents the asynchronous operation.</returns>
-    public async Task DiscoverAsync(IAddressSpaceBuilder builder, CancellationToken cancellationToken)
+    /// <returns>
+    ///     The hierarchy this pass fetched, so a caller that needs the object set for something
+    ///     else (the driver selects its <c>$WinPlatform</c> / <c>$AppEngine</c> probe targets from
+    ///     it) reuses the one sweep instead of asking the gateway for the full galaxy twice.
+    /// </returns>
+    public async Task<IReadOnlyList<GalaxyObject>> DiscoverAsync(IAddressSpaceBuilder builder, CancellationToken cancellationToken)
    {
        ArgumentNullException.ThrowIfNull(builder);
        var objects = await _source.GetHierarchyAsync(cancellationToken).ConfigureAwait(false);
@@ -132,6 +136,8 @@ public sealed class GalaxyDiscoverer
                }
            }
        }
+
+        return objects;
    }

    // Resolve (and memoise) the folder for a gobject, recursively creating its parent's folder
@@ -454,8 +454,11 @@ public sealed class GalaxyDriver
        }
    }

+    // The watcher publishes its probe handles into the driver's registry (reserved id) so the
+    // EventPump's fan-out reaches OnPumpDataChange for ScanState events — without that every
+    // probe event was dropped before it got anywhere near the watcher.
    private PerPlatformProbeWatcher BuildProbeWatcher(IGalaxySubscriber subscriber) =>
-        new(subscriber, _hostStatuses, _logger,
+        new(subscriber, _hostStatuses, _subscriptions, _logger,
            bufferedUpdateIntervalMs: _options.MxAccess.PublishingIntervalMs);

    /// <summary>
@@ -738,6 +741,15 @@ public sealed class GalaxyDriver
        var tagCount = 0;
        foreach (var (subscriptionId, oldBindings) in entries)
        {
+            if (subscriptionId == SubscriptionRegistry.ProbeSubscriptionId && _probeWatcher is { } probeWatcher)
+            {
+                // The probe watcher owns its handles: it re-subscribes its platforms, records the
+                // fresh handles and rebinds the reserved registry entry itself, so ScanState host
+                // status keeps updating after a reconnect (it used to go permanently deaf).
+                tagCount += await probeWatcher.ResubscribeAsync(cancellationToken).ConfigureAwait(false);
+                continue;
+            }
+
            var refs = oldBindings
                .Select(b => b.FullReference)
                .Distinct(StringComparer.OrdinalIgnoreCase)
@@ -1058,11 +1070,12 @@ public sealed class GalaxyDriver
        // keyed by RawPath, matching what WriteAsync resolves against.
        var discoverer = new GalaxyDiscoverer(
            source, attributeRef => _rawTagByAttributeRef.GetValueOrDefault(attributeRef));
-        await discoverer.DiscoverAsync(capturingBuilder, cancellationToken).ConfigureAwait(false);
+        // One sweep: the discoverer hands back the hierarchy it materialised, and the platform
+        // selection below reads that instead of asking the gateway for the whole galaxy again.
+        var hierarchy = await discoverer.DiscoverAsync(capturingBuilder, cancellationToken).ConfigureAwait(false);

        if (_probeWatcher is not null)
        {
-            var hierarchy = await source.GetHierarchyAsync(cancellationToken).ConfigureAwait(false);
            var platforms = hierarchy
                .Where(o => o.TemplateChain.Any(t =>
                    string.Equals(t, "$WinPlatform", StringComparison.OrdinalIgnoreCase)
@@ -1744,6 +1757,24 @@ public sealed class GalaxyDriver
    /// </summary>
    private void OnPumpDataChange(object? sender, DataChangeEventArgs args)
    {
+        // Anything the gateway pushes on the stream proves it is alive and serving this session —
+        // the comms-loss watchdog's evidence (G-11: the stamp used to move only at init and on a
+        // supervisor Healthy transition, so a 90 s Degraded tripped a re-init under flowing data).
+        StampLiveness();
+
+        // Probe traffic rides the reserved registry entry and stays in the attributeRef domain —
+        // ScanState attributes are discovered platform attributes, never authored RawPaths. It is
+        // the watcher's alone: the host never subscribed it and must not see it on OnDataChange.
+        if (args.SubscriptionHandle is GalaxySubscriptionHandle { SubscriptionId: SubscriptionRegistry.ProbeSubscriptionId })
+        {
+            // The probe decoder takes a raw quality byte. Recover it via the canonical
+            // StatusCodeMap.ToQualityCategoryByte helper so the mapping lives in one
+            // place next to its inverse (FromQualityByte) and cannot desync silently.
+            var qualityByte = StatusCodeMap.ToQualityCategoryByte(args.Snapshot.StatusCode);
+            _probeWatcher?.OnProbeValueChanged(args.FullReference, args.Snapshot.Value, qualityByte);
+            return;
+        }
+
        // v3: the pump fans out keyed by the dialled attributeRef; the server correlates values to nodes
        // by RawPath (the node's FullName). Reverse-map before surfacing to public OnDataChange consumers.
        // Identity on a miss, so a live-browse tag surfaces under its attributeRef unchanged.
@@ -1754,18 +1785,25 @@ public sealed class GalaxyDriver
            publicArgs = args with { FullReference = rawPath };
        }
        OnDataChange?.Invoke(this, publicArgs);
+    }

-        // Probe watching stays in the attributeRef domain — ScanState attributes are discovered platform
-        // attributes, never authored RawPaths, so use the untranslated reference here.
-        if (_probeWatcher is not null
-            && args.FullReference.EndsWith(PerPlatformProbeWatcher.ProbeSuffix, StringComparison.OrdinalIgnoreCase))
-        {
-            // The probe decoder takes a raw quality byte. Recover it via the canonical
-            // StatusCodeMap.ToQualityCategoryByte helper so the mapping lives in one
-            // place next to its inverse (FromQualityByte) and cannot desync silently.
-            var qualityByte = StatusCodeMap.ToQualityCategoryByte(args.Snapshot.StatusCode);
-            _probeWatcher.OnProbeValueChanged(args.FullReference, args.Snapshot.Value, qualityByte);
-        }
+    /// <summary>Minimum gap between two <c>LastSuccessfulRead</c> stamps from stream data. Internal so tests can widen it.</summary>
+    internal TimeSpan LivenessStampThrottle { get; set; } = TimeSpan.FromSeconds(1);
+
+    private long _lastLivenessStampTicks;
+
+    /// <summary>
+    ///     Advance <c>LastSuccessfulRead</c> from inbound stream data, at most once per
+    ///     <see cref="LivenessStampThrottle"/> — this runs on the pump's dispatch loop for every
+    ///     event, so a 50k-tag galaxy must not pay a health-record allocation per value.
+    /// </summary>
+    private void StampLiveness()
+    {
+        var now = DateTime.UtcNow;
+        var last = Volatile.Read(ref _lastLivenessStampTicks);
+        if (now.Ticks - last < LivenessStampThrottle.Ticks) return;
+        Volatile.Write(ref _lastLivenessStampTicks, now.Ticks);
+        _health = _health with { LastSuccessfulRead = now };
    }

    /// <summary>
@@ -38,6 +38,12 @@ public sealed class PerPlatformProbeWatcher : IDisposable
    private readonly ILogger _logger;
    private readonly int _bufferedUpdateIntervalMs;

+    // The driver's fan-out registry. The EventPump resolves every inbound event through it, so
+    // the probe handles must be registered there (under SubscriptionRegistry.ProbeSubscriptionId)
+    // or every ScanState event is dropped as "stale after unsubscribe". Null when the watcher is
+    // driven directly (unit tests of the decoding table).
+    private readonly SubscriptionRegistry? _registry;
+
    // Tracked platform → gw item handle. Item handle 0 means the gw rejected the subscribe;
    // we keep the entry so SyncPlatformsAsync doesn't try to subscribe it again on every call.
    private readonly ConcurrentDictionary<string, int> _itemHandlesByPlatform =
@@ -55,9 +61,30 @@ public sealed class PerPlatformProbeWatcher : IDisposable
        HostStatusAggregator aggregator,
        ILogger? logger = null,
        int bufferedUpdateIntervalMs = 0)
+        : this(subscriber, aggregator, registry: null, logger, bufferedUpdateIntervalMs)
+    {
+    }
+
+    /// <summary>
+    ///     Driver-side ctor: the watcher publishes its probe handles into <paramref name="registry"/>
+    ///     under <see cref="SubscriptionRegistry.ProbeSubscriptionId"/> so the <see cref="EventPump"/>
+    ///     fans ScanState events out to <see cref="OnProbeValueChanged"/>.
+    /// </summary>
+    /// <param name="subscriber">The Galaxy subscriber for managing probe subscriptions.</param>
+    /// <param name="aggregator">The host status aggregator for tracking platform connectivity.</param>
+    /// <param name="registry">The driver's subscription registry; null to keep the handles private.</param>
+    /// <param name="logger">Optional logger for diagnostic messages.</param>
+    /// <param name="bufferedUpdateIntervalMs">Buffered update interval in milliseconds; must be >= 0.</param>
+    internal PerPlatformProbeWatcher(
+        IGalaxySubscriber subscriber,
+        HostStatusAggregator aggregator,
+        SubscriptionRegistry? registry,
+        ILogger? logger = null,
+        int bufferedUpdateIntervalMs = 0)
    {
        _subscriber = subscriber ?? throw new ArgumentNullException(nameof(subscriber));
        _aggregator = aggregator ?? throw new ArgumentNullException(nameof(aggregator));
+        _registry = registry;
        _logger = logger ?? NullLogger.Instance;
        if (bufferedUpdateIntervalMs < 0)
        {
@@ -67,6 +94,68 @@ public sealed class PerPlatformProbeWatcher : IDisposable
        _bufferedUpdateIntervalMs = bufferedUpdateIntervalMs;
    }

+    /// <summary>Current platform → probe-handle bindings, as the registry sees them. Test seam.</summary>
+    internal IReadOnlyList<TagBinding> CurrentBindings => BuildBindings();
+
+    private IReadOnlyList<TagBinding> BuildBindings() =>
+        [.. _itemHandlesByPlatform.Select(kvp => new TagBinding(kvp.Key + ProbeSuffix, kvp.Value))];
+
+    /// <summary>Mirror the tracked handle set into the driver registry (no-op without one).</summary>
+    private void PublishBindings()
+    {
+        if (_registry is null) return;
+        var bindings = BuildBindings();
+        if (bindings.Count == 0) _registry.Remove(SubscriptionRegistry.ProbeSubscriptionId);
+        else _registry.Upsert(SubscriptionRegistry.ProbeSubscriptionId, bindings);
+    }
+
+    /// <summary>
+    ///     Re-issue SubscribeBulk for every tracked platform and replace the handles — the
+    ///     reconnect replay path, where the gateway session was recreated and every pre-reconnect
+    ///     handle is dead. Platforms the gateway now rejects keep a 0 handle (no events expected).
+    /// </summary>
+    /// <param name="cancellationToken">A cancellation token to cancel the operation.</param>
+    /// <returns>The number of probe references re-subscribed.</returns>
+    public async Task<int> ResubscribeAsync(CancellationToken cancellationToken)
+    {
+        ObjectDisposedException.ThrowIf(_disposed, this);
+        var platforms = _itemHandlesByPlatform.Keys.ToArray();
+        if (platforms.Length == 0) return 0;
+
+        await SubscribeAndRecordAsync(platforms, cancellationToken).ConfigureAwait(false);
+        return platforms.Length;
+    }
+
+    private async Task SubscribeAndRecordAsync(IReadOnlyList<string> platforms, CancellationToken cancellationToken)
+    {
+        var probeAddresses = platforms.Select(p => p + ProbeSuffix).ToArray();
+        // PR 6.3 — use the configured bufferedUpdateIntervalMs (defaults to 0 = gw cadence
+        // when the driver hasn't overridden MxAccess.PublishingIntervalMs). Probe ScanState
+        // changes are rare so a coarser interval is usually fine; deployments that need
+        // tighter health visibility can dial it down through GalaxyDriverOptions.
+        var results = await _subscriber.SubscribeBulkAsync(
+            probeAddresses, _bufferedUpdateIntervalMs, cancellationToken).ConfigureAwait(false);
+
+        for (var i = 0; i < platforms.Count; i++)
+        {
+            var platform = platforms[i];
+            var match = results.FirstOrDefault(r => string.Equals(
+                r.TagAddress, probeAddresses[i], StringComparison.OrdinalIgnoreCase));
+
+            var itemHandle = match is { WasSuccessful: true } ? match.ItemHandle : 0;
+            _itemHandlesByPlatform[platform] = itemHandle;
+
+            if (itemHandle <= 0)
+            {
+                _logger.LogWarning(
+                    "PerPlatformProbeWatcher subscribe failed for {Platform}: {Error}",
+                    platform, match?.ErrorMessage ?? "<no result returned>");
+            }
+        }
+
+        PublishBindings();
+    }
+
    /// <summary>Snapshot of platform tag names currently watched.</summary>
    public IReadOnlyCollection<string> WatchedPlatforms => [.. _itemHandlesByPlatform.Keys];

@@ -109,6 +198,8 @@ public sealed class PerPlatformProbeWatcher : IDisposable

        if (toRemove.Count > 0)
        {
+            // Drop the dead handles from the fan-out map before the (slow, fallible) unsubscribe.
+            PublishBindings();
            try
            {
                await _subscriber.UnsubscribeBulkAsync(
@@ -124,30 +215,7 @@ public sealed class PerPlatformProbeWatcher : IDisposable

        if (toAdd.Count == 0) return;

-        var probeAddresses = toAdd.Select(p => p + ProbeSuffix).ToArray();
-        // PR 6.3 — use the configured bufferedUpdateIntervalMs (defaults to 0 = gw cadence
-        // when the driver hasn't overridden MxAccess.PublishingIntervalMs). Probe ScanState
-        // changes are rare so a coarser interval is usually fine; deployments that need
-        // tighter health visibility can dial it down through GalaxyDriverOptions.
-        var results = await _subscriber.SubscribeBulkAsync(
-            probeAddresses, _bufferedUpdateIntervalMs, cancellationToken).ConfigureAwait(false);
-
-        for (var i = 0; i < toAdd.Count; i++)
-        {
-            var platform = toAdd[i];
-            var match = results.FirstOrDefault(r => string.Equals(
-                r.TagAddress, probeAddresses[i], StringComparison.OrdinalIgnoreCase));
-
-            var itemHandle = match is { WasSuccessful: true } ? match.ItemHandle : 0;
-            _itemHandlesByPlatform[platform] = itemHandle;
-
-            if (itemHandle <= 0)
-            {
-                _logger.LogWarning(
-                    "PerPlatformProbeWatcher subscribe failed for {Platform}: {Error}",
-                    platform, match?.ErrorMessage ?? "<no result returned>");
-            }
-        }
+        await SubscribeAndRecordAsync(toAdd, cancellationToken).ConfigureAwait(false);
    }

    /// <summary>
@@ -211,6 +279,7 @@ public sealed class PerPlatformProbeWatcher : IDisposable

        var liveHandles = _itemHandlesByPlatform.Values.Where(h => h > 0).ToArray();
        _itemHandlesByPlatform.Clear();
+        _registry?.Remove(SubscriptionRegistry.ProbeSubscriptionId);
        if (liveHandles.Length > 0)
        {
            try { await _subscriber.UnsubscribeBulkAsync(liveHandles, cancellationToken).ConfigureAwait(false); }
@@ -285,14 +285,19 @@ internal sealed class EventPump : IAsyncDisposable
                EventsDispatched.Add(1, _clientTag);
            }
        }
-        catch (OperationCanceledException) when (ct.IsCancellationRequested)
+        catch (Exception) when (ct.IsCancellationRequested)
        {
            // Clean shutdown.
        }
        catch (Exception ex)
        {
+            // The read loop is still draining the stream into a channel nobody reads: the pump is
+            // as dead as after a transport fault, but used to raise nothing — health stayed
+            // Healthy and the comms-loss watchdog never tripped (a clean termination evading the
+            // watchdog). Report it as a stream fault so the supervisor recreates the pump.
            _logger.LogWarning(ex,
-                "Galaxy EventPump dispatch loop ended with an exception — events past this point will be lost until restart.");
+                "Galaxy EventPump dispatch loop ended with an exception — notifying reconnect supervisor so the pump is recreated.");
+            NotifyStreamFault(ex);
        }
    }

@@ -18,10 +18,39 @@ namespace ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Runtime;
 /// </remarks>
 internal sealed class SubscriptionRegistry
 {
+    /// <summary>
+    ///     Reserved subscription id for the per-platform <c>ScanState</c> probe handles
+    ///     (<c>PerPlatformProbeWatcher</c>). Driver subscription ids are positive and
+    ///     monotonic (<see cref="NextSubscriptionId"/>), so a negative id can never collide.
+    ///     The probe handles used to live only in the watcher's own map, so the
+    ///     <see cref="EventPump"/> — which resolves subscribers through THIS registry —
+    ///     dropped every probe event as "stale after unsubscribe" and per-platform host
+    ///     status never updated.
+    /// </summary>
+    public const long ProbeSubscriptionId = -1;
+
    private readonly ConcurrentDictionary<long, SubscriptionEntry> _bySubscriptionId = new();
    private readonly ConcurrentDictionary<int, ImmutableHashSet<long>> _subscribersByItemHandle = new();
    private long _nextSubscriptionId;

+    /// <summary>True when <paramref name="subscriptionId"/> is currently registered.</summary>
+    /// <param name="subscriptionId">The subscription identifier.</param>
+    /// <returns>Whether the id is tracked.</returns>
+    public bool Contains(long subscriptionId) => _bySubscriptionId.ContainsKey(subscriptionId);
+
+    /// <summary>
+    ///     <see cref="Rebind"/> when the id is already tracked (so its stale reverse-map entries
+    ///     are dropped), <see cref="Register"/> otherwise. Used by the probe watcher, whose single
+    ///     reserved subscription is re-issued on every platform-set change.
+    /// </summary>
+    /// <param name="subscriptionId">The subscription identifier.</param>
+    /// <param name="bindings">The tag bindings for the subscription.</param>
+    public void Upsert(long subscriptionId, IReadOnlyList<TagBinding> bindings)
+    {
+        if (Contains(subscriptionId)) Rebind(subscriptionId, bindings);
+        else Register(subscriptionId, bindings);
+    }
+
    /// <summary>Gets the number of tracked subscriptions.</summary>
    public int TrackedSubscriptionCount => _bySubscriptionId.Count;
    /// <summary>Gets the number of tracked item handles.</summary>
@@ -0,0 +1,112 @@
+using System.Runtime.CompilerServices;
+using Shouldly;
+using Xunit;
+using ZB.MOM.WW.MxGateway.Contracts.Proto;
+using ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Health;
+using ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Runtime;
+
+namespace ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Tests.Health;
+
+/// <summary>
+///     Review 2026-10-21 G-4: the watcher's probe handles must live in the driver's
+///     <see cref="SubscriptionRegistry"/> under the reserved id, follow the platform set, be
+///     rebound on <see cref="PerPlatformProbeWatcher.ResubscribeAsync"/>, and leave on stop.
+/// </summary>
+public sealed class PerPlatformProbeWatcherRegistryTests
+{
+    private sealed class FakeSubscriber : IGalaxySubscriber
+    {
+        private int _next;
+        public List<int[]> Unsubscribes { get; } = [];
+
+        public Task<IReadOnlyList<SubscribeResult>> SubscribeBulkAsync(
+            IReadOnlyList<string> fullReferences, int bufferedUpdateIntervalMs, CancellationToken cancellationToken)
+            => Task.FromResult<IReadOnlyList<SubscribeResult>>(
+                [.. fullReferences.Select(r => new SubscribeResult { TagAddress = r, ItemHandle = ++_next, WasSuccessful = true })]);
+
+        public Task UnsubscribeBulkAsync(IReadOnlyList<int> itemHandles, CancellationToken cancellationToken)
+        {
+            Unsubscribes.Add([.. itemHandles]);
+            return Task.CompletedTask;
+        }
+
+        public async IAsyncEnumerable<MxEvent> StreamEventsAsync([EnumeratorCancellation] CancellationToken cancellationToken)
+        {
+            await Task.Delay(Timeout.Infinite, cancellationToken);
+            yield break;
+        }
+    }
+
+    private static (long Id, string Ref)[] Resolve(SubscriptionRegistry registry, int handle) =>
+        [.. registry.ResolveSubscribers(handle)];
+
+    /// <summary>Synced platforms resolve through the registry under the reserved id; removed ones stop resolving.</summary>
+    [Fact]
+    public async Task Sync_registers_probe_handles_under_the_reserved_id_and_tracks_removals()
+    {
+        var registry = new SubscriptionRegistry();
+        using var watcher = new PerPlatformProbeWatcher(new FakeSubscriber(), new HostStatusAggregator(), registry);
+
+        await watcher.SyncPlatformsAsync(["A", "B"], CancellationToken.None);
+
+        registry.Contains(SubscriptionRegistry.ProbeSubscriptionId).ShouldBeTrue();
+        var bindings = watcher.CurrentBindings;
+        bindings.Count.ShouldBe(2);
+        foreach (var b in bindings)
+        {
+            Resolve(registry, b.ItemHandle).ShouldBe([(SubscriptionRegistry.ProbeSubscriptionId, b.FullReference)]);
+        }
+
+        var bHandle = bindings.Single(b => b.FullReference == "B.ScanState").ItemHandle;
+        await watcher.SyncPlatformsAsync(["A"], CancellationToken.None);
+        Resolve(registry, bHandle).ShouldBeEmpty("a removed platform's handle must leave the fan-out map");
+        var aHandle = bindings.Single(b => b.FullReference == "A.ScanState").ItemHandle;
+        Resolve(registry, aHandle).ShouldHaveSingleItem("positive control: the kept platform still resolves");
+    }
+
+    /// <summary>Resubscribe replaces the handles and rebinds: old handles stop resolving, new ones resolve.</summary>
+    [Fact]
+    public async Task Resubscribe_rebinds_the_registry_to_the_fresh_handles()
+    {
+        var registry = new SubscriptionRegistry();
+        using var watcher = new PerPlatformProbeWatcher(new FakeSubscriber(), new HostStatusAggregator(), registry);
+        await watcher.SyncPlatformsAsync(["A"], CancellationToken.None);
+        var oldHandle = watcher.CurrentBindings.Single().ItemHandle;
+
+        (await watcher.ResubscribeAsync(CancellationToken.None)).ShouldBe(1);
+
+        var newHandle = watcher.CurrentBindings.Single().ItemHandle;
+        newHandle.ShouldNotBe(oldHandle);
+        Resolve(registry, oldHandle).ShouldBeEmpty();
+        Resolve(registry, newHandle).ShouldBe([(SubscriptionRegistry.ProbeSubscriptionId, "A.ScanState")]);
+        registry.TrackedSubscriptionCount.ShouldBe(1);
+    }
+
+    /// <summary>StopAsync unsubscribes the live handles and removes the reserved entry.</summary>
+    [Fact]
+    public async Task StopAsync_unsubscribes_and_removes_the_reserved_entry()
+    {
+        var registry = new SubscriptionRegistry();
+        var subscriber = new FakeSubscriber();
+        var watcher = new PerPlatformProbeWatcher(subscriber, new HostStatusAggregator(), registry);
+        await watcher.SyncPlatformsAsync(["A", "B"], CancellationToken.None);
+        var handles = watcher.CurrentBindings.Select(b => b.ItemHandle).OrderBy(h => h).ToArray();
+
+        await watcher.StopAsync(CancellationToken.None);
+
+        subscriber.Unsubscribes.ShouldHaveSingleItem().OrderBy(h => h).ShouldBe(handles);
+        registry.Contains(SubscriptionRegistry.ProbeSubscriptionId).ShouldBeFalse();
+        registry.TrackedItemHandleCount.ShouldBe(0);
+        await Should.NotThrowAsync(() => watcher.StopAsync(CancellationToken.None)); // idempotent
+    }
+
+    /// <summary>A watcher without a registry (the decoding-table unit tests) still works and registers nothing.</summary>
+    [Fact]
+    public async Task Without_a_registry_the_watcher_keeps_its_handles_private()
+    {
+        using var watcher = new PerPlatformProbeWatcher(new FakeSubscriber(), new HostStatusAggregator());
+        await watcher.SyncPlatformsAsync(["A"], CancellationToken.None);
+        watcher.CurrentBindings.ShouldHaveSingleItem();
+        (await watcher.ResubscribeAsync(CancellationToken.None)).ShouldBe(1);
+    }
+}
@@ -0,0 +1,96 @@
+using System.Runtime.CompilerServices;
+using System.Threading.Channels;
+using Google.Protobuf.WellKnownTypes;
+using Shouldly;
+using Xunit;
+using ZB.MOM.WW.MxGateway.Contracts.Proto;
+using ZB.MOM.WW.OtOpcUa.Core.Abstractions;
+using ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Runtime;
+
+namespace ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Tests.Runtime;
+
+/// <summary>
+///     Review 2026-10-21 G-10: a dispatch-loop exit left the read loop draining into a channel
+///     nobody read, with health Healthy and nothing telling the supervisor — a clean termination
+///     evading the watchdog. The exit now reports a stream fault. The fault is injected through
+///     the pump's handle factory, the one dispatch-side seam that runs outside the per-handler
+///     isolation.
+/// </summary>
+public sealed class EventPumpDispatchFaultTests
+{
+    private const int WaitMs = 10_000;
+
+    private sealed class ChannelSubscriber : IGalaxySubscriber
+    {
+        private readonly Channel<MxEvent> _events = Channel.CreateUnbounded<MxEvent>();
+
+        public Task<IReadOnlyList<SubscribeResult>> SubscribeBulkAsync(
+            IReadOnlyList<string> fullReferences, int bufferedUpdateIntervalMs, CancellationToken cancellationToken)
+            => Task.FromResult<IReadOnlyList<SubscribeResult>>([]);
+
+        public Task UnsubscribeBulkAsync(IReadOnlyList<int> itemHandles, CancellationToken cancellationToken) => Task.CompletedTask;
+
+        public async IAsyncEnumerable<MxEvent> StreamEventsAsync([EnumeratorCancellation] CancellationToken cancellationToken)
+        {
+            await foreach (var ev in _events.Reader.ReadAllAsync(cancellationToken).ConfigureAwait(false))
+            {
+                yield return ev;
+            }
+        }
+
+        public ValueTask EmitAsync(int handle) => _events.Writer.WriteAsync(new MxEvent
+        {
+            Family = MxEventFamily.OnDataChange,
+            ItemHandle = handle,
+            Value = new MxValue { DoubleValue = 1 },
+            Quality = 192,
+            SourceTimestamp = Timestamp.FromDateTime(DateTime.UtcNow),
+        });
+    }
+
+    /// <summary>A dispatch-loop exception reports a stream fault carrying that exception.</summary>
+    [Fact]
+    public async Task Dispatch_loop_exit_reports_a_stream_fault()
+    {
+        var subscriber = new ChannelSubscriber();
+        var registry = new SubscriptionRegistry();
+        registry.Register(1, [new TagBinding("Tank01.Level", 7)]);
+        var fault = new TaskCompletionSource<Exception>(TaskCreationOptions.RunContinuationsAsynchronously);
+
+        await using var pump = new EventPump(
+            subscriber, registry,
+            handleFactory: _ => throw new InvalidDataException("synthetic dispatch failure"),
+            channelCapacity: 4, clientName: "DispatchFault",
+            onStreamFault: ex => fault.TrySetResult(ex));
+        pump.Start();
+
+        await subscriber.EmitAsync(7);
+
+        (await Task.WhenAny(fault.Task, Task.Delay(WaitMs))).ShouldBe(fault.Task,
+            "a dispatch-loop exit must be reported as a stream fault");
+        (await fault.Task).ShouldBeOfType<InvalidDataException>();
+    }
+
+    /// <summary>Positive control: the same event with a working handle factory dispatches and reports no fault.</summary>
+    [Fact]
+    public async Task Same_event_with_a_working_factory_dispatches_without_a_fault()
+    {
+        var subscriber = new ChannelSubscriber();
+        var registry = new SubscriptionRegistry();
+        registry.Register(1, [new TagBinding("Tank01.Level", 7)]);
+        var faulted = false;
+        var delivered = new TaskCompletionSource<DataChangeEventArgs>(TaskCreationOptions.RunContinuationsAsynchronously);
+
+        await using var pump = new EventPump(
+            subscriber, registry, channelCapacity: 4, clientName: "DispatchFault",
+            onStreamFault: _ => faulted = true);
+        pump.OnDataChange += (_, e) => delivered.TrySetResult(e);
+        pump.Start();
+
+        await subscriber.EmitAsync(7);
+
+        (await Task.WhenAny(delivered.Task, Task.Delay(WaitMs))).ShouldBe(delivered.Task);
+        (await delivered.Task).FullReference.ShouldBe("Tank01.Level");
+        faulted.ShouldBeFalse();
+    }
+}
@@ -0,0 +1,217 @@
+using System.Runtime.CompilerServices;
+using System.Threading.Channels;
+using Google.Protobuf.WellKnownTypes;
+using Shouldly;
+using Xunit;
+using ZB.MOM.WW.MxGateway.Contracts.Proto;
+using ZB.MOM.WW.MxGateway.Contracts.Proto.Galaxy;
+using ZB.MOM.WW.OtOpcUa.Core.Abstractions;
+using ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Browse;
+using ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Config;
+using ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Runtime;
+
+namespace ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Tests.Runtime;
+
+/// <summary>
+///     Review 2026-10-21 G-4 / G-7 / G-11: the per-platform <c>ScanState</c> probing was structurally
+///     dead (its handles were never in the registry the pump resolves through, and a reconnect
+///     never re-subscribed them), discovery fetched the full galaxy twice, and
+///     <c>LastSuccessfulRead</c> never moved on data. These drive the real driver → pump → watcher
+///     → aggregator path with a fake subscriber and a fake hierarchy.
+/// </summary>
+public sealed class GalaxyDriverProbeWiringTests
+{
+    private const int WaitMs = 10_000;
+
+    private static GalaxyDriverOptions Opts() => new(
+        new GalaxyGatewayOptions("https://mxgw.test:5001", "key"),
+        new GalaxyMxAccessOptions("ProbeWiring"),
+        new GalaxyRepositoryOptions(WatchDeployEvents: false),
+        new GalaxyReconnectOptions());
+
+    private sealed class CountingHierarchySource : IGalaxyHierarchySource
+    {
+        public int Calls;
+
+        public Task<IReadOnlyList<GalaxyObject>> GetHierarchyAsync(CancellationToken cancellationToken)
+        {
+            Interlocked.Increment(ref Calls);
+            var platform = new GalaxyObject { GobjectId = 1, TagName = "Plat01", ContainedName = "Plat01" };
+            platform.TemplateChain.Add("$WinPlatform");
+            var tank = new GalaxyObject { GobjectId = 2, TagName = "Tank01", ContainedName = "Tank01", ParentGobjectId = 1 };
+            tank.TemplateChain.Add("$UserDefined");
+            tank.Attributes.Add(new GalaxyAttribute { AttributeName = "Level", FullTagReference = "Tank01.Level", MxDataType = 5 });
+            return Task.FromResult<IReadOnlyList<GalaxyObject>>([platform, tank]);
+        }
+    }
+
+    private sealed class StreamingSubscriber : IGalaxySubscriber
+    {
+        private readonly Channel<MxEvent> _events = Channel.CreateUnbounded<MxEvent>();
+        private int _nextHandle;
+
+        public List<string> Subscribed { get; } = [];
+        public Dictionary<string, int> HandleByRef { get; } = new(StringComparer.OrdinalIgnoreCase);
+
+        public Task<IReadOnlyList<SubscribeResult>> SubscribeBulkAsync(
+            IReadOnlyList<string> fullReferences, int bufferedUpdateIntervalMs, CancellationToken cancellationToken)
+        {
+            var results = new List<SubscribeResult>();
+            lock (Subscribed)
+            {
+                foreach (var r in fullReferences)
+                {
+                    var h = Interlocked.Increment(ref _nextHandle);
+                    Subscribed.Add(r);
+                    HandleByRef[r] = h;
+                    results.Add(new SubscribeResult { TagAddress = r, ItemHandle = h, WasSuccessful = true });
+                }
+            }
+            return Task.FromResult<IReadOnlyList<SubscribeResult>>(results);
+        }
+
+        public Task UnsubscribeBulkAsync(IReadOnlyList<int> itemHandles, CancellationToken cancellationToken) => Task.CompletedTask;
+
+        public async IAsyncEnumerable<MxEvent> StreamEventsAsync([EnumeratorCancellation] CancellationToken cancellationToken)
+        {
+            await foreach (var ev in _events.Reader.ReadAllAsync(cancellationToken).ConfigureAwait(false))
+            {
+                yield return ev;
+            }
+        }
+
+        public ValueTask EmitBoolAsync(int handle, bool value) => _events.Writer.WriteAsync(new MxEvent
+        {
+            Family = MxEventFamily.OnDataChange,
+            ItemHandle = handle,
+            Value = new MxValue { BoolValue = value },
+            Quality = 192,
+            SourceTimestamp = Timestamp.FromDateTime(DateTime.UtcNow),
+        });
+
+        public ValueTask EmitDoubleAsync(int handle, double value) => _events.Writer.WriteAsync(new MxEvent
+        {
+            Family = MxEventFamily.OnDataChange,
+            ItemHandle = handle,
+            Value = new MxValue { DoubleValue = value },
+            Quality = 192,
+            SourceTimestamp = Timestamp.FromDateTime(DateTime.UtcNow),
+        });
+    }
+
+    private sealed class NullBuilder : IAddressSpaceBuilder
+    {
+        public IAddressSpaceBuilder Folder(string browseName, string displayName) => this;
+        public IVariableHandle Variable(string browseName, string displayName, DriverAttributeInfo attributeInfo) => new Handle(attributeInfo.FullName);
+        public void AddProperty(string browseName, DriverDataType dataType, object? value) { }
+        private sealed class Handle(string fullReference) : IVariableHandle
+        {
+            public string FullReference { get; } = fullReference;
+            public IAlarmConditionSink MarkAsAlarmCondition(AlarmConditionInfo alarmInfo) => new NoopSink();
+        }
+        private sealed class NoopSink : IAlarmConditionSink
+        {
+            public void OnTransition(AlarmEventArgs args) { }
+        }
+    }
+
+    private static async Task WaitUntilAsync(Func<bool> condition, string because)
+    {
+        var deadline = DateTime.UtcNow.AddMilliseconds(WaitMs);
+        while (!condition())
+        {
+            if (DateTime.UtcNow > deadline) throw new ShouldAssertException(because);
+            await Task.Delay(10);
+        }
+    }
+
+    private static HostState? StateOf(GalaxyDriver driver, string host) =>
+        driver.GetHostStatuses().FirstOrDefault(s => s.HostName == host)?.State;
+
+    /// <summary>
+    ///     A ScanState event on the probe handle updates GetHostStatuses; after a reconnect replay
+    ///     the probe is re-subscribed and events on the NEW handle keep updating it, while the dead
+    ///     pre-reconnect handle no longer does. Probe traffic never reaches public OnDataChange.
+    /// </summary>
+    [Fact]
+    public async Task ScanState_event_updates_GetHostStatuses_and_survives_a_reconnect()
+    {
+        var sub = new StreamingSubscriber();
+        var source = new CountingHierarchySource();
+        await using var driver = new GalaxyDriver("drv", Opts(), hierarchySource: source, subscriber: sub);
+        var publicRefs = new List<string>();
+        driver.OnDataChange += (_, e) => { lock (publicRefs) publicRefs.Add(e.FullReference); };
+
+        await driver.InitializeAsync("{}", CancellationToken.None);
+        await driver.DiscoverAsync(new NullBuilder(), CancellationToken.None);
+        // The pump starts on the first value subscription (the probe watcher does not start it).
+        await driver.SubscribeAsync(["Tank01.Level"], TimeSpan.Zero, CancellationToken.None);
+
+        var probeHandle = sub.HandleByRef["Plat01.ScanState"];
+        await sub.EmitBoolAsync(probeHandle, true);
+        await WaitUntilAsync(() => StateOf(driver, "Plat01") == HostState.Running,
+            "a ScanState=true event must surface Plat01 as Running");
+
+        // Reconnect replay: every pre-reconnect handle is dead; the probe must rebind.
+        await driver.InvokeReplayForTestAsync(CancellationToken.None);
+        var rebound = sub.HandleByRef["Plat01.ScanState"];
+        rebound.ShouldNotBe(probeHandle, "replay must re-subscribe the probe and record the fresh handle");
+
+        await sub.EmitBoolAsync(rebound, false);
+        await WaitUntilAsync(() => StateOf(driver, "Plat01") == HostState.Stopped,
+            "a ScanState=false event on the post-reconnect handle must surface Plat01 as Stopped");
+
+        // The dead handle is no longer routed (positive control: the live one still is).
+        await sub.EmitBoolAsync(probeHandle, true);
+        await sub.EmitDoubleAsync(sub.HandleByRef["Tank01.Level"], 1.0);
+        await WaitUntilAsync(() => { lock (publicRefs) return publicRefs.Contains("Tank01.Level"); }, "value traffic must still flow");
+        StateOf(driver, "Plat01").ShouldBe(HostState.Stopped, "an event on the pre-reconnect handle must be dropped");
+
+        lock (publicRefs) publicRefs.ShouldNotContain("Plat01.ScanState", "probe traffic is the watcher's, never the host's");
+    }
+
+    /// <summary>Discovery asks the gateway for the hierarchy once; the probe set is selected from that same sweep.</summary>
+    [Fact]
+    public async Task DiscoverAsync_fetches_the_hierarchy_once_and_still_syncs_the_probe_set()
+    {
+        var sub = new StreamingSubscriber();
+        var source = new CountingHierarchySource();
+        await using var driver = new GalaxyDriver("drv", Opts(), hierarchySource: source, subscriber: sub);
+        await driver.InitializeAsync("{}", CancellationToken.None);
+
+        await driver.DiscoverAsync(new NullBuilder(), CancellationToken.None);
+
+        source.Calls.ShouldBe(1, "the probe selection must reuse the discoverer's sweep");
+        sub.Subscribed.ShouldContain("Plat01.ScanState", "positive control: the platform probe was still synced");
+        driver.ProbeWatcherForTests.ShouldNotBeNull().WatchedPlatforms.ShouldBe(["Plat01"]);
+    }
+
+    /// <summary>Inbound stream data advances LastSuccessfulRead (throttled); the throttle holds the stamp still.</summary>
+    [Fact]
+    public async Task Stream_data_advances_LastSuccessfulRead_under_a_throttle()
+    {
+        var sub = new StreamingSubscriber();
+        await using var driver = new GalaxyDriver("drv", Opts(), hierarchySource: null, subscriber: sub)
+        {
+            LivenessStampThrottle = TimeSpan.FromHours(1),
+        };
+        await driver.InitializeAsync("{}", CancellationToken.None);
+        var initStamp = driver.GetHealth().LastSuccessfulRead.ShouldNotBeNull();
+        await Task.Delay(20);
+
+        var delivered = 0;
+        driver.OnDataChange += (_, _) => Interlocked.Increment(ref delivered);
+        await driver.SubscribeAsync(["Tank01.Level"], TimeSpan.Zero, CancellationToken.None);
+        var handle = sub.HandleByRef["Tank01.Level"];
+
+        await sub.EmitDoubleAsync(handle, 1.0);
+        await WaitUntilAsync(() => Volatile.Read(ref delivered) == 1, "first event must be delivered");
+        var first = driver.GetHealth().LastSuccessfulRead.ShouldNotBeNull();
+        first.ShouldBeGreaterThan(initStamp, "data on the stream is liveness evidence");
+
+        await Task.Delay(20);
+        await sub.EmitDoubleAsync(handle, 2.0);
+        await WaitUntilAsync(() => Volatile.Read(ref delivered) == 2, "second event must be delivered");
+        driver.GetHealth().LastSuccessfulRead.ShouldBe(first, "inside the throttle window the stamp holds still");
+    }
+}