feat(mqtt): flatten Sparkplug Template/DataSet metrics to sibling tags (#515)

`SparkplugCodec` handled scalars; `DataSet`, `Template` and `PropertySet` were skipped with a warning. They now flatten to addressable tags (owner decision, recorded on the issue): Template → sibling metrics, `/`-joined, recursively: Motor → Motor/Speed, Motor/Stator/Temp DataSet → one metric per cell, {column}/{rowIndex}: Recipe → Recipe/Step/0, Recipe/Target/1 PropertySet → node metadata, never tags — it describes a metric rather than being one Flattening was chosen over OPC UA Structures because flattened scalars are consumable by any client with no type dictionary, and the result mirrors how the `/raw` tree works. **The row index is always present, including on row 0.** Emitting a bare column name for a single-row table and an indexed one otherwise would make a tag's name depend on how many rows the plant published — a table growing from one row to two would rename every tag already authored against it. **A DATA metric carries the parent's alias and no name**, so member names cannot be composed until the alias table has resolved the parent. Expansion therefore happens in `SparkplugIngestor.PublishStructuredMembers` — after resolution, at the single funnel both the DATA and birth-value paths already pass through — not at decode time. The alias table installs a binding per member so browse enumerates them and a DDATA member that omits its datatype falls back to the birth's (guessing from the CLR type would read a signed Int32 of -42 out of its unsigned field as 4294967254). **Skip-and-warn survives, and refusal is whole** — the explicit requirement on the issue. A template definition, an unnamed member, an extension value, a column/type mismatch, a short row, or a payload past MaxDepth(5)/MaxMembers(512) refuses the ENTIRE metric. A silently half-flattened UDT looks exactly like a complete one in the picker. The browse picker offers the members and deliberately NOT the structured parent, which has no value and no servable datatype. The rediscovery signature likewise tracks members rather than the parent, or a UDT gaining a member would leave the signature unmoved. Member paths contain `/`, which #509 already solved: the slashed name travels intact as `metricName` while the derived tag Name is coerced to a legal RawPath segment. Live-gated 3/3 against the Mosquitto fixture with the project-owned simulator extended to publish a nested Template and a 2x2 DataSet. Falsified at three points: dropping the member bindings, not expanding the parent, and refusing to flatten templates each turn exactly the expected tests red. Also fixes a regression from the #511 commit in the same branch: `AddOtOpcUaDriverFactories` resolved `IDriverPrimaryRoleSource` with `GetRequiredService` but only `AddOtOpcUaRuntime` registered it, so a graph wiring driver factories without the full runtime failed at resolution. Two Host integration suites do exactly that and caught it; both extensions now TryAdd the same concrete `RedundancyRoleView`, so the registration is order-independent and the fail-loud resolve is kept. Claude-Session: https://claude.ai/code/session_015p7wGqy3YpZNCpDzTpGMKo
2026-07-31 08:14:04 -04:00
parent 7f2108dcf2
commit c74ccc55b5
15 changed files with 1900 additions and 53 deletions
@@ -81,7 +81,14 @@ foreach (var node in nodes)
        "Filler1",
        new SparkplugMetricSeed("Temperature", 10, DataType.Float, 55.5f),
        new SparkplugMetricSeed("FillCount", 11, DataType.Int64, 1000L),
-        new SparkplugMetricSeed("Jammed", 12, DataType.Boolean, false));
+        new SparkplugMetricSeed("Jammed", 12, DataType.Boolean, false),
+
+        // Structured metrics for the Gitea #515 gate. They live on the DEVICE rather than the node
+        // deliberately: a DDATA template is the case a naive implementation gets wrong, because its
+        // metric carries the parent's ALIAS and no name, so the member names cannot be composed until
+        // the alias table has resolved the parent.
+        new SparkplugMetricSeed("Motor", 13, DataType.Template, BuildMotorTemplate()),
+        new SparkplugMetricSeed("Recipe", 14, DataType.DataSet, BuildRecipeDataSet()));

    edges.Add(edge);
 }
@@ -112,3 +119,55 @@ return 0;

 static string Env(string name, string fallback) =>
    Environment.GetEnvironmentVariable(name) is { Length: > 0 } value ? value : fallback;
+
+// A UDT instance: `Motor` with a `Speed` double, a `Running` bool and a nested `Stator` template.
+// Flattens to `Motor/Speed`, `Motor/Running` and `Motor/Stator/Temp` (Gitea #515).
+static Payload.Types.Template BuildMotorTemplate()
+{
+    var stator = new Payload.Types.Template { Version = "1", IsDefinition = false };
+    stator.Metrics.Add(new Payload.Types.Metric
+    {
+        Name = "Temp", Datatype = (uint)DataType.Double, DoubleValue = 71.5,
+    });
+
+    var motor = new Payload.Types.Template { Version = "1", IsDefinition = false };
+    motor.Metrics.Add(new Payload.Types.Metric
+    {
+        Name = "Speed", Datatype = (uint)DataType.Double, DoubleValue = 1500,
+    });
+    motor.Metrics.Add(new Payload.Types.Metric
+    {
+        Name = "Running", Datatype = (uint)DataType.Boolean, BooleanValue = true,
+    });
+
+    // Nested one level: proves the recursion, and proves the `/`-joined path composes correctly rather
+    // than only working for flat templates.
+    motor.Metrics.Add(new Payload.Types.Metric
+    {
+        Name = "Stator", Datatype = (uint)DataType.Template, TemplateValue = stator,
+    });
+
+    return motor;
+}
+
+// A two-column, two-row table. Flattens to `Recipe/Step/0`, `Recipe/Target/0`, `Recipe/Step/1`,
+// `Recipe/Target/1` — the row index is always present, so a table that grows a row never renames the
+// tags already authored against it.
+static Payload.Types.DataSet BuildRecipeDataSet()
+{
+    var dataSet = new Payload.Types.DataSet { NumOfColumns = 2UL };
+    dataSet.Columns.Add("Step");
+    dataSet.Columns.Add("Target");
+    dataSet.Types_.Add((uint)DataType.Int32);
+    dataSet.Types_.Add((uint)DataType.Double);
+
+    for (var row = 0; row < 2; row++)
+    {
+        var r = new Payload.Types.DataSet.Types.Row();
+        r.Elements.Add(new Payload.Types.DataSet.Types.DataSetValue { IntValue = (uint)row });
+        r.Elements.Add(new Payload.Types.DataSet.Types.DataSetValue { DoubleValue = 20d + row });
+        dataSet.Rows.Add(r);
+    }
+
+    return dataSet;
+}
@@ -756,6 +756,18 @@ public sealed class SparkplugEdgeNode : IAsyncDisposable
            case DataType.File:
                metric.BytesValue = ByteString.CopyFrom((byte[])value);
                break;
+
+            // Structured values (Gitea #515). The seed carries the already-built protobuf message
+            // rather than a scalar, because a Template's shape IS its value — there is nothing simpler
+            // to convert from, and inventing a mini-DSL here would put a second, divergent notion of
+            // "what a template is" between the simulator and the wire it is supposed to reproduce.
+            case DataType.Template:
+                metric.TemplateValue = (Payload.Types.Template)value;
+                break;
+            case DataType.DataSet:
+                metric.DatasetValue = (Payload.Types.DataSet)value;
+                break;
+
            default:
                throw new NotSupportedException(
                    $"The simulator does not encode Sparkplug datatype {dataType}; v1 of the driver does "
@@ -824,9 +836,83 @@ public sealed class SparkplugEdgeNode : IAsyncDisposable
        DataType.Uint8 or DataType.Uint16 or DataType.Uint32 or DataType.Uint64 => (uint)tick,
        DataType.Boolean => tick % 2 == 0,
        DataType.String or DataType.Text or DataType.Uuid => $"tick-{tick}",
+
+        // Structured values move too, or the live gate could not tell a flattened member that is being
+        // fed from one that was merely published once at birth and then went quiet.
+        DataType.Template => AdvanceTemplate((Payload.Types.Template)seed.Value!, tick),
+        DataType.DataSet => AdvanceDataSet((Payload.Types.DataSet)seed.Value!, tick),
+
        _ => seed.Value,
    };

+    /// <summary>
+    ///     Rebuilds a template with its numeric members nudged, preserving names, datatypes and order.
+    /// </summary>
+    private static Payload.Types.Template AdvanceTemplate(Payload.Types.Template seed, int tick)
+    {
+        var next = new Payload.Types.Template { Version = seed.Version, IsDefinition = false };
+        foreach (var member in seed.Metrics)
+        {
+            var clone = member.Clone();
+            switch ((DataType)clone.Datatype)
+            {
+                case DataType.Double:
+                    clone.DoubleValue = 1500d + (tick % 100);
+                    break;
+                case DataType.Float:
+                    clone.FloatValue = 40f + (tick % 20) / 10f;
+                    break;
+                case DataType.Int32:
+                    clone.IntValue = unchecked((uint)(-tick));
+                    break;
+                case DataType.Boolean:
+                    clone.BooleanValue = tick % 2 == 0;
+                    break;
+                default:
+                    break;   // leave anything else at its seeded value
+            }
+
+            clone.Timestamp = NowMs();
+            next.Metrics.Add(clone);
+        }
+
+        return next;
+    }
+
+    /// <summary>Rebuilds a dataset with its numeric cells nudged, preserving columns, types and shape.</summary>
+    private static Payload.Types.DataSet AdvanceDataSet(Payload.Types.DataSet seed, int tick)
+    {
+        var next = new Payload.Types.DataSet { NumOfColumns = seed.NumOfColumns };
+        next.Columns.AddRange(seed.Columns);
+        next.Types_.AddRange(seed.Types_);
+
+        for (var row = 0; row < seed.Rows.Count; row++)
+        {
+            var cloned = new Payload.Types.DataSet.Types.Row();
+            for (var col = 0; col < seed.Rows[row].Elements.Count; col++)
+            {
+                var cell = seed.Rows[row].Elements[col].Clone();
+                switch ((DataType)seed.Types_[col])
+                {
+                    case DataType.Double:
+                        cell.DoubleValue = 20d + row + (tick % 10) / 10d;
+                        break;
+                    case DataType.Int32:
+                        cell.IntValue = (uint)(tick + row);
+                        break;
+                    default:
+                        break;
+                }
+
+                cloned.Elements.Add(cell);
+            }
+
+            next.Rows.Add(cloned);
+        }
+
+        return next;
+    }
+
    private void ConfigureTls(MqttClientTlsOptionsBuilder tls)
    {
        if (!_options.UseTls)
@@ -0,0 +1,332 @@
+using Org.Eclipse.Tahu.Protobuf;
+using Shouldly;
+using Xunit;
+using ZB.MOM.WW.OtOpcUa.Core.Abstractions;
+using ZB.MOM.WW.OtOpcUa.Driver.Mqtt.SparkplugSimulator;
+
+namespace ZB.MOM.WW.OtOpcUa.Driver.Mqtt.IntegrationTests;
+
+/// <summary>
+///     Live gate for structured Sparkplug metrics (Gitea #515): a <c>Template</c> and a
+///     <c>DataSet</c> published by a real edge node over the Mosquitto fixture, flattened into
+///     addressable OPC UA tags.
+/// </summary>
+/// <remarks>
+///     <para>
+///         <b>What this adds over the offline suite.</b> The offline tests build decoded payloads by
+///         hand, so they prove the mapping but not the <i>wire</i>: that a nested template really does
+///         round-trip through protobuf with its member names intact, and — the case a naive
+///         implementation gets wrong — that a DDATA structured metric arrives carrying the parent's
+///         <b>alias and nothing else</b>, so the member names cannot be composed until the alias table
+///         has resolved the parent.
+///     </para>
+///     <para>
+///         The issue's acceptance criterion, verbatim: "a simulator publishing a <c>Template</c> and a
+///         <c>DataSet</c> metric produces addressable, correctly-typed OPC UA nodes, live-verified;
+///         every still-unsupported kind continues to skip with a warning."
+///     </para>
+/// </remarks>
+[Trait("Category", "Integration")]
+[Collection("MqttFixture")]
+public sealed class SparkplugStructuredLiveTests(MqttFixture fixture)
+{
+    private const uint Good = 0x00000000u;
+    private static readonly TimeSpan Timeout = TimeSpan.FromSeconds(20);
+
+    private void SkipIfUnconfigured() => Assert.SkipWhen(fixture.SkipReason is not null, fixture.SkipReason ?? "");
+
+    [Fact]
+    public async Task A_template_metric_flattens_to_addressable_tags_across_birth_and_data()
+    {
+        SkipIfUnconfigured();
+
+        var ct = TestContext.Current.CancellationToken;
+        var group = NewGroup("tmpl");
+        const string node = "EdgeA";
+        const string device = "Filler1";
+
+        const string speedPath = "Fixture/Sparkplug/Motor/Speed";
+        const string runningPath = "Fixture/Sparkplug/Motor/Running";
+        const string statorPath = "Fixture/Sparkplug/Motor/Stator/Temp";
+
+        await using var sim = new SparkplugEdgeNode(SimOptions(group, node));
+        sim.SetDeviceMetrics(device, new SparkplugMetricSeed("Motor", 13, DataType.Template, MotorTemplate(speed: 1500, running: true, statorTemp: 71.5)));
+
+        await sim.ConnectAsync(ct);
+        await sim.BirthAsync(ct);
+
+        await using var driver = new MqttDriver(
+            SparkplugOptions(
+                group,
+                Tag(speedPath, group, node, device, "Motor/Speed"),
+                Tag(runningPath, group, node, device, "Motor/Running"),
+                Tag(statorPath, group, node, device, "Motor/Stator/Temp")),
+            "mqtt-live-sp-template");
+        var recorder = new Recorder(driver);
+
+        await driver.InitializeAsync(string.Empty, ct);
+        await driver.SubscribeAsync([speedPath, runningPath, statorPath], TimeSpan.FromSeconds(1), ct);
+
+        // The DBIRTH is a full snapshot, so these are the birth's own member values — including the
+        // NESTED template's, which is what proves the `/`-joined path composes past one level.
+        var birth = await recorder.WaitAsync(speedPath, v => Equals(v.Value, 1500d), Timeout, ct);
+        birth.Snapshot.StatusCode.ShouldBe(Good);
+        await recorder.WaitAsync(runningPath, v => Equals(v.Value, true), Timeout, ct);
+        await recorder.WaitAsync(statorPath, v => Equals(v.Value, 71.5d), Timeout, ct);
+
+        // …and now DDATA. The metric on the wire carries the PARENT's alias and no name at all, so a
+        // driver that flattened only at decode time — without resolving the parent alias to its name
+        // first — would have nothing to compose `Motor/Speed` from and would feed none of these.
+        await sim.PublishDeviceDataAsync(
+            device, [("Motor", MotorTemplate(speed: 1750, running: false, statorTemp: 68.25))], ct);
+
+        var data = await recorder.WaitAsync(speedPath, v => Equals(v.Value, 1750d), Timeout, ct);
+        data.Snapshot.StatusCode.ShouldBe(Good);
+        data.FullReference.ShouldBe(speedPath, "values publish under the tag's RawPath, never the metric name");
+
+        await recorder.WaitAsync(runningPath, v => Equals(v.Value, false), Timeout, ct);
+        await recorder.WaitAsync(statorPath, v => Equals(v.Value, 68.25d), Timeout, ct);
+
+        // The read path serves the same last values under the same keys.
+        var read = await driver.ReadAsync([speedPath, runningPath, statorPath], ct);
+        read.Select(r => r.StatusCode).ShouldAllBe(s => s == Good);
+        read[0].Value.ShouldBe(1750d);
+    }
+
+    [Fact]
+    public async Task A_dataset_metric_flattens_to_one_tag_per_cell()
+    {
+        SkipIfUnconfigured();
+
+        var ct = TestContext.Current.CancellationToken;
+        var group = NewGroup("dset");
+        const string node = "EdgeA";
+        const string device = "Filler1";
+
+        const string step0Path = "Fixture/Sparkplug/Recipe/Step0";
+        const string target1Path = "Fixture/Sparkplug/Recipe/Target1";
+
+        await using var sim = new SparkplugEdgeNode(SimOptions(group, node));
+        sim.SetDeviceMetrics(device, new SparkplugMetricSeed("Recipe", 14, DataType.DataSet, RecipeDataSet(baseTarget: 20d)));
+
+        await sim.ConnectAsync(ct);
+        await sim.BirthAsync(ct);
+
+        await using var driver = new MqttDriver(
+            SparkplugOptions(
+                group,
+                // The row index is always present, including on row 0 — the naming is uniform so a
+                // table that grows a row never renames the tags already authored against it.
+                Tag(step0Path, group, node, device, "Recipe/Step/0"),
+                Tag(target1Path, group, node, device, "Recipe/Target/1")),
+            "mqtt-live-sp-dataset");
+        var recorder = new Recorder(driver);
+
+        await driver.InitializeAsync(string.Empty, ct);
+        await driver.SubscribeAsync([step0Path, target1Path], TimeSpan.FromSeconds(1), ct);
+
+        // Cells are typed per COLUMN: Step is Int32, Target is Double. A dataset whose columns and
+        // types disagree is refused outright, so a wrong type here is not a near-miss — it is the
+        // difference between a typed tag and no tag.
+        await recorder.WaitAsync(step0Path, v => Equals(v.Value, 0), Timeout, ct);
+        await recorder.WaitAsync(target1Path, v => Equals(v.Value, 21d), Timeout, ct);
+
+        await sim.PublishDeviceDataAsync(device, [("Recipe", RecipeDataSet(baseTarget: 30d))], ct);
+
+        var target = await recorder.WaitAsync(target1Path, v => Equals(v.Value, 31d), Timeout, ct);
+        target.Snapshot.StatusCode.ShouldBe(Good);
+    }
+
+    [Fact]
+    public async Task A_template_definition_is_still_skipped_rather_than_flattened()
+    {
+        // Skip-and-warn survives — the requirement the issue is explicit about. A definition names its
+        // members and carries no values, so flattening it would materialise tags that can never read,
+        // indistinguishable in the picker from an instance's.
+        SkipIfUnconfigured();
+
+        var ct = TestContext.Current.CancellationToken;
+        var group = NewGroup("tdef");
+        const string node = "EdgeA";
+        const string device = "Filler1";
+
+        const string speedPath = "Fixture/Sparkplug/Proto/Speed";
+        const string livePath = "Fixture/Sparkplug/Proto/Live";
+
+        var definition = new Payload.Types.Template { Version = "1", IsDefinition = true };
+        definition.Metrics.Add(new Payload.Types.Metric { Name = "Speed", Datatype = (uint)DataType.Double });
+
+        await using var sim = new SparkplugEdgeNode(SimOptions(group, node));
+        sim.SetDeviceMetrics(
+            device,
+            new SparkplugMetricSeed("MotorType", 15, DataType.Template, definition),
+
+            // The positive control, and it is load-bearing: without it "Speed never arrived" is
+            // satisfied by any failure to connect, birth or subscribe, and the test would pass against
+            // a driver that ingests nothing at all.
+            new SparkplugMetricSeed("Live", 16, DataType.Double, 42.5d));
+
+        await sim.ConnectAsync(ct);
+        await sim.BirthAsync(ct);
+
+        await using var driver = new MqttDriver(
+            SparkplugOptions(
+                group,
+                Tag(speedPath, group, node, device, "MotorType/Speed"),
+                Tag(livePath, group, node, device, "Live")),
+            "mqtt-live-sp-tdef");
+        var recorder = new Recorder(driver);
+
+        await driver.InitializeAsync(string.Empty, ct);
+        await driver.SubscribeAsync([speedPath, livePath], TimeSpan.FromSeconds(1), ct);
+
+        // Wait for the control FIRST, so the absence assertion below is made at a moment the birth is
+        // known to have been processed.
+        await recorder.WaitAsync(livePath, v => Equals(v.Value, 42.5d), Timeout, ct);
+
+        recorder.Count(speedPath).ShouldBe(0, "a template definition carries no values to publish");
+
+        var read = await driver.ReadAsync([speedPath], ct);
+        read[0].StatusCode.ShouldNotBe(Good, "a tag bound to an unflattened member has never had a value");
+    }
+
+    // ---- helpers -----------------------------------------------------------------------------------
+
+    private static string NewGroup(string label) => $"OtOpcUaLive-{label}-{Guid.NewGuid():N}"[..40];
+
+    private static Payload.Types.Template MotorTemplate(double speed, bool running, double statorTemp)
+    {
+        var stator = new Payload.Types.Template { Version = "1", IsDefinition = false };
+        stator.Metrics.Add(new Payload.Types.Metric
+        {
+            Name = "Temp", Datatype = (uint)DataType.Double, DoubleValue = statorTemp,
+        });
+
+        var motor = new Payload.Types.Template { Version = "1", IsDefinition = false };
+        motor.Metrics.Add(new Payload.Types.Metric
+        {
+            Name = "Speed", Datatype = (uint)DataType.Double, DoubleValue = speed,
+        });
+        motor.Metrics.Add(new Payload.Types.Metric
+        {
+            Name = "Running", Datatype = (uint)DataType.Boolean, BooleanValue = running,
+        });
+        motor.Metrics.Add(new Payload.Types.Metric
+        {
+            Name = "Stator", Datatype = (uint)DataType.Template, TemplateValue = stator,
+        });
+        return motor;
+    }
+
+    private static Payload.Types.DataSet RecipeDataSet(double baseTarget)
+    {
+        var dataSet = new Payload.Types.DataSet { NumOfColumns = 2UL };
+        dataSet.Columns.Add("Step");
+        dataSet.Columns.Add("Target");
+        dataSet.Types_.Add((uint)DataType.Int32);
+        dataSet.Types_.Add((uint)DataType.Double);
+
+        for (var row = 0; row < 2; row++)
+        {
+            var r = new Payload.Types.DataSet.Types.Row();
+            r.Elements.Add(new Payload.Types.DataSet.Types.DataSetValue { IntValue = (uint)row });
+            r.Elements.Add(new Payload.Types.DataSet.Types.DataSetValue { DoubleValue = baseTarget + row });
+            dataSet.Rows.Add(r);
+        }
+
+        return dataSet;
+    }
+
+    private MqttDriverOptions SparkplugOptions(string groupId, params RawTagEntry[] rawTags) => new()
+    {
+        Host = fixture.TlsHost,
+        Port = fixture.TlsPort,
+        UseTls = true,
+        CaCertificatePath = fixture.CaCertificatePath,
+        AllowUntrustedServerCertificate = fixture.CaCertificatePath is null,
+        Username = fixture.Username,
+        Password = fixture.Password,
+        ClientId = $"otopcua-live-spx-{Guid.NewGuid():N}",
+        ConnectTimeoutSeconds = 15,
+        Mode = MqttMode.SparkplugB,
+        Sparkplug = new MqttSparkplugOptions { GroupId = groupId, RequestRebirthOnGap = true },
+        RawTags = rawTags,
+    };
+
+    private SparkplugEdgeNodeOptions SimOptions(string groupId, string edgeNodeId) => new()
+    {
+        Host = fixture.TlsHost,
+        Port = fixture.TlsPort,
+        UseTls = true,
+        CaCertificatePath = fixture.CaCertificatePath,
+        AllowUntrustedServerCertificate = fixture.CaCertificatePath is null,
+        Username = fixture.Username,
+        Password = fixture.Password,
+        GroupId = groupId,
+        EdgeNodeId = edgeNodeId,
+    };
+
+    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>Records every published data change, so a test can wait on one and count others.</summary>
+    private sealed class Recorder
+    {
+        private readonly List<DataChangeEventArgs> _events = [];
+        private readonly Lock _gate = new();
+
+        public Recorder(MqttDriver driver) => driver.OnDataChange += (_, e) =>
+        {
+            lock (_gate)
+            {
+                _events.Add(e);
+            }
+        };
+
+        public int Count(string rawPath)
+        {
+            lock (_gate)
+            {
+                return _events.Count(e => e.FullReference == rawPath);
+            }
+        }
+
+        public async Task<DataChangeEventArgs> WaitAsync(
+            string rawPath, Func<DataValueSnapshot, bool> predicate, TimeSpan timeout, CancellationToken ct)
+        {
+            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)
+                {
+                    lock (_gate)
+                    {
+                        throw new TimeoutException(
+                            $"No matching value for '{rawPath}' within {timeout}. Seen: "
+                            + string.Join(", ", _events.Select(e => $"{e.FullReference}={e.Snapshot.Value}")));
+                    }
+                }
+
+                await Task.Delay(100, ct);
+            }
+        }
+    }
+}
@@ -486,6 +486,31 @@ public sealed class MqttBrowseSessionTests
        s.PublishCountForTest.ShouldBe(0);   // passive — browsing a live plant broker publishes nothing
    }

+    [Fact]
+    public async Task SparkplugTree_TemplateMetric_OffersItsMembersAndNotItself()
+    {
+        // Gitea #515. The parent has no value and no servable datatype, so a leaf for `Motor` would
+        // offer the operator an address that can only ever commit a tag which never reads — and the
+        // picker gives no hint that one leaf is real and the other is not.
+        await using var s = new MqttBrowseSession(MqttMode.SparkplugB);
+        s.ObserveTemplateBirthForTest(
+            "Plant1", "EdgeA", "Filler1", "Motor",
+            ("Speed", SparkplugDataType.Double),
+            ("Torque", SparkplugDataType.Double));
+
+        var metrics = await s.ExpandAsync("Plant1/EdgeA/Filler1", TestContext.Current.CancellationToken);
+
+        metrics.Select(n => n.DisplayName).ShouldBe(["Motor/Speed", "Motor/Torque"]);
+        metrics.ShouldNotContain(n => n.DisplayName == "Motor");
+        metrics[0].Kind.ShouldBe(BrowseNodeKind.Leaf);
+
+        // The metric name travels INTACT as the node id — it is the binding key. Only the derived tag
+        // Name is later coerced to a legal RawPath segment, by RawBrowseCommitMapper (Gitea #509).
+        metrics[0].NodeId.ShouldBe("Plant1/EdgeA/Filler1::Motor/Speed");
+
+        s.PublishCountForTest.ShouldBe(0);
+    }
+
    [Fact]
    public async Task SparkplugTree_NodeLevelMetrics_HangDirectlyUnderTheEdgeNode()
    {
@@ -186,35 +186,82 @@ public sealed class SparkplugCodecTests
    }

    /// <summary>
-    ///     <c>DataSet</c> and <c>Template</c> metrics are out of scope for v1. They must decode to an
-    ///     explicit "unsupported" kind — never an exception on the MQTT dispatcher thread, and never a
-    ///     silently-dropped metric that looks identical to a metric that was never published.
+    ///     A <c>DataSet</c> metric decodes to <see cref="SparkplugValueKind.Structured"/> with its cells
+    ///     flattened (Gitea #515) — never an exception on the MQTT dispatcher thread, and never a
+    ///     silently-dropped metric that looks identical to one that was never published.
    /// </summary>
    [Fact]
-    public void Decode_GoldenNbirth_DataSetMetric_IsUnsupported_NotAThrow()
+    public void Decode_GoldenNbirth_DataSetMetric_FlattensToCells()
    {
        var payload = SparkplugCodec.Decode(SparkplugGoldenPayloads.Read(SparkplugGoldenPayloads.NbirthFile));

        var config = payload.Metrics.Single(m => m.Name == "Config");
-        config.ValueKind.ShouldBe(SparkplugValueKind.Unsupported);
+        config.ValueKind.ShouldBe(SparkplugValueKind.Structured);
+
+        // The PARENT carries no value — its members do. A parent that reported a value would be a
+        // guess about which cell "the" value is.
        config.Value.ShouldBeNull();
        config.DataType.ShouldBe(DataType.DataSet);
+
+        var cell = config.MemberList.Single();
+        cell.Path.ShouldBe("col0/0");
+        cell.DataType.ShouldBe(DataType.Int32);
+        cell.Value.ShouldBe(1U);
    }

-    /// <summary>A <c>Template</c> metric takes the same unsupported path as <c>DataSet</c>.</summary>
+    /// <summary>A <c>Template</c> metric flattens its members to sibling paths.</summary>
    [Fact]
-    public void Decode_TemplateMetric_IsUnsupported()
+    public void Decode_TemplateMetric_FlattensMembersToSiblingPaths()
    {
+        var template = new Payload.Types.Template { Version = "1", IsDefinition = false };
+        template.Metrics.Add(new Payload.Types.Metric
+        {
+            Name = "Speed", Datatype = (uint)DataType.Double, DoubleValue = 12.5,
+        });
+        template.Metrics.Add(new Payload.Types.Metric
+        {
+            Name = "Running", Datatype = (uint)DataType.Boolean, BooleanValue = true,
+        });
+
        var payload = new Payload();
        payload.Metrics.Add(new Payload.Types.Metric
        {
            Name = "Motor",
            Datatype = (uint)DataType.Template,
-            TemplateValue = new Payload.Types.Template { Version = "1", IsDefinition = false },
+            TemplateValue = template,
        });

-        SparkplugCodec.Decode(payload.ToByteArray()).Metrics.Single().ValueKind
-            .ShouldBe(SparkplugValueKind.Unsupported);
+        var metric = SparkplugCodec.Decode(payload.ToByteArray()).Metrics.Single();
+
+        metric.ValueKind.ShouldBe(SparkplugValueKind.Structured);
+        metric.MemberList.Select(m => m.Path).ShouldBe(["Speed", "Running"]);
+        metric.MemberList[0].Value.ShouldBe(12.5);
+        metric.MemberList[1].Value.ShouldBe(true);
+    }
+
+    /// <summary>
+    ///     A template <b>definition</b> is refused rather than flattened, and refusal is still the
+    ///     visible <see cref="SparkplugValueKind.Unsupported"/> kind.
+    /// </summary>
+    [Fact]
+    public void Decode_TemplateDefinition_StaysUnsupported()
+    {
+        // A definition is the UDT's SHAPE, published so consumers can understand instances — its member
+        // metrics carry no values. Flattening it would materialise a parallel set of tags that can
+        // never read, indistinguishable in the picker from the real instance's.
+        var definition = new Payload.Types.Template { Version = "1", IsDefinition = true };
+        definition.Metrics.Add(new Payload.Types.Metric { Name = "Speed", Datatype = (uint)DataType.Double });
+
+        var payload = new Payload();
+        payload.Metrics.Add(new Payload.Types.Metric
+        {
+            Name = "MotorType", Datatype = (uint)DataType.Template, TemplateValue = definition,
+        });
+
+        var metric = SparkplugCodec.Decode(payload.ToByteArray()).Metrics.Single();
+
+        metric.ValueKind.ShouldBe(SparkplugValueKind.Unsupported);
+        metric.MemberList.ShouldBeEmpty();
    }

    /// <summary>Payload and per-metric timestamps survive as raw Sparkplug epoch milliseconds.</summary>
@@ -0,0 +1,294 @@
+using Google.Protobuf;
+using Org.Eclipse.Tahu.Protobuf;
+using Shouldly;
+using Xunit;
+using ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Sparkplug;
+using TahuDataType = Org.Eclipse.Tahu.Protobuf.DataType;
+
+namespace ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Tests;
+
+/// <summary>
+///     Unit tests for <see cref="SparkplugFlattener"/> — the Template/DataSet/PropertySet mapping
+///     (Gitea #515).
+/// </summary>
+/// <remarks>
+///     Two properties dominate this file. First, <b>what is flattened is addressable and correctly
+///     typed</b>. Second — and the one that would be easy to lose in a later edit —
+///     <b>skip-and-warn survives</b>: anything that cannot be flattened honestly stays
+///     <see cref="SparkplugValueKind.Unsupported"/> rather than becoming a guessed scalar, because a
+///     tag bound to data that is not what its name says is worse than a tag that never binds.
+/// </remarks>
+[Trait("Category", "Unit")]
+public sealed class SparkplugFlattenerTests
+{
+    // ---- Template -------------------------------------------------------------------------------
+
+    [Fact]
+    public void A_nested_template_flattens_to_slash_joined_paths()
+    {
+        // `/` is the separator Sparkplug itself uses for metric-name hierarchy, so `Motor/Stator/Temp`
+        // is exactly what an edge node would have called a flat metric of the same meaning.
+        var stator = Template(
+            Member("Temp", TahuDataType.Double, m => m.DoubleValue = 71.5),
+            Member("Current", TahuDataType.Float, m => m.FloatValue = 3.25f));
+
+        var motor = Template(
+            Member("Speed", TahuDataType.Double, m => m.DoubleValue = 1500),
+            MemberTemplate("Stator", stator));
+
+        SparkplugFlattener.TryFlattenTemplate(motor, out var members).ShouldBeTrue();
+
+        members.Select(m => m.Path).ShouldBe(["Speed", "Stator/Temp", "Stator/Current"]);
+        members[1].DataType.ShouldBe(TahuDataType.Double);
+        members[2].Value.ShouldBe(3.25f);
+    }
+
+    [Fact]
+    public void A_template_member_with_no_name_refuses_the_whole_metric()
+    {
+        // A template member is addressed by name; there is no alias fallback inside a template. Emitting
+        // the members that DO have names would publish a UDT silently missing part of itself, which
+        // reads in the picker exactly like a complete one.
+        var template = Template(
+            Member("Speed", TahuDataType.Double, m => m.DoubleValue = 1),
+            new Payload.Types.Metric { Datatype = (uint)TahuDataType.Double, DoubleValue = 2 });
+
+        SparkplugFlattener.TryFlattenTemplate(template, out var members).ShouldBeFalse();
+        members.ShouldBeEmpty();
+    }
+
+    [Fact]
+    public void An_extension_value_inside_a_template_refuses_the_whole_metric()
+    {
+        // An extension value is a third-party escape hatch with no defined semantics. There is nothing
+        // honest to flatten it into, and the sibling members cannot be trusted to be the whole story.
+        var template = Template(
+            Member("Speed", TahuDataType.Double, m => m.DoubleValue = 1),
+            Member("Odd", TahuDataType.Unknown, m => m.ExtensionValue = new Payload.Types.Metric.Types.MetricValueExtension()));
+
+        SparkplugFlattener.TryFlattenTemplate(template, out _).ShouldBeFalse();
+    }
+
+    [Fact]
+    public void A_template_definition_is_refused()
+    {
+        // A definition carries the UDT's shape with no instance values — metadata for consumers, not
+        // signal. Flattening it would materialise tags that can never read.
+        var definition = Template(Member("Speed", TahuDataType.Double, _ => { }));
+        definition.IsDefinition = true;
+
+        SparkplugFlattener.TryFlattenTemplate(definition, out _).ShouldBeFalse();
+    }
+
+    [Fact]
+    public void Nesting_past_the_depth_bound_refuses_the_whole_metric_rather_than_truncating()
+    {
+        // This walks an object graph anyone on the broker can publish, so the recursion must be bounded.
+        // Refusing whole beats truncating: a silently half-flattened UDT looks exactly like a complete
+        // one, and the operator would author tags against a subtree that quietly stops partway down.
+        var deepest = Template(Member("Leaf", TahuDataType.Double, m => m.DoubleValue = 1));
+        var current = deepest;
+        for (var i = 0; i < SparkplugFlattener.MaxDepth + 1; i++)
+        {
+            current = Template(MemberTemplate($"L{i}", current));
+        }
+
+        SparkplugFlattener.TryFlattenTemplate(current, out var members).ShouldBeFalse();
+        members.ShouldBeEmpty();
+
+        // Falsifying control: the same shape one level inside the bound flattens fine, so the refusal
+        // above is the bound doing its job rather than nesting being broken outright.
+        var shallow = Template(MemberTemplate("L0", deepest));
+        SparkplugFlattener.TryFlattenTemplate(shallow, out var ok).ShouldBeTrue();
+        ok.Single().Path.ShouldBe("L0/Leaf");
+    }
+
+    // ---- DataSet ---------------------------------------------------------------------------------
+
+    [Fact]
+    public void A_dataset_flattens_to_one_member_per_cell_with_the_row_index_always_present()
+    {
+        // Uniform naming, deliberately. Emitting a bare `{column}` for a single row and
+        // `{column}/{row}` otherwise would make a tag's own name depend on how many rows the plant
+        // happened to publish, so a table growing from one row to two would rename every existing tag
+        // out from under its authored binding.
+        var dataSet = DataSet(
+            columns: ["Temp", "Pressure"],
+            types: [TahuDataType.Double, TahuDataType.Int32],
+            rows:
+            [
+                [Cell(v => v.DoubleValue = 20.5), Cell(v => v.IntValue = 3)],
+                [Cell(v => v.DoubleValue = 21.0), Cell(v => v.IntValue = 4)],
+            ]);
+
+        SparkplugFlattener.TryFlattenDataSet(dataSet, out var members).ShouldBeTrue();
+
+        members.Select(m => m.Path).ShouldBe(["Temp/0", "Pressure/0", "Temp/1", "Pressure/1"]);
+        members[0].DataType.ShouldBe(TahuDataType.Double);
+        members[1].DataType.ShouldBe(TahuDataType.Int32);
+        members[3].Value.ShouldBe(4U);
+    }
+
+    [Fact]
+    public void A_single_row_dataset_still_carries_the_row_index()
+        => Flatten(DataSet(["Temp"], [TahuDataType.Double], [[Cell(v => v.DoubleValue = 1)]]))
+            .Single().Path.ShouldBe("Temp/0");
+
+    [Fact]
+    public void An_empty_cell_is_a_null_member_not_a_zero()
+    {
+        // A DataSetValue has no is_null field, so an empty oneof IS the null cell. Reading it as the
+        // proto default would publish 0 for a cell the plant left blank — plausible, Good, and wrong.
+        var members = Flatten(DataSet(["Temp"], [TahuDataType.Double], [[new Payload.Types.DataSet.Types.DataSetValue()]]));
+
+        members.Single().ValueKind.ShouldBe(SparkplugValueKind.Null);
+        members.Single().Value.ShouldBeNull();
+    }
+
+    [Fact]
+    public void A_column_and_type_count_mismatch_refuses_the_dataset()
+        // Columns and types are parallel arrays. Guessing which to trust would type a column by its
+        // neighbour, and every cell in it would publish at Good quality as the wrong type.
+        => SparkplugFlattener.TryFlattenDataSet(
+                DataSet(["Temp", "Pressure"], [TahuDataType.Double], [] ), out _)
+            .ShouldBeFalse();
+
+    [Fact]
+    public void A_short_row_refuses_the_dataset()
+        // Not a row of nulls: emitting the cells that ARE present publishes a table silently missing
+        // its right-hand columns.
+        => SparkplugFlattener.TryFlattenDataSet(
+                DataSet(["Temp", "Pressure"], [TahuDataType.Double, TahuDataType.Int32],
+                    [[Cell(v => v.DoubleValue = 1)]]),
+                out _)
+            .ShouldBeFalse();
+
+    [Fact]
+    public void An_unnamed_column_refuses_the_dataset()
+        => SparkplugFlattener.TryFlattenDataSet(
+                DataSet(["", "Pressure"], [TahuDataType.Double, TahuDataType.Int32],
+                    [[Cell(v => v.DoubleValue = 1), Cell(v => v.IntValue = 2)]]),
+                out _)
+            .ShouldBeFalse();
+
+    [Fact]
+    public void A_dataset_past_the_member_bound_refuses_rather_than_truncating()
+    {
+        var columns = Enumerable.Range(0, 8).Select(i => $"c{i}").ToArray();
+        var types = Enumerable.Repeat(TahuDataType.Int32, 8).ToArray();
+        var rows = Enumerable.Range(0, 100)
+            .Select(_ => Enumerable.Range(0, 8).Select(_ => Cell(v => v.IntValue = 1)).ToArray())
+            .ToArray();
+
+        // 800 cells, past MaxMembers (512).
+        SparkplugFlattener.TryFlattenDataSet(DataSet(columns, types, rows), out var members).ShouldBeFalse();
+        members.ShouldBeEmpty();
+    }
+
+    // ---- PropertySet ------------------------------------------------------------------------------
+
+    [Fact]
+    public void Properties_project_to_flat_metadata_including_one_nested_level()
+    {
+        var nested = new Payload.Types.PropertySet();
+        nested.Keys.Add("Low");
+        nested.Values.Add(new Payload.Types.PropertyValue { DoubleValue = 0 });
+
+        var set = new Payload.Types.PropertySet();
+        set.Keys.Add("engUnit");
+        set.Values.Add(new Payload.Types.PropertyValue { StringValue = "degC" });
+        set.Keys.Add("Range");
+        set.Values.Add(new Payload.Types.PropertyValue { PropertysetValue = nested });
+
+        var projected = SparkplugFlattener.ProjectProperties(set);
+
+        projected.Select(p => p.Name).ShouldBe(["engUnit", "Range/Low"]);
+        projected[0].Value.ShouldBe("degC");
+    }
+
+    [Fact]
+    public void Properties_never_become_members()
+    {
+        // The recorded decision, and the honest reading of the spec: a PropertySet describes a metric
+        // (units, ranges, quality hints) — it is not a value the metric HAS. Surfacing it as sibling
+        // tags would invent plant signals out of documentation.
+        var set = new Payload.Types.PropertySet();
+        set.Keys.Add("engUnit");
+        set.Values.Add(new Payload.Types.PropertyValue { StringValue = "degC" });
+
+        var payload = new Payload();
+        var metric = new Payload.Types.Metric
+        {
+            Name = "Temperature", Datatype = (uint)TahuDataType.Double, DoubleValue = 20.5, Properties = set,
+        };
+        payload.Metrics.Add(metric);
+
+        var decoded = SparkplugCodec.Decode(payload.ToByteArray()).Metrics.Single();
+
+        decoded.ValueKind.ShouldBe(SparkplugValueKind.Scalar);
+        decoded.MemberList.ShouldBeEmpty();
+        decoded.PropertyList.Single().Name.ShouldBe("engUnit");
+    }
+
+    [Fact]
+    public void A_malformed_property_set_projects_to_nothing_rather_than_throwing()
+    {
+        var set = new Payload.Types.PropertySet();
+        set.Keys.Add("orphan");   // no matching value
+
+        SparkplugFlattener.ProjectProperties(set).ShouldBeEmpty();
+        SparkplugFlattener.ProjectProperties(null).ShouldBeEmpty();
+    }
+
+    // ---- helpers ----------------------------------------------------------------------------------
+
+    private static IReadOnlyList<SparkplugMember> Flatten(Payload.Types.DataSet dataSet)
+    {
+        SparkplugFlattener.TryFlattenDataSet(dataSet, out var members).ShouldBeTrue();
+        return members;
+    }
+
+    private static Payload.Types.Template Template(params Payload.Types.Metric[] members)
+    {
+        var template = new Payload.Types.Template { Version = "1", IsDefinition = false };
+        template.Metrics.AddRange(members);
+        return template;
+    }
+
+    private static Payload.Types.Metric Member(
+        string name, TahuDataType dataType, Action<Payload.Types.Metric> setValue)
+    {
+        var metric = new Payload.Types.Metric { Name = name, Datatype = (uint)dataType };
+        setValue(metric);
+        return metric;
+    }
+
+    private static Payload.Types.Metric MemberTemplate(string name, Payload.Types.Template value) =>
+        new() { Name = name, Datatype = (uint)TahuDataType.Template, TemplateValue = value };
+
+    private static Payload.Types.DataSet.Types.DataSetValue Cell(
+        Action<Payload.Types.DataSet.Types.DataSetValue> setValue)
+    {
+        var cell = new Payload.Types.DataSet.Types.DataSetValue();
+        setValue(cell);
+        return cell;
+    }
+
+    private static Payload.Types.DataSet DataSet(
+        string[] columns,
+        TahuDataType[] types,
+        Payload.Types.DataSet.Types.DataSetValue[][] rows)
+    {
+        var dataSet = new Payload.Types.DataSet { NumOfColumns = (ulong)columns.Length };
+        dataSet.Columns.AddRange(columns);
+        dataSet.Types_.AddRange(types.Select(t => (uint)t));
+        foreach (var cells in rows)
+        {
+            var row = new Payload.Types.DataSet.Types.Row();
+            row.Elements.AddRange(cells);
+            dataSet.Rows.Add(row);
+        }
+
+        return dataSet;
+    }
+}
@@ -132,8 +132,9 @@ internal static class SparkplugGoldenPayloads
            IntValue = unchecked((uint)CountBirthValue),
        });

-        // DataSet is explicitly out of scope for v1 — present here so "unsupported" is a decoded,
-        // testable outcome rather than an exception on the MQTT dispatcher thread.
+        // A single-column, single-row DataSet. Gitea #515 flattens it to the addressable member
+        // `Config/col0/0`; before that it decoded as an explicit "unsupported" kind. Kept in the
+        // golden vector either way — it is the only structured metric on the wire-format path.
        var dataSet = new Payload.Types.DataSet { NumOfColumns = 1UL };
        dataSet.Columns.Add("col0");
        dataSet.Types_.Add((uint)DataType.Int32);
@@ -0,0 +1,283 @@
+using Google.Protobuf;
+using Org.Eclipse.Tahu.Protobuf;
+using Shouldly;
+using Xunit;
+using ZB.MOM.WW.OtOpcUa.Core.Abstractions;
+using ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Sparkplug;
+using TahuDataType = Org.Eclipse.Tahu.Protobuf.DataType;
+
+namespace ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Tests;
+
+/// <summary>
+///     End-to-end ingest of flattened structured metrics (Gitea #515): a tag authored against
+///     <c>Motor/Speed</c> is fed by a DBIRTH/DDATA carrying a <c>Template</c> metric named
+///     <c>Motor</c>.
+/// </summary>
+/// <remarks>
+///     <see cref="SparkplugFlattenerTests"/> pins the mapping in isolation; this file pins that the
+///     chain actually connects — flatten → alias-table binding → authored-tag lookup → publish. Those
+///     are four separate places a member name is composed, and a mismatch in any one of them produces
+///     a driver that decodes structured metrics perfectly and feeds nothing.
+/// </remarks>
+[Trait("Category", "Unit")]
+public sealed class SparkplugStructuredIngestTests
+{
+    private const string Group = "Plant1";
+    private const string Node = "EdgeA";
+    private const string Device = "Filler1";
+    private const ulong MotorAlias = 20UL;
+    private const ulong TableAlias = 21UL;
+    private const uint StatusGood = 0x00000000u;
+
+    private const string SpeedPath = "Plant/Mqtt/spb/MotorSpeed";
+    private const string TempCellPath = "Plant/Mqtt/spb/TempCell0";
+
+    [Fact]
+    public async Task A_template_member_feeds_the_tag_authored_against_its_flattened_name()
+    {
+        // The whole point of #515, asserted end to end. `Motor` is what the wire carries; `Motor/Speed`
+        // is what an operator authors, and nothing on the wire is ever called that.
+        var (ing, fired) = await NewIngestorAsync(Tag(SpeedPath, Blob("Motor/Speed", "Float64")));
+
+        ing.Dispatch(Topic(SparkplugMessageType.NBIRTH), Nbirth());
+        ing.Dispatch(Topic(SparkplugMessageType.DBIRTH, Device), Birth(seq: 1, MotorTemplateMetric(speed: 1500, running: true)));
+
+        fired.ShouldContainKey(SpeedPath);
+        fired[SpeedPath].Value.ShouldBe(1500d);
+        fired[SpeedPath].StatusCode.ShouldBe(StatusGood);
+    }
+
+    [Fact]
+    public async Task A_ddata_template_carrying_only_the_parents_alias_still_reaches_the_member_tag()
+    {
+        // The load-bearing case, and the one a naive implementation gets wrong. A DDATA metric carries
+        // an ALIAS and no name — so the member names cannot be composed until the alias table has
+        // resolved the parent's name. Flattening at decode time alone is not enough.
+        var (ing, fired) = await NewIngestorAsync(Tag(SpeedPath, Blob("Motor/Speed", "Float64")));
+
+        ing.Dispatch(Topic(SparkplugMessageType.NBIRTH), Nbirth());
+        ing.Dispatch(Topic(SparkplugMessageType.DBIRTH, Device), Birth(seq: 1, MotorTemplateMetric(speed: 1500, running: true)));
+        fired.Clear();
+
+        // seq 2, alias only — no Name on the metric at all.
+        var data = new Payload { Seq = 2UL, Timestamp = 1721822405000UL };
+        data.Metrics.Add(new Payload.Types.Metric
+        {
+            Alias = MotorAlias,
+            TemplateValue = TemplateOf(speed: 1750, running: false),
+        });
+
+        ing.Dispatch(Topic(SparkplugMessageType.DDATA, Device), Decode(data));
+
+        fired.ShouldContainKey(SpeedPath);
+        fired[SpeedPath].Value.ShouldBe(1750d);
+    }
+
+    [Fact]
+    public async Task A_dataset_cell_feeds_the_tag_authored_against_its_indexed_name()
+    {
+        var (ing, fired) = await NewIngestorAsync(Tag(TempCellPath, Blob("Table/Temp/0", "Float64")));
+
+        ing.Dispatch(Topic(SparkplugMessageType.NBIRTH), Nbirth());
+        ing.Dispatch(Topic(SparkplugMessageType.DBIRTH, Device), Birth(seq: 1, TableDataSetMetric(20.5, 21.5)));
+
+        fired.ShouldContainKey(TempCellPath);
+        fired[TempCellPath].Value.ShouldBe(20.5d);
+    }
+
+    [Fact]
+    public async Task A_member_that_carries_no_datatype_falls_back_to_the_birth_binding()
+    {
+        // A DDATA template is not obliged to re-declare its members' datatypes. Guessing from the CLR
+        // type of the wire value would read a signed Int32 of -42 out of its unsigned field as
+        // 4294967254 — plausible, Good quality, and wrong.
+        var (ing, fired) = await NewIngestorAsync(Tag(SpeedPath, Blob("Motor/Offset", "Int32")));
+
+        var birthTemplate = new Payload.Types.Template { Version = "1", IsDefinition = false };
+        birthTemplate.Metrics.Add(new Payload.Types.Metric
+        {
+            Name = "Offset", Datatype = (uint)TahuDataType.Int32, IntValue = unchecked((uint)-42),
+        });
+
+        ing.Dispatch(Topic(SparkplugMessageType.NBIRTH), Nbirth());
+        ing.Dispatch(Topic(SparkplugMessageType.DBIRTH, Device), Birth(seq: 1, new Payload.Types.Metric
+        {
+            Name = "Motor", Alias = MotorAlias, Datatype = (uint)TahuDataType.Template, TemplateValue = birthTemplate,
+        }));
+
+        fired[SpeedPath].Value.ShouldBe(-42);
+        fired.Clear();
+
+        // DDATA member with NO datatype of its own.
+        var dataTemplate = new Payload.Types.Template { Version = "1", IsDefinition = false };
+        dataTemplate.Metrics.Add(new Payload.Types.Metric { Name = "Offset", IntValue = unchecked((uint)-7) });
+
+        var data = new Payload { Seq = 2UL, Timestamp = 1721822405000UL };
+        data.Metrics.Add(new Payload.Types.Metric { Alias = MotorAlias, TemplateValue = dataTemplate });
+        ing.Dispatch(Topic(SparkplugMessageType.DDATA, Device), Decode(data));
+
+        // -7, not 4294967289: the birth's Int32 binding drove ReinterpretSigned.
+        fired[SpeedPath].Value.ShouldBe(-7);
+    }
+
+    [Fact]
+    public async Task A_tag_authored_against_the_structured_parent_itself_is_never_fed()
+    {
+        // The parent has no value — its members do. Feeding it anything would require choosing which
+        // member "the" value is, which is exactly the guess #515 refuses to make.
+        var (ing, fired) = await NewIngestorAsync(Tag(SpeedPath, Blob("Motor", "Float64")));
+
+        ing.Dispatch(Topic(SparkplugMessageType.NBIRTH), Nbirth());
+        ing.Dispatch(Topic(SparkplugMessageType.DBIRTH, Device), Birth(seq: 1, MotorTemplateMetric(speed: 1500, running: true)));
+
+        fired.ShouldNotContainKey(SpeedPath);
+    }
+
+    [Fact]
+    public async Task An_unflattenable_structured_metric_still_feeds_nothing_rather_than_a_guess()
+    {
+        // Skip-and-warn survives — the explicit requirement on the issue. A template DEFINITION is the
+        // realistic instance of this: it names members and carries no values.
+        var definition = new Payload.Types.Template { Version = "1", IsDefinition = true };
+        definition.Metrics.Add(new Payload.Types.Metric { Name = "Speed", Datatype = (uint)TahuDataType.Double });
+
+        var (ing, fired) = await NewIngestorAsync(Tag(SpeedPath, Blob("Motor/Speed", "Float64")));
+
+        ing.Dispatch(Topic(SparkplugMessageType.NBIRTH), Nbirth());
+        ing.Dispatch(Topic(SparkplugMessageType.DBIRTH, Device), Birth(seq: 1, new Payload.Types.Metric
+        {
+            Name = "Motor", Alias = MotorAlias, Datatype = (uint)TahuDataType.Template, TemplateValue = definition,
+        }));
+
+        fired.ShouldNotContainKey(SpeedPath);
+    }
+
+    [Fact]
+    public async Task The_rediscovery_signature_tracks_members_not_the_parent_metric_name()
+    {
+        // A UDT gaining or losing a member leaves the PARENT metric's name identical, so a gate built
+        // on raw wire names would never move — and the operator would never be told the tree changed,
+        // which for a UDT is the change that matters most.
+        var (ing, _) = await NewIngestorAsync(Tag(SpeedPath, Blob("Motor/Speed", "Float64")));
+
+        var observed = new List<IReadOnlyList<string>>();
+        ing.BirthObserved += (_, e) => observed.Add(e.MetricNames);
+
+        ing.Dispatch(Topic(SparkplugMessageType.NBIRTH), Nbirth());
+        ing.Dispatch(Topic(SparkplugMessageType.DBIRTH, Device), Birth(seq: 1, MotorTemplateMetric(speed: 1, running: true)));
+
+        var deviceBirth = observed[^1];
+        deviceBirth.ShouldBe(["Motor/Speed", "Motor/Running"], ignoreOrder: true);
+        deviceBirth.ShouldNotContain("Motor");
+    }
+
+    // ---- helpers ---------------------------------------------------------------------------------
+
+    /// <summary>
+    ///     A registered <b>and subscribed</b> ingestor. The subscribe step is not optional scaffolding:
+    ///     <c>SparkplugIngestor.Raise</c> only fires <c>OnDataChange</c> for a RawPath that holds a
+    ///     subscription handle, so an unsubscribed ingestor caches values silently and every assertion
+    ///     in this file would read as "no value" whether or not the flattening worked.
+    /// </summary>
+    private static async Task<(SparkplugIngestor Ingestor, Dictionary<string, DataValueSnapshot> Fired)> NewIngestorAsync(
+        params RawTagEntry[] tags)
+    {
+        var options = new MqttDriverOptions
+        {
+            Mode = MqttMode.SparkplugB,
+            Sparkplug = new MqttSparkplugOptions { GroupId = Group, RequestRebirthOnGap = false },
+            RawTags = tags,
+        };
+
+        var ingestor = new SparkplugIngestor(
+            options,
+            "drv-1",
+            subscribeTransport: new GrantingSubscribeTransport(),
+            publishTransport: null,
+            logger: null,
+            cache: new LastValueCache());
+        ingestor.Register(tags);
+        await ingestor.SubscribeAsync(
+            [.. ingestor.AuthoredRawPaths], TimeSpan.Zero, TestContext.Current.CancellationToken);
+
+        var fired = new Dictionary<string, DataValueSnapshot>(StringComparer.Ordinal);
+        ingestor.OnDataChange += (_, e) => fired[e.FullReference] = e.Snapshot;
+        return (ingestor, fired);
+    }
+
+    /// <summary>A SUBACK that grants everything; the tests here are about values, not subscribe policy.</summary>
+    private sealed class GrantingSubscribeTransport : IMqttSubscribeTransport
+    {
+        public Task<IReadOnlyList<MqttSubscribeOutcome>> SubscribeAsync(
+            IReadOnlyList<MqttTopicSubscription> filters, CancellationToken cancellationToken)
+            => Task.FromResult<IReadOnlyList<MqttSubscribeOutcome>>(
+                [.. filters.Select(f => new MqttSubscribeOutcome(f.Topic, true, "GrantedQoS1"))]);
+    }
+
+    private static RawTagEntry Tag(string rawPath, string tagConfig) => new(rawPath, tagConfig, WriteIdempotent: false);
+
+    private static string Blob(string metric, string dataType) =>
+        $$"""{"groupId":"{{Group}}","edgeNodeId":"{{Node}}","deviceId":"{{Device}}","metricName":"{{metric}}","dataType":"{{dataType}}"}""";
+
+    private static SparkplugTopic Topic(SparkplugMessageType type, string? device = null) =>
+        new(type, Group, Node, device, null);
+
+    private static SparkplugPayload Decode(Payload payload) => SparkplugCodec.Decode(payload.ToByteArray());
+
+    private static SparkplugPayload Nbirth()
+    {
+        var payload = new Payload { Seq = 0UL, Timestamp = 1721822400000UL };
+        payload.Metrics.Add(new Payload.Types.Metric
+        {
+            Name = "bdSeq", Datatype = (uint)TahuDataType.Uint64, LongValue = 1UL,
+        });
+        return Decode(payload);
+    }
+
+    private static SparkplugPayload Birth(ulong seq, params Payload.Types.Metric[] metrics)
+    {
+        var payload = new Payload { Seq = seq, Timestamp = 1721822400000UL };
+        payload.Metrics.AddRange(metrics);
+        return Decode(payload);
+    }
+
+    private static Payload.Types.Template TemplateOf(double speed, bool running)
+    {
+        var template = new Payload.Types.Template { Version = "1", IsDefinition = false };
+        template.Metrics.Add(new Payload.Types.Metric
+        {
+            Name = "Speed", Datatype = (uint)TahuDataType.Double, DoubleValue = speed,
+        });
+        template.Metrics.Add(new Payload.Types.Metric
+        {
+            Name = "Running", Datatype = (uint)TahuDataType.Boolean, BooleanValue = running,
+        });
+        return template;
+    }
+
+    private static Payload.Types.Metric MotorTemplateMetric(double speed, bool running) => new()
+    {
+        Name = "Motor",
+        Alias = MotorAlias,
+        Datatype = (uint)TahuDataType.Template,
+        TemplateValue = TemplateOf(speed, running),
+    };
+
+    private static Payload.Types.Metric TableDataSetMetric(params double[] temps)
+    {
+        var dataSet = new Payload.Types.DataSet { NumOfColumns = 1UL };
+        dataSet.Columns.Add("Temp");
+        dataSet.Types_.Add((uint)TahuDataType.Double);
+        foreach (var temp in temps)
+        {
+            var row = new Payload.Types.DataSet.Types.Row();
+            row.Elements.Add(new Payload.Types.DataSet.Types.DataSetValue { DoubleValue = temp });
+            dataSet.Rows.Add(row);
+        }
+
+        return new Payload.Types.Metric
+        {
+            Name = "Table", Alias = TableAlias, Datatype = (uint)TahuDataType.DataSet, DatasetValue = dataSet,
+        };
+    }
+}