feat(mqtt): Sparkplug ingest state machine (birth/alias/seq/rebirth/death→stale)

Task 21 — the design doc's §3.6 correctness core. `SparkplugIngestor` owns the authored `(group, node, device?, metric) → RawPath` index, the live `BirthCache`, one `SequenceTracker` per authored edge node, and the rebirth policy; it routes NBIRTH/DBIRTH/NDATA/DDATA/NDEATH/DDEATH/STATE into `OnDataChange` + `LastValueCache`, or into a bounded, per-node-debounced rebirth NCMD. Invariants held (and each pinned by a falsifiability control): - Published reference is the RawPath, never a composed Sparkplug reference — `DriverHostActor`'s dual-namespace fan-out is RawPath-keyed. - Tags bind by stable metric NAME; the alias is only the per-birth lookup, so an alias reused across a rebirth cannot mis-route. - Every value goes through `SparkplugMetricBinding.Reinterpret` before publish. - NDEATH never reaches `SequenceTracker.Accept` (it carries no seq); it is tied to its birth by bdSeq instead. - An invalid payload mutates nothing — no tracker, no alias table, no eviction. - `HandleMessage` never throws on MQTTnet's dispatcher thread. Supporting changes: - `MqttTagDefinitionFactory.FromSparkplugTagConfig` — the driver had no way to parse the Sparkplug descriptor keys at all; the P1 stub fields on `MqttTagDefinition` were never populated. Mode selects the parser, never a blob heuristic. `dataType` becomes optional in the Sparkplug shape (the birth declares it), recorded via the new `DataTypeAuthored` flag. - `MqttConnection` implements `IMqttPublishTransport` (bounded, refusal throws). - `MqttDriver` dispatches every capability by mode, subscribes `spBv1.0/{group}/#` (+ the configured STATE topic) once at connect, and re-subscribes + requests a late-join rebirth on reconnect. `IngestIdentity` now names `MqttSparkplugOptions`, closing the silent same-ingest gap its own ⚠️ comment warned about. Primary-host STATE *publishing* is explicitly out of scope and warned about at construction: it needs the retained-OFFLINE Last Will set at CONNECT time, and shipping the ONLINE half alone is worse than shipping neither. 58 new tests (MQTT suite 455 → 513, all green). Claude-Session: https://claude.ai/code/session_01GASWkNEi68FSCtvr6rLoEW
2026-07-24 22:11:41 -04:00
parent 31a98a1bf5
commit 8538fb798b
6 changed files with 3456 additions and 42 deletions
@@ -9,12 +9,20 @@ namespace ZB.MOM.WW.OtOpcUa.Driver.Mqtt;
 ///     See <c>docs/plans/2026-07-15-mqtt-sparkplug-driver-design.md</c> §5.2 / §5.3.
 ///     <para>
 ///         The record carries <b>both</b> ingest shapes: the Plain-MQTT fields
-///         (<see cref="Topic"/> … <see cref="RetainSeed"/>), populated in P1, and the Sparkplug B
-///         descriptor fields (<see cref="GroupId"/>, <see cref="EdgeNodeId"/>,
-///         <see cref="DeviceId"/>, <see cref="MetricName"/>), which are deliberate <b>stubs</b> in P1
-///         — always <see langword="null"/> until the P2 tasks (15–26) teach the parser to read them.
-///         A Sparkplug tag is resolved by the stable <c>(group, node, device, metricName)</c> tuple,
-///         never by the per-birth metric alias.
+///         (<see cref="Topic"/> … <see cref="RetainSeed"/>), populated by
+///         <see cref="MqttTagDefinitionFactory.FromTagConfig"/>, and the Sparkplug B descriptor fields
+///         (<see cref="GroupId"/>, <see cref="EdgeNodeId"/>, <see cref="DeviceId"/>,
+///         <see cref="MetricName"/>), populated by
+///         <see cref="MqttTagDefinitionFactory.FromSparkplugTagConfig"/> (Task 21). A Sparkplug tag is
+///         resolved by the stable <c>(group, node, device, metricName)</c> tuple, <b>never</b> by the
+///         per-birth metric alias — an alias may be reused across a rebirth for a different metric.
+///     </para>
+///     <para>
+///         <b>Which factory runs is chosen by the driver's <see cref="MqttMode"/>, never sniffed from
+///         the blob.</b> A blob carrying both shapes' keys is legal (the AdminUI editor preserves
+///         unknown keys through a load→save, so a tag retyped from Plain to Sparkplug keeps its old
+///         <c>topic</c>), and a heuristic would make the same authored tag mean two different things
+///         depending on which key happened to survive.
 ///     </para>
 /// </summary>
 /// <param name="Name">
@@ -44,15 +52,23 @@ namespace ZB.MOM.WW.OtOpcUa.Driver.Mqtt;
 ///     Whether the broker's retained message for <see cref="Topic"/> seeds this tag's initial value
 ///     on subscribe (the OPC UA initial-data convention). Defaults to <see langword="true"/>.
 /// </param>
-/// <param name="GroupId">P2 stub — the Sparkplug group id. Always <see langword="null"/> in P1.</param>
-/// <param name="EdgeNodeId">P2 stub — the Sparkplug edge-node id. Always <see langword="null"/> in P1.</param>
+/// <param name="DataTypeAuthored">
+///     Whether <paramref name="DataType"/> came from an authored <c>dataType</c> key or is merely this
+///     record's default. Load-bearing in <b>Sparkplug</b> mode only, where <c>dataType</c> is optional:
+///     a Sparkplug metric's type is declared by its own birth certificate, so an unauthored tag must
+///     take the type the NBIRTH/DBIRTH declared rather than silently coercing every value to the
+///     <see cref="DriverDataType.String"/> default. Both factories set it from the key's presence so it
+///     means the same thing in either mode; nothing on the Plain ingest path reads it.
+/// </param>
+/// <param name="GroupId">The Sparkplug group id; <see langword="null"/> for a Plain-mode tag.</param>
+/// <param name="EdgeNodeId">The Sparkplug edge-node id; <see langword="null"/> for a Plain-mode tag.</param>
 /// <param name="DeviceId">
-///     P2 stub — the Sparkplug device id (absent for node-level metrics). Always
-///     <see langword="null"/> in P1.
+///     The Sparkplug device id, or <see langword="null"/> for a metric published by the edge node
+///     itself (and for every Plain-mode tag).
 /// </param>
 /// <param name="MetricName">
-///     P2 stub — the Sparkplug metric name, the tag's stable identity across rebirths. Always
-///     <see langword="null"/> in P1.
+///     The Sparkplug metric name — the tag's stable identity across rebirths, and the key the ingest
+///     state machine binds by. <see langword="null"/> for a Plain-mode tag.
 /// </param>
 public sealed record MqttTagDefinition(
    string Name,
@@ -62,6 +78,7 @@ public sealed record MqttTagDefinition(
    DriverDataType DataType,
    int? Qos,
    bool RetainSeed,
+    bool DataTypeAuthored = true,
    string? GroupId = null,
    string? EdgeNodeId = null,
    string? DeviceId = null,
@@ -20,9 +20,13 @@ namespace ZB.MOM.WW.OtOpcUa.Driver.Mqtt;
 ///         deploy instead of silently going dark at runtime.
 ///     </para>
 ///     <para>
-///         Sparkplug-descriptor parsing (<c>groupId</c> / <c>edgeNodeId</c> / <c>deviceId</c> /
-///         <c>metricName</c>) is a deliberate P1 stub — the fields exist on
-///         <see cref="MqttTagDefinition"/> but are never populated here until the P2 tasks fill them.
+///         <b>Two runtime entry points, one per ingest shape</b> — <see cref="FromTagConfig"/> (Plain)
+///         and <see cref="FromSparkplugTagConfig"/> (Sparkplug B). The caller picks by the driver's
+///         <see cref="MqttMode"/>; neither sniffs the blob. That is deliberate: the AdminUI editor
+///         preserves unknown keys through a load→save, so a tag retyped from Plain to Sparkplug keeps
+///         its old <c>topic</c> and a presence heuristic would make one authored blob mean two
+///         different things depending on which key happened to survive. The <i>driver's</i> mode is
+///         the single authority.
 ///     </para>
 ///     <para>
 ///         <b>No <c>ToTagConfig</c> inverse — a deliberate YAGNI call.</b> The six sibling factories
@@ -80,6 +84,7 @@ public static class MqttTagDefinitionFactory
            // (→ BadNodeIdUnknown) instead of silently defaulting to a wrong-typed Good.
            if (!TagConfigJson.TryReadEnumStrict(root, "payloadFormat", MqttPayloadFormat.Json, out var payloadFormat))
                return false;
+            var dataTypeAuthored = root.TryGetProperty("dataType", out _);
            if (!TagConfigJson.TryReadEnumStrict(root, "dataType", DriverDataType.String, out var dataType))
                return false;

@@ -98,7 +103,94 @@ public static class MqttTagDefinitionFactory
                JsonPath: jsonPath,
                DataType: dataType,
                Qos: qos,
-                RetainSeed: ReadBoolOrDefault(root, "retainSeed", defaultValue: true));
+                RetainSeed: ReadBoolOrDefault(root, "retainSeed", defaultValue: true),
+                DataTypeAuthored: dataTypeAuthored);
+            return true;
+        }
+        catch (JsonException) { return false; }
+        catch (FormatException) { return false; }
+        catch (InvalidOperationException) { return false; }
+    }
+
+    /// <summary>
+    ///     Maps an authored <c>TagConfig</c> object to a <b>Sparkplug B</b> definition keyed by
+    ///     <paramref name="rawPath"/>. The tag's binding identity is the stable
+    ///     <c>(groupId, edgeNodeId, deviceId?, metricName)</c> tuple; <c>deviceId</c> is optional (absent
+    ///     means the metric is published by the edge node itself), and there is <b>no</b> <c>topic</c> —
+    ///     a Sparkplug driver subscribes one group-wide filter and routes by the decoded topic + birth
+    ///     certificate, never by a per-tag topic. Never throws.
+    /// </summary>
+    /// <remarks>
+    ///     <para>
+    ///         <b><c>dataType</c> is OPTIONAL here, unlike in Plain mode.</b> A Sparkplug metric declares
+    ///         its own datatype in its birth certificate, so an unauthored tag legitimately takes the
+    ///         type the birth declares — that is the whole point of Task 22's <c>UntilStable</c>
+    ///         discovery. It is still read <b>strictly</b> when present (a typo'd value rejects the tag,
+    ///         exactly as in Plain mode) and the outcome is recorded on
+    ///         <see cref="MqttTagDefinition.DataTypeAuthored"/> so the ingest path can tell "the operator
+    ///         declared String" from "nobody declared anything and the record defaulted".
+    ///     </para>
+    ///     <para>
+    ///         <b>Plain-shape keys are read but not required.</b> <c>topic</c>/<c>payloadFormat</c>/
+    ///         <c>jsonPath</c> are meaningless to the Sparkplug ingest path and are deliberately NOT
+    ///         validated — a blob retyped in the editor keeps them, and rejecting the tag for a stale
+    ///         leftover key would take a correctly-authored Sparkplug tag dark with no stated cause.
+    ///     </para>
+    /// </remarks>
+    /// <param name="tagConfig">The authored equipment-tag TagConfig JSON.</param>
+    /// <param name="rawPath">The tag's RawPath — becomes the definition's identity (<c>Name</c>).</param>
+    /// <param name="def">The mapped definition when this returns <see langword="true"/>.</param>
+    /// <returns><see langword="true"/> when the blob carries a usable Sparkplug descriptor.</returns>
+    public static bool FromSparkplugTagConfig(string tagConfig, string rawPath, out MqttTagDefinition def)
+    {
+        def = null!;
+        if (string.IsNullOrWhiteSpace(tagConfig) || tagConfig[0] != '{') return false;
+        try
+        {
+            using var doc = JsonDocument.Parse(tagConfig);
+            var root = doc.RootElement;
+            if (root.ValueKind != JsonValueKind.Object) return false;
+
+            // The binding tuple. Without all three of these there is nothing a birth certificate could
+            // ever bind the tag to, so an absent/blank value is a hard reject (→ BadNodeIdUnknown)
+            // rather than a definition that can never resolve and reports nothing about why.
+            var groupId = ReadString(root, "groupId");
+            var edgeNodeId = ReadString(root, "edgeNodeId");
+            var metricName = ReadString(root, "metricName");
+            if (string.IsNullOrWhiteSpace(groupId)
+                || string.IsNullOrWhiteSpace(edgeNodeId)
+                || string.IsNullOrWhiteSpace(metricName))
+            {
+                return false;
+            }
+
+            // Optional: a node-level metric has no device segment. Blank normalises to absent so
+            // "deviceId":"" and an omitted key describe the same scope rather than two.
+            var deviceId = ReadString(root, "deviceId");
+            if (string.IsNullOrWhiteSpace(deviceId)) deviceId = null;
+
+            // Strict, but only when present — see the remarks.
+            var dataTypeAuthored = root.TryGetProperty("dataType", out _);
+            if (!TagConfigJson.TryReadEnumStrict(root, "dataType", DriverDataType.String, out var dataType))
+                return false;
+
+            if (!TryReadQosStrict(root, out var qos)) return false;
+
+            def = new MqttTagDefinition(
+                Name: rawPath,
+                // Plain-shape fields carried through verbatim; the Sparkplug ingest path reads none of
+                // them, and a retyped blob's leftovers must not change what the tag means.
+                Topic: ReadString(root, "topic"),
+                PayloadFormat: MqttPayloadFormat.Json,
+                JsonPath: RootJsonPath,
+                DataType: dataType,
+                Qos: qos,
+                RetainSeed: ReadBoolOrDefault(root, "retainSeed", defaultValue: true),
+                DataTypeAuthored: dataTypeAuthored,
+                GroupId: groupId,
+                EdgeNodeId: edgeNodeId,
+                DeviceId: deviceId,
+                MetricName: metricName);
            return true;
        }
        catch (JsonException) { return false; }
@@ -237,10 +237,17 @@ public sealed class MqttConnectRejectedException : Exception
 ///         socket no later dispose could ever reach. Now the losing side of that race disposes the
 ///         client it created and reports <see cref="ObjectDisposedException"/>.
 ///     </para>
-///     Sparkplug rebirth-on-reconnect (Task 21, which hangs off <see cref="Reconnected"/>) is
-///     deliberately not implemented here.
+///     <para>
+///         <b>The publish leg is deliberately narrow.</b> <see cref="PublishAsync"/> exists to satisfy
+///         <see cref="Sparkplug.IMqttPublishTransport"/> — whose entire production caller is
+///         <see cref="Sparkplug.RebirthRequester"/> — not to make this a general MQTT publisher. This
+///         driver has no <c>IWritable</c> leg in v1, so a Sparkplug rebirth NCMD is the <i>only</i>
+///         outbound application message it ever sends. Sparkplug rebirth-on-reconnect itself is decided
+///         by <see cref="Sparkplug.SparkplugIngestor"/> hanging off <see cref="Reconnected"/>; this type
+///         only carries the bytes.
+///     </para>
 /// </remarks>
-public sealed class MqttConnection : IAsyncDisposable, IMqttSubscribeTransport
+public sealed class MqttConnection : IAsyncDisposable, IMqttSubscribeTransport, Sparkplug.IMqttPublishTransport
 {
    private readonly string _driverId;

@@ -994,6 +1001,86 @@ public sealed class MqttConnection : IAsyncDisposable, IMqttSubscribeTransport
        return outcomes;
    }

+    /// <summary>
+    ///     Publishes one application message, bounded by
+    ///     <see cref="MqttDriverOptions.ConnectTimeoutSeconds"/> so a broker that accepts PUBLISH and
+    ///     never completes it cannot park the caller — the same frozen-peer rule every other wait here
+    ///     follows.
+    /// </summary>
+    /// <remarks>
+    ///     <para>
+    ///         <b>A refused PUBLISH throws.</b> MQTTnet reports a broker rejection as a
+    ///         <i>result</i> (as it does for CONNACK — see <see cref="MqttConnectRejectedException"/>),
+    ///         so a caller that treated "did not throw" as "published" would silently drop a rebirth
+    ///         NCMD against a broker whose ACL denies the NCMD topic and then wait forever for the
+    ///         birth it never asked for.
+    ///     </para>
+    ///     <para>
+    ///         <b>Publishing on a down session is refused up front</b> rather than left to MQTTnet's
+    ///         own exception, so the message names the driver and the broker. A rebirth request lost to
+    ///         a reconnect is not a fault: the reconnect path re-requests one itself.
+    ///     </para>
+    /// </remarks>
+    /// <param name="topic">The concrete topic to publish on.</param>
+    /// <param name="payload">The message body.</param>
+    /// <param name="qos">Requested QoS, 0–2; out-of-range values clamp.</param>
+    /// <param name="retain">The MQTT <c>retain</c> flag.</param>
+    /// <param name="cancellationToken">Caller cancellation; linked with the publish deadline.</param>
+    /// <returns>A task that completes when the broker has accepted the message.</returns>
+    /// <exception cref="TimeoutException">The publish deadline elapsed.</exception>
+    /// <exception cref="OperationCanceledException"><paramref name="cancellationToken"/> was cancelled.</exception>
+    /// <exception cref="ObjectDisposedException">The connection was disposed.</exception>
+    /// <exception cref="InvalidOperationException">There is no established session, or the broker refused the message.</exception>
+    public async Task PublishAsync(
+        string topic,
+        byte[] payload,
+        int qos,
+        bool retain,
+        CancellationToken cancellationToken)
+    {
+        ArgumentException.ThrowIfNullOrEmpty(topic);
+        ArgumentNullException.ThrowIfNull(payload);
+        ObjectDisposedException.ThrowIf(Disposed, this);
+
+        var client = _client;
+        if (client is null || !client.IsConnected)
+        {
+            throw new InvalidOperationException(
+                $"MQTT driver '{_driverId}': cannot publish '{topic}' — there is no established session to "
+                + $"{_options.Host}:{_options.Port}.");
+        }
+
+        var message = new MqttApplicationMessageBuilder()
+            .WithTopic(topic)
+            .WithPayload(payload)
+            .WithQualityOfServiceLevel(MapQos(qos))
+            .WithRetainFlag(retain)
+            .Build();
+
+        using var deadline = new CancellationTokenSource(TimeSpan.FromSeconds(_options.ConnectTimeoutSeconds));
+        using var linked = CancellationTokenSource.CreateLinkedTokenSource(
+            cancellationToken,
+            deadline.Token,
+            _lifetimeCts.Token);
+
+        MqttClientPublishResult result;
+        try
+        {
+            result = await client.PublishAsync(message, linked.Token).ConfigureAwait(false);
+        }
+        catch (Exception ex)
+        {
+            throw Classify(ex, cancellationToken, deadline, operation: "publish");
+        }
+
+        if (result is not null && !result.IsSuccess)
+        {
+            throw new InvalidOperationException(
+                $"MQTT driver '{_driverId}': broker at {_options.Host}:{_options.Port} refused PUBLISH "
+                + $"'{topic}': {result.ReasonCode}.");
+        }
+    }
+
    /// <summary>Maps a configured QoS integer onto MQTTnet's enum, clamping an out-of-range value.</summary>
    private static MqttQualityOfServiceLevel MapQos(int qos) => qos switch
    {
@@ -1,6 +1,7 @@
 using System.Text.Json;
 using Microsoft.Extensions.Logging;
 using ZB.MOM.WW.OtOpcUa.Core.Abstractions;
+using ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Sparkplug;

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

@@ -22,11 +23,21 @@ namespace ZB.MOM.WW.OtOpcUa.Driver.Mqtt;
 ///         synchronously, so re-running discovery on a timer can only produce the same tree.
 ///     </para>
 ///     <para>
-///         <b>P1 scope — no Sparkplug.</b> <see cref="IRediscoverable"/> is implemented but
-///         <see cref="OnRediscoveryNeeded"/> never fires in <see cref="MqttMode.Plain"/>: the
-///         authored set only changes by redeploy. Sparkplug B flips the policy to
-///         <see cref="DiscoveryRediscoverPolicy.UntilStable"/> and raises the event on DBIRTH — that
-///         is a later task, and the seam here (<see cref="RaiseRediscoveryNeeded"/>) exists for it.
+///         <b>Two ingest paths, one driver.</b> <see cref="MqttMode.Plain"/> routes through
+///         <see cref="MqttSubscriptionManager"/> (authored topics); <see cref="MqttMode.SparkplugB"/>
+///         routes through <see cref="SparkplugIngestor"/> (one group-wide filter, birth certificates,
+///         alias resolution, seq gaps, rebirth NCMDs). Exactly one is live, selected by
+///         <see cref="MqttDriverOptions.Mode"/> and rebuilt whenever
+///         <see cref="ReinitializeAsync"/> changes anything either captures. They share the
+///         driver-owned <see cref="LastValueCache"/>, so <c>IReadable</c> reads one store either way.
+///     </para>
+///     <para>
+///         <b><see cref="OnRediscoveryNeeded"/> still never fires.</b> Sparkplug B is the mode that
+///         eventually raises it — a DBIRTH can introduce metrics the previous birth did not carry —
+///         but the policy flip to <see cref="DiscoveryRediscoverPolicy.UntilStable"/> and the wiring
+///         from <see cref="SparkplugIngestor.BirthObserved"/> to
+///         <see cref="RaiseRediscoveryNeeded"/> are Task 22. The seams exist on both sides; nothing
+///         connects them yet.
 ///     </para>
 ///     <para>
 ///         <b>Composition order is load-bearing.</b> <see cref="MqttSubscriptionManager.AttachTo"/>
@@ -83,6 +94,14 @@ public sealed class MqttDriver
    private MqttDriverOptions _options;
    private MqttSubscriptionManager _subscriptions;

+    /// <summary>
+    ///     The Sparkplug B ingest path, or <see langword="null"/> in <see cref="MqttMode.Plain"/>.
+    ///     Exactly one of this and <see cref="_subscriptions"/> is fed authored tags and attached to the
+    ///     connection; the other exists but is never registered against, so it can neither route a
+    ///     message nor answer a resolve.
+    /// </summary>
+    private SparkplugIngestor? _sparkplug;
+
    /// <summary>
    ///     RawPaths of the currently registered authored tags, in authoring order — the discovery
    ///     enumeration set. The manager resolves a RawPath to its definition but does not enumerate,
@@ -115,6 +134,7 @@ public sealed class MqttDriver
        _driverInstanceId = driverInstanceId;
        _logger = logger;
        _subscriptions = CreateSubscriptionManager(options);
+        _sparkplug = CreateSparkplugIngestor(options);
        RegisterAuthoredTags(options);
    }

@@ -141,6 +161,17 @@ public sealed class MqttDriver
    /// </summary>
    internal MqttSubscriptionManager Subscriptions => _subscriptions;

+    /// <summary>
+    ///     The Sparkplug B ingest path, or <see langword="null"/> outside
+    ///     <see cref="MqttMode.SparkplugB"/>. Internal: the shell tests drive
+    ///     <see cref="SparkplugIngestor.Dispatch"/> to simulate an edge node without a broker, and
+    ///     Tasks 22/23 read its <see cref="SparkplugIngestor.Births"/>.
+    /// </summary>
+    internal SparkplugIngestor? Sparkplug => _sparkplug;
+
+    /// <summary>Whether this driver instance ingests Sparkplug B rather than plain topics.</summary>
+    private bool IsSparkplug => _options.Mode == MqttMode.SparkplugB;
+
    /// <summary>
    ///     Test seam: awaited inside <see cref="InitializeCoreAsync"/> <b>while the lifecycle gate is
    ///     held</b>, immediately before the broker connect. Lets a test park a lifecycle operation
@@ -360,7 +391,7 @@ public sealed class MqttDriver
        var folder = builder.Folder("Mqtt", "Mqtt");
        foreach (var rawPath in _authoredRawPaths)
        {
-            if (!_subscriptions.TryResolve(rawPath, out var def))
+            if (!TryResolve(rawPath, out var def))
            {
                continue;
            }
@@ -388,11 +419,24 @@ public sealed class MqttDriver
        IReadOnlyList<string> fullReferences,
        TimeSpan publishingInterval,
        CancellationToken cancellationToken)
-        => _subscriptions.SubscribeAsync(fullReferences, publishingInterval, cancellationToken);
+        => _sparkplug is { } sparkplug
+            ? sparkplug.SubscribeAsync(fullReferences, publishingInterval, cancellationToken)
+            : _subscriptions.SubscribeAsync(fullReferences, publishingInterval, cancellationToken);

    /// <inheritdoc />
    public Task UnsubscribeAsync(ISubscriptionHandle handle, CancellationToken cancellationToken)
-        => _subscriptions.UnsubscribeAsync(handle, cancellationToken);
+        => _sparkplug is { } sparkplug
+            ? sparkplug.UnsubscribeAsync(handle, cancellationToken)
+            : _subscriptions.UnsubscribeAsync(handle, cancellationToken);
+
+    /// <summary>Resolves a RawPath through whichever ingest path this driver's mode selected.</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 tag of the live ingest path.</returns>
+    private bool TryResolve(string rawPath, out MqttTagDefinition def)
+        => _sparkplug is { } sparkplug
+            ? sparkplug.TryResolve(rawPath, out def)
+            : _subscriptions.TryResolve(rawPath, out def);

    // ---- IReadable ----

@@ -417,7 +461,9 @@ public sealed class MqttDriver
        var results = new DataValueSnapshot[fullReferences.Count];
        for (var i = 0; i < fullReferences.Count; i++)
        {
-            results[i] = _subscriptions.Values.Read(fullReferences[i]);
+            // The driver-owned cache, not the live ingest path's: both are handed this same instance,
+            // and reading it directly means a mid-flight ingest rebuild can never blank a read.
+            results[i] = _values.Read(fullReferences[i]);
        }

        return Task.FromResult<IReadOnlyList<DataValueSnapshot>>(results);
@@ -539,9 +585,16 @@ public sealed class MqttDriver
            var connection = new MqttConnection(_options, _driverInstanceId, _logger);

            // MUST precede the connect: this wires message delivery AND the reconnect re-subscribe.
-            // The manager's Reconnected handler is passed through unwrapped on purpose — its
-            // throw-on-total-failure is what tears a deaf session down and retries it.
-            _subscriptions.AttachTo(connection);
+            // The handler is passed through unwrapped on purpose — its throw-on-total-failure is what
+            // tears a deaf session down and retries it.
+            if (_sparkplug is { } attaching)
+            {
+                attaching.AttachTo(connection);
+            }
+            else
+            {
+                _subscriptions.AttachTo(connection);
+            }

            if (BeforeConnectHookForTests is { } hook)
            {
@@ -561,6 +614,29 @@ public sealed class MqttDriver
                throw new ObjectDisposedException(nameof(MqttDriver));
            }

+            // Sparkplug subscribes ONCE, here, rather than on the OPC UA server's first subscribe: the
+            // driver must see birth certificates whether or not anything is subscribed yet, because a
+            // birth is where every alias and datatype comes from. This also issues the late-join
+            // rebirth (§3.6 #5) — the initial connect does not fire Reconnected.
+            //
+            // It runs BEFORE `_connection` is published, and disposes the session it was given if it
+            // fails. Publishing first would leave a live, supervised connection behind a throw: the
+            // driver-host resilience layer re-runs this method, the retry overwrites `_connection`, and
+            // the first session becomes an orphan no later dispose can reach — the exact leak
+            // MqttConnection's own remarks describe having already fixed once.
+            if (_sparkplug is { } sparkplug)
+            {
+                try
+                {
+                    await sparkplug.EstablishAsync(cancellationToken).ConfigureAwait(false);
+                }
+                catch
+                {
+                    await connection.DisposeAsync().ConfigureAwait(false);
+                    throw;
+                }
+            }
+
            _connection = connection;

            WriteHealth(new DriverHealth(DriverState.Healthy, connection.LastMessageUtc, null));
@@ -623,6 +699,7 @@ public sealed class MqttDriver
        if (!SameIngest(next, _options))
        {
            _subscriptions = CreateSubscriptionManager(next);
+            _sparkplug = CreateSparkplugIngestor(next);
        }

        _options = next;
@@ -645,10 +722,59 @@ public sealed class MqttDriver
        return manager;
    }

+    /// <summary>
+    ///     Builds the Sparkplug B ingest path for <paramref name="options"/>, or <see langword="null"/>
+    ///     outside <see cref="MqttMode.SparkplugB"/>. Shares the driver-owned
+    ///     <see cref="LastValueCache"/> for the same reason the plain manager does: a rebuild for
+    ///     changed settings must not blank every observed value.
+    /// </summary>
+    private SparkplugIngestor? CreateSparkplugIngestor(MqttDriverOptions options)
+    {
+        if (options.Mode != MqttMode.SparkplugB)
+        {
+            return null;
+        }
+
+        if (options.Sparkplug is { ActAsPrimaryHost: true })
+        {
+            // Loud, because the flag reads as if it does something. Being a Sparkplug Primary Host is a
+            // three-part contract — a retained ONLINE after CONNECT, a matching retained OFFLINE
+            // registered as the MQTT Last Will AT CONNECT TIME, and an explicit OFFLINE on clean
+            // shutdown — and edge nodes stop publishing while their host is offline. Shipping the
+            // ONLINE half without the Will is strictly worse than shipping neither: a crashed node
+            // would leave a retained ONLINE asserting forever that a dead host is alive, which is the
+            // exact state the mechanism exists to prevent. The Will belongs in
+            // MqttConnection.BuildClientOptions; until it is there this flag stays inert and said so.
+            _logger?.LogWarning(
+                "MQTT driver '{DriverId}': Sparkplug actAsPrimaryHost is set but primary-host STATE "
+                + "publishing is not implemented — this driver observes STATE and never publishes one. "
+                + "Edge nodes gated on this host id will not see it come online.",
+                _driverInstanceId);
+        }
+
+        var ingestor = new SparkplugIngestor(
+            options,
+            _driverInstanceId,
+            subscribeTransport: null,
+            publishTransport: null,
+            logger: _logger,
+            cache: _values,
+            maxPayloadBytes: options.MaxPayloadBytes);
+
+        // The published reference is the RawPath the ingestor already stamped on the args; the driver
+        // only re-raises with itself as sender.
+        ingestor.OnDataChange += (_, args) => OnDataChange?.Invoke(this, args);
+        return ingestor;
+    }
+
    /// <summary>Registers the authored raw tags wholesale and refreshes the discovery enumeration set.</summary>
    private void RegisterAuthoredTags(MqttDriverOptions options)
    {
-        var mapped = _subscriptions.Register(options.RawTags);
+        // Wholesale into the LIVE path only. Feeding Sparkplug tags to the plain parser would log a
+        // "did not map" warning for every one of them (they carry no topic) and vice versa.
+        var mapped = _sparkplug is { } sparkplug
+            ? sparkplug.Register(options.RawTags)
+            : _subscriptions.Register(options.RawTags);

        // Enumerate in authoring order; the manager owns which of these actually mapped, and
        // DiscoverAsync skips the rest.
@@ -676,16 +802,19 @@ public sealed class MqttDriver
    private static bool SameSession(MqttDriverOptions a, MqttDriverOptions b)
        => SessionIdentity(a).Equals(SessionIdentity(b));

-    /// <summary>Whether two option sets produce an identical <see cref="MqttSubscriptionManager"/>.</summary>
+    /// <summary>
+    ///     Whether two option sets produce an identical ingest path — <see cref="MqttSubscriptionManager"/>
+    ///     in Plain mode, <see cref="SparkplugIngestor"/> in Sparkplug B.
+    /// </summary>
    /// <remarks>
-    ///     ⚠️ <b>P2 (Sparkplug, tasks 21/22) must extend <see cref="IngestIdentity"/>.</b> It names
-    ///     only the settings the <b>P1</b> manager captures at construction — <c>Mode</c>,
-    ///     <c>Plain.DefaultQos</c>, <c>MaxPayloadBytes</c> — and deliberately no
-    ///     <see cref="MqttSparkplugOptions"/> field, because nothing reads one yet. The moment a
-    ///     Sparkplug handler captures its own options (group id, host id, rebirth policy, birth
-    ///     window) at construction, a delta changing one of them would be judged "same ingest",
-    ///     applied in place, and silently never reach the component that needed rebuilding — the
-    ///     driver would keep subscribing to the OLD group id and report perfect health.
+    ///     ⚠️ <b>Every setting an ingest path captures at construction MUST be named by
+    ///     <see cref="IngestIdentity"/>.</b> A setting that is captured but unnamed makes a delta
+    ///     changing it compare "same ingest", get applied in place, and silently never reach the
+    ///     component that needed rebuilding — the driver would keep subscribing to the OLD Sparkplug
+    ///     group id and report perfect health. Task 21 closed the Sparkplug half of that gap by naming
+    ///     the whole <see cref="MqttSparkplugOptions"/> record (it has value equality, so a member
+    ///     added there is covered automatically — which is the point of naming the record rather than
+    ///     enumerating its fields).
    /// </remarks>
    private static bool SameIngest(MqttDriverOptions a, MqttDriverOptions b)
        => IngestIdentity(a).Equals(IngestIdentity(b));
@@ -714,12 +843,18 @@ public sealed class MqttDriver
        Ingest = IngestIdentity(o),
    };

-    /// <summary>The settings <see cref="MqttSubscriptionManager"/> captures at construction.</summary>
+    /// <summary>The settings the live ingest path captures at construction.</summary>
+    /// <remarks>
+    ///     <see cref="MqttSparkplugOptions"/> is named as a whole record rather than field-by-field: it
+    ///     has structural value equality, so a member added to it in a later task is covered here with
+    ///     no edit — and the failure mode of forgetting one is silent (see <see cref="SameIngest"/>).
+    /// </remarks>
    private static object IngestIdentity(MqttDriverOptions o) => new
    {
        o.Mode,
        DefaultQos = o.Plain?.DefaultQos,
        o.MaxPayloadBytes,
+        o.Sparkplug,
    };

    /// <summary>Shared teardown for <see cref="ShutdownAsync"/>, <see cref="DisposeAsync"/> and the session rebuild.</summary>