fix(mqtt): session-rebind subscriptions, re-init teardown, timeout clamp

Findings from the per-module review of Driver.Mqtt at b990adae
(code-reviews/Driver.Mqtt/findings.md records all ten; these are the fixed
ones).

Driver.Mqtt-001 (High) — MqttSubscriptionManager._subscribedTopics is a claim
about the broker session it established filters on, and AttachTo never cleared
it, but the manager OUTLIVES a session rebuild (AdoptOptions keeps it whenever
the ingest identity is unchanged). After a driver-host reconnect
(ShutdownAsync + InitializeAsync, the #527 watchdog's retry) or a
ReinitializeAsync that only changed the endpoint/credentials, the host's
re-subscribe resolved an empty pending set, sent no SUBSCRIBE at all, and left
the driver Connected/Healthy on a session carrying ZERO subscriptions — no
error, no Bad status, values frozen until the new session happened to drop.
AttachTo now forgets them, mirroring SparkplugIngestor.AttachTo's reset.

Driver.Mqtt-002 (High) — InitializeCoreAsync published _connection and
_primaryHost over whatever the driver already held, so a second initialize
stranded a connected, self-supervising broker session (plus a second Sparkplug
primary-host claim) that no later dispose could reach. Reachable without any
teardown of ours via DriverInstanceActor.AdoptConfigDuringInit. Now tears down
first, under the gate both callers already hold.

Driver.Mqtt-003 (Medium) — connectTimeoutSeconds is [Range]-annotated but
unvalidated on the runtime path; 0 makes every deadline fire immediately and a
negative value throws ArgumentOutOfRangeException out of ShutdownAsync and
DisposeAsync. Normalized once, at the seam every options instance passes.

Also: ShutdownAsync keeps the live last-message instant (-004), three usings
left behind by #512 removed (-005), stale remarks corrected (-009, -010), and
docs/drivers/Mqtt.md now distinguishes the two kinds of reconnect.

Tests 708 -> 713. All three regression tests were confirmed to fail against
the un-fixed code, with two controls that pass in either state.

docs/drivers/Mqtt.md

@@ -404,6 +404,17 @@ nothing forever, so `OnReconnectedAsync` **throws** when zero filters are grante
 down rather than swallowing it. A SUBACK failure degrades just the affected RawPaths to
 `BadCommunicationError`.
 
+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`
+re-establishes every live filter. A **session rebuild** — `ShutdownAsync` + `InitializeAsync` (the #527
+comms-loss watchdog's retry path, and an `ApplyDelta` that changes the endpoint or credentials) — builds
+a *new* connection and re-attaches the *same* `MqttSubscriptionManager`, and the re-subscribe then comes
+from the host (`DriverInstanceActor.ResubscribeDesired`) rather than from the callback. `AttachTo`
+therefore **forgets the filters the previous session had established**: the dedupe set is a claim about
+one session, and carrying it across made the host's re-subscribe resolve an empty pending set, send no
+SUBSCRIBE at all, and leave the driver Connected/Healthy and permanently deaf. Sparkplug's
+`SparkplugIngestor.AttachTo` resets its own session state for the same reason.
+
 ## Write-through
 
 `IWritable` landed with [#508](https://gitea.dohertylan.com/dohertj2/lmxopcua/issues/508). Before it,

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

@@ -1,7 +1,6 @@
+// System.Buffers is load-bearing despite no type here naming it: OnApplicationMessageReceivedAsync
+// calls ReadOnlySequence<byte>.ToArray(), which is BuffersExtensions', not a member.
 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;
@@ -238,13 +237,14 @@ public sealed class MqttConnectRejectedException : Exception
 ///         client it created and reports <see cref="ObjectDisposedException"/>.
 ///     </para>
 ///     <para>
-///         <b>The publish leg is deliberately narrow.</b> <see cref="PublishAsync"/> exists to satisfy
+///         <b>The publish leg is deliberately narrow.</b> <see cref="PublishAsync"/> exists to carry
-///         <see cref="Sparkplug.IMqttPublishTransport"/> — whose entire production caller is
+///         outbound application messages for callers that decide <i>what</i> to send, not to make this a
-///         <see cref="Sparkplug.RebirthRequester"/> — not to make this a general MQTT publisher. This
+///         general MQTT publisher: <see cref="Sparkplug.IMqttPublishTransport"/> (implemented for
-///         driver has no <c>IWritable</c> leg in v1, so a Sparkplug rebirth NCMD is the <i>only</i>
+///         <see cref="Sparkplug.RebirthRequester"/>'s NCMDs), <c>MqttDriver.WriteAsync</c>'s encoded
-///         outbound application message it ever sends. Sparkplug rebirth-on-reconnect itself is decided
+///         write (Gitea #508 — the statement that this driver has no <c>IWritable</c> leg was true only
-///         by <see cref="Sparkplug.SparkplugIngestor"/> hanging off <see cref="Reconnected"/>; this type
+///         before it) and <see cref="Sparkplug.SparkplugPrimaryHost"/>'s retained STATE. Sparkplug
-///         only carries the bytes.
+///         rebirth-on-reconnect itself is decided by <see cref="Sparkplug.SparkplugIngestor"/> hanging
+///         off <see cref="Reconnected"/>; this type only carries the bytes.
 ///     </para>
 /// </remarks>
 public sealed class MqttConnection : IAsyncDisposable, IMqttSubscribeTransport, Sparkplug.IMqttPublishTransport

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

@@ -104,6 +104,18 @@ public sealed class MqttDriver
     /// <summary>How often the host-connectivity poll samples <see cref="MqttConnection.State"/>.</summary>
     private static readonly TimeSpan HostProbeInterval = TimeSpan.FromSeconds(1);
 
+    /// <summary>
+    ///     Stand-in for a non-positive authored <c>connectTimeoutSeconds</c> — see <see cref="Normalize"/>.
+    ///     Matches <see cref="MqttDriverOptions.ConnectTimeoutSeconds"/>'s own default.
+    /// </summary>
+    private const int DefaultConnectTimeoutSeconds = 15;
+
+    /// <summary>
+    ///     Stand-in for a negative authored <c>keepAliveSeconds</c> — see <see cref="Normalize"/>.
+    ///     Matches <see cref="MqttDriverOptions.KeepAliveSeconds"/>'s own default.
+    /// </summary>
+    private const int DefaultKeepAliveSeconds = 30;
+
     private readonly string _driverInstanceId;
     private readonly ILogger? _logger;
 
@@ -221,13 +233,13 @@ public sealed class MqttDriver
         ArgumentNullException.ThrowIfNull(options);
         ArgumentException.ThrowIfNullOrWhiteSpace(driverInstanceId);
 
-        _options = options;
+        _options = Normalize(options);
         _driverInstanceId = driverInstanceId;
         _logger = logger;
         _primaryRoleSource = primaryRoleSource ?? UnmanagedDriverPrimaryRoleSource.Instance;
-        _subscriptions = CreateSubscriptionManager(options);
+        _subscriptions = CreateSubscriptionManager(_options);
-        _sparkplug = CreateSparkplugIngestor(options);
+        _sparkplug = CreateSparkplugIngestor(_options);
-        RegisterAuthoredTags(options);
+        RegisterAuthoredTags(_options);
     }
 
     /// <inheritdoc />
@@ -372,7 +384,12 @@ public sealed class MqttDriver
     /// <inheritdoc />
     public async Task ShutdownAsync(CancellationToken cancellationToken)
     {
-        var lastMessage = ReadHealth().LastSuccessfulRead;
+        // GetHealth(), not ReadHealth(): the live connection's LastMessageUtc is the real last-message
+        // instant, and the stored snapshot only ever carries what was true at the last initialize. Taking
+        // the stored one here throws away the evidence a shutdown-then-initialize cycle (the #527
+        // watchdog's reconnect) is measured against, reporting a driver that was delivering a moment ago
+        // as one that has never delivered.
+        var lastMessage = GetHealth().LastSuccessfulRead;
         if (await _lifecycleGate.WaitAsync(TimeSpan.FromSeconds(_options.ConnectTimeoutSeconds), cancellationToken)
                 .ConfigureAwait(false))
         {
@@ -932,6 +949,17 @@ public sealed class MqttDriver
         WriteHealth(new DriverHealth(DriverState.Initializing, ReadHealth().LastSuccessfulRead, null));
         try
         {
+            // Tear down whatever this driver already holds BEFORE building a new session. Both fields
+            // published below (_connection, _primaryHost) are IAsyncDisposable and were previously
+            // overwritten, so a second initialize on a live driver stranded a connected, supervised
+            // broker session — plus a whole second primary-host connection — that no later dispose
+            // could reach. Reachable without any teardown of ours: DriverInstanceActor re-runs
+            // InitializeAsync with no ShutdownAsync when an ApplyDelta lands while the driver is in
+            // Connecting/Reconnecting (AdoptConfigDuringInit). Idempotent, so the ReinitializeAsync
+            // rebuild branch calling TeardownAsync first costs a no-op, and a first initialize sees
+            // three nulls.
+            await TeardownAsync().ConfigureAwait(false);
+
             // A blank / unusable blob keeps the constructor's options rather than degrading the
             // driver to defaults pointing at localhost.
             var parsed = ParseOptions(driverConfigJson);
@@ -1030,7 +1058,8 @@ public sealed class MqttDriver
             // deliberately: enums must round-trip by NAME. A second, divergent
             // JsonSerializerOptions is this repo's documented systemic enum bug, where an
             // AdminUI-authored config with a string enum field faults the driver.
-            return JsonSerializer.Deserialize<MqttDriverOptions>(driverConfigJson, MqttJson.Options);
+            var parsed = JsonSerializer.Deserialize<MqttDriverOptions>(driverConfigJson, MqttJson.Options);
+            return parsed is null ? null : Normalize(parsed);
         }
         catch (JsonException ex)
         {
@@ -1050,6 +1079,54 @@ public sealed class MqttDriver
         }
     }
 
+    /// <summary>
+    ///     Replaces out-of-range duration knobs with their documented defaults, once, at the single seam
+    ///     every options instance this driver holds passes through (the constructor and
+    ///     <see cref="ParseOptions"/>).
+    /// </summary>
+    /// <remarks>
+    ///     <para>
+    ///         <see cref="MqttDriverOptions"/> carries <c>[Range]</c> annotations, but nothing validates
+    ///         them on the runtime path: the blob is deserialized straight out of the deploy artifact, so
+    ///         a config authored (or hand-edited) with <c>connectTimeoutSeconds: 0</c> or a negative value
+    ///         reaches the driver unchecked. Both are load-bearing, and neither fails where an operator
+    ///         would look:
+    ///     </para>
+    ///     <list type="bullet">
+    ///         <item>
+    ///             zero makes every <c>new CancellationTokenSource(TimeSpan.Zero)</c> in the connect,
+    ///             subscribe and publish paths fire immediately, so the driver can never connect and
+    ///             reports a timeout against a perfectly healthy broker;
+    ///         </item>
+    ///         <item>
+    ///             a negative value is outside <see cref="SemaphoreSlim.WaitAsync(TimeSpan)"/>'s and
+    ///             <see cref="CancellationTokenSource"/>'s legal range, so it throws
+    ///             <see cref="ArgumentOutOfRangeException"/> out of <see cref="ShutdownAsync"/> and
+    ///             <see cref="DisposeAsync"/> — the two paths whose whole job is to not fail.
+    ///         </item>
+    ///     </list>
+    ///     <para>
+    ///         Normalising here rather than at each use site keeps one answer for the whole driver: the
+    ///         value also rides into <see cref="MqttConnection"/>, <see cref="SparkplugPrimaryHost"/> and
+    ///         <c>MqttClientOptionsFactory</c> through <see cref="_options"/>. <c>MaxPayloadBytes</c> and
+    ///         the reconnect backoffs are deliberately left alone — the manager's constructor and
+    ///         <see cref="MqttConnection.NextBackoff"/> already clamp those at their own use site.
+    ///     </para>
+    /// </remarks>
+    private static MqttDriverOptions Normalize(MqttDriverOptions options)
+    {
+        var connectTimeout = options.ConnectTimeoutSeconds > 0
+            ? options.ConnectTimeoutSeconds
+            : DefaultConnectTimeoutSeconds;
+
+        // Zero keep-alive is legal MQTT ("no keep-alive"); only a negative one is nonsense.
+        var keepAlive = options.KeepAliveSeconds >= 0 ? options.KeepAliveSeconds : DefaultKeepAliveSeconds;
+
+        return connectTimeout == options.ConnectTimeoutSeconds && keepAlive == options.KeepAliveSeconds
+            ? options
+            : options with { ConnectTimeoutSeconds = connectTimeout, KeepAliveSeconds = keepAlive };
+    }
+
     /// <summary>
     ///     Adopts new options, rebuilding the subscription manager only when a setting it captures at
     ///     construction actually changed. The rebuilt manager inherits the driver-owned

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

@@ -183,17 +183,35 @@ public sealed class MqttSubscriptionManager
     ///     transport.
     /// </summary>
     /// <remarks>
-    ///     The manager does not own the connection's lifetime and never detaches — the driver disposes
+    ///     <para>
-    ///     both together. Call once per connection, and <b>before</b> the first
+    ///         The manager does not own the connection's lifetime and never detaches — the driver disposes
-    ///     <see cref="SubscribeAsync"/>: a subscribe issued with no transport attached records its
+    ///         both together. Call once per connection, and <b>before</b> the first
-    ///     intent (and warns) but sends nothing to a broker, and would only be established if the
+    ///         <see cref="SubscribeAsync"/>: a subscribe issued with no transport attached records its
-    ///     connection later dropped and recovered.
+    ///         intent (and warns) but sends nothing to a broker, and would only be established if the
+    ///         connection later dropped and recovered.
+    ///     </para>
+    ///     <para>
+    ///         <b>Binding to a session forgets the previous session's established filters</b> — the same
+    ///         reset <see cref="Sparkplug.SparkplugIngestor.AttachTo"/> performs, and for the same reason:
+    ///         this manager <i>outlives</i> a session rebuild. <c>MqttDriver.AdoptOptions</c> only
+    ///         replaces it when the ingest identity changed, so a driver-host reconnect (ShutdownAsync +
+    ///         InitializeAsync, #527's watchdog path) and a <c>ReinitializeAsync</c> that changes only the
+    ///         endpoint/credentials both re-attach <i>this</i> instance to a brand-new broker session.
+    ///         <c>_subscribedTopics</c> is a claim about the session that has just gone: carried across,
+    ///         the host's post-reconnect re-subscribe resolves an empty pending set, issues no SUBSCRIBE
+    ///         at all, and leaves a driver that reports Connected/Healthy and receives nothing until the
+    ///         <i>new</i> session happens to drop — the silent-death outcome every other path in this
+    ///         type is written to prevent.
+    ///     </para>
     /// </remarks>
     /// <param name="connection">The connection to observe.</param>
     public void AttachTo(MqttConnection connection)
     {
         ArgumentNullException.ThrowIfNull(connection);
 
+        // FIRST, before delivery is wired: a new session holds none of the old one's subscriptions.
+        _subscribedTopics.Clear();
+
         _transport = connection;
         connection.MessageReceived += HandleMessage;
         connection.Reconnected += OnReconnectedAsync;

src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt/Sparkplug/SequenceTracker.cs

@@ -285,11 +285,19 @@ public sealed class SequenceTracker
     ///     session token.
     /// </summary>
     /// <remarks>
-    ///     The reconnect seam (§3.6 invariant #5 — late join). After a dropped connection the driver
+    ///     <para>
-    ///     has missed an unknown number of messages, so the baseline it was holding is no longer
+    ///         The rationale is §3.6 invariant #5 (late join): after a dropped connection the driver has
-    ///     evidence of anything: carrying it across would let the stream look contiguous purely
+    ///         missed an unknown number of messages, so the baseline it was holding is no longer evidence
-    ///     because the edge node happened to publish the right number of messages while the driver was
+    ///         of anything — carrying it across would let the stream look contiguous purely because the
-    ///     away.
+    ///         edge node happened to publish the right number of messages while the driver was away.
+    ///     </para>
+    ///     <para>
+    ///         <b>The reconnect path does not call this</b>, despite the reasoning being its.
+    ///         <c>SparkplugIngestor.ResetSessionState</c> drops the whole per-node tracker <i>map</i>
+    ///         instead, which additionally releases the entry for a node the plant has stopped
+    ///         publishing. This member is the same reset for a single tracker, kept for a caller holding
+    ///         one directly (the unit suite is currently the only one).
+    ///     </para>
     /// </remarks>
     public void Reset()
     {

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

@@ -0,0 +1,133 @@
+using System.Diagnostics;
+using System.Text.Json;
+using Shouldly;
+using Xunit;
+using ZB.MOM.WW.OtOpcUa.Core.Abstractions;
+
+namespace ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Tests;
+
+/// <summary>
+///     The driver's own session lifecycle, against the <see cref="MiniBroker"/> fake: what
+///     <see cref="MqttDriver.InitializeAsync"/> does to the session it is <i>replacing</i>, and what a
+///     mis-authored duration knob does to the two paths whose whole job is to not fail
+///     (<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.
+/// </remarks>
+[Trait("Category", "Unit")]
+public sealed class MqttDriverSessionLifecycleTests
+{
+    /// <summary>
+    ///     <b>Driver.Mqtt-002.</b> <c>InitializeCoreAsync</c> publishes <c>_connection</c> — an
+    ///     <see cref="IAsyncDisposable"/> — so re-initializing a live driver without tearing down first
+    ///     stranded a connected, supervised broker session that no later dispose could reach, and the
+    ///     abandoned session's reconnect supervisor kept it alive against the broker indefinitely.
+    /// </summary>
+    /// <remarks>
+    ///     <para>
+    ///         Reachable with no teardown of the driver's own: <c>DriverInstanceActor</c> re-runs
+    ///         <c>InitializeAsync</c> with no <c>ShutdownAsync</c> when an <c>ApplyDelta</c> arrives while
+    ///         the driver is Connecting/Reconnecting (<c>AdoptConfigDuringInit</c>). The
+    ///         <c>teardownFirst</c> flag on the #527 retry path covers only the retry path.
+    ///     </para>
+    ///     <para>
+    ///         The assertion is the broker's <b>live socket count</b>, not the driver's own view: a leak
+    ///         is precisely the case where the driver believes it holds one session and the broker is
+    ///         serving two. <see cref="MiniBroker.AcceptedCount"/> is asserted alongside it as the
+    ///         control — without it, a second initialize that never dialled at all would satisfy
+    ///         "exactly one live connection" just as well.
+    ///     </para>
+    /// </remarks>
+    [Fact]
+    public async Task InitializeAsync_OnALiveDriver_TearsDownThePreviousSession()
+    {
+        using var broker = new MiniBroker();
+        var options = BrokerOptions(broker.Port);
+        await using var driver = new MqttDriver(options, "d", null);
+
+        await driver.InitializeAsync(Json(options), TestContext.Current.CancellationToken);
+        await WaitForAsync(() => broker.LiveConnections == 1, "the first session to be established");
+
+        await driver.InitializeAsync(Json(options), TestContext.Current.CancellationToken);
+
+        broker.AcceptedCount.ShouldBe(2, "the second initialize must genuinely open a new session");
+        await WaitForAsync(
+            () => broker.LiveConnections == 1,
+            "the replaced session to be torn down rather than left live and unreachable");
+        driver.GetHealth().State.ShouldBe(DriverState.Healthy);
+    }
+
+    /// <summary>
+    ///     <b>Driver.Mqtt-003.</b> <c>connectTimeoutSeconds</c> carries a <c>[Range]</c> annotation that
+    ///     nothing validates on the runtime path — the blob is deserialized straight out of the deploy
+    ///     artifact. A negative value is outside the legal range of both
+    ///     <see cref="SemaphoreSlim.WaitAsync(TimeSpan)"/> and <see cref="CancellationTokenSource"/>, so
+    ///     it threw <see cref="ArgumentOutOfRangeException"/> out of shutdown and dispose — the two calls
+    ///     that must always complete, and the ones a host makes while tearing a bad deployment down.
+    /// </summary>
+    [Fact]
+    public async Task ShutdownAndDispose_WithANegativeConnectTimeout_DoNotThrow()
+    {
+        var options = BrokerOptions(1) with { ConnectTimeoutSeconds = -1, KeepAliveSeconds = -1 };
+        var driver = new MqttDriver(options, "d", null);
+
+        await Should.NotThrowAsync(async () =>
+            await driver.ShutdownAsync(TestContext.Current.CancellationToken));
+        await Should.NotThrowAsync(async () => await driver.DisposeAsync());
+    }
+
+    /// <summary>
+    ///     The control for the test above: the normalization must not quietly rewrite a <i>legal</i>
+    ///     value, or a deployment's authored connect budget would silently become the default.
+    /// </summary>
+    [Fact]
+    public async Task InitializeAsync_WithAnAuthoredConnectTimeout_KeepsIt()
+    {
+        using var broker = new MiniBroker { Blackhole = true };
+        var options = BrokerOptions(broker.Port) with { ConnectTimeoutSeconds = 2 };
+        await using var driver = new MqttDriver(options, "d", null);
+
+        var sw = Stopwatch.StartNew();
+        await Should.ThrowAsync<TimeoutException>(async () =>
+            await driver.InitializeAsync(Json(options), TestContext.Current.CancellationToken));
+        sw.Stop();
+
+        sw.Elapsed.ShouldBeLessThan(
+            TimeSpan.FromSeconds(10),
+            "the authored 2s budget must still bound the connect, not the 15s default");
+    }
+
+    private static async Task WaitForAsync(Func<bool> condition, string what)
+    {
+        var deadline = DateTime.UtcNow + TimeSpan.FromSeconds(10);
+        while (DateTime.UtcNow < deadline)
+        {
+            if (condition())
+            {
+                return;
+            }
+
+            await Task.Delay(25, TestContext.Current.CancellationToken);
+        }
+
+        throw new Xunit.Sdk.XunitException($"Timed out waiting for {what}.");
+    }
+
+    private static MqttDriverOptions BrokerOptions(int port) => new()
+    {
+        Mode = MqttMode.Plain,
+        Host = "127.0.0.1",
+        Port = port,
+        UseTls = false,
+        ProtocolVersion = MqttProtocolVersion.V311,
+        KeepAliveSeconds = 300,
+        ConnectTimeoutSeconds = 5,
+        ReconnectMinBackoffSeconds = 1,
+        ReconnectMaxBackoffSeconds = 2,
+    };
+
+    private static string Json(MqttDriverOptions options) => JsonSerializer.Serialize(options, MqttJson.Options);
+}

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

@@ -827,6 +827,66 @@ public sealed class MqttSubscriptionManagerTests
         transport.Calls.Count.ShouldBe(1);   // the initial subscribe only — no zombie re-subscribe
     }
 
+    // ---------------------------------------------------------------------------------
+    // Session rebind — this manager OUTLIVES the broker session it established filters on
+    // ---------------------------------------------------------------------------------
+
+    /// <summary>
+    ///     <b>Driver.Mqtt-001.</b> A driver-host reconnect (ShutdownAsync + InitializeAsync, the #527
+    ///     watchdog's path) builds a brand-new <see cref="MqttConnection"/> but keeps <i>this</i> manager
+    ///     — <c>MqttDriver.AdoptOptions</c> only replaces it when the ingest identity changed. The new
+    ///     session holds none of the old one's subscriptions, so the host's post-reconnect re-subscribe
+    ///     must actually reach the broker. While <c>_subscribedTopics</c> survived the rebind it resolved
+    ///     an empty pending set, sent no SUBSCRIBE at all, and left the driver Connected/Healthy and
+    ///     permanently deaf.
+    /// </summary>
+    /// <remarks>
+    ///     The rebound connection is deliberately a real, <b>unconnected</b> one: its
+    ///     <c>SubscribeAsync</c> refuses ("no established session"), which <c>EstablishAsync</c> catches
+    ///     and turns into <c>BadCommunicationError</c> on the affected references. That Bad IS the proof
+    ///     the subscribe was attempted — with the bug the tag keeps the Good value it was last delivered,
+    ///     because nothing is attempted at all.
+    /// </remarks>
+    [Fact]
+    public async Task AttachTo_ForgetsTheFiltersEstablishedOnThePreviousSession()
+    {
+        var transport = new FakeTransport();
+        var mgr = NewManager(transport);
+        mgr.Register([Tag(OvenTempPath, $$"""{"topic":"{{OvenTopic}}","payloadFormat":"Scalar","dataType":"Float64"}""")]);
+        await mgr.SubscribeAsync([OvenTempPath], TimeSpan.FromSeconds(1), TestContext.Current.CancellationToken);
+        mgr.HandleMessage(OvenTopic, "21.5"u8.ToArray(), retained: false);
+        mgr.Values.Read(OvenTempPath).StatusCode.ShouldBe(Good);
+
+        // The session underneath is rebuilt and this manager is re-attached to it.
+        await using var rebound = new MqttConnection(
+            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);
+
+        mgr.Values.Read(OvenTempPath).StatusCode.ShouldBe(
+            BadCommunicationError,
+            "the re-subscribe must be issued against the NEW session, not skipped as already-established");
+    }
+
+    /// <summary>
+    ///     The falsifiability control for the test above: within ONE session the dedupe must still hold,
+    ///     or the fix would simply be "re-SUBSCRIBE everything on every subscribe" — which re-triggers
+    ///     every topic's retained seed and is the behaviour <c>_subscribedTopics</c> exists to prevent.
+    /// </summary>
+    [Fact]
+    public async Task SubscribeAsync_TwiceOnTheSameSession_StillIssuesOneSubscribe()
+    {
+        var transport = new FakeTransport();
+        var mgr = NewManager(transport);
+        mgr.Register([Tag(OvenTempPath, $$"""{"topic":"{{OvenTopic}}","payloadFormat":"Scalar","dataType":"Float64"}""")]);
+
+        await mgr.SubscribeAsync([OvenTempPath], TimeSpan.FromSeconds(1), TestContext.Current.CancellationToken);
+        await mgr.SubscribeAsync([OvenTempPath], TimeSpan.FromSeconds(1), TestContext.Current.CancellationToken);
+
+        transport.Calls.Count.ShouldBe(1);
+    }
+
     /// <summary>Records every subscribe call so dedupe / re-subscribe can be asserted without a broker.</summary>
     private sealed class FakeTransport : IMqttSubscribeTransport
     {