# MQTT / Sparkplug B Driver Implementation Plan > **For Claude:** REQUIRED SUB-SKILL: Use superpowers-extended-cc:executing-plans (or subagent-driven-development) to implement this plan task-by-task. **Goal:** Ship a standard Equipment-kind `Mqtt` driver that ingests plain MQTT (P1) then Sparkplug B (P2) into the OtOpcUa dual-namespace address space, with a bespoke observation browser, typed AdminUI editor, and TLS+auth-enforced fixtures. **Architecture:** Three new `net10.0` projects mirror the OpcUaClient triad — `.Contracts` (transport-free options/DTOs/parser/enums), `.Driver` (subscribe-first `IDriver` holding one live MQTTnet-5 client for its whole lifetime, raising `OnDataChange` from the receive callback, with a hand-rolled reconnect loop since v5 dropped `ManagedMqttClient`), and `.Browser` (a passive `#`/birth observation-window `IBrowseSession`). Sparkplug B decodes vendored Eclipse Tahu protobuf via `Google.Protobuf`/`Grpc.Tools` over that same single client — no SparkplugNet. **Tech Stack:** MQTTnet v5 (MIT, .NET Foundation, ships `net10.0`), vendored Eclipse Tahu `sparkplug_b.proto` + `Google.Protobuf` (already pinned 3.34.1) + `Grpc.Tools` (already pinned 2.76.0, build-time), bespoke browser. **Source design:** docs/plans/2026-07-15-mqtt-sparkplug-driver-design.md **Phases:** P1 (plain MQTT) = Tasks 0–14 (a complete shippable milestone); P2 (Sparkplug B ingest) = Tasks 15–26 (built on the P1 skeleton). Write-through (NCMD/DCMD/plain-publish) is deferred — see "Deferred / out of scope". --- ## Cross-cutting rules (apply to every task — program design §3.1) - **Enum-serialization trap (systemic bug):** every JSON seam that (de)serializes `MqttDriverOptions` or a tag config — factory, probe, browser, **and** the AdminUI driver-config page + probe DTO — MUST share a `JsonSerializerOptions` carrying `new JsonStringEnumConverter()` + `PropertyNameCaseInsensitive=true` + `UnmappedMemberHandling=Skip`. `MqttMode`, `MqttPayloadFormat`, `protocolVersion` are enums; serialize them as **names**. Mirror `OpcUaClientDriverFactoryExtensions`. - **Per-op deadline (R2-01 frozen-peer lesson):** every network call (connect / subscribe / probe / rebirth-publish) has a bounded linked-CTS deadline. No unbounded waits on a dead broker. - **Ctor is connection-free:** `MqttDriver`/`MqttDriverBrowser` constructors touch no network; all connects happen in `InitializeAsync`/`OpenAsync` (the universal-browser `CanBrowse` throwaway-instance pattern depends on this). - **Secrets from env:** broker `password` is blank in committed JSON, supplied via env (mirror `ServerHistorian__ApiKey`). Never commit or log creds. - **Broker security — never ship anonymous/public-broker defaults:** default `useTls=true`, real auth. `allowUntrustedServerCertificate` + `caCertificatePath` mirror the ServerHistorian TLS knobs (dev/on-prem escape hatch, off by default). Fixtures run auth+TLS. - **DriverType string is `"Mqtt"`** everywhere (driver page, probe, factory, editor map, validator, browser). One constant `DriverTypeNames.Mqtt`; grep to enforce (heed the `ModbusTcp`/`Modbus` mismatch lesson). - **Live-verify discipline:** Razor binding + deploy-inertness bugs pass unit tests. Every picker/editor/deploy path gets a docker-dev `/run` live-verify (Tasks 14 + 26). - **New-project csproj:** `net10.0`, `Nullable` + `ImplicitUsings` enabled, `TreatWarningsAsErrors=true` opted in per-csproj (not global). --- ## Task 0 (P1): Dependency-validation spike — MQTTnet-5 pin + net10 restore/build **Classification:** standard **Estimated implement time:** ~5 min **Parallelizable with:** none (this is the gate — the design's #1 dependency risk; nothing else starts until it is green) **Files:** - Modify: `Directory.Packages.props` (add ``) - Create: `src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Contracts/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Contracts.csproj` (temporary spike form — references `MQTTnet` only to prove the graph restores; the transport ref is removed in Task 1, `.Contracts` is transport-free) - Modify: `ZB.MOM.WW.OtOpcUa.slnx` (add the project) **Why first:** The design (§2.1, §10) makes this the top dependency risk — the repo uses central package management with `CentralPackageTransitivePinningEnabled` **deliberately OFF** (Roslyn 5.0.0/4.12.0 split, `Directory.Packages.props:103` comment + "Transitive pinning breaks Roslyn build" memory). We must prove **MQTTnet v5 restores and builds clean on net10 in this exact pinning configuration** before writing any driver code. Fresh-restore red-vs-local-green (the NU1903 memory) also means we validate a clean restore, not just an incremental one. ### Steps 1. Confirm on NuGet the exact latest **MQTTnet v5** version carrying a `net10.0` TFM; pin that exact version in `Directory.Packages.props` (alphabetical position near `MessagePack`). 2. Scaffold the `.Contracts` csproj (net10.0, nullable, implicit usings, TWAE) with a single `` (no `Version` — central management supplies it). Add to `slnx`. 3. Re-verify the two external-library record claims (design §2.1 checkbox): (a) MQTTnet v5 net10.0 TFM exists; (b) SparkplugNet still transitively pins MQTTnet 4.3.x with no net10.0 TFM. Record the confirmed versions in the tasks.json note. The hand-roll decision does not hinge on (b) — it is a confirm-the-record check. 4. Prove a **clean** restore + build: ```bash dotnet restore src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Contracts # expect: no NU1605/NU1608 version-conflict, no NU1903 audit error dotnet build src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Contracts # expect: Build succeeded, 0 warnings (TWAE on) rm -rf ~/.nuget/packages/mqttnet && dotnet restore src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Contracts # fresh-restore proof ``` **Expected:** all three succeed with zero version-conflict / transitive-pin / audit errors. If restore fails on a 4/5 diamond or a missing net10 TFM, STOP — the hand-roll-Tahu decision (§2.1) is vindicated but the MQTTnet-5 pin itself is the blocker; resolve the exact version before proceeding. 5. Commit: ```bash git commit -am "feat(mqtt): validate MQTTnet-5/net10 restore under central pinning (spike, P1 gate) Claude-Session: https://claude.ai/code/session_01GASWkNEi68FSCtvr6rLoEW" ``` --- ## Task 1 (P1): `.Contracts` — enums + `MqttDriverOptions` DTO **Classification:** small **Estimated implement time:** ~5 min **Parallelizable with:** none (Task 2 depends on this) **Files:** - Modify: `src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Contracts/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Contracts.csproj` (drop the temporary MQTTnet ref from Task 0 — `.Contracts` is transport-free; reference only `Core.Abstractions`) - Create: `src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Contracts/MqttMode.cs` (`Plain | SparkplugB`) - Create: `src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Contracts/MqttPayloadFormat.cs` (`Json | Raw | Scalar`) - Create: `src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Contracts/MqttProtocolVersion.cs` (`V311 | V500`) - Create: `src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Contracts/MqttDriverOptions.cs` (broker conn + mode + `sparkplug`/`plain` sub-objects, §5.1) - Create: `tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Tests/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Tests.csproj` (net10, xUnit + Shouldly) - Create: `tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Tests/MqttDriverOptionsTests.cs` - Modify: `ZB.MOM.WW.OtOpcUa.slnx` ### Steps (TDD) 1. Write the failing test — options round-trip enum-as-name: ```csharp public sealed class MqttDriverOptionsTests { private static readonly JsonSerializerOptions J = new() { PropertyNameCaseInsensitive = true, UnmappedMemberHandling = JsonUnmappedMemberHandling.Skip, Converters = { new JsonStringEnumConverter() }, }; [Fact] public void Deserialize_SparkplugConfig_ReadsModeAndSubObject() { const string json = """ { "host":"10.100.0.35","port":8883,"useTls":true,"mode":"SparkplugB", "sparkplug":{"groupId":"Plant1","hostId":"h1","requestRebirthOnGap":true} } """; var o = JsonSerializer.Deserialize(json, J)!; o.Mode.ShouldBe(MqttMode.SparkplugB); o.UseTls.ShouldBeTrue(); o.Sparkplug!.GroupId.ShouldBe("Plant1"); o.Plain.ShouldBeNull(); } [Fact] public void Serialize_WritesEnumsAsNames_NotOrdinals() { var s = JsonSerializer.Serialize(new MqttDriverOptions { Mode = MqttMode.Plain }, J); s.ShouldContain("\"Plain\""); s.ShouldNotContain("\"mode\":0"); } } ``` 2. Run — expect FAIL (types don't exist): ```bash dotnet test tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Tests # RED ``` 3. Implement the enums + `MqttDriverOptions` (with nested `MqttSparkplugOptions` / `MqttPlainOptions`) matching §5.1 defaults (`useTls=true`, `connectTimeoutSeconds=15`, `reconnectMin/MaxBackoffSeconds=1/30`). Wire the test csproj into `slnx`. 4. Run — expect PASS. 5. Commit: `git commit -am "feat(mqtt): Contracts options DTO + enums (name-serialized)"` (+ session trailer). --- ## Task 2 (P1): `.Contracts` — `MqttTagDefinition` + `MqttEquipmentTagParser` (plain) **Classification:** standard **Estimated implement time:** ~5 min **Parallelizable with:** none (Task 6 depends on the resolver) **Files:** - Create: `src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Contracts/MqttTagDefinition.cs` (parsed per-tag descriptor — carries plain OR sparkplug variant fields; plain fields populated in P1) - Create: `src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Contracts/MqttEquipmentTagParser.cs` (`TryParse(reference, out def)` + `Inspect(reference)`, mirrors `ModbusEquipmentTagParser` / `ModbusTagDefinitionFactory`) - Create: `tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Tests/MqttEquipmentTagParserTests.cs` ### Steps (TDD) 1. Failing tests — parse a plain TagConfig blob, reject a typo'd enum (strict), and reject a wildcard topic: ```csharp [Fact] public void TryParse_PlainJsonBlob_PopulatesTopicAndPath() { const string r = """{"topic":"factory/oven/temp","payloadFormat":"Json","jsonPath":"$.value","dataType":"Double","qos":1}"""; MqttEquipmentTagParser.TryParse(r, out var def).ShouldBeTrue(); def!.Topic.ShouldBe("factory/oven/temp"); def.PayloadFormat.ShouldBe(MqttPayloadFormat.Json); def.Name.ShouldBe(r); // the def Name == reference string (forward-router key) } [Fact] public void TryParse_TypoedPayloadFormat_RejectsStrict() => MqttEquipmentTagParser.TryParse( """{"topic":"a/b","payloadFormat":"Jason","dataType":"Double"}""", out _).ShouldBeFalse(); [Fact] public void Inspect_WildcardTopic_ReturnsWarning() => MqttEquipmentTagParser.Inspect("""{"topic":"a/+/c","payloadFormat":"Raw"}""").ShouldNotBeEmpty(); ``` 2. Run — expect FAIL. 3. Implement: a leading `{` marks a TagConfig blob; use `TagConfigJson.TryReadEnumStrict` for `payloadFormat`/`dataType` (typo → `false` → `BadNodeIdUnknown` upstream). `def.Name = reference`. `Inspect` returns a deploy-time warning list for present-but-invalid enums / unparseable blobs / wildcard tag-topics. Leave Sparkplug-descriptor parsing as a stub the P2 tasks fill (`groupId`/`edgeNodeId`/`deviceId`/`metricName`). 4. Run — expect PASS. 5. Commit: `git commit -am "feat(mqtt): plain tag parser + strict-enum descriptor"` (+ trailer). --- ## Task 3 (P1): `.Driver` project + `MqttConnection` connect/TLS/auth (bounded) **Classification:** standard **Estimated implement time:** ~5 min **Parallelizable with:** none (Task 4 extends `MqttConnection`) **Files:** - Create: `src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.csproj` (net10, refs `.Contracts` + `Core.Abstractions` + `Core` + `MQTTnet`) - Create: `src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt/MqttConnection.cs` (MQTTnet-5 client wrapper: build options from `MqttDriverOptions` incl. TLS + CA-pin + credentials; `ConnectAsync(ct)` under `connectTimeoutSeconds` linked-CTS) - Create: `tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Tests/MqttConnectionTests.cs` - Modify: `ZB.MOM.WW.OtOpcUa.slnx` ### Steps (TDD) 1. Failing test — bounded connect against a dead endpoint fails fast (no hang), and TLS-option assembly honours the knobs. Use a closed loopback port for the deadline test: ```csharp [Fact] public async Task ConnectAsync_DeadBroker_FailsWithinDeadline_NoHang() { var opts = new MqttDriverOptions { Host = "127.0.0.1", Port = 1, UseTls = false, ConnectTimeoutSeconds = 2 }; var conn = new MqttConnection(opts, driverId: "t", logger: null); var sw = Stopwatch.StartNew(); await Should.ThrowAsync(() => conn.ConnectAsync(CancellationToken.None)); sw.Elapsed.ShouldBeLessThan(TimeSpan.FromSeconds(5)); // deadline honoured, not wedged } [Fact] public void BuildClientOptions_UseTlsWithCaPin_SetsTlsAndValidatesChain() { var opts = new MqttDriverOptions { Host="h", Port=8883, UseTls=true, AllowUntrustedServerCertificate=false, CaCertificatePath="/tmp/ca.pem" }; var built = MqttConnection.BuildClientOptions(opts, clientIdSuffix: null); built.ChannelOptions.ShouldBeOfType().TlsOptions.UseTls.ShouldBeTrue(); } ``` 2. Run — expect FAIL. 3. Implement `MqttConnection`: static `BuildClientOptions(opts, clientIdSuffix)` (host/port, protocol version, clientId + optional suffix, keep-alive, clean-session, credentials, TLS with `AllowUntrustedServerCertificate` → custom cert validator, `CaCertificatePath` → chain pin). `ConnectAsync(ct)` links `ct` with a `connectTimeoutSeconds` CTS. Ctor stores options only — connection-free. 4. Run — expect PASS. 5. Commit: `git commit -am "feat(mqtt): MqttConnection connect/TLS/auth under bounded deadline"` (+ trailer). --- ## Task 4 (P1): `MqttConnection` hand-rolled reconnect loop (backoff + resubscribe) **Classification:** high-risk **Estimated implement time:** ~5 min **Parallelizable with:** none **Files:** - Modify: `src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt/MqttConnection.cs` (reconnect loop: exponential backoff `min→max`; on `DisconnectedAsync` schedule reconnect; on reconnect fire a `Reconnected` callback so the driver re-subscribes; expose `State` = Connected/Reconnecting/Faulted + `LastMessageAgeUtc`) - Modify: `tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Tests/MqttConnectionTests.cs` **Why high-risk:** MQTTnet v5 dropped v4's `ManagedMqttClient` (§8) — there is no library-managed reconnect. Subscriptions do not survive a clean session; a reconnect that forgets to re-subscribe silently goes dark. Backoff that ignores its cap can hot-loop a dead broker. ### Steps (TDD) 1. Failing tests — backoff schedule is bounded + monotone-to-cap, and a `Reconnected` event drives a re-subscribe callback. Test the backoff calculator as a pure function (no live broker): ```csharp [Theory] [InlineData(0, 1)] [InlineData(1, 2)] [InlineData(2, 4)] [InlineData(10, 30)] // caps at max=30 public void NextBackoff_IsExponentialClampedToMax(int attempt, int expectedSeconds) => MqttConnection.NextBackoff(attempt, minSeconds: 1, maxSeconds: 30) .ShouldBe(TimeSpan.FromSeconds(expectedSeconds)); [Fact] public async Task OnReconnect_InvokesResubscribeCallback() { var conn = new MqttConnection(new MqttDriverOptions(), "t", null); var fired = 0; conn.Reconnected += () => { fired++; return Task.CompletedTask; }; await conn.RaiseReconnectedForTest(); // test seam fired.ShouldBe(1); } ``` 2. Run — expect FAIL. 3. Implement: `NextBackoff(attempt, min, max)` pure; a reconnect worker that loops on disconnect with backoff, honours `cleanSession`/session-expiry, re-subscribes via the `Reconnected` callback (idempotent insurance even for persistent sessions), and (Sparkplug, P2) re-requests rebirth. `State` transitions Connected↔Reconnecting; Faulted only on unrecoverable config. Never block the MQTTnet dispatcher thread. 4. Run — expect PASS. 5. Commit: `git commit -am "feat(mqtt): hand-rolled reconnect loop with bounded backoff + resubscribe"` (+ trailer). --- ## Task 5 (P1): `LastValueCache` + `IReadable` **Classification:** small **Estimated implement time:** ~5 min **Parallelizable with:** Task 6 **Files:** - Create: `src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt/LastValueCache.cs` (`FullReference → DataValueSnapshot`, thread-safe) - Create: `tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Tests/LastValueCacheTests.cs` ### Steps (TDD) 1. Failing test — unseen reference returns a per-ref `GoodNoData`/uncertain snapshot (never throws the batch); a seeded ref returns last value: ```csharp [Fact] public void Read_UnseenRef_ReturnsPerRefNoData_DoesNotThrow() { var c = new LastValueCache(); var snap = c.Read("factory/oven/temp#$.value"); snap.StatusCode.ShouldBe(StatusCodes.GoodNoData); // or Uncertain — per IReadable contract, per-ref status } [Fact] public void Update_ThenRead_ReturnsLastValue() { var c = new LastValueCache(); c.Update("k", DataValueSnapshot.Good(42.0, DateTime.UtcNow)); c.Read("k").Value.ShouldBe(42.0); } ``` 2. Run — expect FAIL. 3. Implement `LastValueCache` (concurrent dict); `MqttDriver.ReadAsync` (Task 7) returns `references.Select(cache.Read)` — batch never throws, per-ref `StatusCode` carries "not yet observed". 4. Run — expect PASS. 5. Commit: `git commit -am "feat(mqtt): last-value cache backing IReadable (per-ref no-data)"` (+ trailer). --- ## Task 6 (P1): `ISubscribable` — plain topic subscribe → `OnDataChange` + retained seed **Classification:** standard **Estimated implement time:** ~5 min **Parallelizable with:** Task 5 **Files:** - Create: `src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt/MqttSubscriptionManager.cs` (registers `FullReference → MqttTagDefinition` via the shared `EquipmentTagRefResolver`; dedupes/coalesces topics; on message: match topic → authored tag(s), extract value at `jsonPath`/raw/scalar, raise `OnDataChange`; seed from retained message on subscribe) - Modify: `src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt/MqttConnection.cs` (expose `SubscribeAsync(filters, ct)` under a bounded deadline + a message-received event) - Create: `tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Tests/MqttSubscriptionManagerTests.cs` ### Steps (TDD) 1. Failing test — feed a synthetic received message through the manager (no live broker); assert routing + JSONPath extraction fire `OnDataChange` with the right `FullReference`: ```csharp [Fact] public void OnMessage_MatchingTopic_RaisesDataChangeAtJsonPath() { var mgr = new MqttSubscriptionManager(); var handle = mgr.Register(new[] { """{"topic":"f/oven/temp","payloadFormat":"Json","jsonPath":"$.value","dataType":"Double"}""" }); string? gotRef = null; object? gotVal = null; mgr.OnDataChange += (_, e) => { gotRef = e.FullReference; gotVal = e.Snapshot.Value; }; mgr.HandleMessage("f/oven/temp", """{"value":21.5}"""u8.ToArray(), retained: false); gotVal.ShouldBe(21.5); gotRef.ShouldContain("f/oven/temp"); } [Fact] public void OnMessage_UnauthoredTopic_RaisesNothing() // chatty broker must not auto-provision { var mgr = new MqttSubscriptionManager(); mgr.Register(Array.Empty()); var fired = false; mgr.OnDataChange += (_, _) => fired = true; mgr.HandleMessage("random/topic", "1"u8.ToArray(), retained: false); fired.ShouldBeFalse(); } ``` 2. Run — expect FAIL. 3. Implement using `EquipmentTagRefResolver` (parse-once cache, as Modbus does). `HandleMessage` matches topic → tag(s), extracts (Json→JSONPath token→typed value; Raw→bytes-as-string; Scalar→parse), updates `LastValueCache`, raises `OnDataChange`. Retained-flagged messages seed initial value. `SubscribeAsync` returns an `ISubscriptionHandle` whose `DiagnosticId` names mode+filter; completes under a bounded SUBACK deadline (SUBACK failure → per-ref Bad, not hang). 4. Run — expect PASS. 5. Commit: `git commit -am "feat(mqtt): plain subscribe→OnDataChange with retained seed + ref resolver"` (+ trailer). --- ## Task 7 (P1): `MqttDriver` shell — `IDriver` + authored-only `ITagDiscovery` + `IHostConnectivityProbe` **Classification:** standard **Estimated implement time:** ~5 min **Parallelizable with:** none (Task 8/9 wire it) **Files:** - Create: `src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt/MqttDriver.cs` (`IDriver, ITagDiscovery, ISubscribable, IReadable, IHostConnectivityProbe, IRediscoverable`; composes `MqttConnection` + `MqttSubscriptionManager` + `LastValueCache`) - Create: `tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Tests/MqttDriverDiscoveryTests.cs` ### Steps (TDD) 1. Failing test — `DiscoverAsync` streams **only authored tags** into a capturing `IAddressSpaceBuilder`; plain-mode `RediscoverPolicy == Once`; `SupportsOnlineDiscovery == false`: ```csharp [Fact] public async Task DiscoverAsync_Plain_StreamsAuthoredTagsOnly_PolicyOnce() { var driver = new MqttDriver(new MqttDriverOptions { Mode = MqttMode.Plain }, "d", null); driver.SetAuthoredTagsForTest(new[] { """{"topic":"f/t","payloadFormat":"Raw","dataType":"String"}""" }); var b = new CapturingAddressSpaceBuilder(); await driver.DiscoverAsync(b, CancellationToken.None); b.Variables.Count.ShouldBe(1); driver.RediscoverPolicy.ShouldBe(DiscoveryRediscoverPolicy.Once); ((ITagDiscovery)driver).SupportsOnlineDiscovery.ShouldBeFalse(); } ``` 2. Run — expect FAIL. 3. Implement: `InitializeAsync` deserializes options (shared `JsonSerializerOptions`), builds + connects `MqttConnection`, subscribes the mode-appropriate filter. `ReinitializeAsync` applies deltas in place (never crash → Faulted). `ShutdownAsync` disconnects/disposes. `GetHealth` = Connected/Reconnecting/Faulted + last-message-age. `GetMemoryFootprint` = caches; `FlushOptionalCachesAsync` = no-op (birth/alias are correctness state — forbidden to flush; last-value backs `IReadable`). `DiscoverAsync` replays authored tags only; `RediscoverPolicy => Once` (plain). `IRediscoverable.OnRediscoveryNeeded` never fires in plain mode. 4. Run — expect PASS. 5. Commit: `git commit -am "feat(mqtt): MqttDriver shell — lifecycle + authored-only discovery (Once)"` (+ trailer). --- ## Task 8 (P1): `MqttDriverProbe` — CONNECT handshake **Classification:** small **Estimated implement time:** ~5 min **Parallelizable with:** Task 10 **Files:** - Create: `src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt/MqttDriverProbe.cs` (`IDriverProbe`, `DriverType => "Mqtt"`; parse config with the shared options, open a CONNECT under the timeout, green + latency on CONNACK-accepted, targeted error on refused/TLS/auth/timeout — mirrors `ModbusDriverProbe`) - Create: `tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Tests/MqttDriverProbeTests.cs` ### Steps (TDD) 1. Failing test — `DriverType` is `"Mqtt"`; a dead endpoint yields a failed (not thrown) probe result within the deadline: ```csharp [Fact] public void DriverType_IsCanonicalMqtt() => new MqttDriverProbe().DriverType.ShouldBe("Mqtt"); [Fact] public async Task ProbeAsync_DeadBroker_ReturnsFailedResult_WithinDeadline() { var r = await new MqttDriverProbe().ProbeAsync( """{"host":"127.0.0.1","port":1,"useTls":false,"connectTimeoutSeconds":2}""", CancellationToken.None); r.Success.ShouldBeFalse(); r.Message.ShouldNotBeNullOrEmpty(); } ``` 2. Run — expect FAIL. 3. Implement; CONNACK-accepted is the MQTT "device is answering" proof. Use the **shared** `JsonSerializerOptions` (enum-as-name). 4. Run — expect PASS. 5. Commit: `git commit -am "feat(mqtt): MqttDriverProbe CONNECT handshake"` (+ trailer). --- ## Task 9 (P1): Factory + `DriverTypeNames.Mqtt` + Host registration **Classification:** standard **Estimated implement time:** ~5 min **Parallelizable with:** none **Files:** - Create: `src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt/MqttDriverFactoryExtensions.cs` (`DriverTypeName = "Mqtt"`, shared `JsonSerializerOptions`, `Register(registry, loggerFactory)` — mirror `OpcUaClientDriverFactoryExtensions`) - Modify: `src/Core/ZB.MOM.WW.OtOpcUa.Core.Abstractions/DriverTypeNames.cs` (add `public const string Mqtt = "Mqtt";` + append to the `All` list) - Modify: `src/Server/ZB.MOM.WW.OtOpcUa.Host/Drivers/DriverFactoryBootstrap.cs` (add `Driver.Mqtt.MqttDriverFactoryExtensions.Register(registry, loggerFactory);` in `Register(...)` near line 146; add `using MqttProbe = Driver.Mqtt.MqttDriverProbe;` + `services.TryAddEnumerable(ServiceDescriptor.Singleton());` in `AddOtOpcUaDriverProbes` near line 124) - Modify: `src/Server/ZB.MOM.WW.OtOpcUa.Host/ZB.MOM.WW.OtOpcUa.Host.csproj` (ProjectReference the `.Driver` assembly) - Create: `tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Tests/MqttDriverFactoryExtensionsTests.cs` ### Steps (TDD) 1. Failing test — `Register` binds the `"Mqtt"` type, and the factory builds an `MqttDriver` from JSON with string enums: ```csharp [Fact] public void Register_ThenCreate_BuildsMqttDriver() { var registry = new DriverFactoryRegistry(); MqttDriverFactoryExtensions.Register(registry); var d = registry.Create("Mqtt", "d1", """{"host":"h","port":1883,"mode":"Plain"}"""); d.ShouldBeOfType(); } [Fact] public void DriverTypeName_MatchesConstant() => MqttDriverFactoryExtensions.DriverTypeName.ShouldBe(DriverTypeNames.Mqtt); ``` 2. Run — expect FAIL. 3. Implement + wire both host sites. Build the whole solution to confirm registration compiles: ```bash dotnet build ZB.MOM.WW.OtOpcUa.slnx # expect: Build succeeded ``` 4. Run — expect PASS. 5. Commit: `git commit -am "feat(mqtt): factory + DriverTypeNames.Mqtt + host factory/probe registration"` (+ trailer). --- ## Task 10 (P1): `.Browser` — bespoke `#`-observation browser (passive) **Classification:** standard **Estimated implement time:** ~5 min **Parallelizable with:** Task 8 **Files:** - Create: `src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Browser/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Browser.csproj` (net10, refs `.Contracts` + `Commons(.Browsing)` + `MQTTnet`) - Create: `src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Browser/MqttDriverBrowser.cs` (`IDriverBrowser`, `DriverType => "Mqtt"`; `OpenAsync` connects with a `{clientId}-browse-{guid8}` suffix under a clamped 5–30 s budget; subscribes `#`/`{topicPrefix}#`; returns `MqttBrowseSession`; **publishes nothing**) - Create: `src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Browser/MqttBrowseSession.cs` (`IBrowseSession` over a thread-safe accumulating observed tree; `RootAsync`/`ExpandAsync` split topic segments on `/`, leaf = `Kind=Leaf`; `AttributesAsync` = synthetic attribute w/ inferred type + last payload snippet; `DisposeAsync` disconnects best-effort) - Modify: `ZB.MOM.WW.OtOpcUa.slnx` - Create: `tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Tests/MqttBrowseSessionTests.cs` ### Steps (TDD) 1. Failing test — feed observed topics into the session's accumulating tree (test seam, no live broker); assert `RootAsync`/`ExpandAsync` build the segment tree and **no publish** occurs on any browse call: ```csharp [Fact] public async Task ExpandAsync_BuildsTopicSegmentTree_FromObservedTopics() { var s = new MqttBrowseSession(MqttMode.Plain); s.ObserveTopicForTest("factory/line3/oven/temp"); var root = await s.RootAsync(CancellationToken.None); root.ShouldContain(n => n.BrowseName == "factory"); var lvl2 = await s.ExpandAsync("factory", CancellationToken.None); lvl2.ShouldContain(n => n.BrowseName == "line3"); } [Fact] public async Task BrowseCalls_PublishNothing() // browse is read-only { var s = new MqttBrowseSession(MqttMode.Plain); await s.RootAsync(CancellationToken.None); s.PublishCountForTest.ShouldBe(0); } ``` 2. Run — expect FAIL. 3. Implement the accumulating tree + passive `OpenAsync`. In P1 only the plain `#` path is live; leave a Sparkplug hook the P2 tasks fill (`Group→EdgeNode→Device→Metric` + `RequestRebirthAsync`). 4. Run — expect PASS. 5. Commit: `git commit -am "feat(mqtt): bespoke passive #-observation browser (plain)"` (+ trailer). --- ## Task 11 (P1): Register the bespoke browser (overrides universal fallback) **Classification:** trivial **Estimated implement time:** ~3 min **Parallelizable with:** Task 12 **Files:** - Modify: `src/Server/ZB.MOM.WW.OtOpcUa.AdminUI/EndpointRouteBuilderExtensions.cs` (add `services.AddSingleton();` alongside the existing two near line 75–76) - Modify: `src/Server/ZB.MOM.WW.OtOpcUa.AdminUI/ZB.MOM.WW.OtOpcUa.AdminUI.csproj` (ProjectReference `.Browser`) ### Steps 1. Add the registration + project reference. Registering for `DriverType="Mqtt"` overrides the universal fallback (`BrowserSessionService` resolves bespoke-first). 2. Build: `dotnet build src/Server/ZB.MOM.WW.OtOpcUa.AdminUI` — expect success. 3. Commit: `git commit -am "feat(mqtt): register MqttDriverBrowser (bespoke-first for Mqtt)"` (+ trailer). --- ## Task 12 (P1): Typed AdminUI editor + validator (plain shape) **Classification:** standard **Estimated implement time:** ~5 min **Parallelizable with:** Task 11 **Files:** - Create: `src/Server/ZB.MOM.WW.OtOpcUa.AdminUI/Uns/TagEditors/MqttTagConfigModel.cs` (thin typed model over a preserved `JsonObject` key bag; `FromJson`/`ToJson`/`Validate`, preserves unknown keys incl. history keys; carries a `Mode` field selecting sub-shape — P1 implements Plain `Validate`, Sparkplug stub filled in Task 24) - Modify: `src/Server/ZB.MOM.WW.OtOpcUa.AdminUI/Uns/TagEditors/TagConfigEditorMap.cs` (`[DriverTypeNames.Mqtt] = typeof(Components.Shared.Uns.TagEditors.MqttTagConfigEditor)`) - Modify: `src/Server/ZB.MOM.WW.OtOpcUa.AdminUI/Uns/TagEditors/TagConfigValidator.cs` (`[DriverTypeNames.Mqtt] = j => MqttTagConfigModel.FromJson(j).Validate()`) - Create: `src/Server/ZB.MOM.WW.OtOpcUa.AdminUI/Components/Shared/Uns/TagEditors/MqttTagConfigEditor.razor` (copy the Modbus editor template — mode-switch at top, per-mode field group, client-side `Validate()`) - Create: `tests/Server/.../MqttTagConfigModelTests.cs` (co-locate with the existing AdminUI test project — verify path with `find tests/Server -name "*TagConfigModel*Tests.cs"`) ### Steps (TDD) 1. Failing test — round-trip preserves unknown/history keys; plain `Validate` requires concrete topic + jsonPath-when-Json; re-derives `FullName`: ```csharp [Fact] public void FromJson_ToJson_PreservesUnknownKeys() { var m = MqttTagConfigModel.FromJson("""{"topic":"a/b","payloadFormat":"Raw","dataType":"String","isHistorized":true}"""); m.ToJson().ShouldContain("isHistorized"); } [Fact] public void Validate_PlainWildcardTopic_Fails() => MqttTagConfigModel.FromJson("""{"topic":"a/+/c","payloadFormat":"Raw","dataType":"String"}""") .Validate().ShouldNotBeEmpty(); [Fact] public void Validate_JsonWithoutPath_Fails() => MqttTagConfigModel.FromJson("""{"topic":"a/b","payloadFormat":"Json","dataType":"Double"}""") .Validate().ShouldNotBeEmpty(); ``` 2. Run — expect FAIL. 3. Implement the model (mirror `OpcUaClientTagConfigModel`); `ToJson` writes PascalCase `FullName` re-derived from descriptor (`{topic}#{jsonPath}` plain). Build the `.razor`. Register both map entries. 4. Run — expect PASS + `dotnet build src/Server/ZB.MOM.WW.OtOpcUa.AdminUI`. 5. Commit: `git commit -am "feat(mqtt): typed tag editor + validator (plain mode)"` (+ trailer). --- ## Task 13 (P1): Mosquitto + JSON-publisher fixture (TLS+auth) + env-gated live suite **Classification:** standard **Estimated implement time:** ~5 min **Parallelizable with:** none **Files:** - Create: `tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.IntegrationTests/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.IntegrationTests.csproj` - Create: `tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.IntegrationTests/Docker/docker-compose.yml` (`eclipse-mosquitto` with **auth + TLS** on `:8883`, plain `:1883` for smoke only; a JSON-publisher container emitting `retain=true` JSON on a few topics — `mosquitto_pub` loop or `paho-mqtt` script; `project=lmxopcua` label applied host-side) - Create: `tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.IntegrationTests/Docker/mosquitto.conf` + password/cert material generator script (creds via env, never real secrets committed) - Create: `tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.IntegrationTests/PlainMqttLiveTests.cs` (category `LiveIntegration`, gated on `MQTT_FIXTURE_ENDPOINT` — skips clean when unset → macOS-offline-safe) - Modify: `ZB.MOM.WW.OtOpcUa.slnx` ### Steps 1. Write the compose with **auth + TLS** (never anonymous). Add the JSON publisher with `retain=true`. 2. Write env-gated tests: connect/TLS/auth; plain subscribe + retained-read seed; unauthored-topic silence. `[SkippableFact]` reading `MQTT_FIXTURE_ENDPOINT`. 3. Confirm offline skip: ```bash dotnet test tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.IntegrationTests # expect: all Skipped (no env var) ``` 4. Deploy the fixture to the docker host and run live (design §9): ```bash lmxopcua-fix sync mqtt && lmxopcua-fix up mqtt export MQTT_FIXTURE_ENDPOINT=10.100.0.35:8883 dotnet test tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.IntegrationTests --filter "Category=LiveIntegration" # expect: PASS ``` 5. Commit: `git commit -am "test(mqtt): Mosquitto TLS+auth fixture + env-gated plain live suite"` (+ trailer). --- ## Task 14 (P1): Live `/run` verify on docker-dev — **P1 MILESTONE COMPLETE** **Classification:** small **Estimated implement time:** ~5 min **Parallelizable with:** none **Files:** - Modify: `CLAUDE.md` (add the MQTT fixture endpoint to the Docker Workflow endpoint list: `10.100.0.35:1883/8883`, `MQTT_FIXTURE_ENDPOINT`) - Modify: `docs/plans/2026-07-15-driver-expansion-tracking.md` (mark MQTT P1 code-complete) ### Steps 1. On docker-dev (`:9200`, login disabled — run it yourself, do not defer): author an `Mqtt` driver (Plain mode) + a tag via the `/uns` picker (bespoke `#`-observation browser) or the typed editor; deploy. 2. Confirm via Client.CLI the node carries live broker values: ```bash dotnet run --project src/Client/ZB.MOM.WW.OtOpcUa.Client.CLI -- read -u opc.tcp://localhost:4840 -n "ns=2;s=" ``` 3. Confirm the typed editor renders (Razor binding bugs pass unit tests — live-verify is mandatory). Update CLAUDE.md + tracking doc. 4. Commit: `git commit -am "docs(mqtt): P1 plain-MQTT milestone live-verified; endpoint recorded"` (+ trailer). --- # ── P2: Sparkplug B ingest (builds on the P1 skeleton) ── ## Task 15 (P2): Vendor Tahu proto + `Grpc.Tools` codegen **Classification:** standard **Estimated implement time:** ~5 min **Parallelizable with:** none (P2 root) **Files:** - Create: `src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Contracts/Protos/sparkplug_b.proto` (vendored Eclipse Tahu `sparkplug_b.proto`, pinned to a known Tahu commit with a provenance comment header) - Modify: `src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Contracts/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Contracts.csproj` (add `Google.Protobuf` + build-time `Grpc.Tools` (`PrivateAssets=all`); `` — message-only codegen, this repo's first in-repo protoc step) - Create: `tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Tests/SparkplugProtoCodegenTests.cs` ### Steps (TDD) 1. Failing test — the generated `Payload`/`Metric` types exist and round-trip a hand-built payload: ```csharp [Fact] public void GeneratedPayload_RoundTrips() { var p = new Org.Eclipse.Tahu.Protobuf.Payload { Seq = 3 }; p.Metrics.Add(new Org.Eclipse.Tahu.Protobuf.Payload.Types.Metric { Name = "Temperature", Alias = 5 }); var back = Org.Eclipse.Tahu.Protobuf.Payload.Parser.ParseFrom(p.ToByteArray()); back.Seq.ShouldBe(3ul); back.Metrics[0].Alias.ShouldBe(5ul); } ``` 2. Run — expect FAIL (no generated types). 3. Vendor the proto (provenance comment: Tahu commit hash + URL); wire `Grpc.Tools`. If in-repo protoc proves objectionable, fall back to checking in the generated C# with the same provenance comment (design §2.1). Build: ```bash dotnet build src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Contracts # expect: proto compiles, 0 warnings ``` 4. Run — expect PASS. 5. Commit: `git commit -am "feat(mqtt): vendor Tahu sparkplug_b.proto + Grpc.Tools codegen"` (+ trailer). --- ## Task 16 (P2): `SparkplugCodec` decode + golden payloads **Classification:** standard **Estimated implement time:** ~5 min **Parallelizable with:** Task 17 **Files:** - Create: `src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt/Sparkplug/SparkplugCodec.cs` (decode `Payload`/`Metric` from wire bytes → a driver-side struct; encode NCMD deferred to `RebirthRequester` Task 20 / write-through P3) - Create: `tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Tests/SparkplugCodecTests.cs` - Create: `tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Tests/Golden/nbirth.bin` + `ndata.bin` (golden byte vectors, generated once from a hand-built payload and committed) ### Steps (TDD) 1. Failing test — decode a golden NBIRTH: seq + metric name/alias/datatype/value survive: ```csharp [Fact] public void Decode_GoldenNbirth_ExtractsMetrics() { var payload = SparkplugCodec.Decode(File.ReadAllBytes("Golden/nbirth.bin")); payload.Seq.ShouldBe((byte)0); payload.Metrics.ShouldContain(m => m.Name == "Temperature" && m.Alias == 5 && m.DataType == SparkplugDataType.Float); } ``` 2. Run — expect FAIL. 3. Implement `SparkplugCodec.Decode`; generate the golden vectors in a one-off `[Fact(Skip="generator")]` or a small helper, commit the `.bin` files. 4. Run — expect PASS. 5. Commit: `git commit -am "feat(mqtt): SparkplugCodec decode + golden payload vectors"` (+ trailer). --- ## Task 17 (P2): `SparkplugTopic` + `SparkplugDataType.ToDriverDataType` **Classification:** small **Estimated implement time:** ~5 min **Parallelizable with:** Task 16 **Files:** - Create: `src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt/Sparkplug/SparkplugTopic.cs` (parse/format `spBv1.0/{group}/{type}/{node}[/{device}]`; `type` ∈ NBIRTH/DBIRTH/NDATA/DDATA/NDEATH/DDEATH/NCMD/DCMD/STATE) - Modify: `src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Contracts/SparkplugDataType.cs` (enum + `ToDriverDataType()` per the §3.5 map: Int8→Int16, UInt8→UInt16, DataSet/Template unsupported, `*Array`→element+IsArray) - Create: `tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Tests/SparkplugTopicTests.cs` ### Steps (TDD) 1. Failing tests — topic parse extracts group/type/node/device; datatype map widens Int8→Int16 and marks DataSet unsupported: ```csharp [Fact] public void Parse_DeviceData_ExtractsAllSegments() { var t = SparkplugTopic.Parse("spBv1.0/Plant1/DDATA/EdgeA/Filler1"); t.GroupId.ShouldBe("Plant1"); t.Type.ShouldBe(SparkplugMessageType.DDATA); t.EdgeNodeId.ShouldBe("EdgeA"); t.DeviceId.ShouldBe("Filler1"); } [Theory] [InlineData(SparkplugDataType.Int8, DriverDataType.Int16)] [InlineData(SparkplugDataType.UInt8, DriverDataType.UInt16)] [InlineData(SparkplugDataType.Float, DriverDataType.Float32)] public void ToDriverDataType_MapsAndWidens(SparkplugDataType s, DriverDataType d) => s.ToDriverDataType().ShouldBe(d); ``` 2. Run — expect FAIL. 3. Implement. 4. Run — expect PASS. 5. Commit: `git commit -am "feat(mqtt): Sparkplug topic parse + datatype map"` (+ trailer). --- ## Task 18 (P2): `BirthCache` + `AliasTable` — bind-by-name, rebuild-per-birth **Classification:** high-risk **Estimated implement time:** ~5 min **Parallelizable with:** Task 19 **Files:** - Create: `src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt/Sparkplug/AliasTable.cs` (per `(edgeNode,device)` alias→(name,datatype); **rebuilt wholesale each birth, never merged**) - Create: `src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt/Sparkplug/BirthCache.cs` (NBIRTH/DBIRTH metric catalog: name/alias/datatype/last value) - Create: `tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Tests/AliasBindingTests.cs` **Why high-risk:** §3.6 invariants #1–#2 — binding data by alias silently mis-routes when an alias is reused for a different metric across a rebirth. Bind by **stable metric NAME**; the alias is a per-birth cache only. ### Steps (TDD) 1. Failing tests — the two load-bearing invariants: ```csharp [Fact] public void Rebirth_ReusesAliasForDifferentMetric_ResolvesByName_NotAlias() { var t = new AliasTable(); t.RebuildFromBirth(new[] { (name:"Temperature", alias:5ul, dt:SparkplugDataType.Float) }); t.Resolve(alias: 5).Name.ShouldBe("Temperature"); // rebirth: alias 5 now means a DIFFERENT metric t.RebuildFromBirth(new[] { (name:"Pressure", alias:5ul, dt:SparkplugDataType.Float) }); t.Resolve(alias: 5).Name.ShouldBe("Pressure"); // wholesale replace, not merge } [Fact] public void RebuildFromBirth_DoesNotMergeStaleAliases() { var t = new AliasTable(); t.RebuildFromBirth(new[] { (name:"A", alias:1ul, dt:SparkplugDataType.Int32) }); t.RebuildFromBirth(new[] { (name:"B", alias:2ul, dt:SparkplugDataType.Int32) }); t.TryResolve(alias: 1, out _).ShouldBeFalse(); // alias 1 gone after rebirth } ``` 2. Run — expect FAIL. 3. Implement — `RebuildFromBirth` replaces the whole map. `BirthCache` keeps the metric catalog + last value keyed by name. 4. Run — expect PASS. 5. Commit: `git commit -am "feat(mqtt): AliasTable/BirthCache — bind-by-name, rebuild-per-birth"` (+ trailer). --- ## Task 19 (P2): `SequenceTracker` — seq-gap + bdSeq death-tie **Classification:** high-risk **Estimated implement time:** ~5 min **Parallelizable with:** Task 18 **Files:** - Create: `src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt/Sparkplug/SequenceTracker.cs` (per edge-node `seq` 0–255 wrap gap detection; `bdSeq` ties NDEATH↔NBIRTH) - Create: `tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Tests/SequenceTrackerTests.cs` **Why high-risk:** §3.6 invariant #3 — a wrong wrap boundary (255→0 is NOT a gap) either misses real gaps (stale data) or false-positives every wrap (rebirth-storm). ### Steps (TDD) 1. Failing tests — contiguous ok, wrap ok, gap detected: ```csharp [Fact] public void Sequence_WrapAt255IsContiguous_NotAGap() { var s = new SequenceTracker(); s.Accept(254).ShouldBeTrue(); s.Accept(255).ShouldBeTrue(); s.Accept(0).ShouldBeTrue(); // wrap } [Fact] public void Sequence_SkippedValue_IsGap() { var s = new SequenceTracker(); s.Accept(10).ShouldBeTrue(); s.Accept(12).ShouldBeFalse(); // gap → caller requests rebirth } ``` 2. Run — expect FAIL. 3. Implement (next-expected = `(last+1) & 0xFF`); `bdSeq` compare helper. 4. Run — expect PASS. 5. Commit: `git commit -am "feat(mqtt): SequenceTracker seq-gap + bdSeq death-tie"` (+ trailer). --- ## Task 20 (P2): `RebirthRequester` — NCMD encode **Classification:** small **Estimated implement time:** ~5 min **Parallelizable with:** none **Files:** - Create: `src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt/Sparkplug/RebirthRequester.cs` (encode an NCMD writing `Node Control/Rebirth = true` to `spBv1.0/{group}/NCMD/{node}`; publish under a bounded deadline) - Create: `tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Tests/RebirthRequesterTests.cs` ### Steps (TDD) 1. Failing test — encoded NCMD carries the `Node Control/Rebirth`=true metric and targets the right topic: ```csharp [Fact] public void BuildRebirthNcmd_EncodesControlMetric_AndTopic() { var (topic, bytes) = RebirthRequester.Build("Plant1", "EdgeA"); topic.ShouldBe("spBv1.0/Plant1/NCMD/EdgeA"); var p = SparkplugCodec.Decode(bytes); p.Metrics.ShouldContain(m => m.Name == "Node Control/Rebirth" && Equals(m.Value, true)); } ``` 2. Run — expect FAIL. 3. Implement (encode path via the generated proto). 4. Run — expect PASS. 5. Commit: `git commit -am "feat(mqtt): RebirthRequester NCMD encode"` (+ trailer). --- ## Task 21 (P2): Sparkplug ingest state machine — the §3.6 matrix **Classification:** high-risk **Estimated implement time:** ~5 min **Parallelizable with:** none (integrates Tasks 16–20 into the driver) **Files:** - Create: `src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt/Sparkplug/SparkplugIngestor.cs` (routes decoded messages: N/DBIRTH → rebuild `AliasTable` + `BirthCache`; N/DDATA → resolve alias→name → map to authored `FullReference` **by (group,node,device,metricName)** → `OnDataChange`; N/DDEATH → emit STALE/Bad snapshots; seq-gap / unknown-alias / data-before-birth → `RebirthRequester` gated on `requestRebirthOnGap`; STATE/primary-host handling; late-join rebirth on connect) - Modify: `src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt/MqttDriver.cs` (Sparkplug-mode subscribe `spBv1.0/{group}/#` (+ STATE) → `SparkplugIngestor`; reconnect → re-subscribe + request rebirth) - Create: `tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Tests/SparkplugIngestorTests.cs` **Why high-risk:** this is the §3.6 correctness core — the #1 risk in the design. Test the full matrix: rebirth, missed-birth, seq-gap, alias-reuse-across-rebirth, death→stale→rebirth. ### Steps (TDD) 1. Failing tests — the §3.6 matrix, driven through the ingestor with synthetic decoded messages (no live broker): ```csharp [Fact] public void Birth_Then_Data_ResolvesByAlias_RaisesOnDataChangeByName() { var ing = new SparkplugIngestor(...); // authored tag: Plant1/EdgeA/Filler1:Temperature string? firedRef = null; ing.OnDataChange += (_, e) => firedRef = e.FullReference; ing.HandleDbirth("Plant1","EdgeA","Filler1", new[] { (name:"Temperature", alias:5ul, dt:SparkplugDataType.Float, val:(object)0f) }); ing.HandleDdata("Plant1","EdgeA","Filler1", seq:1, new[] { (alias:5ul, val:(object)21.5f) }); firedRef.ShouldContain("Filler1:Temperature"); } [Fact] public void DataBeforeBirth_RequestsRebirth() { var ncmds = new List(); var ing = new SparkplugIngestor(..., publishNcmd: (t,_) => ncmds.Add(t)); ing.HandleDdata("Plant1","EdgeA","Filler1", seq:0, new[] { (alias:9ul, val:(object)1f) }); ncmds.ShouldContain("spBv1.0/Plant1/NCMD/EdgeA"); } [Fact] public void Ddeath_EmitsStaleForDeviceMetrics_NextBirthRestoresGood() { var ing = new SparkplugIngestor(...); var quals = new List(); ing.OnDataChange += (_, e) => quals.Add(e.Snapshot.StatusCode); ing.HandleDbirth("Plant1","EdgeA","Filler1", new[] { (name:"Temperature", alias:5ul, dt:SparkplugDataType.Float, val:(object)0f) }); ing.HandleDdeath("Plant1","EdgeA","Filler1"); quals.ShouldContain(q => StatusCode.IsBad(q)); // STALE/Bad on death } ``` 2. Run — expect FAIL. 3. Implement the ingestor holding the §3.6 invariants. Route via `EquipmentTagRefResolver` keyed by `(group,node,device,metricName)`. 4. Run — expect PASS (all matrix cases). 5. Commit: `git commit -am "feat(mqtt): Sparkplug ingest state machine (birth/alias/seq/rebirth/death→stale)"` (+ trailer). --- ## Task 22 (P2): `ITagDiscovery` `UntilStable` + `IRediscoverable` **Classification:** standard **Estimated implement time:** ~5 min **Parallelizable with:** none **Files:** - Modify: `src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt/MqttDriver.cs` (`RediscoverPolicy => UntilStable` in Sparkplug mode — authored tags' datatypes fill in as births arrive; `DiscoverAsync` re-streams authored tags with resolved datatypes from `BirthCache`; fire `OnRediscoveryNeeded` on a **new DBIRTH** or a rebirth with a changed metric set, `ScopeHint` = edge-node/device folder path) - Modify: `tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Tests/MqttDriverDiscoveryTests.cs` ### Steps (TDD) 1. Failing tests — Sparkplug policy is `UntilStable`; a new DBIRTH fires `OnRediscoveryNeeded`; a tag authored before its birth picks up the birth datatype: ```csharp [Fact] public void SparkplugMode_RediscoverPolicy_IsUntilStable() => new MqttDriver(new MqttDriverOptions { Mode = MqttMode.SparkplugB }, "d", null) .RediscoverPolicy.ShouldBe(DiscoveryRediscoverPolicy.UntilStable); [Fact] public async Task NewDbirth_FiresOnRediscoveryNeeded() { var d = new MqttDriver(new MqttDriverOptions { Mode = MqttMode.SparkplugB }, "d", null); var fired = false; ((IRediscoverable)d).OnRediscoveryNeeded += (_, _) => fired = true; d.SimulateNewDbirthForTest("Plant1","EdgeA","Filler2"); fired.ShouldBeTrue(); } ``` 2. Run — expect FAIL. 3. Implement. 4. Run — expect PASS. 5. Commit: `git commit -am "feat(mqtt): Sparkplug UntilStable discovery + rediscover-on-DBIRTH"` (+ trailer). --- ## Task 23 (P2): Sparkplug browser tree + `AttributesAsync` + `RequestRebirthAsync` **Classification:** standard **Estimated implement time:** ~5 min **Parallelizable with:** Task 24 **Files:** - Modify: `src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Browser/MqttBrowseSession.cs` (Sparkplug tree Group→EdgeNode→Device→Metric from observed births; `AttributesAsync` = self-describing metric `AttributeInfo{Name, DriverDataType from birth, IsArray, SecurityClass}`; `RequestRebirthAsync(scope)` — the **only** publishing session member, scoped to selected group/edge-node, via the `RebirthRequester` path) - Modify: `src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Browser/MqttDriverBrowser.cs` (Sparkplug discovery filter `spBv1.0/{groupId}/#`; `OpenAsync` still publishes nothing) - Modify: `src/Server/ZB.MOM.WW.OtOpcUa.AdminUI/Browsing/BrowserSessionService.cs` (dispatch the MQTT-specific `RequestRebirthAsync` for MQTT sessions, gated by the same `DriverOperator` policy that gates the picker; Info-log the target scope) - Modify: `tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Tests/MqttBrowseSessionTests.cs` ### Steps (TDD) 1. Failing tests — birth-driven tree fills; `OpenAsync`/`RootAsync`/`ExpandAsync` publish **nothing**; `RequestRebirthAsync` publishes exactly one scoped NCMD: ```csharp [Fact] public async Task SparkplugTree_FillsFromObservedBirths_BrowseNeverPublishes() { var s = new MqttBrowseSession(MqttMode.SparkplugB); s.ObserveBirthForTest("Plant1","EdgeA","Filler1","Temperature", SparkplugDataType.Float); var groups = await s.RootAsync(default); groups.ShouldContain(n => n.BrowseName == "Plant1"); s.PublishCountForTest.ShouldBe(0); // passive } [Fact] public async Task RequestRebirthAsync_ScopedToNode_PublishesOneNcmd() { var s = new MqttBrowseSession(MqttMode.SparkplugB); s.ObserveBirthForTest("Plant1","EdgeA","Filler1","T", SparkplugDataType.Float); await s.RequestRebirthAsync(scope: "Plant1/EdgeA"); s.PublishCountForTest.ShouldBe(1); } ``` 2. Run — expect FAIL. 3. Implement. `RequestRebirthAsync` is never fired by `OpenAsync`/`RootAsync`/`ExpandAsync` — only on explicit operator click. 4. Run — expect PASS + `dotnet build` the AdminUI. 5. Commit: `git commit -am "feat(mqtt): Sparkplug birth-driven browser tree + scoped Request-rebirth action"` (+ trailer). --- ## Task 24 (P2): AdminUI editor Sparkplug mode shape + validator **Classification:** small **Estimated implement time:** ~5 min **Parallelizable with:** Task 23 **Files:** - Modify: `src/Server/ZB.MOM.WW.OtOpcUa.AdminUI/Uns/TagEditors/MqttTagConfigModel.cs` (Sparkplug `Validate`: `groupId`/`edgeNodeId`/`metricName` required, `deviceId` optional, `dataType` a known Sparkplug type; `ToJson` re-derives `FullName` = `{group}/{node}[/{device}]:{metric}`) - Modify: `src/Server/ZB.MOM.WW.OtOpcUa.AdminUI/Components/Shared/Uns/TagEditors/MqttTagConfigEditor.razor` (Sparkplug field group under the mode switch) - Modify: `tests/Server/.../MqttTagConfigModelTests.cs` ### Steps (TDD) 1. Failing tests — Sparkplug validation + FullName derivation: ```csharp [Fact] public void Validate_SparkplugMissingMetricName_Fails() => MqttTagConfigModel.FromJson("""{"groupId":"Plant1","edgeNodeId":"EdgeA"}""").Validate().ShouldNotBeEmpty(); [Fact] public void ToJson_Sparkplug_DerivesFullName() => MqttTagConfigModel.FromJson("""{"groupId":"Plant1","edgeNodeId":"EdgeA","deviceId":"Filler1","metricName":"Temperature","dataType":"Float"}""") .ToJson().ShouldContain("Plant1/EdgeA/Filler1:Temperature"); ``` 2. Run — expect FAIL. 3. Implement. 4. Run — expect PASS + AdminUI build. 5. Commit: `git commit -am "feat(mqtt): typed tag editor Sparkplug mode + validation"` (+ trailer). --- ## Task 25 (P2): C# edge-node simulator fixture + §3.6 live matrix **Classification:** standard **Estimated implement time:** ~5 min **Parallelizable with:** none **Files:** - Create: `tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.IntegrationTests/SparkplugSimulator/` (a project-owned C# edge-node simulator using the **same** MQTTnet-5 + Tahu-proto path the driver uses — publishes NBIRTH/DBIRTH → periodic N/DDATA; honours a rebirth NCMD; injects seq-gap / alias-reuse / N/DDEATH on demand) - Modify: `tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.IntegrationTests/Docker/docker-compose.yml` (add the simulator container against the same TLS+auth Mosquitto) - Create: `tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.IntegrationTests/SparkplugLiveTests.cs` (category `LiveIntegration`, env-gated: birth→data→alias-resolve→OnDataChange; rebirth recovery; death→STALE; browser passive window asserts **no NCMD published by open/browse**; explicit `RequestRebirthAsync` node-vs-group enumeration) ### Steps 1. Build the simulator (encode via the same generated proto — validates encode/decode symmetry). 2. Write env-gated live tests covering the §3.6 matrix + browser passivity. 3. Offline skip proof: ```bash dotnet test tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.IntegrationTests # expect: Skipped without MQTT_FIXTURE_ENDPOINT ``` 4. Deploy + run live: ```bash lmxopcua-fix sync mqtt && lmxopcua-fix up mqtt sparkplug export MQTT_FIXTURE_ENDPOINT=10.100.0.35:8883 dotnet test tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.IntegrationTests --filter "Category=LiveIntegration" # expect: PASS ``` 5. Commit: `git commit -am "test(mqtt): C# Sparkplug edge-node simulator + §3.6 live matrix"` (+ trailer). --- ## Task 26 (P2): Live `/run` verify Sparkplug — **P2 COMPLETE** **Classification:** small **Estimated implement time:** ~5 min **Parallelizable with:** none **Files:** - Modify: `docs/plans/2026-07-15-driver-expansion-tracking.md` (mark MQTT/Sparkplug P1+P2 code-complete + live-verified) - Modify: `CLAUDE.md` (if a driver-fleet or endpoint fact changed; propagate to `../scadaproj/CLAUDE.md` OtOpcUa entry per the cross-repo rule) ### Steps 1. On docker-dev (`:9200`): author an `Mqtt` driver (SparkplugB mode) against the simulator; browse the birth-driven picker, click **Request rebirth**, pick a metric, deploy. 2. Confirm live values + rediscover-on-DBIRTH + death→STALE via Client.CLI reads/subscribe. 3. Confirm the typed editor renders both modes (live-verify — Razor bugs pass unit tests). 4. Update tracking + (if needed) CLAUDE.md/scadaproj index. 5. Commit: `git commit -am "docs(mqtt): Sparkplug P2 live-verified; MQTT driver complete"` (+ trailer). --- ## Deferred / out of scope - **Write-through (P3 in the design):** `IWritable` — Sparkplug NCMD/DCMD + plain publish, optimistic-Good + self-correct on echo (mirrors Galaxy fire-and-forget), `WriteIdempotent` default-off for non-idempotent Sparkplug commands, write-topic config. See design §3.7 + §10 (P3 row). `MqttDriver` ships **read/subscribe/discover only** in P1+P2 — its absence of `IWritable` means nodes materialize read-only. - **`DataSet`/`Template` Sparkplug metrics** — unsupported v1 (skip + warn, or raw-JSON as String). §3.5. - **`Bytes`/`File` metrics** beyond a base64/raw-String fallback. §3.5. - **EMQX broker** — Mosquitto is the default fixture; EMQX is the documented alternative when a dashboard/built-in Sparkplug tooling helps (§9), not built here. ## Notes on ordering / parallelism - **P2 is blocked on the P1 milestone (Task 14).** Every P2 task's `blockedBy` chain roots at Task 14 so plain MQTT ships and is live-verified before Sparkplug work begins — and MQTTnet-5/net10 is proven (Task 0) before any proto/Sparkplug effort. - Genuine parallel pairs: 5‖6, 8‖10, 11‖12, 16‖17, 18‖19, 23‖24. - DRY: reuse `EquipmentTagRefResolver` (as Modbus/OpcUaClient do), the shared `JsonSerializerOptions`, and one `SparkplugCodec` for both decode (ingest) and encode (rebirth NCMD). YAGNI: no write-through plumbing, no DataSet/Template, no EMQX until a need lands.