test(mtconnect): make the STAB-20 retention test deterministic again (review I-2)

29c829b5 let the test accept 'already finished' for the stale-session pump, which
since T-1 is nearly always true, so deleting the retention reintroduced STAB-20
silently. A new CannedAgentClient.HoldSampleExit knob keeps that pump alive past
the retire point; the test now asserts strictly that it is still running
(positive control) AND retained, then releases it and asserts shutdown joined
it. Red with the _supersededPumps.Add removed. The pump's own self-exit on a
gone generation is covered by MTConnectStaleCursorTests (pump A completes, the
current pump keeps streaming).

Refs #743

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

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

@@ -178,6 +178,13 @@ internal sealed class CannedAgentClient : IMTConnectAgentClient, IDisposable
     /// </summary>
     public TaskCompletionSource? CurrentGate { get; set; }
 
+    /// <summary>
+    ///     When set, a <c>/sample</c> enumeration that is cancelled or whose client is disposed does NOT end until this
+    ///     source completes — a pump that cannot unwind (a wedged socket read). The deterministic way to keep a retired
+    ///     pump alive past the point a lifecycle restart retires it.
+    /// </summary>
+    public TaskCompletionSource? HoldSampleExit { get; set; }
+
     /// <summary>
     ///     Signalled the moment a <c>/current</c> request <b>lands</b> — before
     ///     <see cref="CurrentGate"/> parks it. This is the deterministic "the caller is now inside
@@ -360,6 +367,11 @@ internal sealed class CannedAgentClient : IMTConnectAgentClient, IDisposable
             {
                 scripted = await chunks.Reader.ReadAsync(lifetime.Token).ConfigureAwait(false);
             }
+            catch (Exception ex) when (HoldSampleExit is { } hold && ex is OperationCanceledException or ChannelClosedException)
+            {
+                await hold.Task.ConfigureAwait(false);
+                throw;
+            }
             catch (ChannelClosedException)
             {
                 // Mirrors the production client: a stream that stops for any reason other than the

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

@@ -37,6 +37,7 @@ public sealed class MTConnectPumpGenerationTests
         await driver.InitializeAsync("{}", Ct);
         Task[] retired = [];
 
+        var holdStalePump = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously);
         var blockOnce = 0;
         var blocking = new ManualResetEventSlim(false);
         var entered = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously);
@@ -68,7 +69,10 @@ public sealed class MTConnectPumpGenerationTests
                 (++spins).ShouldBeLessThan(500, "restart never cleared the pump slot");
             }
 
-            // A second subscriber arrives mid-restart: it starts a pump on the (still-live) old client.
+            // A second subscriber arrives mid-restart: it starts a pump on the (still-live) old client. That pump is
+            // held alive past the retire point (it cannot unwind until released), so the retention is observable
+            // deterministically rather than racing the pump's own exit.
+            clientA.HoldSampleExit = holdStalePump;
             await driver.SubscribeAsync(["dev1_pos"], TimeSpan.FromMilliseconds(50), Ct);
             var staleSessionPump = driver.SampleStreamTask.ShouldNotBeNull();
             await clientA.WaitForSampleCallsAsync(2).WaitAsync(Watchdog, Ct);
@@ -82,8 +86,8 @@ public sealed class MTConnectPumpGenerationTests
             // pump also stops by itself once it sees its session generation is gone, so it may already have
             // finished before the restart got to retire it — either way it is not left running unjoined.
             retired = driver.SupersededPumpTasks;
-            (retired.Contains(staleSessionPump) || staleSessionPump.IsCompleted).ShouldBeTrue(
-                "the stale-session pump must be retained for shutdown to join, or already be gone");
+            staleSessionPump.IsCompleted.ShouldBeFalse("positive control: the stale pump is still running here");
+            retired.ShouldContain(staleSessionPump, "a retired pump that is still running must be retained for shutdown to join");
 
             // The restart installed a new client; the standing subscription must be streaming from IT.
             await clientB.WaitForSampleCallsAsync(1).WaitAsync(Watchdog, Ct);
@@ -92,6 +96,7 @@ public sealed class MTConnectPumpGenerationTests
         finally
         {
             blocking.Set();
+            holdStalePump.TrySetResult();
             await driver.ShutdownAsync(CancellationToken.None);
         }