lmxopcua/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.IntegrationTests/SparkplugLiveTests.cs

using System.Buffers;
using System.Collections.Concurrent;
using System.Text.Json;
using MQTTnet;
using Org.Eclipse.Tahu.Protobuf;
using Shouldly;
using Xunit;
using ZB.MOM.WW.OtOpcUa.Commons.Browsing;
using ZB.MOM.WW.OtOpcUa.Core.Abstractions;
using ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Browser;
using ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Sparkplug;
using ZB.MOM.WW.OtOpcUa.Driver.Mqtt.SparkplugSimulator;

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

/// <summary>
///     The Sparkplug B live gate: the §3.6 state-machine matrix driven end-to-end over the real
///     Mosquitto fixture, against a real edge node (<see cref="SparkplugEdgeNode"/>) that encodes with
///     the same generated Tahu schema the driver decodes with.
///     <para>
///         <b>What the live gate adds over the offline suite.</b> Every case here already has an
///         offline pin driven through <c>SparkplugIngestor.Dispatch</c> with hand-built decoded
///         payloads. Those prove the <i>logic</i>. They cannot prove the <b>wire</b>: that a negative
///         Int32 really does arrive as an unsigned two's-complement <c>int_value</c>, that a DATA
///         metric really does carry an alias and nothing else, that an NDEATH really is a broker-issued
///         Will with no <c>seq</c>, or that the NCMD this driver encodes is one an independent decoder
///         accepts as a rebirth request. A hand-built <c>SparkplugPayload</c> shaped by the same
///         assumptions as the decoder cannot falsify any of that; a byte stream through a broker can.
///     </para>
///     <para>
///         <b>Env-gated + skip-clean</b>, exactly as <see cref="PlainMqttLiveTests"/>: without
///         <c>MQTT_FIXTURE_ENDPOINT</c> (and credentials) every test calls <c>Assert.Skip</c>. See
///         <see cref="MqttFixture"/> for the env-var list.
///     </para>
///     <para>
///         <b>Every test uses its own Sparkplug group id.</b> The fixture broker is shared and the
///         driver subscribes <c>spBv1.0/{group}/#</c>; a fixed group would let one test's edge node
///         feed another's driver, and — worse — let a previous run's traffic satisfy an assertion.
///     </para>
/// </summary>
[Collection(MqttFixtureCollection.Name)]
[Trait("Category", "LiveIntegration")]
public sealed class SparkplugLiveTests(MqttFixture fixture)
{
    /// <summary>OPC UA <c>Good</c>.</summary>
    private const uint Good = 0x00000000u;

    /// <summary>
    ///     OPC UA <c>BadCommunicationError</c> — the quality the ingest state machine stamps on a tag
    ///     whose source announced it is offline (NDEATH/DDEATH).
    /// </summary>
    private const uint BadCommunicationError = 0x80050000u;

    /// <summary>
    ///     Bound on every wait for broker traffic. Generous enough to absorb a TLS handshake plus a
    ///     rebirth round-trip; short enough that a genuinely dark path fails rather than hangs.
    /// </summary>
    private static readonly TimeSpan Timeout = TimeSpan.FromSeconds(30);

    /// <summary>
    ///     How long a "this must NOT happen" assertion watches for. Long enough that the message would
    ///     have arrived (a broker round-trip on this rig is single-digit milliseconds), short enough not
    ///     to dominate the suite.
    /// </summary>
    private static readonly TimeSpan NegativeWindow = TimeSpan.FromSeconds(4);

    private readonly MqttFixture _fx = fixture;

    // -----------------------------------------------------------------------
    // birth → data → alias resolve → OnDataChange   (+ §3.6 case iv: negative Int32)
    // -----------------------------------------------------------------------

    /// <summary>
    ///     The spine of the driver: an NBIRTH/DBIRTH builds the alias table, an alias-only NDATA/DDATA
    ///     resolves through it, and the value is published under the tag's <b>RawPath</b> — the v3
    ///     driver wire reference, not the Sparkplug tuple and not the topic.
    ///     <para>
    ///         It simultaneously carries <b>§3.6 matrix case (iv), the negative Int32</b>. The metric's
    ///         value of −1234 travels as <c>int_value = 4294966062</c> because that is what Sparkplug
    ///         says; a driver that skipped <c>ReinterpretSigned</c> would publish 4294966062 at Good
    ///         quality — a plausible number, a wrong one, and invisible to any assertion that only
    ///         checks "a value arrived".
    ///     </para>
    /// </summary>
    [Fact]
    public async Task Birth_then_alias_only_data_publishes_under_rawpath_with_signed_ints_intact()
    {
        SkipIfUnconfigured();

        var ct = TestContext.Current.CancellationToken;
        var group = NewGroup("birth");
        const string node = "EdgeA";
        const string device = "Filler1";

        const string tempPath = "Fixture/Sparkplug/EdgeA/Temperature";
        const string countPath = "Fixture/Sparkplug/EdgeA/Count";
        const string levelPath = "Fixture/Sparkplug/EdgeA/Filler1/Level";

        await using var sim = new SparkplugEdgeNode(SimOptions(group, node));
        sim.SetNodeMetrics(
            new SparkplugMetricSeed("Temperature", 1, DataType.Float, 21.5f),
            new SparkplugMetricSeed("Count", 2, DataType.Int32, -42));
        sim.SetDeviceMetrics(device, new SparkplugMetricSeed("Level", 10, DataType.Double, 3.5d));

        await sim.ConnectAsync(ct);
        await sim.BirthAsync(ct);

        await using var driver = new MqttDriver(
            SparkplugOptions(
                group,
                Tag(tempPath, group, node, null, "Temperature"),
                Tag(countPath, group, node, null, "Count"),
                Tag(levelPath, group, node, device, "Level")),
            "mqtt-live-sp-birth");
        var recorder = new SparkplugRecorder(driver);

        await driver.InitializeAsync(string.Empty, ct);
        await driver.SubscribeAsync([tempPath, countPath, levelPath], TimeSpan.FromSeconds(1), ct);

        // The birth is a full snapshot, so these are the BIRTH's own values.
        var birthTemp = await recorder.WaitAsync(tempPath, v => Equals(v.Value, 21.5f), Timeout, ct);
        birthTemp.FullReference.ShouldBe(
            tempPath,
            "the driver must publish under the tag's RawPath, never the Sparkplug tuple or the topic");
        birthTemp.Snapshot.StatusCode.ShouldBe(Good);

        await recorder.WaitAsync(countPath, v => Equals(v.Value, -42), Timeout, ct);
        await recorder.WaitAsync(levelPath, v => Equals(v.Value, 3.5d), Timeout, ct);

        // …and now DATA: alias only, no name, no datatype — the shape a bind-by-alias driver survives
        // and a bind-by-name driver must too.
        await sim.PublishNodeDataAsync([("Temperature", 33.25f), ("Count", -1234)], ct);
        await sim.PublishDeviceDataAsync(device, [("Level", 9.75d)], ct);

        var temp = await recorder.WaitAsync(tempPath, v => Equals(v.Value, 33.25f), Timeout, ct);
        temp.Snapshot.StatusCode.ShouldBe(Good);

        // The predicate names the DATA value, not just "an int arrived": the birth already delivered a
        // negative Int32 (−42), so a laxer wait would be satisfied by the earlier message and the
        // assertion below would compare the wrong notification.
        var count = await recorder.WaitAsync(countPath, v => Equals(v.Value, -1234), Timeout, ct);
        count.Snapshot.StatusCode.ShouldBe(Good);
        count.Snapshot.Value.ShouldBe(
            -1234,
            "a Sparkplug Int32 rides as two's complement in the UNSIGNED int_value field (−1234 goes on "
            + "the wire as 4294966062); without ReinterpretSigned that is what would be published, at "
            + "Good quality");

        await recorder.WaitAsync(levelPath, v => Equals(v.Value, 9.75d), Timeout, ct);

        // The read path serves the same last values under the same keys.
        var read = await driver.ReadAsync([tempPath, countPath, levelPath], ct);
        read.Select(r => r.StatusCode).ShouldAllBe(s => s == Good);
        read[1].Value.ShouldBe(-1234);
    }

    // -----------------------------------------------------------------------
    // rebirth recovery (§3.6 #5)
    // -----------------------------------------------------------------------

    /// <summary>
    ///     Late join: the driver connects to a node that has already birthed and gone quiet, so it holds
    ///     no alias table at all. It must request a rebirth, and the request must be one a real edge
    ///     node accepts.
    ///     <para>
    ///         <b>This is the NCMD encode/decode symmetry check.</b> The driver's
    ///         <c>RebirthRequester.Build</c> output is decoded here by the simulator — an independent
    ///         reader that looks for a Boolean metric literally named <c>Node Control/Rebirth</c> set to
    ///         true. Offline tests assert the driver's own bytes against the driver's own expectations;
    ///         only this one proves the command means what the spec says to a party that did not write it.
    ///     </para>
    /// </summary>
    [Fact]
    public async Task Late_join_requests_a_rebirth_the_edge_node_honours_and_metadata_recovers()
    {
        SkipIfUnconfigured();

        var ct = TestContext.Current.CancellationToken;
        var group = NewGroup("rebirth");
        const string node = "EdgeA";
        const string rawPath = "Fixture/Sparkplug/Rebirth/Temperature";

        await using var sim = new SparkplugEdgeNode(SimOptions(group, node));
        sim.SetNodeMetrics(new SparkplugMetricSeed("Temperature", 7, DataType.Double, 64.5d));

        // Connected and listening for NCMD, but DELIBERATELY not birthed: the driver joins a stream it
        // has no metadata for, which is exactly the state a restart or a long outage leaves it in.
        await sim.ConnectAsync(ct);

        await using var driver = new MqttDriver(
            SparkplugOptions(group, Tag(rawPath, group, node, null, "Temperature")),
            "mqtt-live-sp-rebirth");
        var recorder = new SparkplugRecorder(driver);

        await driver.InitializeAsync(string.Empty, ct);
        await driver.SubscribeAsync([rawPath], TimeSpan.FromSeconds(1), ct);

        var honoured = await sim.WaitForRebirthAsync(1, Timeout, ct);
        honoured.ShouldBeGreaterThanOrEqualTo(
            1,
            "the driver must issue a late-join rebirth NCMD on connect (§3.6 #5), and it must be one a "
            + "spec-reading edge node recognises");

        var change = await recorder.WaitAsync(rawPath, v => Equals(v.Value, 64.5d), Timeout, ct);
        change.Snapshot.StatusCode.ShouldBe(
            Good,
            "the rebirth's NBIRTH is a full snapshot — its values are what restores the tag");
    }

    // -----------------------------------------------------------------------
    // §3.6 case (i) — alias reuse across a rebirth
    // -----------------------------------------------------------------------

    /// <summary>
    ///     <b>The single most dangerous Sparkplug defect, on the wire.</b> An edge node may reassign an
    ///     alias to a <i>different</i> metric across a rebirth — legal, and something real edge nodes do
    ///     whenever their metric set changes. A driver that bound its tags to aliases keeps routing by
    ///     the old mapping and publishes Pressure's value into the Temperature tag: Good quality,
    ///     entirely plausible values, wrong tag, and nothing anywhere reports an error.
    ///     <para>
    ///         The test swaps aliases 5 and 6 between Temperature and Pressure at the rebirth, then
    ///         publishes a DATA message carrying <b>alias 5 only</b>. Bind-by-name puts it in Pressure;
    ///         bind-by-alias puts it in Temperature. The assertion checks both tags, because "Pressure
    ///         got the value" alone would still pass if the driver had fed both.
    ///     </para>
    /// </summary>
    [Fact]
    public async Task Alias_reused_across_a_rebirth_routes_by_metric_name_not_by_alias()
    {
        SkipIfUnconfigured();

        var ct = TestContext.Current.CancellationToken;
        var group = NewGroup("alias");
        const string node = "EdgeA";
        const string tempPath = "Fixture/Sparkplug/Alias/Temperature";
        const string pressPath = "Fixture/Sparkplug/Alias/Pressure";

        await using var sim = new SparkplugEdgeNode(SimOptions(group, node));
        sim.SetNodeMetrics(
            new SparkplugMetricSeed("Temperature", 5, DataType.Double, 1d),
            new SparkplugMetricSeed("Pressure", 6, DataType.Double, 2d));

        await sim.ConnectAsync(ct);
        await sim.BirthAsync(ct);

        await using var driver = new MqttDriver(
            SparkplugOptions(
                group,
                Tag(tempPath, group, node, null, "Temperature"),
                Tag(pressPath, group, node, null, "Pressure")),
            "mqtt-live-sp-alias");
        var recorder = new SparkplugRecorder(driver);

        await driver.InitializeAsync(string.Empty, ct);
        await driver.SubscribeAsync([tempPath, pressPath], TimeSpan.FromSeconds(1), ct);

        await recorder.WaitAsync(tempPath, v => Equals(v.Value, 1d), Timeout, ct);
        await recorder.WaitAsync(pressPath, v => Equals(v.Value, 2d), Timeout, ct);

        // Sanity, pre-swap: alias 5 IS Temperature right now.
        await sim.PublishNodeDataAsync([("Temperature", 11d)], ct);
        await recorder.WaitAsync(tempPath, v => Equals(v.Value, 11d), Timeout, ct);

        // The rebirth: aliases 5 and 6 change hands, and the birth republishes both values.
        sim.SetNodeMetrics(
            new SparkplugMetricSeed("Temperature", 6, DataType.Double, 100d),
            new SparkplugMetricSeed("Pressure", 5, DataType.Double, 200d));
        await sim.BirthAsync(ct);

        await recorder.WaitAsync(tempPath, v => Equals(v.Value, 100d), Timeout, ct);
        await recorder.WaitAsync(pressPath, v => Equals(v.Value, 200d), Timeout, ct);

        // …and now the message that discriminates: alias 5, which is Pressure NOW and was Temperature
        // a moment ago.
        await sim.PublishNodeDataAsync([("Pressure", 777d)], ct);

        var pressure = await recorder.WaitAsync(pressPath, v => Equals(v.Value, 777d), Timeout, ct);
        pressure.Snapshot.StatusCode.ShouldBe(Good);

        // The other half of the assertion. Without it, a driver that fed BOTH tags would pass.
        var read = await driver.ReadAsync([tempPath], ct);
        read[0].Value.ShouldBe(
            100d,
            "alias 5 now belongs to Pressure; a driver still routing it to Temperature would show 777 "
            + "here — Good quality, plausible value, wrong tag");
    }

    // -----------------------------------------------------------------------
    // §3.6 case (ii) — a stale-bdSeq NDEATH arriving after the reconnect NBIRTH
    // -----------------------------------------------------------------------

    /// <summary>
    ///     An edge node's link drops, it reconnects and births under a new <c>bdSeq</c>, and only
    ///     <i>then</i> does the broker get round to delivering the old connection's Will. Without the
    ///     <c>bdSeq</c> tie that stale death marks a live, freshly-born node dead and drives every one
    ///     of its tags Bad, with nothing coming to correct it.
    ///     <para>
    ///         Both arms are asserted, and they must be: a driver that ignored <b>every</b> NDEATH would
    ///         pass the first half. So the matching-token death follows immediately and must land.
    ///     </para>
    /// </summary>
    [Fact]
    public async Task Ndeath_with_a_stale_bdseq_is_ignored_and_the_current_one_stales()
    {
        SkipIfUnconfigured();

        var ct = TestContext.Current.CancellationToken;
        var group = NewGroup("bdseq");
        const string node = "EdgeA";
        const string rawPath = "Fixture/Sparkplug/BdSeq/Temperature";

        await using var sim = new SparkplugEdgeNode(SimOptions(group, node));
        sim.SetNodeMetrics(new SparkplugMetricSeed("Temperature", 3, DataType.Double, 10d));

        await sim.ConnectAsync(ct);                 // session 0
        sim.BdSeq.ShouldBe(0UL);
        await sim.BirthAsync(ct);

        await using var driver = new MqttDriver(
            SparkplugOptions(group, Tag(rawPath, group, node, null, "Temperature")),
            "mqtt-live-sp-bdseq");
        var recorder = new SparkplugRecorder(driver);

        await driver.InitializeAsync(string.Empty, ct);
        await driver.SubscribeAsync([rawPath], TimeSpan.FromSeconds(1), ct);
        await recorder.WaitAsync(rawPath, v => Equals(v.Value, 10d), Timeout, ct);

        // The node "reconnects": a clean DISCONNECT (so no Will fires) followed by a new session, whose
        // bdSeq is 1, and a fresh birth carrying a distinguishable value.
        sim.SetNodeMetrics(new SparkplugMetricSeed("Temperature", 3, DataType.Double, 20d));
        await sim.ConnectAsync(ct);                 // session 1
        sim.BdSeq.ShouldBe(1UL);
        await sim.BirthAsync(ct);
        await recorder.WaitAsync(rawPath, v => Equals(v.Value, 20d), Timeout, ct);

        // The old session's Will, delivered late. bdSeq 0 ≠ the live session's 1 ⇒ it must be discarded.
        await sim.PublishNodeDeathAsync(bdSeq: 0UL, ct);

        var deadline = DateTime.UtcNow + NegativeWindow;
        while (DateTime.UtcNow < deadline)
        {
            var sample = await driver.ReadAsync([rawPath], ct);
            sample[0].StatusCode.ShouldBe(
                Good,
                "an NDEATH carrying a PREVIOUS session's bdSeq is a stale Will; acting on it kills a live node");
            await Task.Delay(200, ct);
        }

        // …and the falsifiability control: the SAME message with the CURRENT token must stale the tag.
        // Without this leg, a driver that ignored every death would pass the assertion above.
        await sim.PublishNodeDeathAsync(bdSeq: 1UL, ct);

        var stale = await recorder.WaitAsync(
            rawPath,
            v => v.StatusCode == BadCommunicationError,
            Timeout,
            ct);
        stale.Snapshot.Value.ShouldBeNull();
    }

    // -----------------------------------------------------------------------
    // death → STALE, through the real Last-Will path
    // -----------------------------------------------------------------------

    /// <summary>
    ///     The realistic death: the edge node's session ends and the <b>broker</b> publishes the NDEATH
    ///     Will it registered at CONNECT. Nothing in this test publishes the death message — which is
    ///     the point, because a Will is the only death an edge node that crashed can produce, and it is
    ///     also the one that carries no <c>seq</c>.
    ///     <para>
    ///         The recovery half matters as much: §3.6 #4 says the next birth restores Good, so a driver
    ///         that latched Bad on death would be wrong in the direction nobody notices until a plant
    ///         comes back and its tags do not.
    ///     </para>
    /// </summary>
    [Fact]
    public async Task Broker_published_ndeath_will_stales_the_tags_and_the_next_birth_restores_them()
    {
        SkipIfUnconfigured();

        var ct = TestContext.Current.CancellationToken;
        var group = NewGroup("death");
        const string node = "EdgeA";
        const string rawPath = "Fixture/Sparkplug/Death/Temperature";

        await using var sim = new SparkplugEdgeNode(SimOptions(group, node));
        sim.SetNodeMetrics(new SparkplugMetricSeed("Temperature", 4, DataType.Double, 42d));

        await sim.ConnectAsync(ct);
        await sim.BirthAsync(ct);

        await using var driver = new MqttDriver(
            SparkplugOptions(group, Tag(rawPath, group, node, null, "Temperature")),
            "mqtt-live-sp-death");
        var recorder = new SparkplugRecorder(driver);

        await driver.InitializeAsync(string.Empty, ct);
        await driver.SubscribeAsync([rawPath], TimeSpan.FromSeconds(1), ct);
        await recorder.WaitAsync(rawPath, v => Equals(v.Value, 42d), Timeout, ct);

        // The broker fires the Will — we publish nothing.
        await sim.KillAsync(ct);

        var stale = await recorder.WaitAsync(rawPath, v => v.StatusCode == BadCommunicationError, Timeout, ct);
        stale.FullReference.ShouldBe(rawPath);
        stale.Snapshot.Value.ShouldBeNull("a staled tag must not keep serving the dead node's last value");

        // §3.6 #4: the next birth restores Good.
        sim.SetNodeMetrics(new SparkplugMetricSeed("Temperature", 4, DataType.Double, 43d));
        await sim.ConnectAsync(ct);
        await sim.BirthAsync(ct);

        var revived = await recorder.WaitAsync(rawPath, v => Equals(v.Value, 43d), Timeout, ct);
        revived.Snapshot.StatusCode.ShouldBe(Good);
    }

    // -----------------------------------------------------------------------
    // §3.6 case (iii) — seq wrap 255 → 0, and its falsifiability control
    // -----------------------------------------------------------------------

    /// <summary>
    ///     <b>255 → 0 is the sequence continuing, not a gap.</b> Read the wrap as a gap and every edge
    ///     node is commanded to rebirth once per 256 messages — a busy node would spend its life
    ///     republishing metadata. This drives a full lap of the counter over a real broker and asserts
    ///     that <b>no NCMD appears on the wire</b>.
    ///     <para>
    ///         <b>Two things make this test able to fail.</b> First, the ingestor is built with
    ///         <c>rebirthDebounce: TimeSpan.Zero</c>: at the shipped 10 s default a wrap-triggered
    ///         request would be swallowed by the debounce and the test would pass for the wrong reason.
    ///         Second, the negative control — a deliberately skipped <c>seq</c> immediately afterwards
    ///         <b>must</b> produce an NCMD. Silence proves nothing unless noise is also proven.
    ///     </para>
    ///     <para>
    ///         The assertion is made on the wire, by a third MQTT client watching
    ///         <c>spBv1.0/{group}/NCMD/#</c> — not on a counter inside the driver.
    ///     </para>
    /// </summary>
    [Fact]
    public async Task Seq_wrap_255_to_0_requests_no_rebirth_but_a_real_gap_does()
    {
        SkipIfUnconfigured();

        var ct = TestContext.Current.CancellationToken;
        var group = NewGroup("seq");
        const string node = "EdgeA";
        const string rawPath = "Fixture/Sparkplug/Seq/Counter";

        await using var watcher = await NcmdWatcher.StartAsync(_fx, group, ct);

        // QoS 1 for the 256-message lap: the seq semantics under test are identical at either QoS, and
        // a broker-side drop at QoS 0 would surface as a "gap" — a real detection, but a flaky test.
        await using var sim = new SparkplugEdgeNode(SimOptions(group, node) with { DataQos = 1 });
        sim.SetNodeMetrics(new SparkplugMetricSeed("Counter", 1, DataType.Double, 0d));

        await sim.ConnectAsync(ct);
        await sim.BirthAsync(ct);

        var options = SparkplugOptions(group, Tag(rawPath, group, node, null, "Counter"));

        // The production ingest path — AttachTo + EstablishAsync is exactly what MqttDriver does — but
        // constructed directly so the rebirth debounce can be taken out of the picture.
        await using var connection = new MqttConnection(options, "mqtt-live-sp-seq");
        var ingestor = new SparkplugIngestor(
            options,
            "mqtt-live-sp-seq",
            logger: null,
            maxPayloadBytes: options.MaxPayloadBytes,
            rebirthDebounce: TimeSpan.Zero);
        ingestor.Register(options.RawTags);
        ingestor.AttachTo(connection);

        var recorder = new SparkplugRecorder(ingestor);
        await connection.ConnectAsync(ct);
        await ingestor.EstablishAsync(ct);
        await ingestor.SubscribeAsync([rawPath], TimeSpan.FromSeconds(1), ct);

        // Establish issues the late-join NCMD; let the node answer it and settle before measuring.
        await sim.WaitForRebirthAsync(1, Timeout, ct);
        await recorder.WaitAsync(rawPath, v => v.StatusCode == Good, Timeout, ct);
        await Task.Delay(500, ct);

        var beforeWrap = watcher.Count;

        // A full lap: the birth left seq at 0, so these carry seq 1…255 and then 0 — the wrap.
        for (var i = 1; i <= 256; i++)
        {
            await sim.PublishNodeDataAsync([("Counter", (double)i)], ct);
        }

        sim.LastSeq.ShouldBe((byte)0, "256 messages after a birth at seq 0 must land exactly on the wrap");

        await recorder.WaitAsync(rawPath, v => Equals(v.Value, 256d), Timeout, ct);
        await Task.Delay(500, ct);      // any NCMD the wrap provoked would have arrived by now

        watcher.Count.ShouldBe(
            beforeWrap,
            $"a seq wrap 255 → 0 is the stream continuing, not a gap; NCMDs seen: "
            + $"{string.Join(", ", watcher.Topics)}");

        // ---- the falsifiability control -------------------------------------------------
        // Burn one sequence number without publishing it. That IS a lost message from the host's point
        // of view, and it MUST produce a rebirth request — otherwise the silence above proves only that
        // this driver never asks for anything.
        sim.SkipSeq();
        await sim.PublishNodeDataAsync([("Counter", 999d)], ct);

        await watcher.WaitForCountAsync(beforeWrap + 1, Timeout, ct);

        watcher.Topics.ShouldAllBe(t => t == $"spBv1.0/{group}/NCMD/{node}");

        // A gap resynchronizes rather than latching: the message that exposed it still lands.
        await recorder.WaitAsync(rawPath, v => Equals(v.Value, 999d), Timeout, ct);
    }

    // -----------------------------------------------------------------------
    // browser: passive window
    // -----------------------------------------------------------------------

    /// <summary>
    ///     <b>Opening an address picker against a live plant must publish nothing.</b> The browse
    ///     session counts its own publishes, and the offline suite asserts that counter — but a counter
    ///     can only see messages that went through the seam it guards. This asserts the property where
    ///     it actually matters: on the broker, with an independent client subscribed to the group's NCMD
    ///     topic while the picker opens, roots, expands every level and reads every attribute.
    /// </summary>
    [Fact]
    public async Task Browser_open_and_full_tree_walk_publish_no_ncmd()
    {
        SkipIfUnconfigured();

        var ct = TestContext.Current.CancellationToken;
        var group = NewGroup("browse");

        await using var watcher = await NcmdWatcher.StartAsync(_fx, group, ct);
        await using var edgeA = await StartBrowseNodeAsync(group, "EdgeA", withDevice: true, ct);
        await using var edgeB = await StartBrowseNodeAsync(group, "EdgeB", withDevice: false, ct);

        var browser = new MqttDriverBrowser();
        await using var session = await browser.OpenAsync(BrowseConfigJson(group), ct);

        // The window is passive, so it can only see births published AFTER it subscribed — Sparkplug
        // births are never retained. This is the same round-trip the picker's "Request rebirth" button
        // exists for; here the test plays the edge node's part directly, so the browser stays untouched.
        await edgeA.BirthAsync(ct);
        await edgeB.BirthAsync(ct);

        var roots = await WaitForRootAsync(session, group, ct);
        roots.ShouldContain(n => n.NodeId == group);

        // Walk everything the picker could walk: every level expanded, every node's attributes read.
        var visited = await WalkAsync(session, roots, ct);

        visited.ShouldContain($"{group}/EdgeA");
        visited.ShouldContain($"{group}/EdgeB");
        visited.ShouldContain($"{group}/EdgeA/Filler1");
        visited.ShouldContain($"{group}/EdgeA::Temperature");
        visited.ShouldContain($"{group}/EdgeA/Filler1::Level");

        // Give anything the walk might have kicked off time to reach the broker before concluding.
        await Task.Delay(NegativeWindow, ct);

        watcher.Count.ShouldBe(
            0,
            "no browse path may publish — an operator opening a picker must not be able to command a "
            + $"running plant. Published: {string.Join(", ", watcher.Topics)}");
        edgeA.RebirthCount.ShouldBe(0);
        edgeB.RebirthCount.ShouldBe(0);
    }

    // -----------------------------------------------------------------------
    // browser: explicit rebirth, node vs. group scope
    // -----------------------------------------------------------------------

    /// <summary>
    ///     The one browse-session member that publishes, and the enumeration it performs: an edge-node
    ///     scope addresses exactly that node, a group scope fans out to every edge node observed beneath
    ///     it. Asserted on the wire (which topics were published) and at the far end (which simulated
    ///     nodes actually honoured a rebirth), because "the call returned 2" says nothing about where
    ///     the commands went.
    /// </summary>
    [Fact]
    public async Task Browser_request_rebirth_addresses_one_node_or_every_node_in_the_group()
    {
        SkipIfUnconfigured();

        var ct = TestContext.Current.CancellationToken;
        var group = NewGroup("rbscope");

        await using var watcher = await NcmdWatcher.StartAsync(_fx, group, ct);
        await using var edgeA = await StartBrowseNodeAsync(group, "EdgeA", withDevice: true, ct);
        await using var edgeB = await StartBrowseNodeAsync(group, "EdgeB", withDevice: false, ct);

        var browser = new MqttDriverBrowser();
        await using var session = await browser.OpenAsync(BrowseConfigJson(group), ct);
        var rebirthable = session.ShouldBeAssignableTo<IRebirthCapableBrowseSession>();

        await edgeA.BirthAsync(ct);
        await edgeB.BirthAsync(ct);
        await WaitForRootAsync(session, group, ct);

        // Both edge nodes must be in the tree before the group scope is exercised — it enumerates what
        // the window OBSERVED, so asserting the fan-out against a half-built tree would prove nothing.
        await WaitForChildrenAsync(session, group, expected: 2, ct);

        // ---- node scope ----
        var nodeTargets = await rebirthable.RequestRebirthAsync($"{group}/EdgeA", ct);
        nodeTargets.ShouldBe(1);

        await watcher.WaitForCountAsync(1, Timeout, ct);
        watcher.Topics.ShouldBe([$"spBv1.0/{group}/NCMD/EdgeA"]);
        await edgeA.WaitForRebirthAsync(1, Timeout, ct);
        edgeB.RebirthCount.ShouldBe(0, "a node-scoped rebirth must not touch the node next to it");

        // ---- group scope ----
        var groupTargets = await rebirthable.RequestRebirthAsync(group, ct);
        groupTargets.ShouldBe(2);

        await watcher.WaitForCountAsync(3, Timeout, ct);
        watcher.Topics.ShouldBe(
            [
                $"spBv1.0/{group}/NCMD/EdgeA",
                $"spBv1.0/{group}/NCMD/EdgeA",
                $"spBv1.0/{group}/NCMD/EdgeB",
            ],
            "a group scope addresses every observed edge node, once each, in a deterministic order");

        await edgeA.WaitForRebirthAsync(2, Timeout, ct);
        await edgeB.WaitForRebirthAsync(1, Timeout, ct);
    }

    // -----------------------------------------------------------------------
    // helpers
    // -----------------------------------------------------------------------

    private void SkipIfUnconfigured()
    {
        if (_fx.NotConfigured)
        {
            Assert.Skip(_fx.SkipReason!);
        }
    }

    /// <summary>
    ///     A Sparkplug group id unique to this run. The fixture broker is shared and the driver
    ///     subscribes the whole group, so a fixed id would let tests — and previous runs — feed each
    ///     other.
    /// </summary>
    private static string NewGroup(string label) => $"OtOpcUaLive-{label}-{Guid.NewGuid():N}"[..40];

    /// <summary>Driver options for Sparkplug mode over the fixture's TLS listener.</summary>
    private MqttDriverOptions SparkplugOptions(string groupId, params RawTagEntry[] rawTags) => new()
    {
        Host = _fx.TlsHost,
        Port = _fx.TlsPort,
        UseTls = true,
        CaCertificatePath = _fx.CaCertificatePath,
        AllowUntrustedServerCertificate = _fx.CaCertificatePath is null,
        Username = _fx.Username,
        Password = _fx.Password,
        ClientId = $"otopcua-live-sp-{Guid.NewGuid():N}",
        ConnectTimeoutSeconds = 15,
        Mode = MqttMode.SparkplugB,
        Sparkplug = new MqttSparkplugOptions { GroupId = groupId, RequestRebirthOnGap = true },
        RawTags = rawTags,
    };

    /// <summary>Simulator connection settings, pointed at the same listener with the same credentials.</summary>
    private SparkplugEdgeNodeOptions SimOptions(string groupId, string edgeNodeId) => new()
    {
        Host = _fx.TlsHost,
        Port = _fx.TlsPort,
        UseTls = true,
        CaCertificatePath = _fx.CaCertificatePath,
        AllowUntrustedServerCertificate = _fx.CaCertificatePath is null,
        Username = _fx.Username,
        Password = _fx.Password,
        GroupId = groupId,
        EdgeNodeId = edgeNodeId,
    };

    /// <summary>An authored Sparkplug raw tag — the binding tuple, with the datatype left to the birth.</summary>
    private static RawTagEntry Tag(
        string rawPath,
        string groupId,
        string edgeNodeId,
        string? deviceId,
        string metricName)
    {
        var device = deviceId is null ? "" : $"""
                                              "deviceId":"{deviceId}",
                                              """;
        return new RawTagEntry(
            rawPath,
            $$"""
              {"groupId":"{{groupId}}","edgeNodeId":"{{edgeNodeId}}",{{device}}"metricName":"{{metricName}}"}
              """,
            WriteIdempotent: false);
    }

    /// <summary>The browse configuration blob, parsed through the driver's ONE shared JSON options.</summary>
    private string BrowseConfigJson(string groupId) =>
        JsonSerializer.Serialize(SparkplugOptions(groupId), MqttJson.Options);

    /// <summary>Brings up one simulated edge node for the browse legs, connected but not yet birthed.</summary>
    private async Task<SparkplugEdgeNode> StartBrowseNodeAsync(
        string group,
        string node,
        bool withDevice,
        CancellationToken ct)
    {
        var sim = new SparkplugEdgeNode(SimOptions(group, node));
        sim.SetNodeMetrics(
            new SparkplugMetricSeed("Temperature", 1, DataType.Float, 21.5f),
            new SparkplugMetricSeed("Count", 2, DataType.Int32, -42));

        if (withDevice)
        {
            sim.SetDeviceMetrics("Filler1", new SparkplugMetricSeed("Level", 10, DataType.Double, 3.5d));
        }

        await sim.ConnectAsync(ct);
        return sim;
    }

    /// <summary>Polls <c>RootAsync</c> until the observation window has recorded the group.</summary>
    private static async Task<IReadOnlyList<BrowseNode>> WaitForRootAsync(
        IBrowseSession session,
        string group,
        CancellationToken ct)
    {
        var deadline = DateTime.UtcNow + Timeout;
        while (DateTime.UtcNow < deadline)
        {
            var roots = await session.RootAsync(ct);
            if (roots.Any(n => n.NodeId == group))
            {
                return roots;
            }

            await Task.Delay(200, ct);
        }

        throw new TimeoutException(
            $"The browse window never observed Sparkplug group '{group}'. A birth must be published "
            + "AFTER the window opens — Sparkplug births are not retained.");
    }

    /// <summary>Polls until a node reports at least <paramref name="expected"/> children.</summary>
    private static async Task WaitForChildrenAsync(
        IBrowseSession session,
        string nodeId,
        int expected,
        CancellationToken ct)
    {
        var deadline = DateTime.UtcNow + Timeout;
        while (DateTime.UtcNow < deadline)
        {
            var children = await session.ExpandAsync(nodeId, ct);
            if (children.Count >= expected)
            {
                return;
            }

            await Task.Delay(200, ct);
        }

        throw new TimeoutException($"'{nodeId}' never reported {expected} children in the browse window.");
    }

    /// <summary>Expands every node depth-first and reads every node's attributes; returns the node ids seen.</summary>
    private static async Task<List<string>> WalkAsync(
        IBrowseSession session,
        IReadOnlyList<BrowseNode> level,
        CancellationToken ct)
    {
        var seen = new List<string>();
        foreach (var node in level)
        {
            seen.Add(node.NodeId);
            await session.AttributesAsync(node.NodeId, ct);

            var children = await session.ExpandAsync(node.NodeId, ct);
            if (children.Count > 0)
            {
                seen.AddRange(await WalkAsync(session, children, ct));
            }
        }

        return seen;
    }

    // -----------------------------------------------------------------------
    // fixtures used by the tests
    // -----------------------------------------------------------------------

    /// <summary>
    ///     Records every <c>OnDataChange</c> raised by a driver or an ingestor and lets a test await a
    ///     specific RawPath reaching a specific state.
    ///     <para>
    ///         <b>Awaits a predicate, not merely an arrival.</b> A Sparkplug tag is fed by births,
    ///         rebirths and data alike, so "a notification arrived for this RawPath" is satisfied by the
    ///         wrong message roughly half the time; every wait here names the value or quality it is
    ///         waiting for. It also pre-scans, so a test that registers after the message landed is
    ///         served rather than timing out.
    ///     </para>
    /// </summary>
    private sealed class SparkplugRecorder
    {
        private readonly List<DataChangeEventArgs> _events = [];
        private readonly Lock _gate = new();

        public SparkplugRecorder(MqttDriver driver) => driver.OnDataChange += Record;

        public SparkplugRecorder(SparkplugIngestor ingestor) => ingestor.OnDataChange += Record;

        public async Task<DataChangeEventArgs> WaitAsync(
            string rawPath,
            Func<DataValueSnapshot, bool> predicate,
            TimeSpan timeout,
            CancellationToken cancellationToken)
        {
            var deadline = DateTime.UtcNow + timeout;
            while (true)
            {
                lock (_gate)
                {
                    foreach (var candidate in _events)
                    {
                        if (candidate.FullReference == rawPath && predicate(candidate.Snapshot))
                        {
                            return candidate;
                        }
                    }
                }

                if (DateTime.UtcNow >= deadline)
                {
                    throw new TimeoutException(
                        $"No notification for RawPath '{rawPath}' satisfied the expected state within "
                        + $"{timeout.TotalSeconds:0}s. Recorded: {Describe()}");
                }

                await Task.Delay(100, cancellationToken).ConfigureAwait(false);
            }
        }

        private string Describe()
        {
            lock (_gate)
            {
                return _events.Count == 0
                    ? "(nothing)"
                    : string.Join(
                        " | ",
                        _events
                            .GroupBy(e => e.FullReference)
                            .Select(g =>
                                $"{g.Key}: [{string.Join(", ", g.Select(e => $"{e.Snapshot.Value ?? "null"}@0x{e.Snapshot.StatusCode:X8}"))}]"));
            }
        }

        private void Record(object? sender, DataChangeEventArgs args)
        {
            lock (_gate)
            {
                _events.Add(args);
            }
        }
    }

    /// <summary>
    ///     An independent MQTT client subscribed to <c>spBv1.0/{group}/NCMD/#</c>, recording every
    ///     command published into the group.
    ///     <para>
    ///         <b>This is what makes the passivity assertions mean something.</b> The browse session's
    ///         own <c>PublishCountForTest</c> can only observe messages that pass through the seam it
    ///         guards; this observes the broker. If a future change published an NCMD by some other
    ///         route — a raw client, a Last Will, a stray retained command — the counter would still
    ///         read zero and this would not.
    ///     </para>
    /// </summary>
    private sealed class NcmdWatcher : IAsyncDisposable
    {
        private readonly IMqttClient _client;
        private readonly ConcurrentQueue<string> _topics = new();

        private NcmdWatcher(IMqttClient client) => _client = client;

        /// <summary>Every NCMD topic observed, in arrival order.</summary>
        public IReadOnlyList<string> Topics => [.. _topics];

        /// <summary>How many NCMDs have been observed.</summary>
        public int Count => _topics.Count;

        public static async Task<NcmdWatcher> StartAsync(MqttFixture fixture, string group, CancellationToken ct)
        {
            var client = new MqttClientFactory().CreateMqttClient();
            var watcher = new NcmdWatcher(client);

            client.ApplicationMessageReceivedAsync += args =>
            {
                // Only a genuine rebirth command counts. A malformed or unrelated NCMD would be a
                // different defect, and lumping them together would make the count uninterpretable.
                if (watcher.IsRebirth(args.ApplicationMessage))
                {
                    watcher._topics.Enqueue(args.ApplicationMessage.Topic);
                }

                return Task.CompletedTask;
            };

            var options = new MqttClientOptionsBuilder()
                .WithTcpServer(fixture.PlainHost, fixture.PlainPort)
                .WithCredentials(fixture.Username, fixture.Password)
                .WithClientId($"otopcua-live-ncmdwatch-{Guid.NewGuid():N}")
                .WithTimeout(TimeSpan.FromSeconds(10))
                .Build();

            await client.ConnectAsync(options, ct).ConfigureAwait(false);
            await client.SubscribeAsync(
                new MqttClientSubscribeOptionsBuilder()
                    .WithTopicFilter(f => f.WithTopic($"spBv1.0/{group}/NCMD/#"))
                    .Build(),
                ct).ConfigureAwait(false);

            return watcher;
        }

        public async Task WaitForCountAsync(int count, TimeSpan timeout, CancellationToken ct)
        {
            var deadline = DateTime.UtcNow + timeout;
            while (Count < count)
            {
                if (DateTime.UtcNow >= deadline)
                {
                    throw new TimeoutException(
                        $"Only {Count} rebirth NCMD(s) reached the broker within {timeout.TotalSeconds:0}s; "
                        + $"expected {count}. Seen: {string.Join(", ", Topics)}");
                }

                await Task.Delay(100, ct).ConfigureAwait(false);
            }
        }

        public async ValueTask DisposeAsync()
        {
            try
            {
                using var cts = new CancellationTokenSource(TimeSpan.FromSeconds(5));
                await _client.DisconnectAsync(new MqttClientDisconnectOptions(), cts.Token).ConfigureAwait(false);
            }
            catch (Exception)
            {
                // Teardown is best-effort; the broker may already have gone.
            }

            _client.Dispose();
        }

        private bool IsRebirth(MqttApplicationMessage message)
        {
            var body = message.Payload;
            var bytes = body.IsSingleSegment ? body.FirstSpan.ToArray() : body.ToArray();

            try
            {
                var payload = Payload.Parser.ParseFrom(bytes);
                return payload.Metrics.Any(m =>
                    m.Name == SparkplugEdgeNode.RebirthMetricName
                    && m.ValueCase == Payload.Types.Metric.ValueOneofCase.BooleanValue
                    && m.BooleanValue);
            }
            catch (Google.Protobuf.InvalidProtocolBufferException)
            {
                return false;
            }
        }
    }
}