fix(mqtt): prune LastValueCache to the authored set on Register (R06:M-2)

Both ingest paths now drop cache entries whose RawPath is no longer authored when a
deploy re-registers, so a removed tag's value and every non-authored subscribe
reference's BadNodeIdUnknown marker no longer accumulate for the driver's lifetime.
Still-authored tags keep their observed values.

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

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

@@ -45,6 +45,30 @@ public sealed class LastValueCache
         _values[rawPath] = snapshot;
     }
 
+    /// <summary>How many RawPaths currently hold an entry. Diagnostics and the bounded-growth tests.</summary>
+    internal int Count => _values.Count;
+
+    /// <summary>
+    ///     Drops every entry whose RawPath <paramref name="keep"/> rejects — called by each ingest path's
+    ///     <c>Register</c> with "is this still an authored tag", so the cache is bounded by the CURRENT
+    ///     authored set rather than by the union of every RawPath (and every non-authored reference ever
+    ///     subscribed) across the driver's lifetime (R06:M-2).
+    /// </summary>
+    /// <param name="keep">Returns <see langword="true"/> for a RawPath whose entry must survive.</param>
+    /// <returns>How many entries were dropped.</returns>
+    public int RetainOnly(Func<string, bool> keep)
+    {
+        ArgumentNullException.ThrowIfNull(keep);
+        var dropped = 0;
+        foreach (var rawPath in _values.Keys)
+        {
+            if (!keep(rawPath) && _values.TryRemove(rawPath, out _))
+                dropped++;
+        }
+
+        return dropped;
+    }
+
     /// <summary>
     ///     Read the last observed value for <paramref name="rawPath"/>. Never throws: a
     ///     <c>null</c>/empty <paramref name="rawPath"/> or a RawPath never observed both return a

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

@@ -325,6 +325,11 @@ public sealed class MqttSubscriptionManager
         _authored = table;
         _warnedRefs.Clear();
 
+        // Bound the shared last-value cache by the CURRENT authored set (R06:M-2): a dropped tag's value
+        // and every non-authored reference's BadNodeIdUnknown marker would otherwise live forever. A
+        // still-authored tag keeps its value, so a redeploy never blanks what has been observed.
+        Values.RetainOnly(table.ByRawPath.ContainsKey);
+
         // Drop state belonging to tags this deploy removed. Without this, a topic whose last tag was
         // deleted stays in _subscribedTopics and is dutifully re-SUBSCRIBEd on every future reconnect,
         // forever — and its RawPaths keep a handle mapping that can never fire again.

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

@@ -416,6 +416,11 @@ public sealed class SparkplugIngestor
         _authored = table;
         _warned.Clear();
 
+        // Bound the shared last-value cache by the CURRENT authored set (R06:M-2): a dropped tag's value
+        // and every non-authored reference's BadNodeIdUnknown marker would otherwise live forever. A
+        // still-authored tag keeps its value, so a redeploy never blanks what has been observed.
+        Values.RetainOnly(table.ByRawPath.ContainsKey);
+
         // Drop state belonging to edge nodes this deploy removed. Without this a tracker (and its
         // rebirth debounce slot) for a node nobody reads survives every future redeploy.
         foreach (var key in _trackers.Keys)

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

@@ -664,6 +664,33 @@ public sealed class MqttSubscriptionManagerTests
             .ShouldBe(["f/oven/press", OvenTopic]);
     }
 
+    /// <summary>
+    ///     <b>R06:M-2.</b> Register prunes the cache to the authored set — an unauthored reference's
+    ///     BadNodeIdUnknown marker and a dropped tag's value go. Control: a still-authored tag keeps its
+    ///     value across the re-register (a redeploy must not blank observed values).
+    /// </summary>
+    [Fact]
+    public async Task Register_PrunesCacheEntriesForReferencesNoLongerAuthored()
+    {
+        var mgr = NewManager(new FakeTransport());
+        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, "Plant/Mqtt/broker1/Nope"], TimeSpan.Zero, TestContext.Current.CancellationToken);
+        mgr.HandleMessage(OvenTopic, "1"u8.ToArray(), retained: false);
+        mgr.HandleMessage("f/oven/press", "2"u8.ToArray(), retained: false);
+        mgr.Values.Read("Plant/Mqtt/broker1/Nope").StatusCode.ShouldBe(BadNodeIdUnknown);
+
+        mgr.Register([Tag(OvenTempPath, $$"""{"topic":"{{OvenTopic}}","payloadFormat":"Scalar","dataType":"Float64"}""")]);
+
+        mgr.Values.Read(OvenTempPath).Value.ShouldBe(1.0);
+        mgr.Values.Read(OvenPressPath).StatusCode.ShouldBe(0x80320000u);
+        mgr.Values.Read("Plant/Mqtt/broker1/Nope").StatusCode.ShouldBe(0x80320000u);
+        mgr.Values.Count.ShouldBe(1);
+    }
+
     /// <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

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

@@ -852,6 +852,28 @@ public sealed class SparkplugIngestorTests
         ing.TryResolve(TempPath, out _).ShouldBeFalse();
     }
 
+    /// <summary>
+    ///     <b>R06:M-2.</b> Register prunes the shared cache to the authored set: an unauthored reference's
+    ///     BadNodeIdUnknown marker and a dropped tag's value go, so the cache cannot grow with every RawPath
+    ///     and stray reference ever seen. Control: a still-authored tag's value survives the re-register.
+    /// </summary>
+    [Fact]
+    public async Task Register_PrunesCacheEntriesForReferencesNoLongerAuthored()
+    {
+        var ing = NewIngestor();
+        await ing.SubscribeAsync(["Plant/Mqtt/spb/NotAThing"], TimeSpan.Zero, TestContext.Current.CancellationToken);
+        var kept = new DataValueSnapshot(1.0f, 0u, null, DateTime.UtcNow);
+        ing.Values.Update(TempPath, kept);
+        ing.Values.Update(PressPath, kept);
+
+        ing.Register([Tag(TempPath, Blob(Group, Node, Device, "Temperature", "Float32"))]);
+
+        ing.Values.Read(TempPath).ShouldBe(kept);
+        ing.Values.Read(PressPath).StatusCode.ShouldBe(0x80320000u);              // dropped tag: no stale value
+        ing.Values.Read("Plant/Mqtt/spb/NotAThing").StatusCode.ShouldBe(0x80320000u);
+        ing.Values.Count.ShouldBe(1);
+    }
+
     [Fact]
     public async Task SubscribeAsync_UnauthoredReference_ReportsBadNodeIdUnknown()
     {