lmxopcua/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt/Sparkplug/SparkplugIngestor.cs

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

namespace ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Sparkplug;

/// <summary>
///     The Sparkplug B ingest state machine — design doc §3.6, the driver's correctness core. It owns
///     the authored <c>(group, node, device?, metric) → RawPath</c> index, the live
///     <see cref="BirthCache"/>, one <see cref="SequenceTracker"/> per edge node, and the rebirth
///     policy; it turns every decoded NBIRTH/DBIRTH/NDATA/DDATA/NDEATH/DDEATH/STATE into
///     <see cref="OnDataChange"/> notifications plus <see cref="LastValueCache"/> updates, or into a
///     bounded, debounced rebirth NCMD.
/// </summary>
/// <remarks>
///     <para>
///         <b>The published reference is the RawPath.</b> Every <see cref="DataChangeEventArgs"/> this
///         type raises carries <see cref="MqttTagDefinition.Name"/> — the tag's RawPath — never a
///         composed Sparkplug reference such as <c>Plant1/EdgeA/Filler1:Temperature</c> and never the
///         topic. <c>DriverHostActor</c>'s dual-namespace fan-out is keyed by RawPath, so publishing
///         under a Sparkplug-shaped key produces a driver that satisfies a naive unit test and delivers
///         nothing at all in production. The Sparkplug tuple is the <i>lookup</i>; the RawPath is the
///         <i>identity</i>.
///     </para>
///     <para>
///         <b>Bind by metric NAME, resolve by alias (§3.6 #1).</b> A DATA metric carries only an alias;
///         the alias is resolved through the scope's <see cref="AliasTable"/> — rebuilt wholesale by
///         every birth — back to the stable metric name, and <i>that</i> name is what the authored index
///         is keyed by. Binding a tag to an alias silently mis-routes Pressure into a Temperature tag
///         the first time an edge node reuses an alias across a rebirth: good quality, plausible values,
///         completely wrong.
///     </para>
///     <para>
///         <b>Every signed integer goes through <see cref="SparkplugMetricBinding.Reinterpret"/>.</b>
///         Sparkplug carries Int8/16/32/64 as two's complement in an <i>unsigned</i> proto field, so a
///         metric whose value is -42 arrives as 4294967254. Skipping the call publishes a plausible,
///         Good-quality, completely wrong number.
///     </para>
///     <para>
///         <b>An invalid payload never mutates state.</b> An undecodable body advances no sequence
///         tracker, clears no alias table and evicts no birth — including for an NDEATH. Garbage bytes
///         are not evidence of anything, and a hostile or merely broken publisher on a group-wide
///         <c>#</c> subscription must not be able to blank the address space with them.
///     </para>
///     <para>
///         <b>NDEATH is not part of the sequence.</b> It is the broker-published Last Will and carries
///         no <c>seq</c>; it is tied to its birth by <see cref="SequenceTracker.IsDeathForCurrentSession"/>
///         instead. Feeding it to <see cref="SequenceTracker.Accept"/> reports a gap on every death and
///         answers a node that has just died with a rebirth request it can never serve.
///     </para>
///     <para>
///         <b>Rebirth requests are debounced per edge node.</b> A single lost message on a busy node
///         produces one unknown-alias or data-before-birth event per metric per message until the birth
///         lands; without a floor between requests that is the NCMD storm the design warns about. See
///         <see cref="DefaultRebirthDebounce"/>.
///     </para>
///     <para>
///         <b>Unauthored edge nodes are ignored outright.</b> The driver subscribes
///         <c>spBv1.0/{group}/#</c>, so a large plant delivers traffic for every node in the group. Only
///         messages whose <c>(group, edgeNode)</c> appears in the authored index are processed at all —
///         which is what bounds the sequence-tracker and birth-cache maps by <i>configuration</i> rather
///         than by whatever the plant happens to publish. Device-level filtering happens later, at the
///         metric-resolution step, because an unauthored device's DDATA still advances its edge node's
///         sequence and dropping it early would manufacture a gap.
///     </para>
///     <para>
///         <b>Nothing here throws on MQTTnet's dispatcher thread.</b> <see cref="HandleMessage"/> runs on
///         it: every failure degrades a tag or drops a message, a throwing <see cref="OnDataChange"/>
///         subscriber is contained, and the rebirth publish is dispatched off-thread.
///     </para>
///     <para>
///         <b>Primary-host STATE publishing is deliberately NOT implemented</b> — see
///         <see cref="MqttSparkplugOptions.ActAsPrimaryHost"/> and the remarks on
///         <see cref="HandleState"/>. This type <i>observes</i> STATE; it never publishes one.
///     </para>
/// </remarks>
public sealed class SparkplugIngestor
{
    /// <summary>
    ///     Minimum interval between two rebirth NCMDs aimed at the same edge node, when the caller does
    ///     not supply one. Sized against the observation that a rebirth is a full metadata republish an
    ///     edge node needs a moment to assemble: repeatedly commanding a node that is already
    ///     republishing is the storm, not the cure.
    /// </summary>
    public static readonly TimeSpan DefaultRebirthDebounce = TimeSpan.FromSeconds(10);

    // OPC UA status codes (values verified against Opc.Ua.StatusCodes, not recalled).
    private const uint StatusGood = 0x00000000u;

    /// <summary>
    ///     The source has announced it is offline (NDEATH/DDEATH), or a birth voided the scope that was
    ///     feeding the tag. <c>BadCommunicationError</c> rather than the arguably more literal
    ///     <c>BadNoCommunication</c> (0x80310000) because it is what this repo's drivers actually
    ///     <i>produce</i> for a link that has stopped delivering — Modbus, S7, TwinCAT, FOCAS,
    ///     OpcUaClient and this driver's own plain-mode path all emit it, while no driver emits
    ///     BadNoCommunication at all (it appears once, in a CLI formatter's label map). Following the
    ///     precedent keeps one code meaning "not communicating" across the whole fleet.
    /// </summary>
    private const uint StatusBadCommunicationError = 0x80050000u;

    /// <summary>The metric's value does not fit the tag's effective data type — including an explicit Sparkplug null.</summary>
    private const uint StatusBadTypeMismatch = 0x80740000u;

    /// <summary>The subscribed reference is not an authored Sparkplug tag.</summary>
    private const uint StatusBadNodeIdUnknown = 0x80340000u;

    /// <summary>
    ///     Hard ceiling on <see cref="_warned"/>. Every key fed to <see cref="WarnOnce"/> is derived
    ///     from authored configuration, so the set is bounded by construction — this is the belt to that
    ///     braces. A key set that grows off <i>wire</i> input is a memory leak on the dispatcher path,
    ///     and getting one of the derivations subtly wrong later must degrade to "stops warning" rather
    ///     than to unbounded growth.
    /// </summary>
    private const int MaxWarnKeys = 1024;

    /// <summary>
    ///     How many consecutive <i>metadata</i> rebirths one edge node may be asked for before the
    ///     driver gives up and says so. Bounds the rebirth loop the per-node debounce cannot break: the
    ///     debounce caps the RATE, not the LIFETIME, so a node whose birth this driver can never use —
    ///     one over <see cref="MaxPayloadBytes"/>, or one that simply never comes — would otherwise be
    ///     commanded every debounce period for the life of the process, answering each time with the
    ///     same unusable payload.
    /// </summary>
    private const int MaxConsecutiveMetadataRebirths = 3;

    private readonly string _driverId;
    private readonly ILogger? _logger;
    private readonly TimeSpan _rebirthDebounce;
    private readonly Func<DateTime> _utcNow;

    /// <summary>RawPath → the handle of the subscription currently covering it.</summary>
    private readonly ConcurrentDictionary<string, ISubscriptionHandle> _handleByRawPath = new(StringComparer.Ordinal);

    /// <summary>Handle → the references it was created for, so unsubscribe can undo exactly its own.</summary>
    private readonly ConcurrentDictionary<ISubscriptionHandle, string[]> _refsByHandle = new();

    /// <summary>One stream-continuity tracker per authored edge node; bounded by the authored set.</summary>
    private readonly ConcurrentDictionary<SparkplugNodeKey, SequenceTracker> _trackers = new();

    /// <summary>Last rebirth-NCMD instant per edge node, in UTC ticks — the debounce floor.</summary>
    private readonly ConcurrentDictionary<SparkplugNodeKey, long> _lastRebirthTicks = new();

    /// <summary>
    ///     Edge nodes whose NBIRTH has been <i>seen</i> in the current session, whether or not its
    ///     <c>seq</c> was readable. Distinct from <see cref="SequenceTracker.IsBirthSynchronized"/>,
    ///     which additionally requires a usable <c>seq</c> — see <see cref="CheckSequence"/>.
    /// </summary>
    private readonly ConcurrentDictionary<SparkplugNodeKey, byte> _nodeBirthSeen = new();

    /// <summary>
    ///     Consecutive <i>metadata</i> rebirths requested from an edge node since its last applied
    ///     birth. Capped by <see cref="MaxConsecutiveMetadataRebirths"/> — see
    ///     <see cref="RequestMetadataRebirth"/>.
    /// </summary>
    private readonly ConcurrentDictionary<SparkplugNodeKey, int> _metadataRebirths = new();

    /// <summary>Keys whose anomaly has already been logged loudly once, cleared by <see cref="Register"/>.</summary>
    private readonly ConcurrentDictionary<string, byte> _warned = new(StringComparer.Ordinal);

    /// <summary>In-flight rebirth publishes, so teardown and tests can drain rather than race them.</summary>
    private readonly List<Task> _pendingRebirths = [];

    private readonly EquipmentTagRefResolver<MqttTagDefinition> _resolver;

    private long _handleSeq;

    /// <summary>The authored table + its Sparkplug indexes, swapped atomically by <see cref="Register"/>.</summary>
    private volatile AuthoredTable _authored = AuthoredTable.Empty;

    private readonly MqttDriverOptions _options;
    private volatile SparkplugHostState? _hostState;
    private IMqttSubscribeTransport? _subscribeTransport;
    private IMqttPublishTransport? _publishTransport;

    /// <summary>Initializes a new instance of the <see cref="SparkplugIngestor"/> class.</summary>
    /// <param name="options">
    ///     Driver configuration. <see cref="MqttDriverOptions.Sparkplug"/> supplies the group id, the
    ///     host id and the rebirth policy; a <see langword="null"/> section is treated as the defaults.
    /// </param>
    /// <param name="driverId">Driver instance id, used only for log correlation.</param>
    /// <param name="subscribeTransport">
    ///     The SUBSCRIBE seam. <see langword="null"/> records the group filter without dialling a broker
    ///     (the shape tests use); <see cref="AttachTo"/> supplies the live connection.
    /// </param>
    /// <param name="publishTransport">
    ///     The NCMD publish seam. <see langword="null"/> disables rebirth requests, which are then
    ///     logged and skipped rather than throwing on the dispatcher thread.
    /// </param>
    /// <param name="logger">Optional logger; never receives credentials or payload bodies.</param>
    /// <param name="cache">The last-value store backing <c>IReadable</c>; defaults to a fresh one.</param>
    /// <param name="maxPayloadBytes">
    ///     Ceiling on an inbound message body. Non-positive falls back to
    ///     <see cref="MqttSubscriptionManager.DefaultMaxPayloadBytes"/> — "unbounded" is not offered,
    ///     because the bound protects a shared dispatcher thread.
    /// </param>
    /// <param name="rebirthDebounce">
    ///     Minimum interval between rebirth NCMDs aimed at one edge node. Defaults to
    ///     <see cref="DefaultRebirthDebounce"/>; <see cref="TimeSpan.Zero"/> disables the floor (tests).
    /// </param>
    /// <param name="utcNow">Clock seam for the debounce; defaults to <see cref="DateTime.UtcNow"/>.</param>
    public SparkplugIngestor(
        MqttDriverOptions options,
        string driverId,
        IMqttSubscribeTransport? subscribeTransport = null,
        IMqttPublishTransport? publishTransport = null,
        ILogger? logger = null,
        LastValueCache? cache = null,
        int maxPayloadBytes = MqttSubscriptionManager.DefaultMaxPayloadBytes,
        TimeSpan? rebirthDebounce = null,
        Func<DateTime>? utcNow = null)
    {
        ArgumentNullException.ThrowIfNull(options);
        ArgumentException.ThrowIfNullOrWhiteSpace(driverId);

        _options = options;
        _driverId = driverId;
        _subscribeTransport = subscribeTransport;
        _publishTransport = publishTransport;
        _logger = logger;
        Values = cache ?? new LastValueCache();
        MaxPayloadBytes = maxPayloadBytes > 0 ? maxPayloadBytes : MqttSubscriptionManager.DefaultMaxPayloadBytes;
        _rebirthDebounce = rebirthDebounce is { } d && d >= TimeSpan.Zero ? d : DefaultRebirthDebounce;
        _utcNow = utcNow ?? (() => DateTime.UtcNow);
        _resolver = new EquipmentTagRefResolver<MqttTagDefinition>(
            rawPath => _authored.ByRawPath.TryGetValue(rawPath, out var def) ? def : null);
    }

    /// <inheritdoc cref="ISubscribable.OnDataChange"/>
    public event EventHandler<DataChangeEventArgs>? OnDataChange;

    /// <summary>
    ///     Raised after a birth certificate has been applied — the seam Task 22 wires to
    ///     <c>IRediscoverable.OnRediscoveryNeeded</c> so a DBIRTH introducing metrics the previous birth
    ///     did not carry rebuilds the address space. Deliberately not wired here: this type detects, the
    ///     driver decides.
    /// </summary>
    public event EventHandler<SparkplugBirthObservedEventArgs>? BirthObserved;

    /// <summary>Raised when a primary-host STATE message is observed. Informational; gates nothing.</summary>
    public event EventHandler<SparkplugHostStateEventArgs>? HostStateObserved;

    /// <summary>The last observed value per RawPath — the push→poll bridge <c>IReadable</c> serves from.</summary>
    public LastValueCache Values { get; }

    /// <summary>The live birth state: one <see cref="AliasTable"/> per observed scope. Read by Tasks 22/23.</summary>
    public BirthCache Births { get; } = new();

    /// <summary>Ceiling on an inbound message body; a larger message is dropped before any decode.</summary>
    public int MaxPayloadBytes { get; }

    /// <summary>
    ///     The last observed primary-host STATE, or <see langword="null"/> when none has been seen.
    /// </summary>
    public SparkplugHostState? HostState => _hostState;

    /// <summary>The group-wide filter this ingestor subscribes, or <see langword="null"/> when no group is configured.</summary>
    public string? GroupFilter
    {
        get
        {
            var group = _options.Sparkplug?.GroupId;
            return string.IsNullOrWhiteSpace(group) ? null : $"spBv1.0/{group}/#";
        }
    }

    /// <summary>
    ///     The primary-host STATE filter this ingestor subscribes, or <see langword="null"/> when no host
    ///     id is configured. Deliberately narrowed to the configured host id rather than
    ///     <c>spBv1.0/STATE/#</c>: this driver has no use for a stranger's host state, and a wildcard
    ///     would grow an unbounded observation map off traffic nobody asked for.
    /// </summary>
    public string? StateFilter
    {
        get
        {
            var hostId = _options.Sparkplug?.HostId;
            return string.IsNullOrWhiteSpace(hostId) ? null : SparkplugTopic.FormatState(hostId);
        }
    }

    /// <summary>
    ///     The <b>pre-3.0</b> primary-host STATE filter (<c>STATE/{hostId}</c>, no namespace prefix), or
    ///     <see langword="null"/> when no host id is configured. Subscribed alongside
    ///     <see cref="StateFilter"/> so the legacy form this driver already parses can actually arrive.
    /// </summary>
    public string? LegacyStateFilter
    {
        get
        {
            var hostId = _options.Sparkplug?.HostId;
            return string.IsNullOrWhiteSpace(hostId) ? null : $"STATE/{hostId}";
        }
    }

    /// <summary>
    ///     Wires this ingestor onto a live connection: inbound messages route to
    ///     <see cref="HandleMessage"/>, the connection becomes both transports, and every reconnect runs
    ///     <see cref="OnReconnectedAsync"/> (re-subscribe + birth-state reset + late-join rebirth).
    /// </summary>
    /// <param name="connection">The connection to observe.</param>
    public void AttachTo(MqttConnection connection)
    {
        ArgumentNullException.ThrowIfNull(connection);

        // Binding to a session means the previous one is gone. Reset FIRST, before delivery is wired:
        // with a persistent session a broker may push queued messages the moment CONNACK lands, ahead
        // of our own SUBSCRIBE, and those must not resolve against the old session's aliases.
        ResetSessionState("attach to a new broker session");

        _subscribeTransport = connection;
        _publishTransport = connection;
        connection.MessageReceived += HandleMessage;
        connection.Reconnected += OnReconnectedAsync;
    }

    /// <summary>
    ///     Replaces the authored tag table with the deploy's raw tags, rebuilding the Sparkplug indexes.
    ///     Wholesale, not additive: a redeploy that drops a tag must stop feeding it.
    /// </summary>
    /// <remarks>
    ///     <b>A redeploy that ADDS an edge node does not rebirth it here.</b> Until that deploy the node
    ///     was unauthored, so its traffic was filtered out and the driver holds no birth for it. It
    ///     recovers on the node's next DATA message, which lands as data-before-birth and asks for one —
    ///     self-correcting, and deliberately not a fresh NCMD fan-out on every redeploy, which on a large
    ///     plant would make a routine config change a group-wide rebirth storm.
    /// </remarks>
    /// <param name="rawTags">The authored raw tags delivered by the deploy artifact.</param>
    /// <returns>How many entries mapped to a usable Sparkplug definition.</returns>
    public int Register(IEnumerable<RawTagEntry> rawTags)
    {
        ArgumentNullException.ThrowIfNull(rawTags);

        var byRawPath = new Dictionary<string, MqttTagDefinition>(StringComparer.Ordinal);
        foreach (var entry in rawTags)
        {
            if (MqttTagDefinitionFactory.FromSparkplugTagConfig(entry.TagConfig, entry.RawPath, out var def))
            {
                byRawPath[entry.RawPath] = def;
            }
            else
            {
                _logger?.LogWarning(
                    "MQTT driver '{DriverId}': tag config did not map to a Sparkplug definition "
                    + "(groupId/edgeNodeId/metricName are all required); skipping. RawPath={RawPath}",
                    _driverId,
                    entry.RawPath);
            }
        }

        var table = AuthoredTable.Build(byRawPath);
        _authored = table;
        _warned.Clear();

        // Drop state belonging to edge nodes this deploy removed. Without this a tracker (and its
        // rebirth debounce slot) for a node nobody reads survives every future redeploy.
        foreach (var key in _trackers.Keys)
        {
            if (!table.EdgeNodes.Contains(key))
            {
                _trackers.TryRemove(key, out _);
                _lastRebirthTicks.TryRemove(key, out _);
                _nodeBirthSeen.TryRemove(key, out _);
                _metadataRebirths.TryRemove(key, out _);
            }
        }

        // …and their BIRTHS, which is the half that actually mis-routes. While an edge node is
        // unauthored its traffic is dropped at Dispatch, so its cached alias table stops refreshing —
        // it FREEZES rather than going stale-and-obvious. Deploy A authors Plant1/EdgeA/Filler1,
        // deploy B drops it, deploy C re-adds it a week later having missed every rebirth in between:
        // Births.Find still answers with the week-old table and the next DDATA publishes the wrong
        // metric's value under the right RawPath at Good quality. It is also why Births would otherwise
        // grow monotonically across redeploys.
        //
        // The predicate is edge-node departure ONLY, deliberately narrower than "absent from ByScope".
        // A device with no authored tags but whose EDGE NODE is still authored keeps receiving traffic
        // (the Dispatch filter is node-level, and OnDeviceBirth runs ahead of the authored-scope
        // filter), so its birth is live, not frozen — evicting it would throw away correct state and
        // cost a needless rebirth round-trip the moment a later deploy authored a tag on it.
        foreach (var scope in Births.Scopes)
        {
            var owner = new SparkplugNodeKey(scope.GroupId, scope.EdgeNodeId);
            if (!table.EdgeNodes.Contains(owner))
            {
                // Forgets the node scope and every device under it in one call; idempotent, so a node
                // contributing several scopes to this loop costs one eviction and some no-ops.
                Births.ForgetNode(owner.GroupId, owner.EdgeNodeId);
            }
        }

        foreach (var rawPath in _handleByRawPath.Keys)
        {
            if (!table.ByRawPath.ContainsKey(rawPath))
            {
                _handleByRawPath.TryRemove(rawPath, out _);
            }
        }

        return byRawPath.Count;
    }

    /// <summary>Resolves a RawPath to its authored definition through the shared v3 resolver.</summary>
    /// <param name="rawPath">The driver wire reference.</param>
    /// <param name="def">The definition, when this returns <see langword="true"/>.</param>
    /// <returns><see langword="true"/> when the reference is an authored Sparkplug tag.</returns>
    public bool TryResolve(string rawPath, out MqttTagDefinition def) => _resolver.TryResolve(rawPath, out def!);

    /// <summary>The authored RawPaths, in no particular order — the discovery enumeration source.</summary>
    public IReadOnlyCollection<string> AuthoredRawPaths => _authored.ByRawPath.Keys;

    /// <summary>
    ///     Establishes the group-wide (and, when configured, STATE) subscription and requests a
    ///     late-join rebirth from every authored edge node.
    /// </summary>
    /// <remarks>
    ///     <para>
    ///         <b>Unlike plain mode, the subscribe is NOT driven by <see cref="SubscribeAsync"/>.</b> A
    ///         Sparkplug driver must see birth certificates whether or not the OPC UA server has
    ///         subscribed anything — the birth is where datatypes and aliases come from, and Task 22's
    ///         discovery depends on having seen one. So the driver establishes the filter once, at
    ///         connect, and <see cref="SubscribeAsync"/> only records which references may be published.
    ///     </para>
    ///     <para>
    ///         <b>Total failure throws.</b> There is exactly one filter and it covers every tag, so a
    ///         broker that refuses it leaves the whole driver deaf — the one outcome that must be a
    ///         visible fault (the driver-host resilience layer re-runs initialize under backoff) rather
    ///         than a healthy-looking driver receiving nothing. This is the same rule
    ///         <see cref="OnReconnectedAsync"/> applies, stated once.
    ///     </para>
    /// </remarks>
    /// <param name="cancellationToken">Cancellation for the SUBSCRIBE round-trip.</param>
    /// <returns>A task that completes when the subscribe has been answered.</returns>
    /// <exception cref="InvalidOperationException">No group id is configured, or every filter was refused.</exception>
    public async Task EstablishAsync(CancellationToken cancellationToken)
    {
        // The SAME reset OnReconnectedAsync performs, and for the same reason — this ingestor OUTLIVES
        // a session rebuild. `MqttDriver.AdoptOptions` only replaces it when `!SameIngest`, but
        // `SessionIdentity` is a strict SUPERSET of `IngestIdentity`: changing Host, Port, ClientId,
        // credentials, TLS or a timeout alone yields `SameSession == false` AND `SameIngest == true`,
        // so `ReinitializeAsync` tears the session down and re-establishes on this very instance.
        // `TeardownAsync` touches no ingest state. The driver-host resilience layer re-running
        // `InitializeAsync` after a fault is the second path onto this line.
        //
        // Without the reset, an edge node that restarted during the changeover and rebound alias 5 from
        // Temperature to Pressure would have its first DDATA on the new session resolved against the
        // OLD table — publishing Pressure's value under the Temperature RawPath at Good quality. The
        // late-join rebirth below does not close that: it is dispatched off-thread (DATA can arrive
        // first), passes the debounce, and is a no-op under `requestRebirthOnGap: false` or against a
        // node that ignores the NCMD.
        ResetSessionState("establish a broker session");

        await SubscribeGroupAsync(cancellationToken).ConfigureAwait(false);
        RequestLateJoinRebirths("initial connect");
    }

    /// <summary>
    ///     Records a batch of RawPath references against a new handle. Issues no SUBSCRIBE — the
    ///     group-wide filter is established once by <see cref="EstablishAsync"/>; see its remarks.
    /// </summary>
    /// <param name="fullReferences">The RawPath references to subscribe.</param>
    /// <param name="publishingInterval">Ignored — Sparkplug is push-based.</param>
    /// <param name="cancellationToken">Unused; this call touches no network.</param>
    /// <returns>The handle every resulting <see cref="OnDataChange"/> is attributed to.</returns>
    public Task<ISubscriptionHandle> SubscribeAsync(
        IReadOnlyList<string> fullReferences,
        TimeSpan publishingInterval,
        CancellationToken cancellationToken)
    {
        ArgumentNullException.ThrowIfNull(fullReferences);

        var defs = new List<MqttTagDefinition>(fullReferences.Count);
        var now = _utcNow();
        foreach (var reference in fullReferences)
        {
            if (_resolver.TryResolve(reference, out var def))
            {
                defs.Add(def);
                continue;
            }

            // Not an authored Sparkplug tag. Say so rather than leaving it in "waiting for initial
            // data" forever — a reference that will never be fed is a different fault from one that
            // has simply not been fed yet.
            _logger?.LogWarning(
                "MQTT driver '{DriverId}': subscribe reference '{RawPath}' is not an authored Sparkplug tag.",
                _driverId,
                reference);
            Values.Update(reference, new DataValueSnapshot(null, StatusBadNodeIdUnknown, null, now));
        }

        var handle = new SparkplugSubscriptionHandle(
            $"sparkplug:{Interlocked.Increment(ref _handleSeq)}:[{defs.Count} ref(s)]");
        _refsByHandle[handle] = [.. defs.Select(d => d.Name)];
        foreach (var def in defs)
        {
            _handleByRawPath[def.Name] = handle;
        }

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

    /// <summary>
    ///     Stops attributing notifications to <paramref name="handle"/>. References another live
    ///     subscription also covers keep publishing; the broker-side group filter is untouched.
    /// </summary>
    /// <param name="handle">The handle returned by <see cref="SubscribeAsync"/>.</param>
    /// <param name="cancellationToken">Unused; present for the <c>ISubscribable</c> shape.</param>
    /// <returns>A completed task.</returns>
    public Task UnsubscribeAsync(ISubscriptionHandle handle, CancellationToken cancellationToken)
    {
        if (handle is not null && _refsByHandle.TryRemove(handle, out var refs))
        {
            foreach (var rawPath in refs)
            {
                // Only clear the mapping if it is still OURS — a later overlapping subscription may
                // have taken the reference over, and it must keep publishing.
                _handleByRawPath.TryRemove(new KeyValuePair<string, ISubscriptionHandle>(rawPath, handle));
            }
        }

        return Task.CompletedTask;
    }

    /// <summary>
    ///     Recovers the ingest state after a reconnect: <b>drops every birth</b>, resets every sequence
    ///     tracker, re-establishes the group filter and requests a late-join rebirth from every authored
    ///     edge node (§3.6 invariant #5).
    /// </summary>
    /// <remarks>
    ///     <para>
    ///         <b>Clearing the birth cache is not housekeeping — it is the correctness step.</b> Across
    ///         an outage of unknown length an edge node may have restarted and rebound its aliases; a
    ///         DDATA resolved against the pre-outage table would route a value to whichever tag used to
    ///         own that alias. Same for the sequence baseline: carried across, a stream could look
    ///         contiguous purely because the node published the right number of messages while the
    ///         driver was away.
    ///     </para>
    ///     <para>
    ///         Values are deliberately <b>not</b> staled here. Reconnect is a driver-side event, not a
    ///         statement about the plant; the last observed values remain the best available answer
    ///         until the rebirth lands, and the connection's own health surface already reports
    ///         Reconnecting. Only a death — the source saying it is offline — stales a tag.
    ///     </para>
    /// </remarks>
    /// <param name="cancellationToken">The connection's lifetime token; cancelled by its dispose.</param>
    /// <returns>A task that completes when the re-subscribe has been answered.</returns>
    /// <exception cref="InvalidOperationException">Every filter was refused — the caller tears the session down.</exception>
    public async Task OnReconnectedAsync(CancellationToken cancellationToken)
    {
        ResetSessionState("reconnect");

        await SubscribeGroupAsync(cancellationToken).ConfigureAwait(false);
        RequestLateJoinRebirths("reconnect");
    }

    /// <summary>
    ///     Drops every scrap of per-session ingest state: births, sequence baselines, the birth-seen
    ///     set and the late-join counters. Called from <b>every</b> seam at which the session underneath
    ///     this ingestor changes — <see cref="AttachTo"/>, <see cref="EstablishAsync"/> and
    ///     <see cref="OnReconnectedAsync"/>.
    /// </summary>
    /// <remarks>
    ///     <para>
    ///         <b>This is the correctness step, not housekeeping</b> — and the reason it lives in one
    ///         method called from three places rather than inline in one. Every piece of state cleared
    ///         here is a claim about a session that no longer exists: an alias table binds names to
    ///         numbers the <i>previous</i> edge-node session chose, and a sequence baseline is only
    ///         evidence relative to the stream it was taken from. Carried across, an alias resolves to
    ///         whichever metric used to own it (good quality, plausible value, wrong tag) and a fresh
    ///         stream can look contiguous purely by coincidence.
    ///     </para>
    ///     <para>
    ///         <b>Unconditionally safe.</b> On a freshly-constructed ingestor every collection is
    ///         already empty, so calling it three times during one connect costs three no-ops.
    ///     </para>
    ///     <para>
    ///         Values are deliberately <b>not</b> staled here. A session change is a driver-side event,
    ///         not a statement about the plant; last observed values remain the best available answer
    ///         until a birth lands, and the connection's own health surface already reports the outage.
    ///         Only a death — the source saying it is offline — stales a tag.
    ///     </para>
    /// </remarks>
    /// <param name="reason">Why the reset is happening, for the diagnostic log.</param>
    private void ResetSessionState(string reason)
    {
        var births = Births.Count;
        Births.Clear();
        _trackers.Clear();
        _nodeBirthSeen.Clear();
        _metadataRebirths.Clear();

        if (births > 0)
        {
            _logger?.LogInformation(
                "MQTT driver '{DriverId}': dropped {Count} Sparkplug birth scope(s) and every sequence "
                + "baseline — {Reason}. Aliases will be rebuilt from the next birth certificate.",
                _driverId,
                births,
                reason);
        }
    }

    /// <summary>
    ///     Routes one inbound MQTT message: parses the topic, decodes the Sparkplug payload and hands it
    ///     to <see cref="Dispatch"/>. Runs on MQTTnet's dispatcher thread and <b>never throws</b>.
    /// </summary>
    /// <param name="topic">The concrete topic the message arrived on.</param>
    /// <param name="payload">The message body; valid only for the duration of this call.</param>
    /// <param name="retained">
    ///     The message's retain flag. Not filtered on: a retained NBIRTH/STATE is exactly the
    ///     late-join metadata this driver wants, and Sparkplug DATA is never published retained.
    /// </param>
    public void HandleMessage(string topic, ReadOnlySpan<byte> payload, bool retained)
    {
        try
        {
            if (!SparkplugTopic.TryParse(topic, out var parsed))
            {
                // A group-wide '#' subscription legitimately delivers topics this driver has no opinion
                // about. Debug, not warn — an operator must not have to read one line per stray message.
                _logger?.LogDebug(
                    "MQTT driver '{DriverId}': '{Topic}' is not a Sparkplug topic; ignored.",
                    _driverId,
                    topic);
                return;
            }

            if (parsed.Type == SparkplugMessageType.STATE)
            {
                // A STATE body is a handful of bytes; an oversized one is a broken publisher, not a
                // dispatcher-thread hazard, and it is logged at Debug rather than through WarnOnce
                // because its key would be a host id this driver did not choose.
                if (payload.Length > MaxPayloadBytes)
                {
                    _logger?.LogDebug(
                        "MQTT driver '{DriverId}': STATE on '{Topic}' exceeded the {Max}-byte ceiling; dropped.",
                        _driverId,
                        topic,
                        MaxPayloadBytes);
                    return;
                }

                HandleState(parsed, payload);
                return;
            }

            // The authored-node filter runs HERE as well as inside Dispatch, and the order matters for
            // two reasons. It keeps the oversize warning's key bounded — keyed on the raw topic it grew
            // once per distinct topic off a group-wide '#' subscription, i.e. off wire input a plant (or
            // a hostile publisher) controls. And it drops an unauthored node's body before the protobuf
            // parse rather than after, which on a busy group is most of the decode work this driver
            // would otherwise do for traffic it discards.
            if (parsed.GroupId is not { } groupId || parsed.EdgeNodeId is not { } edgeNodeId)
            {
                return;
            }

            var node = new SparkplugNodeKey(groupId, edgeNodeId);
            if (!_authored.EdgeNodes.Contains(node))
            {
                _logger?.LogDebug(
                    "MQTT driver '{DriverId}': {Type} for unauthored edge node '{Node}'; ignored.",
                    _driverId,
                    parsed.Type,
                    node);
                return;
            }

            if (payload.Length > MaxPayloadBytes)
            {
                // Refused BEFORE any decode: the whole point of the bound is that one publisher cannot
                // impose an unbounded protobuf parse on the shared dispatcher thread. Nothing mutates.
                //
                // An oversized NBIRTH is the worst case and is called out by name: the driver cannot
                // learn this node's aliases at all, so every subsequent DATA message is unroutable. The
                // late-join cap in CheckSequence is what stops that becoming a permanent NCMD loop.
                WarnOnce(
                    $"oversize:{node}",
                    "MQTT driver '{DriverId}': {Type} from '{Node}' carried {Bytes} bytes, over the "
                    + "{Max}-byte ceiling; dropped. Raise maxPayloadBytes if this node's birth is "
                    + "legitimately this large.",
                    _driverId,
                    parsed.Type,
                    node,
                    payload.Length,
                    MaxPayloadBytes);
                return;
            }

            Dispatch(parsed, SparkplugCodec.Decode(payload));
        }
        catch (Exception ex)
        {
            // Defence in depth: every path below is written not to throw, but an escape here surfaces
            // inside MQTTnet's own pump and stalls delivery for every subscription on the connection.
            _logger?.LogError(
                ex,
                "MQTT driver '{DriverId}': Sparkplug ingest threw for '{Topic}'; the message was dropped.",
                _driverId,
                topic);
        }
    }

    /// <summary>
    ///     Routes one <b>decoded</b> Sparkplug message — the state machine proper, and the seam the
    ///     §3.6 matrix is driven through without a broker.
    /// </summary>
    /// <param name="topic">The parsed topic.</param>
    /// <param name="payload">The decoded payload. <b>Checked for validity here</b>, not by the caller.</param>
    public void Dispatch(SparkplugTopic topic, SparkplugPayload payload)
    {
        ArgumentNullException.ThrowIfNull(topic);
        ArgumentNullException.ThrowIfNull(payload);

        if (topic.Type == SparkplugMessageType.STATE)
        {
            return; // STATE carries no Sparkplug payload; HandleMessage routes it separately.
        }

        // NCMD/DCMD flow host → node. Our own rebirth requests come back on the '#' subscription;
        // treating them as ingest would be, at best, noise.
        if (topic.Type is SparkplugMessageType.NCMD or SparkplugMessageType.DCMD)
        {
            return;
        }

        if (topic.GroupId is not { } groupId || topic.EdgeNodeId is not { } edgeNodeId)
        {
            return; // TryParse guarantees both for a non-STATE topic; belt and braces.
        }

        var node = new SparkplugNodeKey(groupId, edgeNodeId);
        if (!_authored.EdgeNodes.Contains(node))
        {
            _logger?.LogDebug(
                "MQTT driver '{DriverId}': {Type} for unauthored edge node '{Node}'; ignored.",
                _driverId,
                topic.Type,
                node);
            return;
        }

        if (!payload.IsValid)
        {
            // NOT a rebirth trigger and NOT a state mutation — see the type remarks. An undecodable
            // body is not evidence that anything about the node's stream has changed.
            WarnOnce(
                $"invalid:{node}",
                "MQTT driver '{DriverId}': {Type} from '{Node}' could not be decoded as a Sparkplug payload; dropped.",
                _driverId,
                topic.Type,
                node);
            return;
        }

        switch (topic.Type)
        {
            case SparkplugMessageType.NBIRTH:
                OnNodeBirth(node, payload);
                break;

            case SparkplugMessageType.DBIRTH when topic.DeviceId is { } dbirthDevice:
                OnDeviceBirth(node, dbirthDevice, payload);
                break;

            case SparkplugMessageType.NDATA:
                OnData(node, new SparkplugScope(groupId, edgeNodeId, null), payload);
                break;

            case SparkplugMessageType.DDATA when topic.DeviceId is { } ddataDevice:
                OnData(node, new SparkplugScope(groupId, edgeNodeId, ddataDevice), payload);
                break;

            case SparkplugMessageType.NDEATH:
                OnNodeDeath(node, payload);
                break;

            case SparkplugMessageType.DDEATH when topic.DeviceId is { } ddeathDevice:
                OnDeviceDeath(node, ddeathDevice, payload);
                break;

            default:
                break;
        }
    }

    /// <summary>
    ///     Awaits every rebirth publish this ingestor has dispatched and not yet observed. Test seam and
    ///     teardown helper — the publishes themselves are fire-and-forget by design, because a rebirth
    ///     NCMD must never be able to park MQTTnet's dispatcher thread.
    /// </summary>
    /// <returns>A task that completes when no dispatched rebirth publish is outstanding.</returns>
    internal async Task DrainPendingRebirthsAsync()
    {
        Task[] pending;
        lock (_pendingRebirths)
        {
            pending = [.. _pendingRebirths];
        }

        foreach (var task in pending)
        {
            try
            {
                await task.ConfigureAwait(false);
            }
            catch (Exception)
            {
                // Already logged where it was dispatched; a failed rebirth is not a caller's problem.
            }
        }
    }

    // ---- births ----

    /// <summary>
    ///     Applies an NBIRTH: restarts the node's sequence at the birth's <c>seq</c>, adopts its
    ///     <c>bdSeq</c> session token, rebuilds the node's catalog, <b>stales every device the birth
    ///     invalidated</b>, and publishes the birth's own values.
    /// </summary>
    private void OnNodeBirth(SparkplugNodeKey node, SparkplugPayload payload)
    {
        // AcceptNodeBirth, never Accept: an NBIRTH restarts the sequence by definition, so running it
        // through the gap detector would flag a gap on the very message that resynchronized everything
        // and answer a birth with a demand for another one.
        // Recorded BEFORE the seq is judged, and independently of the verdict: "a birth arrived" and
        // "the birth established a sequence baseline" are different facts, and conflating them is the
        // rebirth loop CheckSequence documents.
        _nodeBirthSeen[node] = 0;
        _metadataRebirths.TryRemove(node, out _);

        var bdSeq = SequenceTracker.TryReadBdSeq(payload, out var token) ? token : (ulong?)null;
        if (!TrackerFor(node).AcceptNodeBirth(payload.Seq, bdSeq))
        {
            // Deliberately NOT a rebirth trigger: the metadata just arrived, and commanding another
            // birth from a node whose seq we cannot read would loop against the same malformed payload.
            WarnOnce(
                $"birthseq:{node}",
                "MQTT driver '{DriverId}': NBIRTH from '{Node}' carried no usable seq ({Seq}); "
                + "gap detection is disabled for this node until its next birth.",
                _driverId,
                node,
                payload.Seq);
        }

        var outcome = Births.ApplyNodeBirth(node.GroupId, node.EdgeNodeId, payload.Metrics);
        LogBirthAnomaly(node.ToString(), outcome.Result);

        var sourceUtc = ToUtc(payload.TimestampMs);

        // An NBIRTH voids every prior DBIRTH under the node (Sparkplug spec): those devices MUST
        // re-DBIRTH, and until they do their aliases are meaningless. Staling them is what stops the
        // driver serving values resolved against a table it has just thrown away.
        foreach (var invalidated in outcome.InvalidatedDevices)
        {
            FanStale(invalidated.Scope, sourceUtc, "its edge node re-birthed; awaiting a fresh DBIRTH");
        }

        PublishBirthValues(new SparkplugScope(node.GroupId, node.EdgeNodeId, null), payload, sourceUtc);
        RaiseBirthObserved(new SparkplugScope(node.GroupId, node.EdgeNodeId, null), payload);
    }

    /// <summary>Applies a DBIRTH: rebuilds exactly that device's catalog and publishes its values.</summary>
    private void OnDeviceBirth(SparkplugNodeKey node, string deviceId, SparkplugPayload payload)
    {
        // A DBIRTH is a sequenced member of the edge node's stream — only the NBIRTH restarts it.
        CheckSequence(node, payload, "DBIRTH");

        // Metadata arrived, so the give-up counter re-arms: a device that re-births after a bad patch
        // must not stay permanently un-askable because of requests made before it recovered.
        _metadataRebirths.TryRemove(node, out _);

        var result = Births.ApplyDeviceBirth(node.GroupId, node.EdgeNodeId, deviceId, payload.Metrics);
        var scope = new SparkplugScope(node.GroupId, node.EdgeNodeId, deviceId);
        LogBirthAnomaly(scope.ToString(), result);

        var sourceUtc = ToUtc(payload.TimestampMs);
        PublishBirthValues(scope, payload, sourceUtc);
        RaiseBirthObserved(scope, payload);
    }

    // ---- data ----

    /// <summary>
    ///     Applies an NDATA/DDATA: checks stream continuity, resolves every metric through the scope's
    ///     birth-built alias table, and publishes under the authored tag's <b>RawPath</b>.
    /// </summary>
    private void OnData(SparkplugNodeKey node, SparkplugScope scope, SparkplugPayload payload)
    {
        CheckSequence(node, payload, scope.IsDevice ? "DDATA" : "NDATA");

        // Scope-level filtering happens HERE, after the sequence check, never before it: an unauthored
        // device's DDATA is still a member of its edge node's stream, and skipping it earlier would
        // manufacture a gap on the very next authored message. Past this point there is nothing this
        // scope could publish — so there is also nothing worth demanding a rebirth for. Without the
        // guard, a device nobody authored that publishes DDATA without ever DBIRTHing (spec-violating,
        // but they exist) would drive a permanent NCMD at the debounce cadence, forever.
        if (!_authored.ByScope.ContainsKey(scope))
        {
            return;
        }

        var table = Births.Find(scope);
        if (table is null)
        {
            // Data before any birth (§3.6 #3): we hold no alias table for this scope, so every metric
            // in this payload is unroutable. One rebirth request for the whole message, not one per
            // metric — the debounce would collapse them anyway, but not asking is cheaper than asking
            // and being suppressed.
            WarnOnce(
                $"prebirth:{scope}",
                "MQTT driver '{DriverId}': data for '{Scope}' arrived before any birth certificate; "
                + "requesting a rebirth.",
                _driverId,
                scope);
            RequestMetadataRebirth(node, "data before birth");
            return;
        }

        var sourceUtc = ToUtc(payload.TimestampMs);
        var unknownAlias = false;

        foreach (var metric in payload.Metrics)
        {
            var binding = ResolveBinding(table, metric);
            if (binding is null)
            {
                // The birth this table was built from never declared this alias/name — our metadata is
                // behind the node's. Same remedy as data-before-birth, different log.
                unknownAlias = true;
                continue;
            }

            PublishMetric(scope, binding, metric, sourceUtc);
        }

        if (unknownAlias)
        {
            WarnOnce(
                $"alias:{scope}",
                "MQTT driver '{DriverId}': data for '{Scope}' carried a metric the current birth does not "
                + "declare; requesting a rebirth.",
                _driverId,
                scope);
            RequestMetadataRebirth(node, "unknown alias");
        }
    }

    /// <summary>
    ///     Resolves a DATA metric to its birth binding: by <b>alias</b> when it carries one (the normal
    ///     case — a DATA metric after a birth carries an alias and neither name nor datatype), otherwise
    ///     by name, since aliases are optional in Sparkplug.
    /// </summary>
    private static SparkplugMetricBinding? ResolveBinding(AliasTable table, SparkplugMetric metric)
    {
        // NAME FIRST, deliberately. A DATA metric that carries a name has stated its identity in the
        // one vocabulary that survives a rebirth; the alias is a per-birth compression detail. When the
        // two disagree — a publisher that recycled an alias mid-stream, or a birth this driver applied
        // out of order — preferring the alias would resolve to whichever metric currently occupies that
        // slot, which is precisely the mis-route this type's binding rule exists to prevent. The common
        // case costs nothing: a real post-birth DATA metric carries no name at all, so this is a null
        // check before the alias lookup.
        if (metric.Name is { } name && table.ResolveByName(name) is { } byName)
        {
            return byName;
        }

        return metric.Alias is { } alias ? table.Resolve(alias) : null;
    }

    // ---- deaths ----

    /// <summary>
    ///     Applies an NDEATH: verifies it belongs to the session currently believed alive, then forgets
    ///     the node and every device under it and stales their authored tags.
    /// </summary>
    private void OnNodeDeath(SparkplugNodeKey node, SparkplugPayload payload)
    {
        // NOT SequenceTracker.Accept. An NDEATH is the broker-published Last Will: it carries no seq at
        // all, so Accept would report a gap on every death and answer a node that has just died with a
        // rebirth request. The bdSeq tie is the whole mechanism instead.
        var tracker = TrackerFor(node);
        var deathBdSeq = SequenceTracker.TryReadBdSeq(payload, out var token) ? token : (ulong?)null;
        if (!tracker.IsDeathForCurrentSession(deathBdSeq))
        {
            // A previous session's will, delivered after the node already reconnected and re-birthed.
            // Acting on it would mark a live node dead with nothing coming to correct it.
            _logger?.LogInformation(
                "MQTT driver '{DriverId}': ignoring a stale NDEATH for '{Node}' (bdSeq {DeathBdSeq} ≠ current "
                + "{BirthBdSeq}); the node re-birthed before its old will was delivered.",
                _driverId,
                node,
                deathBdSeq,
                tracker.BirthBdSeq);
            return;
        }

        Births.ForgetNode(node.GroupId, node.EdgeNodeId);

        var sourceUtc = ToUtc(payload.TimestampMs);
        foreach (var scope in _authored.ScopesUnder(node))
        {
            FanStale(scope, sourceUtc, "its edge node published NDEATH");
        }
    }

    /// <summary>Applies a DDEATH: forgets exactly that device and stales its authored tags.</summary>
    private void OnDeviceDeath(SparkplugNodeKey node, string deviceId, SparkplugPayload payload)
    {
        // A DDEATH IS published by the edge node and IS sequenced — unlike an NDEATH.
        CheckSequence(node, payload, "DDEATH");

        Births.TryForgetDevice(node.GroupId, node.EdgeNodeId, deviceId, out _);

        FanStale(
            new SparkplugScope(node.GroupId, node.EdgeNodeId, deviceId),
            ToUtc(payload.TimestampMs),
            "the device published DDEATH");
    }

    // ---- STATE ----

    /// <summary>
    ///     Records an observed primary-host STATE message.
    /// </summary>
    /// <remarks>
    ///     <para>
    ///         <b>Receive only. This driver never publishes STATE, and
    ///         <see cref="MqttSparkplugOptions.ActAsPrimaryHost"/> is therefore inert — deliberately, and
    ///         loudly.</b> Being a Sparkplug Primary Host is not "publish an ONLINE message": it is a
    ///         three-part contract — a retained <c>ONLINE</c> STATE published after CONNECT, a matching
    ///         retained <c>OFFLINE</c> registered as the client's <b>MQTT Last Will at CONNECT time</b>,
    ///         and an explicit OFFLINE on clean shutdown. Edge nodes subscribe to it and stop publishing
    ///         while their host is offline. Shipping the ONLINE half without the Will is strictly worse
    ///         than shipping neither: an OtOpcUa node that crashes would leave a retained ONLINE asserting
    ///         forever that a dead host is alive, which is precisely the state the mechanism exists to
    ///         prevent. The Will has to be set inside <c>MqttConnection.BuildClientOptions</c> — a
    ///         different file, a different task — so this task implements the observation half and
    ///         <see cref="MqttDriver"/> warns when the flag is set. See the design doc §7/§8.
    ///     </para>
    ///     <para>
    ///         Both payload forms are accepted on receive: the v3.0 JSON <c>{"online":bool,...}</c> and
    ///         the pre-3.0 bare <c>ONLINE</c>/<c>OFFLINE</c> text.
    ///     </para>
    /// </remarks>
    private void HandleState(SparkplugTopic topic, ReadOnlySpan<byte> payload)
    {
        if (topic.HostId is not { } hostId)
        {
            return;
        }

        if (!TryReadStateOnline(payload, out var online))
        {
            WarnOnce(
                $"state:{hostId}",
                "MQTT driver '{DriverId}': STATE for host '{HostId}' carried an unrecognised payload; ignored.",
                _driverId,
                hostId);
            return;
        }

        var previous = _hostState;
        var state = new SparkplugHostState(hostId, online, _utcNow());
        _hostState = state;

        if (previous is null || previous.Online != online || !string.Equals(previous.HostId, hostId, StringComparison.Ordinal))
        {
            _logger?.LogInformation(
                "MQTT driver '{DriverId}': Sparkplug host '{HostId}' is {State}.",
                _driverId,
                hostId,
                online ? "ONLINE" : "OFFLINE");
        }

        try
        {
            HostStateObserved?.Invoke(this, new SparkplugHostStateEventArgs(state));
        }
        catch (Exception ex)
        {
            // Same containment as every other raise here — this runs on MQTTnet's dispatcher thread.
            _logger?.LogError(ex, "MQTT driver '{DriverId}': a host-state subscriber threw.", _driverId);
        }
    }

    /// <summary>Reads the online flag out of a STATE body — v3.0 JSON first, then the legacy text form.</summary>
    private static bool TryReadStateOnline(ReadOnlySpan<byte> payload, out bool online)
    {
        online = false;
        if (payload.IsEmpty)
        {
            return false;
        }

        try
        {
            var reader = new Utf8JsonReader(payload);
            if (JsonDocument.TryParseValue(ref reader, out var doc))
            {
                using (doc)
                {
                    if (doc.RootElement.ValueKind == JsonValueKind.Object
                        && doc.RootElement.TryGetProperty("online", out var flag)
                        && flag.ValueKind is JsonValueKind.True or JsonValueKind.False)
                    {
                        online = flag.GetBoolean();
                        return true;
                    }
                }
            }
        }
        catch (JsonException)
        {
            // Fall through to the legacy text form.
        }

        Span<char> text = stackalloc char[16];
        if (payload.Length > text.Length || !Encoding.UTF8.TryGetChars(payload, text, out var written))
        {
            return false;
        }

        var trimmed = text[..written].Trim();
        if (trimmed.Equals("ONLINE", StringComparison.OrdinalIgnoreCase))
        {
            online = true;
            return true;
        }

        if (trimmed.Equals("OFFLINE", StringComparison.OrdinalIgnoreCase))
        {
            online = false;
            return true;
        }

        return false;
    }

    // ---- publish ----

    /// <summary>
    ///     Publishes every named metric a birth certificate carried. This is what restores Good quality
    ///     after a death: a birth is a full snapshot, not a delta.
    /// </summary>
    private void PublishBirthValues(SparkplugScope scope, SparkplugPayload payload, DateTime? sourceUtc)
    {
        var table = Births.Find(scope);
        if (table is null)
        {
            return;
        }

        foreach (var metric in payload.Metrics)
        {
            if (metric.Name is not { } name || table.ResolveByName(name) is not { } binding)
            {
                continue;
            }

            PublishMetric(scope, binding, metric, sourceUtc);
        }
    }

    /// <summary>
    ///     Publishes one resolved metric to every authored tag bound to
    ///     <c>(scope, binding.Name)</c> — under the tag's <b>RawPath</b>.
    /// </summary>
    private void PublishMetric(
        SparkplugScope scope,
        SparkplugMetricBinding binding,
        SparkplugMetric metric,
        DateTime? sourceUtc)
    {
        // Bound by NAME. The alias got us here; it is never the key.
        var key = new SparkplugMetricKey(scope.GroupId, scope.EdgeNodeId, scope.DeviceId, binding.Name);
        if (!_authored.ByMetric.TryGetValue(key, out var defs))
        {
            return; // A published metric nobody authored a tag for. Not an error — that is most of them.
        }

        // Arrays slip past the unsupported-type gate below, because ToDriverDataType maps every
        // *Array variant to its ELEMENT type and is therefore never null for one. Design §3.5 defers
        // array support, and the failure is not benign: the value rides in bytes_value as a byte[] at
        // ValueKind.Scalar, so a numeric target throws its way to BadTypeMismatch (safe, wrong reason)
        // while a String/Reference target reaches Convert.ToBase64String and publishes a packed binary
        // buffer as a plausible string at GOOD quality. Gate on the array bit itself.
        if (binding.DataType.IsSparkplugArray())
        {
            WarnOnce(
                $"array-type:{key}",
                "MQTT driver '{DriverId}': metric '{Metric}' on '{Scope}' is a Sparkplug array "
                + "({DataType}); array metrics are not supported in v1 and its tag(s) are not being fed.",
                _driverId,
                binding.Name,
                scope,
                binding.DataType);
            return;
        }

        var birthType = binding.DataType.ToDriverDataType();
        if (birthType is null)
        {
            // DataSet / Template / PropertySet / Unknown — warn-and-skip per the datatype map's
            // contract, and explicitly NOT a rebirth trigger: the node would answer with the same
            // unsupported metric and the pair would loop.
            WarnOnce(
                $"unsupported-type:{key}",
                "MQTT driver '{DriverId}': metric '{Metric}' on '{Scope}' has unsupported Sparkplug type "
                + "{DataType}; its tag(s) are not being fed.",
                _driverId,
                binding.Name,
                scope,
                binding.DataType);
            return;
        }

        var serverUtc = _utcNow();

        foreach (var def in defs)
        {
            // The authored dataType wins when the operator declared one; otherwise the birth's own
            // declaration does. That is what lets a tag be authored with nothing but the metric tuple.
            var target = def.DataTypeAuthored ? def.DataType : birthType.Value;
            if (BuildSnapshot(def, binding, metric, target, sourceUtc, serverUtc) is not { } snapshot)
            {
                continue;
            }

            Values.Update(def.Name, snapshot);
            Raise(def.Name, snapshot);
        }
    }

    /// <summary>
    ///     Turns one metric's wire value into a snapshot for one authored tag, or <see langword="null"/>
    ///     when the metric carried no value at all — in which case the tag keeps whatever it last had
    ///     rather than being re-published unchanged or given an invented quality.
    /// </summary>
    private DataValueSnapshot? BuildSnapshot(
        MqttTagDefinition def,
        SparkplugMetricBinding binding,
        SparkplugMetric metric,
        DriverDataType target,
        DateTime? sourceUtc,
        DateTime serverUtc)
    {
        // Metric timestamp wins over the payload's — it is the acquisition instant at the source.
        var stamp = ToUtc(metric.TimestampMs) ?? sourceUtc;

        switch (metric.ValueKind)
        {
            case SparkplugValueKind.Scalar:
                break;

            case SparkplugValueKind.Null:
                // An explicit is_null. The tag has a declared scalar type and null is not a value of
                // it — the same verdict the plain-mode path reaches for a JSON null.
                return new DataValueSnapshot(null, StatusBadTypeMismatch, stamp, serverUtc);

            default:
                // Absent (no value oneof at all) or Unsupported (DataSet/Template). Neither is a value;
                // leave the tag holding whatever it last had rather than inventing a quality for it.
                WarnOnce(
                    $"valuekind:{def.Name}",
                    "MQTT driver '{DriverId}': metric '{Metric}' for '{RawPath}' carried {ValueKind} rather "
                    + "than a value; skipped.",
                    _driverId,
                    binding.Name,
                    def.Name,
                    metric.ValueKind);
                return null;
        }

        // THE call. Sparkplug carries signed integers as two's complement in an unsigned field, and a
        // DATA metric carries no datatype of its own — only the birth binding knows which applies.
        // Skipping this publishes 4294967254 for a value of -42: plausible, Good, and wrong.
        var raw = binding.Reinterpret(metric.Value);

        return SparkplugValueCoercion.TryCoerce(raw, target, out var value)
            ? new DataValueSnapshot(value, StatusGood, stamp, serverUtc)
            : BadTypeMismatch(def, binding, target, stamp, serverUtc);
    }

    /// <summary>
    ///     The metric decoded, but does not fit the tag's effective type. Warned once per tag; the value
    ///     itself is deliberately never logged — a payload value is plant data, not diagnostics.
    /// </summary>
    private DataValueSnapshot BadTypeMismatch(
        MqttTagDefinition def,
        SparkplugMetricBinding binding,
        DriverDataType target,
        DateTime? stamp,
        DateTime serverUtc)
    {
        WarnOnce(
            $"coerce:{def.Name}",
            "MQTT driver '{DriverId}': metric '{Metric}' ({Sparkplug}) does not fit tag '{RawPath}'s "
            + "{Target}; status BadTypeMismatch.",
            _driverId,
            binding.Name,
            binding.DataType,
            def.Name,
            target);

        return new DataValueSnapshot(null, StatusBadTypeMismatch, stamp, serverUtc);
    }

    /// <summary>Marks every authored tag in <paramref name="scope"/> as no longer communicating.</summary>
    private void FanStale(SparkplugScope scope, DateTime? sourceUtc, string reason)
    {
        if (!_authored.ByScope.TryGetValue(scope, out var defs))
        {
            return;
        }

        var snapshot = new DataValueSnapshot(null, StatusBadCommunicationError, sourceUtc, _utcNow());
        foreach (var def in defs)
        {
            Values.Update(def.Name, snapshot);
            Raise(def.Name, snapshot);
        }

        _logger?.LogInformation(
            "MQTT driver '{DriverId}': staled {Count} tag(s) on '{Scope}' — {Reason}.",
            _driverId,
            defs.Length,
            scope,
            reason);
    }

    /// <summary>
    ///     Raises <see cref="OnDataChange"/> for a RawPath, if a live subscription covers it. A throwing
    ///     subscriber is contained so it cannot starve the rest of the fan-out or escape onto MQTTnet's
    ///     dispatcher thread.
    /// </summary>
    private void Raise(string rawPath, DataValueSnapshot snapshot)
    {
        if (!_handleByRawPath.TryGetValue(rawPath, out var handle))
        {
            return; // Authored but not subscribed: cached so a read answers, nothing to attribute to.
        }

        try
        {
            OnDataChange?.Invoke(this, new DataChangeEventArgs(handle, rawPath, snapshot));
        }
        catch (Exception ex)
        {
            _logger?.LogError(
                ex,
                "MQTT driver '{DriverId}': a data-change subscriber threw for '{RawPath}'.",
                _driverId,
                rawPath);
        }
    }

    private void RaiseBirthObserved(SparkplugScope scope, SparkplugPayload payload)
    {
        var subscribers = BirthObserved;
        if (subscribers is null)
        {
            return;
        }

        try
        {
            subscribers(this, new SparkplugBirthObservedEventArgs(
                scope,
                [.. payload.Metrics.Where(m => m.Name is not null).Select(m => m.Name!)]));
        }
        catch (Exception ex)
        {
            _logger?.LogError(ex, "MQTT driver '{DriverId}': a birth-observed subscriber threw.", _driverId);
        }
    }

    // ---- sequence + rebirth ----

    private SequenceTracker TrackerFor(SparkplugNodeKey node) =>
        _trackers.GetOrAdd(node, static _ => new SequenceTracker());

    /// <summary>
    ///     Offers a sequenced message's <c>seq</c> and requests a rebirth when the stream is not
    ///     contiguous — or when it is contiguous only because the driver joined mid-stream and has never
    ///     seen this node's NBIRTH.
    /// </summary>
    /// <remarks>
    ///     The second case is the one that is easy to miss: <see cref="SequenceTracker.Accept"/> returns
    ///     <see langword="true"/> for the first message after a reset (there is nothing to measure it
    ///     against), so a driver that only checked its return value would sit happily behind a stream
    ///     whose birth certificate — and therefore whose entire alias table — it never received.
    /// </remarks>
    private void CheckSequence(SparkplugNodeKey node, SparkplugPayload payload, string kind)
    {
        var tracker = TrackerFor(node);
        if (!tracker.Accept(payload.Seq))
        {
            WarnOnce(
                $"gap:{node}",
                "MQTT driver '{DriverId}': {Kind} from '{Node}' broke the sequence (seq {Seq}, last {Last}); "
                + "requesting a rebirth.",
                _driverId,
                kind,
                node,
                payload.Seq,
                tracker.LastSeq);
            RequestRebirth(node, "sequence gap");
            return;
        }

        if (tracker.IsBirthSynchronized)
        {
            return;
        }

        // "Birth-synchronized" in SequenceTracker's sense means a birth established a usable SEQUENCE
        // BASELINE — it is false both when no NBIRTH has arrived (a genuine late join, which a rebirth
        // fixes) and when one arrived carrying a seq this driver could not read (which a rebirth does
        // NOT fix: the node answers with the same unusable birth). Reacting to the two identically is
        // what turns the second into a permanent NCMD loop at the debounce cadence, defeating the guard
        // OnNodeBirth deliberately applies one method over. Whether a birth ARRIVED is this type's own
        // fact to keep, so it keeps it.
        if (_nodeBirthSeen.ContainsKey(node))
        {
            return;
        }

        WarnOnce(
            $"latejoin:{node}",
            "MQTT driver '{DriverId}': {Kind} from '{Node}' arrived with no NBIRTH seen since connect; "
            + "requesting a rebirth.",
            _driverId,
            kind,
            node);

        // The residual loop the birth-seen flag cannot see: an NBIRTH that never reaches Dispatch at
        // all, because it exceeded MaxPayloadBytes or because the node never sends one. Nothing about
        // asking again changes that, so it goes through the capped path.
        RequestMetadataRebirth(node, "late join (no NBIRTH observed)");
    }

    /// <summary>
    ///     Requests a rebirth for a <b>missing-metadata</b> reason — late join, data before birth, or an
    ///     alias the current birth does not declare — under a per-node cap on consecutive attempts.
    /// </summary>
    /// <remarks>
    ///     <para>
    ///         <b>Why these three and not a sequence gap.</b> A gap is self-limiting: the tracker
    ///         resynchronizes onto the observed value, so the very next message is contiguous and asks
    ///         for nothing. The three reasons routed here are not — every one of them means "this
    ///         driver holds no usable catalog for this node", and if the node cannot supply one (its
    ///         birth is over the payload ceiling, or it ignores NCMD entirely) then <i>every</i>
    ///         subsequent message re-raises the same condition. The debounce caps the rate at one
    ///         request per period; only this caps the lifetime.
    ///     </para>
    ///     <para>
    ///         <b>Consecutive, not cumulative.</b> Any applied birth clears the counter, so a node that
    ///         recovers is not permanently penalised for having once been unreachable.
    ///     </para>
    /// </remarks>
    /// <param name="node">The edge node to command.</param>
    /// <param name="reason">Why, for the log.</param>
    private void RequestMetadataRebirth(SparkplugNodeKey node, string reason)
    {
        var attempts = _metadataRebirths.AddOrUpdate(node, 1, static (_, n) => n + 1);
        if (attempts > MaxConsecutiveMetadataRebirths)
        {
            WarnOnce(
                $"metadata-exhausted:{node}",
                "MQTT driver '{DriverId}': '{Node}' has not delivered a usable birth certificate after "
                + "{Attempts} rebirth request(s); giving up until one arrives on its own. Its tags will "
                + "not update. Check the node's birth payload size against maxPayloadBytes, its seq "
                + "field, and that it honours NCMD Node Control/Rebirth.",
                _driverId,
                node,
                MaxConsecutiveMetadataRebirths);
            return;
        }

        RequestRebirth(node, reason);
    }

    /// <summary>
    ///     Requests a rebirth from every authored edge node — §3.6 invariant #5. Called on connect and
    ///     on every reconnect.
    /// </summary>
    /// <remarks>
    ///     <b>Bounded by configuration, not by the plant.</b> Sparkplug has no group-wide rebirth
    ///     command — NCMD addresses exactly one edge node — so this is inherently one publish per node.
    ///     The set is the <i>authored</i> edge nodes rather than every node in the group, which is what
    ///     keeps a 200-node group from producing 200 NCMDs for the two nodes this deployment reads. Each
    ///     one still passes through the per-node debounce, so a connection that flaps cannot multiply
    ///     them.
    /// </remarks>
    private void RequestLateJoinRebirths(string reason)
    {
        foreach (var node in _authored.EdgeNodes)
        {
            RequestRebirth(node, reason);
        }
    }

    /// <summary>
    ///     Publishes a rebirth NCMD to one edge node, subject to the
    ///     <see cref="MqttSparkplugOptions.RequestRebirthOnGap"/> policy and the per-node debounce.
    ///     Never blocks the caller and never throws.
    /// </summary>
    private void RequestRebirth(SparkplugNodeKey node, string reason)
    {
        if (_options.Sparkplug is { RequestRebirthOnGap: false })
        {
            _logger?.LogDebug(
                "MQTT driver '{DriverId}': would request a rebirth from '{Node}' ({Reason}) but "
                + "requestRebirthOnGap is off.",
                _driverId,
                node,
                reason);
            return;
        }

        if (_publishTransport is not { } transport)
        {
            _logger?.LogDebug(
                "MQTT driver '{DriverId}': cannot request a rebirth from '{Node}' ({Reason}) — no publish "
                + "transport is attached.",
                _driverId,
                node,
                reason);
            return;
        }

        if (!TryAcquireRebirthSlot(node))
        {
            _logger?.LogDebug(
                "MQTT driver '{DriverId}': rebirth request to '{Node}' ({Reason}) suppressed; one was sent "
                + "less than {Debounce} ago.",
                _driverId,
                node,
                reason,
                _rebirthDebounce);
            return;
        }

        _logger?.LogInformation(
            "MQTT driver '{DriverId}': requesting a Sparkplug rebirth from '{Node}' — {Reason}.",
            _driverId,
            node,
            reason);

        // Off the dispatcher thread: RebirthRequester.RequestAsync serializes a payload and hands it to
        // MQTTnet's publish path, and neither belongs on the thread that is delivering every other
        // subscription's messages. It is bounded by its own 5s deadline.
        TrackRebirth(Task.Run(async () =>
        {
            try
            {
                await RebirthRequester
                    .RequestAsync(transport, node.GroupId, node.EdgeNodeId, CancellationToken.None)
                    .ConfigureAwait(false);
            }
            catch (Exception ex)
            {
                _logger?.LogWarning(
                    ex,
                    "MQTT driver '{DriverId}': the rebirth NCMD to '{Node}' failed; the next gap will retry.",
                    _driverId,
                    node);
            }
        }));
    }

    /// <summary>
    ///     Claims this node's rebirth slot, or reports that one was claimed too recently. Lock-free: this
    ///     runs on MQTTnet's dispatcher thread.
    /// </summary>
    private bool TryAcquireRebirthSlot(SparkplugNodeKey node)
    {
        if (_rebirthDebounce <= TimeSpan.Zero)
        {
            return true;
        }

        var now = _utcNow().Ticks;
        while (true)
        {
            if (!_lastRebirthTicks.TryGetValue(node, out var last))
            {
                if (_lastRebirthTicks.TryAdd(node, now))
                {
                    return true;
                }

                continue;
            }

            if (now - last < _rebirthDebounce.Ticks)
            {
                return false;
            }

            if (_lastRebirthTicks.TryUpdate(node, now, last))
            {
                return true;
            }
        }
    }

    private void TrackRebirth(Task task)
    {
        lock (_pendingRebirths)
        {
            _pendingRebirths.RemoveAll(static t => t.IsCompleted);
            _pendingRebirths.Add(task);
        }
    }

    // ---- subscribe ----

    /// <summary>Issues the group-wide (and STATE) SUBSCRIBE, throwing when nothing could be established.</summary>
    private async Task SubscribeGroupAsync(CancellationToken cancellationToken)
    {
        if (GroupFilter is not { } groupFilter)
        {
            throw new InvalidOperationException(
                $"MQTT driver '{_driverId}': Sparkplug mode requires a groupId — there is no filter to "
                + "subscribe, so the driver would receive nothing.");
        }

        // QoS 1 for both. Sparkplug publishes NBIRTH/DBIRTH/DATA at QoS 0 and NDEATH/STATE at QoS 1;
        // the SUBSCRIBE's QoS is the ceiling the broker may deliver at, so asking for 1 keeps the
        // death and state messages at their intended guarantee and costs nothing for the rest.
        var filters = new List<MqttTopicSubscription>(2)
        {
            // SeedRetained: a retained NBIRTH/STATE replayed at subscribe time IS the late-join
            // metadata this driver wants — the opposite of the plain-mode tag opt-out.
            new(groupFilter, Qos: 1, SeedRetained: true),
        };

        if (StateFilter is { } stateFilter)
        {
            filters.Add(new MqttTopicSubscription(stateFilter, Qos: 1, SeedRetained: true));

            // The pre-3.0 form, which HandleState already parses. Subscribing only the v3.0 topic made
            // that tolerance unreachable in production — the legacy branch could be exercised by a test
            // calling HandleMessage directly and by nothing else. Design §3.1 is explicit: emit v3.0,
            // tolerate legacy on receive, and tolerating requires asking for it.
            filters.Add(new MqttTopicSubscription(
                LegacyStateFilter!,
                Qos: 1,
                SeedRetained: true));
        }

        var transport = _subscribeTransport;
        if (transport is null)
        {
            _logger?.LogWarning(
                "MQTT driver '{DriverId}': Sparkplug subscription recorded with no connection attached; "
                + "nothing was sent to a broker. Call AttachTo before establishing.",
                _driverId);
            return;
        }

        var outcomes = await transport.SubscribeAsync(filters, cancellationToken).ConfigureAwait(false);

        // The group filter is the one that matters: without it the driver is deaf. A refused STATE
        // filter costs an observability signal, not the data plane, so it warns rather than throws.
        var groupGranted = outcomes.Any(o => o.Granted && string.Equals(o.Topic, groupFilter, StringComparison.Ordinal));
        if (!groupGranted)
        {
            throw new InvalidOperationException(
                $"MQTT driver '{_driverId}': the broker refused '{groupFilter}'; tearing the session down "
                + "rather than serving a Sparkplug driver that receives nothing.");
        }

        foreach (var refused in outcomes.Where(o => !o.Granted))
        {
            _logger?.LogWarning(
                "MQTT driver '{DriverId}': broker refused SUBSCRIBE '{Topic}' ({Reason}).",
                _driverId,
                refused.Topic,
                refused.Reason);
        }
    }

    // ---- helpers ----

    private void LogBirthAnomaly(string scope, BirthApplyResult result)
    {
        if (!result.HasAnomaly)
        {
            return;
        }

        // The only window onto a malformed birth — AliasTable takes no logger by design.
        _logger?.LogWarning(
            "MQTT driver '{DriverId}': birth for '{Scope}' was malformed — {Accepted} accepted, "
            + "{Unnamed} unnamed, {AliasCollisions} alias collision(s), {DuplicateNames} duplicate name(s).",
            _driverId,
            scope,
            result.Accepted,
            result.RejectedUnnamed,
            result.AliasCollisions,
            result.DuplicateNames);
    }

    private static DateTime? ToUtc(ulong? epochMs) =>
        epochMs is { } ms && ms <= (ulong)DateTimeOffset.MaxValue.ToUnixTimeMilliseconds()
            ? DateTimeOffset.FromUnixTimeMilliseconds((long)ms).UtcDateTime
            : null;

    /// <summary>
    ///     Logs a warning the first time <paramref name="key"/> is seen and stays quiet thereafter. A
    ///     10 Hz metric with a mis-authored datatype must be diagnosable without drowning the log; the
    ///     key set is bounded because every path that calls this has already filtered to authored nodes,
    ///     scopes or RawPaths, and it is cleared on every <see cref="Register"/>.
    /// </summary>
    private void WarnOnce(string key, string message, params object?[] args)
    {
        // The ceiling is unreachable by design — see MaxWarnKeys. It exists so that a derivation that
        // turns out NOT to be bounded (as the oversize path's raw-topic key was) degrades to silence
        // instead of to unbounded growth on the dispatcher path.
        if (_warned.Count < MaxWarnKeys && _warned.TryAdd(key, 0))
        {
#pragma warning disable CA2254 // Template is a compile-time constant at every call site.
            _logger?.LogWarning(message, args);
#pragma warning restore CA2254
        }
    }

    /// <summary>Opaque subscription identity for the Sparkplug ingest path.</summary>
    private sealed record SparkplugSubscriptionHandle(string DiagnosticId) : ISubscriptionHandle;

    /// <summary>
    ///     The authored tag table and its Sparkplug indexes, published as one immutable snapshot so the
    ///     dispatcher thread reads a consistent view while a redeploy rebuilds it.
    /// </summary>
    /// <param name="ByRawPath">RawPath → definition; the v3 resolver's backing table.</param>
    /// <param name="ByMetric">
    ///     <c>(group, node, device?, metricName)</c> → every definition bound to it. A <i>list</i>,
    ///     because two raw tags may legitimately project the same Sparkplug metric.
    /// </param>
    /// <param name="ByScope">Scope → its definitions; the death/invalidation STALE fan-out's input.</param>
    /// <param name="EdgeNodes">
    ///     The distinct <c>(group, edgeNode)</c> pairs any authored tag names — the ingest filter, the
    ///     late-join rebirth set, and what bounds the per-node tracker map.
    /// </param>
    private sealed record AuthoredTable(
        FrozenDictionary<string, MqttTagDefinition> ByRawPath,
        FrozenDictionary<SparkplugMetricKey, MqttTagDefinition[]> ByMetric,
        FrozenDictionary<SparkplugScope, MqttTagDefinition[]> ByScope,
        FrozenSet<SparkplugNodeKey> EdgeNodes)
    {
        public static readonly AuthoredTable Empty =
            Build(new Dictionary<string, MqttTagDefinition>(StringComparer.Ordinal));

        /// <summary>Every authored scope belonging to one edge node — the NDEATH fan-out set.</summary>
        /// <param name="node">The edge node.</param>
        /// <returns>The scopes, node-own and device alike.</returns>
        public IEnumerable<SparkplugScope> ScopesUnder(SparkplugNodeKey node) =>
            ByScope.Keys.Where(s => s.IsUnder(node.GroupId, node.EdgeNodeId));

        /// <summary>Builds the snapshot from the mapped definitions.</summary>
        /// <param name="byRawPath">The mapped definitions, keyed by RawPath.</param>
        /// <returns>The immutable snapshot.</returns>
        public static AuthoredTable Build(IReadOnlyDictionary<string, MqttTagDefinition> byRawPath)
        {
            var byMetric = new Dictionary<SparkplugMetricKey, List<MqttTagDefinition>>();
            var byScope = new Dictionary<SparkplugScope, List<MqttTagDefinition>>();
            var nodes = new HashSet<SparkplugNodeKey>();

            foreach (var def in byRawPath.Values)
            {
                // The factory guarantees all three; a definition that reached here without them could
                // never be routed, so it is skipped rather than indexed under a nonsense key.
                if (def.GroupId is not { } group || def.EdgeNodeId is not { } node || def.MetricName is not { } metric)
                {
                    continue;
                }

                var scope = new SparkplugScope(group, node, def.DeviceId);
                nodes.Add(new SparkplugNodeKey(group, node));

                if (!byMetric.TryGetValue(new SparkplugMetricKey(group, node, def.DeviceId, metric), out var metricList))
                {
                    byMetric[new SparkplugMetricKey(group, node, def.DeviceId, metric)] = metricList = [];
                }

                metricList.Add(def);

                if (!byScope.TryGetValue(scope, out var scopeList))
                {
                    byScope[scope] = scopeList = [];
                }

                scopeList.Add(def);
            }

            return new AuthoredTable(
                byRawPath.ToFrozenDictionary(StringComparer.Ordinal),
                byMetric.ToFrozenDictionary(kv => kv.Key, kv => kv.Value.ToArray()),
                byScope.ToFrozenDictionary(kv => kv.Key, kv => kv.Value.ToArray()),
                nodes.ToFrozenSet());
        }
    }
}

/// <summary>
///     One edge node's identity — the unit Sparkplug sequences messages by, and the unit a rebirth NCMD
///     is addressed to.
/// </summary>
/// <param name="GroupId">The Sparkplug group id.</param>
/// <param name="EdgeNodeId">The Sparkplug edge-node id.</param>
public readonly record struct SparkplugNodeKey(string GroupId, string EdgeNodeId)
{
    /// <inheritdoc/>
    public override string ToString() => $"{GroupId}/{EdgeNodeId}";
}

/// <summary>
///     An authored tag's Sparkplug binding key: the stable tuple a tag is resolved by, across every
///     rebirth. The metric <b>name</b> is part of it; the per-birth alias never is.
/// </summary>
/// <param name="GroupId">The Sparkplug group id.</param>
/// <param name="EdgeNodeId">The Sparkplug edge-node id.</param>
/// <param name="DeviceId">The Sparkplug device id, or <see langword="null"/> for a node-level metric.</param>
/// <param name="MetricName">The stable metric name.</param>
public readonly record struct SparkplugMetricKey(
    string GroupId,
    string EdgeNodeId,
    string? DeviceId,
    string MetricName)
{
    /// <inheritdoc/>
    public override string ToString() =>
        DeviceId is null
            ? $"{GroupId}/{EdgeNodeId}:{MetricName}"
            : $"{GroupId}/{EdgeNodeId}/{DeviceId}:{MetricName}";
}

/// <summary>An observed Sparkplug primary-host STATE.</summary>
/// <param name="HostId">The host id the STATE topic named.</param>
/// <param name="Online">Whether the host declared itself online.</param>
/// <param name="ObservedUtc">When this driver observed it.</param>
public sealed record SparkplugHostState(string HostId, bool Online, DateTime ObservedUtc);

/// <summary>Event payload for <see cref="SparkplugIngestor.HostStateObserved"/>.</summary>
/// <param name="state">The observed state.</param>
public sealed class SparkplugHostStateEventArgs(SparkplugHostState state) : EventArgs
{
    /// <summary>The observed state.</summary>
    public SparkplugHostState State { get; } = state;
}

/// <summary>
///     Event payload for <see cref="SparkplugIngestor.BirthObserved"/> — the Task-22 rediscovery seam.
/// </summary>
/// <param name="scope">The scope whose birth certificate was applied.</param>
/// <param name="metricNames">The named metrics the birth declared, in wire order.</param>
public sealed class SparkplugBirthObservedEventArgs(SparkplugScope scope, IReadOnlyList<string> metricNames)
    : EventArgs
{
    /// <summary>The scope whose birth certificate was applied.</summary>
    public SparkplugScope Scope { get; } = scope;

    /// <summary>The named metrics the birth declared.</summary>
    public IReadOnlyList<string> MetricNames { get; } = metricNames;
}

/// <summary>
///     Coerces a Sparkplug wire value — <b>after</b>
///     <see cref="SparkplugMetricBinding.Reinterpret"/> has undone the two's-complement encoding — onto
///     an authored tag's <see cref="DriverDataType"/>.
/// </summary>
/// <remarks>
///     <para>
///         <b>A value that does not fit is refused, never silently converted.</b> Same rule the
///         plain-MQTT path follows: a wrong value delivered at Good quality is worse than no value.
///         The one deliberate latitude is an <i>exactly integral</i> real (<c>5.0</c>) for an integer
///         tag — edge gateways emit those routinely — while a genuinely fractional one is refused
///         rather than rounded.
///     </para>
///     <para>
///         <b>Sparkplug <c>DateTime</c> is epoch milliseconds</b>, carried in the unsigned 64-bit
///         field, not an ISO string. Reading it as a number and calling it a timestamp is the whole
///         conversion.
///     </para>
/// </remarks>
internal static class SparkplugValueCoercion
{
    /// <summary>Coerces <paramref name="raw"/> onto <paramref name="target"/>.</summary>
    /// <param name="raw">The reinterpreted wire value.</param>
    /// <param name="target">The tag's effective data type.</param>
    /// <param name="value">The coerced value when this returns <see langword="true"/>.</param>
    /// <returns><see langword="true"/> when the value fits the target type.</returns>
    public static bool TryCoerce(object? raw, DriverDataType target, out object? value)
    {
        value = null;
        if (raw is null)
        {
            return false;
        }

        try
        {
            switch (target)
            {
                case DriverDataType.String:
                case DriverDataType.Reference:
                    // Bytes/File metrics map to String per the datatype map; base64 is the v1 fallback
                    // the design names, and it round-trips where a lossy UTF-8 decode would not.
                    value = raw switch
                    {
                        string s => s,
                        byte[] bytes => Convert.ToBase64String(bytes),
                        bool b => b ? "true" : "false",
                        _ => Convert.ToString(raw, CultureInfo.InvariantCulture),
                    };
                    return value is not null;

                case DriverDataType.Boolean:
                    return TryCoerceBoolean(raw, out value);

                case DriverDataType.Int16:
                case DriverDataType.Int32:
                case DriverDataType.Int64:
                case DriverDataType.UInt16:
                case DriverDataType.UInt32:
                case DriverDataType.UInt64:
                    return TryCoerceInteger(raw, target, out value);

                case DriverDataType.Float32:
                    if (raw is string f32Text)
                    {
                        return float.TryParse(f32Text, NumberStyles.Float, CultureInfo.InvariantCulture, out var pf)
                            && Assign(pf, out value);
                    }

                    var asSingle = Convert.ToSingle(raw, CultureInfo.InvariantCulture);

                    // Convert.ToSingle SATURATES rather than throwing: a Double of 1e300 onto a Float32
                    // tag becomes +Infinity, which would publish at GOOD quality as a number no gauge
                    // can render and no comparison behaves sensibly against. A source that was itself
                    // infinite is passed through — that is the publisher's own value, not our overflow.
                    if (!float.IsFinite(asSingle) && IsFiniteSource(raw))
                    {
                        return false;
                    }

                    value = asSingle;
                    return true;

                case DriverDataType.Float64:
                    if (raw is string f64Text)
                    {
                        return double.TryParse(f64Text, NumberStyles.Float, CultureInfo.InvariantCulture, out var pd)
                            && Assign(pd, out value);
                    }

                    var asDouble = Convert.ToDouble(raw, CultureInfo.InvariantCulture);
                    if (!double.IsFinite(asDouble) && IsFiniteSource(raw))
                    {
                        return false;
                    }

                    value = asDouble;
                    return true;

                case DriverDataType.DateTime:
                    return TryCoerceDateTime(raw, out value);

                default:
                    return false;
            }
        }
        catch (Exception ex) when (ex is OverflowException or FormatException or InvalidCastException)
        {
            // Out of range, unparseable, or a wire type that cannot become the target at all. All three
            // are "does not fit", which is a Bad status on one tag — never an exception on the
            // dispatcher thread.
            value = null;
            return false;
        }
    }

    private static bool Assign<T>(T parsed, out object? value)
    {
        value = parsed;
        return true;
    }

    /// <summary>
    ///     Whether the wire value was itself a finite number — i.e. whether a non-finite result is
    ///     <i>our</i> narrowing overflow rather than the publisher's own Infinity/NaN.
    /// </summary>
    private static bool IsFiniteSource(object raw) => raw switch
    {
        float f => float.IsFinite(f),
        double d => double.IsFinite(d),
        _ => true,
    };

    private static bool TryCoerceBoolean(object raw, out object? value)
    {
        switch (raw)
        {
            case bool b:
                value = b;
                return true;
            case string s when bool.TryParse(s.Trim(), out var parsed):
                value = parsed;
                return true;
            case string s:
                value = null;
                return long.TryParse(s.Trim(), NumberStyles.Integer, CultureInfo.InvariantCulture, out var n)
                    && Assign(n != 0L, out value);
            case byte[]:
                value = null;
                return false;
            default:
                // Any numeric wire type: nonzero is true, matching the plain-mode path's rule.
                value = Convert.ToDouble(raw, CultureInfo.InvariantCulture) != 0d;
                return true;
        }
    }

    private static bool TryCoerceInteger(object raw, DriverDataType target, out object? value)
    {
        value = null;

        // Reals (and textual reals) must be exactly integral: rounding 5.5 to 5 or 6 is a wrong value,
        // which is worse than no value. Parsed as decimal so the test and the conversion stay exact
        // across the whole Int64/UInt64 range, where double loses precision above 2^53.
        if (raw is float or double or decimal or string)
        {
            var text = raw as string ?? Convert.ToString(raw, CultureInfo.InvariantCulture);
            if (text is null
                || !decimal.TryParse(text, NumberStyles.Float, CultureInfo.InvariantCulture, out var dec)
                || dec != decimal.Truncate(dec))
            {
                return false;
            }

            raw = dec;
        }

        value = target switch
        {
            DriverDataType.Int16 => Convert.ToInt16(raw, CultureInfo.InvariantCulture),
            DriverDataType.Int32 => Convert.ToInt32(raw, CultureInfo.InvariantCulture),
            DriverDataType.Int64 => Convert.ToInt64(raw, CultureInfo.InvariantCulture),
            DriverDataType.UInt16 => Convert.ToUInt16(raw, CultureInfo.InvariantCulture),
            DriverDataType.UInt32 => Convert.ToUInt32(raw, CultureInfo.InvariantCulture),
            DriverDataType.UInt64 => Convert.ToUInt64(raw, CultureInfo.InvariantCulture),
            _ => null,
        };

        return value is not null;
    }

    private static bool TryCoerceDateTime(object raw, out object? value)
    {
        value = null;

        switch (raw)
        {
            case DateTime dt:
                value = dt;
                return true;

            case string s:
                return DateTime.TryParse(
                        s,
                        CultureInfo.InvariantCulture,
                        DateTimeStyles.RoundtripKind | DateTimeStyles.AdjustToUniversal | DateTimeStyles.AssumeUniversal,
                        out var parsed)
                    && Assign(parsed, out value);

            case bool:
            case byte[]:
                return false;

            default:
                // Sparkplug DateTime rides as epoch milliseconds in the unsigned 64-bit value field.
                var ms = Convert.ToInt64(raw, CultureInfo.InvariantCulture);
                if (ms < DateTimeOffset.MinValue.ToUnixTimeMilliseconds()
                    || ms > DateTimeOffset.MaxValue.ToUnixTimeMilliseconds())
                {
                    return false;
                }

                value = DateTimeOffset.FromUnixTimeMilliseconds(ms).UtcDateTime;
                return true;
        }
    }
}