fix(mqtt): make connect idempotent, bound the reconnect callback, stop State lying

Closes the connect-vs-connect defect: MQTTnet throws (and raises no DisconnectedAsync)
on a connect-while-connected, so a caller's ConnectAsync and the reconnect supervisor
corrupted each other in both directions. Both paths now check first; when the supervisor
finds the session already restored it stands down WITHOUT firing Reconnected.

Reconnected becomes Func<CancellationToken, Task>, fed from the lifetime token and capped
at ConnectTimeoutSeconds, so a hung re-subscribe can no longer park the supervisor.
Connected is published only after the re-subscribe succeeds; a supervisor that dies now
reports Faulted instead of Reconnecting forever.

Claude-Session: https://claude.ai/code/session_01GASWkNEi68FSCtvr6rLoEW

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

@@ -22,8 +22,10 @@ public enum MqttConnectionState
     Reconnecting = 2,
 
     /// <summary>
-    ///     Unrecoverable <i>configuration</i>: retrying cannot help, so the supervisor stops. Only
+    ///     Unrecoverable without a driver restart: retrying cannot help, so the supervisor has
-    ///     failures raised while assembling the client options (before any I/O) land here.
+    ///     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,
 
@@ -69,7 +71,64 @@ public enum MqttConnectionState
 ///         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.
+///         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
@@ -99,7 +158,15 @@ public sealed class MqttConnection : IAsyncDisposable
     /// </summary>
     private readonly SemaphoreSlim _lifecycleGate = new(1, 1);
 
-    /// <summary>Cancelled by <see cref="DisposeAsync"/>; aborts the supervisor and any in-flight connect.</summary>
+    /// <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;
@@ -135,12 +202,20 @@ public sealed class MqttConnection : IAsyncDisposable
 
     /// <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).
+    ///     <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>
-    public event Func<Task>? Reconnected;
+    /// <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>Whether the underlying client currently holds an established MQTT session.</summary>
     public bool IsConnected => _client?.IsConnected ?? false;
@@ -285,11 +360,20 @@ public sealed class MqttConnection : IAsyncDisposable
     ///     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
+    ///         concurrently with <see cref="DisposeAsync"/>: the two are serialised, and a connect
-    ///     loses the race disposes whatever it built and throws
+    ///         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>
@@ -297,21 +381,22 @@ public sealed class MqttConnection : IAsyncDisposable
     /// <exception cref="ObjectDisposedException">
     ///     The connection was disposed, either before the attempt started or while it was in flight.
     /// </exception>
-    public Task ConnectAsync(CancellationToken cancellationToken)
+    public async Task ConnectAsync(CancellationToken cancellationToken)
     {
         ObjectDisposedException.ThrowIf(Disposed, this);
 
-        return ConnectCoreAsync(startSupervisor: true, cancellationToken);
+        await ConnectCoreAsync(supervisorAttempt: false, cancellationToken).ConfigureAwait(false);
     }
 
     /// <summary>
     ///     Invokes every <see cref="Reconnected"/> subscriber. Walks the invocation list explicitly:
-    ///     awaiting a multicast <see cref="Func{Task}"/> directly yields only the <i>last</i>
+    ///     awaiting a multicast delegate directly yields only the <i>last</i> subscriber's task and
-    ///     subscriber's task and abandons the earlier ones, so a single throwing subscriber would
+    ///     abandons the earlier ones, so a single throwing subscriber would silently skip every
-    ///     silently skip every subscriber behind it — and a skipped re-subscribe is a topic that goes
+    ///     subscriber behind it — and a skipped re-subscribe is a topic that goes dark. Failures are
-    ///     dark. Failures are collected and rethrown after every subscriber has had its turn.
+    ///     collected and rethrown after every subscriber has had its turn.
     /// </summary>
-    internal async Task FireReconnectedAsync()
+    /// <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)
@@ -320,11 +405,11 @@ public sealed class MqttConnection : IAsyncDisposable
         }
 
         List<Exception>? failures = null;
-        foreach (var subscriber in subscribers.GetInvocationList().Cast<Func<Task>>())
+        foreach (var subscriber in subscribers.GetInvocationList().Cast<Func<CancellationToken, Task>>())
         {
             try
             {
-                await subscriber().ConfigureAwait(false);
+                await subscriber(cancellationToken).ConfigureAwait(false);
             }
             catch (Exception ex)
             {
@@ -340,7 +425,23 @@ public sealed class MqttConnection : IAsyncDisposable
         }
     }
 
-    private async Task ConnectCoreAsync(bool startSupervisor, CancellationToken cancellationToken)
+    /// <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
@@ -397,6 +498,23 @@ public sealed class MqttConnection : IAsyncDisposable
                 _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;
+            }
+
             await client.ConnectAsync(clientOptions, linked.Token).ConfigureAwait(false);
 
             if (AfterConnectHookForTests is { } hook)
@@ -417,12 +535,15 @@ public sealed class MqttConnection : IAsyncDisposable
                     + $"{_options.Host}:{_options.Port} was in flight; the established session was torn down.");
             }
 
-            SetState(MqttConnectionState.Connected);
+            // 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 (startSupervisor)
+            if (!supervisorAttempt)
             {
-                _reconnectWorker ??= Task.Run(() => ReconnectSupervisorAsync(_lifetimeCts.Token));
+                SetState(MqttConnectionState.Connected);
             }
+
+            StartSupervisorIfCallerConnect(supervisorAttempt);
+            return true;
         }
         catch (ObjectDisposedException)
         {
@@ -438,6 +559,18 @@ public sealed class MqttConnection : IAsyncDisposable
         }
     }
 
+    /// <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 connect failure onto this type's documented contract. MQTTnet wraps a cancelled
     ///     connect in <c>MqttConnectingFailedException</c> rather than letting the
@@ -537,6 +670,14 @@ public sealed class MqttConnection : IAsyncDisposable
 
                 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(
@@ -546,11 +687,13 @@ public sealed class MqttConnection : IAsyncDisposable
                             cancellationToken)
                         .ConfigureAwait(false);
 
+                    bool established;
                     try
                     {
-                        await ConnectCoreAsync(startSupervisor: false, cancellationToken).ConfigureAwait(false);
+                        established = await ConnectCoreAsync(supervisorAttempt: true, cancellationToken)
+                            .ConfigureAwait(false);
                     }
-                    catch (ObjectDisposedException)
+                    catch (ObjectDisposedException) when (Disposed)
                     {
                         return;
                     }
@@ -575,25 +718,24 @@ public sealed class MqttConnection : IAsyncDisposable
                         continue;
                     }
 
-                    try
+                    if (!established)
                     {
-                        // ALWAYS, on every reconnect — including persistent sessions, where this is
+                        // Someone else restored the session between the backoff and the connect.
-                        // merely redundant. Subscriptions do not survive a clean session, and a
+                        // Firing Reconnected here would re-subscribe on top of a caller that is about
-                        // reconnect that skips this leaves a healthy-looking, permanently deaf client.
+                        // to do exactly that itself, and — worse — leaving the loop running would
-                        await FireReconnectedAsync().ConfigureAwait(false);
+                        // keep failing against a healthy connection forever.
+                        AdoptSessionRestoredByCaller();
+                        break;
                     }
-                    catch (Exception ex)
+
+                    if (!await TryReSubscribeAsync(cancellationToken).ConfigureAwait(false))
                     {
-                        _logger?.LogError(
-                            ex,
-                            "MQTT driver '{DriverId}': re-subscribe after reconnect failed; tearing the session down "
-                            + "and retrying rather than serving a connection that receives nothing.",
-                            _driverId);
-                        SetState(MqttConnectionState.Reconnecting);
-                        await ForceDisconnectAsync().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.",
@@ -612,7 +754,9 @@ public sealed class MqttConnection : IAsyncDisposable
         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.
+            // 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 "
@@ -621,6 +765,71 @@ public sealed class MqttConnection : IAsyncDisposable
         }
     }
 
+    /// <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
@@ -666,15 +875,28 @@ public sealed class MqttConnection : IAsyncDisposable
         }
     }
 
-    /// <summary><see cref="MqttConnectionState.Disposed"/> is terminal — nothing may move off it.</summary>
+    /// <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)
     {
-        if (Volatile.Read(ref _state) == (int)MqttConnectionState.Disposed)
+        var desired = (int)state;
+        while (true)
+        {
+            var current = Volatile.Read(ref _state);
+            if (current == (int)MqttConnectionState.Disposed || current == desired)
             {
                 return;
             }
 
-        Volatile.Write(ref _state, (int)state);
+            if (Interlocked.CompareExchange(ref _state, desired, current) == current)
+            {
+                return;
+            }
+        }
     }
 
     /// <summary>

tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Tests/MqttConnectionTests.cs

@@ -4,6 +4,7 @@ using System.Net.Security;
 using System.Net.Sockets;
 using System.Security.Cryptography;
 using System.Security.Cryptography.X509Certificates;
+using Microsoft.Extensions.Logging;
 using MQTTnet;
 using Shouldly;
 using Xunit;
@@ -447,10 +448,10 @@ public sealed class MqttConnectionTests
         await using var conn = new MqttConnection(LoopbackOpts(), driverId: "t", logger: null);
         var first = 0;
         var second = 0;
-        conn.Reconnected += () => { Interlocked.Increment(ref first); return Task.CompletedTask; };
+        conn.Reconnected += _ => { Interlocked.Increment(ref first); return Task.CompletedTask; };
-        conn.Reconnected += () => { Interlocked.Increment(ref second); return Task.CompletedTask; };
+        conn.Reconnected += _ => { Interlocked.Increment(ref second); return Task.CompletedTask; };
 
-        await conn.FireReconnectedAsync();
+        await conn.FireReconnectedAsync(TestContext.Current.CancellationToken);
 
         first.ShouldBe(1);
         second.ShouldBe(1);
@@ -467,10 +468,10 @@ public sealed class MqttConnectionTests
     {
         await using var conn = new MqttConnection(LoopbackOpts(), driverId: "t", logger: null);
         var behindTheThrower = 0;
-        conn.Reconnected += () => throw new InvalidOperationException("SUBACK refused");
+        conn.Reconnected += _ => throw new InvalidOperationException("SUBACK refused");
-        conn.Reconnected += () => { Interlocked.Increment(ref behindTheThrower); return Task.CompletedTask; };
+        conn.Reconnected += _ => { Interlocked.Increment(ref behindTheThrower); return Task.CompletedTask; };
 
-        await Should.ThrowAsync<Exception>(async () => await conn.FireReconnectedAsync());
+        await Should.ThrowAsync<Exception>(async () => await conn.FireReconnectedAsync(TestContext.Current.CancellationToken));
 
         behindTheThrower.ShouldBe(1);
     }
@@ -480,7 +481,7 @@ public sealed class MqttConnectionTests
     {
         await using var conn = new MqttConnection(LoopbackOpts(), driverId: "t", logger: null);
 
-        await Should.NotThrowAsync(async () => await conn.FireReconnectedAsync());
+        await Should.NotThrowAsync(async () => await conn.FireReconnectedAsync(TestContext.Current.CancellationToken));
     }
 
     // ---------------------------------------------------------------------------------
@@ -594,7 +595,7 @@ public sealed class MqttConnectionTests
         using var broker = new MiniBroker();
         await using var conn = new MqttConnection(BrokerOpts(broker.Port), driverId: "t", logger: null);
         var resubscribes = 0;
-        conn.Reconnected += () => { Interlocked.Increment(ref resubscribes); return Task.CompletedTask; };
+        conn.Reconnected += _ => { Interlocked.Increment(ref resubscribes); return Task.CompletedTask; };
 
         await conn.ConnectAsync(CancellationToken.None);
 
@@ -684,20 +685,248 @@ public sealed class MqttConnectionTests
     {
         using var broker = new MiniBroker();
         var conn = new MqttConnection(BrokerOpts(broker.Port), driverId: "t", logger: null);
+
+        // No assertions inside the hook: it runs inside the production try, so a Shouldly throw would
+        // be laundered through Classify and resurface as a confusing TimeoutException. Record, assert
+        // outside.
+        var raceReproduced = false;
         conn.AfterConnectHookForTests = async () =>
         {
             _ = Task.Run(async () => await conn.DisposeAsync());
-            (await WaitUntilAsync(() => conn.State == MqttConnectionState.Disposed, TimeSpan.FromSeconds(5)))
+            raceReproduced = await WaitUntilAsync(
-                .ShouldBeTrue("dispose never flagged the instance — the race was not reproduced");
+                () => conn.State == MqttConnectionState.Disposed,
+                TimeSpan.FromSeconds(5));
         };
 
         await Should.ThrowAsync<ObjectDisposedException>(async () => await conn.ConnectAsync(CancellationToken.None));
 
+        raceReproduced.ShouldBeTrue("dispose never flagged the instance — the race was not reproduced");
         (await WaitUntilAsync(() => broker.LiveConnections == 0, TimeSpan.FromSeconds(15)))
             .ShouldBeTrue($"a live broker connection leaked ({broker.LiveConnections} still open)");
         conn.IsConnected.ShouldBeFalse();
     }
 
+    // ---------------------------------------------------------------------------------
+    // C1 — connect-vs-connect: the supervisor and ConnectAsync must not corrupt each other
+    // ---------------------------------------------------------------------------------
+
+    /// <summary>
+    ///     Direction (b), the simplest form: MQTTnet throws
+    ///     <c>InvalidOperationException: It is not allowed to connect with a server after the
+    ///     connection is established</c> for a connect-while-connected. That is outside this type's
+    ///     documented contract, so a caller re-running <c>InitializeAsync</c> against a healthy
+    ///     session would look like a hard driver failure.
+    /// </summary>
+    [Fact]
+    public async Task ConnectAsync_OnAnAlreadyEstablishedSession_IsANoOp_NotAnInvalidOperation()
+    {
+        using var broker = new MiniBroker();
+        await using var conn = new MqttConnection(BrokerOpts(broker.Port), driverId: "t", logger: null);
+
+        await conn.ConnectAsync(CancellationToken.None);
+        await Should.NotThrowAsync(async () => await conn.ConnectAsync(CancellationToken.None));
+
+        conn.IsConnected.ShouldBeTrue();
+        conn.State.ShouldBe(MqttConnectionState.Connected);
+        broker.AcceptedCount.ShouldBe(1, "the second connect opened a second socket instead of no-opping");
+    }
+
+    /// <summary>
+    ///     Direction (b) through the supervisor: the supervisor restores the session, then the driver
+    ///     host re-runs <c>InitializeAsync</c> — exactly the sequence this design relies on.
+    /// </summary>
+    [Fact]
+    public async Task ConnectAsync_AfterTheSupervisorAlreadyReconnected_IsANoOp()
+    {
+        using var broker = new MiniBroker();
+        await using var conn = new MqttConnection(BrokerOpts(broker.Port), driverId: "t", logger: null);
+        var resubscribes = 0;
+        conn.Reconnected += _ => { Interlocked.Increment(ref resubscribes); return Task.CompletedTask; };
+
+        await conn.ConnectAsync(CancellationToken.None);
+        broker.DropAll();
+        (await WaitUntilAsync(
+                () => Volatile.Read(ref resubscribes) >= 1 && conn.State == MqttConnectionState.Connected,
+                TimeSpan.FromSeconds(30)))
+            .ShouldBeTrue();
+
+        await Should.NotThrowAsync(async () => await conn.ConnectAsync(CancellationToken.None));
+
+        conn.IsConnected.ShouldBeTrue();
+        conn.State.ShouldBe(MqttConnectionState.Connected);
+    }
+
+    /// <summary>
+    ///     Direction (a): a caller reconnects while the supervisor is asleep in backoff. The
+    ///     supervisor must find the restored session and stand down — <b>silently</b>, without firing
+    ///     <c>Reconnected</c> (the caller owns its own re-subscribe). Left unfixed it fails forever
+    ///     against a perfectly healthy connection, at Debug level, inflating <c>attempt</c> so the
+    ///     NEXT genuine outage waits max backoff instead of min.
+    /// </summary>
+    [Fact]
+    public async Task CallerReconnectsWhileTheSupervisorIsInBackoff_SupervisorStandsDownSilently()
+    {
+        using var broker = new MiniBroker();
+        var log = new CapturingLogger();
+        var opts = BrokerOpts(broker.Port) with { ReconnectMinBackoffSeconds = 3, ReconnectMaxBackoffSeconds = 3 };
+        await using var conn = new MqttConnection(opts, driverId: "t", logger: log);
+        var resubscribes = 0;
+        conn.Reconnected += _ => { Interlocked.Increment(ref resubscribes); return Task.CompletedTask; };
+
+        await conn.ConnectAsync(CancellationToken.None);
+        broker.DropAll();
+        (await WaitUntilAsync(() => conn.State == MqttConnectionState.Reconnecting, TimeSpan.FromSeconds(10)))
+            .ShouldBeTrue();
+
+        // The caller wins the race while the supervisor sleeps off its 3 s backoff.
+        await conn.ConnectAsync(CancellationToken.None);
+        conn.IsConnected.ShouldBeTrue();
+        var failuresBefore = log.CountContaining("reconnect attempt");
+
+        // Three more supervisor wake points would land in this window if it were still looping.
+        await Task.Delay(TimeSpan.FromSeconds(9), TestContext.Current.CancellationToken);
+
+        log.CountContaining("reconnect attempt")
+            .ShouldBe(failuresBefore, "the supervisor kept failing against a healthy connection");
+        conn.State.ShouldBe(MqttConnectionState.Connected);
+        conn.IsConnected.ShouldBeTrue();
+        Volatile.Read(ref resubscribes).ShouldBe(0, "the caller owns its own re-subscribe; Reconnected must not fire");
+    }
+
+    /// <summary>
+    ///     The operational consequence of direction (a), measured rather than argued: after a caller
+    ///     has won a reconnect race, the supervisor must be parked at its outer wait with its attempt
+    ///     counter reset, so the NEXT genuine outage recovers at MIN backoff.
+    /// </summary>
+    /// <remarks>
+    ///     The timings are load-bearing, not arbitrary. A supervisor still flailing against the healthy
+    ///     connection would have burned attempts 0 and 1 (at t≈2 s and t≈6 s) and be asleep in an 8 s
+    ///     backoff running to t≈14 s, so the outage induced at t≈7 s cannot be answered before then.
+    ///     A stood-down supervisor answers it one min-backoff (2 s) later. The 5 s window separates the
+    ///     two by a clear margin; widening the burn only widens the separation.
+    /// </remarks>
+    [Fact]
+    public async Task AfterACallerWinsTheRace_TheNextGenuineOutageStillRecoversAtMinBackoff()
+    {
+        using var broker = new MiniBroker();
+        var opts = BrokerOpts(broker.Port) with { ReconnectMinBackoffSeconds = 2, ReconnectMaxBackoffSeconds = 30 };
+        await using var conn = new MqttConnection(opts, driverId: "t", logger: null);
+        var resubscribes = 0;
+        conn.Reconnected += _ => { Interlocked.Increment(ref resubscribes); return Task.CompletedTask; };
+
+        await conn.ConnectAsync(CancellationToken.None);
+        broker.DropAll();
+        (await WaitUntilAsync(() => conn.State == MqttConnectionState.Reconnecting, TimeSpan.FromSeconds(10)))
+            .ShouldBeTrue();
+        await conn.ConnectAsync(CancellationToken.None); // caller wins the race
+        await Task.Delay(TimeSpan.FromSeconds(7), TestContext.Current.CancellationToken); // a broken loop climbs here
+
+        broker.DropAll(); // the NEXT genuine outage
+        var recovered = await WaitUntilAsync(() => Volatile.Read(ref resubscribes) >= 1, TimeSpan.FromSeconds(5));
+
+        recovered.ShouldBeTrue("recovery took far longer than min backoff — the attempt counter never reset");
+        conn.State.ShouldBe(MqttConnectionState.Connected);
+    }
+
+    // ---------------------------------------------------------------------------------
+    // I2 / M3 — the Reconnected callback is bounded, and Connected means "connected AND subscribed"
+    // ---------------------------------------------------------------------------------
+
+    [Fact]
+    public async Task FireReconnected_PassesTheSuppliedTokenToEverySubscriber()
+    {
+        await using var conn = new MqttConnection(LoopbackOpts(), driverId: "t", logger: null);
+        using var cts = new CancellationTokenSource();
+        var seen = new List<CancellationToken>();
+        conn.Reconnected += ct => { seen.Add(ct); return Task.CompletedTask; };
+        conn.Reconnected += ct => { seen.Add(ct); return Task.CompletedTask; };
+
+        await conn.FireReconnectedAsync(cts.Token);
+
+        seen.Count.ShouldBe(2);
+        seen.ShouldAllBe(t => t == cts.Token);
+    }
+
+    /// <summary>
+    ///     A re-subscribe that hangs — broker accepts SUBSCRIBE and never SUBACKs, the frozen-peer
+    ///     shape again — must not park the supervisor. This subscriber deliberately <b>ignores</b> its
+    ///     token, so only the fan-out's own ceiling can end the wait.
+    /// </summary>
+    /// <remarks>
+    ///     Asserted at the transport, not at dispose timing: with the ceiling, each hung re-subscribe
+    ///     is abandoned, the session torn down, and another CONNECT reaches the broker; without it,
+    ///     the supervisor is parked forever on the first one and the broker never sees another. A
+    ///     dispose-elapsed assertion cannot tell those apart — <c>DisposeAsync</c>'s own bounded join
+    ///     returns promptly either way, leaving the supervisor alive behind it.
+    /// </remarks>
+    [Fact]
+    public async Task ReSubscribeThatHangsIgnoringItsToken_DoesNotParkTheSupervisor()
+    {
+        using var broker = new MiniBroker();
+        var opts = BrokerOpts(broker.Port) with
+        {
+            ConnectTimeoutSeconds = 2, ReconnectMinBackoffSeconds = 1, ReconnectMaxBackoffSeconds = 2,
+        };
+        var conn = new MqttConnection(opts, driverId: "t", logger: null);
+        var entered = new TaskCompletionSource();
+        conn.Reconnected += async _ =>
+        {
+            entered.TrySetResult();
+            await Task.Delay(Timeout.InfiniteTimeSpan); // no token — only the ceiling bounds this
+        };
+
+        await conn.ConnectAsync(CancellationToken.None);
+        broker.DropAll();
+        await entered.Task.WaitAsync(TimeSpan.FromSeconds(30), TestContext.Current.CancellationToken);
+
+        // The first reconnect is accept #2; a supervisor that recovered from the hang makes at least
+        // one more. A parked one never does.
+        (await WaitUntilAsync(() => broker.AcceptedCount >= 3, TimeSpan.FromSeconds(25)))
+            .ShouldBeTrue($"the supervisor is parked behind a hung re-subscribe ({broker.AcceptedCount} connects seen)");
+
+        var sw = Stopwatch.StartNew();
+        await conn.DisposeAsync();
+        sw.Stop();
+
+        sw.Elapsed.ShouldBeLessThan(TimeSpan.FromSeconds(20), "dispose was parked behind a hung re-subscribe");
+        conn.State.ShouldBe(MqttConnectionState.Disposed);
+        var acceptedAtDispose = broker.AcceptedCount;
+        await Task.Delay(TimeSpan.FromSeconds(5), TestContext.Current.CancellationToken);
+        broker.AcceptedCount.ShouldBe(acceptedAtDispose, "the supervisor outlived DisposeAsync");
+    }
+
+    /// <summary>
+    ///     M3: on the reconnect path the socket comes up before the subscriptions do. Publishing
+    ///     <c>Connected</c> in that window advertises exactly the healthy-looking-but-deaf condition
+    ///     this type exists to prevent. It is also the documented case where <c>State</c> and
+    ///     <c>IsConnected</c> legitimately disagree.
+    /// </summary>
+    [Fact]
+    public async Task Reconnect_DoesNotPublishConnectedUntilTheReSubscribeHasCompleted()
+    {
+        using var broker = new MiniBroker();
+        await using var conn = new MqttConnection(BrokerOpts(broker.Port), driverId: "t", logger: null);
+        var inSubscriber = new TaskCompletionSource();
+        var release = new TaskCompletionSource();
+        conn.Reconnected += async _ =>
+        {
+            inSubscriber.TrySetResult();
+            await release.Task;
+        };
+
+        await conn.ConnectAsync(CancellationToken.None);
+        broker.DropAll();
+        await inSubscriber.Task.WaitAsync(TimeSpan.FromSeconds(30), TestContext.Current.CancellationToken);
+
+        conn.IsConnected.ShouldBeTrue("the socket is up");
+        conn.State.ShouldBe(MqttConnectionState.Reconnecting, "…but it has no subscriptions yet");
+
+        release.SetResult();
+
+        (await WaitUntilAsync(() => conn.State == MqttConnectionState.Connected, TimeSpan.FromSeconds(10)))
+            .ShouldBeTrue();
+    }
+
     // ---------------------------------------------------------------------------------
     // helpers
     // ---------------------------------------------------------------------------------
@@ -720,6 +949,44 @@ public sealed class MqttConnectionTests
             ReconnectMaxBackoffSeconds = 2,
         };
 
+    /// <summary>
+    ///     Captures formatted log messages. The supervisor's failed-attempt path is Debug-only and has
+    ///     no other externally visible effect — MQTTnet refuses a connect-while-connected before
+    ///     opening a socket, so the broker never sees it — which is precisely why the C1 defect was
+    ///     silent. The log is the observable.
+    /// </summary>
+    private sealed class CapturingLogger : ILogger
+    {
+        private readonly List<string> _messages = [];
+
+        public int CountContaining(string fragment)
+        {
+            lock (_messages)
+            {
+                return _messages.Count(m => m.Contains(fragment, StringComparison.OrdinalIgnoreCase));
+            }
+        }
+
+        public IDisposable? BeginScope<TState>(TState state)
+            where TState : notnull
+            => null;
+
+        public bool IsEnabled(LogLevel logLevel) => true;
+
+        public void Log<TState>(
+            LogLevel logLevel,
+            EventId eventId,
+            TState state,
+            Exception? exception,
+            Func<TState, Exception?, string> formatter)
+        {
+            lock (_messages)
+            {
+                _messages.Add(formatter(state, exception));
+            }
+        }
+    }
+
     private static async Task<bool> WaitUntilAsync(Func<bool> condition, TimeSpan timeout)
     {
         var deadline = Stopwatch.StartNew();