fix(sql): attribute a duplicated key to its own tags; key-value cap gets 2x headroom (R06:S-1 review I-1, m-3)

One stray duplicate row used to fail every tag in a key-value group or none of them,
depending on whether an unrelated key was absent. Duplicated keys are now refused
per key: only that key's tags publish BadConfigurationError and the rest of the group
reads normally. The key-value cap reads up to 2x the distinct keys (+1 to detect
excess), so a few stray duplicates are attributed per key and only a gross excess
(history/events table) fails the whole group. Duplicated-key refusals count as
answered refusals for health. The row-cap, duplicate-key and multi-row warnings now
use the per-plan limiter (m-3); the reader-wide limiter is removed. Unit tests pin
both the all-keys-present and absent-key cases with Good controls; live SQL Server
gate added (23/23).

Refs #744
Claude-Session: https://claude.ai/code/session_01Ub9hGmYLQvn9XETzw7CMsT

src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Sql/SqlDriver.cs

@@ -542,7 +542,7 @@ public sealed class SqlDriver
         try
         {
             var result = await _reader.ReadDetailedAsync(fullReferences, cancellationToken).ConfigureAwait(false);
-            ObservePollOutcome(result.Snapshots, result.RowCapExceededGroups, generation);
+            ObservePollOutcome(result.Snapshots, result.AnsweredRefusalGroups, generation);
             return result.Snapshots;
         }
         catch (OperationCanceledException)

src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Sql/SqlGroupPlanner.cs

@@ -230,10 +230,12 @@ public static class SqlGroupPlanner
             values.Add(keyValue);
         }
 
-        // Row cap (R06:S-1): one row per distinct key is all this plan can use, so it asks for ONE more — enough
-        // to see a source breaking the one-row-per-key contract, never enough to pull a history table's every
-        // matching row into memory on every poll.
-        var rowCap = values.Count;
+        // Row cap (R06:S-1): one row per distinct key is all this plan can use. It reads up to TWICE that, so a
+        // handful of stray duplicates stays inside the cap and is attributed PER KEY by the reader (only the
+        // duplicated key's tags go BadConfigurationError) — and asks for one row more than that, so only a
+        // gross excess (a history/events table, unattributable by construction because the rest is unread)
+        // fails the whole group. Memory stays O(keys) either way.
+        var rowCap = values.Count * 2;
         var sql = new StringBuilder("SELECT ")
             .Append(dialect.RowLimitPrefix(rowCap + 1))
             .Append(QuoteList(selected, dialect))

src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Sql/SqlPollReader.cs

@@ -68,7 +68,6 @@ public sealed class SqlPollReader : IDisposable
 
     private int _disposed;
 
-    private long _lastContractWarningTicks;
 
     /// <summary>
     ///     Per-key last-warned instant for the per-poll warnings (R06:S-6): a frozen group's BadTimeout and a
@@ -215,7 +214,7 @@ public sealed class SqlPollReader : IDisposable
                 results[i] = new DataValueSnapshot(null, SqlStatusCodes.BadNodeIdUnknown, null, readAt);
         }
 
-        var rowCapExceeded = shape.Definitions.Count > 0
+        var answeredRefusals = shape.Definitions.Count > 0
             ? await ReadGroupsAsync(shape, results, readAt, cancellationToken).ConfigureAwait(false)
             : 0;
 
@@ -224,7 +223,7 @@ public sealed class SqlPollReader : IDisposable
         for (var i = 0; i < results.Length; i++)
             results[i] ??= new DataValueSnapshot(null, SqlStatusCodes.BadInternalError, null, readAt);
 
-        return new SqlPollResult(results, rowCapExceeded);
+        return new SqlPollResult(results, answeredRefusals);
     }
 
     /// <summary>How many times a reference list has been (re)planned. Test observation of the S-7 cache.</summary>
@@ -301,7 +300,7 @@ public sealed class SqlPollReader : IDisposable
 
     /// <summary>
     ///     Runs every planned group concurrently and slices the results back. Returns how many groups came
-    ///     back over their row cap.
+    ///     back with an answered refusal (over the row cap, or a duplicated key).
     /// </summary>
     private async Task<int> ReadGroupsAsync(
         PollShape shape,
@@ -334,14 +333,14 @@ public sealed class SqlPollReader : IDisposable
         // leaves a sibling group running against a connection nobody is watching.
         var outcomes = await Task.WhenAll(groups).ConfigureAwait(false);
 
-        var rowCapExceeded = 0;
+        var answeredRefusals = 0;
         for (var g = 0; g < plans.Count; g++)
         {
-            Slice(plans[g], outcomes[g], slots, results, readAt);
-            if (ReferenceEquals(outcomes[g], GroupResult.RowCapExceeded)) rowCapExceeded++;
+            var refusedSome = Slice(plans[g], outcomes[g], slots, results, readAt);
+            if (refusedSome || ReferenceEquals(outcomes[g], GroupResult.RowCapExceeded)) answeredRefusals++;
         }
 
-        return rowCapExceeded;
+        return answeredRefusals;
     }
 
     /// <summary>
@@ -622,14 +621,16 @@ public sealed class SqlPollReader : IDisposable
 
         if (plan.RowCap is { } rowCap && rows.Count > rowCap)
         {
-            if (ShouldWarnAboutContract())
+            // Per plan, not reader-wide: with two ambiguous groups the reader-wide limiter could name only the
+            // first table forever, and with health still Healthy this line is the operator's only pointer.
+            if (ShouldWarn("rowcap:" + plan.SqlText))
             {
                 var first = plan.Members[0];
                 var selector = plan.Model == SqlTagModel.KeyValue
-                    ? string.Concat(first.KeyColumn, " IN (", rowCap.ToString(CultureInfo.InvariantCulture), " key(s))")
+                    ? string.Concat(first.KeyColumn, " IN (", (rowCap / 2).ToString(CultureInfo.InvariantCulture), " key(s))")
                     : string.Concat(first.RowSelectorColumn, " = ", first.RowSelectorValue);
                 _logger.LogWarning(
-                    "Sql poll source '{Table}' returned more rows than the {Model} selector '{Selector}' can use " +
+                    "Sql poll source '{Table}' returned more rows than the {Model} selector '{Selector}' can attribute " +
                     "(at most {Cap}) in one poll, so the selector is ambiguous — a history/events table, or a " +
                     "where column that is not unique. Its {Members} tag(s) publish BadConfigurationError and " +
                     "nothing past row {Limit} is read. Point the tag at a current-values table or a latest-row view.",
@@ -658,8 +659,11 @@ public sealed class SqlPollReader : IDisposable
         }
     }
 
-    /// <summary>Maps one group's outcome onto every slot its members occupy in the caller's list.</summary>
-    private void Slice(
+    /// <summary>
+    ///     Maps one group's outcome onto every slot its members occupy in the caller's list. Returns whether any
+    ///     member was refused on an ANSWERED result (a duplicated key) — answer evidence the driver must count.
+    /// </summary>
+    private bool Slice(
         SqlQueryPlan plan,
         GroupResult outcome,
         Dictionary<SqlTagDefinition, List<int>> slots,
@@ -669,9 +673,11 @@ public sealed class SqlPollReader : IDisposable
         // KeyValue indexes rows by the key column; WideRow's members all read the one selected row, so the
         // where-column is not even in the result set (design §5.3). Both selections happen ONCE per group —
         // they are group-wide facts, and doing them per member would report a contract violation N times.
+        HashSet<string>? duplicatedKeys = null;
         var byKey = outcome.Succeeded && plan.Model == SqlTagModel.KeyValue
-            ? IndexByKey(plan, outcome)
+            ? IndexByKey(plan, outcome, out duplicatedKeys)
             : null;
+        var refusedSome = false;
         var wideRow = outcome.Succeeded && plan.Model == SqlTagModel.WideRow
             ? SelectWideRow(plan, outcome)
             : null;
@@ -681,26 +687,45 @@ public sealed class SqlPollReader : IDisposable
             // Members may repeat (two tags on one keyValue, or the same ref twice) — every occurrence is fed.
             if (!slots.TryGetValue(member, out var positions)) continue;
 
-            var snapshot = outcome.Succeeded
-                ? MapMember(plan, member, outcome, byKey, wideRow, readAt)
-                : new DataValueSnapshot(null, outcome.FailureStatus, null, readAt);
+            DataValueSnapshot snapshot;
+            if (!outcome.Succeeded)
+            {
+                snapshot = new DataValueSnapshot(null, outcome.FailureStatus, null, readAt);
+            }
+            else if (duplicatedKeys is not null && member.KeyValue is { } key && duplicatedKeys.Contains(key))
+            {
+                // More than one row for THIS tag's key: its selector is ambiguous — the same rule a where-pair
+                // follows — so it is a configuration error for this key's tags only. Every other key in the
+                // group is unaffected, whatever else is present or absent (R06:S-1 review I-1: it used to be
+                // last-row-wins Good, or the whole group Bad, depending on an unrelated key's absence).
+                snapshot = new DataValueSnapshot(null, SqlStatusCodes.BadConfigurationError, null, readAt);
+                refusedSome = true;
+            }
+            else
+            {
+                snapshot = MapMember(plan, member, outcome, byKey, wideRow, readAt);
+            }
 
             foreach (var position in positions) results[position] = snapshot;
         }
+
+        return refusedSome;
     }
 
     /// <summary>
-    ///     Builds the key → row index a key-value plan slices through. Later rows win, so a source that
-    ///     breaks the one-row-per-key contract (design §3.6) still yields a deterministic value; the
-    ///     violation is reported through a rate-limited warning rather than silently.
+    ///     Builds the key → row index a key-value plan slices through, and the set of keys that came back more
+    ///     than once — a source breaking the one-row-per-key contract (design §3.6). Those keys' tags publish
+    ///     <see cref="SqlStatusCodes.BadConfigurationError"/>; the violation is also reported through a
+    ///     per-plan rate-limited warning.
     /// </summary>
-    private Dictionary<string, object?[]> IndexByKey(SqlQueryPlan plan, GroupResult outcome)
+    private Dictionary<string, object?[]> IndexByKey(
+        SqlQueryPlan plan, GroupResult outcome, out HashSet<string>? duplicatedKeys)
     {
+        duplicatedKeys = null;
         var index = new Dictionary<string, object?[]>(StringComparer.Ordinal);
         if (plan.KeyColumn is null || !outcome.ColumnIndex.TryGetValue(plan.KeyColumn, out var keyAt))
             return index;
 
-        var duplicated = 0;
         foreach (var row in outcome.Rows)
         {
             // A NULL key cell indexes nothing: Convert.ToString would fold it to "", which would then be
@@ -711,20 +736,16 @@ public sealed class SqlPollReader : IDisposable
             // (an INTEGER station id, say) — the provider coerces on the way in, so match on the way out.
             var key = Convert.ToString(keyCell, CultureInfo.InvariantCulture);
             if (key is null) continue;
-            if (!index.TryAdd(key, row))
-            {
-                index[key] = row;
-                duplicated++;
-            }
+            if (!index.TryAdd(key, row)) (duplicatedKeys ??= new HashSet<string>(StringComparer.Ordinal)).Add(key);
         }
 
-        if (duplicated > 0 && ShouldWarnAboutContract())
+        if (duplicatedKeys is not null && ShouldWarn("dupkey:" + plan.SqlText))
         {
             _logger.LogWarning(
-                "Sql poll source '{Table}' returned more than one row for {Count} key(s) in one poll; the " +
-                "last row wins. The key-value model requires one row per key — point the tag at a " +
-                "current-values table or an operator-authored latest-per-key view.",
-                plan.Members[0].Table, duplicated);
+                "Sql poll source '{Table}' returned more than one row for {Count} key(s) in one poll (e.g. " +
+                "'{Example}'); those keys' tags publish BadConfigurationError. The key-value model requires one " +
+                "row per key — point the tag at a current-values table or an operator-authored latest-per-key view.",
+                plan.Members[0].Table, duplicatedKeys.Count, duplicatedKeys.First());
         }
 
         return index;
@@ -759,7 +780,7 @@ public sealed class SqlPollReader : IDisposable
     private object?[]? SelectWideRow(SqlQueryPlan plan, GroupResult outcome)
     {
         if (outcome.Rows.Count == 0) return null;
-        if (outcome.Rows.Count > 1 && ShouldWarnAboutContract())
+        if (outcome.Rows.Count > 1 && ShouldWarn("multirow:" + plan.SqlText))
         {
             var first = plan.Members[0];
             var selector = string.IsNullOrWhiteSpace(first.RowSelectorColumn)
@@ -920,15 +941,6 @@ public sealed class SqlPollReader : IDisposable
         }
     }
 
-    /// <summary>Rate-limits the source-contract warning so a misconfigured table cannot flood the log.</summary>
-    private bool ShouldWarnAboutContract()
-    {
-        var now = Stopwatch.GetTimestamp();
-        var last = Interlocked.Read(ref _lastContractWarningTicks);
-        if (last != 0 && Stopwatch.GetElapsedTime(last, now) < ContractWarningInterval) return false;
-        return Interlocked.CompareExchange(ref _lastContractWarningTicks, now, last) == last;
-    }
-
     /// <summary>One group's outcome: either a materialised result set, or the status its members inherit.</summary>
     private sealed class GroupResult
     {
@@ -993,11 +1005,11 @@ public sealed class SqlPollReader : IDisposable
 
 /// <summary>One poll's snapshots plus the answer evidence they cannot carry themselves.</summary>
 /// <param name="Snapshots">One snapshot per requested reference, at the same index.</param>
-/// <param name="RowCapExceededGroups">
-///     Groups the database answered with more rows than their plan can use; their members carry
-///     <see cref="SqlStatusCodes.BadConfigurationError"/>.
+/// <param name="AnsweredRefusalGroups">
+///     Groups the database ANSWERED but whose result the reader refused in whole or part — an over-cap result
+///     set, or a duplicated key. Those members carry <see cref="SqlStatusCodes.BadConfigurationError"/>.
 /// </param>
-public sealed record SqlPollResult(IReadOnlyList<DataValueSnapshot> Snapshots, int RowCapExceededGroups);
+public sealed record SqlPollResult(IReadOnlyList<DataValueSnapshot> Snapshots, int AnsweredRefusalGroups);
 
 /// <summary>
 ///     The OPC UA status codes the Sql driver publishes, and the quality-class predicates its tests read

src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Sql/SqlQueryPlan.cs

@@ -56,11 +56,13 @@ namespace ZB.MOM.WW.OtOpcUa.Driver.Sql;
 ///     reader's map from result-set ordinal to source column.
 /// </param>
 /// <param name="RowCap">
-///     The most rows this plan can USE — its distinct key count (key-value) or 1 (a where-pair wide row);
-///     <see langword="null"/> when the statement already limits itself (topByTimestamp). <see cref="SqlText"/>
-///     asks for <c>RowCap + 1</c>, and the reader stops there: receiving the extra row proves the selector is
-///     ambiguous (a history table, a non-unique where column) and the group publishes
-///     <c>BadConfigurationError</c> instead of materialising every matching row on every poll (R06:S-1).
+///     The most rows the reader will attribute before refusing the whole group — twice the distinct key count
+///     (key-value: headroom so stray duplicates are attributed per key) or 1 (a where-pair wide row: one
+///     selector, so any second row is the group's ambiguity); <see langword="null"/> when the statement
+///     already limits itself (topByTimestamp). <see cref="SqlText"/> asks for <c>RowCap + 1</c>, and the
+///     reader stops there: receiving the extra row means a history/events table or a non-unique where column,
+///     and the group publishes <c>BadConfigurationError</c> instead of materialising every matching row on
+///     every poll (R06:S-1).
 /// </param>
 public sealed record SqlQueryPlan(
     SqlTagModel Model,