test+docs(mqtt): end-to-end Bad-then-identical-replay recovery

archreview/plans/R3-02-stale-good-family-plan.md

@@ -635,3 +635,19 @@ test could only pass once the production change landed.
   source announcing its own offline state, the stale is our transport losing the broker (and matches
   what the host's S15 fan-out writes). The existing const's "no driver emits BadNoCommunication at all"
   remark was corrected rather than left to rot.
+
+### T5 — MQTT driver-level end-to-end recovery + docs
+
+- The test is green on first run by construction (T2–T4 already landed), so the plan's TDD bar was met
+  the other way: **the wiring was deliberately broken and the test seen to fail.** Deleting
+  `connection.ConnectionLost += OnConnectionLost;` from `MqttSubscriptionManager.AttachTo` ⇒
+  `ABrokerDrop_MarksCachedRefsBadNoCommunication_AndTheIdenticalReplayRecovers` **FAILED** with
+  `Timed out waiting for the broker drop to publish a BadNoCommunication snapshot` (15 s). Restored,
+  green. That is the point of this test: the manager-level suites drive `OnConnectionLost()` directly
+  and stay green against an unforwarded seam.
+- The outage edge is a real socket drop through MiniBroker → MQTTnet `DisconnectedAsync`; ingest is fed
+  through `Subscriptions.HandleMessage` because MiniBroker answers CONNECT/PINGREQ only and cannot
+  PUBLISH.
+- Docs: `docs/drivers/Mqtt.md` §"Quality on connection loss and re-subscribe" (links the contract in
+  `DriverLifecycle.md`, records the two distinct "not communicating" codes and the operator-visible
+  behaviour change).

archreview/plans/R3-02-stale-good-family-plan.md.tasks.json

@@ -33,7 +33,7 @@
     {
       "id": "T5",
       "subject": "MQTT driver-level end-to-end recovery test (broker drop -> Bad -> identical retained replay -> Good) + docs/drivers/Mqtt.md section",
-      "status": "pending",
+      "status": "done",
       "blockedBy": [
         "T4"
       ]

docs/drivers/Mqtt.md

@@ -415,6 +415,41 @@ one session, and carrying it across made the host's re-subscribe resolve an empt
 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.
 
+## Quality on connection loss and re-subscribe
+
+The quality contract this implements is stated once, in
+[`docs/DriverLifecycle.md` §"The quality contract"](../DriverLifecycle.md#the-quality-contract).
+Two edge behaviours make MQTT — a push-model driver — honour it. A poll driver gets both free from
+`PollGroupEngine` (a fresh subscription's first poll runs with `forceRaise: true`); MQTT had neither,
+which is archreview 05/STAB-18.
+
+- **Connection loss stales, keeping the value.** On the `Connected → Reconnecting` edge
+  `MqttConnection.ConnectionLost` fires, and both ingest paths mark every **already-observed**
+  reference `BadNoCommunication` with its **last value retained** and publish the change. A
+  never-observed reference (still `BadWaitingForInitialData`) and an already-Bad one are left alone.
+  The event does **not** fire for a failed *initial* connect or during dispose — neither is an outage.
+- **Every (re)subscribe pushes the cached snapshots.** `SubscribeAsync` republishes each observed
+  reference's cached snapshot onto the new handle before any broker traffic. This is what refills a
+  node an address-space rebuild reseeded `BadWaitingForInitialData` on a quiet topic — a rebuild
+  triggers no retained replay and, in Sparkplug, no rebirth.
+
+**Why the stale is what makes recovery work.** The plain path suppresses a message whose value *and*
+status equal the driver-owned cache, and that cache survives reconnects and redeploys. So the retained
+message a broker replays on reconnect — routinely the identical payload — was suppressed, and every
+node the host had latched Bad stayed Bad until the plant value happened to change. Staling the status
+first makes the identical replay a change. (A regression test for this must replay the **identical**
+payload; a different value defeats the suppression on its own and proves nothing.)
+
+**Two "not communicating" codes, deliberately.** `BadNoCommunication` (0x80310000) means *the link is
+down* — the connection-loss stale, and what the host's comms-loss fan-out writes onto the value nodes.
+`BadCommunicationError` (0x80050000) keeps its narrower Sparkplug meaning: the **source announced its
+own death** (NDEATH/DDEATH), or a birth voided the scope feeding the tag. A refused SUBSCRIBE filter
+also degrades to `BadCommunicationError`.
+
+**Operator-visible change:** an MQTT tag now goes Bad when the broker session drops, rather than
+holding its last value at Good until the host's 90 s comms-loss watchdog notices — which, since
+MQTTnet usually reconnects well inside that window, in practice meant never.
+
 ## Write-through
 
 `IWritable` landed with [#508](https://gitea.dohertylan.com/dohertj2/lmxopcua/issues/508). Before it,

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

@@ -100,6 +100,79 @@ public sealed class MqttDriverSessionLifecycleTests
             "the authored 2s budget must still bound the connect, not the 15s default");
     }
 
+    /// <summary>
+    ///     <b>End-to-end through <see cref="MqttDriver"/> (archreview 05/STAB-18).</b> Deliver a value,
+    ///     let the real broker drop the real session, and the driver publishes
+    ///     <c>BadNoCommunication</c> with the value retained; feed the <b>identical</b> payload back —
+    ///     the retained replay a broker sends on reconnect — and it publishes Good again.
+    /// </summary>
+    /// <remarks>
+    ///     <para>
+    ///         <b>What this adds over <c>MqttSubscriptionManagerRecoveryTests</c>:</b> the wiring. Those
+    ///         drive <c>OnConnectionLost()</c> directly, so they stay green even if
+    ///         <c>MqttSubscriptionManager.AttachTo</c> never subscribes to
+    ///         <see cref="MqttConnection.ConnectionLost"/> — an unforwarded seam is this repo's
+    ///         best-documented way to ship an inert fix. Here the edge comes from a genuine socket drop
+    ///         through MQTTnet's own <c>DisconnectedAsync</c>.
+    ///     </para>
+    ///     <para>
+    ///         <b>Vacuity rule:</b> the recovery payload is byte-identical to the pre-outage one. A test
+    ///         that replays a <i>different</i> value passes against a build with no stale at all — the
+    ///         value change alone would defeat the suppression — and is not the guard.
+    ///     </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
+    ///         can drop a session (which is what this test needs from it) but cannot PUBLISH.
+    ///     </para>
+    /// </remarks>
+    [Fact]
+    public async Task ABrokerDrop_MarksCachedRefsBadNoCommunication_AndTheIdenticalReplayRecovers()
+    {
+        const string rawPath = "Plant/Mqtt/dev1/Temp";
+        const string topic = "factory/oven/temp";
+        const uint good = 0x00000000u;
+        const uint badNoCommunication = 0x80310000u;
+
+        using var broker = new MiniBroker();
+        var options = BrokerOptions(broker.Port) with
+        {
+            RawTags =
+            [
+                new RawTagEntry(
+                    rawPath,
+                    $$"""{"topic":"{{topic}}","payloadFormat":"Json","jsonPath":"$.value","dataType":"Float64"}""",
+                    WriteIdempotent: false),
+            ],
+        };
+        await using var driver = new MqttDriver(options, "d", null);
+        await driver.InitializeAsync(Json(options), TestContext.Current.CancellationToken);
+        await driver.SubscribeAsync([rawPath], TimeSpan.Zero, TestContext.Current.CancellationToken);
+
+        var raised = new List<DataValueSnapshot>();
+        driver.OnDataChange += (_, e) => { lock (raised) raised.Add(e.Snapshot); };
+
+        driver.Subscriptions.HandleMessage(topic, """{"value":21.5}"""u8.ToArray(), retained: false);
+        lock (raised) raised.Count.ShouldBe(1, "the first value must publish");
+
+        broker.DropAll();
+        await WaitForAsync(
+            () => { lock (raised) return raised.Count == 2; },
+            "the broker drop to publish a BadNoCommunication snapshot");
+        DataValueSnapshot staled;
+        lock (raised) staled = raised[1];
+        staled.StatusCode.ShouldBe(badNoCommunication);
+        staled.Value.ShouldBe(21.5, "the last known value is retained through the outage");
+
+        // The reconnect's retained replay — the IDENTICAL payload.
+        driver.Subscriptions.HandleMessage(topic, """{"value":21.5}"""u8.ToArray(), retained: true);
+
+        DataValueSnapshot recovered;
+        lock (raised) { raised.Count.ShouldBe(3); recovered = raised[2]; }
+        recovered.StatusCode.ShouldBe(good);
+        recovered.Value.ShouldBe(21.5);
+    }
+
     private static async Task WaitForAsync(Func<bool> condition, string what)
     {
         var deadline = DateTime.UtcNow + TimeSpan.FromSeconds(10);