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

@@ -0,0 +1,29 @@
+using ZB.MOM.WW.OtOpcUa.Core.Abstractions;
+
+namespace ZB.MOM.WW.OtOpcUa.Driver.MTConnect;
+
+/// <summary>
+///     Driver-internal identity of one <c>/sample</c> subscription, matching the sibling drivers'
+///     shape (Galaxy's <c>GalaxySubscriptionHandle</c>, the shared <c>PollGroupEngine</c>'s
+///     <c>PollSubscriptionHandle</c>): a monotonic per-driver id plus a diagnostic string that
+///     carries it into logs.
+/// </summary>
+/// <remarks>
+///     <para>
+///         <b>Every handle shares ONE Agent stream.</b> Unlike a polled driver, where each
+///         subscription owns a loop, MTConnect's <c>/sample</c> long poll is per-Agent: the driver
+///         opens exactly one and fans each chunk out to whichever handles subscribe the reporting
+///         DataItem. The handle is therefore purely an identity — it owns no connection, no task,
+///         and no cursor — and dropping one only stops the stream when it was the last.
+///     </para>
+///     <para>
+///         A <see langword="record"/> for value equality on the id, so a handle that has round-tripped
+///         through the caller still resolves. The id is never reused within a driver instance.
+///     </para>
+/// </remarks>
+/// <param name="SubscriptionId">The monotonic per-driver subscription id.</param>
+internal sealed record MTConnectSampleHandle(long SubscriptionId) : ISubscriptionHandle
+{
+    /// <inheritdoc/>
+    public string DiagnosticId => $"mtconnect-sub-{SubscriptionId}";
+}

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

@@ -1,3 +1,4 @@
+using System.Collections.Concurrent;
 using System.Runtime.CompilerServices;
 using System.Threading.Channels;
 
@@ -34,9 +35,19 @@ namespace ZB.MOM.WW.OtOpcUa.Driver.MTConnect.Tests;
 /// </remarks>
 internal sealed class CannedAgentClient : IMTConnectAgentClient, IDisposable
 {
-    private readonly Channel<ScriptedChunk> _chunks = Channel.CreateUnbounded<ScriptedChunk>();
     private readonly CancellationTokenSource _disposeCts = new();
 
+    /// <summary>Chunks a test has scripted but not yet pumped — see <see cref="ScriptChunks"/>.</summary>
+    private readonly ConcurrentQueue<MTConnectStreamsResult> _script = new();
+
+    /// <summary>
+    ///     The <b>current</b> <c>/sample</c> stream generation. Replaced (not merely completed) by
+    ///     <see cref="EndStream"/> so that a pump which reconnects after a dropped stream gets a
+    ///     live channel to read from instead of one that is permanently closed — without that, a
+    ///     reconnect test could only ever observe the reconnect failing.
+    /// </summary>
+    private Channel<ScriptedChunk> _chunks = Channel.CreateUnbounded<ScriptedChunk>();
+
     private int _probeCallCount;
     private int _currentCallCount;
     private int _sampleCallCount;
@@ -66,6 +77,44 @@ internal sealed class CannedAgentClient : IMTConnectAgentClient, IDisposable
     public static MTConnectStreamsResult Chunk(string fixture) =>
         MTConnectStreamsParser.Parse(File.ReadAllText(fixture));
 
+    /// <summary>
+    ///     An Agent scripted for the ring-buffer-overflow story: the driver primes from
+    ///     <c>Fixtures/current.xml</c> (<c>nextSequence</c> 108), then the first scripted chunk is
+    ///     <c>Fixtures/sample-gap.xml</c> — whose <c>firstSequence</c> (5000) is strictly newer than
+    ///     the cursor, i.e. the buffer rolled past the driver — and the second is an ordinary chunk
+    ///     that continues contiguously from the re-baseline.
+    /// </summary>
+    /// <remarks>
+    ///     The <b>second</b> <c>/current</c> answer is what makes this a story rather than a single
+    ///     event: it advertises the buffer the gap chunk revealed (<c>firstSequence</c> 5000 /
+    ///     <c>nextSequence</c> 5005), so a driver that re-baselines and resumes from that answer sees
+    ///     the follow-up chunk as contiguous. A driver that re-baselined but resumed from the stale
+    ///     cursor would report a gap on every subsequent chunk instead.
+    /// </remarks>
+    public static CannedAgentClient WithGapThenResume()
+    {
+        var client = FromFixtures();
+        var gap = Chunk("Fixtures/sample-gap.xml");
+
+        // 1st /current = the InitializeAsync prime (the fixture). 2nd = the post-gap re-baseline.
+        client.CurrentAnswers.Enqueue(client.Current);
+        client.CurrentAnswers.Enqueue(
+            new MTConnectStreamsResult(gap.InstanceId, gap.NextSequence, gap.FirstSequence, gap.Observations));
+
+        client.ScriptChunks(
+            gap,
+            new MTConnectStreamsResult(
+                gap.InstanceId,
+                gap.NextSequence + 5,
+                gap.NextSequence,
+                [
+                    new MTConnectObservation(
+                        "dev1_pos", "201.5000", new DateTime(2026, 7, 24, 12, 6, 0, DateTimeKind.Utc)),
+                ]));
+
+        return client;
+    }
+
     /// <summary>The document <c>/probe</c> answers with. Settable so a test can re-shape the model.</summary>
     public MTConnectProbeModel Probe { get; set; }
 
@@ -81,6 +130,31 @@ internal sealed class CannedAgentClient : IMTConnectAgentClient, IDisposable
     /// <summary>When set, <c>/current</c> throws this instead of answering.</summary>
     public Exception? CurrentFailure { get; set; }
 
+    /// <summary>
+    ///     Scripted <c>/current</c> answers, consumed in order: each call dequeues the next one and
+    ///     <b>promotes it to <see cref="Current"/></b>, so once the script runs out the Agent keeps
+    ///     answering with the last thing it said rather than travelling back in time. This is how a
+    ///     test scripts "the prime, then the post-gap re-baseline" without racing the pump for a
+    ///     property setter.
+    /// </summary>
+    public ConcurrentQueue<MTConnectStreamsResult> CurrentAnswers { get; } = new();
+
+    /// <summary>
+    ///     Scripted one-shot <c>/sample</c> failures: each enumeration dequeues one and throws it.
+    ///     Models a transient stream fault the Agent recovers from — e.g. the
+    ///     <see cref="InvalidDataException"/> a real cppagent's <c>OUT_OF_RANGE</c> error document
+    ///     (served under HTTP 200) surfaces as.
+    /// </summary>
+    public ConcurrentQueue<Exception> SampleFailures { get; } = new();
+
+    /// <summary>
+    ///     A <b>sticky</b> <c>/sample</c> failure: every enumeration throws it, forever. Models a
+    ///     configuration-level fault such as
+    ///     <see cref="MTConnectStreamNotSupportedException"/>, where reconnecting reproduces the
+    ///     identical answer — the shape a pump must NOT retry-loop against.
+    /// </summary>
+    public Exception? SampleFailure { get; set; }
+
     /// <summary>
     ///     When set, the <b>next</b> <c>/probe</c> parks here until the test completes it, then the
     ///     gate clears itself so later calls answer immediately.
@@ -101,6 +175,14 @@ internal sealed class CannedAgentClient : IMTConnectAgentClient, IDisposable
     /// </summary>
     public TaskCompletionSource? CurrentGate { 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
+    ///     the request" barrier a test needs before landing a concurrent lifecycle change on it;
+    ///     without it the only alternative is polling <see cref="CurrentCallCount"/> on a timer.
+    /// </summary>
+    public TaskCompletionSource? CurrentEntered { get; set; }
+
     /// <summary>Number of <c>/probe</c> requests issued against this client.</summary>
     public int ProbeCallCount => Volatile.Read(ref _probeCallCount);
 
@@ -152,6 +234,7 @@ internal sealed class CannedAgentClient : IMTConnectAgentClient, IDisposable
         ObjectDisposedException.ThrowIf(IsDisposed, this);
         ct.ThrowIfCancellationRequested();
         Interlocked.Increment(ref _currentCallCount);
+        CurrentEntered?.TrySetResult();
 
         // The request was accepted; the answer is withheld until the test releases the gate or the
         // caller's deadline cancels the token.
@@ -161,6 +244,12 @@ internal sealed class CannedAgentClient : IMTConnectAgentClient, IDisposable
             await gate.Task.WaitAsync(ct).ConfigureAwait(false);
         }
 
+        // A scripted answer supersedes — and then becomes — the standing one.
+        if (CurrentAnswers.TryDequeue(out var scripted))
+        {
+            Current = scripted;
+        }
+
         return CurrentFailure is null ? Current : throw CurrentFailure;
     }
 
@@ -172,6 +261,21 @@ internal sealed class CannedAgentClient : IMTConnectAgentClient, IDisposable
         Interlocked.Increment(ref _sampleCallCount);
         LastSampleFrom = from;
 
+        // Thrown before the first chunk, exactly like a real client that fails its /sample request:
+        // one-shot scripts first, then the sticky "this endpoint will never stream" failure.
+        if (SampleFailures.TryDequeue(out var oneShot))
+        {
+            throw oneShot;
+        }
+
+        if (SampleFailure is { } sticky)
+        {
+            throw sticky;
+        }
+
+        // Bound to THIS generation: EndStream swaps in a fresh channel, so a consumer that
+        // reconnects reads the new one rather than the closed one it just lost.
+        var chunks = Volatile.Read(ref _chunks);
         using var lifetime = CancellationTokenSource.CreateLinkedTokenSource(ct, _disposeCts.Token);
         var delivered = 0L;
 
@@ -180,7 +284,7 @@ internal sealed class CannedAgentClient : IMTConnectAgentClient, IDisposable
             ScriptedChunk scripted;
             try
             {
-                scripted = await _chunks.Reader.ReadAsync(lifetime.Token).ConfigureAwait(false);
+                scripted = await chunks.Reader.ReadAsync(lifetime.Token).ConfigureAwait(false);
             }
             catch (ChannelClosedException)
             {
@@ -192,9 +296,18 @@ internal sealed class CannedAgentClient : IMTConnectAgentClient, IDisposable
 
             delivered++;
 
-            yield return scripted.Result;
+            try
-
+            {
-            scripted.Consumed.TrySetResult();
+                yield return scripted.Result;
+            }
+            finally
+            {
+                // Signalled when the consumer is DONE with this chunk — whether it came back for the
+                // next one or abandoned the enumeration. The finally is load-bearing: a chunk that
+                // makes the pump break out (a sequence gap, an agent restart) is never resumed past,
+                // so signalling only after the resume would hang every re-baseline test.
+                scripted.Consumed.TrySetResult();
+            }
         }
     }
 
@@ -212,7 +325,7 @@ internal sealed class CannedAgentClient : IMTConnectAgentClient, IDisposable
         var scripted = new ScriptedChunk(
             chunk, new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously));
 
-        if (!_chunks.Writer.TryWrite(scripted))
+        if (!Volatile.Read(ref _chunks).Writer.TryWrite(scripted))
         {
             throw new InvalidOperationException("The scripted /sample stream is already closed.");
         }
@@ -221,10 +334,42 @@ internal sealed class CannedAgentClient : IMTConnectAgentClient, IDisposable
     }
 
     /// <summary>
-    ///     Closes the scripted <c>/sample</c> stream, which surfaces to the consumer as an
+    ///     Queues chunks for <see cref="PumpOnce"/> to deliver one at a time. Nothing is sent until
-    ///     <see cref="MTConnectStreamEndedException"/> — the transient, reconnectable end.
+    ///     a test asks for it — the script is a plan, not a schedule.
     /// </summary>
-    public void EndStream() => _chunks.Writer.TryComplete();
+    /// <param name="chunks">The chunks the Agent should send, in order.</param>
+    public void ScriptChunks(params MTConnectStreamsResult[] chunks)
+    {
+        ArgumentNullException.ThrowIfNull(chunks);
+
+        foreach (var chunk in chunks)
+        {
+            _script.Enqueue(chunk);
+        }
+    }
+
+    /// <summary>
+    ///     Pumps the next scripted chunk (see <see cref="ScriptChunks"/>) and waits until the
+    ///     consumer has finished with it.
+    /// </summary>
+    /// <returns>A task completing once the consumer has processed — or abandoned the stream on — that chunk.</returns>
+    /// <exception cref="InvalidOperationException">The script is exhausted.</exception>
+    public Task PumpOnce() =>
+        _script.TryDequeue(out var next)
+            ? PumpAsync(next)
+            : throw new InvalidOperationException(
+                "No scripted /sample chunk left to pump; call ScriptChunks first.");
+
+    /// <summary>
+    ///     Closes the current scripted <c>/sample</c> stream, which surfaces to its consumer as an
+    ///     <see cref="MTConnectStreamEndedException"/> — the transient, reconnectable end — and opens
+    ///     a fresh one so a reconnecting consumer has somewhere to land.
+    /// </summary>
+    public void EndStream()
+    {
+        var closed = Interlocked.Exchange(ref _chunks, Channel.CreateUnbounded<ScriptedChunk>());
+        closed.Writer.TryComplete();
+    }
 
     /// <inheritdoc/>
     public void Dispose()
@@ -237,7 +382,7 @@ internal sealed class CannedAgentClient : IMTConnectAgentClient, IDisposable
         }
 
         _disposeCts.Cancel();
-        _chunks.Writer.TryComplete();
+        Volatile.Read(ref _chunks).Writer.TryComplete();
         _disposeCts.Dispose();
     }
 

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

@@ -0,0 +1,827 @@
+using System.Collections.Concurrent;
+using Shouldly;
+using Xunit;
+using ZB.MOM.WW.OtOpcUa.Core.Abstractions;
+
+namespace ZB.MOM.WW.OtOpcUa.Driver.MTConnect.Tests;
+
+/// <summary>
+///     Task 11 — <see cref="MTConnectDriver"/>'s <see cref="ISubscribable"/> half: one shared
+///     <c>/sample</c> long-poll pump behind every handle, the ring-buffer re-baseline, and the
+///     teardown that must leave nothing running. Every test runs against
+///     <see cref="CannedAgentClient"/>; no socket is opened anywhere in this file.
+/// </summary>
+/// <remarks>
+///     <para>
+///         <b>Nothing here waits on wall-clock time.</b> The fake completes a
+///         <see cref="CannedAgentClient.PumpAsync"/> only once the pump has finished with that
+///         chunk, and exposes the pump's own task
+///         (<see cref="MTConnectDriver.SampleStreamTask"/>) as the teardown barrier — so every
+///         "the pump has now done X" statement is a fact, not a sleep. The one
+///         <see cref="Watchdog"/> below is a <i>failure</i> guard that turns a genuine hang (an
+///         un-terminated retry loop, a deadlocked teardown) into a clean red test; no assertion
+///         is ever made about elapsed time.
+///     </para>
+///     <para>
+///         <b>The load-bearing tests are the ones about a stream that has gone wrong.</b> A pump
+///         that only ever meets contiguous chunks is a dozen lines; the defects live in the four
+///         ways it can be knocked off the sequence — the buffer rolling past the cursor
+///         (<c>IsSequenceGap</c>), the Agent refusing an evicted <c>from</c> outright
+///         (<c>OUT_OF_RANGE</c> under HTTP 200 ⇒ <see cref="InvalidDataException"/>), the Agent
+///         restarting (a new <c>instanceId</c>, which ALSO looks like a gap), and the connection
+///         dropping — plus the one failure it must NOT retry
+///         (<see cref="MTConnectStreamNotSupportedException"/>, where reconnecting reproduces the
+///         identical answer forever).
+///     </para>
+/// </remarks>
+public sealed class MTConnectSubscribeTests
+{
+    private const string AgentUri = "http://fixture-agent:5000";
+
+    /// <summary>The <c>instanceId</c> every canned fixture shares.</summary>
+    private const long FixtureInstanceId = 1655000000L;
+
+    /// <summary>A DIFFERENT <c>instanceId</c> — the Agent came back as a new process.</summary>
+    private const long RestartedInstanceId = 1655999999L;
+
+    /// <summary>The cursor <c>Fixtures/current.xml</c> primes the pump with.</summary>
+    private const long PrimedNextSequence = 108L;
+
+    // Canonical Opc.Ua.StatusCodes numerics, restated so the test asserts the wire value a client
+    // actually sees rather than a driver-private constant it could drift with.
+    private const uint Good = 0x00000000u;
+    private const uint BadNoCommunication = 0x80310000u;
+    private const uint BadWaitingForInitialData = 0x80320000u;
+    private const uint BadNodeIdUnknown = 0x80340000u;
+
+    /// <summary>
+    ///     Failure guard for the handful of awaits that a defect could turn into a permanent hang
+    ///     (a <c>NotSupported</c> retry loop, a teardown that deadlocks on the lifecycle semaphore).
+    ///     Deliberately far longer than any of these operations could legitimately take — it exists
+    ///     to make a hang red, not to assert a duration.
+    /// </summary>
+    private static readonly TimeSpan Watchdog = TimeSpan.FromSeconds(10);
+
+    private static readonly DateTime ObservedAt = new(2026, 7, 24, 12, 30, 0, DateTimeKind.Utc);
+
+    private static CancellationToken Ct => TestContext.Current.CancellationToken;
+
+    private static MTConnectDriverOptions Opts(MTConnectReconnectOptions? reconnect = null) =>
+        new()
+        {
+            AgentUri = AgentUri,
+            RequestTimeoutMs = 5000,
+            Reconnect = reconnect ?? new MTConnectReconnectOptions(),
+            Tags =
+            [
+                new MTConnectTagDefinition("dev1_pos", DriverDataType.Float64),
+                new MTConnectTagDefinition("dev1_execution", DriverDataType.String),
+                new MTConnectTagDefinition("dev1_partcount", DriverDataType.Int32),
+                new MTConnectTagDefinition("dev1_never_reported", DriverDataType.Int32),
+            ],
+        };
+
+    // ---- fixtures ----
+
+    private static (MTConnectDriver Driver, CannedAgentClient Client) NewDriver(
+        CannedAgentClient? client = null, MTConnectDriverOptions? options = null)
+    {
+        var agent = client ?? CannedAgentClient.FromFixtures();
+
+        return (new MTConnectDriver(options ?? Opts(), "mt1", _ => agent), agent);
+    }
+
+    private static async Task<(MTConnectDriver Driver, CannedAgentClient Client)> InitializedDriverAsync(
+        CannedAgentClient? client = null, MTConnectDriverOptions? options = null)
+    {
+        var (driver, agent) = NewDriver(client, options);
+        await driver.InitializeAsync("{}", Ct);
+
+        return (driver, agent);
+    }
+
+    /// <summary>Attaches a recorder to the driver's data-change event and returns its log.</summary>
+    private static ConcurrentQueue<DataChangeEventArgs> Record(MTConnectDriver driver)
+    {
+        var seen = new ConcurrentQueue<DataChangeEventArgs>();
+        driver.OnDataChange += (_, e) => seen.Enqueue(e);
+
+        return seen;
+    }
+
+    /// <summary>Builds a <c>/sample</c> chunk (or <c>/current</c> answer) for the fixture Agent.</summary>
+    private static MTConnectStreamsResult Chunk(
+        long firstSequence, long nextSequence, params (string Id, string Value)[] observations) =>
+        ChunkFrom(FixtureInstanceId, firstSequence, nextSequence, observations);
+
+    private static MTConnectStreamsResult ChunkFrom(
+        long instanceId, long firstSequence, long nextSequence, params (string Id, string Value)[] observations) =>
+        new(
+            instanceId,
+            nextSequence,
+            firstSequence,
+            [.. observations.Select(o => new MTConnectObservation(o.Id, o.Value, ObservedAt))]);
+
+    private static IReadOnlyList<DataChangeEventArgs> For(
+        ConcurrentQueue<DataChangeEventArgs> seen, string reference) =>
+        [.. seen.Where(e => e.FullReference == reference)];
+
+    // ---- initial data ----
+
+    /// <summary>
+    ///     The plan's first TDD case, and the OPC UA Part 4 convention: a subscription reports the
+    ///     current value immediately, out of the index the priming <c>/current</c> already filled —
+    ///     it does not make the caller wait for the Agent's next chunk, which may be a whole
+    ///     heartbeat away.
+    /// </summary>
+    [Fact]
+    public async Task Subscribe_fires_initial_data_from_current()
+    {
+        var (driver, _) = await InitializedDriverAsync();
+        var seen = Record(driver);
+
+        await driver.SubscribeAsync(["dev1_pos"], TimeSpan.FromMilliseconds(50), Ct);
+
+        seen.Select(e => e.FullReference).ShouldContain("dev1_pos");
+        var initial = seen.Single();
+        initial.Snapshot.StatusCode.ShouldBe(Good);
+        initial.Snapshot.Value.ShouldBe(123.4567d);
+    }
+
+    /// <summary>
+    ///     Initial data fires for EVERY subscribed reference, including the ones with no value yet.
+    ///     A driver that fired only for the references it happens to hold a value for would leave
+    ///     the others with no snapshot at all — indistinguishable, from the server's side, from a
+    ///     subscription that never established.
+    /// </summary>
+    [Fact]
+    public async Task Subscribe_fires_initial_data_for_every_ref_including_the_valueless_ones()
+    {
+        var (driver, _) = await InitializedDriverAsync();
+        var seen = Record(driver);
+
+        await driver.SubscribeAsync(
+            ["dev1_pos", "dev1_partcount", "dev1_never_reported", "nope"], TimeSpan.FromMilliseconds(50), Ct);
+
+        seen.Count.ShouldBe(4);
+        For(seen, "dev1_pos")[0].Snapshot.StatusCode.ShouldBe(Good);
+        For(seen, "dev1_partcount")[0].Snapshot.StatusCode.ShouldBe(BadNoCommunication);
+        For(seen, "dev1_never_reported")[0].Snapshot.StatusCode.ShouldBe(BadWaitingForInitialData);
+        For(seen, "nope")[0].Snapshot.StatusCode.ShouldBe(BadNodeIdUnknown);
+    }
+
+    /// <summary>Every event carries the handle the caller was given — that is how it demultiplexes.</summary>
+    [Fact]
+    public async Task Subscribe_returns_a_distinct_handle_per_call_and_stamps_it_on_every_event()
+    {
+        var (driver, _) = await InitializedDriverAsync();
+        var seen = Record(driver);
+
+        var a = await driver.SubscribeAsync(["dev1_pos"], TimeSpan.FromMilliseconds(50), Ct);
+        var b = await driver.SubscribeAsync(["dev1_pos"], TimeSpan.FromMilliseconds(50), Ct);
+
+        a.ShouldNotBe(b);
+        a.DiagnosticId.ShouldNotBe(b.DiagnosticId);
+        a.DiagnosticId.ShouldNotBeNullOrWhiteSpace();
+        For(seen, "dev1_pos").Select(e => e.SubscriptionHandle).ShouldBe([a, b]);
+    }
+
+    /// <summary>Subscribing to nothing costs the Agent nothing — no stream is opened for it.</summary>
+    [Fact]
+    public async Task Subscribe_of_an_empty_ref_list_opens_no_stream()
+    {
+        var (driver, client) = await InitializedDriverAsync();
+
+        var handle = await driver.SubscribeAsync([], TimeSpan.FromMilliseconds(50), Ct);
+
+        handle.ShouldNotBeNull();
+        client.SampleCallCount.ShouldBe(0);
+
+        // Asserted on the pump itself, not only on the call count: a pump that HAD been started
+        // might simply not have reached its first request yet, and would pass the count alone.
+        driver.SampleStreamTask.ShouldBeNull();
+
+        // …and it still unsubscribes cleanly, so a caller that conditionally adds refs later is symmetric.
+        await driver.UnsubscribeAsync(handle, Ct);
+    }
+
+    // ---- the shared stream ----
+
+    /// <summary>
+    ///     One Agent, one <c>/sample</c> long poll — no matter how many handles. MTConnect streams
+    ///     the whole device, so a stream per subscription would multiply the Agent's connection load
+    ///     by the number of OPC UA subscriptions for identical data.
+    /// </summary>
+    [Fact]
+    public async Task Two_subscriptions_share_exactly_one_sample_stream()
+    {
+        var (driver, client) = await InitializedDriverAsync();
+
+        await driver.SubscribeAsync(["dev1_pos"], TimeSpan.FromMilliseconds(50), Ct);
+        await driver.SubscribeAsync(["dev1_execution"], TimeSpan.FromMilliseconds(50), Ct);
+
+        // Barrier: a chunk can only be consumed by a running enumeration, so both subscriptions have
+        // demonstrably landed on the same one by the time this returns.
+        await client.PumpAsync(Chunk(PrimedNextSequence, 113, ("dev1_pos", "1.5"))).WaitAsync(Watchdog, Ct);
+
+        client.SampleCallCount.ShouldBe(1);
+        client.LastSampleFrom.ShouldBe(PrimedNextSequence);
+    }
+
+    /// <summary>
+    ///     A chunk updates the index for everything it carries, but only the subscribed references
+    ///     raise a callback. The Agent streams every DataItem on the device; publishing all of them
+    ///     would flood the server with values nothing asked for.
+    /// </summary>
+    [Fact]
+    public async Task Pump_raises_OnDataChange_only_for_subscribed_refs()
+    {
+        var (driver, client) = await InitializedDriverAsync();
+        var seen = Record(driver);
+        await driver.SubscribeAsync(["dev1_pos"], TimeSpan.FromMilliseconds(50), Ct);
+        seen.Clear();
+
+        await client.PumpAsync(CannedAgentClient.Chunk("Fixtures/sample.xml")).WaitAsync(Watchdog, Ct);
+
+        seen.Select(e => e.FullReference).Distinct().ShouldBe(["dev1_pos"]);
+        For(seen, "dev1_pos")[0].Snapshot.Value.ShouldBe(124.01d);
+
+        // The index took the whole chunk even though only one ref was published.
+        driver.ObservationIndex.Get("dev1_partcount").Value.ShouldBe(42);
+    }
+
+    /// <summary>Each handle hears about the references IT subscribed, and only those.</summary>
+    [Fact]
+    public async Task Pump_fans_each_observation_to_every_handle_that_subscribes_it()
+    {
+        var (driver, client) = await InitializedDriverAsync();
+        var seen = Record(driver);
+        var a = await driver.SubscribeAsync(["dev1_pos", "dev1_execution"], TimeSpan.FromMilliseconds(50), Ct);
+        var b = await driver.SubscribeAsync(["dev1_pos"], TimeSpan.FromMilliseconds(50), Ct);
+        seen.Clear();
+
+        await client
+            .PumpAsync(Chunk(PrimedNextSequence, 113, ("dev1_pos", "1.5"), ("dev1_execution", "READY")))
+            .WaitAsync(Watchdog, Ct);
+
+        For(seen, "dev1_pos").Select(e => e.SubscriptionHandle).ShouldBe([a, b], ignoreOrder: true);
+        For(seen, "dev1_execution").Select(e => e.SubscriptionHandle).ShouldBe([a]);
+    }
+
+    /// <summary>
+    ///     <b>The running-cursor pin.</b> An observation-free heartbeat still advances the sequence —
+    ///     the Agent sends one precisely so a quiet connection can be told from a dead one — and the
+    ///     gap check must be made against the PREVIOUS chunk's <c>nextSequence</c>, never against the
+    ///     <c>from</c> the stream was opened with. A driver that compares against the opening
+    ///     <c>from</c>, or that skips an empty chunk when advancing, reports a gap on the third chunk
+    ///     here and re-baselines against a perfectly healthy stream.
+    /// </summary>
+    [Fact]
+    public async Task Pump_advances_the_cursor_on_every_chunk_including_an_observation_free_one()
+    {
+        var (driver, client) = await InitializedDriverAsync();
+        var seen = Record(driver);
+        await driver.SubscribeAsync(["dev1_pos"], TimeSpan.FromMilliseconds(50), Ct);
+        seen.Clear();
+
+        await client.PumpAsync(Chunk(PrimedNextSequence, 113, ("dev1_pos", "1.5"))).WaitAsync(Watchdog, Ct);
+
+        // A heartbeat: no observations, and firstSequence is the buffer floor, not the cursor.
+        await client.PumpAsync(Chunk(1, 120)).WaitAsync(Watchdog, Ct);
+
+        // Contiguous with the heartbeat's nextSequence — a gap ONLY if the heartbeat was ignored.
+        await client.PumpAsync(Chunk(120, 125, ("dev1_pos", "2.5"))).WaitAsync(Watchdog, Ct);
+
+        client.CurrentCallCount.ShouldBe(1);   // the initialize prime, and nothing else
+        client.SampleCallCount.ShouldBe(1);    // one uninterrupted stream
+        For(seen, "dev1_pos").Last().Snapshot.Value.ShouldBe(2.5d);
+    }
+
+    // ---- ring-buffer overflow ----
+
+    /// <summary>
+    ///     The plan's second TDD case. The Agent's buffer rolled past the cursor
+    ///     (<c>firstSequence</c> 5000 &gt; the driver's 108), so the observations in between are
+    ///     gone: the driver must re-baseline from <c>/current</c> rather than trust an incremental
+    ///     update — and must do it <b>once</b>, then resume, not once per chunk forever.
+    /// </summary>
+    [Fact]
+    public async Task Sequence_gap_triggers_exactly_one_current_rebaseline_then_resumes()
+    {
+        var client = CannedAgentClient.WithGapThenResume();
+        var (driver, _) = await InitializedDriverAsync(client);
+        var seen = Record(driver);
+        await driver.SubscribeAsync(["dev1_pos"], TimeSpan.FromMilliseconds(50), Ct);
+        seen.Clear();
+
+        await client.PumpOnce().WaitAsync(Watchdog, Ct);          // the gap chunk
+
+        client.CurrentCallCount.ShouldBeGreaterThan(1);           // the plan's assertion
+        client.CurrentCallCount.ShouldBe(2);                      // prime + exactly one re-baseline
+
+        await client.PumpOnce().WaitAsync(Watchdog, Ct);          // the contiguous follow-up
+
+        client.CurrentCallCount.ShouldBe(2);                      // no re-baseline storm
+        client.SampleCallCount.ShouldBe(2);                       // the stream was reopened once
+        For(seen, "dev1_pos").Last().Snapshot.Value.ShouldBe(201.5d);
+    }
+
+    /// <summary>
+    ///     …and it resumes from the <b>re-baselined</b> <c>nextSequence</c>, not from the stale
+    ///     cursor the Agent has already evicted. Reopening at the old cursor would earn the identical
+    ///     rejection immediately and forever.
+    /// </summary>
+    [Fact]
+    public async Task Sequence_gap_resumes_the_stream_from_the_rebaselined_next_sequence()
+    {
+        var client = CannedAgentClient.WithGapThenResume();
+        var (driver, _) = await InitializedDriverAsync(client);
+        await driver.SubscribeAsync(["dev1_pos"], TimeSpan.FromMilliseconds(50), Ct);
+
+        await client.PumpOnce().WaitAsync(Watchdog, Ct);   // gap
+        await client.PumpOnce().WaitAsync(Watchdog, Ct);   // resume — only a reopened stream delivers this
+
+        client.LastSampleFrom.ShouldBe(5005L);
+    }
+
+    /// <summary>
+    ///     <b>The other half of the overflow story.</b> A real cppagent answers a <c>from</c> that
+    ///     has already fallen out of its buffer with an <c>MTConnectError</c> / <c>OUT_OF_RANGE</c>
+    ///     document served under <b>HTTP 200</b> — which the client surfaces as an
+    ///     <see cref="InvalidDataException"/>, never as a gap-bearing chunk. A pump that re-baselines
+    ///     only on <c>IsSequenceGap</c> therefore fails precisely when it has fallen furthest behind.
+    /// </summary>
+    [Fact]
+    public async Task Out_of_range_stream_error_also_triggers_a_rebaseline()
+    {
+        var (driver, client) = await InitializedDriverAsync();
+        var seen = Record(driver);
+
+        client.SampleFailures.Enqueue(new InvalidDataException(
+            "MTConnect /sample answered an MTConnectError document (OUT_OF_RANGE) under HTTP 200"));
+        client.CurrentAnswers.Enqueue(Chunk(490, 500, ("dev1_pos", "50.5")));
+
+        await driver.SubscribeAsync(["dev1_pos"], TimeSpan.FromMilliseconds(50), Ct);
+        seen.Clear();
+
+        // Only a pump that re-baselined and reopened the stream can consume this.
+        await client.PumpAsync(Chunk(500, 505, ("dev1_pos", "77.5"))).WaitAsync(Watchdog, Ct);
+
+        client.CurrentCallCount.ShouldBe(2);
+        client.SampleCallCount.ShouldBe(2);
+        client.LastSampleFrom.ShouldBe(500L);
+        For(seen, "dev1_pos").Last().Snapshot.Value.ShouldBe(77.5d);
+    }
+
+    // ---- agent restart ----
+
+    /// <summary>
+    ///     <b>The instanceId check must come BEFORE the gap check.</b> An Agent restart changes
+    ///     <c>instanceId</c> AND resets sequences, so a restart usually trips
+    ///     <c>IsSequenceGap</c> too — and the two demand different handling: a gap keeps the held
+    ///     values (they are still this device's), a restart invalidates every one of them because
+    ///     the device model they describe no longer exists. This chunk is deliberately BOTH: new
+    ///     instanceId and a <c>firstSequence</c> far past the cursor. A driver that tests the gap
+    ///     first keeps serving <c>dev1_execution</c>'s pre-restart value indefinitely — the new
+    ///     Agent never reports it again, so nothing would ever overwrite it.
+    /// </summary>
+    [Fact]
+    public async Task Agent_restart_clears_the_index_and_is_detected_before_the_gap_path()
+    {
+        var (driver, client) = await InitializedDriverAsync();
+        var seen = Record(driver);
+        await driver.SubscribeAsync(["dev1_pos", "dev1_execution"], TimeSpan.FromMilliseconds(50), Ct);
+        seen.Clear();
+
+        // The restarted Agent's /current knows nothing about dev1_execution.
+        client.CurrentAnswers.Enqueue(ChunkFrom(RestartedInstanceId, 40, 42, ("dev1_pos", "9.5")));
+
+        await client
+            .PumpAsync(ChunkFrom(RestartedInstanceId, 5000, 5005, ("dev1_pos", "5.5")))
+            .WaitAsync(Watchdog, Ct);
+
+        driver.AgentInstanceId.ShouldBe(RestartedInstanceId);
+        client.CurrentCallCount.ShouldBe(2);
+
+        // Cleared: the pre-restart value is gone, not merely shadowed.
+        driver.ObservationIndex.Get("dev1_execution").StatusCode.ShouldBe(BadWaitingForInitialData);
+        driver.ObservationIndex.Get("dev1_pos").Value.ShouldBe(9.5d);
+
+        // And the subscriber was TOLD its value went away, rather than being left holding a stale Good.
+        For(seen, "dev1_execution").Last().Snapshot.StatusCode.ShouldBe(BadWaitingForInitialData);
+        For(seen, "dev1_pos").Last().Snapshot.Value.ShouldBe(9.5d);
+    }
+
+    // ---- stream ends ----
+
+    /// <summary>
+    ///     A dropped connection is transient: reconnect from the cursor and keep going. It is
+    ///     emphatically NOT a re-baseline — the sequence is still valid, and spending a
+    ///     <c>/current</c> on every network blip would be a self-inflicted load.
+    /// </summary>
+    [Fact]
+    public async Task Stream_ended_reconnects_from_the_cursor_without_a_rebaseline()
+    {
+        var (driver, client) = await InitializedDriverAsync();
+        var seen = Record(driver);
+        await driver.SubscribeAsync(["dev1_pos"], TimeSpan.FromMilliseconds(50), Ct);
+        seen.Clear();
+
+        await client.PumpAsync(Chunk(PrimedNextSequence, 113, ("dev1_pos", "1.5"))).WaitAsync(Watchdog, Ct);
+
+        client.EndStream();
+
+        // Only a reconnected pump can consume a chunk written to the NEW stream generation.
+        await client.PumpAsync(Chunk(113, 118, ("dev1_pos", "2.5"))).WaitAsync(Watchdog, Ct);
+
+        client.SampleCallCount.ShouldBe(2);
+        client.LastSampleFrom.ShouldBe(113L);
+        client.CurrentCallCount.ShouldBe(1);
+        For(seen, "dev1_pos").Last().Snapshot.Value.ShouldBe(2.5d);
+    }
+
+    /// <summary>
+    ///     <b>The hot-loop pin.</b> <see cref="MTConnectStreamNotSupportedException"/> means the
+    ///     endpoint will never stream to this request — a reverse proxy, a health page, an Agent
+    ///     fronted by something that collapses <c>multipart/x-mixed-replace</c>. Reconnecting
+    ///     reproduces the identical answer forever, so the pump must give up loudly instead of
+    ///     hammering the endpoint until an operator notices. A later subscription must not restart
+    ///     the loop either: the answer has not changed just because someone asked again.
+    /// </summary>
+    [Fact]
+    public async Task Stream_not_supported_stops_the_pump_and_never_retries()
+    {
+        var (driver, client) = await InitializedDriverAsync();
+        client.SampleFailure = new MTConnectStreamNotSupportedException(
+            "the Agent answered /sample with a single non-multipart document");
+
+        var handle = await driver.SubscribeAsync(["dev1_pos"], TimeSpan.FromMilliseconds(50), Ct);
+        var pump = driver.SampleStreamTask.ShouldNotBeNull();
+
+        await pump.WaitAsync(Watchdog, Ct);
+
+        client.SampleCallCount.ShouldBe(1);
+        driver.GetHealth().State.ShouldBe(DriverState.Degraded);
+        driver.GetHealth().LastError.ShouldNotBeNullOrWhiteSpace();
+
+        // A full unsubscribe/resubscribe cycle must not re-open the wound either.
+        await driver.UnsubscribeAsync(handle, Ct).WaitAsync(Watchdog, Ct);
+        await driver.SubscribeAsync(["dev1_pos"], TimeSpan.FromMilliseconds(50), Ct);
+
+        client.SampleCallCount.ShouldBe(1);
+    }
+
+    /// <summary>
+    ///     …but a re-initialize IS an operator changing something, so it clears the verdict and
+    ///     tries again. Otherwise fixing the proxy would need a process restart.
+    /// </summary>
+    [Fact]
+    public async Task Reinitialize_clears_the_stream_unsupported_verdict()
+    {
+        var (driver, client) = await InitializedDriverAsync();
+        client.SampleFailure = new MTConnectStreamNotSupportedException("not multipart");
+        await driver.SubscribeAsync(["dev1_pos"], TimeSpan.FromMilliseconds(50), Ct);
+        await driver.SampleStreamTask!.WaitAsync(Watchdog, Ct);
+
+        client.SampleFailure = null;
+        await driver.ReinitializeAsync("{}", Ct).WaitAsync(Watchdog, Ct);
+
+        await client.PumpAsync(Chunk(PrimedNextSequence, 113, ("dev1_pos", "1.5"))).WaitAsync(Watchdog, Ct);
+
+        client.SampleCallCount.ShouldBe(2);
+    }
+
+    // ---- backoff ----
+
+    /// <summary>
+    ///     The reconnect backoff as a pure function of the attempt number: the first retry honours
+    ///     <c>MinBackoffMs</c> (zero = immediate), and every later one grows geometrically to the
+    ///     cap. The 100 ms growth floor matters more than it looks — the default
+    ///     <c>MinBackoffMs</c> is <b>0</b>, and <c>0 × multiplier</c> is still 0, so without a floor
+    ///     the "geometric" backoff would be an unbounded hot loop against a dead Agent.
+    /// </summary>
+    [Fact]
+    public void Reconnect_backoff_starts_at_min_grows_geometrically_and_caps()
+    {
+        var options = new MTConnectReconnectOptions
+        {
+            MinBackoffMs = 0, MaxBackoffMs = 1000, BackoffMultiplier = 2.0,
+        };
+
+        MTConnectDriver.BackoffFor(1, options).ShouldBe(TimeSpan.Zero);
+        MTConnectDriver.BackoffFor(2, options).ShouldBe(TimeSpan.FromMilliseconds(200));
+        MTConnectDriver.BackoffFor(3, options).ShouldBe(TimeSpan.FromMilliseconds(400));
+        MTConnectDriver.BackoffFor(4, options).ShouldBe(TimeSpan.FromMilliseconds(800));
+        MTConnectDriver.BackoffFor(5, options).ShouldBe(TimeSpan.FromMilliseconds(1000));
+        MTConnectDriver.BackoffFor(500, options).ShouldBe(TimeSpan.FromMilliseconds(1000));
+    }
+
+    /// <summary>
+    ///     Operator-authored nonsense must not un-bound the loop: a multiplier of 1 (or less) would
+    ///     never grow, and a negative delay is not a delay. Both fall back to the shipped default.
+    /// </summary>
+    [Fact]
+    public void Reconnect_backoff_survives_a_degenerate_multiplier()
+    {
+        var options = new MTConnectReconnectOptions
+        {
+            MinBackoffMs = -5, MaxBackoffMs = 10_000, BackoffMultiplier = 0.5,
+        };
+
+        MTConnectDriver.BackoffFor(1, options).ShouldBe(TimeSpan.Zero);
+        MTConnectDriver.BackoffFor(2, options).ShouldBeGreaterThan(TimeSpan.Zero);
+        MTConnectDriver.BackoffFor(3, options)
+            .ShouldBeGreaterThan(MTConnectDriver.BackoffFor(2, options));
+    }
+
+    // ---- unsubscribe ----
+
+    /// <summary>
+    ///     Dropping one handle drops only its references; the shared stream keeps running for the
+    ///     others. Tearing the stream down on the first unsubscribe would silently blind every
+    ///     remaining subscription.
+    /// </summary>
+    [Fact]
+    public async Task Unsubscribe_drops_only_that_handles_refs_and_keeps_the_shared_stream()
+    {
+        var (driver, client) = await InitializedDriverAsync();
+        var seen = Record(driver);
+        var a = await driver.SubscribeAsync(["dev1_pos"], TimeSpan.FromMilliseconds(50), Ct);
+        var b = await driver.SubscribeAsync(["dev1_execution"], TimeSpan.FromMilliseconds(50), Ct);
+        var pump = driver.SampleStreamTask.ShouldNotBeNull();
+
+        await driver.UnsubscribeAsync(a, Ct).WaitAsync(Watchdog, Ct);
+        seen.Clear();
+
+        await client
+            .PumpAsync(Chunk(PrimedNextSequence, 113, ("dev1_pos", "1.5"), ("dev1_execution", "READY")))
+            .WaitAsync(Watchdog, Ct);
+
+        pump.IsCompleted.ShouldBeFalse();
+        seen.Select(e => e.FullReference).Distinct().ShouldBe(["dev1_execution"]);
+
+        await driver.UnsubscribeAsync(b, Ct).WaitAsync(Watchdog, Ct);
+
+        pump.IsCompleted.ShouldBeTrue();
+        driver.SampleStreamTask.ShouldBeNull();
+    }
+
+    /// <summary>The last unsubscribe stops the stream — and nothing is published after it.</summary>
+    [Fact]
+    public async Task Unsubscribe_of_the_last_handle_stops_the_stream_and_silences_the_driver()
+    {
+        var (driver, client) = await InitializedDriverAsync();
+        var seen = Record(driver);
+        var handle = await driver.SubscribeAsync(["dev1_pos"], TimeSpan.FromMilliseconds(50), Ct);
+
+        await driver.UnsubscribeAsync(handle, Ct).WaitAsync(Watchdog, Ct);
+        seen.Clear();
+
+        // The stream is gone, so this chunk has no consumer and nothing may reach the recorder.
+        client.PumpAsync(Chunk(PrimedNextSequence, 113, ("dev1_pos", "1.5"))).IsCompleted.ShouldBeFalse();
+
+        seen.ShouldBeEmpty();
+        driver.SampleStreamTask.ShouldBeNull();
+    }
+
+    /// <summary>
+    ///     Unsubscribing twice, or with a handle this driver never issued, is a no-op — not a throw.
+    ///     Teardown paths run on failure paths, and an exception there would mask the real fault.
+    /// </summary>
+    [Fact]
+    public async Task Unsubscribe_is_idempotent_and_ignores_a_foreign_handle()
+    {
+        var (driver, _) = await InitializedDriverAsync();
+        var handle = await driver.SubscribeAsync(["dev1_pos"], TimeSpan.FromMilliseconds(50), Ct);
+
+        await driver.UnsubscribeAsync(handle, Ct).WaitAsync(Watchdog, Ct);
+        await driver.UnsubscribeAsync(handle, Ct).WaitAsync(Watchdog, Ct);
+        await driver.UnsubscribeAsync(new ForeignHandle(), Ct).WaitAsync(Watchdog, Ct);
+    }
+
+    /// <summary>A null argument is a caller bug, and the one thing these methods are loud about.</summary>
+    [Fact]
+    public async Task Null_arguments_throw_ArgumentNullException()
+    {
+        var (driver, _) = await InitializedDriverAsync();
+
+        await Should.ThrowAsync<ArgumentNullException>(
+            () => driver.SubscribeAsync(null!, TimeSpan.FromMilliseconds(50), Ct));
+        await Should.ThrowAsync<ArgumentNullException>(() => driver.UnsubscribeAsync(null!, Ct));
+    }
+
+    // ---- lifecycle ----
+
+    /// <summary>
+    ///     Subscribing before the driver is initialized registers the interest and reports the
+    ///     honest "no value yet" — it does not throw, and it does not dial an Agent that does not
+    ///     exist yet. Initialize then picks the standing subscription up.
+    /// </summary>
+    [Fact]
+    public async Task Subscribe_before_initialize_registers_and_initialize_starts_the_stream()
+    {
+        var (driver, client) = NewDriver();
+        var seen = Record(driver);
+
+        await driver.SubscribeAsync(["dev1_pos"], TimeSpan.FromMilliseconds(50), Ct);
+
+        client.SampleCallCount.ShouldBe(0);
+        driver.SampleStreamTask.ShouldBeNull();
+        seen.Single().Snapshot.StatusCode.ShouldBe(BadWaitingForInitialData);
+
+        await driver.InitializeAsync("{}", Ct);
+        await client.PumpAsync(Chunk(PrimedNextSequence, 113, ("dev1_pos", "1.5"))).WaitAsync(Watchdog, Ct);
+
+        client.SampleCallCount.ShouldBe(1);
+        For(seen, "dev1_pos").Last().Snapshot.Value.ShouldBe(1.5d);
+    }
+
+    /// <summary>
+    ///     A re-initialize replaces the served state, so it must replace the pump with it: the old
+    ///     one is stopped (it holds the old cursor, and possibly the old client) and a new one starts
+    ///     from the fresh <c>/current</c>. The standing subscription survives — the caller did not
+    ///     ask for it to be dropped — and is re-primed, because every value it holds came from a
+    ///     baseline that has just been thrown away.
+    /// </summary>
+    [Fact]
+    public async Task Reinitialize_stops_the_old_pump_starts_a_new_one_and_republishes()
+    {
+        var (driver, client) = await InitializedDriverAsync();
+        var seen = Record(driver);
+        await driver.SubscribeAsync(["dev1_pos"], TimeSpan.FromMilliseconds(50), Ct);
+        var first = driver.SampleStreamTask.ShouldNotBeNull();
+
+        // Barrier: the first pump is demonstrably enumerating before the re-initialize lands on it.
+        await client.PumpAsync(Chunk(PrimedNextSequence, 113, ("dev1_pos", "1.5"))).WaitAsync(Watchdog, Ct);
+
+        client.CurrentAnswers.Enqueue(Chunk(300, 310, ("dev1_pos", "310.5")));
+        seen.Clear();
+
+        await driver.ReinitializeAsync("{}", Ct).WaitAsync(Watchdog, Ct);
+
+        first.IsCompleted.ShouldBeTrue();
+        var second = driver.SampleStreamTask.ShouldNotBeNull();
+        second.ShouldNotBeSameAs(first);
+
+        await client.PumpAsync(Chunk(310, 315, ("dev1_pos", "315.5"))).WaitAsync(Watchdog, Ct);
+
+        client.SampleCallCount.ShouldBe(2);
+        client.LastSampleFrom.ShouldBe(310L);
+        For(seen, "dev1_pos").Select(e => e.Snapshot.Value).ShouldBe([310.5d, 315.5d]);
+    }
+
+    /// <summary>
+    ///     Shutdown stops the pump BEFORE disposing the client. Getting that order wrong leaves the
+    ///     pump enumerating a disposed client — which it would read as a dropped connection and try
+    ///     to reconnect, forever, against an object that no longer exists.
+    /// </summary>
+    [Fact]
+    public async Task Shutdown_stops_the_pump_before_disposing_the_client_and_leaks_nothing()
+    {
+        var (driver, client) = await InitializedDriverAsync();
+        var seen = Record(driver);
+        await driver.SubscribeAsync(["dev1_pos"], TimeSpan.FromMilliseconds(50), Ct);
+        var pump = driver.SampleStreamTask.ShouldNotBeNull();
+
+        // Barrier: the stream is demonstrably open before the shutdown lands, so the call count
+        // below is a statement about reconnect churn rather than about who won a race.
+        await client.PumpAsync(Chunk(PrimedNextSequence, 113, ("dev1_pos", "1.5"))).WaitAsync(Watchdog, Ct);
+        seen.Clear();
+
+        await driver.ShutdownAsync(Ct).WaitAsync(Watchdog, Ct);
+
+        pump.IsCompleted.ShouldBeTrue();
+        pump.IsFaulted.ShouldBeFalse();
+        driver.SampleStreamTask.ShouldBeNull();
+        client.IsDisposed.ShouldBeTrue();
+
+        // No reconnect churn against the disposed client, and nothing published after teardown.
+        client.SampleCallCount.ShouldBe(1);
+        seen.ShouldBeEmpty();
+    }
+
+    /// <summary>
+    ///     <b>The deadlock pin.</b> The lifecycle semaphore is non-reentrant, so a pump that reached
+    ///     back into <c>InitializeAsync</c>/<c>ReinitializeAsync</c>/<c>ShutdownAsync</c> — the
+    ///     natural way to write "on a gap, just re-prime" — would hang the driver permanently, with
+    ///     no exception and no log. Here the pump is parked INSIDE its re-baseline <c>/current</c>
+    ///     when a shutdown lands and takes that semaphore: the shutdown must cancel the pump and
+    ///     complete. If the pump were waiting on the lifecycle lock instead, neither side would ever
+    ///     move.
+    /// </summary>
+    [Fact]
+    public async Task Shutdown_while_the_pump_is_mid_rebaseline_does_not_deadlock()
+    {
+        var (driver, client) = await InitializedDriverAsync();
+        await driver.SubscribeAsync(["dev1_pos"], TimeSpan.FromMilliseconds(50), Ct);
+
+        var entered = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously);
+        var release = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously);
+        client.CurrentEntered = entered;
+        client.CurrentGate = release;
+
+        // A gap forces the re-baseline; the gate holds the pump inside the /current request.
+        _ = client.PumpAsync(Chunk(5000, 5005, ("dev1_pos", "5.5")));
+        await entered.Task.WaitAsync(Watchdog, Ct);
+
+        try
+        {
+            await driver.ShutdownAsync(Ct).WaitAsync(Watchdog, Ct);
+        }
+        finally
+        {
+            release.TrySetResult();
+        }
+
+        driver.SampleStreamTask.ShouldBeNull();
+        client.IsDisposed.ShouldBeTrue();
+    }
+
+    // ---- robustness + health ----
+
+    /// <summary>
+    ///     A subscriber that throws is a bug in the consumer, not a reason to stop delivering data to
+    ///     every other subscriber. The pump absorbs it and keeps pumping.
+    /// </summary>
+    [Fact]
+    public async Task A_throwing_subscriber_does_not_kill_the_pump()
+    {
+        var (driver, client) = await InitializedDriverAsync();
+        var seen = Record(driver);
+        driver.OnDataChange += (_, _) => throw new InvalidOperationException("subscriber blew up");
+
+        await driver.SubscribeAsync(["dev1_pos"], TimeSpan.FromMilliseconds(50), Ct);
+        seen.Clear();
+
+        await client.PumpAsync(Chunk(PrimedNextSequence, 113, ("dev1_pos", "1.5"))).WaitAsync(Watchdog, Ct);
+        await client.PumpAsync(Chunk(113, 118, ("dev1_pos", "2.5"))).WaitAsync(Watchdog, Ct);
+
+        driver.SampleStreamTask!.IsCompleted.ShouldBeFalse();
+        For(seen, "dev1_pos").Select(e => e.Snapshot.Value).ShouldBe([1.5d, 2.5d]);
+    }
+
+    /// <summary>
+    ///     <b>Health precedence, read side.</b> A healthy stream must not paper over a broken read
+    ///     path: <c>/sample</c> and <c>/current</c> are different Agent requests and can fail
+    ///     independently, and "Healthy" while every OPC UA Read returns Bad is exactly the
+    ///     failure-wearing-success shape this driver is written against.
+    /// </summary>
+    [Fact]
+    public async Task A_pumped_chunk_does_not_paper_over_a_failed_read()
+    {
+        var (driver, client) = await InitializedDriverAsync();
+        await driver.SubscribeAsync(["dev1_pos"], TimeSpan.FromMilliseconds(50), Ct);
+
+        client.CurrentFailure = new HttpRequestException("connection refused");
+        await driver.ReadAsync(["dev1_pos"], Ct);
+        driver.GetHealth().State.ShouldBe(DriverState.Degraded);
+
+        await client.PumpAsync(Chunk(PrimedNextSequence, 113, ("dev1_pos", "1.5"))).WaitAsync(Watchdog, Ct);
+
+        driver.GetHealth().State.ShouldBe(DriverState.Degraded);
+
+        // …and clearing the read failure does restore it, because the stream never failed.
+        client.CurrentFailure = null;
+        await driver.ReadAsync(["dev1_pos"], Ct);
+        driver.GetHealth().State.ShouldBe(DriverState.Healthy);
+    }
+
+    /// <summary>
+    ///     <b>Health precedence, stream side.</b> The mirror image: a successful read must not paper
+    ///     over a stream that has stopped delivering. Only the path that failed may clear its own
+    ///     degradation.
+    /// </summary>
+    [Fact]
+    public async Task A_successful_read_does_not_paper_over_a_broken_stream()
+    {
+        var (driver, client) = await InitializedDriverAsync();
+        client.SampleFailure = new MTConnectStreamNotSupportedException("not multipart");
+        await driver.SubscribeAsync(["dev1_pos"], TimeSpan.FromMilliseconds(50), Ct);
+        await driver.SampleStreamTask!.WaitAsync(Watchdog, Ct);
+
+        var res = await driver.ReadAsync(["dev1_pos"], Ct);
+
+        res[0].StatusCode.ShouldBe(Good);
+        driver.GetHealth().State.ShouldBe(DriverState.Degraded);
+    }
+
+    /// <summary>A recovered stream clears its own degradation once chunks flow again.</summary>
+    [Fact]
+    public async Task A_recovered_stream_restores_health()
+    {
+        var (driver, client) = await InitializedDriverAsync();
+        client.SampleFailures.Enqueue(new TimeoutException("no chunk within the heartbeat window"));
+
+        await driver.SubscribeAsync(["dev1_pos"], TimeSpan.FromMilliseconds(50), Ct);
+        await client.PumpAsync(Chunk(PrimedNextSequence, 113, ("dev1_pos", "1.5"))).WaitAsync(Watchdog, Ct);
+
+        driver.GetHealth().State.ShouldBe(DriverState.Healthy);
+        client.SampleCallCount.ShouldBe(2);
+    }
+
+    /// <summary>A handle from some other driver — the type check must not be a cast.</summary>
+    private sealed class ForeignHandle : ISubscriptionHandle
+    {
+        public string DiagnosticId => "not-ours";
+    }
+}