feat(mtconnect): /sample lag measurement + adaptive count (R06:T-5)

Subscription throughput was hard-capped at SampleCount / SampleIntervalMs with no lag detection: a machine producing more fell behind the ring buffer forever, re-baselining every buffer-length with every value backlog-stale. - The parser carries the header's optional lastSequence; each chunk measures lag = lastSequence + 1 - nextSequence and whether it was saturated (full count). - Three consecutive saturated chunks with a positive (or unknown) lag reopen the stream at the cursor reached with count doubled, up to 16x SampleCount (max 65536), through a new IMTConnectAgentClient.SampleAsync(from, count, ct) overload (default implementation ignores count). A raised stream that delivers nothing steps back to the last good count; at the ceiling one Warning per pump. - MTConnectDriver implements IDriverDiagnostics: sampleCount, sampleLag, sampleSaturatedChunks, sampleCountRaises on the /hosts bag. The cap is documented on SampleCount. Refs #743 Claude-Session: https://claude.ai/code/session_01Ub9hGmYLQvn9XETzw7CMsT
2026-10-22 15:50:42 -04:00
parent 29c829b540
commit 21b4f00f46
12 changed files with 505 additions and 26 deletions
@@ -73,7 +73,7 @@ below).

 ## Capability surface

-`MTConnectDriver : IDriver, IReadable, IWritable, ISubscribable, ITagDiscovery, IHostConnectivityProbe, IRediscoverable`
+`MTConnectDriver : IDriver, IReadable, IWritable, ISubscribable, ITagDiscovery, IHostConnectivityProbe, IRediscoverable, IDriverDiagnostics`
 (`src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.MTConnect/MTConnectDriver.cs`).

 `IWritable` landed 2026-08-05 (`b29623cf`), against the Agent's `AllowPut`
@@ -92,6 +92,7 @@ request/response handshake — is a different mechanism and remains deferred; se
 | `ISubscribable` | `SubscribeAsync`/`OnDataChange` — the shared `/sample` long-poll pump | **The production data path** |
 | `IHostConnectivityProbe` | periodic `/probe` under `Probe.*` | No consumer wires it today — see [Known limitations](#known-limitations) |
 | `IRediscoverable` | watches the Agent's `Header.instanceId` | No consumer wires it today — see [Known limitations](#known-limitations) |
+| `IDriverDiagnostics` | `GetDiagnostics()` — `/sample` effective count, lag, saturated chunks, count raises | Shown on `/hosts` (review 2026-10-21 T-5) |
 | `IAlarmSource` | CONDITION observations off the same `/sample` pump → Part 9 transitions keyed by the alarm tag's RawPath | Opt-in per raw tag (`alarm` object) — see [CONDITION alarms](#condition-alarms-626) |

 ### Feature status at a glance (#613)
@@ -159,7 +160,7 @@ Read off `MTConnectDriverOptions` / `MTConnectDriverConfigDto`
 | `DeviceName` | `null` | **Double duty.** (1) Scopes requests to `{AgentUri}/{DeviceName}/...`; `null` = whole Agent. (2) The write path's fallback device — the device a write is addressed to when the probe model does not say which device declared the DataItem (see [Write-back](#write-back--observation-injection-off-by-default)) |
 | `RequestTimeoutMs` | `5000` | Per-call deadline for `/probe` and `/current` |
 | `SampleIntervalMs` | `1000` | The Agent's `/sample?interval=` query param |
-| `SampleCount` | `1000` | The Agent's `/sample?count=` query param |
+| `SampleCount` | `1000` | The Agent's `/sample?count=` query param. **Caps subscription throughput** at `SampleCount / SampleIntervalMs` — see [Subscribe](#subscribe--the-sample-pump) for the adaptive raise |
 | `HeartbeatMs` | `10000` | The Agent's `/sample?heartbeat=` query param — also the watchdog's liveness window |
 | `Probe.Enabled` / `Probe.IntervalMs` / `Probe.TimeoutMs` | `true` / `5000` / `2000` | Background connectivity-probe knobs (mirrors `ModbusProbeOptions`) |
 | `Reconnect.MinBackoffMs` / `MaxBackoffMs` / `BackoffMultiplier` | `0` / `30000` / `2.0` | Geometric backoff after a failed request or a dropped `/sample` stream |
@@ -333,6 +334,17 @@ 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.

+**Lag measurement and the adaptive `count` (review 2026-10-21 T-5).** Throughput is capped at `SampleCount`
+observations per chunk, one chunk per `SampleIntervalMs`; a machine producing more used to fall behind the ring
+buffer forever, re-baselining every buffer-length (dropping intermediate transitions) with nothing surfacing it.
+Each chunk is now measured: `lag = lastSequence + 1 − nextSequence` (unknown when the Agent omits `lastSequence`)
+and whether it was **saturated** (carried the full `count`). Three consecutive saturated chunks with a positive (or
+unknown) lag reopen the stream at the cursor reached with `count` doubled, up to 16× `SampleCount` (max 65 536). A
+raised stream that delivers nothing (an Agent refusing the larger count) steps back to the last count that delivered
+and stops raising; at the ceiling a still-behind stream logs one Warning per pump. Every new pump starts from the
+configured `SampleCount`. The driver now implements `IDriverDiagnostics`: `sampleCount` (effective), `sampleLag`,
+`sampleSaturatedChunks` and `sampleCountRaises` ride the `/hosts` diagnostics bag.
+
 **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
@@ -44,6 +44,13 @@ public sealed class MTConnectDriverOptions
    ///     The <c>count</c> query parameter passed to the Agent's <c>/sample</c> request — the
    ///     maximum number of DataItem observations returned per chunk. Default <c>1000</c>, the
    ///     MTConnect Agent's own conventional default.
+    ///     <para>
+    ///         <b>It caps subscription throughput</b> at <c>SampleCount / SampleIntervalMs</c> observations per
+    ///         millisecond (default 1000 a second): a machine producing more falls behind the Agent's ring buffer. The
+    ///         pump measures that lag and, when it is persistently behind, reopens the stream with the count doubled — up
+    ///         to 16× this value, at most 65 536 — and reports the effective count and the lag on the <c>/hosts</c>
+    ///         diagnostics (review 2026-10-21 T-5). Each new stream starts again from this value.
+    ///     </para>
    /// </summary>
    public int SampleCount { get; init; } = 1000;

@@ -59,7 +59,7 @@ public sealed record MTConnectWriteOutcome(bool Accepted, string? AgentError);
 ///         <see cref="CancellationTokenSource"/>. An <see cref="IAsyncDisposable"/> here would be a
 ///         synchronous method wearing a <c>ValueTask</c>, and it would oblige every canned-XML fake
 ///         to grow one too. The <b>async</b> unwind that a streaming client does need is already
-///         covered elsewhere and is not this method's job: the <see cref="SampleAsync"/> enumerator
+///         covered elsewhere and is not this method's job: the <see cref="SampleAsync(long, CancellationToken)"/> enumerator
 ///         is itself <see cref="IAsyncDisposable"/> and is owned by the pump that enumerates it,
 ///         while <see cref="IDisposable.Dispose"/> on the client converts any read still in flight
 ///         into an ordinary <see cref="OperationCanceledException"/>.
@@ -79,7 +79,7 @@ public interface IMTConnectAgentClient : IDisposable
    /// <summary>
    ///     Issues an Agent <c>/current</c> request and returns the latest observed value for every
    ///     data item. Used both to prime the observation index at startup and to re-baseline after
-    ///     a detected sequence gap in <see cref="SampleAsync"/>.
+    ///     a detected sequence gap in <see cref="SampleAsync(long, CancellationToken)"/>.
    /// </summary>
    /// <param name="ct">Cancellation token.</param>
    /// <returns>The latest observation for every data item.</returns>
@@ -113,6 +113,20 @@ public interface IMTConnectAgentClient : IDisposable
    /// <returns>An asynchronous sequence of streams results, one per multipart chunk.</returns>
    IAsyncEnumerable<MTConnectStreamsResult> SampleAsync(long from, CancellationToken ct);

+    /// <summary>
+    ///     As <see cref="SampleAsync(long, CancellationToken)"/>, but asking the Agent for up to
+    ///     <paramref name="count"/> observations per chunk instead of the configured
+    ///     <see cref="MTConnectDriverOptions.SampleCount"/> — how the pump catches up when it falls behind an Agent
+    ///     producing more than <c>SampleCount / SampleIntervalMs</c> observations a second (R06:T-5). The default
+    ///     ignores <paramref name="count"/>, so a client that cannot honour it keeps working at the configured one.
+    /// </summary>
+    /// <param name="from">The sequence to open the stream at.</param>
+    /// <param name="count">The per-chunk observation cap for this stream.</param>
+    /// <param name="ct">Cancellation token; the only contracted way the enumeration ends quietly.</param>
+    /// <returns>The stream's chunks.</returns>
+    IAsyncEnumerable<MTConnectStreamsResult> SampleAsync(long from, int count, CancellationToken ct) =>
+        SampleAsync(from, ct);
+
    /// <summary>
    ///     Sets the Agent's observation for one DataItem, through the Agent's <c>AllowPut</c>
    ///     surface (<c>POST {agent}/{deviceName}?{dataItemId}={value}</c>).
@@ -183,7 +197,7 @@ public interface IMTConnectAgentClient : IDisposable
    /// <remarks>
    ///     <para>
    ///         <b><paramref name="expectedFrom"/> is a running cursor, not the original
-    ///         <c>from</c>.</b> It equals the <c>from</c> passed to <see cref="SampleAsync"/> only
+    ///         <c>from</c>.</b> It equals the <c>from</c> passed to <see cref="SampleAsync(long, CancellationToken)"/> only
    ///         for the <i>first</i> chunk; from the second chunk on it is the <b>previous</b>
    ///         chunk's <see cref="MTConnectStreamsResult.NextSequence"/>. Comparing every chunk
    ///         against the original <c>from</c> instead reports a gap on every chunk once the
@@ -199,7 +213,7 @@ public interface IMTConnectAgentClient : IDisposable
    ///         a chunk the Agent was willing to send — and 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 this client surfaces as an
-    ///         <see cref="InvalidDataException"/> out of <see cref="SampleAsync"/>, never as a chunk.
+    ///         <see cref="InvalidDataException"/> out of <see cref="SampleAsync(long, CancellationToken)"/>, never as a chunk.
    ///         A pump that re-baselines only on <c>IsSequenceGap</c> will therefore fault instead of
    ///         recovering exactly when it has fallen furthest behind. Treat an <c>OUT_OF_RANGE</c>
    ///         parse failure as a re-baseline trigger (re-prime from <see cref="CurrentAsync"/>,
@@ -17,7 +17,7 @@ namespace ZB.MOM.WW.OtOpcUa.Driver.MTConnect;
 ///         <b>The constructor opens no connection.</b> It only composes the request URIs and builds
 ///         its <see cref="HttpClient"/>s (which themselves dial nothing until a request is issued),
 ///         so a throwaway instance is safe to construct — the universal discovery browser's
-///         <c>CanBrowse</c> pattern depends on that. <see cref="SampleAsync"/> likewise dials
+///         <c>CanBrowse</c> pattern depends on that. <see cref="SampleAsync(long, CancellationToken)"/> likewise dials
 ///         nothing until it is enumerated.
 ///     </para>
 ///     <para>
@@ -54,7 +54,7 @@ namespace ZB.MOM.WW.OtOpcUa.Driver.MTConnect;
 ///     </para>
 ///     <para>
 ///         <b>Disposal is safe against an in-flight enumeration.</b> Re-initialization disposes and rebuilds this client, which can happen while a pump is still
-///         enumerating <see cref="SampleAsync"/>. Rather than letting the in-flight read fault with
+///         enumerating <see cref="SampleAsync(long, CancellationToken)"/>. Rather than letting the in-flight read fault with
 ///         an unclassified <see cref="ObjectDisposedException"/> or <see cref="IOException"/> that no
 ///         caller is written to expect, every request runs under a token linked to an internal
 ///         dispose token, so disposal surfaces as an ordinary
@@ -79,6 +79,8 @@ public sealed class MTConnectAgentClient : IMTConnectAgentClient, IDisposable
    private readonly string _agentRoot;
    private readonly string _sampleUriBase;
    private readonly string _sampleQuerySuffix;
+    private readonly int _sampleCount;
+    private readonly int _heartbeatMs;
    private readonly TimeSpan _requestTimeout;
    private readonly TimeSpan _streamIdleTimeout;

@@ -151,9 +153,10 @@ public sealed class MTConnectAgentClient : IMTConnectAgentClient, IDisposable
        _probeUri = BuildRequestUri(options, "probe");
        _currentUri = BuildRequestUri(options, "current");
        _sampleUriBase = BuildRequestUri(options, "sample").AbsoluteUri;
-        _sampleQuerySuffix = string.Create(
-            CultureInfo.InvariantCulture,
-            $"&interval={options.SampleIntervalMs}&count={options.SampleCount}&heartbeat={options.HeartbeatMs}");
+        // count is appended per request: the driver may raise it for one stream when it falls behind (R06:T-5).
+        _sampleQuerySuffix = string.Create(CultureInfo.InvariantCulture, $"&interval={options.SampleIntervalMs}");
+        _sampleCount = options.SampleCount;
+        _heartbeatMs = options.HeartbeatMs;

        _http = handler is null ? new HttpClient() : new HttpClient(handler, disposeHandler: false);
        _http.Timeout = _requestTimeout;
@@ -195,10 +198,32 @@ public sealed class MTConnectAgentClient : IMTConnectAgentClient, IDisposable
    public async IAsyncEnumerable<MTConnectStreamsResult> SampleAsync(
        long from,
        [EnumeratorCancellation] CancellationToken ct)
+    {
+        await foreach (var chunk in SampleAsync(from, count: null, ct).ConfigureAwait(false))
+        {
+            yield return chunk;
+        }
+    }
+
+    /// <inheritdoc/>
+    public IAsyncEnumerable<MTConnectStreamsResult> SampleAsync(long from, int count, CancellationToken ct)
+    {
+        if (count <= 0)
+        {
+            throw new ArgumentOutOfRangeException(nameof(count), count, "a /sample count must be positive");
+        }
+
+        return SampleAsync(from, (int?)count, ct);
+    }
+
+    private async IAsyncEnumerable<MTConnectStreamsResult> SampleAsync(
+        long from,
+        int? count,
+        [EnumeratorCancellation] CancellationToken ct)
    {
        ObjectDisposedException.ThrowIf(Volatile.Read(ref _disposed) != 0, this);

-        var uri = BuildSampleUri(from);
+        var uri = BuildSampleUri(from, count);

        // Linked to the caller's token AND to disposal, so tearing this client down mid-enumeration
        // surfaces as a cancellation rather than as a raw ObjectDisposedException from the socket.
@@ -411,8 +436,10 @@ public sealed class MTConnectAgentClient : IMTConnectAgentClient, IDisposable
        }
    }

-    private Uri BuildSampleUri(long from) =>
-        new(string.Create(CultureInfo.InvariantCulture, $"{_sampleUriBase}?from={from}{_sampleQuerySuffix}"),
+    private Uri BuildSampleUri(long from, int? count) =>
+        new(string.Create(
+                CultureInfo.InvariantCulture,
+                $"{_sampleUriBase}?from={from}{_sampleQuerySuffix}&count={count ?? _sampleCount}&heartbeat={_heartbeatMs}"),
            UriKind.Absolute);

    private async Task<HttpResponseMessage> SendAsync(Uri uri, CancellationToken ct)
@@ -0,0 +1,156 @@
+using System.Globalization;
+using Microsoft.Extensions.Logging;
+using ZB.MOM.WW.OtOpcUa.Core.Abstractions;
+
+namespace ZB.MOM.WW.OtOpcUa.Driver.MTConnect;
+
+/// <summary>
+///     Review 2026-10-21 R06:T-5 — <c>/sample</c> throughput is capped at
+///     <see cref="MTConnectDriverOptions.SampleCount"/> observations per chunk, one chunk per
+///     <see cref="MTConnectDriverOptions.SampleIntervalMs"/>. A machine producing more than that falls steadily behind
+///     the Agent's ring buffer: every buffer-length the driver re-baselines (losing every intermediate transition) and
+///     every delivered value is backlog-stale. This half measures the lag, surfaces it, and raises the per-stream
+///     <c>count</c> so the pump can catch up.
+/// </summary>
+/// <remarks>
+///     <para>
+///         <b>Measured per chunk.</b> <c>lag = lastSequence + 1 − nextSequence</c> — the observations the Agent holds
+///         that this stream has not been sent yet (unknown when the Agent omits <c>lastSequence</c>). A chunk is
+///         <b>saturated</b> when it carries the full <c>count</c>: the Agent had more and stopped. A stream is behind
+///         when <see cref="BehindChunksBeforeRaise"/> consecutive chunks are saturated with a positive (or unknown) lag.
+///     </para>
+///     <para>
+///         <b>Adapted per pump.</b> A behind stream is reopened at the cursor it reached with <c>count</c> doubled, up
+///         to <see cref="SampleCountCeiling"/>. A raised stream that then delivers nothing (an Agent that refuses the
+///         larger count) steps back to the last count that delivered and stops raising for that pump. At the ceiling,
+///         a stream still behind logs one Warning per pump. A new pump (a re-initialize, a resubscribe) starts again
+///         from the configured <see cref="MTConnectDriverOptions.SampleCount"/> — the count is an upper bound, so a
+///         raised one costs nothing while the Agent has little to send.
+///     </para>
+///     <para>
+///         <b>Surfaced</b> on the <see cref="IDriverDiagnostics"/> bag the <c>/hosts</c> page shows: the effective
+///         count, the last measured lag, saturated chunks and count raises.
+///     </para>
+/// </remarks>
+public sealed partial class MTConnectDriver : IDriverDiagnostics
+{
+    /// <summary>Consecutive behind chunks before the count is raised — one saturated chunk is just a burst.</summary>
+    internal const int BehindChunksBeforeRaise = 3;
+
+    /// <summary>The most the per-stream count is ever raised to, relative to the configured one.</summary>
+    internal const int SampleCountMaxGrowth = 16;
+
+    /// <summary>Absolute ceiling on a raised count (a stock cppagent buffer holds 131072 observations).</summary>
+    internal const int SampleCountAbsoluteCeiling = 65_536;
+
+    /// <summary>The count the current pump is streaming with. Diagnostics only.</summary>
+    private int _effectiveSampleCount;
+
+    /// <summary>The last measured lag (observations), or -1 when unknown. Diagnostics only.</summary>
+    private long _sampleLag = -1;
+
+    /// <summary>Chunks that carried the full count. Diagnostics only.</summary>
+    private long _saturatedChunks;
+
+    /// <summary>Times a pump raised its count. Diagnostics only.</summary>
+    private long _sampleCountRaises;
+
+    /// <summary>The ceiling a pump started with <paramref name="configured"/> may raise its count to.</summary>
+    /// <param name="configured">The configured <see cref="MTConnectDriverOptions.SampleCount"/>.</param>
+    /// <returns>The ceiling — never below <paramref name="configured"/>.</returns>
+    internal static int SampleCountCeiling(int configured) =>
+        Math.Max(configured, (int)Math.Min((long)configured * SampleCountMaxGrowth, SampleCountAbsoluteCeiling));
+
+    /// <inheritdoc />
+    public IReadOnlyDictionary<string, string> GetDiagnostics()
+    {
+        var bag = new Dictionary<string, string>(StringComparer.Ordinal);
+        var count = Volatile.Read(ref _effectiveSampleCount);
+        if (count > 0)
+        {
+            bag["sampleCount"] = count.ToString(CultureInfo.InvariantCulture);
+        }
+
+        var lag = Interlocked.Read(ref _sampleLag);
+        if (lag >= 0)
+        {
+            bag["sampleLag"] = lag.ToString(CultureInfo.InvariantCulture);
+        }
+
+        bag["sampleSaturatedChunks"] = Interlocked.Read(ref _saturatedChunks).ToString(CultureInfo.InvariantCulture);
+        bag["sampleCountRaises"] = Interlocked.Read(ref _sampleCountRaises).ToString(CultureInfo.InvariantCulture);
+
+        return bag;
+    }
+
+    /// <summary>
+    ///     Measures one delivered chunk against the count its stream was opened with and records the result.
+    /// </summary>
+    /// <param name="chunk">The chunk just applied.</param>
+    /// <param name="sampleCount">The count the stream was opened with.</param>
+    /// <returns>Whether this chunk says the stream is behind the Agent.</returns>
+    private bool MeasureSampleLag(MTConnectStreamsResult chunk, int sampleCount)
+    {
+        long? lag = chunk.LastSequence is { } last ? Math.Max(0, last + 1 - chunk.NextSequence) : null;
+        Interlocked.Exchange(ref _sampleLag, lag ?? -1);
+
+        var saturated = chunk.Observations.Count >= sampleCount;
+        if (saturated)
+        {
+            Interlocked.Increment(ref _saturatedChunks);
+        }
+
+        return saturated && lag is not 0;
+    }
+
+    /// <summary>One pump's adaptive-count state. Only that pump's task touches it.</summary>
+    private sealed class SampleCountState(int configured)
+    {
+        /// <summary>The count the next stream opens with.</summary>
+        public int Count { get; set; } = configured;
+
+        /// <summary>The last count a stream actually delivered at.</summary>
+        public int LastDelivered { get; set; } = configured;
+
+        /// <summary>The most <see cref="Count"/> may be raised to; lowered to the last good count on a refusal.</summary>
+        public int Ceiling { get; set; } = SampleCountCeiling(configured);
+
+        /// <summary>Whether the at-the-ceiling Warning has been logged for this pump.</summary>
+        public bool CeilingWarned { get; set; }
+    }
+
+    /// <summary>
+    ///     The count a behind stream should reopen with, or <see langword="null"/> when it is already at its ceiling
+    ///     (logged once per pump).
+    /// </summary>
+    private int? NextSampleCount(SampleCountState state, MTConnectDriverOptions options, long cursor)
+    {
+        var sampleCount = state.Count;
+        var next = (int)Math.Min((long)sampleCount * 2, state.Ceiling);
+        if (next > sampleCount)
+        {
+            Interlocked.Increment(ref _sampleCountRaises);
+            _logger.LogInformation(
+                "MTConnect driver {DriverInstanceId} is falling behind {AgentUri}'s /sample stream (chunks full at count {Count}, lag {Lag} observation(s)); reopening at sequence {Cursor} with count {NextCount}.",
+                _driverInstanceId, options.AgentUri, sampleCount, DescribeLag(), cursor, next);
+
+            return next;
+        }
+
+        if (!state.CeilingWarned)
+        {
+            state.CeilingWarned = true;
+            _logger.LogWarning(
+                "MTConnect driver {DriverInstanceId} is still behind {AgentUri}'s /sample stream at its count ceiling {Count} (lag {Lag} observation(s)): the machine produces more than count / SampleIntervalMs observations a second, so values are backlog-stale and the stream will re-baseline (dropping intermediate transitions) whenever it falls out of the Agent's buffer. Lower SampleIntervalMs or raise SampleCount.",
+                _driverInstanceId, options.AgentUri, sampleCount, DescribeLag());
+        }
+
+        return null;
+    }
+
+    private string DescribeLag()
+    {
+        var lag = Interlocked.Read(ref _sampleLag);
+        return lag >= 0 ? lag.ToString(CultureInfo.InvariantCulture) : "unknown";
+    }
+}
@@ -1971,6 +1971,8 @@ public sealed partial class MTConnectDriver
        var cursor = from;
        var attempt = 0;
        _reconnectAttempts = 0;
+        var counts = new SampleCountState(options.SampleCount);
+        Volatile.Write(ref _effectiveSampleCount, counts.Count);

        try
        {
@@ -2001,7 +2003,7 @@ public sealed partial class MTConnectDriver
                    }
                }

-                var outcome = await ConsumeStreamAsync(client, options, cursor, instanceId, generation, ct).ConfigureAwait(false);
+                var outcome = await ConsumeStreamAsync(client, options, cursor, instanceId, generation, counts, ct).ConfigureAwait(false);
                cursor = outcome.Cursor;
                instanceId = outcome.InstanceId;

@@ -2012,6 +2014,30 @@ public sealed partial class MTConnectDriver
                    return;
                }

+                // R06:T-5 — the adaptive count. A stream that delivered vindicates the count it was opened with; a
+                // RAISED count whose stream delivered nothing may be one the Agent refuses (count above its buffer),
+                // so step back to the last count that worked and stop raising for this pump.
+                if (outcome.DeliveredAny)
+                {
+                    counts.LastDelivered = counts.Count;
+                }
+                else if (counts.Count > counts.LastDelivered && outcome.Verdict is not StreamVerdict.Cancelled)
+                {
+                    _logger.LogWarning(
+                        outcome.Error,
+                        "MTConnect driver {DriverInstanceId}: the /sample stream opened with the raised count {Count} delivered nothing; stepping back to {Previous} and not raising it again for this stream.",
+                        _driverInstanceId, counts.Count, counts.LastDelivered);
+                    counts.Count = counts.LastDelivered;
+                    counts.Ceiling = counts.Count;
+                    Volatile.Write(ref _effectiveSampleCount, counts.Count);
+                }
+
+                if (outcome.RaiseCountTo is { } raised)
+                {
+                    counts.Count = raised;
+                    Volatile.Write(ref _effectiveSampleCount, raised);
+                }
+
                switch (outcome.Verdict)
                {
                    case StreamVerdict.Cancelled:
@@ -2118,6 +2144,7 @@ public sealed partial class MTConnectDriver
        long cursor,
        long instanceId,
        long generation,
+        SampleCountState counts,
        CancellationToken ct)
    {
        // The RUNNING cursor the gap check is made against: the previous chunk's nextSequence, equal
@@ -2131,9 +2158,12 @@ public sealed partial class MTConnectDriver
        // that cannot work at all. They must not share a backoff ladder — see RunSampleStreamAsync.
        var delivered = false;

+        // Consecutive chunks that said this stream is behind the Agent (R06:T-5).
+        var behindChunks = 0;
+
        try
        {
-            await foreach (var chunk in client.SampleAsync(cursor, ct).ConfigureAwait(false))
+            await foreach (var chunk in client.SampleAsync(cursor, counts.Count, ct).ConfigureAwait(false))
            {
                if (chunk is null)
                {
@@ -2198,6 +2228,21 @@ public sealed partial class MTConnectDriver
                _streamDegraded = false;
                _pumpLive = true;
                PublishHealthy(DateTime.UtcNow);
+
+                // R06:T-5 — a stream persistently behind the Agent reopens at the cursor it reached with a larger
+                // count, rather than falling out of the ring buffer and re-baselining every buffer-length.
+                if (!MeasureSampleLag(chunk, counts.Count))
+                {
+                    behindChunks = 0;
+                }
+                else if (++behindChunks >= BehindChunksBeforeRaise)
+                {
+                    behindChunks = 0;
+                    if (NextSampleCount(counts, options, expected) is { } next)
+                    {
+                        return new StreamOutcome(StreamVerdict.Reopen, expected, instanceId, delivered, null, next);
+                    }
+                }
            }

            // The seam contracts that cancellation is the ONLY way this enumeration ends quietly, so
@@ -3894,8 +3939,12 @@ public sealed partial class MTConnectDriver
    ///     the reconnect backoff from ratcheting monotonically over a process lifetime.
    /// </param>
    /// <param name="Error">The failure, when there was one.</param>
+    /// <param name="RaiseCountTo">
+    ///     Set on a <see cref="StreamVerdict.Reopen"/> the adaptive count asked for: the stream was persistently behind
+    ///     the Agent, and the next one opens with this <c>count</c> (R06:T-5).
+    /// </param>
    private sealed record StreamOutcome(
-        StreamVerdict Verdict, long Cursor, long InstanceId, bool DeliveredAny, Exception? Error);
+        StreamVerdict Verdict, long Cursor, long InstanceId, bool DeliveredAny, Exception? Error, int? RaiseCountTo = null);

    // ---- config DTOs ----

@@ -56,7 +56,7 @@ public sealed record MTConnectComponent(
 /// <summary>
 ///     One MTConnect DataItem declaration from the Agent's probe response. This is pure metadata —
 ///     it carries no observed value; values arrive separately via
-///     <see cref="IMTConnectAgentClient.CurrentAsync"/> / <see cref="IMTConnectAgentClient.SampleAsync"/>
+///     <see cref="IMTConnectAgentClient.CurrentAsync"/> / <see cref="IMTConnectAgentClient.SampleAsync(long, CancellationToken)"/>
 ///     and are correlated back to a data item by <see cref="Id"/>.
 /// </summary>
 /// <param name="Id">
@@ -89,7 +89,7 @@ public sealed record MTConnectDataItem(
 /// <summary>
 ///     Result of an Agent <c>/current</c> request, or of a single multipart chunk from a
 ///     <c>/sample</c> stream. Produced by <see cref="IMTConnectAgentClient.CurrentAsync"/> and
-///     <see cref="IMTConnectAgentClient.SampleAsync"/>.
+///     <see cref="IMTConnectAgentClient.SampleAsync(long, CancellationToken)"/>.
 /// </summary>
 /// <param name="InstanceId">
 ///     The Agent's current instance id (its MTConnectStreams header <c>instanceId</c>). Changes
@@ -98,7 +98,7 @@ public sealed record MTConnectDataItem(
 /// </param>
 /// <param name="NextSequence">
 ///     The header <c>nextSequence</c> — the sequence number to resume a subsequent
-///     <see cref="IMTConnectAgentClient.SampleAsync"/> call <c>from</c>.
+///     <see cref="IMTConnectAgentClient.SampleAsync(long, CancellationToken)"/> call <c>from</c>.
 /// </param>
 /// <param name="FirstSequence">
 ///     The header <c>firstSequence</c> — the oldest sequence number still held in the Agent's
@@ -110,11 +110,17 @@ public sealed record MTConnectDataItem(
 ///     The observations carried in this snapshot/chunk, in Agent-supplied order. Never <c>null</c>;
 ///     empty when the chunk carries no new observations (e.g. a heartbeat).
 /// </param>
+/// <param name="LastSequence">
+///     The header <c>lastSequence</c> — the newest sequence number the Agent holds, or <c>null</c> when the
+///     document omits it. With <see cref="NextSequence"/> it measures how far a <c>/sample</c> consumer is behind
+///     the Agent: <c>LastSequence + 1 - NextSequence</c> observations it has not been sent yet (R06:T-5).
+/// </param>
 public sealed record MTConnectStreamsResult(
    long InstanceId,
    long NextSequence,
    long FirstSequence,
-    IReadOnlyList<MTConnectObservation> Observations);
+    IReadOnlyList<MTConnectObservation> Observations,
+    long? LastSequence = null);

 /// <summary>
 ///     One observed value for a single DataItem, as reported in a <c>/current</c> snapshot or a
@@ -7,7 +7,7 @@ namespace ZB.MOM.WW.OtOpcUa.Driver.MTConnect;
 /// <remarks>
 ///     <para>
 ///         <b>Why these are exceptions rather than a normal end of enumeration.</b>
-///         <see cref="IMTConnectAgentClient.SampleAsync"/> is contracted to yield indefinitely until
+///         <see cref="IMTConnectAgentClient.SampleAsync(long, CancellationToken)"/> is contracted to yield indefinitely until
 ///         its token is cancelled, so a pump written to that contract has no reason to handle normal
 ///         completion — it would either drop the subscription silently or hot-loop reconnecting on
 ///         an enumerator that ends immediately. Returning normally would make "the data path
@@ -330,7 +330,8 @@ internal static class MTConnectStreamsParser
            sequences.InstanceId,
            sequences.NextSequence,
            sequences.FirstSequence,
-            observations);
+            observations,
+            sequences.LastSequence);
    }

    /// <summary>
@@ -342,7 +343,14 @@ internal static class MTConnectStreamsParser
        new(
            MTConnectXml.RequiredLongAttribute(reader, "instanceId", HeaderOwner, Subject),
            MTConnectXml.RequiredLongAttribute(reader, "nextSequence", HeaderOwner, Subject),
-            MTConnectXml.RequiredLongAttribute(reader, "firstSequence", HeaderOwner, Subject));
+            MTConnectXml.RequiredLongAttribute(reader, "firstSequence", HeaderOwner, Subject),
+            // Optional on purpose: only the /sample lag measurement reads it, and a document without it must
+            // still parse (the pump then falls back to chunk saturation alone — R06:T-5).
+            long.TryParse(
+                MTConnectXml.OptionalAttribute(reader, "lastSequence"),
+                NumberStyles.Integer,
+                CultureInfo.InvariantCulture,
+                out var last) ? last : null);

    /// <summary>
    ///     Consumes the element the reader is positioned on, collecting any observation it is and
@@ -653,7 +661,7 @@ internal static class MTConnectStreamsParser
    }

    /// <summary>The three sequence numbers read off a <c>&lt;Header&gt;</c>.</summary>
-    private readonly record struct HeaderSequences(long InstanceId, long NextSequence, long FirstSequence);
+    private readonly record struct HeaderSequences(long InstanceId, long NextSequence, long FirstSequence, long? LastSequence);

    /// <summary>
    ///     Accumulates one observation's text across however many nodes the parser hands it, without
@@ -20,7 +20,7 @@ namespace ZB.MOM.WW.OtOpcUa.Driver.MTConnect.Tests;
 ///     </para>
 ///     <para>
 ///         <b>It honours the seam's stream-end contract</b> (see
-///         <see cref="IMTConnectAgentClient.SampleAsync"/>): cancelling the token is the only way the
+///         <see cref="IMTConnectAgentClient.SampleAsync(long, CancellationToken)"/>): cancelling the token is the only way the
 ///         enumeration ends without throwing. Closing the scripted stream via
 ///         <see cref="EndStream"/> raises <see cref="MTConnectStreamEndedException"/>, exactly as the
 ///         production client does when an Agent drops the connection — so a pump written against the
@@ -297,6 +297,20 @@ internal sealed class CannedAgentClient : IMTConnectAgentClient, IDisposable
        return CurrentFailure is null ? Current : throw CurrentFailure;
    }

+    /// <summary>The <c>count</c> the most recent <c>/sample</c> enumeration was opened with (null = the 2-arg overload).</summary>
+    public int? LastSampleCount { get; private set; }
+
+    /// <summary>Every <c>count</c> a <c>/sample</c> enumeration was opened with, in order.</summary>
+    public ConcurrentQueue<int> SampleCounts { get; } = new();
+
+    /// <inheritdoc/>
+    public IAsyncEnumerable<MTConnectStreamsResult> SampleAsync(long from, int count, CancellationToken ct)
+    {
+        LastSampleCount = count;
+        SampleCounts.Enqueue(count);
+        return SampleAsync(from, ct);
+    }
+
    /// <inheritdoc/>
    public async IAsyncEnumerable<MTConnectStreamsResult> SampleAsync(
        long from, [EnumeratorCancellation] CancellationToken ct)
@@ -0,0 +1,154 @@
+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-5 — <c>/sample</c> throughput was hard-capped at <c>SampleCount / SampleIntervalMs</c>
+///     with no lag detection: a machine producing more fell behind the Agent's ring buffer forever, re-baselining
+///     every buffer-length with every value backlog-stale and nothing surfacing it. The pump now measures the lag
+///     (<c>lastSequence + 1 − nextSequence</c>, plus chunk saturation), surfaces it on the diagnostics bag, and raises
+///     the per-stream <c>count</c> when it is persistently behind.
+/// </summary>
+public sealed class MTConnectSampleLagTests
+{
+    private const long Instance = 1655000000L;
+    private const long Primed = 108L;
+    private static readonly TimeSpan Watchdog = TimeSpan.FromSeconds(30);
+
+    private static CancellationToken Ct => TestContext.Current.CancellationToken;
+
+    /// <summary>A chunk of <paramref name="observations"/> dev1_pos values from <paramref name="first"/>.</summary>
+    private static MTConnectStreamsResult Chunk(long first, int observations, long? last)
+    {
+        var obs = Enumerable.Range(0, observations)
+            .Select(i => new MTConnectObservation(
+                "dev1_pos", (first + i).ToString(System.Globalization.CultureInfo.InvariantCulture),
+                new DateTime(2026, 10, 22, 12, 0, 0, DateTimeKind.Utc).AddSeconds(first + i)))
+            .ToList();
+        return new MTConnectStreamsResult(Instance, first + observations, first, obs, last);
+    }
+
+    private static async Task<(MTConnectDriver Driver, CannedAgentClient Client)> SubscribedDriverAsync(int sampleCount)
+    {
+        var client = CannedAgentClient.FromFixtures();
+        var driver = new MTConnectDriver(
+            new MTConnectDriverOptions
+            {
+                AgentUri = "http://fixture-agent:5000",
+                RequestTimeoutMs = 5000,
+                SampleCount = sampleCount,
+                Tags = [new MTConnectTagDefinition("dev1_pos", DriverDataType.Float64)],
+            },
+            "mt-t5",
+            _ => client);
+        await driver.InitializeAsync("{}", Ct);
+        await driver.SubscribeAsync(["dev1_pos"], TimeSpan.FromMilliseconds(50), Ct);
+        await client.WaitForSampleCallsAsync(1).WaitAsync(Watchdog, Ct);
+        return (driver, client);
+    }
+
+    /// <summary>Three full chunks with a positive lag reopen the stream, at the cursor reached, with count doubled.</summary>
+    [Fact]
+    public async Task A_persistently_behind_stream_reopens_with_a_doubled_count()
+    {
+        var (driver, client) = await SubscribedDriverAsync(sampleCount: 2);
+        try
+        {
+            await client.PumpAsync(Chunk(Primed, 2, last: 900)).WaitAsync(Watchdog, Ct);
+            await client.PumpAsync(Chunk(Primed + 2, 2, last: 900)).WaitAsync(Watchdog, Ct);
+            await client.PumpAsync(Chunk(Primed + 4, 2, last: 900)).WaitAsync(Watchdog, Ct);
+
+            await client.WaitForSampleCallsAsync(2).WaitAsync(Watchdog, Ct);
+            client.SampleCounts.ShouldBe([2, 4]);
+            client.LastSampleFrom.ShouldBe(Primed + 6, "the raised stream resumes where the behind one got to");
+
+            var diag = driver.GetDiagnostics();
+            diag["sampleCount"].ShouldBe("4");
+            diag["sampleLag"].ShouldBe((900 + 1 - (Primed + 6)).ToString(System.Globalization.CultureInfo.InvariantCulture));
+            diag["sampleSaturatedChunks"].ShouldBe("3");
+            diag["sampleCountRaises"].ShouldBe("1");
+        }
+        finally
+        {
+            await driver.ShutdownAsync(CancellationToken.None);
+        }
+    }
+
+    /// <summary>An Agent that omits lastSequence: persistent saturation alone is enough.</summary>
+    [Fact]
+    public async Task Persistent_saturation_without_a_lastSequence_still_raises_the_count()
+    {
+        var (driver, client) = await SubscribedDriverAsync(sampleCount: 2);
+        try
+        {
+            for (var i = 0; i < 3; i++)
+                await client.PumpAsync(Chunk(Primed + (2 * i), 2, last: null)).WaitAsync(Watchdog, Ct);
+
+            await client.WaitForSampleCallsAsync(2).WaitAsync(Watchdog, Ct);
+            client.SampleCounts.ShouldBe([2, 4]);
+            driver.GetDiagnostics().ContainsKey("sampleLag").ShouldBeFalse("an unknown lag is not reported as a number");
+        }
+        finally
+        {
+            await driver.ShutdownAsync(CancellationToken.None);
+        }
+    }
+
+    /// <summary>
+    ///     Full chunks with NO lag (the Agent happened to have exactly count observations) are not "behind": the stream
+    ///     is kept. Positive control: the first test, through the same pump, does reopen.
+    /// </summary>
+    [Fact]
+    public async Task Full_chunks_with_no_lag_keep_the_stream()
+    {
+        var (driver, client) = await SubscribedDriverAsync(sampleCount: 2);
+        try
+        {
+            for (var i = 0; i < 4; i++)
+            {
+                var first = Primed + (2 * i);
+                await client.PumpAsync(Chunk(first, 2, last: first + 1)).WaitAsync(Watchdog, Ct);
+            }
+
+            client.SampleCallCount.ShouldBe(1, "a caught-up stream must not be reopened");
+            driver.GetDiagnostics()["sampleLag"].ShouldBe("0");
+            driver.GetDiagnostics()["sampleCountRaises"].ShouldBe("0");
+        }
+        finally
+        {
+            await driver.ShutdownAsync(CancellationToken.None);
+        }
+    }
+
+    /// <summary>A raised count the Agent refuses (its stream delivers nothing) steps back to the last count that worked.</summary>
+    [Fact]
+    public async Task A_refused_raised_count_steps_back_to_the_last_count_that_delivered()
+    {
+        var (driver, client) = await SubscribedDriverAsync(sampleCount: 2);
+        try
+        {
+            // The NEXT enumeration (the raised one) is refused the way a real Agent refuses a count above its buffer.
+            client.SampleFailures.Enqueue(new InvalidDataException("MTConnectError: INVALID_REQUEST count too large"));
+            for (var i = 0; i < 3; i++)
+                await client.PumpAsync(Chunk(Primed + (2 * i), 2, last: 900)).WaitAsync(Watchdog, Ct);
+
+            await client.WaitForSampleCallsAsync(3).WaitAsync(Watchdog, Ct);
+            client.SampleCounts.Take(3).ShouldBe([2, 4, 2]);
+            driver.GetDiagnostics()["sampleCount"].ShouldBe("2");
+        }
+        finally
+        {
+            await driver.ShutdownAsync(CancellationToken.None);
+        }
+    }
+
+    /// <summary>The ceiling is 16× the configured count, capped at 65 536, and never below the configured count.</summary>
+    [Theory]
+    [InlineData(1000, 16000)]
+    [InlineData(10_000, 65_536)]
+    [InlineData(100_000, 100_000)]
+    public void The_count_ceiling_is_bounded(int configured, int expected) =>
+        MTConnectDriver.SampleCountCeiling(configured).ShouldBe(expected);
+}
@@ -593,6 +593,38 @@ public sealed class MTConnectStreamsParseTests
            "http://agent:5000/VMC-3Axis/sample?from=108&interval=250&count=64&heartbeat=7000");
    }

+    /// <summary>R06:T-5 — the count overload sends the raised count in place of the configured one.</summary>
+    [Fact]
+    public async Task SampleAsync_with_a_count_sends_that_count_instead_of_the_configured_one()
+    {
+        var handler = StubHandler.RespondingWithMultipart("bnd", await File.ReadAllTextAsync(SampleFixture, Ct));
+        using var client = NewClient(handler, new MTConnectDriverOptions
+        {
+            AgentUri = "http://agent:5000",
+            DeviceName = "VMC-3Axis",
+            SampleIntervalMs = 250,
+            SampleCount = 64,
+            HeartbeatMs = 7000
+        });
+
+        await DrainAsync(client.SampleAsync(108, 256, Ct));
+
+        handler.LastRequestUri!.AbsoluteUri.ShouldBe(
+            "http://agent:5000/VMC-3Axis/sample?from=108&interval=250&count=256&heartbeat=7000");
+    }
+
+    /// <summary>R06:T-5 — the header's lastSequence is carried (it measures the stream's lag); absent ⇒ null.</summary>
+    [Fact]
+    public async Task The_header_lastSequence_is_carried_when_present_and_null_when_absent()
+    {
+        var current = MTConnectStreamsParser.Parse(await File.ReadAllTextAsync("Fixtures/current.xml", Ct));
+        current.LastSequence.ShouldBe(107);
+
+        var stripped = (await File.ReadAllTextAsync("Fixtures/current.xml", Ct)).Replace("lastSequence=\"107\"", string.Empty);
+        stripped.ShouldNotContain("lastSequence");
+        MTConnectStreamsParser.Parse(stripped).LastSequence.ShouldBeNull();
+    }
+
    [Fact]
    public async Task SampleAsync_yields_one_result_per_multipart_chunk()
    {