fix(mtconnect): generation guard on the pump's cursor publish, chunks and re-baseline (R06:T-1)

A pump the lifecycle retired or abandoned that finished after the restart
installed the new session published its stale cursor onto it (PublishAgentCursor
only checked for a null or identical session), so the next pump read its first
chunk as an Agent restart: index cleared, a Bad flicker on every subscribed tag,
a needless re-baseline. The pump now carries the session generation it was
started against; the cursor publish is checked under _subscriptionLock (where the
session install and generation bump are atomic) and a superseded pump also stops
folding late chunks / re-baselines into the new session's index.

Refs #743

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

docs/drivers/MTConnect.md

@@ -324,6 +324,13 @@ the two need disambiguating, not just OR-ing together). On a changed
 `/hosts` page — see [Known limitations](#known-limitations) for what that does
 and does not do.
 
+**A pump carries the session generation it was started against (review 2026-10-21 T-1).** A lifecycle restart
+retires the running pump (or abandons it after its bounded teardown wait); if that pump finishes after the new
+session is installed, its cursor publish, a late chunk and a late re-baseline are all dropped — checked under the
+same lock the new session is installed and the generation bumped under. It used to land the OLD Agent's
+`instanceId` + sequence on the new session, so the next pump read its first chunk as an Agent restart (index
+cleared, one Bad flicker on every subscribed tag, a needless re-baseline).
+
 ## Write-back — observation injection, off by default
 
 > ⚠️ **A write sets what the Agent *reports*, not what the machine does.** The value

src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.MTConnect/MTConnectDriver.cs

@@ -1856,9 +1856,10 @@ public sealed partial class MTConnectDriver
         var cts = new CancellationTokenSource();
         _pumpCts = cts;
         _pumpGeneration = _sessionGeneration;
+        var generation = _sessionGeneration;
         _pumpTask = Task.Run(
             () => RunSampleStreamAsync(
-                session.Client, options, session.NextSequence, session.InstanceId, republishOnStart, cts.Token),
+                session.Client, options, session.NextSequence, session.InstanceId, republishOnStart, generation, cts.Token),
             CancellationToken.None);
     }
 
@@ -1894,6 +1895,7 @@ public sealed partial class MTConnectDriver
         long from,
         long instanceId,
         bool republishOnStart,
+        long generation,
         CancellationToken ct)
     {
         var cursor = from;
@@ -1929,7 +1931,7 @@ public sealed partial class MTConnectDriver
                     }
                 }
 
-                var outcome = await ConsumeStreamAsync(client, options, cursor, instanceId, ct).ConfigureAwait(false);
+                var outcome = await ConsumeStreamAsync(client, options, cursor, instanceId, generation, ct).ConfigureAwait(false);
                 cursor = outcome.Cursor;
                 instanceId = outcome.InstanceId;
 
@@ -2015,8 +2017,9 @@ public sealed partial class MTConnectDriver
             // would therefore reopen at a sequence the Agent has long since evicted: it self-heals
             // via OUT_OF_RANGE, but only after a wasted round trip, and a still-buffered stale
             // cursor replays historical observations to subscribers first. One publish per pump
-            // lifetime, dropped if a lifecycle change has already installed a newer session.
-            PublishAgentCursor(instanceId, cursor);
+            // lifetime, dropped if a lifecycle change has already installed a newer session (R06:T-1: the
+            // generation this pump was started against is what says so — the session alone cannot).
+            PublishAgentCursor(generation, instanceId, cursor);
         }
     }
 
@@ -2030,6 +2033,7 @@ public sealed partial class MTConnectDriver
         MTConnectDriverOptions options,
         long cursor,
         long instanceId,
+        long generation,
         CancellationToken ct)
     {
         // The RUNNING cursor the gap check is made against: the previous chunk's nextSequence, equal
@@ -2052,6 +2056,14 @@ public sealed partial class MTConnectDriver
                     continue;
                 }
 
+                // R06:T-1 — a pump whose session a lifecycle restart has replaced (it was retired, or its
+                // teardown wait was abandoned) must not fold a late chunk of the OLD Agent connection into
+                // the NEW session's index — nor read it as an Agent restart and clear that index.
+                if (!IsCurrentSessionGeneration(generation))
+                {
+                    return new StreamOutcome(StreamVerdict.Cancelled, expected, instanceId, delivered, null);
+                }
+
                 // CHECKED BEFORE THE GAP: an Agent restart changes instanceId *and* resets
                 // sequences, so a restart usually trips IsSequenceGap too — and the two need
                 // different handling. A gap keeps the held values (they are still this device's);
@@ -2076,7 +2088,7 @@ public sealed partial class MTConnectDriver
                     // DiscoverAsync.
                     AnnounceAgentRestart(chunk.InstanceId, options);
 
-                    return await RebaselineAsync(client, options, expected, instanceId, delivered, ct)
+                    return await RebaselineAsync(client, options, expected, instanceId, delivered, generation, ct)
                         .ConfigureAwait(false);
                 }
 
@@ -2086,7 +2098,7 @@ public sealed partial class MTConnectDriver
                         "MTConnect driver {DriverInstanceId} fell out of {AgentUri}'s buffer (expected sequence {Expected}, oldest retained {FirstSequence}); re-baselining from /current.",
                         _driverInstanceId, options.AgentUri, expected, chunk.FirstSequence);
 
-                    return await RebaselineAsync(client, options, expected, instanceId, delivered, ct)
+                    return await RebaselineAsync(client, options, expected, instanceId, delivered, generation, ct)
                         .ConfigureAwait(false);
                 }
 
@@ -2136,7 +2148,7 @@ public sealed partial class MTConnectDriver
                 "MTConnect driver {DriverInstanceId} could not resume {AgentUri}'s /sample stream at sequence {Cursor} (the Agent rejected it as out of range); re-baselining from /current.",
                 _driverInstanceId, options.AgentUri, cursor);
 
-            return await RebaselineAsync(client, options, expected, instanceId, delivered, ct)
+            return await RebaselineAsync(client, options, expected, instanceId, delivered, generation, ct)
                 .ConfigureAwait(false);
         }
         catch (Exception ex)
@@ -2166,6 +2178,7 @@ public sealed partial class MTConnectDriver
         long fallbackCursor,
         long fallbackInstanceId,
         bool deliveredAny,
+        long generation,
         CancellationToken ct)
     {
         try
@@ -2173,7 +2186,13 @@ public sealed partial class MTConnectDriver
             var current = await BoundedAsync(client.CurrentAsync, "/current", options.RequestTimeoutMs, ct)
                 .ConfigureAwait(false);
 
-            PublishAgentCursor(current.InstanceId, current.NextSequence);
+            if (!IsCurrentSessionGeneration(generation))
+            {
+                return new StreamOutcome(
+                    StreamVerdict.Cancelled, fallbackCursor, fallbackInstanceId, deliveredAny, null);
+            }
+
+            PublishAgentCursor(generation, current.InstanceId, current.NextSequence);
             ApplyAndFanOut(current, baseline: true);
 
             _streamDegraded = false;
@@ -2488,25 +2507,51 @@ public sealed partial class MTConnectDriver
     /// </remarks>
     /// <param name="instanceId">The Agent instanceId last observed.</param>
     /// <param name="nextSequence">The sequence a <c>/sample</c> stream should resume from.</param>
-    private void PublishAgentCursor(long instanceId, long nextSequence)
+    /// <param name="generation">The session generation the publishing pump was started against.</param>
+    private void PublishAgentCursor(long generation, long instanceId, long nextSequence)
     {
-        while (true)
+        // R06:T-1 — under _subscriptionLock, the lock StartCoreAsync installs a new session AND bumps the
+        // generation under: inside it the two are always observed together, so a pump bound to a retired
+        // session can never land its cursor on the session that replaced it (a lock-free generation check
+        // could read the new session beside the not-yet-bumped generation). Nothing is awaited here.
+        lock (_subscriptionLock)
         {
-            var session = Volatile.Read(ref _session);
-            if (session is null ||
-                (session.InstanceId == instanceId && session.NextSequence == nextSequence))
+            if (_sessionGeneration != generation)
             {
                 return;
             }
 
-            var updated = session with { InstanceId = instanceId, NextSequence = nextSequence };
-            if (ReferenceEquals(Interlocked.CompareExchange(ref _session, updated, session), session))
+            while (true)
             {
-                return;
+                var session = Volatile.Read(ref _session);
+                if (session is null ||
+                    (session.InstanceId == instanceId && session.NextSequence == nextSequence))
+                {
+                    return;
+                }
+
+                // Still a compare-and-swap: InvalidateProbeModel / GetOrFetchProbeModelAsync swap the session
+                // lock-free, and a plain write here could undo a probe-model drop.
+                var updated = session with { InstanceId = instanceId, NextSequence = nextSequence };
+                if (ReferenceEquals(Interlocked.CompareExchange(ref _session, updated, session), session))
+                {
+                    return;
+                }
             }
         }
     }
 
+    /// <summary>Whether <paramref name="generation"/> is still the installed session's generation.</summary>
+    /// <param name="generation">A pump's session generation.</param>
+    /// <returns><see langword="false"/> once a lifecycle restart has installed a newer session.</returns>
+    private bool IsCurrentSessionGeneration(long generation)
+    {
+        lock (_subscriptionLock)
+        {
+            return _sessionGeneration == generation;
+        }
+    }
+
     /// <summary>
     ///     The <c>/probe</c> device model: the cached copy when warm, otherwise a fresh
     ///     deadline-bounded <c>/probe</c> that re-populates the cache. This is the accessor every

tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.MTConnect.Tests/MTConnectStaleCursorTests.cs

@@ -0,0 +1,110 @@
+using Shouldly;
+using Xunit;
+using ZB.MOM.WW.OtOpcUa.Core.Abstractions;
+
+namespace ZB.MOM.WW.OtOpcUa.Driver.MTConnect.Tests;
+
+/// <summary>
+///     Review 2026-10-21 R06:T-1 — a pump the lifecycle abandoned (its teardown wait timed out) that finished after
+///     the restart installed the NEW session published its stale cursor onto it: the old Agent's instanceId and
+///     sequence replaced the new session's, so the next pump to start treated its first chunk as an Agent restart
+///     (index cleared, a Bad flicker on every subscribed tag, a needless re-baseline). The pump now carries the
+///     session generation it was started against and its cursor publish is dropped once that generation is gone.
+/// </summary>
+public sealed class MTConnectStaleCursorTests
+{
+    private const long InstanceB = 1655000099L;
+    private const long NextSequenceB = 500L;
+    private static readonly TimeSpan Watchdog = TimeSpan.FromSeconds(30);
+
+    private static CancellationToken Ct => TestContext.Current.CancellationToken;
+
+    /// <summary>The abandoned pump's late cursor publish leaves the new session's cursor alone.</summary>
+    [Fact]
+    public async Task An_abandoned_pump_does_not_publish_its_cursor_onto_the_new_session()
+    {
+        var clientA = CannedAgentClient.FromFixtures();
+        var clientB = CannedAgentClient.FromFixtures();
+        clientB.Current = new MTConnectStreamsResult(InstanceB, NextSequenceB, NextSequenceB - 10, clientB.Current.Observations);
+        var clients = new Queue<CannedAgentClient>([clientA, clientB]);
+
+        var driver = new MTConnectDriver(
+            new MTConnectDriverOptions
+            {
+                AgentUri = "http://fixture-agent:5000",
+                RequestTimeoutMs = 5000,
+                Tags = [new MTConnectTagDefinition("dev1_pos", DriverDataType.Float64)],
+            },
+            "mt-t1",
+            _ => clients.Dequeue());
+        await driver.InitializeAsync("{}", Ct);
+
+        var wedgeOnce = 0;
+        var wedge = new ManualResetEventSlim(false);
+        var entered = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously);
+        driver.OnDataChange += (_, e) =>
+        {
+            if (e.Snapshot.Value is not 1.5d || Interlocked.Exchange(ref wedgeOnce, 1) != 0) return;
+            entered.TrySetResult();
+            wedge.Wait();
+        };
+
+        try
+        {
+            await driver.SubscribeAsync(["dev1_pos"], TimeSpan.FromMilliseconds(50), Ct);
+            var pumpA = driver.SampleStreamTask.ShouldNotBeNull();
+            _ = clientA.PumpAsync(MTConnectPumpGenerationChunks.Pos(108, 113, "1.5"));
+            await entered.Task.WaitAsync(Watchdog, Ct);
+
+            // The restart gives up on the wedged pump (bounded teardown wait), retires clientA and installs B.
+            await driver.InitializeAsync("{}", Ct).WaitAsync(Watchdog, Ct);
+            driver.AgentInstanceId.ShouldBe(InstanceB, "positive control: the new session is installed");
+            pumpA.IsCompleted.ShouldBeFalse("pump A must still be alive — it is the late writer under test");
+
+            // Now let the abandoned pump finish: its finally publishes (instanceA, 113).
+            wedge.Set();
+            await pumpA.WaitAsync(Watchdog, Ct);
+
+            driver.AgentInstanceId.ShouldBe(InstanceB, "a superseded pump must not write the old Agent's instanceId");
+            driver.AgentNextSequence.ShouldBe(NextSequenceB, "nor its cursor");
+        }
+        finally
+        {
+            wedge.Set();
+            await driver.ShutdownAsync(CancellationToken.None);
+        }
+    }
+
+    /// <summary>
+    ///     Positive control: the CURRENT pump's own cursor publish still lands — a last unsubscribe stops it, and the
+    ///     session's cursor moves to where that pump got to.
+    /// </summary>
+    [Fact]
+    public async Task The_current_pump_still_publishes_its_cursor_when_it_stops()
+    {
+        var client = CannedAgentClient.FromFixtures();
+        var driver = new MTConnectDriver(
+            new MTConnectDriverOptions
+            {
+                AgentUri = "http://fixture-agent:5000",
+                RequestTimeoutMs = 5000,
+                Tags = [new MTConnectTagDefinition("dev1_pos", DriverDataType.Float64)],
+            },
+            "mt-t1-ok",
+            _ => client);
+        await driver.InitializeAsync("{}", Ct);
+        try
+        {
+            var handle = await driver.SubscribeAsync(["dev1_pos"], TimeSpan.FromMilliseconds(50), Ct);
+            await client.PumpAsync(MTConnectPumpGenerationChunks.Pos(108, 113, "1.5")).WaitAsync(Watchdog, Ct);
+
+            await driver.UnsubscribeAsync(handle, Ct);
+
+            driver.AgentNextSequence.ShouldBe(113);
+        }
+        finally
+        {
+            await driver.ShutdownAsync(CancellationToken.None);
+        }
+    }
+}