lmxopcua/docs/plans/2026-07-15-mqtt-sparkplug-driver-design.md

MQTT / Sparkplug B driver — executable implementation design

Status: design, build-ready. 2026-07-15. Research input: docs/research/drivers/mqtt-sparkplug.md (verdicts + protocol/library survey — this doc turns that into a build plan; it does not re-litigate the research). Related: docs/plans/2026-07-15-universal-discovery-browser-design.md (why MQTT needs a bespoke browser despite the universal one — §4), docs/plans/2026-06-12-galaxy-standard-driver-design.md (standard Equipment-kind driver shape + fire-and-forget write precedent), docs/plans/2026-05-28-driver-browsers-design.md (IDriverBrowser/IBrowseSession pattern reused by the observation browser).

---

1. Motivation + scope

Build a standard Equipment-kind driver — same composable-IDriver shape as Modbus / S7 / AbCip / TwinCAT / FOCAS / OpcUaClient — that ingests plain MQTT and Sparkplug B into the OtOpcUa OPC UA / UNS address space.

• OtOpcUa is a Unified-Namespace product, and Sparkplug B is the de-facto UNS-over-MQTT payload standard (mandated topic namespace + protobuf schema + birth-certificate discovery + stateful sessions). Sparkplug alignment is therefore the strategic headline; plain MQTT/JSON is the broad-compatibility fallback for brokers that carry raw JSON topics.
• This is a subscribe-first driver. Unlike the polled drivers (Modbus/S7/AbCip/FOCAS with PollGroupEngine), it holds a live broker connection for its whole lifetime and raises OnDataChange from the MQTT receive callback. The closest existing analog is the native-push side of OpcUaClientDriver.
• In scope (this design): connect/TLS/auth/reconnect; topic + Sparkplug subscribe → OnDataChange; last-value IReadable; authored-only ITagDiscovery; IRediscoverable; IHostConnectivityProbe; a bespoke observation browser; typed AdminUI tag editor + validator; Docker fixtures + env-gated live suite.
• Out of scope for v1: write-through (IWritable — Sparkplug NCMD/DCMD + plain publish) is deferred to phase 3 (§3.4, §10); DataSet/Template Sparkplug metrics; Bytes/File metrics beyond a raw-String fallback.

Canonical DriverType string: "Mqtt" — one string used identically across the AdminUI driver page, probe, factory, editor map, validator, and browser. (Heed the Modbus "ModbusTcp" vs "Modbus" mismatch lesson: a divergent string silently drops tags to the raw-JSON editor / faults resolution. Pick "Mqtt" and grep to enforce it.)

---

2. Project layout

Three new projects under src/Drivers/, mirroring the OpcUaClient triad (.Driver + .Contracts + .Browser). All net10.0, Nullable+ImplicitUsings enabled, TreatWarningsAsErrors=true opted in per-csproj (mandatory for new projects — Directory.Build.props deliberately does not set it globally because legacy test projects would fail the build).

src/Drivers/
  ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Contracts/      # options + tag DTOs + parser — zero transport deps
    MqttDriverOptions.cs                          #   driver-config DTO (broker conn + mode + sparkplug/plain sub-objects)
    MqttMode.cs, MqttPayloadFormat.cs             #   enums (Plain|SparkplugB ; Json|Raw|Scalar)
    MqttTagDefinition.cs                          #   parsed per-tag descriptor (plain OR sparkplug variant)
    MqttEquipmentTagParser.cs                     #   TagConfig JSON → MqttTagDefinition (mirrors ModbusEquipmentTagParser)
    SparkplugDataType.cs                          #   Sparkplug metric-datatype enum + → DriverDataType map

  ZB.MOM.WW.OtOpcUa.Driver.Mqtt/                  # runtime driver — references .Contracts + Core.Abstractions + Core
    MqttDriver.cs                                 #   IDriver, ITagDiscovery, ISubscribable, IReadable,
                                                  #     IHostConnectivityProbe, IRediscoverable (+ IWritable phase 3)
    MqttDriverProbe.cs                            #   IDriverProbe — broker CONNECT handshake
    MqttDriverFactoryExtensions.cs                #   DriverTypeName="Mqtt"; Register(registry, loggerFactory)
    MqttConnection.cs                             #   MQTTnet client wrapper: connect/TLS/auth/reconnect/backoff
    LastValueCache.cs                             #   FullReference → last DataValueSnapshot (seeds IReadable)
    Sparkplug/
      SparkplugCodec.cs                           #   Tahu-proto decode (Payload/Metric); encode NCMD (phase 3)
      AliasTable.cs                               #   per edge-node/device alias→(name,datatype); rebuilt each birth
      BirthCache.cs                               #   NBIRTH/DBIRTH metric catalog (name/alias/datatype/last value)
      SequenceTracker.cs                          #   seq (0-255 wrap) gap detection; bdSeq death→birth tie
      RebirthRequester.cs                         #   publishes NCMD Node Control/Rebirth=true
      SparkplugTopic.cs                           #   parse/format spBv1.0/{group}/{type}/{node}[/{device}]

  ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Browser/          # bespoke observation browser — references .Contracts + Commons(.Browsing) + MQTTnet
    MqttDriverBrowser.cs                          #   IDriverBrowser (DriverType="Mqtt"); OpenAsync → MqttBrowseSession
    MqttBrowseSession.cs                          #   IBrowseSession over an accumulating observed tree

src/Server/.../AdminUI/
  Uns/TagEditors/MqttTagConfigModel.cs            #   typed working model (FromJson/ToJson/Validate; preserves unknown keys)
  Uns/TagEditors/TagConfigEditorMap.cs            #   + ["Mqtt"] = MqttTagConfigEditor
  Uns/TagEditors/TagConfigValidator.cs            #   + ["Mqtt"] = j => MqttTagConfigModel.FromJson(j).Validate()
  Components/Shared/Uns/TagEditors/MqttTagConfigEditor.razor
  EndpointRouteBuilderExtensions.cs               #   + AddSingleton<IDriverBrowser, MqttDriverBrowser>()

src/Server/.../Host/Drivers/DriverFactoryBootstrap.cs
                                                  #   + MqttDriverFactoryExtensions.Register(registry, loggerFactory)
                                                  #   + TryAddEnumerable(IDriverProbe, MqttDriverProbe)

tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Tests/          # unit — codec/alias/seq/parser (net10, macOS-safe)
tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.IntegrationTests/  # env-gated live suite (MQTT_FIXTURE_ENDPOINT)
tests/Drivers/.../Docker/docker-compose.yml       # mosquitto + sparkplug-sim + json-publisher (project=lmxopcua)

Why the split .Contracts project: it carries MqttDriverOptions + MqttTagDefinition + MqttEquipmentTagParser + the enums with zero transport (MQTTnet) dependency, exactly like Driver.Modbus.Contracts depends only on Modbus.Addressing + Core.Abstractions. This lets the AdminUI typed editor (and any tooling) reference the DTO shape/enums without dragging MQTTnet into AdminUI. The runtime .Driver and .Browser both reference .Contracts and add MQTTnet.

2.1 Libraries + the pinning caveat

Concern	Choice	Package
MQTT transport (both modes)	MQTTnet v5 — MIT, .NET Foundation (dotnet/MQTTnet), the standard .NET MQTT client. v5 explicitly ships a net10.0 TFM. Client + TLS + MQTT 3.1.1/5.0.	MQTTnet (add PackageVersion Include="MQTTnet" Version="5.x" to Directory.Packages.props)
Sparkplug decode	Hand-rolled Tahu protobuf (recommended) — generate C# from Tahu's sparkplug_b.proto with Google.Protobuf (Grpc.Tools/protoc), decode over a single MQTTnet-5 client. Small, stable schema; we own the alias/seq/rebirth state machine exactly as we want it.	Google.Protobuf + Grpc.Tools (build-time)
Alternative Sparkplug decode	SparkplugNet — MIT, provides Application/Node/Device base classes, saves the birth/alias plumbing. Rejected as primary for two repo-specific reasons (below).	SparkplugNet (only if the risk clears)

Pinning caveat (load-bearing — this is why hand-roll is recommended). This repo uses central package management (Directory.Packages.props, ManagePackageVersionsCentrally=true) and deliberately leaves CentralPackageTransitivePinningEnabled OFF because turning it on breaks the Roslyn 5.0.0/4.12.0 split (see the "Transitive pinning breaks Roslyn build" memory + Directory.Packages.props:103 comment). SparkplugNet pins MQTTnet 4.3.x transitively and has no explicit net10.0 TFM yet. A direct MQTTnet-5 reference (needed for the plain-mode transport + net10) will coexist in the graph with SparkplugNet's transitive 4.x, and because transitive pinning is off we cannot cleanly force-align the two. Net: SparkplugNet risks a hard-to-resolve MQTTnet 4/5 diamond in exactly the pinning configuration this repo can't change.

Decision: hand-roll Tahu-proto decode over one MQTTnet-5 client. It removes the diamond and the net10-TFM concern entirely, and the Sparkplug payload schema (Payload { seq, timestamp, Metric[], body }, Metric { name, alias, timestamp, datatype, value, is_null, ... }) is ~90 lines of .proto — a bounded, well-understood effort. Phase 1 (plain MQTT) references only MQTTnet-5, so we validate MQTTnet-v5-on-net10 before ever touching Sparkplug. (If a future need for SparkplugNet arises, the phase-2 codec seam SparkplugCodec is the single swap point.)

Vendor the proto: commit Tahu sparkplug_b.proto into Driver.Mqtt.Contracts/Protos/ (pinned to a known Tahu commit, with a provenance comment) and generate at build via Grpc.Tools. Don't fetch at build time.

Existing protobuf tooling, honestly: Google.Protobuf is already centrally pinned (Directory.Packages.props:31, 3.34.1) and consumed by Driver.Galaxy — but the Galaxy proto types ship pre-generated inside the ZB.MOM.WW.MxGateway.* NuGets; there is no .proto file or Grpc.Tools protoc codegen anywhere in this repo today. So Grpc.Tools is a new (build-time-only, PrivateAssets=all) package and this is the repo's first in-repo proto codegen step. If that step proves objectionable, the fallback is checking in the generated C# (the schema is small and stable) with the same provenance comment.

---

3. Capability mapping

MqttDriver : IDriver, ITagDiscovery, ISubscribable, IReadable, IHostConnectivityProbe, IRediscoverable (+ IWritable in phase 3). Not IHistoryProvider (historization is the server-side isHistorized path, unchanged). Not a poller — no PollGroupEngine.

3.0 Lifecycle (IDriver)

• InitializeAsync(configJson, ct) — deserialize MqttDriverOptions (shared JsonSerializerOptions with JsonStringEnumConverter, §6); build MqttConnection; connect to broker under a bounded connect deadline (§8); set LWT/clean-session; subscribe to the mode-appropriate filter (§3.1). Publish STATE birth if actAsPrimaryHost (Sparkplug).
• ReinitializeAsync — apply a config delta in place: if only tag membership changed, re-subscribe; if broker connection changed, tear down + reconnect. Never crash the process (marks Faulted → nodes go Bad, per driver-stability Tier A/B).
• ShutdownAsync — publish STATE OFFLINE (if primary host), disconnect the MQTTnet client, dispose.
• GetHealth — Connected/Reconnecting/Faulted + last-message-age; drives ServiceLevel + status dashboard.
• GetMemoryFootprint/FlushOptionalCachesAsync — footprint = last-value cache + birth caches + alias tables. The runtime driver has no optional caches to flush: birth/alias tables are required-for-correctness (flushing them mis-routes aliased DDATA — IDriver.FlushOptionalCachesAsync forbids flushing correctness state) and the last-value cache is what IReadable serves. Flush is a no-op; all three structures are bounded by the authored tag set + observed births. (The observation browser's accumulating tree lives in the AdminUI-side MqttBrowseSession, not in the runtime driver.)

3.1 Subscribe (ISubscribable — primary)

• Plain mode: subscribe to each authored tag's topic at its qos (dedupe identical topics; coalesce shared wildcards where the operator authored one). On message: match topic → authored tag(s), extract value at jsonPath (or raw/scalar), raise OnDataChange(handle, FullReference, snapshot). Seed initial value from the broker's retained message replayed on subscribe (OPC UA initial-data convention).
• Sparkplug mode: subscribe spBv1.0/{groupId}/# (+ the host-state topic when acting as primary host — spBv1.0/STATE/{hostId} with a JSON {online, timestamp} payload in spec v3.0; pre-3.0 peers used top-level STATE/{hostId} with ONLINE/OFFLINE text — target the v3.0 form, tolerate the legacy one on receive). Maintain a BirthCache + AliasTable per (edgeNode, device) from NBIRTH/DBIRTH. On NDATA/DDATA: resolve each metric by alias → (name, datatype) via the alias table, map to the authored tag's FullReference by stable metric name, raise OnDataChange. On NDEATH/DDEATH: emit STALE / Bad-quality snapshots for that node's/device's metrics. On missed birth / unknown alias / seq gap: issue a rebirth NCMD (§3.6).
• publishingInterval is advisory (MQTT/Sparkplug are event-driven). Use it only for optional server-side coalescing/deadband, mirroring Modbus's ShouldPublish. SubscribeAsync returns an ISubscriptionHandle whose DiagnosticId names the mode+filter; OnDataChange fires from the MQTT receive handler for the whole driver lifetime.

SubscribeAsync maps the requested FullReferences onto the live receive stream: it registers them in a FullReference → MqttTagDefinition resolver (parsed once, cached — reuse the shared EquipmentTagRefResolver<TDef> from Core.Abstractions, exactly as Modbus does), so an inbound message/metric routes to the right node.

3.2 Read (IReadable)

MQTT has no request/response read. ReadAsync(fullReferences) returns the last-value cache (LastValueCache): last message seen on the topic (plain) or last metric value from data/birth (Sparkplug). Seed from the retained message at connect (plain) and from birth values (Sparkplug). If nothing has been observed yet, return a GoodNoData/uncertain snapshot per reference (never throw the batch — per IReadable contract, per-reference status is the snapshot's StatusCode). This satisfies OPC UA reads + HistoryRead-AtTime without a live round-trip.

3.3 Discover (ITagDiscovery) — authored-only

DiscoverAsync replays authored tags only, both modes — it does not auto-provision from wildcard/birth observation. Rationale (from research §2.2): a chatty broker or a large Sparkplug plant would produce an unbounded auto-provisioned address space. Wildcard/birth observation is a browse-time concern (§4), not a runtime-provisioning one.

• The operator authors tags (topic+jsonPath, or group/node/device/metric) via the typed editor / observation browser. DiscoverAsync streams exactly those authored tags into the IAddressSpaceBuilder, each as a Variable(browseName, displayName, DriverAttributeInfo{ FullName, DriverDataType, ... }).
• RediscoverPolicy:
  • Plain: Once (authored set is static per published config — like Modbus).
  • Sparkplug: UntilStable (metric datatypes fill in asynchronously as births arrive after connect; keep re-streaming until the authored tags' resolved datatypes stop changing — like FOCAS's FixedTree). This lets a tag authored before its first birth pick up its true datatype once the birth lands.
• SupportsOnlineDiscovery => false (the default-member from the universal-browser design). This is deliberate and decoupled from browseability (§4): runtime discovery is authored-only, so the universal Discover-backed browser would only replay authored tags — useless for discovering new topics/metrics. Leaving the flag false keeps MQTT off the universal fallback; the bespoke observation browser (§4) is what backs the picker.

3.4 IRediscoverable

Sparkplug fires OnRediscoveryNeeded when a new DBIRTH (a device joins) or a rebirth with a changed metric set is observed — the Core re-runs DiscoverAsync + diffs. ScopeHint = the edge-node/device folder path so the rebuild is surgical. Plain mode never fires it (no backend change signal — a config republish is the only tag-set change), so plain-only drivers effectively no-op this interface.

3.5 Data-type mapping

Sparkplug metric datatype (or inferred/authored plain type) → DriverDataType (Boolean, Int16, Int32, Int64, UInt16, UInt32, UInt64, Float32, Float64, String, DateTime, Reference):

Sparkplug DataType	DriverDataType	OPC UA
Int8, Int16	Int16 (Int8 widens)	Int16
Int32	Int32	Int32
UInt8, UInt16	UInt16 (UInt8 widens)	UInt16
UInt32	UInt32	UInt32
Int64	Int64	Int64
UInt64	UInt64	UInt64
Float	Float32	Float
Double	Float64	Double
Boolean	Boolean	Boolean
String, Text, UUID	String	String
DateTime	DateTime	DateTime
Bytes, File	String (base64/raw fallback, v1)	String
DataSet, Template	unsupported v1 — skip + warn, or expose raw-JSON as String	—
*Array variants	element type + IsArray=true, ArrayDim from birth	ValueRank=1

Live map in SparkplugDataType.ToDriverDataType() (.Contracts). Plain mode: honour an explicit per-tag dataType (preferred — inference is brittle); fall back to inferring from the JSON token at jsonPath (number→Float64/Int64, bool→Boolean, string→String). Int8/UInt8 widen (no OPC UA signed-byte distinction needed).

3.6 Sparkplug state-machine correctness (the #1 risk)

The alias→metric map, seq-gap detection, late-join rebirth, and death→STALE are load-bearing and easy to get subtly wrong. Invariants the implementation must hold:

1. Bind tags by stable metric NAME, never by alias. The authored TagConfig stores metricName; the alias is a per-birth cache only. An alias can be reused across a rebirth for a different metric — binding by alias silently mis-routes data to the wrong tag.
2. Rebuild the alias table on every (re)birth. NBIRTH/DBIRTH replaces the (edgeNode,device) AliasTable wholesale; never merge.
3. Track seq (0–255, wraps) per edge-node. A gap (or an unknown alias, or a data message before any birth) ⇒ request rebirth (RebirthRequester publishes NCMD writing Node Control/Rebirth = true), gated on requestRebirthOnGap. bdSeq ties an NDEATH to its NBIRTH.
4. Death → STALE. NDEATH/DDEATH (LWT-driven) emits Bad/uncertain-quality snapshots for all affected metrics; the next birth restores Good.
5. Late join. On connect the driver may have missed births — proactively request rebirth (or wait for the retained STATE + natural rebirth) so metadata is recovered. This is the same primitive the browser uses (§4).

Unit-test these explicitly against a controllable simulator: rebirth, missed-birth, seq-gap, alias-reuse-across-rebirth, death→stale→rebirth.

3.7 Write (IWritable) — verdict: NOT in v1; phase 3, fire-and-forget optimistic-Good

Writes are possible but deferred (research §2.4). When shipped in phase 3:

• Sparkplug: publish NCMD/DCMD to spBv1.0/{group}/NCMD/{node}[/{device}] with the target metric (by name+alias) + value. No synchronous write ack — success is only observable when the edge echoes the change in the next N/DDATA. This mirrors the Galaxy gateway's fire-and-forget write ("Galaxy can never surface a write failure"): return Good optimistically; the value self-corrects on the next data message. Respect WriteIdempotent — Sparkplug commands (rebirth, pulse) are frequently non-idempotent, so default retry off.
• Plain: publish to a configured command/write topic with a formatted payload.

v1 ships read/subscribe/discover only — a genuinely useful UNS-ingest driver. MqttDriver does not implement IWritable in v1 (the interface is composable; its absence means nodes materialize read-only).

---

4. Bespoke observation browser

Decision — reconciling with the universal Discover-backed browser

The universal browser (docs/plans/2026-07-15-universal-discovery-browser-design.md) captures whatever DiscoverAsync streams and re-presents it, gated on SupportsOnlineDiscovery. MQTT/Sparkplug is the one roadmap driver that needs a bespoke browser despite the universal one, because:

• Runtime DiscoverAsync is authored-only (§3.3). A universal Discover-backed browser would replay only tags the operator already authored — it would not discover new topics/metrics, which is the entire point of a picker.
• MQTT/Sparkplug discovery is observational: you learn what exists only by listening for a bounded window. That's a fundamentally different session shape than "connect → one-shot enumerate → disconnect" — the tree fills in over time, and Sparkplug can be accelerated by forcing a rebirth NCMD.

Considered + rejected: feeding the Sparkplug BirthCache into DiscoverAsync device-enumeration to plug into universal. Rejected because it would make runtime discovery auto-provision from births (the unbounded-address-space failure §3.3 explicitly avoids) and still wouldn't cover plain MQTT. The birth cache stays a browse-time + subscribe-time structure, not a discovery source.

Decision: keep SupportsOnlineDiscovery=false (authored-only runtime discovery) and build a bespoke IDriverBrowser (MqttDriverBrowser) with an observation-window IBrowseSession (MqttBrowseSession). Registering the bespoke browser for DriverType="Mqtt" overrides the universal fallback for MQTT (the universal browser resolves bespoke-first — BrowserSessionService §3.1 of that doc).

4.1 MqttDriverBrowser : IDriverBrowser

DriverType => "Mqtt". OpenAsync(configJson, ct):

1. Parse the same MqttDriverOptions JSON the runtime driver consumes (shared JsonSerializerOptions), but connect with a distinct client-id suffix ({clientId}-browse-{guid8}) so the browse session never collides with the running driver's MQTT session / clean-session state.
2. Connect an MQTTnet client under a bounded connect budget (clamp like OpcUaClient's PerEndpointConnectTimeout, 5–30 s).
3. Subscribe to the discovery filter: spBv1.0/{groupId}/# (Sparkplug) or #/{topicPrefix}# (plain).
4. Kick off the observation window immediately and, in Sparkplug mode, optionally publish a rebirth NCMD to force every edge node to re-announce (turns the passive wait into a near-synchronous enumeration).
5. Return an MqttBrowseSession whose internal tree fills in over the window and keeps filling while the session lives. Because IBrowseSession methods are async and the session lands in the AdminUI BrowseSessionRegistry (TTL-reaped, per-call 20 s timeout), the observation cost is amortised across the user's clicks in the picker.

4.2 MqttBrowseSession : IBrowseSession

Serves from an accumulating observed tree (thread-safe; the MQTT receive handler mutates it, browse calls read it). Refreshes LastUsedUtc on each call.

• RootAsync — Sparkplug: observed groups (or edge nodes under the configured group). Plain: top-level topic segments. BrowseNode { Kind=Folder, HasChildrenHint=true }. If the window hasn't yielded yet, return what's accumulated (possibly empty) and let the user re-expand; in Sparkplug mode trigger a rebirth on first RootAsync to populate fast.
• ExpandAsync(nodeId) — one level down the accumulated tree. Sparkplug: Group → EdgeNode → Device → Metric. Plain: next topic segment (split on /). Metrics / leaf-topics are Kind=Leaf.
• AttributesAsync(nodeId) — Sparkplug: the metric's AttributeInfo { Name, DriverDataType (from birth), IsArray, SecurityClass } — self-describing, so the picker side-panel is populated (unlike the OPC UA browser which returns empty). Plain: a single synthetic attribute carrying the leaf topic's inferred type + last-seen payload snippet.
• DisposeAsync — disconnect the MQTTnet client, best-effort (registry reaper may race), same pattern as OpcUaClientBrowseSession.

4.3 Picked node → TagConfig

On commit the picker projects the selected BrowseNode/AttributeInfo into the §5 TagConfig JSON:

• Sparkplug: the NodeId encodes group/node/device:metric → split into groupId/edgeNodeId/deviceId/metricName; dataType from AttributeInfo.
• Plain: the NodeId is the topic path → topic + a default jsonPath ($, operator-editable) + inferred dataType.

4.4 Addressing the passive/time-bounded wrinkle explicitly

1. Picker shows a live "listening… N nodes/topics discovered" affordance so the operator understands coverage is accumulating, not instantaneous.
2. Sparkplug mode actively forces a rebirth to convert the passive wait into a fast near-complete enumeration.
3. The browse session stays alive (registry TTL) so re-expanding later reflects newly observed topics without reconnecting.
4. Manual entry of a topic/metric via the typed editor (§6) is always available as an escape hatch for a tag that's silent during the window.

4.5 Registration

// EndpointRouteBuilderExtensions.cs — alongside the existing two (near line 49-50)
services.AddSingleton<IDriverBrowser, OpcUaClientDriverBrowser>();
services.AddSingleton<IDriverBrowser, GalaxyDriverBrowser>();
services.AddSingleton<IDriverBrowser, MqttDriverBrowser>();   // NEW — overrides universal fallback for "Mqtt"

---

5. TagConfig + driver-config JSON

Two layers, mirroring every other driver: driver options (broker connection, one per driver instance, parsed by the factory) and per-tag config (TagConfig JSON authored on the /uns Equipment → Tags tab; the driver's FullName is the driver-side reference the router resolves).

5.1 Driver options — MqttDriverOptions (.Contracts)

{
  "host": "10.100.0.35",
  "port": 8883,
  "clientId": "otopcua-uns-1",
  "useTls": true,
  "allowUntrustedServerCertificate": false,   // dev/on-prem only
  "caCertificatePath": null,                   // PEM CA pin; null ⇒ OS trust store
  "username": "otopcua",
  "password": "",                              // supply via env, NEVER commit
  "protocolVersion": "V500",                   // "V311" | "V500"
  "cleanSession": true,
  "keepAliveSeconds": 30,
  "connectTimeoutSeconds": 15,                 // bounded connect deadline (§8)
  "reconnectMinBackoffSeconds": 1,
  "reconnectMaxBackoffSeconds": 30,

  "mode": "SparkplugB",                        // "Plain" | "SparkplugB"

  // ---- Sparkplug-only (null in Plain mode) ----
  "sparkplug": {
    "groupId": "Plant1",                       // subscribe spBv1.0/Plant1/#
    "hostId": "otopcua-host-1",                // spBv1.0/STATE/{hostId} identity (v3.0)
    "actAsPrimaryHost": false,                 // at most ONE primary host per host-id per broker
    "requestRebirthOnGap": true,
    "birthObservationWindowSeconds": 15        // browse/discovery: how long to collect births
  },

  // ---- Plain-only (null in SparkplugB mode) ----
  "plain": {
    "topicPrefix": "factory/",
    "defaultQos": 1
  }
}

5.2 Per-tag — Sparkplug

{
  "FullName": "Plant1/Line3EdgeNode/Filler1:Temperature/degC",  // PascalCase key (composer/walker contract)
  "groupId": "Plant1",
  "edgeNodeId": "Line3EdgeNode",
  "deviceId": "Filler1",                       // omit/null for node-level metrics
  "metricName": "Temperature/degC",            // STABLE identity across rebirths — bind by this, not alias
  "dataType": "Float"                          // from DBIRTH; explicit for safety
}

FullName convention: "{group}/{node}[/{device}]:{metric}". The driver parses FullName (or the descriptor fields) once, caches it, and resolves inbound metrics by (group,node,device,metricName).

5.3 Per-tag — Plain MQTT

{
  "FullName": "factory/line3/oven/temp#$.value",  // topic#jsonPath — stable per-tag key
  "topic": "factory/line3/oven/temp",
  "payloadFormat": "Json",                     // "Json" | "Raw" | "Scalar"
  "jsonPath": "$.value",                        // JSONPath into payload (Json only)
  "dataType": "Double",                         // explicit — inference is the fallback
  "qos": 1,
  "retainSeed": true                            // seed last-value from retained msg on subscribe
}

FullName convention: "{topic}#{jsonPath}" (Json) or just "{topic}" (Raw/Scalar). MqttEquipmentTagParser.TryParse(reference, out def) follows ModbusEquipmentTagParser: a leading { marks a TagConfig blob; parse it, use strict enum reads (TagConfigJson.TryReadEnumStrict) so a typo'd payloadFormat/dataType rejects the tag (→ BadNodeIdUnknown) rather than silently defaulting to a wrong type; the def's Name = the reference string so published values key the forward router. Add an Inspect(reference) deploy-time warning pass (like Modbus) for present-but-invalid enums / unparseable blobs.

Historization: the server-side isHistorized / historianTagname keys are owned by the TagModal-merge seam (TagHistorizeConfig) and survive load→save of the typed model as preserved unknown keys — the driver does not model them (identical to OpcUaClientTagConfigModel).

---

6. Typed editor + validator

6.1 MqttTagConfigModel (.AdminUI/Uns/TagEditors/)

Thin typed working model over a preserved JsonObject key bag, per the OpcUaClientTagConfigModel template: FromJson(json) / ToJson() / Validate(), preserving unknown keys across a load→save (so history keys + fields this editor doesn't expose survive). It carries both mode shapes — a Mode field (Plain|SparkplugB, defaulted from which fields are present) selects which sub-shape the editor renders + validates.

• FromJson reads the descriptor fields for the active mode plus the always-present FullName (PascalCase, case-sensitive — the composer/artifact-decoder/walker read it via case-sensitive TryGetProperty("FullName", …)).
• ToJson writes FullName PascalCase and re-derives it from the descriptor fields ({group}/{node}[/{device}]:{metric} or {topic}#{jsonPath}) so a hand-edited descriptor stays consistent with the routed key.
• Validate():
  • Sparkplug: groupId, edgeNodeId, metricName required; deviceId optional; dataType must be a known Sparkplug type.
  • Plain: topic required and non-wildcard (a subscription topic for a tag must be concrete — reject +/#); payloadFormat=Json ⇒ jsonPath required; dataType a known DriverDataType.

6.2 MqttTagConfigEditor.razor

Copy the Modbus editor template (Components/Shared/Uns/TagEditors/): a razor shell over the pure model, mode-switch at top, per-mode field group below, client-side validation via Validate(). Register:

// TagConfigEditorMap.cs
["Mqtt"] = typeof(Components.Shared.Uns.TagEditors.MqttTagConfigEditor),
// TagConfigValidator.cs
["Mqtt"] = j => MqttTagConfigModel.FromJson(j).Validate(),

6.3 The JsonStringEnumConverter enum-serialization trap

Every JSON seam that (de)serializes MqttDriverOptions / tag configs — factory, probe, browser, and the AdminUI driver-config page + probe DTO — MUST use a JsonSerializerOptions that includes new JsonStringEnumConverter() (plus PropertyNameCaseInsensitive=true, UnmappedMemberHandling=Skip). The repo's SYSTEMIC bug (memory: "Driver enum-serialization bug"): AdminUI pages that serialize enums numerically while factory DTOs are string-typed fault the driver on any enum field. MqttMode, MqttPayloadFormat, protocolVersion are all enums — the AdminUI Mqtt driver page and its probe must serialize them as strings (mirror the OpcUaClient page). JsonStringEnumConverter also accepts numeric ordinals, so existing numeric configs still load. Keep one shared JsonSerializerOptions definition referenced by factory + probe + browser (as OpcUaClient does).

---

7. Factory + registration

MqttDriverFactoryExtensions (mirror OpcUaClientDriverFactoryExtensions):

public static class MqttDriverFactoryExtensions
{
    public const string DriverTypeName = "Mqtt";
    private static readonly JsonSerializerOptions JsonOptions = new()
    {
        PropertyNameCaseInsensitive = true,
        UnmappedMemberHandling = JsonUnmappedMemberHandling.Skip,
        Converters = { new JsonStringEnumConverter() },
    };

    public static void Register(DriverFactoryRegistry registry, ILoggerFactory? loggerFactory = null)
    {
        ArgumentNullException.ThrowIfNull(registry);
        registry.Register(DriverTypeName, (id, json) => CreateInstance(id, json, loggerFactory));
    }

    public static MqttDriver CreateInstance(string driverInstanceId, string driverConfigJson, ILoggerFactory? lf = null)
    {
        var options = JsonSerializer.Deserialize<MqttDriverOptions>(driverConfigJson, JsonOptions)
            ?? throw new InvalidOperationException($"Mqtt driver config for '{driverInstanceId}' deserialised to null");
        return new MqttDriver(options, driverInstanceId, lf?.CreateLogger<MqttDriver>());
    }
}

Wire in the two host bootstrap sites:

// Host/Drivers/DriverFactoryBootstrap.cs — factory registration (near line 133)
Driver.Mqtt.MqttDriverFactoryExtensions.Register(registry, loggerFactory);
// … and probe registration in AddOtOpcUaDriverProbes (near line 114)
services.TryAddEnumerable(ServiceDescriptor.Singleton<IDriverProbe, Driver.Mqtt.MqttDriverProbe>());

MqttDriverProbe : IDriverProbe (DriverType => "Mqtt"): parse the config with the shared JsonSerializerOptions, open an MQTTnet CONNECT to host:port under the timeout, return green + latency on CONNACK-accepted, or a targeted error (connection refused / TLS failure / auth failure / timeout). Analogous to ModbusDriverProbe's TCP+FC03 handshake — CONNACK is the MQTT "device is answering" proof. Probes are admin-pinned (the admin-operations singleton dispatches by DriverType) — the TryAddEnumerable line must be on admin nodes (it is, via AddOtOpcUaDriverProbes in Program.cs's hasAdmin block).

---

8. Resilience / timeout / security

• Bounded connect. InitializeAsync connects under connectTimeoutSeconds (default 15) via a linked CTS — no unbounded wait on a dead broker (heed the S7/FOCAS read-timeout lessons: an async network call that ignores its deadline wedges the whole driver).
• Reconnect with backoff. MQTTnet's managed reconnect (or a hand-rolled loop) with exponential backoff reconnectMinBackoffSeconds → reconnectMaxBackoffSeconds. On disconnect: GetHealth → Reconnecting; on reconnect: re-subscribe, and in Sparkplug mode request rebirth to recover metadata (late-join, §3.6). Nodes go Bad-quality while disconnected via the standard fan-out.
• Subscribe deadline. SubscribeAsync completes under a bounded deadline; SUBACK failure surfaces as a per-reference Bad, not a hang.
• No poll loop — the receive handler must be non-blocking (offload heavy decode off the MQTT callback thread; never block the MQTTnet dispatcher).
• Broker security — enforce, never ship public-broker defaults. Default useTls=true; real auth (username/password or client-cert). allowUntrustedServerCertificate + caCertificatePath mirror the ServerHistorian TLS knobs (dev/on-prem escape hatch, off by default). Never put Sparkplug production-shaped data on public brokers (test.mosquitto.org etc. — manual smoke only). Credentials come from env (password blank in committed JSON), same posture as ServerHistorian__ApiKey.
• Primary-host guard. At most one primary-host client per hostId per broker (Sparkplug MUST). actAsPrimaryHost=true on two OtOpcUa instances sharing a hostId (e.g. a redundancy pair) is a misconfiguration — the redundancy pair should use distinct host-ids (mirrors the distinct MxAccess.ClientName redundancy rule).

---

9. Test fixtures

Follow the driver-fixture pattern: tests/Drivers/.../Docker/docker-compose.yml with the project: lmxopcua label, deployed to 10.100.0.35 via lmxopcua-fix sync/up (CLAUDE.md Docker Workflow). Repo files are source-of-truth; /opt/otopcua-mqtt/ is the mirrored deployment.

• Broker: Eclipse Mosquitto (eclipse-mosquitto, tiny, carries both plain + Sparkplug unchanged) as the default fixture; EMQX (emqx/emqx, Apache-2.0 CE) as the alternative when a dashboard / built-in Sparkplug tooling helps. Configure with auth + TLS so the fixtures exercise the real security path (not anonymous).
• Sparkplug edge-node simulator: prefer a project-owned C# simulator using the same MQTTnet + Tahu-proto path the driver uses — validates encode/decode symmetrically and gives full control of rebirth/seq-gap/death scenarios (the §3.6 test matrix). Alternatives: Eclipse Tahu reference (Java/Python) edge-node script; Bevywise MQTT/IoT Simulator (purpose-built Sparkplug B). The simulator must publish NBIRTH/DBIRTH → periodic NDATA/DDATA, honour a rebirth NCMD, and be able to inject a seq gap / alias reuse / N-D-DEATH on demand.
• Plain-MQTT fixture: a tiny publisher container (mosquitto_pub loop or a paho-mqtt script) emitting JSON on a handful of topics with retain=true so the last-value/read path and the #-observation browse are testable.
• Env-gated live suite: Driver.Mqtt.IntegrationTests, category LiveIntegration (à la HistorianGateway), skips cleanly when MQTT_FIXTURE_ENDPOINT is unset → macOS-offline-safe. Covers: connect/TLS/auth; plain subscribe+retained-read; Sparkplug birth→data→alias-resolve→OnDataChange; rebirth recovery; death→STALE; the browser observation window (birth-driven tree + rebirth-forced enumeration).
• Live-verify (docker-dev): author an Mqtt driver + tag via the /uns picker on :9200, deploy, confirm the node carries live broker values and the typed editor renders both modes (the usual live-/run discipline — Razor binding bugs pass unit tests).

Add the fixture endpoint to CLAUDE.md's endpoint list once landed (e.g. 10.100.0.35:1883/8883, MQTT_FIXTURE_ENDPOINT).

---

10. Phasing + effort

Order: plain MQTT first, then Sparkplug B, then write. Plain MQTT is the dependency-light slice that stands up the whole skeleton and validates MQTTnet-v5-on-net10 before any Sparkplug/proto work — front-loading the reusable plumbing and de-risking the library decision.

Phase	Scope	Rel. effort
P1 — Plain MQTT	3 projects; MQTTnet-5 connect/TLS/auth/reconnect+backoff (MqttConnection); ISubscribable (topic subscribe → OnDataChange); IReadable last-value (retained seed); authored-tag ITagDiscovery (Once); IHostConnectivityProbe; MqttDriverProbe; factory + host wiring; #-observation browser (MqttDriverBrowser/MqttBrowseSession); typed editor + validator + AdminUI driver page; Mosquitto + JSON-publisher fixtures + env-gated suite. Validate MQTTnet-5/net10 here.	M
P2 — Sparkplug B ingest	Vendored Tahu proto + Google.Protobuf codegen; SparkplugCodec decode; spBv1.0/{group}/# subscribe; N/DBIRTH → BirthCache/AliasTable; bind-by-name + seq-gap detect + rebirth NCMD; N/DDEATH → STALE; STATE/primary-host option; ITagDiscovery UntilStable + IRediscoverable; birth-driven browser tree + AttributesAsync; Sparkplug datatype map; C# edge-node simulator fixture + §3.6 test matrix.	L
P3 — Write-through (optional)	IWritable: Sparkplug NCMD/DCMD + plain publish; optimistic-Good + self-correct on echo; WriteIdempotent respect (default off for non-idempotent Sparkplug commands); write-topic config.	M

Top risks: (1) Sparkplug state-machine correctness (aliases + rebirth + seq — §3.6; bind-by-name is the mitigation, test against a controllable simulator); (2) library/dependency friction — SparkplugNet's MQTTnet-4.x transitive pin ✕ this repo's fragile central pinning (transitive pinning off), mitigated by hand-rolling the Tahu proto over one MQTTnet-5 client (§2.1). Secondary: passive-browse coverage gaps (rebirth-forcing + manual entry), unbounded address space (authored-only runtime discovery), write-has-no-ack (accepted, optimistic-Good like Galaxy), broker security (TLS + real auth enforced).