fix(mqtt): retry failed SUBSCRIBEs via a pending set; throw on no-verdict subscribe (R06:M-1, M-11)

A failed SUBSCRIBE (refusal or no broker verdict) used to drop its topic from every
retry set while SubscribeAsync returned a live handle, leaving tags dark behind a
Healthy connection until a redeploy. Failed topics now join a pending set retried by
the next SubscribeAsync and by every reconnect (not counted toward the reconnect's
total-failure verdict, so an ACL-denied topic cannot reconnect-loop). A no-verdict
subscribe degrades its refs then throws so the actor's reconcile re-subscribes. A
subscribe with no transport attached is recorded pending, never established (M-11).
MiniBroker now answers SUBSCRIBE.

Refs #744
Claude-Session: https://claude.ai/code/session_01Ub9hGmYLQvn9XETzw7CMsT

docs/drivers/Mqtt.md

@@ -408,9 +408,18 @@ value. `MqttConnection` therefore raises `MqttConnectRejectedException`:
   code. (`ServerMoved` is permanent, `UseAnotherServer` is temporary — per the MQTT 5 spec.)
 
 A reconnect that completes without re-subscribing would be a healthy-looking connection that receives
-nothing forever, so `OnReconnectedAsync` **throws** when zero filters are granted, tearing the session
-down rather than swallowing it. A SUBACK failure degrades just the affected RawPaths to
-`BadCommunicationError`.
+nothing forever, so `OnReconnectedAsync` **throws** when none of the previously live filters is granted
+(or the SUBSCRIBE gets no verdict), tearing the session down rather than swallowing it. A SUBACK failure
+degrades just the affected RawPaths to `BadCommunicationError`.
+
+**A failed SUBSCRIBE is retried, never forgotten (R06:M-1).** Every topic whose SUBSCRIBE was refused or got
+no verdict joins a *pending* set that the next `SubscribeAsync` and every reconnect re-request; a grant moves
+it to the live set, and a redeploy that drops its tags drops it. Pending topics do not count toward the
+reconnect's total-failure verdict, so one ACL-denied topic cannot reconnect-loop the driver. A plain-mode
+`SubscribeAsync` whose SUBSCRIBE got **no verdict at all** (transport throw, SUBSCRIBE deadline) degrades its
+refs and then **throws**, so `DriverInstanceActor` holds no handle and its health-tick reconcile re-subscribes;
+a SUBACK *refusal* still returns a handle (the broker answered). A subscribe issued with no transport attached
+records its topics as pending, never as established (M-11).
 
 There are **two** kinds of reconnect and only one of them fires that callback. A drop the connection's
 own supervisor recovers keeps the same `MqttConnection`, so `Reconnected` → `OnReconnectedAsync`

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

@@ -113,6 +113,14 @@ public sealed class MqttSubscriptionManager
     /// <summary>Topics already SUBSCRIBEd on the live session — the dedupe set and the re-subscribe set.</summary>
     private readonly ConcurrentDictionary<string, byte> _subscribedTopics = new(StringComparer.Ordinal);
 
+    /// <summary>
+    ///     Topics a subscribe asked for whose SUBSCRIBE failed (refused, or no verdict at all). Retried by
+    ///     every later <see cref="SubscribeAsync"/> and by <see cref="OnReconnectedAsync"/>; moved to
+    ///     <see cref="_subscribedTopics"/> when granted. Without it a failed topic left every retry set and
+    ///     stayed dark behind a Healthy connection until a redeploy.
+    /// </summary>
+    private readonly ConcurrentDictionary<string, byte> _pendingTopics = new(StringComparer.Ordinal);
+
     /// <summary>RawPaths whose extraction failure has already been logged loudly once.</summary>
     private readonly ConcurrentDictionary<string, byte> _warnedRefs = new(StringComparer.Ordinal);
 
@@ -212,8 +220,8 @@ public sealed class MqttSubscriptionManager
     ///         The manager does not own the connection's lifetime and never detaches — the driver disposes
     ///         both together. Call once per connection, and <b>before</b> the first
     ///         <see cref="SubscribeAsync"/>: a subscribe issued with no transport attached records its
-    ///         intent (and warns) but sends nothing to a broker, and would only be established if the
-    ///         connection later dropped and recovered.
+    ///         topics as <i>pending</i> (and warns) but sends nothing to a broker; the next subscribe or
+    ///         reconnect on an attached session retries them.
     ///     </para>
     ///     <para>
     ///         <b>Binding to a session forgets the previous session's established filters</b> — the same
@@ -328,6 +336,14 @@ public sealed class MqttSubscriptionManager
             }
         }
 
+        foreach (var topic in _pendingTopics.Keys)
+        {
+            if (!table.ByFilter.ContainsKey(topic))
+            {
+                _pendingTopics.TryRemove(topic, out _);
+            }
+        }
+
         foreach (var rawPath in _handleByRawPath.Keys)
         {
             if (!table.ByRawPath.ContainsKey(rawPath))
@@ -356,12 +372,13 @@ public sealed class MqttSubscriptionManager
     ///         opening SUBSCRIBE itself; <see cref="OnReconnectedAsync"/> only re-establishes it later.
     ///     </para>
     ///     <para>
-    ///         <b>Never throws, never hangs.</b> A reference that is not an authored tag is recorded as
-    ///         <c>BadNodeIdUnknown</c>; a SUBACK rejection, transport failure or subscribe deadline
-    ///         degrades exactly the affected references to <c>BadCommunicationError</c>. Either way a
-    ///         handle is returned — the OPC UA server's subscribe call must not fail because one topic
-    ///         is ACL-denied. Recovery for a transient failure comes from the reconnect path;
-    ///         a persistent broker-side refusal is visible as a Bad status on the affected nodes.
+    ///         <b>Never hangs; throws only when the broker gave no verdict.</b> A reference that is not an
+    ///         authored tag is recorded as <c>BadNodeIdUnknown</c>. A SUBACK rejection degrades exactly the
+    ///         affected references to <c>BadCommunicationError</c> and still returns a handle — the OPC UA
+    ///         server's subscribe must not fail because one topic is ACL-denied. A transport failure or
+    ///         subscribe deadline degrades the same way and then <b>throws</b>, so the host holds no handle
+    ///         and its reconcile re-subscribes. Every failed topic also joins a pending set that the next
+    ///         subscribe and every reconnect retry.
     ///     </para>
     /// </remarks>
     /// <param name="fullReferences">The RawPath references to subscribe.</param>
@@ -414,12 +431,43 @@ public sealed class MqttSubscriptionManager
 
         // One SUBSCRIBE per distinct topic, and only for topics not already live: re-SUBSCRIBEing an
         // established filter is legal but pointless, and it re-triggers the retained seed.
-        var pending = defs.Where(d => !_subscribedTopics.ContainsKey(d.Topic)).ToList();
-        await EstablishAsync(BuildFilters(pending, _defaultQos), cancellationToken).ConfigureAwait(false);
+        // Topics an EARLIER subscribe failed to establish ride along: a redeploy's bulk re-subscribe is then a
+        // retry of a refusal the broker may since have lifted, instead of a no-op that leaves them dark.
+        var authored = _authored;
+        var pending = defs.Where(d => !_subscribedTopics.ContainsKey(d.Topic))
+            .Concat(PendingDefinitions(authored))
+            .ToList();
+        var outcome = await EstablishAsync(BuildFilters(pending, _defaultQos), cancellationToken).ConfigureAwait(false);
 
+        if (outcome.TransportFailure is { } failure)
+        {
+            // No verdict at all (transport throw / SUBSCRIBE deadline): THROW, so the caller holds no handle
+            // and DriverInstanceActor's health-tick reconcile re-subscribes. Returning a handle here was the
+            // silent-death case — the actor thought the subscription live and never retried. The refs already
+            // carry BadCommunicationError (raised under this handle above), so the throw costs no visibility.
+            _refsByHandle.TryRemove(handle, out _);
+            foreach (var def in defs)
+            {
+                _handleByRawPath.TryRemove(new KeyValuePair<string, ISubscriptionHandle>(def.Name, handle));
+            }
+
+            throw new InvalidOperationException(
+                $"MQTT driver '{_driverId}': SUBSCRIBE for {pending.Count} tag(s) got no broker verdict "
+                + $"({failure.GetType().Name}); the subscription will be retried.",
+                failure);
+        }
+
+        // A SUBACK refusal (an ACL denial, say) does NOT throw: the broker answered, the refused topics are
+        // Bad + pending, and failing the whole subscribe would take every other topic in it dark too.
         return handle;
     }
 
+    /// <summary>The authored definitions behind every still-authored pending topic.</summary>
+    private IEnumerable<MqttTagDefinition> PendingDefinitions(AuthoredTable authored)
+        => _pendingTopics.Keys
+            .Select(topic => authored.ByFilter.TryGetValue(topic, out var defs) ? defs : [])
+            .SelectMany(defs => defs);
+
     /// <summary>
     ///     Stops attributing notifications to <paramref name="handle"/>. References another live
     ///     subscription also covers keep publishing.
@@ -498,8 +546,13 @@ public sealed class MqttSubscriptionManager
             .SelectMany(defs => defs)
             .ToList();
 
-        var filters = BuildFilters(live, _defaultQos);
-        if (filters.Count == 0)
+        // Pending topics (a failed earlier SUBSCRIBE) are retried here too — a reconnect re-subscribing
+        // only the granted set is how a failed topic stayed dark forever. They do NOT count toward the
+        // total-failure verdict below: re-refusing a topic that was never live loses nothing, and throwing
+        // for it would turn one ACL-denied topic into a reconnect loop for the whole driver.
+        var pendingDefs = PendingDefinitions(authored).ToList();
+        var filters = BuildFilters(live.Concat(pendingDefs), _defaultQos);
+        if (live.Count == 0 && pendingDefs.Count == 0)
         {
             // Silence here is how the wildcard defect hid: no exception, no log, no SUBSCRIBE. Live
             // topics resolving to zero filters re-establishes nothing, which is the total-failure case
@@ -519,13 +572,17 @@ public sealed class MqttSubscriptionManager
             return;
         }
 
-        var granted = await EstablishAsync(filters, cancellationToken).ConfigureAwait(false);
-        if (granted == 0)
+        var outcome = await EstablishAsync(filters, cancellationToken).ConfigureAwait(false);
+        var liveTopics = live.Select(d => d.Topic).Distinct(StringComparer.Ordinal).ToList();
+        var liveLost = outcome.TransportFailure is not null
+                       || (subscribed.Count > 0 && (liveTopics.Count == 0 || !liveTopics.Any(outcome.Granted.Contains)));
+        if (liveLost)
         {
             throw new InvalidOperationException(
                 $"MQTT driver '{_driverId}': no subscription could be re-established after reconnect "
                 + $"({filters.Count} filter(s) requested); tearing the session down rather than serving a "
-                + "connection that receives nothing.");
+                + "connection that receives nothing.",
+                outcome.TransportFailure);
         }
     }
 
@@ -633,35 +690,35 @@ public sealed class MqttSubscriptionManager
     /// </summary>
     /// <param name="filters">The filters to establish; an empty list is a no-op.</param>
     /// <param name="cancellationToken">Cancellation for the round-trip.</param>
-    /// <returns>How many filters the broker granted.</returns>
-    private async Task<int> EstablishAsync(
+    /// <returns>The granted topics, plus the transport failure when the broker gave no verdict at all.</returns>
+    private async Task<EstablishOutcome> EstablishAsync(
         IReadOnlyList<MqttTopicSubscription> filters,
         CancellationToken cancellationToken)
     {
         if (filters.Count == 0)
         {
-            return 0;
+            return EstablishOutcome.Nothing;
         }
 
         var transport = _transport;
         if (transport is null)
         {
-            // No transport: the intent is recorded so the reconnect path establishes it, but NOTHING
-            // reaches a broker now. Loud, because outside tests this means the composition order is
-            // wrong — AttachTo must run before the OPC UA server's first subscribe, or these topics
-            // stay unestablished until the connection happens to drop and recover.
+            // No transport: NOTHING reaches a broker, so nothing is recorded as subscribed — recording it
+            // was a silent "subscribed, receives nothing" if the AttachTo-before-subscribe ordering ever
+            // changed. The topics wait in the pending set, which the next subscribe and every reconnect
+            // retry. Loud, because outside tests this means the composition order is wrong.
             foreach (var filter in filters)
             {
-                _subscribedTopics[filter.Topic] = 0;
+                _pendingTopics[filter.Topic] = 0;
             }
 
             _logger?.LogWarning(
-                "MQTT driver '{DriverId}': {FilterCount} subscription(s) recorded with no connection attached; "
-                + "nothing was sent to a broker. Call AttachTo before subscribing.",
+                "MQTT driver '{DriverId}': {FilterCount} subscription(s) requested with no connection attached; "
+                + "nothing was sent to a broker and they stay pending. Call AttachTo before subscribing.",
                 _driverId,
                 filters.Count);
 
-            return filters.Count;
+            return EstablishOutcome.Nothing;
         }
 
         IReadOnlyList<MqttSubscribeOutcome> outcomes;
@@ -681,16 +738,17 @@ public sealed class MqttSubscriptionManager
                 DegradeTopic(filter.Topic, ex.GetType().Name);
             }
 
-            return 0;
+            return new EstablishOutcome(new HashSet<string>(StringComparer.Ordinal), ex);
         }
 
-        var granted = 0;
+        var granted = new HashSet<string>(StringComparer.Ordinal);
         foreach (var outcome in outcomes)
         {
             if (outcome.Granted)
             {
                 _subscribedTopics[outcome.Topic] = 0;
-                granted++;
+                _pendingTopics.TryRemove(outcome.Topic, out _);
+                granted.Add(outcome.Topic);
             }
             else
             {
@@ -703,7 +761,16 @@ public sealed class MqttSubscriptionManager
             }
         }
 
-        return granted;
+        return new EstablishOutcome(granted, null);
+    }
+
+    /// <summary>One SUBSCRIBE round-trip's verdict.</summary>
+    /// <param name="Granted">The filters the broker granted.</param>
+    /// <param name="TransportFailure">Set when the call produced no verdict at all (throw / deadline).</param>
+    private sealed record EstablishOutcome(IReadOnlySet<string> Granted, Exception? TransportFailure)
+    {
+        public static readonly EstablishOutcome Nothing =
+            new(new HashSet<string>(StringComparer.Ordinal), null);
     }
 
     /// <summary>Marks every authored reference on <paramref name="topic"/> as unreachable.</summary>
@@ -722,6 +789,7 @@ public sealed class MqttSubscriptionManager
 
         if (!_authored.ByFilter.TryGetValue(topic, out var defs))
         {
+            _pendingTopics.TryRemove(topic, out _);
             _logger?.LogWarning(
                 "MQTT driver '{DriverId}': cannot degrade tags for unestablished filter '{Topic}' ({Reason}) — "
                 + "no authored tag resolves to it.",
@@ -731,6 +799,9 @@ public sealed class MqttSubscriptionManager
             return;
         }
 
+        // Still authored, so it must be retried — see _pendingTopics.
+        _pendingTopics[topic] = 0;
+
         var now = DateTime.UtcNow;
         var snapshot = new DataValueSnapshot(null, StatusBadCommunicationError, null, now);
         foreach (var def in defs)

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

@@ -59,6 +59,54 @@ internal sealed class MiniBroker : IDisposable
     /// </summary>
     public byte ConnAckReturnCode { get; set; } = ReturnCodeAccepted;
 
+    /// <summary>
+    ///     When true (the default) a SUBSCRIBE is answered with a SUBACK granting each filter's requested
+    ///     QoS. False is a broker that accepts SUBSCRIBE and never answers — the subscribe-deadline shape.
+    /// </summary>
+    public bool AnswerSubscribe { get; set; } = true;
+
+    /// <summary>
+    ///     Builds the SUBACK for one SUBSCRIBE packet read whole into <paramref name="buffer"/>, or null when
+    ///     the packet is not parseable from one read (the fake handles one packet per read, like CONNECT).
+    /// </summary>
+    private static byte[]? TryBuildSubAck(byte[] buffer, int read)
+    {
+        // Fixed header: type byte, then the variable-length "remaining length".
+        var at = 1;
+        var remaining = 0;
+        var multiplier = 1;
+        byte digit;
+        do
+        {
+            if (at >= read) return null;
+            digit = buffer[at++];
+            remaining += (digit & 0x7F) * multiplier;
+            multiplier *= 128;
+        }
+        while ((digit & 0x80) != 0);
+
+        var end = at + remaining;
+        if (end > read || remaining < 2) return null;
+
+        var packetIdHigh = buffer[at];
+        var packetIdLow = buffer[at + 1];
+        at += 2;
+
+        var grants = new List<byte>();
+        while (at + 2 <= end)
+        {
+            var length = (buffer[at] << 8) | buffer[at + 1];
+            at += 2 + length;
+            if (at >= end) return null;
+            grants.Add((byte)(buffer[at] & 0x03));
+            at++;
+        }
+
+        var ack = new List<byte> { 0x90, (byte)(2 + grants.Count), packetIdHigh, packetIdLow };
+        ack.AddRange(grants);
+        return [.. ack];
+    }
+
     /// <summary>Total TCP connections accepted since construction — counts reconnect attempts.</summary>
     public int AcceptedCount => Volatile.Read(ref _acceptedCount);
 
@@ -163,6 +211,10 @@ internal sealed class MiniBroker : IDisposable
                 {
                     await stream.WriteAsync(PingResp, _cts.Token);
                 }
+                else if (packetType == 0x80 && AnswerSubscribe && TryBuildSubAck(buffer, read) is { } subAck)
+                {
+                    await stream.WriteAsync(subAck, _cts.Token);
+                }
             }
         }
         catch (Exception)

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

@@ -13,9 +13,9 @@ namespace ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Tests;
 ///     (<see cref="MqttDriver.ShutdownAsync"/> / <see cref="MqttDriver.DisposeAsync"/>).
 /// </summary>
 /// <remarks>
-///     Plain mode deliberately: it is the only mode whose initialize needs no SUBSCRIBE, and
-///     <see cref="MiniBroker"/> answers CONNECT and PINGREQ and nothing else — so a Sparkplug driver
-///     would park on its group SUBSCRIBE until the deadline and prove nothing about connections.
+///     Plain mode deliberately: it is the only mode whose initialize needs no SUBSCRIBE.
+///     <see cref="MiniBroker"/> answers CONNECT, SUBSCRIBE (unless told not to) and PINGREQ, and never
+///     PUBLISHes, so ingest is driven through <c>Subscriptions.HandleMessage</c>.
 /// </remarks>
 [Trait("Category", "Unit")]
 public sealed class MqttDriverSessionLifecycleTests
@@ -122,7 +122,7 @@ public sealed class MqttDriverSessionLifecycleTests
     ///     </para>
     ///     <para>
     ///         Ingest is fed through <c>Subscriptions.HandleMessage</c>, the method MQTTnet's dispatcher
-    ///         calls, because <see cref="MiniBroker"/> answers CONNECT and PINGREQ and nothing else: it
+    ///         calls, because <see cref="MiniBroker"/> answers CONNECT, SUBSCRIBE and PINGREQ only: it
     ///         can drop a session (which is what this test needs from it) but cannot PUBLISH.
     ///     </para>
     /// </remarks>
@@ -174,6 +174,51 @@ public sealed class MqttDriverSessionLifecycleTests
         recovered.Value.ShouldBe(21.5);
     }
 
+    /// <summary>
+    ///     <b>R06:M-1, end to end.</b> A SUBSCRIBE the broker never answers used to come back from
+    ///     <see cref="MqttDriver.SubscribeAsync"/> as a live handle while the topic left every retry set, so
+    ///     <c>DriverInstanceActor</c> never re-subscribed and the tag stayed dark behind a Healthy connection.
+    ///     It must throw (the actor's reconcile then retries), and the retry against a broker that answers
+    ///     must establish and deliver — the positive control.
+    /// </summary>
+    [Trait("Category", "SlowUnit")]
+    [Fact]
+    public async Task ASubscribeTheBrokerNeverAnswers_Throws_AndTheRetryEstablishes()
+    {
+        const string rawPath = "Plant/Mqtt/dev1/Temp";
+        const string topic = "factory/oven/temp";
+
+        using var broker = new MiniBroker { AnswerSubscribe = false };
+        var options = BrokerOptions(broker.Port) with
+        {
+            RawTags =
+            [
+                new RawTagEntry(
+                    rawPath,
+                    $$"""{"topic":"{{topic}}","payloadFormat":"Scalar","dataType":"Float64"}""",
+                    WriteIdempotent: false),
+            ],
+        };
+        await using var driver = new MqttDriver(options, "d", null);
+        await driver.InitializeAsync(Json(options), TestContext.Current.CancellationToken);
+
+        await Should.ThrowAsync<InvalidOperationException>(() =>
+            driver.SubscribeAsync([rawPath], TimeSpan.Zero, TestContext.Current.CancellationToken));
+        driver.GetHealth().State.ShouldBe(DriverState.Healthy, "the connection itself is fine — that was the trap");
+
+        broker.AnswerSubscribe = true;
+        var handle = await driver.SubscribeAsync([rawPath], TimeSpan.Zero, TestContext.Current.CancellationToken);
+
+        var raised = new List<DataChangeEventArgs>();
+        driver.OnDataChange += (_, e) => { lock (raised) raised.Add(e); };
+        driver.Subscriptions.HandleMessage(topic, "7"u8.ToArray(), retained: false);
+        lock (raised)
+        {
+            raised.ShouldHaveSingleItem().SubscriptionHandle.ShouldBe(handle);
+            raised[0].Snapshot.StatusCode.ShouldBe(0u);
+        }
+    }
+
     private static async Task WaitForAsync(Func<bool> condition, string what)
     {
         var deadline = DateTime.UtcNow + TimeSpan.FromSeconds(10);

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

@@ -538,17 +538,171 @@ public sealed class MqttSubscriptionManagerTests
         mgr.Values.Read(OvenTempPath).StatusCode.ShouldBe(BadCommunicationError);
     }
 
+    /// <summary>
+    ///     A SUBSCRIBE that never got a verdict (transport throw / deadline) degrades its refs AND throws,
+    ///     so <c>DriverInstanceActor</c> holds no handle and its health-tick reconcile re-subscribes. It
+    ///     used to return a live-looking handle with the topic dropped from every retry set — dark behind a
+    ///     Healthy connection until a redeploy.
+    /// </summary>
     [Fact]
-    public async Task SubscribeAsync_TransportThrows_DegradesRefsToBad_DoesNotThrow()
+    public async Task SubscribeAsync_TransportThrows_DegradesRefsToBad_AndThrows()
     {
         var transport = new FakeTransport { Throw = new InvalidOperationException("broker gone") };
         var mgr = NewManager(transport);
         mgr.Register([Tag(OvenTempPath, $$"""{"topic":"{{OvenTopic}}","payloadFormat":"Scalar","dataType":"Float64"}""")]);
+        var raised = new List<DataChangeEventArgs>();
+        mgr.OnDataChange += (_, e) => raised.Add(e);
 
-        await Should.NotThrowAsync(() =>
+        await Should.ThrowAsync<Exception>(() =>
             mgr.SubscribeAsync([OvenTempPath], TimeSpan.FromSeconds(1), TestContext.Current.CancellationToken));
 
         mgr.Values.Read(OvenTempPath).StatusCode.ShouldBe(BadCommunicationError);
+        // The Bad verdict still reached the subscriber before the throw.
+        raised.ShouldContain(e => e.FullReference == OvenTempPath && e.Snapshot.StatusCode == BadCommunicationError);
+    }
+
+    /// <summary>
+    ///     The reconcile path: once the actor re-subscribes after the throw, the topic IS re-SUBSCRIBEd and
+    ///     delivery resumes on the new handle — the positive control for the throw above.
+    /// </summary>
+    [Fact]
+    public async Task SubscribeAsync_AfterATransportThrow_TheRetrySubscribesAndDelivers()
+    {
+        var transport = new FakeTransport { Throw = new InvalidOperationException("broker gone") };
+        var mgr = NewManager(transport);
+        mgr.Register([Tag(OvenTempPath, $$"""{"topic":"{{OvenTopic}}","payloadFormat":"Scalar","dataType":"Float64"}""")]);
+        await Should.ThrowAsync<Exception>(() =>
+            mgr.SubscribeAsync([OvenTempPath], TimeSpan.FromSeconds(1), TestContext.Current.CancellationToken));
+
+        transport.Throw = null;
+        var handle = await mgr.SubscribeAsync([OvenTempPath], TimeSpan.FromSeconds(1), TestContext.Current.CancellationToken);
+        var raised = new List<DataChangeEventArgs>();
+        mgr.OnDataChange += (_, e) => raised.Add(e);
+        mgr.HandleMessage(OvenTopic, "21.5"u8.ToArray(), retained: false);
+
+        transport.Calls.Count.ShouldBe(2);
+        transport.Calls[1].Select(f => f.Topic).ShouldBe([OvenTopic]);
+        raised.ShouldHaveSingleItem().SubscriptionHandle.ShouldBe(handle);
+        raised[0].Snapshot.Value.ShouldBe(21.5);
+    }
+
+    /// <summary>
+    ///     A topic whose first SUBSCRIBE was refused stays in a pending set the reconnect re-subscribe
+    ///     covers. The reconnect used to re-SUBSCRIBE only the granted set, so it was never retried.
+    /// </summary>
+    [Fact]
+    public async Task OnReconnectedAsync_RetriesATopicWhoseFirstSubscribeWasRefused()
+    {
+        var transport = new FakeTransport { Reject = _ => true };
+        var mgr = NewManager(transport);
+        mgr.Register([Tag(OvenTempPath, $$"""{"topic":"{{OvenTopic}}","payloadFormat":"Scalar","dataType":"Float64"}""")]);
+        var handle = await mgr.SubscribeAsync([OvenTempPath], TimeSpan.FromSeconds(1), TestContext.Current.CancellationToken);
+
+        transport.Reject = _ => false;
+        await mgr.OnReconnectedAsync(TestContext.Current.CancellationToken);
+
+        transport.Calls.Count.ShouldBe(2);
+        transport.Calls[1].Select(f => f.Topic).ShouldBe([OvenTopic]);
+
+        var raised = new List<DataChangeEventArgs>();
+        mgr.OnDataChange += (_, e) => raised.Add(e);
+        mgr.HandleMessage(OvenTopic, "3"u8.ToArray(), retained: false);
+        raised.ShouldHaveSingleItem().SubscriptionHandle.ShouldBe(handle);
+        raised[0].Snapshot.StatusCode.ShouldBe(Good);
+    }
+
+    /// <summary>
+    ///     A persistently ACL-denied topic must not turn every reconnect into a teardown loop: a pending
+    ///     topic refused again is not "re-established nothing" — nothing that was live was lost. Both a
+    ///     mixed set (the granted topic still re-subscribes) and a pending-only set are covered.
+    /// </summary>
+    [Fact]
+    public async Task OnReconnectedAsync_PendingTopicsRefusedAgain_DoesNotThrow()
+    {
+        var transport = new FakeTransport { Reject = t => t == "f/oven/press" };
+        var mgr = NewManager(transport);
+        mgr.Register(
+        [
+            Tag(OvenTempPath, $$"""{"topic":"{{OvenTopic}}","payloadFormat":"Scalar","dataType":"Float64"}"""),
+            Tag(OvenPressPath, """{"topic":"f/oven/press","payloadFormat":"Scalar","dataType":"Float64"}"""),
+        ]);
+        await mgr.SubscribeAsync([OvenTempPath, OvenPressPath], TimeSpan.FromSeconds(1), TestContext.Current.CancellationToken);
+
+        await Should.NotThrowAsync(() => mgr.OnReconnectedAsync(TestContext.Current.CancellationToken));
+        transport.Calls[1].Select(f => f.Topic).OrderBy(t => t, StringComparer.Ordinal)
+            .ShouldBe(["f/oven/press", OvenTopic]);
+
+        var lone = new FakeTransport { Reject = _ => true };
+        var mgr2 = NewManager(lone);
+        mgr2.Register([Tag(OvenPressPath, """{"topic":"f/oven/press","payloadFormat":"Scalar","dataType":"Float64"}""")]);
+        await mgr2.SubscribeAsync([OvenPressPath], TimeSpan.FromSeconds(1), TestContext.Current.CancellationToken);
+
+        await Should.NotThrowAsync(() => mgr2.OnReconnectedAsync(TestContext.Current.CancellationToken));
+        lone.Calls.Count.ShouldBe(2);   // it WAS retried
+    }
+
+    /// <summary>
+    ///     A later subscribe on the same session (a redeploy's bulk re-subscribe) also retries the pending
+    ///     set, so a broker ACL fixed while connected recovers without waiting for a reconnect.
+    /// </summary>
+    [Fact]
+    public async Task SubscribeAsync_RetriesPendingTopicsFromAnEarlierRefusal()
+    {
+        var transport = new FakeTransport { Reject = t => t == "f/oven/press" };
+        var mgr = NewManager(transport);
+        mgr.Register(
+        [
+            Tag(OvenTempPath, $$"""{"topic":"{{OvenTopic}}","payloadFormat":"Scalar","dataType":"Float64"}"""),
+            Tag(OvenPressPath, """{"topic":"f/oven/press","payloadFormat":"Scalar","dataType":"Float64"}"""),
+        ]);
+        await mgr.SubscribeAsync([OvenPressPath], TimeSpan.FromSeconds(1), TestContext.Current.CancellationToken);
+
+        transport.Reject = _ => false;
+        await mgr.SubscribeAsync([OvenTempPath], TimeSpan.FromSeconds(1), TestContext.Current.CancellationToken);
+
+        transport.Calls[1].Select(f => f.Topic).OrderBy(t => t, StringComparer.Ordinal)
+            .ShouldBe(["f/oven/press", OvenTopic]);
+    }
+
+    /// <summary>
+    ///     <b>R06:M-11.</b> A subscribe issued before any transport is attached reaches no broker, so it must
+    ///     not be recorded as established. It used to be — "subscribed, receives nothing" — and the first
+    ///     reconnect then had nothing to re-establish. Now it stays pending, and the first reconnect on the
+    ///     attached session attempts it: the Bad verdict from the unconnected session is the proof.
+    /// </summary>
+    [Fact]
+    public async Task SubscribeAsync_WithNoTransport_StaysPending_AndTheFirstReconnectAttemptsIt()
+    {
+        var mgr = NewManager(transport: null);
+        mgr.Register([Tag(OvenTempPath, $$"""{"topic":"{{OvenTopic}}","payloadFormat":"Scalar","dataType":"Float64"}""")]);
+        await mgr.SubscribeAsync([OvenTempPath], TimeSpan.FromSeconds(1), TestContext.Current.CancellationToken);
+        mgr.Values.Read(OvenTempPath).StatusCode.ShouldBe(0x80320000u);   // BadWaitingForInitialData
+
+        await using var attached = new MqttConnection(
+            new MqttDriverOptions { Host = "127.0.0.1", Port = 1, UseTls = false }, "mqtt-1");
+        mgr.AttachTo(attached);
+
+        // The unconnected session gives no verdict, so the reconnect throws — and it ATTEMPTED the topic.
+        await Should.ThrowAsync<Exception>(() => mgr.OnReconnectedAsync(TestContext.Current.CancellationToken));
+        mgr.Values.Read(OvenTempPath).StatusCode.ShouldBe(BadCommunicationError);
+    }
+
+    /// <summary>
+    ///     Register drops a pending topic whose tags the redeploy removed — the same housekeeping the
+    ///     granted set gets — so a deleted tag's denied topic is not retried forever.
+    /// </summary>
+    [Fact]
+    public async Task Register_DroppingATag_RemovesItsTopicFromThePendingSet()
+    {
+        var transport = new FakeTransport { Reject = _ => true };
+        var mgr = NewManager(transport);
+        mgr.Register([Tag(OvenPressPath, """{"topic":"f/oven/press","payloadFormat":"Scalar","dataType":"Float64"}""")]);
+        await mgr.SubscribeAsync([OvenPressPath], TimeSpan.FromSeconds(1), TestContext.Current.CancellationToken);
+
+        mgr.Register([Tag(OvenTempPath, $$"""{"topic":"{{OvenTopic}}","payloadFormat":"Scalar","dataType":"Float64"}""")]);
+        await mgr.OnReconnectedAsync(TestContext.Current.CancellationToken);
+
+        transport.Calls.Count.ShouldBe(1);   // nothing left to retry
     }
 
     // ---------------------------------------------------------------------------------
@@ -939,7 +1093,10 @@ public sealed class MqttSubscriptionManagerTests
             new MqttDriverOptions { Host = "127.0.0.1", Port = 1, UseTls = false }, "mqtt-1");
         mgr.AttachTo(rebound);
 
-        await mgr.SubscribeAsync([OvenTempPath], TimeSpan.FromSeconds(1), TestContext.Current.CancellationToken);
+        // The unconnected session gives no verdict, so the subscribe now THROWS (M-1) — the throw is a
+        // second proof the SUBSCRIBE was attempted against the new session.
+        await Should.ThrowAsync<Exception>(() =>
+            mgr.SubscribeAsync([OvenTempPath], TimeSpan.FromSeconds(1), TestContext.Current.CancellationToken));
 
         mgr.Values.Read(OvenTempPath).StatusCode.ShouldBe(
             BadCommunicationError,