lmxopcua/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy/GalaxyDriver.cs

using System.Text.Json;
using Microsoft.Extensions.Logging;
using Microsoft.Extensions.Logging.Abstractions;
using ZB.MOM.WW.MxGateway.Client;
using ZB.MOM.WW.MxGateway.Contracts.Proto;
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.Health;
using ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Runtime;

namespace ZB.MOM.WW.OtOpcUa.Driver.Galaxy;

/// <summary>
///     In-process .NET 10 Galaxy driver — the only Galaxy backend since PR 7.2 retired
///     the legacy <c>Galaxy.Host</c> / <c>Galaxy.Proxy</c> / <c>Galaxy.Shared</c>
///     projects and the <c>OtOpcUaGalaxyHost</c> Windows service. Implements the full
///     capability surface: <see cref="ITagDiscovery"/>, <see cref="IReadable"/>,
///     <see cref="IWritable"/>, <see cref="ISubscribable"/>, <see cref="IRediscoverable"/>,
///     <see cref="IHostConnectivityProbe"/>, and <see cref="IAlarmSource"/>. Galaxy
///     access flows through the in-process driver over gRPC to the separately
///     installed <c>mxaccessgw</c> gateway (sibling repo), which owns the MXAccess
///     COM apartment server-side.
/// </summary>
/// <remarks>
///     <para>
///         Registered as a Tier A in-process driver alongside Modbus / S7 / etc. via
///         <see cref="GalaxyDriverFactoryExtensions"/> under driver-type name
///         "GalaxyMxGateway".
///     </para>
///     <para>
///         Tests inject capability seams (<see cref="IGalaxyHierarchySource"/>,
///         <see cref="IGalaxyDataReader"/>, <see cref="IGalaxyDataWriter"/>,
///         <see cref="IGalaxySubscriber"/>, <see cref="IGalaxyAlarmAcknowledger"/>,
///         <see cref="IGalaxyAlarmFeed"/>) via the internal ctor so capability flow
///         can be exercised without a real gw round-trip.
///     </para>
/// </remarks>
public sealed class GalaxyDriver
    : IDriver, ITagDiscovery, IReadable, IWritable, ISubscribable, IRediscoverable, IHostConnectivityProbe, IAlarmSource, IDisposable, IAsyncDisposable
{
    private readonly string _driverInstanceId;
    private readonly GalaxyDriverOptions _options;
    private readonly ILogger<GalaxyDriver> _logger;

    // PR 4.1 — IGalaxyHierarchySource is the test seam for browse. When null, the driver
    // lazily builds a GatewayGalaxyHierarchySource around a GalaxyRepositoryClient on
    // first DiscoverAsync. Tests inject a fake source via the internal ctor to exercise
    // GalaxyDiscoverer's translation logic without a real gRPC channel.
    private IGalaxyHierarchySource? _hierarchySource;
    private GalaxyRepositoryClient? _ownedRepositoryClient;

    // IGalaxyDataReader is the test seam for IReadable. Production ReadAsync calls
    // ReadViaSubscribeOnceAsync (SubscribeBulk + first-event wait + Unsubscribe) through
    // the live GalaxyMxSession built by InitializeAsync; no legacy-host backend exists.
    // When a test injects a reader via the internal ctor, that injected reader is used
    // instead so capability flow can be exercised without a real gRPC round-trip.
    private IGalaxyDataReader? _dataReader;

    // PR 4.3 — IGalaxyDataWriter is the test seam for IWritable. Production wraps
    // GalaxyMxSession via GatewayGalaxyDataWriter (Write / WriteSecured routing). The
    // per-tag SecurityClassification map is populated during ITagDiscovery and consumed
    // here at write time.
    private IGalaxyDataWriter? _dataWriter;
    private readonly System.Collections.Concurrent.ConcurrentDictionary<string, SecurityClassification>
        _securityByFullRef = new(StringComparer.OrdinalIgnoreCase);

    // v3 RawPath keying. The server hands every read/write/subscribe reference as a RawPath; MXAccess
    // dials the Galaxy attributeRef (tag_name.AttributeName). These two tables — built once in the ctor
    // from options.RawTags (each entry's TagConfig carries the camelCase "attributeRef" key) — bridge the
    // two identity domains at the driver boundary:
    //   _attributeRefByRawPath : RawPath      -> attributeRef  (forward dial:  read/write/subscribe/ack)
    //   _rawTagByAttributeRef  : attributeRef -> RawTagEntry    (reverse + WriteIdempotent: discovery,
    //                                                            OnDataChange, alarm ConditionId)
    // Both fall back to identity on a miss, so the live-browse discovery path (no authored tags) and any
    // synthetic alarm sub-ref (attributeRef.InAlarm, never an authored RawPath) dial straight through
    // unchanged — reproducing pre-v3 behaviour exactly.
    private readonly Dictionary<string, string> _attributeRefByRawPath;
    private readonly Dictionary<string, RawTagEntry> _rawTagByAttributeRef;

    // PR 4.4 — subscription lifecycle. The pump consumes the gw event stream and fans
    // out OnDataChange events to every registered driver subscription via the registry's
    // reverse map. The subscriber is the test seam — production uses
    // GatewayGalaxySubscriber over a connected GalaxyMxSession.
    private IGalaxySubscriber? _subscriber;
    private readonly SubscriptionRegistry _subscriptions = new();
    private EventPump? _eventPump;
    private readonly Lock _pumpLock = new();

    // IAlarmSource implementation. Production-side acks route through
    // GatewayGalaxyAlarmAcknowledger which calls the session-less
    // MxGatewayClient.AcknowledgeAlarmAsync RPC; alarm transitions arrive on the
    // gateway's session-less StreamAlarms feed via GatewayGalaxyAlarmFeed. Tests inject
    // IGalaxyAlarmAcknowledger + IGalaxyAlarmFeed via the internal ctor to exercise the
    // wiring without a running gateway. This driver bridges the feed's OnAlarmTransition
    // onto IAlarmSource.OnAlarmEvent.
    private IGalaxyAlarmAcknowledger? _alarmAcknowledger;
    private IGalaxyAlarmFeed? _alarmFeed;
    private readonly Lock _alarmHandlersLock = new();
    private readonly Lock _alarmFeedLock = new();
    private bool _alarmFeedWired;
    // List preserves insertion order so OnAlarmFeedTransition always picks the
    // earliest-registered handle — a deterministic choice that doesn't vary as
    // handles are added/removed (HashSet.First() is unstable).
    private readonly List<GalaxyAlarmSubscriptionHandle> _alarmSubscriptions = new();

    // Production runtime owned by InitializeAsync (BuildProductionRuntimeAsync). The
    // driver builds these when it opens a real gw session; tests bypass them by
    // injecting seams via the internal ctor, which leaves both fields null and skips
    // BuildProductionRuntimeAsync entirely.
    private GalaxyMxSession? _ownedMxSession;
    private MxGatewayClient? _ownedMxClient;

    // PR 4.5 — reconnect supervisor. Reflects in DriverState.Degraded while not Healthy.
    private ReconnectSupervisor? _supervisor;

    // PR 4.6 — IRediscoverable plumbing.
    private DeployWatcher? _deployWatcher;

    // PR 4.7 — IHostConnectivityProbe plumbing. The aggregator owns the merged
    // transport+per-platform view; the forwarder is fed from the supervisor on
    // transport state transitions; the probe watcher subscribes ScanState attributes
    // for every discovered platform and pushes value changes to the aggregator.
    private readonly HostStatusAggregator _hostStatuses = new();
    private HostConnectivityForwarder? _transportForwarder;
    private PerPlatformProbeWatcher? _probeWatcher;

    private DriverHealth _health = new(DriverState.Unknown, null, null);
    private bool _disposed;

    /// <summary>
    ///     Server-pushed data-change notification. Fires from the
    ///     <see cref="EventPump"/>'s background loop; handlers should be cheap (or queue
    ///     onto another thread) to avoid blocking the gw event stream.
    /// </summary>
    public event EventHandler<DataChangeEventArgs>? OnDataChange;

    /// <summary>Fires when the gateway signals a deploy-time change (PR 4.6 DeployWatcher).</summary>
    public event EventHandler<RediscoveryEventArgs>? OnRediscoveryNeeded;

    /// <summary>Fires when a host transitions Running ↔ Stopped (PR 4.7 HostStatusAggregator).</summary>
    public event EventHandler<HostStatusChangedEventArgs>? OnHostStatusChanged;

    /// <inheritdoc />
    public event EventHandler<AlarmEventArgs>? OnAlarmEvent;

    /// <summary>Initializes a new instance of the <see cref="GalaxyDriver"/> class.</summary>
    /// <param name="driverInstanceId">The unique identifier for this driver instance.</param>
    /// <param name="options">The Galaxy driver configuration options.</param>
    /// <param name="logger">Optional logger instance for diagnostics.</param>
    public GalaxyDriver(
        string driverInstanceId,
        GalaxyDriverOptions options,
        ILogger<GalaxyDriver>? logger = null)
        : this(driverInstanceId, options,
              hierarchySource: null, dataReader: null, dataWriter: null, subscriber: null,
              alarmAcknowledger: null, alarmFeed: null, logger)
    {
    }

    /// <summary>
    ///     Test-visible ctor — inject custom seams so <see cref="DiscoverAsync"/>,
    ///     <see cref="ReadAsync"/>, <see cref="WriteAsync"/>, and
    ///     <see cref="SubscribeAsync"/> can be exercised against canned data without
    ///     building real gRPC channels.
    /// </summary>
    /// <param name="driverInstanceId">The unique identifier for this driver instance.</param>
    /// <param name="options">The Galaxy driver configuration options.</param>
    /// <param name="hierarchySource">Optional custom hierarchy source for testing.</param>
    /// <param name="dataReader">Optional custom data reader for testing.</param>
    /// <param name="dataWriter">Optional custom data writer for testing.</param>
    /// <param name="subscriber">Optional custom subscriber for testing.</param>
    /// <param name="alarmAcknowledger">Optional custom alarm acknowledger for testing.</param>
    /// <param name="alarmFeed">Optional custom alarm feed for testing.</param>
    /// <param name="logger">Optional logger instance for diagnostics.</param>
    internal GalaxyDriver(
        string driverInstanceId,
        GalaxyDriverOptions options,
        IGalaxyHierarchySource? hierarchySource,
        IGalaxyDataReader? dataReader = null,
        IGalaxyDataWriter? dataWriter = null,
        IGalaxySubscriber? subscriber = null,
        IGalaxyAlarmAcknowledger? alarmAcknowledger = null,
        IGalaxyAlarmFeed? alarmFeed = null,
        ILogger<GalaxyDriver>? logger = null)
    {
        _driverInstanceId = !string.IsNullOrWhiteSpace(driverInstanceId)
            ? driverInstanceId
            : throw new ArgumentException("Driver instance id required.", nameof(driverInstanceId));
        _options = options ?? throw new ArgumentNullException(nameof(options));
        _logger = logger ?? NullLogger<GalaxyDriver>.Instance;
        _hierarchySource = hierarchySource;
        _dataReader = dataReader;
        _dataWriter = dataWriter;
        _subscriber = subscriber;
        _alarmAcknowledger = alarmAcknowledger;
        _alarmFeed = alarmFeed;

        // Build the v3 RawPath ⇄ attributeRef tables from the authored raw tags. Empty when no tags
        // are authored (live-browse path), leaving every translation an identity no-op.
        (_attributeRefByRawPath, _rawTagByAttributeRef) = BuildRawTagTables(_options.RawTags, _logger);

        // Forward the aggregator's transitions through IHostConnectivityProbe.
        _hostStatuses.OnHostStatusChanged += (_, args) => OnHostStatusChanged?.Invoke(this, args);
    }

    /// <summary>
    ///     Build the RawPath → attributeRef and attributeRef → <see cref="RawTagEntry"/> tables from the
    ///     authored raw tags. Each entry's <see cref="RawTagEntry.TagConfig"/> carries the Galaxy dialled
    ///     address under the camelCase key <c>attributeRef</c>. Entries whose blob has no usable
    ///     <c>attributeRef</c> are skipped (they can't be dialled). On a duplicate attributeRef (two
    ///     authored RawPaths pointing at the same Galaxy attribute) the first entry wins the reverse map —
    ///     the value fan-out / alarm ConditionId then surfaces under that first RawPath; the forward map
    ///     still dials both. Case-insensitive to match the driver's reference handling.
    /// </summary>
    private static (Dictionary<string, string> Forward, Dictionary<string, RawTagEntry> Reverse)
        BuildRawTagTables(IReadOnlyList<RawTagEntry> rawTags, ILogger logger)
    {
        var forward = new Dictionary<string, string>(StringComparer.OrdinalIgnoreCase);
        var reverse = new Dictionary<string, RawTagEntry>(StringComparer.OrdinalIgnoreCase);
        foreach (var entry in rawTags)
        {
            if (string.IsNullOrEmpty(entry.RawPath)) continue;
            var attributeRef = TryReadAttributeRef(entry.TagConfig);
            if (string.IsNullOrEmpty(attributeRef))
            {
                logger.LogWarning(
                    "GalaxyDriver raw tag {RawPath} has no 'attributeRef' in its TagConfig — skipped (cannot dial MXAccess).",
                    entry.RawPath);
                continue;
            }
            forward[entry.RawPath] = attributeRef;
            reverse.TryAdd(attributeRef, entry); // first-wins on a shared attributeRef
        }
        return (forward, reverse);
    }

    /// <summary>
    ///     Read the Galaxy dialled address from a tag's <c>TagConfig</c> JSON — the camelCase
    ///     <c>attributeRef</c> key (v3 rename of the pre-v3 PascalCase <c>FullName</c>). Case-insensitive
    ///     on the property name for authoring tolerance; returns null on an absent/blank/non-string value
    ///     or unparseable blob (the entry is then skipped, never dialled).
    /// </summary>
    private static string? TryReadAttributeRef(string tagConfigJson)
    {
        if (string.IsNullOrWhiteSpace(tagConfigJson)) return null;
        try
        {
            using var doc = JsonDocument.Parse(tagConfigJson);
            if (doc.RootElement.ValueKind != JsonValueKind.Object) return null;
            foreach (var prop in doc.RootElement.EnumerateObject())
            {
                if (!prop.NameEquals("attributeRef")
                    && !string.Equals(prop.Name, "attributeRef", StringComparison.OrdinalIgnoreCase))
                {
                    continue;
                }
                if (prop.Value.ValueKind != JsonValueKind.String) return null;
                var v = prop.Value.GetString();
                return string.IsNullOrWhiteSpace(v) ? null : v;
            }
        }
        catch (JsonException)
        {
            // Malformed blob — treat as un-dialable; BuildRawTagTables logs the skip.
        }
        return null;
    }

    /// <summary>RawPath → Galaxy attributeRef for MXAccess dialling; identity on a miss.</summary>
    private string ToAttributeRef(string rawPath) =>
        _attributeRefByRawPath.GetValueOrDefault(rawPath, rawPath);

    /// <summary>Galaxy attributeRef → RawPath (the server-facing identity); identity on a miss.</summary>
    private string ToRawPath(string attributeRef) =>
        _rawTagByAttributeRef.TryGetValue(attributeRef, out var e) ? e.RawPath : attributeRef;

    /// <inheritdoc />
    public string DriverInstanceId => _driverInstanceId;

    /// <inheritdoc />
    public string DriverType => GalaxyDriverFactoryExtensions.DriverTypeName;

    /// <summary>Test-visible options snapshot.</summary>
    internal GalaxyDriverOptions Options => _options;

    /// <summary>
    ///     Test-visible entry into <see cref="ReplayAsync"/>. The supervisor's
    ///     <see cref="ReconnectSupervisor.ReportTransportFailure"/> drives this on a
    ///     background task in production; tests prefer to invoke it directly so the
    ///     <see cref="GalaxyReconnectOptions.ReplayOnSessionLost"/> branch can be
    ///     asserted deterministically.
    /// </summary>
    /// <param name="cancellationToken">Cancellation token for the replay operation.</param>
    /// <returns>A task that represents the asynchronous operation.</returns>
    internal Task InvokeReplayForTestAsync(CancellationToken cancellationToken) =>
        ReplayAsync(cancellationToken);

    /// <inheritdoc />
    public async Task InitializeAsync(string driverConfigJson, CancellationToken cancellationToken)
    {
        ObjectDisposedException.ThrowIf(_disposed, this);

        // Tests inject seams via the internal ctor; production InitializeAsync builds
        // the gateway client + session + per-capability runtime components from
        // GalaxyDriverOptions. When seams are pre-injected we leave them alone (the
        // test exercises the wired surface without a real gw round-trip).
        if (_subscriber is null && _dataWriter is null && _hierarchySource is null)
        {
            await BuildProductionRuntimeAsync(cancellationToken).ConfigureAwait(false);
        }
        else
        {
            _logger.LogDebug(
                "GalaxyDriver {InstanceId} initializing with pre-injected seams — production runtime build skipped",
                _driverInstanceId);
        }

        StartDeployWatcher();
        _logger.LogInformation(
            "GalaxyDriver {InstanceId} initialized — endpoint={Endpoint} clientName={ClientName}",
            _driverInstanceId, _options.Gateway.Endpoint, _options.MxAccess.ClientName);
        _health = new DriverHealth(DriverState.Healthy, DateTime.UtcNow, null);
    }

    /// <summary>
    ///     Build the production gw client + session + per-capability runtime components
    ///     from <c>_options</c>. Sets up the reconnect supervisor's reopen / replay
    ///     callbacks so a transport drop replays every active subscription on the
    ///     restored session.
    /// </summary>
    private async Task BuildProductionRuntimeAsync(CancellationToken cancellationToken)
    {
        var clientOptions = BuildClientOptions(_options.Gateway);
        _ownedMxClient = MxGatewayClient.Create(clientOptions);
        _ownedMxSession = new GalaxyMxSession(_options.MxAccess, _logger);
        await _ownedMxSession.ConnectAsync(clientOptions, cancellationToken).ConfigureAwait(false);

        // PR 6.1 — wrap the gw-facing seams in tracing decorators so every Subscribe /
        // Unsubscribe / Write / StreamEvents call emits a span on the
        // "ZB.MOM.WW.OtOpcUa.Driver.Galaxy" ActivitySource. The host process's tracing
        // listener (OTLP exporter, dotnet-trace, etc.) consumes these without the driver
        // taking a dependency on the OpenTelemetry packages.
        _subscriber = new TracedGalaxySubscriber(
            new GatewayGalaxySubscriber(_ownedMxSession, _logger), _options.MxAccess.ClientName);
        _dataWriter = new TracedGalaxyDataWriter(
            // Let the writer borrow live MXAccess item handles the subscription registry already
            // holds, so the first write to an already-subscribed tag skips a redundant AddItem.
            new GatewayGalaxyDataWriter(
                _ownedMxSession, _options.MxAccess.WriteUserId, _logger,
                _subscriptions.TryResolveItemHandle),
            _options.MxAccess.ClientName);

        _supervisor = new ReconnectSupervisor(
            reopen: ReopenAsync,
            replay: ReplayAsync,
            options: new ReconnectOptions(
                InitialBackoffOverride: TimeSpan.FromMilliseconds(_options.Reconnect.InitialBackoffMs),
                MaxBackoffOverride: TimeSpan.FromMilliseconds(_options.Reconnect.MaxBackoffMs)),
            logger: _logger);

        _transportForwarder = new HostConnectivityForwarder(_options.MxAccess.ClientName, _hostStatuses, _logger);
        _transportForwarder.SetTransport(HostState.Running); // initial state — we just connected

        _supervisor.StateChanged += OnSupervisorStateChanged;

        _probeWatcher = new PerPlatformProbeWatcher(
            _subscriber, _hostStatuses, _logger,
            bufferedUpdateIntervalMs: _options.MxAccess.PublishingIntervalMs);

        // Wire the alarm acknowledger + feed to the live gateway client. Both are
        // session-less — the gateway serves alarms from an always-on central monitor —
        // so they hang off the owned MxGatewayClient, not the worker session.
        _alarmAcknowledger ??= new GatewayGalaxyAlarmAcknowledger(_ownedMxClient, _logger);
        _alarmFeed ??= new GatewayGalaxyAlarmFeed(
            _ownedMxClient.StreamAlarmsAsync, _logger, _options.MxAccess.ClientName);
    }

    /// <summary>
    ///     Reopen callback for <see cref="ReconnectSupervisor"/>: recreate the gw session. After a
    ///     gateway restart the existing session handle is faulted / stale but non-null, so a plain
    ///     <c>ConnectAsync</c> would no-op and the supervisor would loop forever replaying against a
    ///     dead session. <see cref="GalaxyMxSession.RecreateAsync"/> disposes the stale session +
    ///     owned client and rebuilds, so a never-connected session still opens fresh.
    /// </summary>
    private async Task ReopenAsync(CancellationToken cancellationToken)
    {
        if (_ownedMxSession is null) return;
        var clientOptions = BuildClientOptions(_options.Gateway);
        await _ownedMxSession.RecreateAsync(clientOptions, cancellationToken).ConfigureAwait(false);
        // The recreated session invalidates every prior gw item handle; drop the writer's handle/advise
        // caches so the next write re-AddItems + re-AdviseSupervisory against the fresh session.
        _dataWriter?.InvalidateHandleCaches();
    }

    /// <summary>
    ///     Replay callback. Walks every active subscription, re-issues SubscribeBulk for
    ///     its tag list, and <see cref="SubscriptionRegistry.Rebind">rebinds</see> the
    ///     registry with the fresh item handles the gateway returned — the pre-reconnect
    ///     handles are dead once the session reopened. The faulted <see cref="EventPump"/>
    ///     is recreated first so the replayed subscriptions have a live StreamEvents
    ///     consumer; without that restart the replayed tags are subscribed on the gw but
    ///     never reach <c>OnDataChange</c>. PR 6.x can swap this for
    ///     the gw's batched <c>ReplaySubscriptionsCommand</c> once it ships.
    /// </summary>
    private async Task ReplayAsync(CancellationToken cancellationToken)
    {
        if (_subscriber is null) return;
        var entries = _subscriptions.SnapshotEntries();
        if (entries.Count == 0) return;

        // Honor ReplayOnSessionLost. When operators opt out (false) we skip the
        // per-tag SubscribeBulk fan-out — they're delegating to the gateway's
        // session-level ReplaySubscriptions or accept post-reconnect tag loss. We
        // still restart the EventPump so a future Subscribe call lands on a live
        // consumer.
        if (!_options.Reconnect.ReplayOnSessionLost)
        {
            RestartEventPumpForReplay();
            _logger.LogInformation(
                "GalaxyDriver {InstanceId} reconnect replay skipped — ReplayOnSessionLost=false ({SubCount} subscriptions tracked)",
                _driverInstanceId, entries.Count);
            return;
        }

        // The stream-fault that triggered this recovery left the old pump's RunAsync loop
        // exited and its channel completed; EventPump.Start() is a no-op on a non-null but
        // completed loop. Recreate the pump so the replayed subscriptions have a consumer.
        RestartEventPumpForReplay();

        var tagCount = 0;
        foreach (var (subscriptionId, oldBindings) in entries)
        {
            var refs = oldBindings
                .Select(b => b.FullReference)
                .Distinct(StringComparer.OrdinalIgnoreCase)
                .ToArray();
            if (refs.Length == 0) continue;

            var results = await _subscriber
                .SubscribeBulkAsync(refs, _options.MxAccess.PublishingIntervalMs, cancellationToken)
                .ConfigureAwait(false);

            var byAddress = BuildResultIndex(results);
            var newBindings = new List<TagBinding>(refs.Length);
            foreach (var fullRef in refs)
            {
                var itemHandle = byAddress.TryGetValue(fullRef, out var match) && match.WasSuccessful
                    ? match.ItemHandle
                    : 0;
                newBindings.Add(new TagBinding(fullRef, itemHandle));
            }

            // Rebind so the EventPump fan-out reverse map points at the post-reconnect
            // handles; otherwise events on the new handles miss every subscription.
            _subscriptions.Rebind(subscriptionId, newBindings);
            tagCount += refs.Length;
        }

        _logger.LogInformation(
            "GalaxyDriver {InstanceId} replay completed — {SubCount} subscriptions, {TagCount} tags re-subscribed",
            _driverInstanceId, entries.Count, tagCount);
    }

    /// <summary>
    ///     Index a SubscribeBulk result list by tag address (OrdinalIgnoreCase) so the
    ///     subscribe / replay correlation loops are O(1) per reference rather than a
    ///     linear scan. Last-write-wins on a duplicate address (the gw shouldn't emit one).
    /// </summary>
    private static Dictionary<string, SubscribeResult> BuildResultIndex(IReadOnlyList<SubscribeResult> results)
    {
        var index = new Dictionary<string, SubscribeResult>(results.Count, StringComparer.OrdinalIgnoreCase);
        foreach (var result in results)
        {
            if (!string.IsNullOrEmpty(result.TagAddress)) index[result.TagAddress] = result;
        }
        return index;
    }

    /// <summary>
    ///     Dispose the faulted <see cref="EventPump"/> and create a fresh one bound to the
    ///     same subscriber / registry, started immediately. Invoked from the reconnect
    ///     replay path. No-op when no pump was ever started (no active subscriptions).
    /// </summary>
    private void RestartEventPumpForReplay()
    {
        EventPump? old;
        lock (_pumpLock)
        {
            old = _eventPump;
            if (old is null) return; // pump never started — nothing to restart
            _eventPump = null;
        }

        // Detach + dispose the faulted pump outside the lock so a slow shutdown doesn't
        // block a concurrent EnsureEventPumpStarted; the old loop already exited on fault.
        old.OnDataChange -= OnPumpDataChange;
        try { old.DisposeAsync().AsTask().GetAwaiter().GetResult(); }
        catch (Exception ex)
        {
            _logger.LogWarning(ex,
                "GalaxyDriver {InstanceId} faulted EventPump dispose failed during replay — continuing.",
                _driverInstanceId);
        }

        // EnsureEventPumpStarted creates + starts a fresh pump under the lock. Skip the
        // recreate if the driver is being disposed — Dispose already tore the pump down
        // and a fresh one here would leak past the driver's lifetime.
        if (_disposed) return;
        EnsureEventPumpStarted();
    }

    private void OnSupervisorStateChanged(object? sender, StateTransition transition)
    {
        // Reflect supervisor state in DriverHealth + transport forwarder.
        _health = transition.Next switch
        {
            ReconnectSupervisor.State.Healthy => new DriverHealth(DriverState.Healthy, DateTime.UtcNow, null),
            _ => new DriverHealth(DriverState.Degraded, _health.LastSuccessfulRead, transition.Cause),
        };
        if (_transportForwarder is not null)
        {
            var hostState = transition.Next == ReconnectSupervisor.State.Healthy
                ? HostState.Running
                : HostState.Stopped;
            _transportForwarder.SetTransport(hostState);
        }
    }

    private MxGatewayClientOptions BuildClientOptions(GalaxyGatewayOptions gw) => new()
    {
        Endpoint = new Uri(gw.Endpoint, UriKind.Absolute),
        // Pass the logger so the literal-arm cleartext fallback surfaces a startup
        // warning rather than silently shipping the key. The resolver lives in
        // Driver.Galaxy.Contracts (GalaxySecretRef) so the runtime driver and the
        // AdminUI browser share one implementation.
        ApiKey = GalaxySecretRef.ResolveApiKey(gw.ApiKeySecretRef, _logger),
        UseTls = gw.UseTls,
        CaCertificatePath = gw.CaCertificatePath,
        ConnectTimeout = TimeSpan.FromSeconds(gw.ConnectTimeoutSeconds),
        DefaultCallTimeout = TimeSpan.FromSeconds(gw.DefaultCallTimeoutSeconds),
        StreamTimeout = gw.StreamTimeoutSeconds > 0 ? TimeSpan.FromSeconds(gw.StreamTimeoutSeconds) : null,
    };

    private void StartDeployWatcher()
    {
        if (!_options.Repository.WatchDeployEvents) return;
        if (_ownedRepositoryClient is null && _hierarchySource is null) return;

        // Reuse the lazily-built repository client (DiscoverAsync constructs it on demand).
        // If discovery hasn't run yet, build the client here so the watcher has a target.
        // Guard with ??= so if BuildDefaultHierarchySource later runs it reuses this client
        // rather than overwriting the field and leaking the first instance.
        _ownedRepositoryClient ??= ZB.MOM.WW.MxGateway.Client.GalaxyRepositoryClient.Create(
            BuildClientOptions(_options.Gateway));

        var source = new GatewayGalaxyDeployWatchSource(_ownedRepositoryClient);
        _deployWatcher = new DeployWatcher(source, _logger);
        _deployWatcher.OnRediscoveryNeeded += (_, args) => OnRediscoveryNeeded?.Invoke(this, args);

        // StartAsync schedules the background loop and returns Task.CompletedTask immediately.
        // It throws InvalidOperationException synchronously if called twice (programming error).
        // Don't discard the return value — observe any synchronous throw.
        var startTask = _deployWatcher.StartAsync(CancellationToken.None);
        // The task is already completed (StartAsync is synchronous); surface any synchronous fault.
        if (startTask.IsFaulted) startTask.GetAwaiter().GetResult();
    }

    /// <inheritdoc />
    public Task ReinitializeAsync(string driverConfigJson, CancellationToken cancellationToken)
    {
        ObjectDisposedException.ThrowIf(_disposed, this);
        if (!string.IsNullOrWhiteSpace(driverConfigJson))
        {
            // In-place config reapply. The driver does not currently support hot-swapping
            // GalaxyDriverOptions at runtime — changing the gateway endpoint, MxAccess
            // client name, or reconnect policy requires tearing down the gw session,
            // supervisor, event pump, and address space. The host stack handles that via
            // DriverInstance restart, so this method only accepts an equivalent config (no
            // meaningful change) and refreshes health; a non-equivalent reapply throws
            // NotSupportedException so the caller knows the change wasn't applied.
            //
            // Materialise the incoming config and compare against the live options. We
            // refuse any change that would require a session teardown rather than
            // pretending to apply it.
            GalaxyDriverOptions incoming;
            try
            {
                // Reuse the factory's parse pipeline so any missing-required-field
                // error surfaces with the same diagnostic text as InitializeAsync.
                var transient = GalaxyDriverFactoryExtensions.CreateInstance(_driverInstanceId, driverConfigJson);
                incoming = transient.Options;
                // The transient instance never started a runtime — disposing is cheap.
                transient.Dispose();
            }
            catch (Exception ex) when (ex is not NotSupportedException and not ObjectDisposedException)
            {
                throw new NotSupportedException(
                    $"GalaxyDriver.ReinitializeAsync could not parse the incoming DriverConfig JSON for '{_driverInstanceId}': {ex.Message}",
                    ex);
            }

            if (!OptionsAreEquivalent(_options, incoming))
            {
                throw new NotSupportedException(
                    "GalaxyDriver.ReinitializeAsync does not support hot-swapping driver options at runtime " +
                    "(gateway endpoint, MxAccess client name, reconnect policy, etc.). Restart the DriverInstance " +
                    "through the host stack to apply a config change.");
            }
        }
        _health = new DriverHealth(DriverState.Healthy, DateTime.UtcNow, null);
        return Task.CompletedTask;
    }

    /// <summary>
    ///     Compare two <see cref="GalaxyDriverOptions"/> for runtime equivalence — every
    ///     field that drives gw session shape or reconnect behaviour must match. Compared
    ///     field-wise (the nested records supply value-equality) and <b>excluding
    ///     <see cref="GalaxyDriverOptions.RawTags"/></b>: the authored tag set is address-space
    ///     state re-applied through rediscovery, not a session-teardown trigger, and a record's
    ///     synthesized equality would compare the two <c>RawTags</c> lists by reference anyway.
    /// </summary>
    private static bool OptionsAreEquivalent(GalaxyDriverOptions a, GalaxyDriverOptions b) =>
        a.Gateway == b.Gateway
        && a.MxAccess == b.MxAccess
        && a.Repository == b.Repository
        && a.Reconnect == b.Reconnect
        && a.ProbeTimeoutSeconds == b.ProbeTimeoutSeconds;

    /// <inheritdoc />
    public Task ShutdownAsync(CancellationToken cancellationToken)
    {
        if (_disposed) return Task.CompletedTask;
        _logger.LogInformation("GalaxyDriver {InstanceId} shutting down", _driverInstanceId);
        _health = new DriverHealth(DriverState.Unknown, _health.LastSuccessfulRead, null);
        return Task.CompletedTask;
    }

    /// <inheritdoc />
    public DriverHealth GetHealth()
    {
        // Reconnect supervisor wins when degraded — the cached _health reflects the last
        // successful operation, but ongoing recovery should surface as Degraded.
        if (_supervisor?.IsDegraded == true)
        {
            return new DriverHealth(DriverState.Degraded, _health.LastSuccessfulRead, _supervisor.LastError);
        }
        return _health;
    }

    // ===== IHostConnectivityProbe (PR 4.7 wire-up) =====

    /// <inheritdoc />
    public IReadOnlyList<HostConnectivityStatus> GetHostStatuses() => _hostStatuses.Snapshot();

    /// <inheritdoc />
    public long GetMemoryFootprint()
    {
        // Estimated footprint: 64 bytes × tracked item handles (one gw subscription entry
        // per bound tag) + 256 bytes × tracked driver subscriptions (registry overhead per
        // OPC UA monitored item). Returns 0 when no subscriptions are active. These
        // constants are conservative — a 50k-tag set occupies ~3 MB and registers clearly
        // with the server's cache-flush heuristic.
        const long BytesPerItemHandle = 64L;     // TagBinding + reverse-map entry
        const long BytesPerSubscription = 256L;  // SubscriptionEntry overhead
        return (_subscriptions.TrackedItemHandleCount * BytesPerItemHandle)
             + (_subscriptions.TrackedSubscriptionCount * BytesPerSubscription);
    }

    /// <inheritdoc />
    public Task FlushOptionalCachesAsync(CancellationToken cancellationToken) => Task.CompletedTask;

    // ===== ITagDiscovery (PR 4.1) =====

    /// <inheritdoc />
    // Run-once: DiscoverAsync fetches the full Galaxy hierarchy inline and streams the complete
    // node set within a single awaited call — there is no FOCAS-style background cache that fills
    // in after connect. Galaxy is a heavy network driver, so the bounded post-connect retry loop
    // is deliberately avoided; re-discovery on Galaxy redeploy is handled separately via
    // IRediscoverable + the deploy-event watcher.
    public DiscoveryRediscoverPolicy RediscoverPolicy => DiscoveryRediscoverPolicy.Once;

    /// <inheritdoc />
    public async Task DiscoverAsync(IAddressSpaceBuilder builder, CancellationToken cancellationToken)
    {
        ObjectDisposedException.ThrowIf(_disposed, this);
        ArgumentNullException.ThrowIfNull(builder);

        // Wrap the builder with a SecurityCapturingBuilder so each discovered attribute's
        // SecurityClassification is recorded in _securityByFullRef for write-time routing.
        // After discovery, SyncPlatformsAsync refreshes the probe watcher's membership so
        // newly added $WinPlatform / $AppEngine objects start advising their ScanState attribute.
        var capturingBuilder = new SecurityCapturingBuilder(builder, _securityByFullRef);
        var source = _hierarchySource ??= BuildDefaultHierarchySource();
        // Thread the v3 reverse resolver so the discoverer emits each authored attribute's RawPath as
        // its node reference (FullName) — the security map SecurityCapturingBuilder captures is then
        // keyed by RawPath, matching what WriteAsync resolves against.
        var discoverer = new GalaxyDiscoverer(
            source, attributeRef => _rawTagByAttributeRef.GetValueOrDefault(attributeRef));
        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)
                    || string.Equals(t, "$AppEngine", StringComparison.OrdinalIgnoreCase)))
                .Select(o => o.TagName)
                .Where(name => !string.IsNullOrEmpty(name));
            await _probeWatcher.SyncPlatformsAsync(platforms, cancellationToken).ConfigureAwait(false);
        }
    }

    private SecurityClassification ResolveSecurity(string fullReference) =>
        _securityByFullRef.TryGetValue(fullReference, out var sec) ? sec : SecurityClassification.FreeAccess;

    // ===== IReadable =====

    /// <inheritdoc />
    public Task<IReadOnlyList<DataValueSnapshot>> ReadAsync(
        IReadOnlyList<string> fullReferences, CancellationToken cancellationToken)
    {
        ObjectDisposedException.ThrowIf(_disposed, this);
        ArgumentNullException.ThrowIfNull(fullReferences);
        if (fullReferences.Count == 0) return Task.FromResult<IReadOnlyList<DataValueSnapshot>>([]);

        // v3: the server addresses by RawPath; translate to the Galaxy attributeRef MXAccess dials.
        // Read returns positionally, so no reverse translation is needed on the result. An unmapped
        // RawPath dials as-is (identity) — the gateway rejects an unknown address → Bad, same as today.
        var dialledRefs = new string[fullReferences.Count];
        for (var i = 0; i < fullReferences.Count; i++) dialledRefs[i] = ToAttributeRef(fullReferences[i]);

        if (_dataReader is not null)
        {
            // Test-only path — tests inject a canned reader via the internal ctor.
            return _dataReader.ReadAsync(dialledRefs, cancellationToken);
        }

        if (_subscriber is null)
        {
            throw new NotSupportedException(
                "GalaxyDriver.ReadAsync requires a connected GalaxyMxSession (production runtime not built). " +
                "Either inject a test seam via the internal ctor or call InitializeAsync against a real gateway.");
        }

        return ReadViaSubscribeOnceAsync(dialledRefs, cancellationToken);
    }

    /// <summary>
    ///     Production read path. MxAccess has no one-shot Read RPC — every value comes
    ///     through the event stream. We synthesise a Read by:
    ///     <list type="number">
    ///       <item>Subscribing the requested tags through the existing
    ///         <see cref="SubscriptionRegistry"/> + <see cref="EventPump"/>.</item>
    ///       <item>Waiting for the first <c>OnDataChange</c> per item handle (the gateway
    ///         pushes the current value as the initial event after a SubscribeBulk).</item>
    ///       <item>Unsubscribing.</item>
    ///     </list>
    ///     Tags the gw rejects at SubscribeBulk time, or that never publish before the
    ///     caller's cancellation token fires, return a Bad-status snapshot in input order
    ///     so the caller still sees one snapshot per requested reference.
    /// </summary>
    private async Task<IReadOnlyList<DataValueSnapshot>> ReadViaSubscribeOnceAsync(
        IReadOnlyList<string> fullReferences, CancellationToken cancellationToken)
    {
        var pump = EnsureEventPumpStarted();
        var subscriptionId = _subscriptions.NextSubscriptionId();

        // Pre-allocate one TaskCompletionSource per full-reference so the OnDataChange
        // handler can complete them out-of-order as events arrive. Wired BEFORE the
        // SubscribeBulk call so we don't race with the first event the gw pushes.
        var pendingByRef = new Dictionary<string, TaskCompletionSource<DataValueSnapshot>>(
            StringComparer.OrdinalIgnoreCase);
        foreach (var fullRef in fullReferences.Distinct(StringComparer.OrdinalIgnoreCase))
        {
            pendingByRef[fullRef] = new TaskCompletionSource<DataValueSnapshot>(
                TaskCreationOptions.RunContinuationsAsynchronously);
        }

        EventHandler<DataChangeEventArgs> handler = (_, args) =>
        {
            // Filter to OUR subscription — the pump's OnDataChange fans out across all
            // subscriptions on the driver, and we don't want a parallel ISubscribable
            // caller's events to leak into our read.
            if (args.SubscriptionHandle is GalaxySubscriptionHandle gsh
                && gsh.SubscriptionId == subscriptionId
                && pendingByRef.TryGetValue(args.FullReference, out var tcs))
            {
                tcs.TrySetResult(args.Snapshot);
            }
        };
        pump.OnDataChange += handler;

        var bufferedIntervalMs = _options.MxAccess.PublishingIntervalMs;
        IReadOnlyList<SubscribeResult> results;
        try
        {
            results = await _subscriber!
                .SubscribeBulkAsync(fullReferences, bufferedIntervalMs, cancellationToken)
                .ConfigureAwait(false);
        }
        catch
        {
            pump.OnDataChange -= handler;
            throw;
        }

        // Register bindings so the pump knows to dispatch events for these handles.
        // Index the SubscribeBulk results once and correlate to references in O(1)
        // instead of FirstOrDefault per element (O(n²) over the batch). On the
        // 50k-tag soak path this turns a 2.5G-comparison loop into a single
        // Dictionary build + linear scan.
        var resultIndex = BuildResultIndex(results);
        var bindings = new List<TagBinding>(fullReferences.Count);
        for (var i = 0; i < fullReferences.Count; i++)
        {
            var fullRef = fullReferences[i];
            var itemHandle = resultIndex.TryGetValue(fullRef, out var match) && match is { WasSuccessful: true }
                ? match.ItemHandle
                : 0;
            bindings.Add(new TagBinding(fullRef, itemHandle));

            // Tags the gw rejected up front — complete with Bad status now so the
            // wait below doesn't time out on them.
            if (itemHandle <= 0
                && pendingByRef.TryGetValue(fullRef, out var rejectedTcs))
            {
                rejectedTcs.TrySetResult(new DataValueSnapshot(
                    Value: null,
                    StatusCode: 0x80000000u, // Bad
                    SourceTimestampUtc: null,
                    ServerTimestampUtc: DateTime.UtcNow));
            }
        }
        _subscriptions.Register(subscriptionId, bindings);

        try
        {
            // Wait for every pending TCS to complete or the caller's CT to fire. When the
            // CT fires before all values arrive, fill the still-pending entries with a
            // Bad-status snapshot rather than throwing — Read semantics let callers see
            // partial results.
            using var registration = cancellationToken.Register(() =>
            {
                foreach (var tcs in pendingByRef.Values)
                {
                    tcs.TrySetResult(new DataValueSnapshot(
                        Value: null,
                        StatusCode: 0x800B0000u, // BadTimeout
                        SourceTimestampUtc: null,
                        ServerTimestampUtc: DateTime.UtcNow));
                }
            });

            var snapshots = new DataValueSnapshot[fullReferences.Count];
            for (var i = 0; i < fullReferences.Count; i++)
            {
                snapshots[i] = await pendingByRef[fullReferences[i]].Task.ConfigureAwait(false);
            }
            return snapshots;
        }
        finally
        {
            pump.OnDataChange -= handler;
            // Drop the bindings + unsubscribe the live handles. UnsubscribeBulkAsync's
            // failure isn't fatal — the registry is already cleared, so any straggling
            // event from the gw would be a no-op fan-out.
            _subscriptions.Remove(subscriptionId);
            var liveHandles = bindings.Where(b => b.ItemHandle > 0).Select(b => b.ItemHandle).ToArray();
            if (liveHandles.Length > 0)
            {
                try
                {
                    await _subscriber!.UnsubscribeBulkAsync(liveHandles, CancellationToken.None)
                        .ConfigureAwait(false);
                }
                catch (Exception ex)
                {
                    _logger.LogWarning(ex,
                        "GalaxyDriver.ReadViaSubscribeOnceAsync UnsubscribeBulk failed for {Count} handle(s) — registry already cleared.",
                        liveHandles.Length);
                }
            }
        }
    }

    // ===== IWritable (PR 4.3) =====

    /// <inheritdoc />
    public Task<IReadOnlyList<WriteResult>> WriteAsync(
        IReadOnlyList<WriteRequest> writes, CancellationToken cancellationToken)
    {
        ObjectDisposedException.ThrowIf(_disposed, this);
        ArgumentNullException.ThrowIfNull(writes);
        if (writes.Count == 0) return Task.FromResult<IReadOnlyList<WriteResult>>([]);

        if (_dataWriter is null)
        {
            // Mirror the IReadable fallback: production write path runs on top of
            // GatewayGalaxyDataWriter, which is wired by InitializeAsync when the gRPC session
            // connects. This guard fires only if the driver is called before initialization completes.
            throw new NotSupportedException(
                "GalaxyDriver.WriteAsync requires GatewayGalaxyDataWriter wired against a connected " +
                "GalaxyMxSession. Call InitializeAsync and wait for a successful connection before " +
                "issuing writes.");
        }

        // v3: WriteRequest.FullReference is a RawPath. Translate to the Galaxy attributeRef the writer
        // dials to MXAccess. Security was captured at discovery keyed by RawPath (= the node's FullName),
        // so the resolver the writer invokes with the DIALLED attributeRef must reverse it back to RawPath
        // before the lookup. Both legs are identity on a miss, so an un-authored tag behaves as before.
        var dialledWrites = new List<WriteRequest>(writes.Count);
        foreach (var w in writes) dialledWrites.Add(w with { FullReference = ToAttributeRef(w.FullReference) });

        return _dataWriter.WriteAsync(
            dialledWrites, attributeRef => ResolveSecurity(ToRawPath(attributeRef)), cancellationToken);
    }

    // ===== ISubscribable (PR 4.4) =====

    /// <inheritdoc />
    public async Task<ISubscriptionHandle> SubscribeAsync(
        IReadOnlyList<string> fullReferences, TimeSpan publishingInterval, CancellationToken cancellationToken)
    {
        ObjectDisposedException.ThrowIf(_disposed, this);
        ArgumentNullException.ThrowIfNull(fullReferences);

        if (_subscriber is null)
        {
            throw new NotSupportedException(
                "GalaxyDriver.SubscribeAsync requires a connected GalaxyMxSession + GatewayGalaxySubscriber. " +
                "Call InitializeAsync and wait for a successful connection before subscribing.");
        }

        var pump = EnsureEventPumpStarted();
        var subscriptionId = _subscriptions.NextSubscriptionId();

        if (fullReferences.Count == 0)
        {
            // Empty subscriptions register but never bind anything — keeps Unsubscribe
            // symmetric for callers that conditionally add tags later.
            _subscriptions.Register(subscriptionId, []);
            return new GalaxySubscriptionHandle(subscriptionId);
        }

        // v3: the server subscribes by RawPath; the registry + gateway operate on the Galaxy attributeRef.
        // Translate here (identity on a miss — a synthetic alarm sub-ref or live-browse tag dials straight
        // through). The value fan-out reverse-translates back to RawPath in OnPumpDataChange.
        var dialledRefs = new string[fullReferences.Count];
        for (var i = 0; i < fullReferences.Count; i++) dialledRefs[i] = ToAttributeRef(fullReferences[i]);

        // PR 6.3 — when the caller doesn't set a publishing interval (TimeSpan.Zero or
        // negative), fall back to the configured MxAccess.PublishingIntervalMs. The
        // server's UA subscription publishingInterval drives this in production; tests
        // and infrastructure callers (probe watcher, deploy watcher) hit the fallback.
        var requested = (int)Math.Max(0, publishingInterval.TotalMilliseconds);
        var bufferedIntervalMs = requested > 0 ? requested : _options.MxAccess.PublishingIntervalMs;
        var results = await _subscriber
            .SubscribeBulkAsync(dialledRefs, bufferedIntervalMs, cancellationToken)
            .ConfigureAwait(false);

        // Build the binding list in input order. Failed entries (gw rejected the tag) are
        // recorded with a non-positive ItemHandle so the caller can detect partial failure
        // by inspecting the returned handle's diagnostic context — full per-tag error
        // surface lands in PR 5.3's parity tests.
        // Index results once, correlate in O(1) per reference rather than
        // FirstOrDefault inside the loop (O(n²) on the 50k-tag path).
        // Bindings key on the dialled attributeRef — that's what the gw returns as the tag address and
        // what the EventPump fans out on. OnPumpDataChange maps it back to RawPath for public consumers.
        var resultIndex = BuildResultIndex(results);
        var bindings = new List<TagBinding>(dialledRefs.Length);
        for (var i = 0; i < dialledRefs.Length; i++)
        {
            var fullRef = dialledRefs[i];
            var hasMatch = resultIndex.TryGetValue(fullRef, out var match);
            var itemHandle = hasMatch && match is { WasSuccessful: true } ? match.ItemHandle : 0;
            bindings.Add(new TagBinding(fullRef, itemHandle));
            if (match is null || !match.WasSuccessful)
            {
                _logger.LogWarning(
                    "Galaxy subscribe for {FullRef} failed: {Error}",
                    fullRef, match?.ErrorMessage ?? "<no result returned>");
            }
        }

        _subscriptions.Register(subscriptionId, bindings);
        _ = pump; // keep the pump alive for the subscription's lifetime
        return new GalaxySubscriptionHandle(subscriptionId);
    }

    /// <inheritdoc />
    public async Task UnsubscribeAsync(ISubscriptionHandle handle, CancellationToken cancellationToken)
    {
        ObjectDisposedException.ThrowIf(_disposed, this);
        ArgumentNullException.ThrowIfNull(handle);
        if (handle is not GalaxySubscriptionHandle gsh)
        {
            throw new ArgumentException(
                $"Subscription handle was not issued by this driver (expected GalaxySubscriptionHandle, got {handle.GetType().Name}).",
                nameof(handle));
        }

        var bindings = _subscriptions.Remove(gsh.SubscriptionId);
        if (bindings is null) return; // already removed or never registered

        var liveItemHandles = bindings.Where(b => b.ItemHandle > 0).Select(b => b.ItemHandle).ToArray();
        if (liveItemHandles.Length == 0 || _subscriber is null) return;

        try
        {
            await _subscriber.UnsubscribeBulkAsync(liveItemHandles, cancellationToken).ConfigureAwait(false);
        }
        catch (Exception ex)
        {
            _logger.LogWarning(ex,
                "Galaxy UnsubscribeBulk failed for subscription {SubscriptionId} — registry already cleared on driver side.",
                gsh.SubscriptionId);
        }
    }

    /// <summary>
    ///     Lazily start the <see cref="EventPump"/> on the first subscribe. The pump is
    ///     shared across every subscription on this driver — fan-out happens through the
    ///     <see cref="SubscriptionRegistry"/> reverse map, not by spinning a pump per
    ///     subscription.
    /// </summary>
    private EventPump EnsureEventPumpStarted()
    {
        lock (_pumpLock)
        {
            if (_eventPump is not null) return _eventPump;
            _eventPump = new EventPump(
                _subscriber!, _subscriptions, _logger,
                channelCapacity: _options.MxAccess.EventPumpChannelCapacity,
                clientName: _options.MxAccess.ClientName,
                onStreamFault: OnEventPumpStreamFault);
            _eventPump.OnDataChange += OnPumpDataChange;
            _eventPump.Start();
            return _eventPump;
        }
    }

    /// <summary>
    ///     Stream-fault callback for the <see cref="EventPump"/>. The gw StreamEvents
    ///     stream faulted (transient gateway drop, network blip, gw restart). Forward
    ///     the cause to the <see cref="ReconnectSupervisor"/> so it drives reopen →
    ///     replay; without this hand-off a transient transport drop permanently kills
    ///     the event stream and <c>GetHealth()</c> keeps reporting Healthy.
    /// </summary>
    private void OnEventPumpStreamFault(Exception cause)
    {
        var supervisor = _supervisor;
        if (supervisor is null)
        {
            // No production runtime (skeleton / injected-seam path) — nothing to drive.
            _logger.LogWarning(cause,
                "GalaxyDriver {InstanceId} event stream faulted but no reconnect supervisor is wired.",
                _driverInstanceId);
            return;
        }

        try
        {
            supervisor.ReportTransportFailure(cause);
        }
        catch (ObjectDisposedException)
        {
            // Driver is being disposed — the stream fault is just shutdown noise.
        }
    }

    // ===== IAlarmSource =====

    /// <summary>
    ///     Start the gateway alarm feed (idempotent) and wire its transitions onto this
    ///     driver's <see cref="OnAlarmEvent"/> bridge. The feed is session-less — it does
    ///     not depend on a data subscription or the <see cref="EventPump"/>.
    /// </summary>
    private void EnsureAlarmFeedStarted()
    {
        lock (_alarmFeedLock)
        {
            if (_alarmFeed is null)
            {
                throw new InvalidOperationException(
                    "GalaxyDriver alarm feed is not wired. InitializeAsync must run (or a feed " +
                    "seam must be injected via the internal ctor) before subscribing to alarms.");
            }
            if (_alarmFeedWired) return;
            _alarmFeed.OnAlarmTransition += OnAlarmFeedTransition;
            _alarmFeed.Start();
            _alarmFeedWired = true;
        }
    }

    /// <inheritdoc />
    public Task<IAlarmSubscriptionHandle> SubscribeAlarmsAsync(
        IReadOnlyList<string> sourceNodeIds, CancellationToken cancellationToken)
    {
        ObjectDisposedException.ThrowIf(_disposed, this);
        ArgumentNullException.ThrowIfNull(sourceNodeIds);

        // The driver doesn't multiplex alarm subscriptions per source-node-id today —
        // every active handle receives every transition off the gateway's session-less
        // StreamAlarms feed, and the server filters by source node before raising Part 9
        // conditions. The subscription handle is a sentinel the server uses for
        // symmetric Unsubscribe. Same shape AbCip uses.
        EnsureAlarmFeedStarted();
        var handle = new GalaxyAlarmSubscriptionHandle(Guid.NewGuid().ToString("N"));
        lock (_alarmHandlersLock)
        {
            // Collapse to a SINGLE live handle. The session-less StreamAlarms feed survives an in-place
            // reconnect, so DriverInstanceActor re-subscribing on every Connected re-entry (after dropping
            // its own cached handle) would otherwise grow this list by one untracked handle per reconnect —
            // a slow leak (the gate only checks Count > 0 and OnAlarmFeedTransition only reads [0], so the
            // extras were pure dead weight). There is exactly one consumer per driver instance — enforced
            // structurally by the factory-per-actor lifecycle (one GalaxyDriver is created for one owning
            // DriverInstanceActor, the only caller of SubscribeAlarmsAsync), not by mere convention — so
            // replacing the set with the latest handle is semantically faithful and bounds it at 1.
            _alarmSubscriptions.Clear();
            _alarmSubscriptions.Add(handle);
        }
        return Task.FromResult<IAlarmSubscriptionHandle>(handle);
    }

    /// <inheritdoc />
    public Task UnsubscribeAlarmsAsync(IAlarmSubscriptionHandle handle, CancellationToken cancellationToken)
    {
        ObjectDisposedException.ThrowIf(_disposed, this);
        ArgumentNullException.ThrowIfNull(handle);
        if (handle is not GalaxyAlarmSubscriptionHandle gash)
        {
            throw new ArgumentException(
                $"Subscription handle was not issued by this driver (expected GalaxyAlarmSubscriptionHandle, got {handle.GetType().Name}).",
                nameof(handle));
        }
        lock (_alarmHandlersLock)
        {
            _alarmSubscriptions.Remove(gash);
        }
        return Task.CompletedTask;
    }

    /// <inheritdoc />
    public async Task AcknowledgeAsync(
        IReadOnlyList<AlarmAcknowledgeRequest> acknowledgements, CancellationToken cancellationToken)
    {
        ObjectDisposedException.ThrowIf(_disposed, this);
        ArgumentNullException.ThrowIfNull(acknowledgements);
        if (acknowledgements.Count == 0) return;

        if (_alarmAcknowledger is null)
        {
            throw new NotSupportedException(
                "GalaxyDriver.AcknowledgeAsync requires GatewayGalaxyAlarmAcknowledger wired against a connected " +
                "GalaxyMxSession. InitializeAsync must complete successfully before alarm acknowledgements can flow.");
        }

        // Acks are issued one-by-one — the gateway RPC accepts a single alarm
        // reference per call. AlarmConditionState's per-condition Acknowledge in the
        // server-side ACL layer is the natural rate-limit, so issuing in series here
        // keeps the operator-comment ordering deterministic without bursting the
        // worker's STA queue.
        foreach (var ack in acknowledgements)
        {
            // ConditionId carries the alarm's RawPath (v3) for the Galaxy driver — SourceNodeId is the
            // OPC UA browse path, which the gateway can't address. Translate the RawPath back to the
            // Galaxy attributeRef the gateway dials (identity on a miss). The server-side condition state
            // pairs them through AlarmConditionService.
            var conditionReference = !string.IsNullOrEmpty(ack.ConditionId)
                ? ack.ConditionId
                : ack.SourceNodeId;
            var alarmFullReference = ToAttributeRef(conditionReference);
            await _alarmAcknowledger.AcknowledgeAsync(
                alarmFullReference,
                ack.Comment ?? string.Empty,
                // Honor the authenticated principal carried on the request when present;
                // fall back to empty (server-side ACL fills this from the OPC UA session).
                operatorUser: ack.OperatorUser ?? string.Empty,
                cancellationToken).ConfigureAwait(false);
        }
    }

    /// <summary>
    ///     Receives <see cref="GalaxyAlarmTransition"/> events from the gateway alarm
    ///     feed and reshapes them into <see cref="AlarmEventArgs"/> for OPC UA-side
    ///     consumers. Fires <see cref="OnAlarmEvent"/> only when at least one alarm
    ///     subscription is active so a server that hasn't called
    ///     <see cref="SubscribeAlarmsAsync"/> yet doesn't surface untracked transitions.
    /// </summary>
    private void OnAlarmFeedTransition(object? sender, GalaxyAlarmTransition transition)
    {
        GalaxyAlarmSubscriptionHandle? handle;
        int subCount;
        lock (_alarmHandlersLock)
        {
            // Pick the earliest-registered handle as the event owner. The server routes
            // by SourceNodeId (not by handle), so every active subscriber sees the same
            // transition regardless of which handle is attached here. Using the first
            // insertion-order entry is deterministic and stable as long as at least one
            // subscription remains — HashSet.First() was unstable across mutations.
            // _alarmSubscriptions is a List, so [0] is always the earliest-registered
            // handle.
            subCount = _alarmSubscriptions.Count;
            handle = subCount > 0 ? _alarmSubscriptions[0] : null;
        }
        // Native-alarm delivery observability (was a black box): one line per feed transition with the
        // resolved alarm ref, the live subscription count, and whether it fans out to OnAlarmEvent. Debug
        // so a flapping galaxy doesn't flood prod, but available on demand to answer "is the feed
        // delivering / is the subscription un-gating it" — the exact question this driver couldn't answer
        // before the SubscribeAlarmsAsync wiring fix.
        _logger.LogDebug(
            "GalaxyDriver alarm feed: {Kind} {Ref} (subs={Subs}, fanout={Fanout})",
            transition.TransitionKind, transition.AlarmFullReference, subCount, handle is not null);
        if (handle is null) return;

        // v3: the ConditionId the driver reports must be the tag's RawPath — the same key the value
        // fan-out uses (OnPumpDataChange) and the identity the discoverer stamped as SourceName — so the
        // server-side condition state correlates the alarm to its node and a subsequent ack round-trips
        // back through AcknowledgeAsync. The feed reports the Galaxy attributeRef; reverse-map it (identity
        // on a miss).
        var conditionRawPath = ToRawPath(transition.AlarmFullReference);
        var args = new AlarmEventArgs(
            SubscriptionHandle: handle,
            SourceNodeId: transition.SourceObjectReference,
            ConditionId: conditionRawPath,
            AlarmType: transition.AlarmTypeName,
            Message: transition.Description,
            Severity: transition.SeverityBucket,
            SourceTimestampUtc: transition.TransitionTimestampUtc,
            OperatorComment: string.IsNullOrEmpty(transition.OperatorComment) ? null : transition.OperatorComment,
            OriginalRaiseTimestampUtc: transition.OriginalRaiseTimestampUtc,
            AlarmCategory: string.IsNullOrEmpty(transition.Category) ? null : transition.Category,
            // Fully-qualify the Core.Abstractions enum: this file also imports
            // MxGateway.Contracts.Proto, which defines a same-named AlarmTransitionKind.
            Kind: transition.TransitionKind switch
            {
                GalaxyAlarmTransitionKind.Raise => Core.Abstractions.AlarmTransitionKind.Raise,
                GalaxyAlarmTransitionKind.Acknowledge => Core.Abstractions.AlarmTransitionKind.Acknowledge,
                GalaxyAlarmTransitionKind.Clear => Core.Abstractions.AlarmTransitionKind.Clear,
                GalaxyAlarmTransitionKind.Retrigger => Core.Abstractions.AlarmTransitionKind.Retrigger,
                _ => Core.Abstractions.AlarmTransitionKind.Unspecified,
            });
        try
        {
            OnAlarmEvent?.Invoke(this, args);
        }
        catch (Exception ex)
        {
            _logger.LogWarning(ex,
                "GalaxyDriver OnAlarmEvent handler threw for {AlarmRef} — continuing.",
                transition.AlarmFullReference);
        }
    }

    /// <summary>
    ///     Forwards every fan-out event to the public <see cref="OnDataChange"/> for
    ///     ISubscribable consumers, AND routes ScanState changes to the per-platform
    ///     probe watcher (PR 4.7) so platform health entries update without the watcher
    ///     consuming the event stream itself.
    /// </summary>
    private void OnPumpDataChange(object? sender, DataChangeEventArgs args)
    {
        // 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.
        var publicArgs = args;
        var rawPath = ToRawPath(args.FullReference);
        if (!string.Equals(rawPath, args.FullReference, StringComparison.Ordinal))
        {
            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>
    ///     Lazily builds the default <see cref="IGalaxyHierarchySource"/> from
    ///     <c>_options.Gateway</c>. Owned <see cref="GalaxyRepositoryClient"/> is disposed in
    ///     <see cref="Dispose"/>. Tests bypass this by injecting their own source via the
    ///     internal ctor.
    /// </summary>
    private IGalaxyHierarchySource BuildDefaultHierarchySource()
    {
        // Reuse a client that StartDeployWatcher may have already created (??=) rather
        // than always overwriting the field and leaking the first instance. Both paths
        // produce equivalent clients from the same options.
        _ownedRepositoryClient ??= GalaxyRepositoryClient.Create(BuildClientOptions(_options.Gateway));
        return new TracedGalaxyHierarchySource(
            new GatewayGalaxyHierarchySource(_ownedRepositoryClient), _options.MxAccess.ClientName);
    }

    /// <summary>
    ///     Asynchronous disposal. Prefer <c>await using</c> over <c>using</c> — the
    ///     async path does not block the caller while awaiting EventPump / session /
    ///     client shutdown (the sync path blocks on <c>GetAwaiter().GetResult()</c>
    ///     for every async sub-component, risking a deadlock under thread-pool
    ///     starvation).
    /// </summary>
    /// <returns>A task that represents the asynchronous operation.</returns>
    public async ValueTask DisposeAsync()
    {
        if (_disposed) return;
        _disposed = true;

        // Synchronous sub-components first — none of these block.
        try { _deployWatcher?.Dispose(); } catch (Exception ex) { _logger.LogWarning(ex, "DeployWatcher dispose failed"); }
        try { _supervisor?.Dispose(); } catch (Exception ex) { _logger.LogWarning(ex, "ReconnectSupervisor dispose failed"); }
        try { _probeWatcher?.Dispose(); } catch (Exception ex) { _logger.LogWarning(ex, "ProbeWatcher dispose failed"); }
        try { _transportForwarder?.Dispose(); } catch (Exception ex) { _logger.LogWarning(ex, "Transport forwarder dispose failed"); }

        // Async sub-components: await each so we don't block a thread-pool thread
        // on a slow shutdown (e.g. EventPump draining its channel, gRPC stream closing).
        EventPump? pump;
        lock (_pumpLock) { pump = _eventPump; _eventPump = null; }
        if (pump is not null)
        {
            try { await pump.DisposeAsync().ConfigureAwait(false); }
            catch (Exception ex) { _logger.LogWarning(ex, "EventPump dispose failed"); }
        }

        IGalaxyAlarmFeed? alarmFeed;
        lock (_alarmFeedLock) { alarmFeed = _alarmFeed; _alarmFeed = null; }
        if (alarmFeed is not null)
        {
            try { await alarmFeed.DisposeAsync().ConfigureAwait(false); }
            catch (Exception ex) { _logger.LogWarning(ex, "Alarm feed dispose failed"); }
        }

        if (_ownedMxSession is not null)
        {
            try { await _ownedMxSession.DisposeAsync().ConfigureAwait(false); }
            catch (Exception ex) { _logger.LogWarning(ex, "MxSession dispose failed"); }
            _ownedMxSession = null;
        }

        if (_ownedMxClient is not null)
        {
            try { await _ownedMxClient.DisposeAsync().ConfigureAwait(false); }
            catch (Exception ex) { _logger.LogWarning(ex, "MxClient dispose failed"); }
            _ownedMxClient = null;
        }

        if (_ownedRepositoryClient is not null)
        {
            try { await _ownedRepositoryClient.DisposeAsync().ConfigureAwait(false); }
            catch (Exception ex) { _logger.LogWarning(ex, "RepositoryClient dispose failed"); }
            _ownedRepositoryClient = null;
        }

        _hierarchySource = null;
    }

    /// <summary>
    ///     Synchronous disposal. Prefer <see cref="DisposeAsync"/> in async contexts —
    ///     this path must block on every async sub-component shutdown. Provided for
    ///     compatibility with <c>using</c> statements that cannot <c>await</c>.
    /// </summary>
    public void Dispose() => DisposeAsync().AsTask().GetAwaiter().GetResult();

    /// <summary>
    ///     Address-space builder wrapper that records each variable's
    ///     <see cref="DriverAttributeInfo.SecurityClass"/> into the supplied dictionary
    ///     before delegating to the inner builder. Used by <see cref="DiscoverAsync"/>
    ///     to capture per-tag classifications for the IWritable routing decision —
    ///     PR 4.3 needs the data, but the discoverer itself doesn't (and shouldn't)
    ///     know about the driver's internal state.
    /// </summary>
    private sealed class SecurityCapturingBuilder(
        IAddressSpaceBuilder inner,
        System.Collections.Concurrent.ConcurrentDictionary<string, SecurityClassification> map)
        : IAddressSpaceBuilder
    {
        /// <inheritdoc />
        public IAddressSpaceBuilder Folder(string browseName, string displayName)
            => new SecurityCapturingBuilder(inner.Folder(browseName, displayName), map);

        /// <inheritdoc />
        public IVariableHandle Variable(string browseName, string displayName, DriverAttributeInfo attributeInfo)
        {
            map[attributeInfo.FullName] = attributeInfo.SecurityClass;
            return inner.Variable(browseName, displayName, attributeInfo);
        }

        /// <inheritdoc />
        public void AddProperty(string browseName, DriverDataType dataType, object? value)
            => inner.AddProperty(browseName, dataType, value);
    }
}