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

using System.Globalization;
using System.Text.Json;
using Microsoft.Extensions.Logging;
using Microsoft.Extensions.Logging.Abstractions;
using ZB.MOM.WW.OtOpcUa.Core.Abstractions;

namespace ZB.MOM.WW.OtOpcUa.Driver.MTConnect;

/// <summary>
///     MTConnect Agent implementation of <see cref="IDriver"/> — the lifecycle half. Everything the
///     driver serves sits behind one <see cref="IMTConnectAgentClient"/>, so the whole class is
///     exercised against canned XML with no socket.
/// </summary>
/// <remarks>
///     <para>
///         <b>The constructor opens nothing and builds nothing.</b> It does not even call
///         <c>agentClientFactory</c> — the Wave-0 universal discovery browser constructs a throwaway
///         instance purely to ask whether the driver can browse, and a constructor that materialized
///         a client would open a connection pool per browse probe against a possibly-unreachable
///         Agent. Every connection is owned by <see cref="InitializeAsync"/>.
///     </para>
///     <para>
///         <b>The <c>driverConfigJson</c> argument is honoured, not decorative.</b> The runtime hands
///         a config change to a <i>live</i> instance — <c>DriverInstanceActor</c>'s <c>ApplyDelta</c>
///         calls <see cref="ReinitializeAsync"/> with the new document and never rebuilds the object
///         — so a driver that read only its constructor options would turn every MTConnect config
///         edit into a silent no-op that still seals the deployment green. This class therefore
///         re-parses the document (<see cref="ParseOptions"/>) whenever it carries a real body,
///         matching S7/AbCip/TwinCAT/Galaxy rather than the Modbus/FOCAS/OpcUaClient drivers that
///         ignore the argument. A blank / <c>{}</c> / <c>[]</c> document means "no config supplied"
///         and keeps the constructor options, which is what unit tests and the factory path rely on.
///     </para>
///     <para>
///         <b>Failure is never dressed as success (#485).</b> A failed initialize disposes whatever
///         client it built, drops the probe cache and the observation index, publishes
///         <see cref="DriverState.Faulted"/> with the error text, and rethrows — the runtime's
///         retry/backoff loop is the recovery path. In particular a <c>/probe</c> that succeeds while
///         the priming <c>/current</c> fails is Faulted, not "Healthy with no data": the index IS the
///         read surface, so serving that state would render every tag
///         <c>BadWaitingForInitialData</c> indefinitely behind a green driver, and the
///         <c>/sample</c> cursor (Task 11) only exists because <c>/current</c> reported a
///         <c>nextSequence</c>.
///     </para>
///     <para>
///         <b>Scope.</b> This type implements <see cref="IDriver"/>, <see cref="IReadable"/>,
///         <see cref="ISubscribable"/>, <see cref="ITagDiscovery"/>,
///         <see cref="IHostConnectivityProbe"/> and <see cref="IRediscoverable"/> — the last two
///         built on state this class already caches (the Agent <c>instanceId</c> and the probe
///         model). There is deliberately no <c>IWritable</c>: MTConnect is a read-only protocol,
///         which is also why every discovered leaf is
///         <see cref="SecurityClassification.ViewOnly"/>.
///     </para>
///     <para>
///         <b>The read path never writes the shared observation index</b> — see
///         <see cref="ReadAsync"/>. That index has exactly one writer: the <c>/sample</c> pump
///         (<see cref="RunSampleStreamAsync"/>). The same single-writer discipline covers the two
///         Task-13 fields: the <b>connectivity probe loop</b> is the sole writer of
///         <see cref="HostState"/>, and it publishes <b>nothing</b> into
///         <see cref="AgentSession.ProbeModel"/> (see <see cref="RunProbeLoopAsync"/>).
///     </para>
/// </remarks>
public sealed class MTConnectDriver : IDriver, IReadable, ISubscribable, ITagDiscovery, IHostConnectivityProbe, IRediscoverable
{
    /// <summary>
    ///     Coarse per-entry byte weights behind <see cref="GetMemoryFootprint"/>. The contract asks
    ///     for an <i>approximate</i> driver-attributable footprint that Core polls every 30 s to
    ///     decide whether to ask for a flush, so an order-of-magnitude estimate from the two things
    ///     that actually scale with a plant (declared data items, live observations) is the right
    ///     precision. Measuring real retained size would need a heap walk per poll.
    /// </summary>
    private const int ProbeModelBytesPerDataItem = 512;

    /// <summary>Estimated retained bytes per live <c>dataItemId → snapshot</c> entry.</summary>
    private const int IndexBytesPerEntry = 256;

    /// <summary>Estimated retained bytes per authored <see cref="MTConnectTagDefinition"/>.</summary>
    private const int TagBytesPerEntry = 256;

    /// <summary>
    ///     The Agent could not be reached, answered unusably, or blew the per-call deadline. Distinct
    ///     from the index's <c>BadNoCommunication</c>, which means the Agent answered and said the
    ///     machine has no value — the two failures need different operator responses.
    /// </summary>
    private const uint BadCommunicationError = 0x80050000u;

    /// <summary>
    ///     There is no Agent client at all: the driver has not been initialized, its initialize
    ///     faulted, or it has been shut down. Deliberately not one of the index's codes — none of
    ///     them would say "this driver is not running", which is the only true statement available.
    /// </summary>
    private const uint BadNotConnected = 0x808A0000u;

    /// <summary>
    ///     Floor the reconnect backoff grows from, in milliseconds. Load-bearing: the shipped
    ///     <see cref="MTConnectReconnectOptions.MinBackoffMs"/> default is <b>0</b> (an immediate
    ///     first retry, which is what an operator wants after a one-off blip) and
    ///     <c>0 × multiplier</c> is still 0 — so without a floor, the "geometric" backoff would be
    ///     an unbounded hot loop against an Agent that stays down. Mirrors
    ///     <c>ModbusTcpTransport.ConnectWithBackoffAsync</c>.
    /// </summary>
    private const int BackoffGrowthFloorMs = 100;

    /// <summary>Growth factor used when the authored multiplier could not grow the delay (≤ 1).</summary>
    private const double DefaultBackoffMultiplier = 2.0;

    /// <summary>
    ///     Backoff cap used when <see cref="MTConnectReconnectOptions.MaxBackoffMs"/> is authored
    ///     non-positive. Treated as "unset" rather than as "no cap" for the same reason a
    ///     multiplier ≤ 1 falls back to <see cref="DefaultBackoffMultiplier"/>: a zero cap clamps
    ///     EVERY delay to zero, which turns the reconnect loop into a spin against a refused
    ///     connection — one Warning per iteration, as fast as the socket can fail. Mirrors the
    ///     shipped <see cref="MTConnectReconnectOptions.MaxBackoffMs"/> default.
    /// </summary>
    private const int DefaultMaxBackoffMs = 30_000;

    /// <summary>
    ///     How long teardown waits for a background loop (the <c>/sample</c> pump, the connectivity
    ///     probe) to unwind before abandoning the wait and carrying on.
    /// </summary>
    /// <remarks>
    ///     <b>An unbounded wait here is not merely slow — it wedges an actor-system thread.</b>
    ///     <c>DriverInstanceActor.PostStop</c> calls <see cref="ShutdownAsync"/> with
    ///     <c>GetAwaiter().GetResult()</c>, i.e. a synchronous block on an Akka dispatcher thread,
    ///     while this driver holds <see cref="_lifecycle"/>. So a subscriber callback that never
    ///     returns would take the dispatcher thread AND every subsequent lifecycle call on this
    ///     driver with it, permanently. Five seconds matches the budget
    ///     <c>DriverInstanceActor</c> already puts around <see cref="UnsubscribeAsync"/>.
    /// </remarks>
    private static readonly TimeSpan TeardownWait = TimeSpan.FromSeconds(5);

    /// <summary>
    ///     Config-JSON reader options, mirroring the sibling driver factories. Note there is
    ///     deliberately <b>no</b> <c>JsonStringEnumConverter</c>: enum-carrying DTO fields stay
    ///     <c>string?</c> and go through <see cref="ParseEnum{T}"/>, which is the project-wide
    ///     guard against the numerically-serialized-enum trap that faults a driver at deploy time.
    /// </summary>
    private static readonly JsonSerializerOptions JsonOptions = new()
    {
        PropertyNameCaseInsensitive = true,
        ReadCommentHandling = JsonCommentHandling.Skip,
        AllowTrailingCommas = true,
    };

    /// <summary>
    ///     Reader options for the emptiness probe in <see cref="HasConfigBody"/>. Kept in step with
    ///     <see cref="JsonOptions"/> so a document the real parse would accept is never judged
    ///     malformed here (a stray trailing comma must not change which branch a config takes).
    /// </summary>
    private static readonly JsonDocumentOptions DocumentOptions = new()
    {
        CommentHandling = JsonCommentHandling.Skip,
        AllowTrailingCommas = true,
    };

    private readonly string _driverInstanceId;
    private readonly ILogger<MTConnectDriver> _logger;
    private readonly Func<MTConnectDriverOptions, IMTConnectAgentClient> _agentClientFactory;

    /// <summary>
    ///     How the connectivity probe loop waits between ticks. Injected so the cadence is a seam
    ///     rather than a wall clock: the loop is a timer, and a suite that drove it by sleeping would
    ///     be flaky by construction. Defaults to <see cref="Task.Delay(TimeSpan, CancellationToken)"/>.
    /// </summary>
    private readonly Func<TimeSpan, CancellationToken, Task> _probeScheduler;

    /// <summary>
    ///     Serializes the three lifecycle entry points against each other. Initialize / Reinitialize
    ///     / Shutdown all swap the client and the served caches; overlapping them would let a
    ///     shutdown dispose a client an in-flight initialize is about to publish.
    /// </summary>
    /// <remarks>
    ///     <para>
    ///         <b>⚠️ NON-REENTRANT — never call a public lifecycle method from code that already
    ///         holds this.</b> <see cref="SemaphoreSlim"/> has no owner tracking, so a re-entrant
    ///         call does not fail: it <b>deadlocks the driver permanently</b>, with no exception and
    ///         no log line to diagnose it from.
    ///     </para>
    ///     <para>
    ///         <b>This is aimed squarely at the <c>/sample</c> pump (Task 11).</b> Its ring-buffer
    ///         re-baseline ("on a sequence gap, re-<c>/current</c> and resume") is naturally written
    ///         as "just call the existing re-init logic" — and
    ///         <see cref="ReinitializeAsync"/>/<see cref="InitializeAsync"/> both take this
    ///         semaphore, so from inside the pump loop that is a hang, not a re-prime. Re-baseline
    ///         must instead call <see cref="IMTConnectAgentClient.CurrentAsync"/> directly on the
    ///         client the pump already holds (through <see cref="BoundedAsync"/>, to keep the
    ///         deadline) and <c>Apply</c> the result to <see cref="ObservationIndex"/>. Any future
    ///         shared step belongs in a <b>private, non-locking</b> helper that both the pump and the
    ///         lock-holding lifecycle methods can call — never in the public entry points.
    ///     </para>
    ///     <para>
    ///         The inverse direction is already safe: the lifecycle methods call
    ///         <see cref="StopSampleStreamAsync"/> while holding this, so that method — and whatever
    ///         Task 11 fills it with — must never re-enter the lifecycle either.
    ///     </para>
    /// </remarks>
    private readonly SemaphoreSlim _lifecycle = new(1, 1);

    /// <summary>
    ///     Guards the subscription registry and the pump's control block (<see cref="_pumpCts"/>,
    ///     <see cref="_pumpTask"/>, <see cref="_streamUnsupported"/>).
    /// </summary>
    /// <remarks>
    ///     <para>
    ///         <b>Lock order is <see cref="_lifecycle"/> → this, never the reverse.</b> The lifecycle
    ///         methods take this lock (to start/stop the pump) while holding the semaphore;
    ///         <see cref="UnsubscribeAsync"/> therefore releases this lock <i>before</i> awaiting
    ///         <see cref="StopSampleStreamAsync"/>, and nothing under this lock ever waits on
    ///         <see cref="_lifecycle"/>.
    ///     </para>
    ///     <para>
    ///         <b>Nothing is awaited while it is held, and no subscriber callback is raised under
    ///         it.</b> A handler is caller code: it may block, throw, or re-enter this driver, and
    ///         holding a lock across it would turn a slow subscriber into a stalled pump.
    ///     </para>
    /// </remarks>
    private readonly Lock _subscriptionLock = new();

    /// <summary>Live subscriptions by id. Guarded by <see cref="_subscriptionLock"/>.</summary>
    private readonly Dictionary<long, SubscriptionState> _subscriptions = [];

    /// <summary>
    ///     Guards the two host-connectivity fields, which the probe loop writes and
    ///     <see cref="GetHostStatuses"/> reads from an arbitrary caller thread. Held for a field
    ///     read/write only — <b>never</b> across an await, and never while a status handler runs.
    /// </summary>
    private readonly Lock _probeLock = new();

    /// <summary>
    ///     The Agent's last observed reachability. Written <b>only</b> by
    ///     <see cref="RunProbeLoopAsync"/>, so the state and the event stream can never disagree:
    ///     every value this field has ever held was announced by exactly one
    ///     <see cref="OnHostStatusChanged"/>. That is why a successful <see cref="InitializeAsync"/>
    ///     does not seed it <see cref="HostState.Running"/> — raising the event from there would run
    ///     caller code while the non-reentrant <see cref="_lifecycle"/> semaphore is held (the same
    ///     hazard that makes the pump, not <see cref="StartCoreAsync"/>, republish to subscribers),
    ///     and seeding it silently would leave a consumer that tracks the event stream disagreeing
    ///     with <see cref="GetHostStatuses"/>. <see cref="HostState.Unknown"/> until the first tick
    ///     is the honest answer. Guarded by <see cref="_probeLock"/>.
    /// </summary>
    private HostState _hostState = HostState.Unknown;

    /// <summary>When <see cref="_hostState"/> last changed. Guarded by <see cref="_probeLock"/>.</summary>
    private DateTime _hostStateChangedUtc = DateTime.UtcNow;

    /// <summary>
    ///     Cancels the connectivity probe loop. Written only under <see cref="_lifecycle"/>, which
    ///     serializes every start and stop of the loop.
    /// </summary>
    private CancellationTokenSource? _probeCts;

    /// <summary>The connectivity probe loop's task, or <c>null</c> when none runs. See <see cref="_probeCts"/>.</summary>
    private Task? _probeTask;

    /// <summary>
    ///     The Agent <c>instanceId</c> the last <see cref="OnRediscoveryNeeded"/> was raised for —
    ///     seeded at the commit point of <see cref="StartCoreAsync"/> with the id the priming
    ///     <c>/current</c> reported. This is what makes rediscovery fire once per <b>change</b>
    ///     rather than once per chunk: every chunk after a restart carries the new id, and a
    ///     re-baseline that fails leaves the pump still holding the old one.
    /// </summary>
    private long _announcedInstanceId;

    private MTConnectDriverOptions _options;
    private MTConnectObservationIndex _index;
    private DriverHealth _health = new(DriverState.Unknown, null, null);
    private long _nextSubscriptionId;

    /// <summary>Cancels the shared <c>/sample</c> pump. Guarded by <see cref="_subscriptionLock"/>.</summary>
    private CancellationTokenSource? _pumpCts;

    /// <summary>The shared <c>/sample</c> pump task. Guarded by <see cref="_subscriptionLock"/>.</summary>
    private Task? _pumpTask;

    /// <summary>
    ///     The Agent answered <c>/sample</c> with something that cannot be streamed at all, so the
    ///     pump gave up. Latched — reconnecting reproduces the identical answer forever, and a
    ///     later <see cref="SubscribeAsync"/> is not new information. Cleared only by a lifecycle
    ///     start, which is the one event that means an operator changed something. Guarded by
    ///     <see cref="_subscriptionLock"/>.
    /// </summary>
    private bool _streamUnsupported;

    /// <summary>
    ///     Whether the <c>/sample</c> pump is currently failing. Kept separate from
    ///     <see cref="_readDegraded"/> because <c>/sample</c> and <c>/current</c> are different
    ///     Agent requests that fail independently — see <see cref="PublishHealthy"/>.
    /// </summary>
    private volatile bool _streamDegraded;

    /// <summary>Whether the <c>/current</c> read path is currently failing. See <see cref="PublishHealthy"/>.</summary>
    private volatile bool _readDegraded;

    /// <summary>
    ///     Consecutive <c>/sample</c> reconnect attempts that have <b>not</b> yet been vindicated by
    ///     a delivered chunk — the input to <see cref="BackoffFor"/>. Reset by any stream that
    ///     actually delivered something, so an agent that drops a connection once an hour is never
    ///     treated as one that has been failing all day.
    /// </summary>
    private volatile int _reconnectAttempts;

    /// <summary>
    ///     Latch behind the once-per-stream-generation subscriber-fault Warning. A consumer that
    ///     throws on every value would otherwise emit one Warning per observation <i>per handle</i>
    ///     — thousands a second on a busy Agent, which buries the very log it is trying to raise.
    /// </summary>
    private int _subscriberFaultLogged;

    /// <summary>
    ///     The live agent client plus the <c>/probe</c> cache derived from it, held as <b>one
    ///     immutable snapshot behind a single reference</b> and only ever replaced by a
    ///     <see cref="Volatile.Write{T}"/> of a whole new instance — the same discipline
    ///     <see cref="_index"/> uses, and for the same reason.
    /// </summary>
    /// <remarks>
    ///     <para>
    ///         <b>Why one field instead of three.</b> These three values are read <i>outside</i>
    ///         <see cref="_lifecycle"/> — by <see cref="ReadAsync"/>, by
    ///         <see cref="GetOrFetchProbeModelAsync"/> (Task 12's browse), and by
    ///         <see cref="GetMemoryFootprint"/> — while every write happens under it. As three
    ///         separate fields they could be observed mid-update: a reader could see a dropped
    ///         <c>ProbeModel</c> beside a stale non-zero data-item count and report a footprint for
    ///         a cache that no longer exists, or pair one initialize's client with the previous
    ///         one's model. Packing them makes every observation internally consistent by
    ///         construction, with no lock on the read path.
    ///     </para>
    ///     <para>
    ///         <b>Why not simply take the lock in those readers.</b> Both readers <c>await</c> the
    ///         network, so they would hold the lifecycle lock across an HTTP round trip — a browse
    ///         could stall a <see cref="ShutdownAsync"/> for a full <c>RequestTimeoutMs</c> — and
    ///         every added lock site widens the non-reentrancy hazard documented on
    ///         <see cref="_lifecycle"/>. It would also make the read path inconsistent with
    ///         <see cref="ReadAsync"/>, which is deliberately lock-free.
    ///     </para>
    ///     <para>
    ///         <b>What this does NOT fix.</b> A reader can still capture a session moments before a
    ///         concurrent teardown disposes its client, and then call it — no lock-free scheme can
    ///         prevent that, and holding the lock across the call is the trade rejected above.
    ///         What it guarantees is that the failure is <i>named</i>: the resulting
    ///         <see cref="ObjectDisposedException"/> is caught at the seam and reported as the same
    ///         "driver is not connected" error a caller already has to handle, rather than escaping
    ///         raw out of a browse.
    ///     </para>
    /// </remarks>
    private AgentSession? _session;

    /// <summary>Creates a driver instance. Opens no connection and builds no agent client.</summary>
    /// <param name="options">
    ///     The typed configuration this instance starts from. A config document supplied to
    ///     <see cref="InitializeAsync"/> / <see cref="ReinitializeAsync"/> supersedes it.
    /// </param>
    /// <param name="driverInstanceId">Stable logical id of this driver instance, from the config DB.</param>
    /// <param name="agentClientFactory">
    ///     Builds the Agent client from the effective options. Defaults to the production
    ///     <see cref="MTConnectAgentClient"/>; tests inject a canned-XML fake. Called from
    ///     <see cref="InitializeAsync"/>, never from this constructor.
    /// </param>
    /// <param name="logger">Optional logger; defaults to the null logger.</param>
    /// <param name="probeScheduler">
    ///     How the connectivity probe loop waits between ticks. Defaults to
    ///     <see cref="Task.Delay(TimeSpan, CancellationToken)"/>; tests inject a hand-driven ticker so
    ///     the probe's cadence is asserted deterministically instead of slept through.
    /// </param>
    public MTConnectDriver(
        MTConnectDriverOptions options,
        string driverInstanceId,
        Func<MTConnectDriverOptions, IMTConnectAgentClient>? agentClientFactory = null,
        ILogger<MTConnectDriver>? logger = null,
        Func<TimeSpan, CancellationToken, Task>? probeScheduler = null)
    {
        ArgumentNullException.ThrowIfNull(options);
        ArgumentException.ThrowIfNullOrWhiteSpace(driverInstanceId);

        _options = options;
        _driverInstanceId = driverInstanceId;
        _logger = logger ?? NullLogger<MTConnectDriver>.Instance;
        _agentClientFactory = agentClientFactory ?? (o => new MTConnectAgentClient(o, logger));
        _probeScheduler = probeScheduler ?? Task.Delay;

        // Never null, so every consumer (Task 10's ReadAsync included) can ask it for a snapshot
        // without a null check: an un-primed index answers BadWaitingForInitialData for an authored
        // tag and BadNodeIdUnknown otherwise, which is exactly the truth before Initialize runs.
        _index = new MTConnectObservationIndex(options.Tags);
    }

    /// <inheritdoc/>
    public string DriverInstanceId => _driverInstanceId;

    /// <inheritdoc/>
    /// <remarks>
    ///     Sourced from the <see cref="DriverTypeNames"/> constant rather than a literal — the
    ///     repo-wide fix for the historical <c>TwinCat</c>/<c>Focas</c> drift, where a driver's own
    ///     spelling diverged from the constant every dispatch map keys off.
    /// </remarks>
    public string DriverType => DriverTypeNames.MTConnect;

    /// <summary>
    ///     The live <c>dataItemId → snapshot</c> map that backs the <b>subscription</b> surface:
    ///     primed by <see cref="InitializeAsync"/> and thereafter written only by the <c>/sample</c>
    ///     pump (Task 11). Exposed to tests as the observable proof that initialize actually primed
    ///     it — and, in <c>MTConnectReadTests</c>, that <see cref="ReadAsync"/> leaves it alone.
    /// </summary>
    /// <remarks>
    ///     <b>Read does not consume this.</b> <see cref="ReadAsync"/> indexes its own <c>/current</c>
    ///     into a per-call snapshot instead, so the pump keeps a single writer — see the failure
    ///     analysis in <see cref="ReadAsync"/>'s remarks.
    /// </remarks>
    internal MTConnectObservationIndex ObservationIndex => Volatile.Read(ref _index);

    /// <summary>
    ///     The cached <c>/probe</c> device model, or <c>null</c> when it has not been fetched or was
    ///     dropped by <see cref="FlushOptionalCachesAsync"/>. Prefer
    ///     <see cref="GetOrFetchProbeModelAsync"/>, which cannot observe the flushed hole.
    /// </summary>
    internal MTConnectProbeModel? CachedProbeModel => Volatile.Read(ref _session)?.ProbeModel;

    /// <summary>
    ///     The Agent's <c>instanceId</c> as of the last successful <c>/current</c> — including the
    ///     re-baseline the <c>/sample</c> pump runs when it sees the id change mid-stream. Task 13
    ///     watches this for change to raise rediscovery (an Agent restart invalidates every cached
    ///     id).
    /// </summary>
    internal long? AgentInstanceId => Volatile.Read(ref _session)?.InstanceId;

    /// <summary>
    ///     The sequence the next <c>/sample</c> stream will open at — the priming <c>/current</c>'s
    ///     <c>nextSequence</c>, advanced by every re-baseline and by the pump as it stops. Paired
    ///     with <see cref="AgentInstanceId"/> in one session record, so the two are never observed
    ///     out of step.
    /// </summary>
    internal long? AgentNextSequence => Volatile.Read(ref _session)?.NextSequence;

    /// <summary>
    ///     Consecutive <c>/sample</c> reconnect attempts not yet vindicated by a delivered chunk —
    ///     the input to <see cref="BackoffFor"/>, exposed so the reset policy can be asserted
    ///     without measuring how long anything slept.
    /// </summary>
    internal int ReconnectAttempts => _reconnectAttempts;

    /// <summary>
    ///     The shared <c>/sample</c> pump's task while one is running, else <c>null</c>. Exposed to
    ///     tests as the deterministic teardown barrier: "the pump has stopped" is a task completion,
    ///     never a sleep. Guarded, so a test never observes a half-installed pump.
    /// </summary>
    internal Task? SampleStreamTask
    {
        get
        {
            lock (_subscriptionLock)
            {
                return _pumpTask;
            }
        }
    }

    /// <summary>
    ///     The connectivity probe loop's task while one is running, else <c>null</c>. Exposed to
    ///     tests as the deterministic teardown barrier — "the probe loop has stopped" is a task
    ///     completion, never a sleep — and as the proof that <c>Probe.Enabled = false</c> starts
    ///     nothing at all.
    /// </summary>
    internal Task? ProbeLoopTask => Volatile.Read(ref _probeTask);

    /// <summary>The options currently in force — the constructor's, or the last document applied.</summary>
    internal MTConnectDriverOptions EffectiveOptions => _options;

    /// <inheritdoc/>
    /// <remarks>
    ///     Builds the Agent client from the effective options, runs one deadline-bounded
    ///     <c>/probe</c> (caching the device model and its data-item count), then one
    ///     deadline-bounded <c>/current</c> to prime the observation index and record the Agent's
    ///     <c>instanceId</c>. Publishes <see cref="DriverState.Healthy"/> on success; on any failure
    ///     disposes the client, clears the caches, publishes <see cref="DriverState.Faulted"/> with
    ///     the error text, and rethrows.
    /// </remarks>
    public async Task InitializeAsync(string driverConfigJson, CancellationToken cancellationToken)
    {
        await _lifecycle.WaitAsync(cancellationToken).ConfigureAwait(false);
        try
        {
            // Parse BEFORE tearing down — see ResolveIncomingOptions. Initialize is reachable on a
            // live instance (DriverInstanceActor re-initialises in Connecting/Reconnecting), and
            // destroying a working client before discovering the new document is unreadable would
            // be the exact outcome ReinitializeAsync is written to avoid.
            var options = ResolveIncomingOptions(driverConfigJson);

            // A re-Initialize on a live instance must not orphan the previous client — nor the pump
            // that is enumerating it. Stopping the stream FIRST is the same ordering
            // ReinitializeAsync and ShutdownAsync use, and for the same reason: a pump left
            // enumerating a disposed client reads the disposal as a dropped connection and
            // reconnects against an object that no longer exists.
            await StopSampleStreamAsync(cancellationToken).ConfigureAwait(false);
            await StopProbeLoopAsync(cancellationToken).ConfigureAwait(false);
            await TeardownCoreAsync().ConfigureAwait(false);

            await StartCoreAsync(options, existingClient: null, cancellationToken).ConfigureAwait(false);
        }
        finally
        {
            _lifecycle.Release();
        }
    }

    /// <inheritdoc/>
    /// <remarks>
    ///     <para>
    ///         Stops the <c>/sample</c> stream, then takes one of two paths depending on whether the
    ///         new document changes anything the <see cref="IMTConnectAgentClient"/> reads at
    ///         <b>construction</b>:
    ///     </para>
    ///     <para>
    ///         <b>Endpoint/transport changed</b> (<c>AgentUri</c>, <c>DeviceName</c>,
    ///         <c>RequestTimeoutMs</c>, <c>HeartbeatMs</c>, <c>SampleIntervalMs</c>,
    ///         <c>SampleCount</c>) ⇒ full teardown and rebuild. The plan names only the first two,
    ///         but the other four are baked into the client's deadlines, its watchdog window, and its
    ///         <c>/sample</c> query string, so keeping the old client because the URI happened not to
    ///         change would silently discard the operator's edit — the same class of defect as
    ///         ignoring the config document altogether.
    ///     </para>
    ///     <para>
    ///         <b>Otherwise</b> ("config-only": authored tags, probe cadence, reconnect backoff) ⇒
    ///         the live client and its connection pool are kept, and everything derived from config
    ///         is rebuilt: the probe model is re-fetched, the observation index is recreated against
    ///         the new tag types, and one <c>/current</c> re-primes it. Nothing config-derived
    ///         survives, so a retyped tag cannot keep coercing to its old type.
    ///     </para>
    /// </remarks>
    public async Task ReinitializeAsync(string driverConfigJson, CancellationToken cancellationToken)
    {
        await _lifecycle.WaitAsync(cancellationToken).ConfigureAwait(false);
        try
        {
            var incoming = ResolveIncomingOptions(driverConfigJson);

            var existing = Volatile.Read(ref _session)?.Client;

            await StopSampleStreamAsync(cancellationToken).ConfigureAwait(false);
            await StopProbeLoopAsync(cancellationToken).ConfigureAwait(false);

            if (existing is null || RequiresClientRebuild(_options, incoming))
            {
                await TeardownCoreAsync().ConfigureAwait(false);
                await StartCoreAsync(incoming, existingClient: null, cancellationToken).ConfigureAwait(false);

                return;
            }

            await StartCoreAsync(incoming, existing, cancellationToken).ConfigureAwait(false);
        }
        finally
        {
            _lifecycle.Release();
        }
    }

    /// <inheritdoc/>
    /// <remarks>
    ///     Stops the <c>/sample</c> stream, disposes the Agent client, and drops the probe cache and
    ///     the observation index — values sourced from a torn-down connection must never keep
    ///     reading Good. <c>LastSuccessfulRead</c> is retained because it is diagnostic ("when did
    ///     this driver last hear from the Agent"), not served data. Idempotent.
    /// </remarks>
    public async Task ShutdownAsync(CancellationToken cancellationToken)
    {
        await _lifecycle.WaitAsync(cancellationToken).ConfigureAwait(false);
        try
        {
            var lastRead = ReadHealth().LastSuccessfulRead;

            await StopSampleStreamAsync(cancellationToken).ConfigureAwait(false);
            await StopProbeLoopAsync(cancellationToken).ConfigureAwait(false);
            await TeardownCoreAsync().ConfigureAwait(false);

            WriteHealth(new DriverHealth(DriverState.Unknown, lastRead, null));
        }
        finally
        {
            _lifecycle.Release();
        }
    }

    /// <inheritdoc/>
    public DriverHealth GetHealth() => ReadHealth();

    /// <inheritdoc/>
    /// <remarks>
    ///     Sums the three caches that scale with plant size: the <c>/probe</c> device model (by
    ///     declared data item), the live observation index (by indexed data item), and the authored
    ///     tag table. See <see cref="ProbeModelBytesPerDataItem"/> for why the weights are coarse
    ///     constants.
    /// </remarks>
    public long GetMemoryFootprint() =>
        ((long)(Volatile.Read(ref _session)?.ProbeModelDataItemCount ?? 0) * ProbeModelBytesPerDataItem)
        + ((long)ObservationIndex.Count * IndexBytesPerEntry)
        + ((long)_options.Tags.Count * TagBytesPerEntry);

    /// <inheritdoc/>
    /// <remarks>
    ///     Drops the <c>/probe</c> device model — the browse cache, which is re-derivable from the
    ///     Agent at any time — and keeps the observation index, which is required for correctness
    ///     (it is the read surface, and re-deriving it would mean a full re-prime the caller did not
    ///     ask for). Flushing cannot wedge the driver: every consumer of the model goes through
    ///     <see cref="GetOrFetchProbeModelAsync"/>, which re-fetches and re-caches on a miss.
    ///     <para>
    ///         The drop is a single <see cref="Interlocked.CompareExchange{T}"/> of a whole new
    ///         <see cref="AgentSession"/>, so the model and its data-item count can never be seen
    ///         out of step (which would skew <see cref="GetMemoryFootprint"/>), and a flush that
    ///         races a concurrent teardown or re-initialize is discarded rather than resurrecting
    ///         the session it was reading.
    ///     </para>
    /// </remarks>
    public Task FlushOptionalCachesAsync(CancellationToken cancellationToken)
    {
        var session = Volatile.Read(ref _session);
        if (session?.ProbeModel is null)
        {
            return Task.CompletedTask;
        }

        var flushed = session with { ProbeModel = null, ProbeModelDataItemCount = 0 };
        if (!ReferenceEquals(Interlocked.CompareExchange(ref _session, flushed, session), session))
        {
            // Someone swapped the session while we were building the flushed copy — their state is
            // newer than ours, and publishing would undo it. Whatever they installed is either a
            // teardown (no cache to flush) or a fresh initialize (a cache the operator's budget
            // breach never saw), so dropping this flush is correct, not merely safe.
            return Task.CompletedTask;
        }

        _logger.LogInformation(
            "MTConnect driver {DriverInstanceId} flushed its cached /probe device model ({DataItemCount} data items); it is re-fetched on the next browse.",
            _driverInstanceId, session.ProbeModelDataItemCount);

        return Task.CompletedTask;
    }

    /// <inheritdoc/>
    /// <remarks>
    ///     <para>
    ///         <b>One <c>/current</c> per batch, whatever the batch size.</b> MTConnect's
    ///         <c>/current</c> answers for the whole device (or the whole Agent), so N references
    ///         cost one deadline-bounded round-trip and are answered out of that one document. There
    ///         is no per-reference request to make and none is made.
    ///     </para>
    ///     <para>
    ///         <b>The response is indexed into a per-call snapshot, NOT into the driver's shared
    ///         observation index.</b> That index is written by the <c>/sample</c> pump (Task 11) and
    ///         is deliberately last-write-wins with no timestamp arbitration. A read that folded its
    ///         <c>/current</c> into it would be a second writer racing the first, and the
    ///         <c>/current</c> document is frequently the <i>older</i> of the two — the Agent
    ///         composes it while the pump's newer delta is already in flight. Losing that race rolls
    ///         a subscribed value backwards and <b>leaves it there until the data item next
    ///         changes</b>, which for an EVENT such as <c>Execution</c> can be hours. Keeping the
    ///         read path a pure function of (document, authored tags) costs one index construction
    ///         per batch — negligible beside the HTTP round-trip that precedes it — and makes it
    ///         impossible for a read to corrupt subscription state. The reference source-of-truth is
    ///         unchanged either way: both paths coerce through the same
    ///         <see cref="MTConnectObservationIndex"/> logic, so a read and a subscription report a
    ///         given observation identically.
    ///     </para>
    ///     <para>
    ///         <b>Nothing but genuine caller cancellation crosses this boundary.</b> An unreachable
    ///         Agent, a blown deadline, an unusable answer — each fails the batch's <i>values</i>
    ///         (every reference codes <see cref="BadCommunicationError"/>) rather than the batch: an
    ///         exception here would also fail the references that were perfectly readable, and the
    ///         node-manager cannot tell a thrown read from a broken driver. This is the opposite
    ///         posture to <see cref="InitializeAsync"/>, which is specified to throw so the
    ///         runtime's retry/backoff loop can own recovery.
    ///     </para>
    ///     <para>
    ///         The status-code distinctions the index draws — <c>BadWaitingForInitialData</c> for an
    ///         authored-but-unreported id, <c>BadNodeIdUnknown</c> for an unknown one,
    ///         <c>BadNoCommunication</c> for the Agent's <c>UNAVAILABLE</c> — are passed through
    ///         unmodified. They tell an operator three different things.
    ///     </para>
    /// </remarks>
    /// <exception cref="ArgumentNullException">
    ///     <paramref name="fullReferences"/> is <c>null</c> — a caller bug, not device data, and the
    ///     one thing this method is loud about (matching <see cref="MTConnectObservationIndex"/>'s
    ///     documented posture).
    /// </exception>
    public async Task<IReadOnlyList<DataValueSnapshot>> ReadAsync(
        IReadOnlyList<string> fullReferences, CancellationToken cancellationToken)
    {
        ArgumentNullException.ThrowIfNull(fullReferences);

        // Checked before anything else: an empty batch must cost the Agent nothing, and there is no
        // failure it could report.
        if (fullReferences.Count == 0)
        {
            return [];
        }

        // One read of the session reference: the client used below is the one that was live at this
        // instant, never a half-swapped pairing (see the _session remarks).
        var client = Volatile.Read(ref _session)?.Client;
        var options = _options;

        if (client is null)
        {
            // Before InitializeAsync, after a faulted one, or after ShutdownAsync. Health is left
            // exactly as it is: a read attempted against a driver that was never started (or was
            // deliberately stopped) is not evidence of degradation.
            return BadBatch(fullReferences.Count, BadNotConnected);
        }

        try
        {
            var current = await BoundedAsync(client.CurrentAsync, "/current", options.RequestTimeoutMs, cancellationToken)
                .ConfigureAwait(false);

            // Built from the SAME authored tags as the shared index, so the authored-vs-unknown
            // distinction and every coercion behave identically — it is a snapshot, not a variant.
            var snapshot = new MTConnectObservationIndex(options.Tags);
            snapshot.Apply(current);

            var results = new DataValueSnapshot[fullReferences.Count];
            for (var i = 0; i < results.Length; i++)
            {
                // Get never throws and never returns null, including for a null/blank reference.
                results[i] = snapshot.Get(fullReferences[i]);
            }

            _readDegraded = false;
            PublishHealthy(DateTime.UtcNow);

            return results;
        }
        catch (OperationCanceledException) when (cancellationToken.IsCancellationRequested)
        {
            // The caller asked us to stop. This is the ONLY exception that may propagate: a blown
            // per-call deadline arrives as the TimeoutException BoundedAsync raises instead, and is
            // handled below as the Agent failure it is.
            throw;
        }
        catch (Exception ex)
        {
            DegradeAfterReadFailure(ex);

            return BadBatch(fullReferences.Count, BadCommunicationError);
        }
    }

    // ---- ISubscribable: ONE shared /sample long poll behind every handle ----

    /// <inheritdoc/>
    public event EventHandler<DataChangeEventArgs>? OnDataChange;

    /// <inheritdoc/>
    /// <remarks>
    ///     <para>
    ///         <b>Initial data comes from the primed <c>/current</c>, not from the wire.</b> Every
    ///         subscribed reference gets one callback immediately (the OPC UA Part 4 convention),
    ///         answered out of the observation index — including the references with no value yet,
    ///         which report <c>BadWaitingForInitialData</c>. Waiting for the Agent's next chunk
    ///         instead would leave a fresh subscription silent for up to a whole <c>heartbeat</c>,
    ///         and permanently silent for any data item that simply is not changing.
    ///     </para>
    ///     <para>
    ///         <b>Subscribe does not wait for the stream.</b> The first subscription starts the
    ///         shared pump; the pump then opens <c>/sample</c> on its own task under its own
    ///         cancellation token. This is deliberate on both counts. The opening handshake cannot
    ///         be awaited inside the caller's Subscribe timeout in any useful sense — the first
    ///         chunk may legitimately be a heartbeat away — and the long-lived poll must NOT run
    ///         under the caller's token, which belongs to one Subscribe call and is disposed the
    ///         moment it returns. What bounds the running stream is the client's own heartbeat
    ///         watchdog, and what stops it is <see cref="UnsubscribeAsync"/> or the lifecycle.
    ///     </para>
    ///     <para>
    ///         <b><paramref name="publishingInterval"/> is not honoured per subscription</b>, and
    ///         cannot be: MTConnect's publish cadence is the Agent-side <c>interval</c> query
    ///         parameter of the single shared <c>/sample</c> request
    ///         (<see cref="MTConnectDriverOptions.SampleIntervalMs"/>), fixed when the client is
    ///         built. Honouring a per-handle interval would mean one stream per subscription — the
    ///         Agent load this driver exists to avoid — or client-side throttling that would drop
    ///         transient EVENT values the Agent bothered to send. The parameter is accepted and
    ///         ignored, which is also what the natively-subscribing sibling drivers do.
    ///     </para>
    ///     <para>
    ///         Subscribing before <see cref="InitializeAsync"/> is legal: the interest is recorded,
    ///         every reference reports "no value yet", and the initialize that follows starts the
    ///         pump for it. It does not throw and it does not dial an Agent that is not there yet.
    ///     </para>
    /// </remarks>
    /// <exception cref="ArgumentNullException">
    ///     <paramref name="fullReferences"/> is <c>null</c> — a caller bug, and the only thing this
    ///     method is loud about, matching <see cref="ReadAsync"/>'s posture.
    /// </exception>
    public Task<ISubscriptionHandle> SubscribeAsync(
        IReadOnlyList<string> fullReferences, TimeSpan publishingInterval, CancellationToken cancellationToken)
    {
        ArgumentNullException.ThrowIfNull(fullReferences);

        // De-duplicated: a reference asked for twice in one subscription is one subscribed value,
        // and would otherwise raise two identical callbacks per chunk forever.
        var references = new HashSet<string>(StringComparer.Ordinal);
        foreach (var reference in fullReferences)
        {
            if (reference is not null)
            {
                references.Add(reference);
            }
        }

        var handle = new MTConnectSampleHandle(Interlocked.Increment(ref _nextSubscriptionId));

        lock (_subscriptionLock)
        {
            _subscriptions[handle.SubscriptionId] = new SubscriptionState(handle, references);

            // No-op when a pump is already running: every handle shares the one stream.
            StartSampleStreamCore(republishOnStart: false);
        }

        // Raised OUTSIDE the lock — a handler is caller code (see the _subscriptionLock remarks).
        var index = ObservationIndex;
        foreach (var reference in references)
        {
            RaiseDataChange(handle, reference, index.Get(reference));
        }

        _logger.LogDebug(
            "MTConnect driver {DriverInstanceId} opened subscription {SubscriptionId} over {ReferenceCount} reference(s).",
            _driverInstanceId, handle.DiagnosticId, references.Count);

        return Task.FromResult<ISubscriptionHandle>(handle);
    }

    /// <inheritdoc/>
    /// <remarks>
    ///     Drops the handle's references and stops the shared stream when the last subscription
    ///     goes. An unknown handle — already unsubscribed, or issued by a different driver — is a
    ///     no-op rather than an error: this runs on teardown paths, including failure paths, where
    ///     throwing would mask whatever prompted the teardown.
    /// </remarks>
    /// <exception cref="ArgumentNullException"><paramref name="handle"/> is <c>null</c>.</exception>
    public async Task UnsubscribeAsync(ISubscriptionHandle handle, CancellationToken cancellationToken)
    {
        ArgumentNullException.ThrowIfNull(handle);

        if (handle is not MTConnectSampleHandle ours)
        {
            return;
        }

        bool stopStream;
        lock (_subscriptionLock)
        {
            if (!_subscriptions.Remove(ours.SubscriptionId))
            {
                return;
            }

            stopStream = !HasSubscribedReferencesCore();
        }

        if (stopStream)
        {
            // Outside the lock: StopSampleStreamAsync awaits the pump, and the pump takes this same
            // lock to read the subscription registry. Holding it here would deadlock the two.
            // The caller's token bounds the wait: DriverInstanceActor puts a 5 s budget around this
            // call, and until now that budget was inert.
            await StopSampleStreamAsync(cancellationToken).ConfigureAwait(false);
        }

        _logger.LogDebug(
            "MTConnect driver {DriverInstanceId} closed subscription {SubscriptionId}{StreamNote}.",
            _driverInstanceId, ours.DiagnosticId, stopStream ? " and stopped the shared /sample stream" : string.Empty);
    }

    // ---- ITagDiscovery: the /probe device model, streamed as folders + variables ----

    /// <inheritdoc/>
    /// <remarks>
    ///     <b>This one member is the whole browse opt-in.</b> The Wave-0 universal
    ///     <c>DiscoveryDriverBrowser</c> turns any driver that answers <c>true</c> here into a live
    ///     address picker by capturing <see cref="DiscoverAsync"/> — so this driver ships no browser
    ///     code, no per-driver picker component, and no entry in the browser's browse-patch table
    ///     (its discovery is unconditional; there is no "enable browse" option to turn on first).
    ///     It is answered on an <b>un-initialized</b> instance without dialling anything, because
    ///     <c>CanBrowse</c> constructs a throwaway driver purely to read it — which is also why the
    ///     constructor opens nothing.
    /// </remarks>
    public bool SupportsOnlineDiscovery => true;

    /// <inheritdoc/>
    /// <remarks>
    ///     <c>Once</c>, not the default <c>UntilStable</c>: <see cref="DiscoverAsync"/> streams the
    ///     complete device model within the single call — <c>/probe</c> is one synchronous request
    ///     that either answers with the whole hierarchy or fails — so nothing fills in asynchronously
    ///     after connect and a settle loop would only re-fetch an identical answer. (The one thing
    ///     that <i>does</i> change the model is an Agent restart, and that is
    ///     <c>IRediscoverable</c>'s job in Task 13, not a retry cadence's.)
    /// </remarks>
    public DiscoveryRediscoverPolicy RediscoverPolicy => DiscoveryRediscoverPolicy.Once;

    /// <inheritdoc/>
    /// <remarks>
    ///     <para>
    ///         Streams the Agent's <c>/probe</c> hierarchy verbatim: one folder per <c>Device</c>,
    ///         one per nested <c>Component</c> to arbitrary depth, and one variable per
    ///         <c>DataItem</c> under whichever container declares it — <b>including the data items a
    ///         device declares directly</b>, outside any component (an availability item almost
    ///         always lives there, and a walker that only recursed <c>Components</c> would silently
    ///         drop it).
    ///     </para>
    ///     <para>
    ///         <b><see cref="DriverAttributeInfo.FullName"/> is the <c>DataItem@id</c>, never the
    ///         browse name.</b> The universal browser commits a browsed leaf's <c>FullName</c>
    ///         straight into <c>TagConfig.FullName</c>, and that is the key
    ///         <see cref="ReadAsync"/> and the <c>/sample</c> pump resolve an observation against
    ///         (<see cref="MTConnectObservationIndex"/> is keyed by <c>DataItem@id</c>). Committing
    ///         the <c>name</c> instead would produce a picker that looks right and authors tags that
    ///         can never report a value. The browse <i>name</i> — cosmetic — prefers the DataItem's
    ///         <c>name</c> and falls back to its <c>id</c>, which for this protocol is the common
    ///         case rather than an edge case.
    ///     </para>
    ///     <para>
    ///         <b>The type shape comes from <see cref="MTConnectDataTypeInference.Infer"/>'s returned
    ///         record, in full</b> — <see cref="MTConnectInferredType.IsArray"/> and
    ///         <see cref="MTConnectInferredType.ArrayDim"/> included. Recomputing the array shape
    ///         locally from <c>representation</c>/<c>sampleCount</c> is the tempting shortcut and it
    ///         is wrong: the inference also demotes <c>DATA_SET</c>/<c>TABLE</c> to
    ///         <see cref="DriverDataType.String"/> even on a <c>SAMPLE</c>, and honours
    ///         <c>TIME_SERIES</c> only on a <c>SAMPLE</c>. The AdminUI typed editor calls the same
    ///         function, so any divergence here makes the picker and the editor disagree about one
    ///         data item. The <c>type</c> attribute is passed <b>verbatim</b> for the same reason:
    ///         the probe document spells it UPPER_SNAKE (<c>PART_COUNT</c>) where the streams
    ///         document uses PascalCase, and the inference is separator-insensitive precisely to
    ///         bridge the two — pre-normalising it here would defeat that.
    ///     </para>
    ///     <para>
    ///         <b><see cref="DriverAttributeInfo.IsAlarm"/> is this seam's job</b>, deliberately not
    ///         the inference's: a <c>CONDITION</c> data item is an alarm.
    ///         <c>IVariableHandle.MarkAsAlarmCondition</c> is <i>not</i> called — that annotation
    ///         needs the driver-side refs of an alarm's sibling attributes (in-alarm flag, priority,
    ///         ack target), and an MTConnect condition has none: it is one text observation whose
    ///         value is the state word. Core's generic node manager enriches on the flag alone.
    ///     </para>
    ///     <para>
    ///         <b>The model comes from <see cref="GetOrFetchProbeModelAsync"/>, never from
    ///         <see cref="CachedProbeModel"/>.</b> <see cref="FlushOptionalCachesAsync"/> may have
    ///         dropped the cache under memory pressure; reading the field directly would make a
    ///         flushed driver browse as an Agent with no data items, permanently.
    ///     </para>
    ///     <para>
    ///         <b>This throws, and that is the opposite of <see cref="ReadAsync"/>'s posture on
    ///         purpose.</b> A read has a per-reference Bad status code to report a failure in-band; a
    ///         discovery stream has no such channel, so the only way to "fail quietly" would be to
    ///         emit an empty tree — indistinguishable from an Agent that genuinely declares nothing,
    ///         and read by an operator in the picker as a working connection to an empty machine
    ///         (#485: empty is not an answer).
    ///     </para>
    ///     <para>
    ///         <b>What the callers actually do with the throw</b> — worth stating precisely, because
    ///         the reassuring version ("it is logged and retried") is not true here. The live
    ///         consumer is the <c>/raw</c> browse-commit path
    ///         (<c>DiscoveryDriverBrowser</c>/<c>BrowserSessionService</c>), which surfaces it to the
    ///         operator as a <i>failed browse</i> — that is the case this posture is chosen for.
    ///         <c>DriverInstanceActor.HandleRediscoverAsync</c> catches it, logs a Warning, and
    ///         publishes an EMPTY node set; it does <b>not</b> reschedule, because retrying is a
    ///         <see cref="DiscoveryRediscoverPolicy.UntilStable"/> behaviour and this driver is
    ///         <see cref="DiscoveryRediscoverPolicy.Once"/>. And that publish currently reaches
    ///         nothing: <c>DriverHostActor.HandleDiscoveredNodes</c> short-circuits unconditionally
    ///         (discovered-node injection is dormant in v3 — raw tags are authored through
    ///         browse-commit instead), so at that seam the throw-versus-empty choice has no runtime
    ///         effect today. It is made for the browse path, and for whatever re-enables injection.
    ///     </para>
    ///     <para>
    ///         <b>Nothing here writes the observation index.</b> Discovery is a pure read of the
    ///         device model — the <c>/sample</c> pump remains its sole writer.
    ///     </para>
    /// </remarks>
    /// <exception cref="ArgumentNullException"><paramref name="builder"/> is <c>null</c>.</exception>
    /// <exception cref="InvalidOperationException">
    ///     The driver is not initialized, or it was torn down while the <c>/probe</c> fetch was in
    ///     flight (see <see cref="GetOrFetchProbeModelAsync"/>).
    /// </exception>
    public async Task DiscoverAsync(IAddressSpaceBuilder builder, CancellationToken cancellationToken)
    {
        ArgumentNullException.ThrowIfNull(builder);

        var model = await GetOrFetchProbeModelAsync(cancellationToken).ConfigureAwait(false);

        // Read once: a concurrent re-initialize must not scope half this walk to one device and half
        // to another.
        var deviceScope = _options.DeviceName?.Trim();

        var devices = 0;
        var variables = 0;

        foreach (var device in model.Devices)
        {
            if (device is null || !InDeviceScope(device, deviceScope))
            {
                continue;
            }

            var name = FolderName(device.Name, device.Id);
            if (string.IsNullOrWhiteSpace(name))
            {
                // Neither a name nor an id: there is no browse path that could identify this device,
                // and folding it in would emit a blank tree segment (and a blank NodeId component
                // downstream). Its data items are unreachable either way.
                _logger.LogWarning(
                    "MTConnect driver {DriverInstanceId} skipped a device in {AgentUri}'s /probe model that declares neither a name nor an id; it cannot be given a browse path.",
                    _driverInstanceId, _options.AgentUri);

                continue;
            }

            devices++;
            variables += StreamContainer(device, builder.Folder(name, name));
        }

        // A configured scope that matches NO device is an authoring error — almost always a typo or
        // a device renamed in the Agent — and its only symptom is a browse that succeeds and shows
        // nothing. Reported at Warning, because at Debug the operator's evidence is an empty picker
        // and no explanation. An agent-wide browse that finds nothing is a different statement (the
        // Agent declares no devices) and stays a Debug tally.
        if (devices == 0 && !string.IsNullOrEmpty(deviceScope))
        {
            _logger.LogWarning(
                "MTConnect driver {DriverInstanceId} is scoped to device '{DeviceScope}', but {AgentUri}'s /probe model declares no device with that name or id ({DeclaredCount} declared). The browse is empty because the scope matched nothing, not because the Agent has nothing.",
                _driverInstanceId, deviceScope, _options.AgentUri, model.Devices.Count);

            return;
        }

        _logger.LogDebug(
            "MTConnect driver {DriverInstanceId} streamed {DeviceCount} device(s) and {VariableCount} data item(s) from {AgentUri}'s /probe model{DeviceScope}.",
            _driverInstanceId,
            devices,
            variables,
            _options.AgentUri,
            deviceScope is null or "" ? string.Empty : $" (scoped to '{deviceScope}')");
    }

    /// <summary>
    ///     Streams one device/component's own data items into <paramref name="scope"/>, then recurses
    ///     into each nested component under a folder of its own. Returns how many variables it
    ///     streamed, for the diagnostic tally.
    /// </summary>
    /// <remarks>
    ///     Walks the tree rather than using <c>AllDataItems()</c>: that helper flattens, and the
    ///     folder structure — which item belongs to which component — is precisely what a browse
    ///     picker exists to show. Recursion depth is bounded by the parser, which refuses a probe
    ///     document nested beyond its own limit.
    /// </remarks>
    private static int StreamContainer(IMTConnectComponentContainer container, IAddressSpaceBuilder scope)
    {
        var streamed = 0;

        foreach (var dataItem in container.DataItems)
        {
            // An id-less data item cannot be read, subscribed, or committed as a tag — there would be
            // nothing to key it by. The parser already requires the attribute; this is the guard for a
            // model built any other way.
            if (dataItem is null || string.IsNullOrWhiteSpace(dataItem.Id))
            {
                continue;
            }

            // Every field the DataItem declares, straight through — the array shape is READ OFF the
            // result, never recomputed here. See the DiscoverAsync remarks.
            var inferred = MTConnectDataTypeInference.Infer(
                dataItem.Category,
                dataItem.Type,
                dataItem.Units,
                dataItem.Representation,
                dataItem.SampleCount);

            var browseName = string.IsNullOrWhiteSpace(dataItem.Name) ? dataItem.Id : dataItem.Name;

            scope.Variable(
                browseName,
                browseName,
                new DriverAttributeInfo(
                    FullName: dataItem.Id,
                    DriverDataType: inferred.DataType,
                    IsArray: inferred.IsArray,
                    ArrayDim: inferred.ArrayDim,

                    // MTConnect is read-only and this driver implements no IWritable, so the tier
                    // that is read-only from OPC UA is the only honest one.
                    SecurityClass: SecurityClassification.ViewOnly,

                    // Historization is authored per tag, never inferred from a device model.
                    IsHistorized: false,
                    IsAlarm: IsCondition(dataItem.Category)));

            streamed++;
        }

        foreach (var component in container.Components)
        {
            if (component is null)
            {
                continue;
            }

            // Guarded on the RESOLVED folder name, not on the id alone: a component with a blank id
            // but a real name browses perfectly well (the id is only the fallback), while one with
            // neither would open a folder named "" — a blank segment in every browse path beneath
            // it. The data-item guard above is stricter for a different reason: THAT id is the
            // correlation key an observation is resolved by, so a blank one is unreadable whatever
            // it is called.
            var name = FolderName(component.Name, component.Id);
            if (string.IsNullOrWhiteSpace(name))
            {
                continue;
            }

            streamed += StreamContainer(component, scope.Folder(name, name));
        }

        return streamed;
    }

    /// <summary>
    ///     Whether a device is in this instance's configured scope. An unset scope is agent-wide.
    /// </summary>
    /// <remarks>
    ///     <para>
    ///         Matched against the device's <c>name</c> <b>or</b> its <c>id</c> — a device model may
    ///         omit the name, and its id is then what an operator would put in the config and what a
    ///         scoped Agent URL would carry.
    ///     </para>
    ///     <para>
    ///         <b>Belt-and-braces, not dead code.</b> The production client already narrows the
    ///         request to <c>{AgentUri}/{DeviceName}/probe</c>, so a conforming Agent answers with the
    ///         one device anyway. But an Agent (or an intermediary) that ignored the path segment and
    ///         answered agent-wide would otherwise publish every other machine's data items into an
    ///         instance scoped to one, and the operator's only evidence would be a picker full of ids
    ///         they never configured. Matching is <b>exact</b> for the same reason the URL is: a
    ///         differently-cased device name is not the device the Agent would have served.
    ///     </para>
    /// </remarks>
    private static bool InDeviceScope(MTConnectDevice device, string? deviceScope) =>
        string.IsNullOrEmpty(deviceScope)
        || string.Equals(device.Name, deviceScope, StringComparison.Ordinal)
        || string.Equals(device.Id, deviceScope, StringComparison.Ordinal);

    /// <summary>The folder name for a device/component: its <c>name</c>, or its <c>id</c> when it declares none.</summary>
    private static string FolderName(string? name, string id) => string.IsNullOrWhiteSpace(name) ? id : name;

    /// <summary>
    ///     Whether a data item's category makes it an alarm. Lives here rather than in
    ///     <see cref="MTConnectDataTypeInference"/> because that function answers a <i>type</i>
    ///     question; this is an address-space one.
    /// </summary>
    private static bool IsCondition(string? category) =>
        category.AsSpan().Trim().Equals("CONDITION", StringComparison.OrdinalIgnoreCase);

    // ---- IHostConnectivityProbe: one host per Agent, on its own cheap tick ----

    /// <inheritdoc/>
    /// <remarks>
    ///     <b>Raised from the probe loop's task, never under a lock and never under
    ///     <see cref="_lifecycle"/>.</b> A handler is arbitrary caller code: one that blocks forever
    ///     blocks the loop with it (and therefore the teardown that awaits it), exactly as an
    ///     <see cref="OnDataChange"/> handler does to the pump. One that <i>throws</i> is absorbed.
    /// </remarks>
    public event EventHandler<HostStatusChangedEventArgs>? OnHostStatusChanged;

    /// <summary>
    ///     The single host this driver instance talks to, named by its configured
    ///     <see cref="MTConnectDriverOptions.AgentUri"/>. Mirrors <c>ModbusDriver.HostName</c>'s
    ///     <c>host:port</c> convention: several MTConnect drivers in one server disambiguate on the
    ///     Admin dashboard by endpoint rather than by driver-instance id.
    /// </summary>
    public string HostName => _options.AgentUri;

    /// <inheritdoc/>
    /// <remarks>
    ///     One entry, always — an Agent is a single host, and the driver knows of it whether or not
    ///     it has been reached. Before the first probe tick (and for the whole life of an instance
    ///     with <c>Probe.Enabled = false</c>) that entry reads <see cref="HostState.Unknown"/>, which
    ///     is the truth: nothing has measured it. See <see cref="_hostState"/> for why a successful
    ///     initialize does not seed it <see cref="HostState.Running"/>.
    /// </remarks>
    public IReadOnlyList<HostConnectivityStatus> GetHostStatuses()
    {
        lock (_probeLock)
        {
            return [new HostConnectivityStatus(HostName, _hostState, _hostStateChangedUtc)];
        }
    }

    /// <summary>
    ///     Starts the periodic connectivity probe, if the operator asked for one. <b>Caller must hold
    ///     <see cref="_lifecycle"/>.</b>
    /// </summary>
    /// <param name="client">The client this session owns; the loop dials that one, never a later one.</param>
    /// <param name="options">The options this session started with.</param>
    private void StartProbeLoopCore(IMTConnectAgentClient client, MTConnectDriverOptions options)
    {
        if (!options.Probe.Enabled)
        {
            return;
        }

        var cts = new CancellationTokenSource();
        _probeCts = cts;
        Volatile.Write(ref _probeTask, Task.Run(() => RunProbeLoopAsync(client, options, cts.Token), CancellationToken.None));
    }

    /// <summary>
    ///     The connectivity probe: wait, ask the Agent whether it is answering, publish the verdict,
    ///     repeat. Never throws — it is a detached background task, and an escaping exception would
    ///     silently end host reporting while <see cref="GetHostStatuses"/> kept serving the last
    ///     thing it saw.
    /// </summary>
    /// <remarks>
    ///     <para>
    ///         <b>It calls <see cref="IMTConnectAgentClient.ProbeAsync"/> directly and publishes
    ///         nothing.</b> Both halves of that are deliberate.
    ///     </para>
    ///     <para>
    ///         Routing it through <see cref="GetOrFetchProbeModelAsync"/> is the obvious shortcut and
    ///         it is <b>inert</b>: that accessor answers from the cache <see cref="InitializeAsync"/>
    ///         already warmed, so the loop would issue no request at all and report
    ///         <see cref="HostState.Running"/> forever no matter what the Agent was doing — a feature
    ///         that ticks, logs, and measures nothing.
    ///     </para>
    ///     <para>
    ///         Publishing the answer into <see cref="AgentSession.ProbeModel"/> is the opposite
    ///         mistake: it would put a second writer on a field the lifecycle owns under a
    ///         compare-and-swap, and would silently re-warm a cache that
    ///         <see cref="FlushOptionalCachesAsync"/> — or an Agent restart, see
    ///         <see cref="AnnounceAgentRestart"/> — deliberately dropped. The parsed model is
    ///         discarded; only the fact that a request completed is used.
    ///     </para>
    ///     <para>
    ///         <b>Why <c>/probe</c> rather than the <c>/sample</c> heartbeat.</b> The heartbeat is
    ///         free, but it only exists while something is subscribed — an instance with no
    ///         subscriptions would report <see cref="HostState.Unknown"/> indefinitely, and host
    ///         reachability would become a function of OPC UA client behaviour. A dedicated tick is
    ///         the one signal that is true whether or not anyone is watching. Its cost is a parse of
    ///         the device model per interval, which is why <see cref="MTConnectProbeOptions.Interval"/>
    ///         exists.
    ///     </para>
    ///     <para>
    ///         <b>It waits first, then probes.</b> <see cref="StartCoreAsync"/> has just completed a
    ///         successful <c>/probe</c> one moment earlier; a tick that fired immediately would spend
    ///         a second identical request to learn nothing, and would do it on every re-initialize.
    ///     </para>
    ///     <para>
    ///         <b>It never calls a lifecycle method</b> — see the <see cref="_lifecycle"/> remarks.
    ///         <see cref="StopProbeLoopAsync"/> awaits this task while holding that semaphore, so
    ///         reaching back into one would be a permanent hang with no exception and no log line.
    ///     </para>
    /// </remarks>
    private async Task RunProbeLoopAsync(
        IMTConnectAgentClient client, MTConnectDriverOptions options, CancellationToken ct)
    {
        try
        {
            while (!ct.IsCancellationRequested)
            {
                try
                {
                    await _probeScheduler(options.Probe.Interval, ct).ConfigureAwait(false);
                }
                catch (OperationCanceledException)
                {
                    return;
                }

                if (ct.IsCancellationRequested)
                {
                    return;
                }

                bool reachable;
                try
                {
                    using var deadline = CancellationTokenSource.CreateLinkedTokenSource(ct);
                    deadline.CancelAfter(options.Probe.Timeout);

                    // The answer is deliberately discarded — see the remarks.
                    _ = await client.ProbeAsync(deadline.Token).ConfigureAwait(false);
                    reachable = true;
                }
                catch (OperationCanceledException) when (ct.IsCancellationRequested)
                {
                    return;
                }
                catch (Exception ex)
                {
                    // Unreachable, unusable, disposed, or past its deadline — from the host's point
                    // of view these are one fact, and the data paths report the detail.
                    reachable = false;

                    _logger.LogDebug(
                        ex,
                        "MTConnect driver {DriverInstanceId} could not reach {AgentUri} on its connectivity probe.",
                        _driverInstanceId, options.AgentUri);
                }

                TransitionHostTo(reachable ? HostState.Running : HostState.Stopped);
            }
        }
        catch (Exception ex)
        {
            _logger.LogError(
                ex,
                "MTConnect driver {DriverInstanceId} stopped its connectivity probe on an unexpected fault; host status for {AgentUri} will not be updated until the driver is re-initialized.",
                _driverInstanceId, options.AgentUri);
        }
    }

    /// <summary>
    ///     Publishes a host state and announces it — <b>only when it actually changed</b>. Raising on
    ///     every tick would emit one event per interval per Agent forever: Core would re-fan Bad
    ///     quality across the host's subtree each time, and an operator watching the feed could not
    ///     tell a new outage from an ongoing one.
    /// </summary>
    private void TransitionHostTo(HostState newState)
    {
        HostState old;
        lock (_probeLock)
        {
            old = _hostState;
            if (old == newState)
            {
                return;
            }

            _hostState = newState;
            _hostStateChangedUtc = DateTime.UtcNow;
        }

        _logger.LogInformation(
            "MTConnect driver {DriverInstanceId} host {AgentUri} went {Previous} -> {Current}.",
            _driverInstanceId, HostName, old, newState);

        // Outside the lock: a handler is caller code.
        try
        {
            OnHostStatusChanged?.Invoke(this, new HostStatusChangedEventArgs(HostName, old, newState));
        }
        catch (Exception ex)
        {
            _logger.LogWarning(
                ex,
                "MTConnect driver {DriverInstanceId} caught a fault from a host-status subscriber for {AgentUri}; the probe continues.",
                _driverInstanceId, HostName);
        }
    }

    /// <summary>
    ///     Stops the connectivity probe loop and waits for it to be gone. Idempotent, and never
    ///     throws — it runs on teardown paths where a second exception would mask the first.
    /// </summary>
    /// <remarks>
    ///     <b>⚠️ This runs while <see cref="_lifecycle"/> is held</b> and it awaits the loop, on
    ///     exactly the same terms as <see cref="StopSampleStreamAsync"/>: safe only because the loop
    ///     never touches the lifecycle and every await it makes observes the token cancelled below.
    ///     The wait is bounded by <see cref="TeardownWait"/> and the caller's token for the same
    ///     reason: an <see cref="OnHostStatusChanged"/> handler that never returns would otherwise
    ///     wedge the Akka dispatcher thread <c>DriverInstanceActor.PostStop</c> blocks on.
    /// </remarks>
    /// <param name="cancellationToken">The caller's deadline for the teardown wait.</param>
    private async Task StopProbeLoopAsync(CancellationToken cancellationToken = default)
    {
        var cts = _probeCts;
        var task = Volatile.Read(ref _probeTask);
        _probeCts = null;
        Volatile.Write(ref _probeTask, null);

        if (cts is null)
        {
            return;
        }

        try
        {
            await cts.CancelAsync().ConfigureAwait(false);
        }
        catch (ObjectDisposedException)
        {
            // Raced another stop; the loop is already going away.
        }

        var stopped = true;
        if (task is not null)
        {
            try
            {
                await task.WaitAsync(TeardownWait, cancellationToken).ConfigureAwait(false);
            }
            catch (Exception ex) when (ex is TimeoutException or OperationCanceledException)
            {
                stopped = false;

                _logger.LogError(
                    ex,
                    "MTConnect driver {DriverInstanceId} gave up waiting for its connectivity probe to stop after {TeardownWait}; the usual cause is an OnHostStatusChanged subscriber that blocks. Teardown continues.",
                    _driverInstanceId, TeardownWait);
            }
            catch (Exception ex)
            {
                // RunProbeLoopAsync is written not to throw, so this is belt-and-braces.
                _logger.LogDebug(
                    ex,
                    "MTConnect driver {DriverInstanceId} swallowed a fault from its connectivity probe while stopping it.",
                    _driverInstanceId);
            }
        }

        if (stopped)
        {
            // Only once the loop is provably gone — see StopSampleStreamAsync for why.
            cts.Dispose();
        }
    }

    // ---- IRediscoverable: the Agent instanceId watch ----

    /// <inheritdoc/>
    /// <remarks>
    ///     <para>
    ///         <b>This event is what makes <see cref="RediscoverPolicy"/> =
    ///         <see cref="DiscoveryRediscoverPolicy.Once"/> safe.</b> The runtime runs exactly one
    ///         discovery pass per connect, so an Agent that restarts mid-run and renumbers, adds, or
    ///         drops data items would otherwise leave a stale address space sitting behind a
    ///         perfectly Healthy driver.
    ///     </para>
    ///     <para>
    ///         <b>Raised from the <c>/sample</c> pump's task, outside every lock.</b> A handler that
    ///         throws is absorbed and logged — a consumer bug must not tear down the stream serving
    ///         every subscriber. A handler that <i>blocks</i> blocks the pump, and one that
    ///         synchronously awaits a lifecycle method on this driver deadlocks it: teardown waits
    ///         for the pump, and the pump is waiting inside the handler. Core's consumer schedules
    ///         the rediscovery rather than running it inline, which is the only safe shape.
    ///     </para>
    /// </remarks>
    public event EventHandler<RediscoveryEventArgs>? OnRediscoveryNeeded;

    /// <summary>
    ///     Handles an Agent restart observed on the wire: drop the device model it invalidated, then
    ///     tell Core to re-discover — <b>once per distinct <c>instanceId</c></b>.
    /// </summary>
    /// <remarks>
    ///     <para>
    ///         <b>Dropping the cached <c>/probe</c> model is not housekeeping — it is what stops this
    ///         whole capability being inert.</b> Core's response to the event is to re-run
    ///         <see cref="DiscoverAsync"/>, which reads
    ///         <see cref="GetOrFetchProbeModelAsync"/>; with the cache still warm that call answers
    ///         from a model describing a process that no longer exists, and the "rediscovery" would
    ///         diff the stale tree against itself and change nothing. Dropped through the same
    ///         compare-and-swap <see cref="FlushOptionalCachesAsync"/> uses, so a lifecycle change
    ///         landing mid-drop wins rather than being undone.
    ///     </para>
    ///     <para>
    ///         <b>Once per change, not once per chunk.</b> Every chunk after a restart carries the new
    ///         id, and a re-baseline that fails leaves the pump still comparing against the old one —
    ///         so the guard is a latch on the id that was last announced, not on the pump's cursor
    ///         state. A second, different restart is a second announcement; an id the Agent reuses
    ///         after some other id is one too, because that is genuinely a new process.
    ///     </para>
    /// </remarks>
    /// <param name="instanceId">The <c>instanceId</c> the Agent's chunk reported.</param>
    /// <param name="options">The options in force, for the log line's endpoint.</param>
    private void AnnounceAgentRestart(long instanceId, MTConnectDriverOptions options)
    {
        if (Interlocked.Exchange(ref _announcedInstanceId, instanceId) == instanceId)
        {
            return;
        }

        InvalidateProbeModel();

        _logger.LogInformation(
            "MTConnect driver {DriverInstanceId} dropped its cached /probe device model for {AgentUri} and raised rediscovery: agent instanceId is now {InstanceId}.",
            _driverInstanceId, options.AgentUri, instanceId);

        try
        {
            OnRediscoveryNeeded?.Invoke(this, new RediscoveryEventArgs("MTConnect agent instanceId changed", null));
        }
        catch (Exception ex)
        {
            _logger.LogWarning(
                ex,
                "MTConnect driver {DriverInstanceId} caught a fault from a rediscovery subscriber; the /sample stream continues.",
                _driverInstanceId);
        }
    }

    /// <summary>
    ///     Drops the cached <c>/probe</c> device model so the next consumer re-fetches it. Same
    ///     compare-and-swap discipline as <see cref="FlushOptionalCachesAsync"/> — retried, because
    ///     unlike a flush this drop is a correctness requirement rather than a courtesy: losing the
    ///     race must not leave a model from a dead Agent installed.
    /// </summary>
    private void InvalidateProbeModel()
    {
        while (true)
        {
            var session = Volatile.Read(ref _session);
            if (session?.ProbeModel is null)
            {
                return;
            }

            var dropped = session with { ProbeModel = null, ProbeModelDataItemCount = 0 };
            if (ReferenceEquals(Interlocked.CompareExchange(ref _session, dropped, session), session))
            {
                return;
            }
        }
    }

    /// <summary>
    ///     Starts the shared <c>/sample</c> pump, if there is anything to pump and anything to pump
    ///     from. <b>Caller must hold <see cref="_subscriptionLock"/>.</b>
    /// </summary>
    /// <param name="republishOnStart">
    ///     Whether the pump should republish the observation index to every subscriber before it
    ///     opens the stream — true when a lifecycle start replaced the baseline the subscribers'
    ///     current values came from.
    /// </param>
    private void StartSampleStreamCore(bool republishOnStart)
    {
        // `IsCompleted: false`, not `is not null`: a pump that has already exited is not a running
        // stream, and conflating the two means a pump killed by an unexpected fault could never be
        // revived by a resubscribe. The genuinely-unrepeatable case is the latch below, not this.
        if (_pumpTask is { IsCompleted: false })
        {
            return;
        }

        if (_streamUnsupported)
        {
            // Refusing IS correct — reconnecting reproduces the identical answer forever — but it
            // must never be silent. DriverInstanceActor handles a tag-set change as
            // Unsubscribe-then-Subscribe with NO re-initialize, and the unsubscribe stops a stream
            // that is not running; without this, the resubscribe returns a handle, the caller is
            // told SubscriptionEstablished, and the next successful read reports a perfectly green
            // driver over a subscription that can never deliver a value (#485).
            _streamDegraded = true;
            Degrade($"The MTConnect Agent at '{_options.AgentUri}' does not serve a framed /sample stream; subscriptions cannot deliver until the driver is re-initialized.");

            _logger.LogWarning(
                "MTConnect driver {DriverInstanceId} refused to start a /sample stream for {AgentUri}: the endpoint was already found not to stream. The subscription is registered but will receive NO updates until the driver is re-initialized.",
                _driverInstanceId, _options.AgentUri);

            return;
        }

        if (!HasSubscribedReferencesCore())
        {
            return;
        }

        // One read of the session: the client the pump enumerates and the cursor it opens at are
        // guaranteed to belong to the same initialize.
        var session = Volatile.Read(ref _session);
        if (session is null)
        {
            return;
        }

        // A previous pump ran to completion; its cancellation source is ours to release before the
        // field is overwritten, or every revived stream leaks one.
        _pumpCts?.Dispose();
        _pumpTask = null;

        // A new stream generation gets a fresh subscriber-fault Warning budget (see
        // _subscriberFaultLogged).
        Interlocked.Exchange(ref _subscriberFaultLogged, 0);

        var options = _options;
        var cts = new CancellationTokenSource();
        _pumpCts = cts;
        _pumpTask = Task.Run(
            () => RunSampleStreamAsync(
                session.Client, options, session.NextSequence, session.InstanceId, republishOnStart, cts.Token),
            CancellationToken.None);
    }

    /// <summary>
    ///     The shared <c>/sample</c> pump: reads chunks, keeps the observation index and the
    ///     subscribers up to date, and owns every way the stream can stop.
    /// </summary>
    /// <remarks>
    ///     <para>
    ///         <b>Never throws.</b> It is a detached background task; an escaping exception would be
    ///         an unobserved fault that silently ends the driver's subscription surface while every
    ///         handle still looks alive.
    ///     </para>
    ///     <para>
    ///         <b>Never calls a lifecycle method.</b> Its re-baseline goes straight to
    ///         <see cref="IMTConnectAgentClient.CurrentAsync"/> on the client it already holds — see
    ///         <see cref="RebaselineAsync"/> and the <see cref="_lifecycle"/> remarks for why
    ///         "just re-initialize" would be a silent permanent deadlock.
    ///     </para>
    ///     <para>
    ///         <b>Backoff applies to failures only.</b> A re-baseline is a protocol event, not a
    ///         fault, so it reopens the stream immediately and resets the attempt count; only a
    ///         genuine failure (dropped connection, blown heartbeat, a re-baseline that itself
    ///         failed) advances <see cref="BackoffFor"/>.
    ///     </para>
    /// </remarks>
    private async Task RunSampleStreamAsync(
        IMTConnectAgentClient client,
        MTConnectDriverOptions options,
        long from,
        long instanceId,
        bool republishOnStart,
        CancellationToken ct)
    {
        var cursor = from;
        var attempt = 0;
        _reconnectAttempts = 0;

        try
        {
            if (republishOnStart)
            {
                FanOut(SubscribedReferences(), ObservationIndex);
            }

            while (!ct.IsCancellationRequested)
            {
                if (attempt > 0)
                {
                    var delay = BackoffFor(attempt, options.Reconnect);
                    if (delay > TimeSpan.Zero)
                    {
                        try
                        {
                            await Task.Delay(delay, ct).ConfigureAwait(false);
                        }
                        catch (OperationCanceledException)
                        {
                            return;
                        }
                    }
                }

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

                switch (outcome.Verdict)
                {
                    case StreamVerdict.Cancelled:
                        return;

                    case StreamVerdict.Unsupported:
                        // Configuration, not weather: reconnecting reproduces the identical answer
                        // forever, so retrying would be an unbounded hot loop against an endpoint
                        // that will never stream until an operator changes something. Latch it, say
                        // so loudly, and stop. A re-initialize clears the latch.
                        lock (_subscriptionLock)
                        {
                            _streamUnsupported = true;
                        }

                        _streamDegraded = true;
                        Degrade(outcome.Error!);

                        _logger.LogError(
                            outcome.Error,
                            "MTConnect driver {DriverInstanceId} cannot stream /sample from {AgentUri}: the endpoint did not answer with a framed multipart stream. Subscriptions stay registered but will receive no updates until the driver is re-initialized; check for a proxy or load balancer in front of the Agent.",
                            _driverInstanceId, options.AgentUri);

                        return;

                    case StreamVerdict.Reopen:
                        attempt = 0;
                        _reconnectAttempts = 0;

                        break;

                    default:
                        _streamDegraded = true;
                        Degrade(outcome.Error!);

                        // A stream that DELIVERED before it dropped proves the endpoint works, so
                        // its failure starts a fresh backoff ladder. Without this the counter is
                        // monotonic for the life of the process: a healthy agent that drops a
                        // connection once an hour would be pinned at MaxBackoffMs within a day, and
                        // "the first retry is immediate" would be true exactly once ever.
                        attempt = outcome.DeliveredAny ? 1 : attempt + 1;
                        _reconnectAttempts = attempt;

                        _logger.LogWarning(
                            outcome.Error,
                            "MTConnect driver {DriverInstanceId} lost the /sample stream from {AgentUri}; reconnecting from sequence {Cursor} (attempt {Attempt}, delivered={Delivered}).",
                            _driverInstanceId, options.AgentUri, cursor, attempt, outcome.DeliveredAny);

                        break;
                }
            }
        }
        catch (Exception ex)
        {
            _logger.LogError(
                ex,
                "MTConnect driver {DriverInstanceId} stopped its /sample pump on an unexpected fault; subscriptions will receive no further updates until the driver is re-initialized.",
                _driverInstanceId);

            _streamDegraded = true;
            Degrade(ex);
        }
        finally
        {
            // The session's cursor is the opening `from` of the NEXT pump, and until now it only
            // ever held the priming /current's. A last-unsubscribe + resubscribe (no re-initialize)
            // would therefore reopen at a sequence the Agent has long since evicted: it self-heals
            // via OUT_OF_RANGE, but only after a wasted round trip, and a still-buffered stale
            // cursor replays historical observations to subscribers first. One publish per pump
            // lifetime, dropped if a lifecycle change has already installed a newer session.
            PublishAgentCursor(instanceId, cursor);
        }
    }

    /// <summary>
    ///     Enumerates one <c>/sample</c> connection until it ends, and classifies how it ended.
    ///     Re-baselines are performed here, <b>inside</b> the enumeration, so the index and the
    ///     subscribers are up to date before the stream is reopened.
    /// </summary>
    private async Task<StreamOutcome> ConsumeStreamAsync(
        IMTConnectAgentClient client,
        MTConnectDriverOptions options,
        long cursor,
        long instanceId,
        CancellationToken ct)
    {
        // The RUNNING cursor the gap check is made against: the previous chunk's nextSequence, equal
        // to the opening `from` only for the first chunk. Comparing every chunk against the opening
        // `from` instead reports a gap on every chunk once the Agent's buffer rolls past it — an
        // endless /current re-baseline storm against a perfectly healthy stream.
        var expected = cursor;

        // Whether THIS connection ever handed us a chunk. A stream that delivered and then dropped
        // is a working endpoint having a bad moment; one that never delivered may be an endpoint
        // that cannot work at all. They must not share a backoff ladder — see RunSampleStreamAsync.
        var delivered = false;

        try
        {
            await foreach (var chunk in client.SampleAsync(cursor, ct).ConfigureAwait(false))
            {
                if (chunk is null)
                {
                    continue;
                }

                // CHECKED BEFORE THE GAP: an Agent restart changes instanceId *and* resets
                // sequences, so a restart usually trips IsSequenceGap too — and the two need
                // different handling. A gap keeps the held values (they are still this device's);
                // a restart invalidates every one of them, because the device model they describe
                // no longer exists. Testing the gap first would misdiagnose the restart and keep
                // serving values the new Agent may never report again.
                if (chunk.InstanceId != instanceId)
                {
                    _logger.LogWarning(
                        "MTConnect driver {DriverInstanceId} saw {AgentUri} restart (instanceId {Previous} -> {Current}); dropping every held observation and re-baselining.",
                        _driverInstanceId, options.AgentUri, instanceId, chunk.InstanceId);

                    // Dropped BEFORE the re-baseline, and the subscribers are told: if the /current
                    // then fails, holding stale values from a dead device model would be worse than
                    // holding none.
                    ObservationIndex.Clear();
                    FanOut(SubscribedReferences(), ObservationIndex);

                    // The driver's own state is consistent by this point, so it is safe to hand the
                    // restart to arbitrary caller code — which may call straight back into
                    // DiscoverAsync.
                    AnnounceAgentRestart(chunk.InstanceId, options);

                    return await RebaselineAsync(client, options, expected, instanceId, delivered, ct)
                        .ConfigureAwait(false);
                }

                if (IMTConnectAgentClient.IsSequenceGap(expected, chunk))
                {
                    _logger.LogWarning(
                        "MTConnect driver {DriverInstanceId} fell out of {AgentUri}'s buffer (expected sequence {Expected}, oldest retained {FirstSequence}); re-baselining from /current.",
                        _driverInstanceId, options.AgentUri, expected, chunk.FirstSequence);

                    return await RebaselineAsync(client, options, expected, instanceId, delivered, ct)
                        .ConfigureAwait(false);
                }

                ApplyAndFanOut(chunk);
                delivered = true;

                // EVERY chunk advances the cursor, including an observation-free heartbeat: the
                // Agent sends those precisely so a quiet connection can be told from a dead one, and
                // they carry the sequence forward. Not advancing on one makes the NEXT chunk look
                // like a gap.
                expected = chunk.NextSequence;

                _streamDegraded = false;
                PublishHealthy(DateTime.UtcNow);
            }

            // The seam contracts that cancellation is the ONLY way this enumeration ends quietly, so
            // a quiet end under a live token is a non-conforming client — reported as the lost
            // stream it is rather than treated as "the subscription is simply idle" (#485).
            return ct.IsCancellationRequested
                ? new StreamOutcome(StreamVerdict.Cancelled, expected, instanceId, delivered, null)
                : new StreamOutcome(
                    StreamVerdict.Transient,
                    expected,
                    instanceId,
                    delivered,
                    new MTConnectStreamEndedException(
                        MTConnectStreamEndReason.ConnectionClosed, options.AgentUri, 0));
        }
        catch (OperationCanceledException) when (ct.IsCancellationRequested)
        {
            return new StreamOutcome(StreamVerdict.Cancelled, expected, instanceId, delivered, null);
        }
        catch (MTConnectStreamNotSupportedException ex)
        {
            return new StreamOutcome(StreamVerdict.Unsupported, expected, instanceId, delivered, ex);
        }
        catch (InvalidDataException ex)
        {
            // THE OTHER HALF OF RING-BUFFER OVERFLOW, and the one IsSequenceGap cannot see: a real
            // cppagent answers a `from` that has already fallen out of its buffer with an
            // MTConnectError / OUT_OF_RANGE document served under HTTP 200, which the client
            // surfaces here rather than as a gap-bearing chunk. Re-baselining on the gap alone would
            // mean the driver recovers from falling a little behind but not from falling a lot.
            _logger.LogWarning(
                ex,
                "MTConnect driver {DriverInstanceId} could not resume {AgentUri}'s /sample stream at sequence {Cursor} (the Agent rejected it as out of range); re-baselining from /current.",
                _driverInstanceId, options.AgentUri, cursor);

            return await RebaselineAsync(client, options, expected, instanceId, delivered, ct)
                .ConfigureAwait(false);
        }
        catch (Exception ex)
        {
            // Transient by default: a dropped connection, a closing boundary, the heartbeat
            // watchdog's TimeoutException. Reconnect under backoff.
            return new StreamOutcome(StreamVerdict.Transient, expected, instanceId, delivered, ex);
        }
    }

    /// <summary>
    ///     Re-primes the observation index from a fresh <c>/current</c> and reports where the stream
    ///     should resume. The recovery path for every way the incremental sequence can be broken:
    ///     a ring-buffer overflow (detected as a gap, or refused outright as <c>OUT_OF_RANGE</c>)
    ///     and an Agent restart.
    /// </summary>
    /// <remarks>
    ///     <b>This calls the client directly and takes no lock.</b> Re-baselining by calling
    ///     <see cref="ReinitializeAsync"/> — the obvious way to write it, since that method already
    ///     does exactly this work — would take the non-reentrant <see cref="_lifecycle"/> semaphore
    ///     from inside a pump that a lifecycle method may already be waiting on, and hang the driver
    ///     permanently with no exception and no log line.
    /// </remarks>
    private async Task<StreamOutcome> RebaselineAsync(
        IMTConnectAgentClient client,
        MTConnectDriverOptions options,
        long fallbackCursor,
        long fallbackInstanceId,
        bool deliveredAny,
        CancellationToken ct)
    {
        try
        {
            var current = await BoundedAsync(client.CurrentAsync, "/current", options.RequestTimeoutMs, ct)
                .ConfigureAwait(false);

            PublishAgentCursor(current.InstanceId, current.NextSequence);
            ApplyAndFanOut(current);

            _streamDegraded = false;
            PublishHealthy(DateTime.UtcNow);

            _logger.LogInformation(
                "MTConnect driver {DriverInstanceId} re-baselined from {AgentUri}'s /current ({ObservationCount} observation(s), agent instanceId {InstanceId}) and resumes /sample from sequence {NextSequence}.",
                _driverInstanceId,
                options.AgentUri,
                current.Observations.Count,
                current.InstanceId,
                current.NextSequence);

            return new StreamOutcome(
                StreamVerdict.Reopen, current.NextSequence, current.InstanceId, deliveredAny, null);
        }
        catch (OperationCanceledException) when (ct.IsCancellationRequested)
        {
            return new StreamOutcome(
                StreamVerdict.Cancelled, fallbackCursor, fallbackInstanceId, deliveredAny, null);
        }
        catch (Exception ex)
        {
            // The Agent is unreachable or unusable right now. Treated as an ordinary transient
            // failure so the reconnect backoff applies — without it, an Agent that is down would be
            // re-baselined against as fast as the loop can spin. The cursor is deliberately left
            // where it was: reopening there either works or lands right back here.
            return new StreamOutcome(
                StreamVerdict.Transient, fallbackCursor, fallbackInstanceId, deliveredAny, ex);
        }
    }

    /// <summary>
    ///     The reconnect delay before attempt <paramref name="attempt"/>, as a pure function — the
    ///     backoff policy is asserted directly rather than inferred from how long a test slept.
    /// </summary>
    /// <remarks>
    ///     The first retry honours <see cref="MTConnectReconnectOptions.MinBackoffMs"/> (<c>0</c> by
    ///     default = immediate, which is what an operator wants after a one-off blip); every later
    ///     one multiplies, from a <see cref="BackoffGrowthFloorMs"/> floor, up to
    ///     <see cref="MTConnectReconnectOptions.MaxBackoffMs"/>. Operator-authored nonsense cannot
    ///     un-bound the loop: a multiplier that cannot grow the delay falls back to
    ///     <see cref="DefaultBackoffMultiplier"/>, a non-positive cap falls back to
    ///     <see cref="DefaultMaxBackoffMs"/> (clamping to it would make EVERY delay zero — a spin,
    ///     not a backoff), and a negative minimum clamps to zero.
    /// </remarks>
    /// <param name="attempt">The 1-based reconnect attempt about to be made.</param>
    /// <param name="reconnect">The authored backoff options.</param>
    /// <returns>How long to wait before that attempt.</returns>
    internal static TimeSpan BackoffFor(int attempt, MTConnectReconnectOptions reconnect)
    {
        ArgumentNullException.ThrowIfNull(reconnect);

        // A non-positive cap is "unset", NOT "no cap". Math.Max(0, ...) would clamp every delay to
        // zero and turn the reconnect loop into a spin against a refused connection.
        var capMs = reconnect.MaxBackoffMs > 0 ? reconnect.MaxBackoffMs : DefaultMaxBackoffMs;
        var minMs = Math.Max(0, reconnect.MinBackoffMs);

        if (attempt <= 1)
        {
            return TimeSpan.FromMilliseconds(Math.Min(minMs, capMs));
        }

        var multiplier = reconnect.BackoffMultiplier > 1.0 ? reconnect.BackoffMultiplier : DefaultBackoffMultiplier;
        var delayMs = (double)Math.Max(minMs, BackoffGrowthFloorMs);

        for (var step = 2; step <= attempt; step++)
        {
            delayMs *= multiplier;

            if (delayMs >= capMs)
            {
                return TimeSpan.FromMilliseconds(capMs);
            }
        }

        return TimeSpan.FromMilliseconds(Math.Min(delayMs, capMs));
    }

    /// <summary>
    ///     Folds a <c>/current</c> snapshot or one <c>/sample</c> chunk into the shared index and
    ///     publishes every reference it touched to the handles that subscribe it.
    /// </summary>
    private void ApplyAndFanOut(MTConnectStreamsResult document)
    {
        var index = ObservationIndex;
        index.Apply(document);

        if (document.Observations is not { Count: > 0 })
        {
            return;
        }

        // De-duplicated in document order: one Agent document may report a data item more than once
        // (a Condition container carries several simultaneously-active states), and the index has
        // already reconciled those into a single snapshot.
        var touched = new List<string>(document.Observations.Count);
        var seen = new HashSet<string>(StringComparer.Ordinal);
        foreach (var observation in document.Observations)
        {
            if (observation is null || string.IsNullOrWhiteSpace(observation.DataItemId))
            {
                continue;
            }

            if (seen.Add(observation.DataItemId))
            {
                touched.Add(observation.DataItemId);
            }
        }

        FanOut(touched, index);
    }

    /// <summary>
    ///     Publishes the index's current snapshot of <paramref name="references"/> to every handle
    ///     that subscribes them. References nobody subscribed raise nothing: the Agent streams the
    ///     whole device, and republishing all of it would flood the server with values no client
    ///     asked for.
    /// </summary>
    private void FanOut(IReadOnlyList<string> references, MTConnectObservationIndex index)
    {
        if (references.Count == 0)
        {
            return;
        }

        SubscriptionState[] states;
        lock (_subscriptionLock)
        {
            if (_subscriptions.Count == 0)
            {
                return;
            }

            states = [.. _subscriptions.Values];
        }

        // The lock is released before any callback: handlers are caller code.
        foreach (var reference in references)
        {
            DataValueSnapshot? snapshot = null;

            foreach (var state in states)
            {
                if (!state.References.Contains(reference))
                {
                    continue;
                }

                // Read once per reference, so every handle sees the identical snapshot.
                snapshot ??= index.Get(reference);
                RaiseDataChange(state.Handle, reference, snapshot);
            }
        }
    }

    /// <summary>
    ///     Raises one data-change callback to <b>every</b> subscriber, absorbing anything each one
    ///     throws. A consumer bug must not tear down the pump that serves the others — nor rob them
    ///     of the value.
    /// </summary>
    /// <remarks>
    ///     <para>
    ///         <b>The invocation list is walked by hand, with the try/catch INSIDE the loop.</b> A
    ///         plain <c>OnDataChange?.Invoke(...)</c> is a multicast call: the first handler that
    ///         throws aborts the rest of the list, so one broken consumer silently starves every
    ///         other subscriber of that observation — and a single try/catch around the whole
    ///         invoke absorbs the exception while leaving the starvation in place, which reads as
    ///         "isolated" in a log and is not.
    ///     </para>
    ///     <para>
    ///         One <see cref="DataChangeEventArgs"/> is shared across the list deliberately: it is
    ///         an immutable record, and allocating per handler on the hot path would buy nothing.
    ///     </para>
    /// </remarks>
    private void RaiseDataChange(MTConnectSampleHandle handle, string reference, DataValueSnapshot snapshot)
    {
        var handlers = OnDataChange;
        if (handlers is null)
        {
            return;
        }

        var args = new DataChangeEventArgs(handle, reference, snapshot);

        foreach (var registration in handlers.GetInvocationList())
        {
            try
            {
                ((EventHandler<DataChangeEventArgs>)registration).Invoke(this, args);
            }
            catch (Exception ex)
            {
                LogSubscriberFault(ex, reference, handle);
            }
        }
    }

    /// <summary>
    ///     Reports a faulting data-change subscriber <b>once per stream generation</b> at Warning,
    ///     and at Debug thereafter.
    /// </summary>
    /// <remarks>
    ///     A consumer that throws on every value would otherwise emit one Warning per observation
    ///     per handle — on a busy Agent, thousands a second, which buries the incident it is
    ///     reporting along with everything else in the log. Nothing is lost: the later faults are
    ///     still emitted, at Debug, and the latch is cleared whenever the pump (re)starts.
    /// </remarks>
    private void LogSubscriberFault(Exception ex, string reference, MTConnectSampleHandle handle)
    {
        if (Interlocked.Exchange(ref _subscriberFaultLogged, 1) == 0)
        {
            _logger.LogWarning(
                ex,
                "MTConnect driver {DriverInstanceId} caught a fault from a data-change subscriber for {Reference} on {SubscriptionId}; the stream continues and every other subscriber still received the value. Further subscriber faults are logged at Debug until the stream restarts.",
                _driverInstanceId, reference, handle.DiagnosticId);

            return;
        }

        _logger.LogDebug(
            ex,
            "MTConnect driver {DriverInstanceId} caught a further fault from a data-change subscriber for {Reference} on {SubscriptionId}.",
            _driverInstanceId, reference, handle.DiagnosticId);
    }

    /// <summary>Every reference any live subscription is interested in.</summary>
    private List<string> SubscribedReferences()
    {
        lock (_subscriptionLock)
        {
            var all = new HashSet<string>(StringComparer.Ordinal);
            foreach (var state in _subscriptions.Values)
            {
                all.UnionWith(state.References);
            }

            return [.. all];
        }
    }

    /// <summary>
    ///     Whether any live subscription actually names a reference. <b>Caller must hold
    ///     <see cref="_subscriptionLock"/>.</b> A subscription over an empty reference list is legal
    ///     (a caller that adds tags later) and must not open a stream by itself.
    /// </summary>
    private bool HasSubscribedReferencesCore()
    {
        foreach (var state in _subscriptions.Values)
        {
            if (state.References.Count > 0)
            {
                return true;
            }
        }

        return false;
    }

    /// <summary>
    ///     Records where the Agent now is — its <c>instanceId</c> <b>and</b> the sequence a stream
    ///     should resume from — onto whichever session is installed.
    /// </summary>
    /// <remarks>
    ///     <para>
    ///         <b>Both fields move together, in one compare-and-swap.</b> They are one fact about
    ///         one Agent process, and the session's own contract is that a reader gets the client
    ///         and its cursor as a consistent pair. Publishing the id alone (as this did) left the
    ///         session holding a NEW instanceId beside the cursor of the process that had just
    ///         died — the exact tear the packing exists to prevent.
    ///     </para>
    ///     <para>
    ///         <b>There is no "the id did not change, so skip" early return</b>, because the two
    ///         re-baselines that do NOT change the id — a sequence gap and an <c>OUT_OF_RANGE</c>
    ///         rejection — are precisely the ones whose whole purpose is to move the cursor.
    ///         Suppressing the write there would leave the next pump reopening at the stale
    ///         sequence that had just been rejected.
    ///     </para>
    ///     <para>
    ///         Compare-and-swap for the same reason <see cref="FlushOptionalCachesAsync"/> uses one:
    ///         a lifecycle change landing mid-update has newer state, so this write is dropped
    ///         rather than replayed over it.
    ///     </para>
    /// </remarks>
    /// <param name="instanceId">The Agent instanceId last observed.</param>
    /// <param name="nextSequence">The sequence a <c>/sample</c> stream should resume from.</param>
    private void PublishAgentCursor(long instanceId, long nextSequence)
    {
        while (true)
        {
            var session = Volatile.Read(ref _session);
            if (session is null ||
                (session.InstanceId == instanceId && session.NextSequence == nextSequence))
            {
                return;
            }

            var updated = session with { InstanceId = instanceId, NextSequence = nextSequence };
            if (ReferenceEquals(Interlocked.CompareExchange(ref _session, updated, session), session))
            {
                return;
            }
        }
    }

    /// <summary>
    ///     The <c>/probe</c> device model: the cached copy when warm, otherwise a fresh
    ///     deadline-bounded <c>/probe</c> that re-populates the cache. This is the accessor every
    ///     model consumer (Task 12's <c>DiscoverAsync</c>) must use, so that
    ///     <see cref="FlushOptionalCachesAsync"/> can never leave browse permanently disabled.
    /// </summary>
    /// <remarks>
    ///     Lock-free by design — see the <see cref="_session"/> remarks for why this does not take
    ///     <see cref="_lifecycle"/>. It reads the session <b>once</b>, fetches against that
    ///     session's client, and re-caches only if that same session is still installed, so a
    ///     concurrent <see cref="ShutdownAsync"/>/<see cref="ReinitializeAsync"/> can neither be
    ///     undone by a late re-cache nor leak a model belonging to a client that is gone. A teardown
    ///     that lands mid-fetch surfaces as the <see cref="InvalidOperationException"/> below rather
    ///     than as a raw <see cref="ObjectDisposedException"/> from the disposed client.
    /// </remarks>
    /// <param name="cancellationToken">Cancellation token.</param>
    /// <returns>The Agent's device model.</returns>
    /// <exception cref="InvalidOperationException">
    ///     The driver is not initialized, or it was torn down while this fetch was in flight.
    /// </exception>
    internal async Task<MTConnectProbeModel> GetOrFetchProbeModelAsync(CancellationToken cancellationToken)
    {
        var session = Volatile.Read(ref _session) ?? throw NotConnected();

        if (session.ProbeModel is not null)
        {
            return session.ProbeModel;
        }

        MTConnectProbeModel model;
        try
        {
            model = await BoundedAsync(session.Client.ProbeAsync, "/probe", _options.RequestTimeoutMs, cancellationToken)
                .ConfigureAwait(false);
        }
        catch (ObjectDisposedException ex)
        {
            // The lock-free window the _session remarks name: a teardown disposed this client after
            // we captured it. Reported as the not-connected error the caller already handles, so
            // browse has exactly one failure mode to reason about.
            throw NotConnected(ex);
        }

        // Publish only onto the session we actually fetched against.
        var recached = session with { ProbeModel = model, ProbeModelDataItemCount = CountDataItems(model) };
        _ = Interlocked.CompareExchange(ref _session, recached, session);

        return model;
    }

    private InvalidOperationException NotConnected(Exception? inner = null) =>
        new($"MTConnect driver '{_driverInstanceId}' has no live agent client; the /probe device model cannot be fetched unless InitializeAsync has succeeded and the driver has not since been shut down or re-initialized.",
            inner);

    /// <summary>
    ///     Builds driver options from a <c>DriverConfig</c> JSON document.
    /// </summary>
    /// <remarks>
    ///     Lives here rather than on the factory because the driver itself needs it: the runtime
    ///     delivers config changes as JSON to a live instance (see the class remarks). Task 15's
    ///     <c>MTConnectDriverFactoryExtensions.CreateInstance</c> should <b>delegate to this method</b>
    ///     rather than deserializing a second DTO — two parsers for one document drift.
    /// </remarks>
    /// <param name="driverInstanceId">Driver instance id, used only in error text.</param>
    /// <param name="driverConfigJson">The driver's <c>DriverConfig</c> JSON.</param>
    /// <returns>The parsed options.</returns>
    /// <exception cref="InvalidOperationException">The document is unusable or omits <c>agentUri</c>.</exception>
    internal static MTConnectDriverOptions ParseOptions(string driverInstanceId, string driverConfigJson)
    {
        ArgumentException.ThrowIfNullOrWhiteSpace(driverInstanceId);
        ArgumentException.ThrowIfNullOrWhiteSpace(driverConfigJson);

        MTConnectDriverConfigDto? dto;
        try
        {
            dto = JsonSerializer.Deserialize<MTConnectDriverConfigDto>(driverConfigJson, JsonOptions);
        }
        catch (JsonException ex)
        {
            throw new InvalidOperationException(
                $"MTConnect driver config for '{driverInstanceId}' is not valid JSON: {ex.Message}", ex);
        }

        if (dto is null)
        {
            throw new InvalidOperationException(
                $"MTConnect driver config for '{driverInstanceId}' deserialised to null");
        }

        if (string.IsNullOrWhiteSpace(dto.AgentUri))
        {
            throw new InvalidOperationException(
                $"MTConnect driver config for '{driverInstanceId}' missing required AgentUri");
        }

        return new MTConnectDriverOptions
        {
            AgentUri = dto.AgentUri.Trim(),
            DeviceName = string.IsNullOrWhiteSpace(dto.DeviceName) ? null : dto.DeviceName.Trim(),
            RequestTimeoutMs = dto.RequestTimeoutMs ?? 5_000,
            SampleIntervalMs = dto.SampleIntervalMs ?? 1_000,
            SampleCount = dto.SampleCount ?? 1_000,
            HeartbeatMs = dto.HeartbeatMs ?? 10_000,
            Tags = dto.Tags is { Count: > 0 }
                ? [.. dto.Tags.Select(t => BuildTag(t, driverInstanceId))]
                : [],
            Probe = new MTConnectProbeOptions
            {
                Enabled = dto.Probe?.Enabled ?? true,
                Interval = TimeSpan.FromMilliseconds(dto.Probe?.IntervalMs ?? 5_000),
                Timeout = TimeSpan.FromMilliseconds(dto.Probe?.TimeoutMs ?? 2_000),
            },
            Reconnect = new MTConnectReconnectOptions
            {
                MinBackoffMs = dto.Reconnect?.MinBackoffMs ?? 0,
                MaxBackoffMs = dto.Reconnect?.MaxBackoffMs ?? 30_000,
                BackoffMultiplier = dto.Reconnect?.BackoffMultiplier ?? 2.0,
            },
        };
    }

    // ---- lifecycle internals (all called with _lifecycle held) ----

    /// <summary>
    ///     Brings the driver up against <paramref name="options"/>, reusing
    ///     <paramref name="existingClient"/> when the caller determined the transport is unchanged.
    ///     Nothing is published until BOTH agent calls have succeeded, so a failure can never leave a
    ///     half-initialized instance behind.
    /// </summary>
    private async Task StartCoreAsync(
        MTConnectDriverOptions options, IMTConnectAgentClient? existingClient, CancellationToken ct)
    {
        var lastRead = ReadHealth().LastSuccessfulRead;
        WriteHealth(new DriverHealth(DriverState.Initializing, lastRead, null));

        IMTConnectAgentClient? built = null;
        try
        {
            // arch-review 01/S-6: a 0 authored here must never come to mean "wait forever". Checked
            // in the driver, not only in the production client's constructor, so the guard holds for
            // any client implementation behind the seam.
            RequirePositive(options.RequestTimeoutMs, nameof(MTConnectDriverOptions.RequestTimeoutMs));
            RequirePositive(options.HeartbeatMs, nameof(MTConnectDriverOptions.HeartbeatMs));
            RequirePositive(options.SampleIntervalMs, nameof(MTConnectDriverOptions.SampleIntervalMs));
            RequirePositive(options.SampleCount, nameof(MTConnectDriverOptions.SampleCount));

            // Same 01/S-6 guard for Task 13's two knobs, gated on the feature that reads them. A 0
            // interval turns the "periodic" probe into an unthrottled hot loop against the Agent, and
            // a 0 (or negative) timeout cancels every probe before it can be answered — pinning the
            // host to Stopped forever and reporting an outage that does not exist. Checked only when
            // probing is ON because a disabled probe reads neither value, and faulting a deployment
            // over a field the driver never touches would buy nothing; flipping Enabled back on is a
            // config edit, and ReinitializeAsync runs this same validation.
            if (options.Probe.Enabled)
            {
                RequirePositive(options.Probe.Interval, $"{nameof(MTConnectDriverOptions.Probe)}.{nameof(MTConnectProbeOptions.Interval)}");
                RequirePositive(options.Probe.Timeout, $"{nameof(MTConnectDriverOptions.Probe)}.{nameof(MTConnectProbeOptions.Timeout)}");
            }

            var client = existingClient;
            if (client is null)
            {
                built = _agentClientFactory(options)
                    ?? throw new InvalidOperationException(
                        $"The MTConnect agent-client factory for driver '{_driverInstanceId}' returned null.");
                client = built;
            }

            var probe = await BoundedAsync(client.ProbeAsync, "/probe", options.RequestTimeoutMs, ct)
                .ConfigureAwait(false);
            var current = await BoundedAsync(client.CurrentAsync, "/current", options.RequestTimeoutMs, ct)
                .ConfigureAwait(false);

            // ---- commit point: everything below is non-throwing bookkeeping ----
            _options = options;
            built = null;

            // The Agent this session talks to, as of now. A restart is measured against THIS, so a
            // re-initialize against a different Agent process never re-announces the change the
            // operator's own action caused.
            Volatile.Write(ref _announcedInstanceId, current.InstanceId);

            // One publish, so a concurrent lock-free reader sees the new client, the new probe
            // cache, and the pump's opening cursor together or none of them.
            Volatile.Write(
                ref _session,
                new AgentSession(client, probe, CountDataItems(probe), current.InstanceId, current.NextSequence));

            var index = new MTConnectObservationIndex(options.Tags);
            index.Apply(current);
            Volatile.Write(ref _index, index);

            // A fresh session has no failures behind it — including the /sample verdict, because a
            // re-initialize is precisely the event that means an operator changed something.
            _streamDegraded = false;
            _readDegraded = false;

            WriteHealth(new DriverHealth(DriverState.Healthy, DateTime.UtcNow, null));

            // Standing subscriptions survive a lifecycle restart; everything they hold came from the
            // baseline just thrown away, so the restarted pump republishes as its first act. It is
            // the PUMP that republishes rather than this method, deliberately: raising subscriber
            // callbacks from here would run caller code on the thread holding the non-reentrant
            // lifecycle semaphore, and a handler that touched the driver's lifecycle would deadlock.
            lock (_subscriptionLock)
            {
                _streamUnsupported = false;
                StartSampleStreamCore(republishOnStart: true);
            }

            StartProbeLoopCore(client, options);

            _logger.LogInformation(
                "MTConnect driver {DriverInstanceId} initialized against {AgentUri}{DeviceScope}: {DeviceCount} device(s), {ObservationCount} primed observation(s), agent instanceId {InstanceId}.",
                _driverInstanceId,
                options.AgentUri,
                options.DeviceName is null ? string.Empty : $"/{options.DeviceName}",
                probe.Devices.Count,
                index.Count,
                current.InstanceId);
        }
        catch (Exception ex)
        {
            // A client this attempt built is unreachable from anywhere else; one the caller handed in
            // is disposed by the teardown below. Either way nothing survives a failed start.
            SafeDispose(built);
            await TeardownCoreAsync().ConfigureAwait(false);

            WriteHealth(new DriverHealth(DriverState.Faulted, lastRead, ex.Message));

            _logger.LogError(
                ex,
                "MTConnect driver {DriverInstanceId} failed to initialize against {AgentUri}; the driver is Faulted and serves no values.",
                _driverInstanceId, options.AgentUri);

            throw;
        }
    }

    /// <summary>
    ///     Releases the agent client and drops every piece of served state. Never throws — it runs
    ///     on the failure path of <see cref="StartCoreAsync"/>, where a second exception would mask
    ///     the real one.
    /// </summary>
    private Task TeardownCoreAsync()
    {
        // Retire the whole session in one write, BEFORE disposing: a lock-free reader that still
        // holds the old reference gets a disposed client (caught + named at each seam), never a
        // session that is half-torn-down.
        var session = Volatile.Read(ref _session);
        Volatile.Write(ref _session, null);
        SafeDispose(session?.Client);

        // A fresh index rather than Clear(): the tag table it coerces against belongs to the options
        // that are being torn down.
        Volatile.Write(ref _index, new MTConnectObservationIndex(_options.Tags));

        return Task.CompletedTask;
    }

    /// <summary>
    ///     Stops the shared <c>/sample</c> long-poll stream and waits for the pump to actually be
    ///     gone. Idempotent, and never throws — it runs on teardown paths where a second exception
    ///     would mask the first.
    /// </summary>
    /// <remarks>
    ///     <para>
    ///         <b>It waits, rather than merely signalling.</b> Every caller is about to dispose the
    ///         client the pump is enumerating (or install a new observation index the old pump would
    ///         write into). Returning while the pump is still unwinding would let it publish one more
    ///         value into state its owner has already retired — the exact "torn down but still
    ///         reporting" shape the lifecycle is written to prevent.
    ///     </para>
    ///     <para>
    ///         <b>⚠️ This runs while <see cref="_lifecycle"/> is held</b> (from
    ///         <see cref="InitializeAsync"/>, <see cref="ReinitializeAsync"/> and
    ///         <see cref="ShutdownAsync"/>) — and it awaits the pump. That is only safe because the
    ///         pump never touches <see cref="_lifecycle"/>: its re-baseline calls
    ///         <see cref="IMTConnectAgentClient.CurrentAsync"/> directly (see
    ///         <see cref="RebaselineAsync"/>) rather than re-entering a lifecycle method, and every
    ///         await it makes observes the pump token cancelled below. Break either property and
    ///         this await is a permanent hang with no exception and no log line.
    ///     </para>
    ///     <para>
    ///         <b>The wait is bounded</b> (<see cref="TeardownWait"/>, and the caller's token) — the
    ///         one thing outside the guarantee above is a subscriber callback, and an
    ///         <see cref="OnDataChange"/> handler that never returns would otherwise block teardown
    ///         forever. That is not merely slow: <c>DriverInstanceActor.PostStop</c> blocks an Akka
    ///         dispatcher thread on <see cref="ShutdownAsync"/> while this driver holds
    ///         <see cref="_lifecycle"/>, so an unbounded wait wedges an actor-system thread and
    ///         every later lifecycle call with it. Abandoning the wait is safe because the pump's
    ///         registry slots are cleared BEFORE it: a pump left running cannot be resurrected into
    ///         <see cref="_pumpTask"/>, and it is already cancelled.
    ///     </para>
    /// </remarks>
    /// <param name="cancellationToken">The caller's deadline for the teardown wait.</param>
    private async Task StopSampleStreamAsync(CancellationToken cancellationToken = default)
    {
        CancellationTokenSource? cts;
        Task? task;

        lock (_subscriptionLock)
        {
            cts = _pumpCts;
            task = _pumpTask;
            _pumpCts = null;
            _pumpTask = null;
        }

        if (cts is null)
        {
            return;
        }

        try
        {
            await cts.CancelAsync().ConfigureAwait(false);
        }
        catch (ObjectDisposedException)
        {
            // Raced another stop; the pump is already going away.
        }

        var stopped = true;
        if (task is not null)
        {
            try
            {
                await task.WaitAsync(TeardownWait, cancellationToken).ConfigureAwait(false);
            }
            catch (Exception ex) when (ex is TimeoutException or OperationCanceledException)
            {
                stopped = false;

                _logger.LogError(
                    ex,
                    "MTConnect driver {DriverInstanceId} gave up waiting for its /sample pump to stop after {TeardownWait}. The pump is cancelled and can publish nothing further, but something inside it is not returning — the usual cause is an OnDataChange subscriber that blocks. Teardown continues.",
                    _driverInstanceId, TeardownWait);
            }
            catch (Exception ex)
            {
                // RunSampleStreamAsync is written not to throw, so this is belt-and-braces — but a
                // teardown that rethrew would replace whatever failure prompted it.
                _logger.LogDebug(
                    ex,
                    "MTConnect driver {DriverInstanceId} swallowed a fault from its /sample pump while stopping the stream.",
                    _driverInstanceId);
            }
        }

        if (stopped)
        {
            // Disposed ONLY once the pump is provably gone: an abandoned pump still holds this
            // source's token, and disposing underneath it turns its next linked-token creation into
            // an ObjectDisposedException. One leaked CTS beats a mystery exception.
            cts.Dispose();
        }

        // Nothing to be degraded about once the stream is gone — UNLESS the endpoint is latched as
        // unable to stream at all, in which case there is no prospect of one coming back, and
        // clearing this would let the next successful read report a green driver over a subscription
        // that can never deliver (#485). See StartSampleStreamCore.
        lock (_subscriptionLock)
        {
            if (!_streamUnsupported)
            {
                _streamDegraded = false;
            }
        }
    }

    /// <summary>
    ///     Resolves the options a lifecycle call should apply, reporting an unreadable document
    ///     consistently for both entry points before any state is destroyed.
    /// </summary>
    /// <remarks>
    ///     <para>
    ///         One rule, deliberately shared by <see cref="InitializeAsync"/> and
    ///         <see cref="ReinitializeAsync"/>: <b>a config document that cannot be read never
    ///         destroys working state; it faults only a driver that had none.</b>
    ///     </para>
    ///     <para>
    ///         With a live session the driver keeps serving its previous, valid configuration and
    ///         health is left alone — rejecting the edit already fails the deployment
    ///         (<c>DriverInstanceActor</c> turns the throw into <c>ApplyResult(false)</c>), and
    ///         downing a working driver over an unreadable document is a far larger blast radius
    ///         than the change that was refused. Publishing <see cref="DriverState.Faulted"/> while
    ///         still serving values would also be a lie in the other direction.
    ///     </para>
    ///     <para>
    ///         With no live session there is nothing to protect and every tag is already dark, so
    ///         the failure must reach <see cref="GetHealth"/> as <see cref="DriverState.Faulted"/> —
    ///         otherwise the dashboard keeps showing whatever state the driver was in before, which
    ///         is the failure-that-looks-like-success shape this codebase exists to avoid.
    ///     </para>
    /// </remarks>
    private MTConnectDriverOptions ResolveIncomingOptions(string driverConfigJson)
    {
        try
        {
            return HasConfigBody(driverConfigJson)
                ? ParseOptions(_driverInstanceId, driverConfigJson)
                : _options;
        }
        catch (Exception ex)
        {
            if (Volatile.Read(ref _session) is not null)
            {
                _logger.LogError(
                    ex,
                    "MTConnect driver {DriverInstanceId} rejected a new driver config document and kept running its previous configuration.",
                    _driverInstanceId);
            }
            else
            {
                WriteHealth(new DriverHealth(DriverState.Faulted, ReadHealth().LastSuccessfulRead, ex.Message));
                _logger.LogError(
                    ex,
                    "MTConnect driver {DriverInstanceId} could not read its driver config document and has no running configuration to fall back on; the driver is Faulted.",
                    _driverInstanceId);
            }

            throw;
        }
    }

    /// <summary>
    ///     True when <paramref name="driverConfigJson"/> carries a real config body — i.e. a JSON
    ///     object or array with <b>at least one element</b>. The bootstrapper always passes a
    ///     populated document; tests and probe-only callers pass a semantically empty one, which
    ///     keeps the constructor-supplied options.
    /// </summary>
    /// <remarks>
    ///     <para>
    ///         <b>Emptiness is decided by parsing, not by string comparison.</b> Matching the
    ///         literals <c>"{}"</c> / <c>"[]"</c> misses every other spelling of the same document —
    ///         <c>"{ }"</c>, a pretty-printed <c>"{\n}"</c>, <c>"{ /* nothing yet */ }"</c> — and
    ///         each of those would then reach <see cref="ParseOptions"/>, fail the required-AgentUri
    ///         check, and turn a semantically empty document into a cold-start fault or a rejected
    ///         deployment. That is precisely the asymmetry the keep-the-constructor-options rule
    ///         exists to prevent.
    ///     </para>
    ///     <para>
    ///         <b>A malformed document is emphatically NOT empty</b> and returns <c>true</c>, so
    ///         <see cref="ParseOptions"/> runs and produces the real, quoted parse error. Treating
    ///         unreadable text as "no config supplied" would silently swallow a corrupt document and
    ///         start the driver on stale options — a failure wearing success's clothes.
    ///     </para>
    /// </remarks>
    private static bool HasConfigBody(string? driverConfigJson)
    {
        if (string.IsNullOrWhiteSpace(driverConfigJson))
        {
            return false;
        }

        try
        {
            using var document = JsonDocument.Parse(driverConfigJson, DocumentOptions);

            return document.RootElement.ValueKind switch
            {
                JsonValueKind.Object => document.RootElement.EnumerateObject().Any(),
                JsonValueKind.Array => document.RootElement.EnumerateArray().Any(),

                // A bare `null` document is the JSON spelling of "nothing supplied".
                JsonValueKind.Null => false,

                // A scalar root is not a config object at all; let ParseOptions say so.
                _ => true,
            };
        }
        catch (JsonException)
        {
            return true;
        }
    }

    /// <summary>
    ///     Whether the new options change anything <see cref="IMTConnectAgentClient"/> reads at
    ///     construction, and therefore cannot be applied to a live client. See
    ///     <see cref="ReinitializeAsync"/> for why this is wider than the plan's AgentUri/DeviceName.
    /// </summary>
    private static bool RequiresClientRebuild(MTConnectDriverOptions current, MTConnectDriverOptions incoming) =>
        !string.Equals(current.AgentUri, incoming.AgentUri, StringComparison.Ordinal)
        || !string.Equals(current.DeviceName, incoming.DeviceName, StringComparison.Ordinal)
        || current.RequestTimeoutMs != incoming.RequestTimeoutMs
        || current.HeartbeatMs != incoming.HeartbeatMs
        || current.SampleIntervalMs != incoming.SampleIntervalMs
        || current.SampleCount != incoming.SampleCount;

    /// <summary>
    ///     Counts every data item the device model declares. Done once, when the model is cached,
    ///     rather than re-walking the tree on every 30 s <see cref="GetMemoryFootprint"/> poll.
    /// </summary>
    private static int CountDataItems(MTConnectProbeModel model) =>
        model.Devices.Sum(d => d.AllDataItems().Count());

    /// <summary>
    ///     Runs one unary agent call under a deadline linked to the caller's token. Belt-and-braces
    ///     over the production client's own per-call deadline: the seam does not oblige an
    ///     implementation to bound itself, and an unbounded initialize would wedge the driver-host
    ///     actor (arch-review R2-01).
    /// </summary>
    private async Task<T> BoundedAsync<T>(
        Func<CancellationToken, Task<T>> call, string request, int timeoutMs, CancellationToken ct)
    {
        using var deadline = CancellationTokenSource.CreateLinkedTokenSource(ct);
        deadline.CancelAfter(TimeSpan.FromMilliseconds(timeoutMs));

        try
        {
            return await call(deadline.Token).ConfigureAwait(false);
        }
        catch (OperationCanceledException) when (!ct.IsCancellationRequested)
        {
            // A bare "A task was canceled" names neither the Agent nor the request that stalled.
            throw new TimeoutException(
                $"MTConnect {request} request for driver '{_driverInstanceId}' against '{_options.AgentUri}' did not complete within {timeoutMs} ms.");
        }
    }

    /// <summary>
    ///     One Bad-coded snapshot per requested reference — the shape every read failure degrades
    ///     to. Arity is preserved because the caller indexes the result positionally against the
    ///     references it asked for.
    /// </summary>
    private static DataValueSnapshot[] BadBatch(int count, uint status)
    {
        // No source timestamp: nothing was observed. The server timestamp is when we gave up.
        var serverTimestamp = DateTime.UtcNow;
        var results = new DataValueSnapshot[count];

        for (var i = 0; i < count; i++)
        {
            results[i] = new DataValueSnapshot(null, status, null, serverTimestamp);
        }

        return results;
    }

    /// <summary>
    ///     Routes a read failure to the health surface (05/STAB-9's fleet-wide shape): log it and
    ///     degrade, preserving <c>LastSuccessfulRead</c> — the "when did this driver last hear from
    ///     the Agent" diagnostic. Never downgrades <see cref="DriverState.Faulted"/>, which is a
    ///     stronger, config-level signal that a transient read failure must not paper over.
    /// </summary>
    private void DegradeAfterReadFailure(Exception ex)
    {
        _readDegraded = true;
        Degrade(ex);

        _logger.LogWarning(
            ex,
            "MTConnect driver {DriverInstanceId} could not read /current from {AgentUri}; the batch is reported Bad.",
            _driverInstanceId, _options.AgentUri);
    }

    /// <summary>Publishes <see cref="DriverState.Degraded"/> without downgrading a Faulted driver.</summary>
    private void Degrade(Exception ex) => Degrade(ex.Message);

    /// <summary>
    ///     Publishes <see cref="DriverState.Degraded"/> with an explanatory message, without
    ///     downgrading a Faulted driver.
    /// </summary>
    /// <remarks>
    ///     <b>Deliberately not <see cref="DriverState.Faulted"/>, even for the "an operator must fix
    ///     this" cases</b> such as an endpoint that cannot stream. In this codebase Faulted is not
    ///     merely a stronger word: <c>DriverHealthReport</c> maps any Faulted driver to a
    ///     <c>/readyz</c> <b>503</b>, taking the whole server node out of readiness. A driver whose
    ///     reads are perfectly healthy and whose subscriptions are dead is degraded, not unready —
    ///     de-readying the node over one misconfigured Agent would be a far larger blast radius than
    ///     the fault it reports. Faulted stays reserved for the case that earns it: an initialize
    ///     that failed, after which the driver serves nothing at all.
    /// </remarks>
    private void Degrade(string message)
    {
        var current = ReadHealth();
        if (current.State != DriverState.Faulted)
        {
            WriteHealth(new DriverHealth(DriverState.Degraded, current.LastSuccessfulRead, message));
        }
    }

    /// <summary>
    ///     Publishes <see cref="DriverState.Healthy"/> — <b>unless the other data path is currently
    ///     failing</b>.
    /// </summary>
    /// <remarks>
    ///     <para>
    ///         <b>Health precedence, stated deliberately.</b> This driver has two independent Agent
    ///         data paths: <c>/current</c> (the <see cref="IReadable"/> batch read) and
    ///         <c>/sample</c> (the subscription pump). They are different HTTP requests and fail
    ///         separately — a proxy that collapses <c>multipart/x-mixed-replace</c> breaks streaming
    ///         while reads stay perfect; an Agent that stalls composing a whole-device snapshot
    ///         breaks reads while the stream keeps flowing.
    ///     </para>
    ///     <para>
    ///         The naive last-writer-wins publish would let either path paper over the other's
    ///         failure, and "Healthy" while half the surface returns Bad is precisely the
    ///         failure-wearing-success shape this codebase exists to avoid. So each path owns a
    ///         flag, <b>clears only its own</b>, and Healthy requires both clear: the driver reports
    ///         Degraded while <i>anything</i> is failing and recovers as soon as the failing path
    ///         itself recovers. <see cref="DriverState.Faulted"/> — a config-level verdict — is
    ///         never downgraded by either.
    ///     </para>
    /// </remarks>
    /// <param name="at">When the Agent was last successfully heard from.</param>
    private void PublishHealthy(DateTime at)
    {
        var current = ReadHealth();

        if (_streamDegraded || _readDegraded)
        {
            if (current.State != DriverState.Faulted)
            {
                WriteHealth(new DriverHealth(DriverState.Degraded, at, current.LastError));
            }

            return;
        }

        WriteHealth(new DriverHealth(DriverState.Healthy, at, null));
    }

    /// <summary>
    ///     Disposes an agent client without letting a disposal fault escape. Teardown runs on the
    ///     failure path of <see cref="StartCoreAsync"/>, where a second exception would replace the
    ///     real one — but the swallowed exception is still <b>traced</b> rather than vanishing: a
    ///     client that cannot be released is how a socket-handle leak starts, and an empty catch is
    ///     the only place in this driver where a failure would leave no evidence at all. Debug
    ///     level, because it is diagnostic detail about an already-reported failure.
    /// </summary>
    private void SafeDispose(IMTConnectAgentClient? client)
    {
        if (client is null)
        {
            return;
        }

        try
        {
            client.Dispose();
        }
        catch (Exception ex)
        {
            _logger.LogDebug(
                ex,
                "MTConnect driver {DriverInstanceId} swallowed a fault while disposing its agent client during teardown; the client may not have released its connections.",
                _driverInstanceId);
        }
    }

    private static void RequirePositive(int value, string optionName)
    {
        if (value <= 0)
        {
            throw new ArgumentOutOfRangeException(
                nameof(MTConnectDriverOptions),
                value,
                $"{optionName} must be positive; a non-positive value would either un-bound a deadline or ask the Agent for nothing.");
        }
    }

    /// <summary>
    ///     The <see cref="TimeSpan"/> half of the 01/S-6 guard, for the connectivity probe's two
    ///     knobs. A non-positive interval is an unthrottled loop; a non-positive timeout cancels
    ///     every request before it can be answered.
    /// </summary>
    private static void RequirePositive(TimeSpan value, string optionName)
    {
        if (value <= TimeSpan.Zero)
        {
            throw new ArgumentOutOfRangeException(
                nameof(MTConnectDriverOptions),
                value,
                $"{optionName} must be positive; a non-positive value would either spin the connectivity probe against the Agent or cancel every probe before it could be answered.");
        }
    }

    /// <summary>Barrier-protected read of the cross-thread health snapshot.</summary>
    private DriverHealth ReadHealth() => Volatile.Read(ref _health);

    /// <summary>Barrier-protected publish of a new health snapshot.</summary>
    private void WriteHealth(DriverHealth value) => Volatile.Write(ref _health, value);

    private static MTConnectTagDefinition BuildTag(MTConnectTagDto dto, string driverInstanceId)
    {
        if (string.IsNullOrWhiteSpace(dto.FullName))
        {
            throw new InvalidOperationException(
                $"MTConnect driver config for '{driverInstanceId}' has a tag with no FullName (the Agent DataItem id it binds).");
        }

        return new MTConnectTagDefinition(
            dto.FullName.Trim(),
            // Absent means "no declared type": String is the one coercion that cannot fail, and it
            // carries the Agent's text through untouched.
            dto.DriverDataType is null
                ? DriverDataType.String
                : ParseEnum<DriverDataType>(dto.DriverDataType, dto.FullName, driverInstanceId, nameof(MTConnectTagDefinition.DriverDataType)),
            dto.IsArray ?? false,
            dto.ArrayDim ?? 0,
            dto.MtCategory,
            dto.MtType,
            dto.MtSubType,
            dto.Units);
    }

    /// <summary>
    ///     Parses an enum authored as a <b>name</b>. Config surfaces serialize enums as names
    ///     project-wide; a numerically-serialized enum is the known trap that faults a driver at
    ///     deploy, so this reports the offending field rather than silently taking member 0.
    /// </summary>
    private static T ParseEnum<T>(string raw, string tagName, string driverInstanceId, string field)
        where T : struct, Enum =>
        Enum.TryParse<T>(raw, ignoreCase: true, out var parsed) && Enum.IsDefined(parsed)
            ? parsed
            : throw new InvalidOperationException(string.Create(
                CultureInfo.InvariantCulture,
                $"MTConnect driver config for '{driverInstanceId}' tag '{tagName}' has an unrecognised {field} '{raw}'. Expected one of: {string.Join(", ", Enum.GetNames<T>())}."));

    /// <summary>
    ///     The live agent client and the <c>/probe</c> cache derived from it, as one immutable
    ///     unit. Immutable so that publishing a change is a single reference swap — see the
    ///     <see cref="_session"/> remarks for why the three values must move together.
    /// </summary>
    /// <param name="Client">The agent client this session owns and will dispose on teardown.</param>
    /// <param name="ProbeModel">
    ///     The cached device model, or <c>null</c> once <see cref="FlushOptionalCachesAsync"/> has
    ///     dropped it. Never null at the moment a session is first published.
    /// </param>
    /// <param name="ProbeModelDataItemCount">
    ///     Data items declared by <paramref name="ProbeModel"/>, counted once at cache time and
    ///     always <c>0</c> when it is <c>null</c>.
    /// </param>
    /// <param name="InstanceId">
    ///     The Agent's <c>instanceId</c> as last observed. Updated in place (by
    ///     <see cref="PublishAgentCursor"/>) when the pump sees the Agent restart, so the
    ///     session always names the Agent process the client is actually talking to.
    /// </param>
    /// <param name="NextSequence">
    ///     The sequence the <c>/sample</c> pump should open at — the priming <c>/current</c>'s
    ///     <c>nextSequence</c>. Lives here so that starting the pump reads the client and its cursor
    ///     as one consistent pair; a torn read would open the stream at another session's sequence.
    /// </param>
    private sealed record AgentSession(
        IMTConnectAgentClient Client,
        MTConnectProbeModel? ProbeModel,
        int ProbeModelDataItemCount,
        long InstanceId,
        long NextSequence);

    /// <summary>
    ///     One live subscription: its handle and the references it wants. Both are effectively
    ///     immutable once registered, so the pump reads <see cref="References"/> without a lock —
    ///     it takes <see cref="_subscriptionLock"/> only to snapshot the registry itself.
    /// </summary>
    /// <param name="Handle">The identity handed to the caller and stamped on every event.</param>
    /// <param name="References">The de-duplicated DataItem ids this subscription publishes.</param>
    private sealed record SubscriptionState(MTConnectSampleHandle Handle, HashSet<string> References);

    /// <summary>How one <c>/sample</c> connection ended, and therefore what the pump does next.</summary>
    private enum StreamVerdict
    {
        /// <summary>Cancelled by teardown — stop, publish nothing, report nothing.</summary>
        Cancelled = 0,

        /// <summary>A transient failure: degrade and reconnect under the backoff.</summary>
        Transient = 1,

        /// <summary>Re-baselined successfully: reopen immediately at the new cursor, no backoff.</summary>
        Reopen = 2,

        /// <summary>The endpoint cannot stream at all: latch, report loudly, and stop retrying.</summary>
        Unsupported = 3
    }

    /// <param name="Verdict">How the connection ended.</param>
    /// <param name="Cursor">The sequence the next connection must open at.</param>
    /// <param name="InstanceId">The Agent instance the cursor belongs to.</param>
    /// <param name="DeliveredAny">
    ///     Whether this connection yielded at least one chunk before it ended. Distinguishes a
    ///     working endpoint having a bad moment from one that has never worked, which is what keeps
    ///     the reconnect backoff from ratcheting monotonically over a process lifetime.
    /// </param>
    /// <param name="Error">The failure, when there was one.</param>
    private sealed record StreamOutcome(
        StreamVerdict Verdict, long Cursor, long InstanceId, bool DeliveredAny, Exception? Error);

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

    /// <summary>
    ///     Nullable-init mirror of the driver's <c>DriverConfig</c> JSON. Every property is nullable
    ///     so an omitted key means "use the default" rather than "reset to zero".
    /// </summary>
    private sealed class MTConnectDriverConfigDto
    {
        public string? AgentUri { get; init; }
        public string? DeviceName { get; init; }
        public int? RequestTimeoutMs { get; init; }
        public int? SampleIntervalMs { get; init; }
        public int? SampleCount { get; init; }
        public int? HeartbeatMs { get; init; }
        public List<MTConnectTagDto>? Tags { get; init; }
        public MTConnectProbeDto? Probe { get; init; }
        public MTConnectReconnectDto? Reconnect { get; init; }
    }

    /// <summary>One authored tag. <c>DriverDataType</c> stays a string — see <see cref="ParseEnum{T}"/>.</summary>
    private sealed class MTConnectTagDto
    {
        public string? FullName { get; init; }
        public string? DriverDataType { get; init; }
        public bool? IsArray { get; init; }
        public int? ArrayDim { get; init; }
        public string? MtCategory { get; init; }
        public string? MtType { get; init; }
        public string? MtSubType { get; init; }
        public string? Units { get; init; }
    }

    /// <summary>Background connectivity-probe knobs (Task 13 consumes them).</summary>
    private sealed class MTConnectProbeDto
    {
        public bool? Enabled { get; init; }
        public int? IntervalMs { get; init; }
        public int? TimeoutMs { get; init; }
    }

    /// <summary>Reconnect-backoff knobs (Task 11's pump consumes them).</summary>
    private sealed class MTConnectReconnectDto
    {
        public int? MinBackoffMs { get; init; }
        public int? MaxBackoffMs { get; init; }
        public double? BackoffMultiplier { get; init; }
    }
}