lmxopcua/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt/MqttConnection.cs

using System.Buffers;
using System.Net.Security;
using System.Security.Cryptography;
using System.Security.Cryptography.X509Certificates;
using Microsoft.Extensions.Logging;
using MQTTnet;
using MQTTnet.Protocol;

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

/// <summary>
///     Receives one inbound MQTT application message. Invoked on <b>MQTTnet's own dispatcher
///     thread</b>: an implementation must not block, must not do I/O and must not throw, or it stalls
///     the client's pump for every other subscription.
/// </summary>
/// <param name="topic">The concrete topic the message arrived on (never a filter).</param>
/// <param name="payload">
///     The message body. Only valid for the duration of the call — the underlying buffer belongs to
///     MQTTnet, so anything an implementation wants to keep must be copied out.
/// </param>
/// <param name="retained">
///     The message's MQTT <c>retain</c> flag as delivered to <i>this</i> client, i.e. <c>true</c> for
///     the broker's stored last-known value replayed at subscribe time, and <c>false</c> for an
///     ordinary live publish. This is the retained <b>seed</b> signal.
/// </param>
public delegate void MqttMessageObserver(string topic, ReadOnlySpan<byte> payload, bool retained);

/// <summary>One topic filter to SUBSCRIBE, as the subscription manager wants it established.</summary>
/// <param name="Topic">The topic filter (concrete, or carrying MQTT wildcards).</param>
/// <param name="Qos">Requested QoS, 0–2.</param>
/// <param name="SeedRetained">
///     Whether the broker should replay its retained message for this filter at subscribe time.
///     Honoured natively on MQTT 5.0 (retain handling); on 3.1.1 the broker always replays and the
///     manager drops the seed client-side instead.
/// </param>
public sealed record MqttTopicSubscription(string Topic, int Qos, bool SeedRetained);

/// <summary>The broker's SUBACK verdict for one requested filter.</summary>
/// <param name="Topic">The filter this outcome answers.</param>
/// <param name="Granted">Whether the broker granted the subscription.</param>
/// <param name="Reason">The broker's reason code / string, for logs and diagnostics.</param>
public sealed record MqttSubscribeOutcome(string Topic, bool Granted, string Reason);

/// <summary>
///     The one seam through which the subscription manager establishes MQTT subscriptions.
///     <see cref="MqttConnection"/> is the production implementation; the interface exists so the
///     manager's SUBACK handling (per-filter grant / rejection, total failure) is exercisable without
///     a broker.
/// </summary>
public interface IMqttSubscribeTransport
{
    /// <summary>
    ///     Issues one SUBSCRIBE carrying every filter and returns the broker's per-filter verdict.
    ///     Implementations must be bounded — a broker that accepts SUBSCRIBE and never SUBACKs must
    ///     fail at a deadline, not hang.
    /// </summary>
    /// <param name="filters">The filters to establish; never empty.</param>
    /// <param name="cancellationToken">Caller cancellation, linked with the implementation's deadline.</param>
    /// <returns>One outcome per requested filter.</returns>
    Task<IReadOnlyList<MqttSubscribeOutcome>> SubscribeAsync(
        IReadOnlyList<MqttTopicSubscription> filters,
        CancellationToken cancellationToken);
}

/// <summary>Lifecycle state of an <see cref="MqttConnection"/>.</summary>
public enum MqttConnectionState
{
    /// <summary>Constructed, or the first connect has not yet succeeded. No supervisor is running.</summary>
    Disconnected = 0,

    /// <summary>An MQTT session is established.</summary>
    Connected = 1,

    /// <summary>
    ///     The session dropped and the supervisor is retrying under backoff. A broker that is merely
    ///     down stays here <b>indefinitely</b> — being unreachable is never a fault.
    /// </summary>
    Reconnecting = 2,

    /// <summary>
    ///     Unrecoverable without a driver restart: retrying cannot help, so the supervisor has
    ///     stopped. Two causes — unusable configuration (a failure raised while assembling the client
    ///     options, before any I/O), or the supervisor itself dying on an unexpected exception. A
    ///     broker that is merely unreachable never lands here.
    /// </summary>
    Faulted = 3,

    /// <summary><see cref="MqttConnection.DisposeAsync"/> has run. Terminal.</summary>
    Disposed = 4,
}

/// <summary>
///     The broker completed the handshake and <b>refused</b> the CONNECT, returning a non-success
///     CONNACK reason code.
/// </summary>
/// <remarks>
///     <para>
///         This exists because <b>MQTTnet 5 does not throw on an unsuccessful CONNACK</b> — it returns
///         the outcome in <see cref="MqttClientConnectResult.ResultCode"/> and leaves the client
///         disconnected, and v5 removed v4's <c>ThrowOnNonSuccessfulConnectResponse</c> option, so
///         inspecting the result is the only way to see it. A caller that ignores the result gets a
///         connect that "succeeded" against a broker which rejected it: a driver configured with the
///         wrong broker password would report <c>Healthy</c> and never receive a single value. That is
///         exactly what the Task-13 live fixture caught.
///     </para>
///     <para>
///         <see cref="IsUnrecoverable"/> distinguishes a rejection no amount of retrying can fix
///         (credentials, identity, protocol, an invalid Last-Will) from a genuinely transient refusal
///         (broker unavailable/busy, quota, rate limit). See
///         <see cref="MqttConnection.IsUnrecoverableConnackRejection"/> for the full mapping.
///     </para>
/// </remarks>
public sealed class MqttConnectRejectedException : Exception
{
    internal MqttConnectRejectedException(MqttClientConnectResultCode resultCode, bool isUnrecoverable, string message)
        : base(message)
    {
        ResultCode = resultCode;
        IsUnrecoverable = isUnrecoverable;
    }

    /// <summary>Whether retrying against this broker with this configuration can never succeed.</summary>
    public bool IsUnrecoverable { get; }

    /// <summary>The CONNACK reason code the broker returned.</summary>
    public MqttClientConnectResultCode ResultCode { get; }
}

/// <summary>
///     Owns the driver's single live MQTTnet-5 client: assembles the broker connection options
///     from <see cref="MqttDriverOptions"/> (endpoint, protocol version, session, credentials,
///     TLS + CA pin), connects under a bounded deadline, and keeps the session alive with a
///     hand-rolled reconnect supervisor.
/// </summary>
/// <remarks>
///     <para>
///         <b>Connection-free construction.</b> The constructor stores configuration only — it
///         opens no socket, creates no client and starts no supervisor. The universal-browser
///         <c>CanBrowse</c> pattern constructs a throwaway instance purely to ask what driver type
///         it is, so a constructor that dialled a broker would stall the AdminUI.
///     </para>
///     <para>
///         <b>Bounded connect.</b> <see cref="ConnectAsync"/> links the caller's token with a
///         <see cref="MqttDriverOptions.ConnectTimeoutSeconds"/> deadline, and the same value is
///         fed to MQTTnet's own <see cref="MqttClientOptions.Timeout"/>. A broker that accepts the
///         TCP connection and then never answers CONNACK — the frozen-peer shape that wedged the
///         S7 read path (arch-review R2-01) — fails at the deadline instead of hanging forever.
///         Every wait in this type is bounded the same way, teardown included.
///     </para>
///     <para>
///         <b>Hand-rolled reconnect.</b> MQTTnet v5 dropped v4's <c>ManagedMqttClient</c>, so there
///         is no library-managed reconnect: this type supplies it. The supervisor starts on the
///         first <i>successful</i> connect (a failed first attempt is the caller's to retry — the
///         driver-host resilience layer re-runs <c>InitializeAsync</c> — and must not leave a
///         background task hammering an endpoint the caller has given up on). MQTTnet's
///         <c>DisconnectedAsync</c> callback only signals a semaphore and returns, so the library's
///         own dispatcher thread is never blocked by a backoff sleep.
///     </para>
///     <para>
///         <b><see cref="Reconnected"/> is load-bearing.</b> MQTT subscriptions do not survive a
///         clean session, and a reconnect that completes without re-subscribing produces a
///         healthy-looking connection that receives nothing, forever, with no error, no exception
///         and no bad status. So the callback fires on <b>every</b> successful reconnect — including
///         persistent-session reconnects, where re-subscribing is merely redundant. Re-subscribing
///         an already-subscribed topic is harmless; missing one is silent death. If the callback
///         fails, the freshly-established session is torn down and retried rather than left
///         connected-but-deaf. It is invoked under a cancellation token and a
///         <see cref="MqttDriverOptions.ConnectTimeoutSeconds"/> ceiling: a subscriber that hangs on
///         a broker which accepts SUBSCRIBE and never SUBACKs — the frozen-peer shape again — must
///         not be able to park the supervisor or outlive <see cref="DisposeAsync"/>.
///     </para>
///     <para>
///         <b>Connect is idempotent.</b> MQTTnet throws (and raises no <c>DisconnectedAsync</c>) if
///         asked to connect a client that is already connected, so both connect paths check first and
///         return a no-op. The two paths genuinely race: the supervisor is told to expect the driver
///         host to re-run <c>InitializeAsync</c>, so a caller can restore the session while the
///         supervisor sits in backoff, and the supervisor can restore it a moment before a caller
///         asks. Without the check, the first case leaves the supervisor failing forever against a
///         perfectly healthy connection (inflating <c>attempt</c>, so the <i>next</i> genuine outage
///         waits <see cref="MqttDriverOptions.ReconnectMaxBackoffSeconds"/> instead of min) and the
///         second throws an undocumented <see cref="InvalidOperationException"/> out of
///         <see cref="ConnectAsync"/> on a working session. When the supervisor finds the session
///         already restored it stops retrying <b>without</b> firing <see cref="Reconnected"/> — the
///         caller that reconnected owns its own subscribe, per the <see cref="ConnectAsync"/>
///         contract.
///     </para>
///     <para>
///         <b>State transitions.</b> <see cref="MqttConnectionState.Disposed"/> is terminal; nothing
///         moves off it.
///         <list type="table">
///             <item>
///                 <term><see cref="MqttConnectionState.Connected"/></term>
///                 <description>
///                     set by <see cref="ConnectAsync"/> on success (or on finding the session
///                     already up), and by the supervisor <i>only after</i>
///                     <see cref="Reconnected"/> has completed — so on the reconnect path
///                     <c>Connected</c> means "connected <i>and</i> re-subscribed".
///                 </description>
///             </item>
///             <item>
///                 <term><see cref="MqttConnectionState.Reconnecting"/></term>
///                 <description>
///                     set by the <c>DisconnectedAsync</c> callback when a live session drops, and
///                     held by the supervisor for the whole retry + re-subscribe sequence.
///                 </description>
///             </item>
///             <item>
///                 <term><see cref="MqttConnectionState.Faulted"/></term>
///                 <description>
///                     set when the client options cannot be assembled, or when the supervisor exits
///                     on an unexpected exception. Both mean no further recovery without a restart.
///                 </description>
///             </item>
///             <item>
///                 <term><see cref="MqttConnectionState.Disposed"/></term>
///                 <description>set first thing in <see cref="DisposeAsync"/>.</description>
///             </item>
///         </list>
///         <see cref="State"/> and <see cref="IsConnected"/> can legitimately disagree in <b>both</b>
///         directions and are not interchangeable: <c>Reconnecting</c> with
///         <c>IsConnected == true</c> is the window where the socket is up but the re-subscribe has
///         not finished, and <c>Connected</c> with <c>IsConnected == false</c> is the instant between
///         the socket dropping and MQTTnet raising its callback. <see cref="State"/> is the health
///         surface; <see cref="IsConnected"/> is the transport fact.
///     </para>
///     <para>
///         <b>Credentials never leak.</b> Nothing here logs or formats
///         <see cref="MqttDriverOptions.Password"/>; the options record itself redacts it in
///         <c>ToString()</c>.
///     </para>
///     <para>
///         <b>Lifecycle is serialised.</b> Connect (caller-initiated or supervisor-initiated) and
///         dispose run under a single gate, and a connect that completes re-checks the disposed flag
///         <i>while still holding that gate</i>. This closes the connection leak the Task-3 review
///         traced: dispose could observe a still-null client, dispose nothing and return, after
///         which the in-flight connect assigned a client and connected successfully — leaving a live
///         socket no later dispose could ever reach. Now the losing side of that race disposes the
///         client it created and reports <see cref="ObjectDisposedException"/>.
///     </para>
///     Sparkplug rebirth-on-reconnect (Task 21, which hangs off <see cref="Reconnected"/>) is
///     deliberately not implemented here.
/// </remarks>
public sealed class MqttConnection : IAsyncDisposable, IMqttSubscribeTransport
{
    private readonly string _driverId;

    /// <summary>
    ///     Serialises connect (both callers of it) against dispose. Never disposed: a late MQTTnet
    ///     callback can still reach this instance after teardown, and a disposed
    ///     <see cref="SemaphoreSlim"/> would throw inside the library's own dispatcher.
    /// </summary>
    private readonly SemaphoreSlim _lifecycleGate = new(1, 1);

    /// <summary>
    ///     Cancelled by <see cref="DisposeAsync"/>; aborts the supervisor, any in-flight connect and
    ///     any in-flight <see cref="Reconnected"/> callback. Never disposed — deliberately, like the
    ///     two semaphores: a connect racing dispose reads <see cref="CancellationTokenSource.Token"/>
    ///     after its own disposed check, and disposing the source would turn that benign loser into an
    ///     <see cref="ObjectDisposedException"/> naming the wrong type. It holds no timer and no
    ///     surviving registrations (every linked source here is <c>using</c>-scoped), so leaving it
    ///     undisposed costs nothing.
    /// </summary>
    private readonly CancellationTokenSource _lifetimeCts = new();

    private readonly ILogger? _logger;
    private readonly MqttDriverOptions _options;

    /// <summary>
    ///     Wakes the supervisor. Unbounded max count: spurious releases (a failed connect attempt
    ///     also raises <c>DisconnectedAsync</c>) are drained harmlessly by the still-connected guard,
    ///     whereas a bounded semaphore would throw on the extra release. Never disposed, for the same
    ///     reason as <see cref="_lifecycleGate"/>.
    /// </summary>
    private readonly SemaphoreSlim _reconnectWake = new(0);

    private IMqttClient? _client;
    private int _disposed;
    private long _lastMessageTicksUtc;
    private Task? _reconnectWorker;
    private int _state = (int)MqttConnectionState.Disconnected;

    /// <summary>Stores configuration only — no network, no client instantiation, no supervisor.</summary>
    /// <param name="options">Broker connection settings.</param>
    /// <param name="driverId">Driver instance id, used only for log/diagnostic correlation.</param>
    /// <param name="logger">Optional logger; never receives credentials.</param>
    public MqttConnection(MqttDriverOptions options, string driverId, ILogger? logger = null)
    {
        ArgumentNullException.ThrowIfNull(options);
        ArgumentException.ThrowIfNullOrWhiteSpace(driverId);

        _options = options;
        _driverId = driverId;
        _logger = logger;
    }

    /// <summary>
    ///     Raised after every successful <i>re</i>connect (never after the initial
    ///     <see cref="ConnectAsync"/>, which the caller already sequences its own subscribe behind,
    ///     and never when the supervisor merely finds the session already restored by such a caller).
    ///     Subscribers must re-establish their MQTT subscriptions here — see the type remarks for
    ///     why skipping it is silent death. Every subscriber is invoked even if an earlier one
    ///     throws; a throwing subscriber causes the session to be torn down and retried.
    /// </summary>
    /// <remarks>
    ///     The supplied token is cancelled by <see cref="DisposeAsync"/>, and the whole fan-out is
    ///     additionally capped at <see cref="MqttDriverOptions.ConnectTimeoutSeconds"/> — a subscriber
    ///     that ignores its token cannot park the supervisor or make it outlive teardown. Subscribers
    ///     should flow the token into their own network calls rather than relying on that ceiling,
    ///     which abandons rather than stops the offending work.
    /// </remarks>
    public event Func<CancellationToken, Task>? Reconnected;

    /// <summary>
    ///     Raised for every inbound application message, on MQTTnet's own dispatcher thread. This
    ///     connection does no routing, parsing or typing of its own — that is the subscription
    ///     manager's job; here the message is only stamped onto <see cref="LastMessageUtc"/> and
    ///     handed on.
    /// </summary>
    /// <remarks>
    ///     Handlers must not block, do I/O or throw. A throwing handler is caught and logged here
    ///     rather than being allowed to escape into the library's pump — one bad tag must not stop
    ///     delivery for every other subscriber.
    /// </remarks>
    public event MqttMessageObserver? MessageReceived;

    /// <summary>Whether the underlying client currently holds an established MQTT session.</summary>
    public bool IsConnected => _client?.IsConnected ?? false;

    /// <summary>
    ///     UTC timestamp of the most recent inbound application message, or <c>null</c> if none has
    ///     arrived on this instance. Connection health only — nothing here routes or parses the
    ///     message; that is the subscription manager's job (Task 6).
    /// </summary>
    /// <remarks>
    ///     The plan calls this member <c>LastMessageAgeUtc</c>, which conflates a timestamp with an
    ///     age; it is split here into the instant (<see cref="LastMessageUtc"/>) and the elapsed span
    ///     (<see cref="LastMessageAge"/>).
    /// </remarks>
    public DateTime? LastMessageUtc
    {
        get
        {
            var ticks = Interlocked.Read(ref _lastMessageTicksUtc);
            return ticks == 0 ? null : new DateTime(ticks, DateTimeKind.Utc);
        }
    }

    /// <summary>
    ///     How long ago the most recent inbound application message arrived, or <c>null</c> if none
    ///     has. A connection that is <see cref="MqttConnectionState.Connected"/> with a large age is
    ///     the "reconnected but never re-subscribed" shape.
    /// </summary>
    public TimeSpan? LastMessageAge => LastMessageUtc is { } at ? DateTime.UtcNow - at : null;

    /// <summary>Current lifecycle state. See <see cref="MqttConnectionState"/>.</summary>
    public MqttConnectionState State => (MqttConnectionState)Volatile.Read(ref _state);

    /// <summary>
    ///     Test seam: awaited inside the gated connect immediately after the broker session is
    ///     established and <b>before</b> the disposed re-check, so a test can reproduce the exact
    ///     connect/dispose interleaving that used to leak a live connection. Always <c>null</c> in
    ///     production.
    /// </summary>
    internal Func<Task>? AfterConnectHookForTests { get; set; }

    private bool Disposed => Volatile.Read(ref _disposed) != 0;

    /// <inheritdoc/>
    public async ValueTask DisposeAsync()
    {
        if (Interlocked.Exchange(ref _disposed, 1) != 0)
        {
            return;
        }

        SetState(MqttConnectionState.Disposed);

        // Stop the supervisor and abort any in-flight connect FIRST, so the gate below is not held
        // for a full connect deadline by work that is already pointless.
        try
        {
            await _lifetimeCts.CancelAsync().ConfigureAwait(false);
        }
        catch (ObjectDisposedException)
        {
            // Nothing to cancel.
        }

        var worker = Volatile.Read(ref _reconnectWorker);
        if (worker is not null)
        {
            try
            {
                await worker.WaitAsync(TimeSpan.FromSeconds(_options.ConnectTimeoutSeconds)).ConfigureAwait(false);
            }
            catch (Exception ex)
            {
                _logger?.LogDebug(ex, "MQTT driver '{DriverId}': reconnect supervisor did not stop cleanly.", _driverId);
            }
        }

        // Bounded even here: a wedged connect must not turn dispose into a hang. If the gate cannot
        // be taken we still claim the client — double-disposing an MqttClient is harmless, leaking a
        // live one is not.
        var gated = await _lifecycleGate
            .WaitAsync(TimeSpan.FromSeconds(_options.ConnectTimeoutSeconds))
            .ConfigureAwait(false);
        if (!gated)
        {
            _logger?.LogWarning(
                "MQTT driver '{DriverId}': lifecycle gate still held at dispose; tearing the client down anyway.",
                _driverId);
        }

        try
        {
            var client = Interlocked.Exchange(ref _client, null);
            if (client is not null)
            {
                await CloseAsync(client).ConfigureAwait(false);
            }
        }
        finally
        {
            if (gated)
            {
                _lifecycleGate.Release();
            }
        }
    }

    /// <summary>
    ///     The backoff a supervisor waits before reconnect attempt <paramref name="attempt"/>:
    ///     <c>minSeconds × 2^attempt</c>, saturated at <paramref name="maxSeconds"/>. Pure — no
    ///     clock, no state, no jitter (deliberately: the schedule is asserted by exact equality, and
    ///     with one connection per driver instance there is no thundering herd to spread out).
    /// </summary>
    /// <remarks>
    ///     Computed in <see cref="double"/> rather than by shifting, because <c>min &lt;&lt; attempt</c>
    ///     wraps — <c>1 &lt;&lt; 32</c> is <c>1</c> — which would silently turn a long outage back
    ///     into a once-a-second hammering of a dead broker. A <c>maxSeconds</c> below
    ///     <paramref name="minSeconds"/> is misconfiguration; the <i>floor</i> wins, because the floor
    ///     is what stops the loop going hot.
    /// </remarks>
    /// <param name="attempt">Zero-based attempt index; negatives are treated as the first attempt.</param>
    /// <param name="minSeconds">Delay before the first attempt, and the floor for every later one.</param>
    /// <param name="maxSeconds">Cap on the exponential growth.</param>
    public static TimeSpan NextBackoff(int attempt, int minSeconds, int maxSeconds)
    {
        var min = Math.Max(1d, minSeconds);
        var max = Math.Max(min, maxSeconds);

        if (attempt <= 0)
        {
            return TimeSpan.FromSeconds(min);
        }

        // Math.Pow saturates to +Infinity instead of wrapping, so a huge attempt count clamps to max.
        var seconds = min * Math.Pow(2d, attempt);
        return TimeSpan.FromSeconds(seconds >= max || double.IsNaN(seconds) ? max : seconds);
    }

    /// <summary>
    ///     Connects to the broker, failing at <see cref="MqttDriverOptions.ConnectTimeoutSeconds"/>
    ///     rather than waiting indefinitely on an unresponsive peer. On success the reconnect
    ///     supervisor takes over keeping the session alive.
    /// </summary>
    /// <remarks>
    ///     <para>
    ///         May be retried on the same instance after a failed attempt — the underlying
    ///         <see cref="IMqttClient"/> is created once and reused across attempts. Safe to call
    ///         concurrently with <see cref="DisposeAsync"/>: the two are serialised, and a connect
    ///         that loses the race disposes whatever it built and throws
    ///         <see cref="ObjectDisposedException"/>.
    ///     </para>
    ///     <para>
    ///         <b>Idempotent.</b> Calling it on a session that is already established — including one
    ///         the reconnect supervisor restored a moment earlier — is a no-op that returns normally,
    ///         not the <see cref="InvalidOperationException"/> MQTTnet would raise for a
    ///         connect-while-connected. The caller still owns re-establishing its own subscriptions
    ///         after this returns; <see cref="Reconnected"/> is not fired for this path.
    ///     </para>
    /// </remarks>
    /// <param name="cancellationToken">Caller cancellation; linked with the connect deadline.</param>
    /// <exception cref="TimeoutException">The connect deadline elapsed.</exception>
    /// <exception cref="OperationCanceledException"><paramref name="cancellationToken"/> was cancelled.</exception>
    /// <exception cref="ObjectDisposedException">
    ///     The connection was disposed, either before the attempt started or while it was in flight.
    /// </exception>
    /// <exception cref="MqttConnectRejectedException">
    ///     The broker refused the CONNECT (non-success CONNACK — bad credentials, unsupported protocol
    ///     version, server unavailable …). MQTTnet reports this as a <i>result</i>, not an exception,
    ///     so this method inspects the CONNACK and raises it: a caller that treats "did not throw" as
    ///     "connected" would otherwise report a healthy session the broker never granted.
    /// </exception>
    public async Task ConnectAsync(CancellationToken cancellationToken)
    {
        ObjectDisposedException.ThrowIf(Disposed, this);

        await ConnectCoreAsync(supervisorAttempt: false, cancellationToken).ConfigureAwait(false);
    }

    /// <summary>
    ///     Invokes every <see cref="Reconnected"/> subscriber. Walks the invocation list explicitly:
    ///     awaiting a multicast delegate directly yields only the <i>last</i> subscriber's task and
    ///     abandons the earlier ones, so a single throwing subscriber would silently skip every
    ///     subscriber behind it — and a skipped re-subscribe is a topic that goes dark. Failures are
    ///     collected and rethrown after every subscriber has had its turn.
    /// </summary>
    /// <param name="cancellationToken">Passed to every subscriber; cancelled by <see cref="DisposeAsync"/>.</param>
    internal async Task FireReconnectedAsync(CancellationToken cancellationToken)
    {
        var subscribers = Reconnected;
        if (subscribers is null)
        {
            return;
        }

        List<Exception>? failures = null;
        foreach (var subscriber in subscribers.GetInvocationList().Cast<Func<CancellationToken, Task>>())
        {
            try
            {
                await subscriber(cancellationToken).ConfigureAwait(false);
            }
            catch (Exception ex)
            {
                (failures ??= []).Add(ex);
            }
        }

        if (failures is not null)
        {
            throw new AggregateException(
                $"MQTT driver '{_driverId}': {failures.Count} re-subscribe callback(s) failed after reconnect.",
                failures);
        }
    }

    /// <summary>
    ///     The single gated connect both entry points funnel through.
    /// </summary>
    /// <param name="supervisorAttempt">
    ///     <c>true</c> when the reconnect supervisor is calling. It suppresses publishing
    ///     <see cref="MqttConnectionState.Connected"/> on success, because on the reconnect path
    ///     "connected" is only true once <see cref="Reconnected"/> has re-subscribed — publishing it
    ///     here would advertise a healthy connection during the window in which it has no
    ///     subscriptions at all. It also suppresses starting the supervisor (it is the supervisor).
    /// </param>
    /// <param name="cancellationToken">Caller cancellation; linked with the connect deadline.</param>
    /// <returns>
    ///     <c>true</c> if this call established the session; <c>false</c> if it found one already
    ///     established and did nothing. The supervisor uses <c>false</c> to stop retrying <i>without</i>
    ///     firing <see cref="Reconnected"/> — whoever established that session owns its subscribe.
    /// </returns>
    private async Task<bool> ConnectCoreAsync(bool supervisorAttempt, CancellationToken cancellationToken)
    {
        // The options are rebuilt per attempt so a rotated CA file is picked up on the next connect
        // rather than being pinned for the life of the process. This is also the only step that can
        // fail unrecoverably — it is pure config assembly, so retrying it can never help.
        MqttClientOptions clientOptions;
        try
        {
            clientOptions = BuildClientOptions(_options, clientIdSuffix: null, _logger);
        }
        catch (Exception ex)
        {
            SetState(MqttConnectionState.Faulted);
            _logger?.LogError(
                ex,
                "MQTT driver '{DriverId}': broker configuration is unusable; no amount of retrying will help.",
                _driverId);
            throw;
        }

        using var deadline = new CancellationTokenSource(TimeSpan.FromSeconds(_options.ConnectTimeoutSeconds));
        using var linked = CancellationTokenSource.CreateLinkedTokenSource(
            cancellationToken,
            deadline.Token,
            _lifetimeCts.Token);

        _logger?.LogDebug(
            "MQTT driver '{DriverId}': connecting to {Host}:{Port} (tls={UseTls}, timeout={TimeoutSeconds}s).",
            _driverId,
            _options.Host,
            _options.Port,
            _options.UseTls,
            _options.ConnectTimeoutSeconds);

        try
        {
            await _lifecycleGate.WaitAsync(linked.Token).ConfigureAwait(false);
        }
        catch (Exception ex)
        {
            throw Classify(ex, cancellationToken, deadline);
        }

        try
        {
            // Dispose won the race outright: it has already flagged the instance and disposed
            // whatever existed. Creating a client now would be creating one nobody can reach.
            ObjectDisposedException.ThrowIf(Disposed, this);

            var client = _client;
            if (client is null)
            {
                client = new MqttClientFactory().CreateMqttClient();
                AttachHandlers(client);
                _client = client;
            }

            // Idempotence, and the whole point of the gate: MQTTnet throws
            // "not allowed to connect with a server after the connection is established" for a
            // connect-while-connected AND raises no DisconnectedAsync for it, so neither the retry
            // loop nor Classify would ever recover. Both entry points race for real — see the type
            // remarks — so whichever arrives second must find the session and leave it alone.
            if (client.IsConnected)
            {
                SetState(MqttConnectionState.Connected);
                StartSupervisorIfCallerConnect(supervisorAttempt);
                _logger?.LogDebug(
                    "MQTT driver '{DriverId}': connect to {Host}:{Port} skipped — the session is already established.",
                    _driverId,
                    _options.Host,
                    _options.Port);
                return false;
            }

            var connack = await client.ConnectAsync(clientOptions, linked.Token).ConfigureAwait(false);

            // MQTTnet 5 does NOT throw for a broker that refused the CONNECT — it hands the reason
            // back here and leaves the client disconnected. Skipping this check is how a wrong broker
            // password produced a Healthy driver that never received a value (Task-13 live gate).
            if (connack is not null && connack.ResultCode != MqttClientConnectResultCode.Success)
            {
                throw BuildRejection(connack, supervisorAttempt);
            }

            if (AfterConnectHookForTests is { } hook)
            {
                await hook().ConfigureAwait(false);
            }

            // The race the Task-3 review traced: dispose flagged the instance while this connect was
            // in flight, then blocked on the very gate this call holds — so it disposed nothing and
            // the session below would have been unreachable forever. Hand it back here instead.
            if (Disposed)
            {
                Interlocked.CompareExchange(ref _client, null, client);
                await CloseAsync(client).ConfigureAwait(false);
                throw new ObjectDisposedException(
                    nameof(MqttConnection),
                    $"MQTT driver '{_driverId}': the connection was disposed while a connect to "
                    + $"{_options.Host}:{_options.Port} was in flight; the established session was torn down.");
            }

            // On the supervisor path the session is up but has no subscriptions yet, so State stays
            // Reconnecting until FireReconnectedAsync succeeds — see the state table in the remarks.
            if (!supervisorAttempt)
            {
                SetState(MqttConnectionState.Connected);
            }

            StartSupervisorIfCallerConnect(supervisorAttempt);
            return true;
        }
        catch (ObjectDisposedException)
        {
            throw;
        }
        // A refused CONNECT is already fully classified — it must not be re-wrapped as a timeout or a
        // cancellation just because a token happened to fire while the CONNACK was being read.
        catch (MqttConnectRejectedException)
        {
            throw;
        }
        catch (Exception ex)
        {
            throw Classify(ex, cancellationToken, deadline);
        }
        finally
        {
            _lifecycleGate.Release();
        }
    }

    /// <summary>
    ///     Turns a refused CONNACK into the thrown contract, setting <see cref="MqttConnectionState.Faulted"/>
    ///     first when the refusal is unrecoverable — which is what stops the supervisor
    ///     (<see cref="ReconnectSupervisorAsync"/> returns on <c>Faulted</c>) from retrying every
    ///     backoff period against a broker that will never accept these credentials.
    /// </summary>
    private MqttConnectRejectedException BuildRejection(MqttClientConnectResult connack, bool supervisorAttempt)
    {
        var unrecoverable = IsUnrecoverableConnackRejection(connack.ResultCode);
        var message = DescribeConnackRejection($"{_options.Host}:{_options.Port}", connack.ResultCode);

        if (unrecoverable)
        {
            SetState(MqttConnectionState.Faulted);
            _logger?.LogError(
                "MQTT driver '{DriverId}': {Message} Retrying cannot help; fix the configuration and redeploy.",
                _driverId,
                message);
        }
        else
        {
            // Retryable: leave the state alone — Disconnected before a first connect, Reconnecting
            // under the supervisor — so the backoff loop keeps trying at its bounded rate.
            _logger?.LogWarning(
                "MQTT driver '{DriverId}': {Message} Treating as transient{Suffix}.",
                _driverId,
                message,
                supervisorAttempt ? " and continuing to retry" : string.Empty);
        }

        return new MqttConnectRejectedException(connack.ResultCode, unrecoverable, message);
    }

    /// <summary>
    ///     Whether a refused CONNACK can never be fixed by retrying, and so must fault the connection
    ///     rather than leave it looping.
    /// </summary>
    /// <remarks>
    ///     <para>
    ///         Unrecoverable covers credentials/identity (<c>BadUserNameOrPassword</c>,
    ///         <c>NotAuthorized</c>, <c>ClientIdentifierNotValid</c>, <c>Banned</c>,
    ///         <c>BadAuthenticationMethod</c>), protocol mismatch (<c>UnsupportedProtocolVersion</c>,
    ///         <c>MalformedPacket</c>, <c>ProtocolError</c>, <c>PacketTooLarge</c>) and a CONNECT whose
    ///         Last-Will the broker will not accept (<c>TopicNameInvalid</c>,
    ///         <c>PayloadFormatInvalid</c>, <c>RetainNotSupported</c>, <c>QoSNotSupported</c>). Every
    ///         one is a property of the configuration we keep re-sending.
    ///     </para>
    ///     <para>
    ///         <c>ServerMoved</c> (0x9D) is unrecoverable and <c>UseAnotherServer</c> (0x9C) is not,
    ///         because MQTT 5 defines the first as "permanently use another server" and the second as
    ///         "temporarily use another server" — the spec, not a guess.
    ///     </para>
    ///     <para>
    ///         Everything else — including <c>UnspecifiedError</c>, <c>ImplementationSpecificError</c>
    ///         and any code a future broker invents — is treated as transient. That default is
    ///         deliberate: wrongly faulting a transient refusal takes a recoverable driver down until
    ///         someone redeploys, whereas wrongly retrying a permanent one costs a bounded, visibly
    ///         <c>Reconnecting</c>, backoff-paced attempt that names the reason code in the log.
    ///     </para>
    /// </remarks>
    internal static bool IsUnrecoverableConnackRejection(MqttClientConnectResultCode code) => code switch
    {
        MqttClientConnectResultCode.MalformedPacket
            or MqttClientConnectResultCode.ProtocolError
            or MqttClientConnectResultCode.UnsupportedProtocolVersion
            or MqttClientConnectResultCode.ClientIdentifierNotValid
            or MqttClientConnectResultCode.BadUserNameOrPassword
            or MqttClientConnectResultCode.NotAuthorized
            or MqttClientConnectResultCode.Banned
            or MqttClientConnectResultCode.BadAuthenticationMethod
            or MqttClientConnectResultCode.TopicNameInvalid
            or MqttClientConnectResultCode.PacketTooLarge
            or MqttClientConnectResultCode.PayloadFormatInvalid
            or MqttClientConnectResultCode.RetainNotSupported
            or MqttClientConnectResultCode.QoSNotSupported
            or MqttClientConnectResultCode.ServerMoved => true,
        _ => false,
    };

    /// <summary>
    ///     Renders a refused CONNACK as a targeted, credential-free operator message. Shared with
    ///     <c>MqttDriverProbe</c> so the AdminUI "Test connect" button and the running driver describe
    ///     the same rejection in the same words.
    /// </summary>
    internal static string DescribeConnackRejection(string target, MqttClientConnectResultCode code) => code switch
    {
        MqttClientConnectResultCode.NotAuthorized or MqttClientConnectResultCode.BadUserNameOrPassword
            => $"Broker at {target} rejected the credentials ({code}).",
        MqttClientConnectResultCode.UnsupportedProtocolVersion
            => $"Broker at {target} does not support the configured protocol version ({code}).",
        MqttClientConnectResultCode.Banned
            => $"Broker at {target} has banned this client ({code}).",
        _ => $"Broker at {target} refused CONNECT: {code}.",
    };

    /// <summary>
    ///     Starts the supervisor on the first <i>successful</i> caller connect. Always called while
    ///     holding the lifecycle gate, so the <c>??=</c> needs no further synchronisation.
    /// </summary>
    private void StartSupervisorIfCallerConnect(bool supervisorAttempt)
    {
        if (!supervisorAttempt)
        {
            _reconnectWorker ??= Task.Run(() => ReconnectSupervisorAsync(_lifetimeCts.Token));
        }
    }

    /// <summary>
    ///     Maps a bounded-operation failure onto this type's documented contract. MQTTnet wraps a
    ///     cancelled connect in <c>MqttConnectingFailedException</c> rather than letting the
    ///     <see cref="OperationCanceledException"/> surface, so the legs are told apart by which
    ///     token fired, not by exception type. Precedence: teardown, then caller intent, then the
    ///     deadline.
    /// </summary>
    /// <param name="ex">The failure to classify.</param>
    /// <param name="cancellationToken">The caller's token — cancelled means caller intent.</param>
    /// <param name="deadline">The operation's own deadline source.</param>
    /// <param name="operation">The operation name for the message ("connect", "subscribe").</param>
    private Exception Classify(
        Exception ex,
        CancellationToken cancellationToken,
        CancellationTokenSource deadline,
        string operation = "connect")
    {
        if (Disposed)
        {
            return new ObjectDisposedException(
                nameof(MqttConnection),
                new InvalidOperationException(
                    $"MQTT driver '{_driverId}': {operation} to {_options.Host}:{_options.Port} was aborted because the "
                    + "connection was disposed while the attempt was in flight.",
                    ex));
        }

        if (cancellationToken.IsCancellationRequested)
        {
            return new OperationCanceledException(
                $"MQTT driver '{_driverId}': {operation} to {_options.Host}:{_options.Port} was cancelled.",
                ex,
                cancellationToken);
        }

        if (deadline.IsCancellationRequested)
        {
            return new TimeoutException(
                $"MQTT driver '{_driverId}': {operation} to {_options.Host}:{_options.Port} did not complete within "
                + $"{_options.ConnectTimeoutSeconds}s.",
                ex);
        }

        return ex;
    }

    private void AttachHandlers(IMqttClient client)
    {
        client.DisconnectedAsync += OnDisconnectedAsync;
        client.ApplicationMessageReceivedAsync += OnApplicationMessageReceivedAsync;
    }

    /// <summary>
    ///     Runs on MQTTnet's own dispatcher, so it does exactly two non-blocking things and returns.
    ///     Sleeping the backoff here would stall the client's internal pump.
    /// </summary>
    private Task OnDisconnectedAsync(MqttClientDisconnectedEventArgs args)
    {
        if (Disposed || State is MqttConnectionState.Faulted or MqttConnectionState.Disposed)
        {
            return Task.CompletedTask;
        }

        if (State == MqttConnectionState.Connected)
        {
            SetState(MqttConnectionState.Reconnecting);
        }

        // Released unconditionally rather than only when the client had been connected: a missed
        // wake is a permanently dark connection, whereas a spurious one costs a loop iteration.
        _reconnectWake.Release();
        return Task.CompletedTask;
    }

    private Task OnApplicationMessageReceivedAsync(MqttApplicationMessageReceivedEventArgs args)
    {
        Interlocked.Exchange(ref _lastMessageTicksUtc, DateTime.UtcNow.Ticks);

        var observers = MessageReceived;
        if (observers is null)
        {
            return Task.CompletedTask;
        }

        var message = args.ApplicationMessage;
        var payload = message.Payload;

        // Same idiom the browse session uses: the common single-segment case is served straight off
        // the library's buffer; a fragmented sequence is flattened once. Either way the span is only
        // valid for this call, which is exactly what MqttMessageObserver documents.
        ReadOnlySpan<byte> body = payload.IsSingleSegment ? payload.FirstSpan : payload.ToArray();

        try
        {
            observers(message.Topic, body, message.Retain);
        }
        catch (Exception ex)
        {
            // An exception escaping here surfaces inside MQTTnet's own pump. Contain it: a broken
            // observer must degrade its own tags, never stop delivery for every other subscription.
            _logger?.LogError(
                ex,
                "MQTT driver '{DriverId}': an inbound-message observer threw; the message was dropped.",
                _driverId);
        }

        return Task.CompletedTask;
    }

    /// <summary>
    ///     Issues one SUBSCRIBE carrying every requested filter, bounded by
    ///     <see cref="MqttDriverOptions.ConnectTimeoutSeconds"/> so a broker that accepts SUBSCRIBE and
    ///     never answers SUBACK — the frozen-peer shape again — fails at the deadline instead of
    ///     parking the caller (and, on the reconnect path, the supervisor) forever.
    /// </summary>
    /// <remarks>
    ///     The deadline is this method's own linked source, deliberately <b>not</b>
    ///     <see cref="MqttClientOptions.Timeout"/>: that knob already governs every MQTTnet operation
    ///     and repurposing it would couple the subscribe budget to the connect budget in a way neither
    ///     side could change independently.
    ///     <para>
    ///         A rejected filter is <b>not</b> an exception: the returned outcomes carry the broker's
    ///         per-filter verdict so the caller can degrade exactly the affected references. Only a
    ///         failure of the SUBSCRIBE itself (transport error, deadline, teardown) throws.
    ///     </para>
    /// </remarks>
    /// <param name="filters">The filters to establish.</param>
    /// <param name="cancellationToken">Caller cancellation; linked with the subscribe deadline.</param>
    /// <returns>One outcome per requested filter, in request order.</returns>
    /// <exception cref="TimeoutException">The subscribe deadline elapsed.</exception>
    /// <exception cref="OperationCanceledException"><paramref name="cancellationToken"/> was cancelled.</exception>
    /// <exception cref="ObjectDisposedException">The connection was disposed.</exception>
    /// <exception cref="InvalidOperationException">There is no established session to subscribe on.</exception>
    public async Task<IReadOnlyList<MqttSubscribeOutcome>> SubscribeAsync(
        IReadOnlyList<MqttTopicSubscription> filters,
        CancellationToken cancellationToken)
    {
        ArgumentNullException.ThrowIfNull(filters);
        ObjectDisposedException.ThrowIf(Disposed, this);

        if (filters.Count == 0)
        {
            return [];
        }

        var client = _client
            ?? throw new InvalidOperationException(
                $"MQTT driver '{_driverId}': cannot subscribe before a session to {_options.Host}:{_options.Port} "
                + "has been established.");

        var builder = new MqttClientSubscribeOptionsBuilder();
        foreach (var filter in filters)
        {
            builder = builder.WithTopicFilter(f => f
                .WithTopic(filter.Topic)
                .WithQualityOfServiceLevel(MapQos(filter.Qos))
                // Only meaningful on MQTT 5.0. On 3.1.1 the broker always replays its retained
                // message, so the manager also drops unwanted seeds client-side off the retain flag.
                .WithRetainHandling(filter.SeedRetained
                    ? MqttRetainHandling.SendAtSubscribe
                    : MqttRetainHandling.DoNotSendOnSubscribe));
        }

        using var deadline = new CancellationTokenSource(TimeSpan.FromSeconds(_options.ConnectTimeoutSeconds));
        using var linked = CancellationTokenSource.CreateLinkedTokenSource(
            cancellationToken,
            deadline.Token,
            _lifetimeCts.Token);

        MqttClientSubscribeResult result;
        try
        {
            result = await client.SubscribeAsync(builder.Build(), linked.Token).ConfigureAwait(false);
        }
        catch (Exception ex)
        {
            throw Classify(ex, cancellationToken, deadline, operation: "subscribe");
        }

        // Pair each requested filter with its SUBACK item positionally: MQTT guarantees SUBACK reason
        // codes arrive in the order of the SUBSCRIBE's filters. A short/absent SUBACK (a
        // specification-violating broker) is reported as ungranted rather than silently assumed good.
        var items = result.Items as IList<MqttClientSubscribeResultItem> ?? [.. result.Items];
        var outcomes = new List<MqttSubscribeOutcome>(filters.Count);
        for (var i = 0; i < filters.Count; i++)
        {
            if (i >= items.Count)
            {
                outcomes.Add(new MqttSubscribeOutcome(filters[i].Topic, Granted: false, "NoSubAckReasonCode"));
                continue;
            }

            var code = items[i].ResultCode;
            var granted = code is MqttClientSubscribeResultCode.GrantedQoS0
                or MqttClientSubscribeResultCode.GrantedQoS1
                or MqttClientSubscribeResultCode.GrantedQoS2;
            outcomes.Add(new MqttSubscribeOutcome(filters[i].Topic, granted, code.ToString()));
        }

        return outcomes;
    }

    /// <summary>Maps a configured QoS integer onto MQTTnet's enum, clamping an out-of-range value.</summary>
    private static MqttQualityOfServiceLevel MapQos(int qos) => qos switch
    {
        <= 0 => MqttQualityOfServiceLevel.AtMostOnce,
        1 => MqttQualityOfServiceLevel.AtLeastOnce,
        _ => MqttQualityOfServiceLevel.ExactlyOnce,
    };

    /// <summary>
    ///     The reconnect loop MQTTnet v5 no longer provides. Started on the first successful connect
    ///     and stopped by <see cref="DisposeAsync"/> cancelling the lifetime token; it owns every
    ///     reconnect attempt so the library's dispatcher never waits on one.
    /// </summary>
    private async Task ReconnectSupervisorAsync(CancellationToken cancellationToken)
    {
        try
        {
            while (!cancellationToken.IsCancellationRequested)
            {
                await _reconnectWake.WaitAsync(cancellationToken).ConfigureAwait(false);

                if (Disposed || State == MqttConnectionState.Faulted)
                {
                    return;
                }

                if (IsConnected)
                {
                    continue; // Spurious wake (e.g. a failed attempt's own disconnect callback).
                }

                SetState(MqttConnectionState.Reconnecting);

                for (var attempt = 0; !cancellationToken.IsCancellationRequested && !Disposed; attempt++)
                {
                    // A caller's ConnectAsync may have restored the session while we slept. Bail
                    // before burning the backoff, not just before burning an attempt.
                    if (IsConnected)
                    {
                        AdoptSessionRestoredByCaller();
                        break;
                    }

                    await Task
                        .Delay(
                            NextBackoff(
                                attempt,
                                _options.ReconnectMinBackoffSeconds,
                                _options.ReconnectMaxBackoffSeconds),
                            cancellationToken)
                        .ConfigureAwait(false);

                    bool established;
                    try
                    {
                        established = await ConnectCoreAsync(supervisorAttempt: true, cancellationToken)
                            .ConfigureAwait(false);
                    }
                    catch (ObjectDisposedException) when (Disposed)
                    {
                        return;
                    }
                    catch (Exception ex)
                    {
                        if (State == MqttConnectionState.Faulted)
                        {
                            _logger?.LogError(
                                ex,
                                "MQTT driver '{DriverId}': reconnect abandoned — the broker configuration is unusable.",
                                _driverId);
                            return;
                        }

                        _logger?.LogDebug(
                            ex,
                            "MQTT driver '{DriverId}': reconnect attempt {Attempt} to {Host}:{Port} failed.",
                            _driverId,
                            attempt + 1,
                            _options.Host,
                            _options.Port);
                        continue;
                    }

                    if (!established)
                    {
                        // Someone else restored the session between the backoff and the connect.
                        // Firing Reconnected here would re-subscribe on top of a caller that is about
                        // to do exactly that itself, and — worse — leaving the loop running would
                        // keep failing against a healthy connection forever.
                        AdoptSessionRestoredByCaller();
                        break;
                    }

                    if (!await TryReSubscribeAsync(cancellationToken).ConfigureAwait(false))
                    {
                        continue;
                    }

                    // Only now is "Connected" true in the sense this type sells it: connected AND
                    // subscribed.
                    SetState(MqttConnectionState.Connected);
                    _logger?.LogInformation(
                        "MQTT driver '{DriverId}': reconnected to {Host}:{Port} after {Attempts} attempt(s); "
                        + "subscriptions re-established.",
                        _driverId,
                        _options.Host,
                        _options.Port,
                        attempt + 1);
                    break;
                }
            }
        }
        catch (OperationCanceledException)
        {
            // Dispose cancelled the lifetime token — the expected way this loop ends.
        }
        catch (Exception ex)
        {
            // Reaching here means the connection is permanently dark, which is exactly the failure
            // this supervisor exists to prevent — it must never be swallowed quietly, and State must
            // not keep reporting Reconnecting, which every consumer reads as "recovering".
            SetState(MqttConnectionState.Faulted);
            _logger?.LogError(
                ex,
                "MQTT driver '{DriverId}': reconnect supervisor stopped unexpectedly; the broker session will not "
                + "recover without a driver restart.",
                _driverId);
        }
    }

    /// <summary>
    ///     The session came back without this supervisor's help — a caller's <see cref="ConnectAsync"/>
    ///     won the race while we were in backoff. Publish the truth and stand down; deliberately does
    ///     <b>not</b> fire <see cref="Reconnected"/>, because that caller owns its own re-subscribe.
    /// </summary>
    private void AdoptSessionRestoredByCaller()
    {
        SetState(MqttConnectionState.Connected);
        _logger?.LogDebug(
            "MQTT driver '{DriverId}': reconnect stood down — the session to {Host}:{Port} was restored elsewhere.",
            _driverId,
            _options.Host,
            _options.Port);
    }

    /// <summary>
    ///     Fires <see cref="Reconnected"/> — ALWAYS, on every reconnect, including persistent
    ///     sessions where it is merely redundant, because subscriptions do not survive a clean
    ///     session and a reconnect that skips this leaves a healthy-looking, permanently deaf client.
    /// </summary>
    /// <returns>
    ///     <c>false</c> if the re-subscribe failed or overran its deadline, having torn the session
    ///     down so the next attempt starts from a clean CONNECT — serving a connection that receives
    ///     nothing is the one outcome this type must never produce.
    /// </returns>
    private async Task<bool> TryReSubscribeAsync(CancellationToken cancellationToken)
    {
        // Bounded like every other wait here. A subscriber that flows the token cancels promptly; one
        // that ignores it is abandoned at the ceiling rather than being allowed to park the
        // supervisor past DisposeAsync.
        var fanOut = FireReconnectedAsync(cancellationToken);

        // The ceiling abandons the task rather than stopping it, so its eventual failure must still
        // be observed or it resurfaces as an unobserved TaskException on the finalizer thread.
        _ = fanOut.ContinueWith(
            static abandoned => _ = abandoned.Exception,
            CancellationToken.None,
            TaskContinuationOptions.OnlyOnFaulted | TaskContinuationOptions.ExecuteSynchronously,
            TaskScheduler.Default);

        try
        {
            await fanOut
                .WaitAsync(TimeSpan.FromSeconds(_options.ConnectTimeoutSeconds), cancellationToken)
                .ConfigureAwait(false);
            return true;
        }
        catch (OperationCanceledException) when (cancellationToken.IsCancellationRequested)
        {
            throw; // Teardown — let the supervisor's own cancellation handling end the loop.
        }
        catch (Exception ex)
        {
            _logger?.LogError(
                ex,
                "MQTT driver '{DriverId}': re-subscribe after reconnect failed or overran {TimeoutSeconds}s; tearing "
                + "the session down and retrying rather than serving a connection that receives nothing.",
                _driverId,
                _options.ConnectTimeoutSeconds);
            SetState(MqttConnectionState.Reconnecting);
            await ForceDisconnectAsync().ConfigureAwait(false);
            return false;
        }
    }

    /// <summary>
    ///     Drops a session that is connected but unusable, so the next supervisor pass starts from a
    ///     clean CONNECT. Best-effort and bounded — failure here just means the next attempt sees a
    ///     client that is already down.
    /// </summary>
    private async Task ForceDisconnectAsync()
    {
        var client = _client;
        if (client is null)
        {
            return;
        }

        try
        {
            using var deadline = new CancellationTokenSource(TimeSpan.FromSeconds(_options.ConnectTimeoutSeconds));
            await client.DisconnectAsync(new MqttClientDisconnectOptions(), deadline.Token).ConfigureAwait(false);
        }
        catch (Exception ex)
        {
            _logger?.LogDebug(ex, "MQTT driver '{DriverId}': forced disconnect failed.", _driverId);
        }
    }

    private async Task CloseAsync(IMqttClient client)
    {
        try
        {
            if (client.IsConnected)
            {
                // Bounded even on teardown — a wedged broker must not stall driver shutdown.
                using var deadline = new CancellationTokenSource(TimeSpan.FromSeconds(_options.ConnectTimeoutSeconds));
                await client.DisconnectAsync(new MqttClientDisconnectOptions(), deadline.Token).ConfigureAwait(false);
            }
        }
        catch (Exception ex)
        {
            _logger?.LogDebug(ex, "MQTT driver '{DriverId}': disconnect during dispose failed; disposing anyway.", _driverId);
        }
        finally
        {
            client.Dispose();
        }
    }

    /// <summary>
    ///     <see cref="MqttConnectionState.Disposed"/> is terminal — nothing may move off it. A
    ///     check-then-write would let a thread preempted between the two steps resurrect a disposed
    ///     connection's state, so the transition is a compare-and-swap: the terminal check and the
    ///     write are the same atomic operation.
    /// </summary>
    private void SetState(MqttConnectionState state)
    {
        var desired = (int)state;
        while (true)
        {
            var current = Volatile.Read(ref _state);
            if (current == (int)MqttConnectionState.Disposed || current == desired)
            {
                return;
            }

            if (Interlocked.CompareExchange(ref _state, desired, current) == current)
            {
                return;
            }
        }
    }

    /// <summary>
    ///     Assembles the MQTTnet client options from <paramref name="options"/>. Pure — it performs
    ///     no I/O and touches no network, so it is safe to call on a config-validation path. A
    ///     configured <see cref="MqttDriverOptions.CaCertificatePath"/> is read lazily by the
    ///     returned certificate-validation callback at handshake time, not here.
    /// </summary>
    /// <param name="options">Broker connection settings.</param>
    /// <param name="clientIdSuffix">
    ///     Appended to <see cref="MqttDriverOptions.ClientId"/>, so a browse/probe session can share
    ///     the configured identity without colliding with the driver's own client id. When both are
    ///     empty MQTTnet generates a random client id.
    /// </param>
    /// <param name="logger">Optional logger for the deferred CA-pin load; never receives credentials.</param>
    public static MqttClientOptions BuildClientOptions(
        MqttDriverOptions options,
        string? clientIdSuffix,
        ILogger? logger = null)
    {
        ArgumentNullException.ThrowIfNull(options);

        var builder = new MqttClientOptionsBuilder()
            .WithTcpServer(options.Host, options.Port)
            .WithProtocolVersion(MapProtocolVersion(options.ProtocolVersion))
            .WithCleanSession(options.CleanSession)
            .WithKeepAlivePeriod(TimeSpan.FromSeconds(options.KeepAliveSeconds))
            .WithTimeout(TimeSpan.FromSeconds(options.ConnectTimeoutSeconds));

        var clientId = string.Concat(options.ClientId ?? string.Empty, clientIdSuffix ?? string.Empty);
        if (!string.IsNullOrWhiteSpace(clientId))
        {
            builder = builder.WithClientId(clientId);
        }

        // An empty username means "connect anonymously"; MQTTnet would otherwise send an empty
        // CONNECT username, which some brokers treat as a failed auth rather than as anonymous.
        if (!string.IsNullOrEmpty(options.Username))
        {
            builder = builder.WithCredentials(options.Username, options.Password ?? string.Empty);
        }

        builder = builder.WithTlsOptions(tls => ConfigureTls(tls, options, logger));

        return builder.Build();
    }

    private static void ConfigureTls(MqttClientTlsOptionsBuilder tls, MqttDriverOptions options, ILogger? logger)
    {
        if (!options.UseTls)
        {
            tls.UseTls(false);
            return;
        }

        tls.UseTls(true)
            // Explicit SNI / hostname-verification target. Left unset, name validation depends on
            // how MQTTnet infers the host from the endpoint — pin it to the configured host.
            .WithTargetHost(options.Host);

        if (options.AllowUntrustedServerCertificate)
        {
            // Dev / on-prem escape hatch, off unless explicitly enabled — mirrors the
            // ServerHistorian TLS knobs. This is the ONLY branch that accepts a bad certificate.
            tls.WithAllowUntrustedCertificates(true)
                .WithIgnoreCertificateChainErrors(true)
                .WithCertificateValidationHandler(static _ => true);
            return;
        }

        if (string.IsNullOrWhiteSpace(options.CaCertificatePath))
        {
            // No pin: MQTTnet's default handler validates against the OS trust store.
            return;
        }

        tls.WithCertificateValidationHandler(CreatePinnedCaValidator(options.CaCertificatePath, logger));
    }

    /// <summary>
    ///     Builds a certificate-validation callback that pins the broker chain to the PEM CA at
    ///     <paramref name="caCertificatePath"/>. The file is loaded lazily on first validation
    ///     (i.e. during the TLS handshake) so that this — and therefore
    ///     <see cref="BuildClientOptions"/> — stays free of I/O. Any failure to load or to chain
    ///     up to the pinned CA rejects the certificate: fail closed.
    /// </summary>
    private static Func<MqttClientCertificateValidationEventArgs, bool> CreatePinnedCaValidator(
        string caCertificatePath,
        ILogger? logger)
    {
        var trustedRoots = new Lazy<X509Certificate2Collection?>(
            () => LoadPemCa(caCertificatePath, logger),
            LazyThreadSafetyMode.ExecutionAndPublication);

        return args => ValidateAgainstPinnedCa(args, trustedRoots.Value, caCertificatePath, logger);
    }

    private static X509Certificate2Collection? LoadPemCa(string caCertificatePath, ILogger? logger)
    {
        try
        {
            var collection = new X509Certificate2Collection();
            collection.ImportFromPemFile(caCertificatePath);
            if (collection.Count != 0)
            {
                return collection;
            }

            logger?.LogError("MQTT CA pin '{CaCertificatePath}' contains no certificates; broker TLS will be rejected.", caCertificatePath);
        }
        catch (Exception ex)
        {
            logger?.LogError(ex, "MQTT CA pin '{CaCertificatePath}' could not be loaded; broker TLS will be rejected.", caCertificatePath);
        }

        return null;
    }

    private static bool ValidateAgainstPinnedCa(
        MqttClientCertificateValidationEventArgs args,
        X509Certificate2Collection? trustedRoots,
        string caCertificatePath,
        ILogger? logger)
    {
        if (trustedRoots is null || trustedRoots.Count == 0)
        {
            // Unreadable pin ⇒ no trust anchor ⇒ refuse. Never silently degrade to the OS store.
            return false;
        }

        // The pin replaces the trust anchor only. Anything else the platform flagged — a hostname
        // mismatch, an absent certificate — remains fatal.
        if ((args.SslPolicyErrors & ~SslPolicyErrors.RemoteCertificateChainErrors) != SslPolicyErrors.None)
        {
            logger?.LogError("MQTT broker certificate rejected: {SslPolicyErrors}.", args.SslPolicyErrors);
            return false;
        }

        if (args.Certificate is null)
        {
            return false;
        }

        var presented = args.Certificate as X509Certificate2;
        X509Certificate2? owned = null;
        if (presented is null)
        {
            owned = X509CertificateLoader.LoadCertificate(args.Certificate.Export(X509ContentType.Cert));
            presented = owned;
        }

        try
        {
            using var chain = new X509Chain();
            chain.ChainPolicy.TrustMode = X509ChainTrustMode.CustomRootTrust;
            chain.ChainPolicy.CustomTrustStore.AddRange(trustedRoots);

            // Revocation is deliberately not checked: the pin exists precisely for self-issued
            // dev / on-prem CAs, which typically publish no CRL or OCSP responder, so checking
            // would fail every such chain. The pin itself is the revocation story — remove the CA.
            chain.ChainPolicy.RevocationMode = X509RevocationMode.NoCheck;

            // Seed the intermediates the broker presented during the handshake. A leaf signed by
            // an intermediate that chains up to the pinned root is a common broker topology, and
            // that intermediate arrives on the wire rather than being installed locally — without
            // it in the candidate pool a legitimate trust path simply cannot be built.
            // CustomRootTrust still means only `trustedRoots` may terminate the chain, so a rogue
            // self-signed cert smuggled in here cannot become an anchor.
            // Both sources are read: platforms differ in whether the peer-supplied intermediates
            // land in the incoming chain's elements, its ExtraStore, or both.
            if (args.Chain is not null)
            {
                foreach (var element in args.Chain.ChainElements)
                {
                    chain.ChainPolicy.ExtraStore.Add(element.Certificate);
                }

                chain.ChainPolicy.ExtraStore.AddRange(args.Chain.ChainPolicy.ExtraStore);
            }

            if (chain.Build(presented))
            {
                return true;
            }

            logger?.LogError(
                "MQTT broker certificate does not chain to the pinned CA '{CaCertificatePath}': {ChainStatus}.",
                caCertificatePath,
                string.Join(", ", chain.ChainStatus.Select(s => s.Status)));
            return false;
        }
        catch (CryptographicException ex)
        {
            // X509Chain.Build throws on a malformed certificate or an unusable policy. An exception
            // escaping a TLS validation callback surfaces as an opaque handshake crash, so classify
            // it here and refuse: an unverifiable certificate is a rejected certificate.
            logger?.LogError(
                ex,
                "MQTT broker certificate could not be validated against the pinned CA '{CaCertificatePath}'; rejecting.",
                caCertificatePath);
            return false;
        }
        finally
        {
            owned?.Dispose();
        }
    }

    private static MQTTnet.Formatter.MqttProtocolVersion MapProtocolVersion(MqttProtocolVersion version)
        => version switch
        {
            MqttProtocolVersion.V311 => MQTTnet.Formatter.MqttProtocolVersion.V311,
            MqttProtocolVersion.V500 => MQTTnet.Formatter.MqttProtocolVersion.V500,
            _ => throw new ArgumentOutOfRangeException(nameof(version), version, "Unsupported MQTT protocol version."),
        };
}