feat(historian): Gateway serves the per-sample aggregates from raw reads with the shared Sql.Common evaluator (#681)

Average, Count, Minimum, Maximum, Range, Start, End, Delta and Minimum/MaximumActualTime are evaluated on the Gateway backend from forward ReadRaw reads with BucketPlan + AggregateEvaluator, so Gateway matches SqlServer/QuestDB exactly (values, status bits, interval-start stamps, #677 reverse). - Window read [start - 1 ms, end] so AVEVA's initial-value row never lands in a bucket; a pre-window row proves earlier data only with a non-Bad OPC quality (histsdk puts a value on every NULL row). - Earlier data otherwise from a 2-row [1970, start) probe; later data from a 2-row [end, now+1d] look-ahead; both ignore rows stamped at their query start/end. - No gateway stream is ever abandoned (a client-side cancel makes the gateway evict its shared historian session); past ServerHistorian:MaxTimeWeightedRawRows the bounded stream is drained, then BadTooManyOperations. The gateway's max_values-ceiling InvalidArgument maps to the same limit; a caller cancellation is not a backend fault. - TimeAverage/Total stay AVEVA-native (no prior-sample read for exact Part 13 bounds), now stamped with the interval start (PENDING the live leg: histsdk's tail+43 start_time field is unverified) and carrying the Calculated bits. - Served set: GatewayHistorianDataSource.ServedAggregates (12). Live AVEVA leg open (VPN).
2026-10-12 09:55:32 -04:00
parent dbdf29c8fc
commit 78f604e0cd
16 changed files with 1187 additions and 77 deletions
@@ -1,7 +1,10 @@
+using System.Collections.Frozen;
+using Grpc.Core;
 using Microsoft.Extensions.Logging;
 using ZB.MOM.WW.HistorianGateway.Contracts.Grpc;
 using ZB.MOM.WW.OtOpcUa.Core.Abstractions;
 using ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway.Mapping;
+using ZB.MOM.WW.OtOpcUa.Driver.Historian.Sql.Common;

 namespace ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway;

@@ -23,8 +26,26 @@ namespace ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway;
 /// </remarks>
 public sealed class GatewayHistorianDataSource : IHistorianDataSource, IAsyncDisposable
 {
+    /// <summary>
+    ///     The processed (aggregate) functions the Gateway backend serves — the single source of truth for this backend's
+    ///     aggregate set (#681): the per-sample aggregates evaluated from raw reads (every member of the
+    ///     <see cref="AggregateFamily.Simple"/> / <see cref="AggregateFamily.StartEnd"/> /
+    ///     <see cref="AggregateFamily.ActualTime"/> families of <see cref="AggregateFamilies"/>: Average, Count, Minimum,
+    ///     Maximum, Range, Start, End, Delta, MinimumActualTime, MaximumActualTime) plus the two answered natively by AVEVA
+    ///     (TimeAverage, Total — <see cref="AggregateModeMapper"/>). Any other aggregate throws
+    ///     <see cref="HistoryAggregateNotSupportedException"/> without calling the gateway. It is what
+    ///     <see cref="SupportedAggregates"/> returns, so the advertised <c>AggregateFunctions</c> folders and the server's
+    ///     up-front <c>BadAggregateNotSupported</c> read it too (#682).
+    /// </summary>
+    public static IReadOnlySet<HistoryAggregateType> ServedAggregates { get; } =
+        Enum.GetValues<HistoryAggregateType>()
+            .Where(a => RawAggregateReader.Serves(a) || AggregateModeMapper.NativeAggregates.Contains(a))
+            .ToFrozenSet();
+
    private readonly IHistorianGatewayClient _client;
    private readonly ILogger<GatewayHistorianDataSource> _logger;
+    private readonly int _maxRawRows;
+    private readonly TimeProvider _time;

    private readonly object _healthLock = new();
    private DateTime? _lastSuccessUtc;
@@ -38,12 +59,20 @@ public sealed class GatewayHistorianDataSource : IHistorianDataSource, IAsyncDis
    /// <summary>Creates a gateway-backed historian data source.</summary>
    /// <param name="client">The gateway client seam used for all reads.</param>
    /// <param name="logger">Diagnostic logger; failures are recorded without leaking tag/host detail.</param>
-    public GatewayHistorianDataSource(IHistorianGatewayClient client, ILogger<GatewayHistorianDataSource> logger)
+    /// <param name="maxRawRows">
+    ///     In-window raw-row cap of one raw-evaluated processed read (<c>ServerHistorian:MaxTimeWeightedRawRows</c>;
+    ///     &lt;= 0 = unbounded). Past it the node fails <c>BadTooManyOperations</c>.
+    /// </param>
+    /// <param name="timeProvider">Clock for result server timestamps and the raw look-ahead horizon; null = system.</param>
+    public GatewayHistorianDataSource(IHistorianGatewayClient client, ILogger<GatewayHistorianDataSource> logger,
+        int maxRawRows = TimeWeightedAggregator.DefaultMaxWindowRows, TimeProvider? timeProvider = null)
    {
        ArgumentNullException.ThrowIfNull(client);
        ArgumentNullException.ThrowIfNull(logger);
        _client = client;
        _logger = logger;
+        _maxRawRows = maxRawRows;
+        _time = timeProvider ?? TimeProvider.System;
    }

    /// <inheritdoc />
@@ -75,17 +104,38 @@ public sealed class GatewayHistorianDataSource : IHistorianDataSource, IAsyncDis
    }

    /// <inheritdoc />
+    /// <remarks>
+    ///     <para>
+    ///         <b>Per-sample aggregates (#681)</b> — Average, Count, Minimum, Maximum, Range, Start, End, Delta,
+    ///         Minimum/MaximumActualTime — are evaluated from a raw read with the SQL backends' shared evaluator
+    ///         (<see cref="RawAggregateReader"/>): identical values, status, Calculated/Partial bits and timestamps to
+    ///         SqlServer/QuestDB for the same samples, forward and reverse (#677). The in-window raw rows are capped by
+    ///         <c>ServerHistorian:MaxTimeWeightedRawRows</c> (<see cref="HistoryReadLimitExceededException"/>, not counted as
+    ///         a backend failure).
+    ///     </para>
+    ///     <para>
+    ///         <b>TimeAverage / Total</b> stay native AVEVA retrieval modes (<see cref="AggregateModeMapper"/>), stamped with
+    ///         the interval start and the Calculated bit. Everything else is <see cref="HistoryAggregateNotSupportedException"/>
+    ///         (see <see cref="ServedAggregates"/>).
+    ///     </para>
+    /// </remarks>
+    /// <exception cref="ArgumentOutOfRangeException">A raw-evaluated read whose window holds more than
+    ///     <see cref="BucketPlan.MaxBuckets"/> intervals.</exception>
    public async Task<HistoryReadResult> ReadProcessedAsync(
        string fullReference, DateTime startUtc, DateTime endUtc, TimeSpan interval,
        HistoryAggregateType aggregate, CancellationToken cancellationToken)
    {
+        if (RawAggregateReader.Serves(aggregate))
+            return await ReadProcessedFromRawAsync(fullReference, startUtc, endUtc, interval, aggregate, cancellationToken)
+                .ConfigureAwait(false);
+
        // Mapped OUTSIDE the failure-recording try: an aggregate the gateway has no mode for is a client
        // request the server answers BadAggregateNotSupported, not a backend fault to count (#285).
        var mode = AggregateModeMapper.ToRetrievalMode(aggregate);
        try
        {
-            // Total is a native gateway retrieval mode (Integral) — no client-side scaling. Average and Count
-            // have no AVEVA mode and were rejected by the mapper above (#679).
+            // TimeAverage / Total are native gateway retrieval modes (TimeWeightedAverage / Integral) — no
+            // client-side scaling.
            var buckets = new List<HistorianAggregateSample>();
            await foreach (var bucket in _client
                .ReadAggregateAsync(fullReference, startUtc, endUtc, mode, interval, cancellationToken)
@@ -98,6 +148,10 @@ public sealed class GatewayHistorianDataSource : IHistorianDataSource, IAsyncDis
            RecordOutcome(success: true, error: null);
            return new HistoryReadResult(snapshots, ContinuationPoint: null);
        }
+        catch (OperationCanceledException) when (cancellationToken.IsCancellationRequested)
+        {
+            throw; // caller-requested, not a backend fault
+        }
        catch (Exception ex)
        {
            RecordReadFailure(ex);
@@ -105,6 +159,62 @@ public sealed class GatewayHistorianDataSource : IHistorianDataSource, IAsyncDis
        }
    }

+    /// <summary>Serves a per-sample aggregate from a raw read (#681); see <see cref="RawAggregateReader"/>.</summary>
+    private async Task<HistoryReadResult> ReadProcessedFromRawAsync(
+        string fullReference, DateTime startUtc, DateTime endUtc, TimeSpan interval,
+        HistoryAggregateType aggregate, CancellationToken cancellationToken)
+    {
+        // Planned OUTSIDE the failure-recording try (as on the SQL backends): too many intervals is a client request
+        // limit, not a backend fault.
+        var plan = BucketPlan.Create(startUtc, endUtc, interval);
+        try
+        {
+            var now = _time.GetUtcNow().UtcDateTime;
+            var values = await RawAggregateReader
+                .ReadAsync(_client, fullReference, aggregate, plan, _maxRawRows, now, cancellationToken)
+                .ConfigureAwait(false);
+            RecordOutcome(success: true, error: null);
+            return new HistoryReadResult(values, ContinuationPoint: null);
+        }
+        catch (HistoryReadLimitExceededException)
+        {
+            // A client-request limit (the raw-row cap), not a backend fault: not counted as a failure.
+            throw;
+        }
+        catch (OperationCanceledException) when (cancellationToken.IsCancellationRequested)
+        {
+            // The caller (client cancel / HistoryReadDeadline) asked for this — not a backend fault.
+            throw;
+        }
+        catch (Exception ex) when (IsMaxValuesCeilingRejection(ex))
+        {
+            // The gateway refused the row budget (Historian:ReadLimits:MaxValuesCeiling with
+            // AllowStreamingBeyondCeiling=false): the same client-request limit as the cap, not a backend fault.
+            throw new HistoryReadLimitExceededException(_maxRawRows,
+                $"The HistorianGateway refused the raw row budget of the processed read of '{fullReference}' (its " +
+                "Historian:ReadLimits:MaxValuesCeiling); lower ServerHistorian:MaxTimeWeightedRawRows below that ceiling " +
+                "or narrow the window.");
+        }
+        catch (Exception ex)
+        {
+            RecordReadFailure(ex);
+            throw;
+        }
+    }
+
+    /// <summary>
+    ///     True when <paramref name="ex"/> (or an inner exception — the client package wraps the <see cref="RpcException"/>)
+    ///     is the gateway's <c>InvalidArgument</c> "max_values exceeds the server ceiling" rejection.
+    /// </summary>
+    internal static bool IsMaxValuesCeilingRejection(Exception ex)
+    {
+        for (var e = ex; e is not null; e = e.InnerException)
+            if (e is RpcException { StatusCode: StatusCode.InvalidArgument } rpc
+                && rpc.Status.Detail.Contains("max_values exceeds the server ceiling", StringComparison.Ordinal))
+                return true;
+        return false;
+    }
+
    /// <inheritdoc />
    public async Task<HistoryReadResult> ReadAtTimeAsync(
        string fullReference, IReadOnlyList<DateTime> timestampsUtc, CancellationToken cancellationToken)
@@ -132,13 +242,6 @@ public sealed class GatewayHistorianDataSource : IHistorianDataSource, IAsyncDis
    /// </remarks>
    public bool SupportsReverseEvents => false;

-    /// <summary>
-    ///     The aggregates the Gateway backend serves — the single static the advertised <c>AggregateFunctions</c> folders
-    ///     and the server's up-front <c>BadAggregateNotSupported</c> read through <see cref="SupportedAggregates"/> (#682).
-    ///     Today: <see cref="AggregateModeMapper.Served"/>, the members the gateway has a retrieval mode for.
-    /// </summary>
-    public static readonly IReadOnlySet<HistoryAggregateType> ServedAggregates = AggregateModeMapper.Served;
-
    /// <inheritdoc />
    /// <remarks><see cref="ServedAggregates"/>; every other member answers <c>BadAggregateNotSupported</c> and is not advertised.</remarks>
    public IReadOnlySet<HistoryAggregateType> SupportedAggregates => ServedAggregates;
@@ -38,7 +38,9 @@ public static class GatewayHistorian

        return new GatewayHistorianDataSource(
            HistorianGatewayClientAdapter.Create(options, loggerFactory),
-            logger);
+            logger,
+            // #681: the raw-evaluated per-sample aggregates share the SQL backends' raw-row cap.
+            maxRawRows: options.MaxTimeWeightedRawRows);
    }

    /// <summary>
@@ -6,29 +6,33 @@ namespace ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway.Mapping;

 /// <summary>
 /// Maps the driver-agnostic <see cref="HistoryAggregateType"/> (OPC UA Part 13 aggregate) onto the
-/// gateway's native <see cref="RetrievalMode"/>.
+/// gateway's native <see cref="RetrievalMode"/> — for the aggregates still answered by AVEVA itself.
 /// </summary>
 /// <remarks>
 /// <para>
-/// Served: <see cref="HistoryAggregateType.TimeAverage"/> → <see cref="RetrievalMode.TimeWeightedAverage"/>,
-/// Minimum → <see cref="RetrievalMode.MinimumWithTime"/>, Maximum → <see cref="RetrievalMode.MaximumWithTime"/>,
-/// Total → <see cref="RetrievalMode.Integral"/>.
+/// Native (AVEVA-computed): <see cref="HistoryAggregateType.TimeAverage"/> → <see cref="RetrievalMode.TimeWeightedAverage"/>
+/// and <see cref="HistoryAggregateType.Total"/> → <see cref="RetrievalMode.Integral"/>. They stay native because an exact
+/// Part 13 time-weighted evaluation needs the real bounding samples on both sides of the window (time, value and
+/// quality of the latest sample before it), and the gateway has no "previous stored sample" read — a raw read streams
+/// forward from its start and AVEVA replaces whatever lies before with an initial-value row stamped at the query start,
+/// losing the original time a sloped segment across the window start needs (#681).
 /// </para>
 /// <para>
-/// Deliberately NOT served (#679): <see cref="HistoryAggregateType.Average"/> (OPC UA Average is the arithmetic mean;
-/// AVEVA's modes include no arithmetic mean — only the time-weighted one, which is TimeAverage) and
-/// <see cref="HistoryAggregateType.Count"/> (AVEVA <c>Counter</c> is a rollover-counter increase, not a sample count).
-/// Both surface as <c>BadAggregateNotSupported</c> rather than returning a different statistic under the requested name.
+/// Since #681 the per-sample aggregates (Average, Count, Minimum, Maximum, Range, Start, End, Delta,
+/// Minimum/MaximumActualTime) are NOT native: <see cref="GatewayHistorianDataSource"/> evaluates them from raw reads
+/// with the SQL backends' shared evaluator (Minimum/Maximum used to map to AVEVA's <c>MinimumWithTime</c> /
+/// <c>MaximumWithTime</c>). Every other member — incl. TimeAverage2, Total2, Interpolative, PercentGood, PercentBad — is
+/// <see cref="HistoryAggregateNotSupportedException"/> here.
 /// </para>
 /// </remarks>
 internal static class AggregateModeMapper
 {
    /// <summary>
-    ///     The aggregates the gateway serves — every defined member <see cref="TryMap"/> maps, derived from it so the
-    ///     advertised set and the dispatch can never disagree. <see cref="GatewayHistorianDataSource.SupportedAggregates"/>
-    ///     returns it (#682); it is the ONLY declaration of the gateway's served set.
+    ///     The aggregates answered by a native AVEVA retrieval mode — every defined member <see cref="TryMap"/> maps, derived
+    ///     from it so the dispatch and the declared set can never disagree. Part of
+    ///     <see cref="GatewayHistorianDataSource.ServedAggregates"/> (#681/#682).
    /// </summary>
-    public static readonly IReadOnlySet<HistoryAggregateType> Served =
+    public static readonly FrozenSet<HistoryAggregateType> NativeAggregates =
        Enum.GetValues<HistoryAggregateType>().Where(a => TryMap(a) is not null).ToFrozenSet();

    /// <summary>Maps an aggregate function to the gateway retrieval mode.</summary>
@@ -48,12 +52,10 @@ internal static class AggregateModeMapper
            nameof(aggregate), aggregate, "Unmapped HistoryAggregateType — add a RetrievalMode mapping.");
    }

-    /// <summary>The single mapping table: the retrieval mode for a served aggregate, or null.</summary>
+    /// <summary>The single mapping table: the retrieval mode for a natively served aggregate, or null.</summary>
    private static RetrievalMode? TryMap(HistoryAggregateType aggregate) => aggregate switch
    {
        HistoryAggregateType.TimeAverage => RetrievalMode.TimeWeightedAverage,
-        HistoryAggregateType.Minimum => RetrievalMode.MinimumWithTime,
-        HistoryAggregateType.Maximum => RetrievalMode.MaximumWithTime,
        HistoryAggregateType.Total => RetrievalMode.Integral,
        _ => null,
    };
@@ -1,5 +1,6 @@
 using ZB.MOM.WW.HistorianGateway.Contracts.Grpc;
 using ZB.MOM.WW.OtOpcUa.Core.Abstractions;
+using ZB.MOM.WW.OtOpcUa.Driver.Historian.Sql.Common;

 namespace ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway.Mapping;

@@ -30,16 +31,23 @@ internal static class SampleMapper
        else
            value = null;

+        return new DataValueSnapshot(
+            Value: value,
+            StatusCode: ToStatus(sample),
+            SourceTimestampUtc: sample.Timestamp?.ToDateTime(),  // Utc kind
+            ServerTimestampUtc: DateTime.UtcNow);
+    }
+
+    /// <summary>The OPC UA status of a raw sample (shared by raw reads and the #681 raw-evaluated aggregates).</summary>
+    /// <param name="sample">The gateway raw sample.</param>
+    /// <returns>The mapped status code.</returns>
+    public static uint ToStatus(HistorianSample sample)
+    {
        // Prefer the OPC DA quality byte (opc_quality); the gateway populates it directly from the
        // SDK's OpcQuality, so it is the authoritative byte for GatewayQualityMapper. Fall back to
        // the historian quality field only when opc_quality is unset (0).
        byte qualityByte = sample.OpcQuality != 0 ? (byte)sample.OpcQuality : (byte)sample.Quality;
-
-        return new DataValueSnapshot(
-            Value: value,
-            StatusCode: GatewayQualityMapper.Map(qualityByte),
-            SourceTimestampUtc: sample.Timestamp?.ToDateTime(),  // Utc kind
-            ServerTimestampUtc: DateTime.UtcNow);
+        return GatewayQualityMapper.Map(qualityByte);
    }

    /// <summary>Maps a batch of raw samples to value snapshots, in order.</summary>
@@ -57,20 +65,28 @@ internal static class SampleMapper
    /// <param name="aggregate">The gateway aggregate sample.</param>
    /// <returns>The driver-agnostic snapshot.</returns>
    /// <remarks>
+    /// <para>
    /// Unlike the legacy Wonderware DTO (a nullable <c>double?</c>), the gateway proto carries a
    /// non-optional <c>double value</c>, so an unavailable (no-data) bucket cannot be signalled by a
    /// null value. Instead it is signalled by a non-Good <c>opc_quality</c>: a Good bucket
    /// (<c>opc_quality &gt;= 192</c>) yields its value with a Good status, anything else maps to
    /// <c>BadNoData</c> with a null value — preserving the Wonderware aggregate contract (binary
    /// Good-with-value / BadNoData-null).
+    /// </para>
+    /// <para>
+    /// #681: both statuses carry the Calculated info bits (<c>0x401</c>) like every other backend's processed value, and
+    /// the bucket is stamped with its interval START (Part 13; it was the interval end), falling back to the end only when
+    /// the start is absent. Pending the live leg: histsdk decodes <c>start_time</c> from a row field (<c>tail + 43</c>) its
+    /// reverse-engineering notes still mark unverified.
+    /// </para>
    /// </remarks>
    public static DataValueSnapshot ToAggregateSnapshot(HistorianAggregateSample aggregate)
    {
        bool hasData = aggregate.OpcQuality >= GoodQualityFloor;
        return new DataValueSnapshot(
            Value: hasData ? aggregate.Value : null,                 // boxes as System.Double when present
-            StatusCode: hasData ? StatusGood : StatusBadNoData,
-            SourceTimestampUtc: (aggregate.EndTime ?? aggregate.StartTime)?.ToDateTime(),  // bucket timestamp
+            StatusCode: HistorianInfoBits.WithCalculated(hasData ? StatusGood : StatusBadNoData),
+            SourceTimestampUtc: (aggregate.StartTime ?? aggregate.EndTime)?.ToDateTime(),  // bucket timestamp = interval start
            ServerTimestampUtc: DateTime.UtcNow);
    }

@@ -0,0 +1,296 @@
+using ZB.MOM.WW.HistorianGateway.Contracts.Grpc;
+using ZB.MOM.WW.OtOpcUa.Core.Abstractions;
+using ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway.Mapping;
+using ZB.MOM.WW.OtOpcUa.Driver.Historian.Sql.Common;
+
+namespace ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway;
+
+/// <summary>
+///     #681: serves the per-sample processed aggregates (the <see cref="AggregateFamily.Simple"/>,
+///     <see cref="AggregateFamily.StartEnd"/> and <see cref="AggregateFamily.ActualTime"/> families) on the Gateway backend by
+///     streaming the window's RAW samples through the gateway and evaluating them with the shared
+///     <c>Driver.Historian.Sql.Common</c> code — <see cref="BucketPlan"/> for the bucket math (forward and reverse, #677) and
+///     <see cref="AggregateEvaluator.EvaluateAll"/> for values, status and the Partial/Calculated bits — so the Gateway answers
+///     exactly what SqlServer and QuestDB answer for the same stored samples.
+/// </summary>
+/// <remarks>
+///     <para>
+///         <b>Forward raw reads, each consumed to completion.</b> The gateway's <c>ReadRaw</c> streams oldest-first over an
+///         inclusive <c>[start, end]</c> and refuses <c>end &lt; start</c>, so a reverse read streams the same ascending range
+///         and <see cref="BucketPlan"/> assembles the result newest-first. The window stream is classified sample by sample
+///         with <see cref="BucketPlan.IndexOf"/> (<c>[start, end)</c> forward, <c>(start, end]</c> reverse) — memory is
+///         O(buckets), never O(samples). No gateway stream is ever abandoned mid-way: a client-side break cancels the call,
+///         and the gateway currently classifies that cancellation as a connection fault and evicts its shared historian
+///         session (HistorianGateway <c>HistorianRpcFaults.IsConnectionFault</c>: any <c>RpcException</c> ⇒ true).
+///     </para>
+///     <para>
+///         <b>The AVEVA start-boundary row.</b> An AVEVA <c>Full</c> raw read emits an initial-value row stamped exactly at
+///         the query start (histsdk reverse-engineering: <c>Quality = 133</c>, <c>OpcQuality = 192</c>) carrying the value
+///         in effect there, not a stored sample. The window stream is therefore requested from <see cref="BoundaryMargin"/>
+///         (1 ms) BEFORE the window, so that row lands outside every bucket and can never be counted as a sample. A row
+///         before the window counts only as evidence that the tag has earlier data, and only when its OPC quality is not
+///         Bad — a "no earlier data" NULL row still arrives WITH a value (histsdk sets <c>NumericValue = 0</c> on every
+///         numeric row, <c>StringValue = ""</c> on a string one) but with OPC quality 0.
+///     </para>
+///     <para>
+///         <b>The data range.</b> The start/end-of-data Partial rules and Count's in-range empty interval need the tag's
+///         earliest sample at or before the window's later edge and latest at or after its earlier edge
+///         (<see cref="AggregateEvaluator.NeedsDataRange"/>; a plain <c>Average</c> needs neither). Earlier data is a non-Bad
+///         pre-window row, or — when none arrived — a <see cref="ProbeRows"/>-row read <c>[</c><see cref="ProbeFloor"/><c>,
+///         start)</c>. Later data is a sample exactly at the window end (forward) from the window read, or else a
+///         <see cref="ProbeRows"/>-row read <c>[end, now + </c><see cref="LookAheadHorizon"/><c>]</c> (the historian stores
+///         nothing further in the future). In both small reads a row stamped exactly at the query start (a boundary row) or
+///         at the query end (a possible terminal row) is not a stored sample, whatever its value.
+///     </para>
+///     <para>
+///         <b>Bounded.</b> The in-window rows are capped by <c>ServerHistorian:MaxTimeWeightedRawRows</c> (the same knob the
+///         SQL backends' time-weighted reads use): past it the node fails with
+///         <see cref="HistoryReadLimitExceededException"/> (⇒ <c>BadTooManyOperations</c>) once the stream — bounded by the
+///         requested row budget — has been drained. The gateway is asked for at most <c>cap + 2 + </c>
+///         <see cref="PreWindowAllowance"/> rows; a stream that delivers all of them cannot be proven complete and fails the
+///         same way rather than answering from a truncated window.
+///     </para>
+///     <para>
+///         <b>Values and status.</b> A sample's value is its numeric value (Double), else its string value, else Empty; its
+///         status is <see cref="SampleMapper.ToStatus"/> — the same quality mapping a raw HistoryRead serves (C-002: the
+///         gateway stores no quality, so AVEVA history is normally all Good). An Empty sample is clamped Bad
+///         (<see cref="StoredSample.FromStored"/>), exactly as on the SQL backends.
+///     </para>
+/// </remarks>
+internal static class RawAggregateReader
+{
+    /// <summary>How far before the window the raw stream starts, so the AVEVA initial-value row lands outside it.</summary>
+    internal static readonly TimeSpan BoundaryMargin = TimeSpan.FromMilliseconds(1);
+
+    /// <summary>How far past "now" the stream may look for a sample after the window (end-of-data rules).</summary>
+    internal static readonly TimeSpan LookAheadHorizon = TimeSpan.FromDays(1);
+
+    /// <summary>Rows the stream may spend before the window (the boundary row plus samples inside the margin).</summary>
+    internal const int PreWindowAllowance = 16;
+
+    /// <summary>Start of the "is there data before the window?" probe (the Unix epoch — before any AVEVA history).</summary>
+    internal static readonly DateTime ProbeFloor = DateTime.UnixEpoch;
+
+    /// <summary>Rows the earlier-data probe asks for (a possible boundary row plus the first stored sample).</summary>
+    internal const int ProbeRows = 2;
+
+    /// <summary>True for the aggregates this reader serves: the per-sample families of <see cref="AggregateFamilies"/>.</summary>
+    /// <param name="aggregate">Requested aggregate.</param>
+    /// <returns>Whether the aggregate is evaluated from raw samples.</returns>
+    public static bool Serves(HistoryAggregateType aggregate) =>
+        AggregateFamilies.FamilyOf(aggregate) is AggregateFamily.Simple or AggregateFamily.StartEnd or AggregateFamily.ActualTime;
+
+    /// <summary>The gateway <c>max_values</c> for a configured window cap (<see cref="int.MaxValue"/> when unbounded).</summary>
+    /// <param name="maxWindowRows">The configured cap (&lt;= 0 = unbounded).</param>
+    /// <returns>The row limit to request.</returns>
+    public static int MaxValuesFor(int maxWindowRows) =>
+        maxWindowRows <= 0 ? int.MaxValue : (int)Math.Min(int.MaxValue, (long)maxWindowRows + 2 + PreWindowAllowance);
+
+    /// <summary>Maps one gateway raw sample to a stored sample (value + mapped status, Empty clamped Bad).</summary>
+    /// <param name="sample">The gateway sample; its timestamp must be set.</param>
+    /// <returns>The stored sample.</returns>
+    public static StoredSample ToStored(HistorianSample sample)
+    {
+        var value = sample.HasNumericValue ? HistorianValue.FromDouble(sample.NumericValue)
+            : sample.HasStringValue ? HistorianValue.FromString(sample.StringValue)
+            : HistorianValue.Empty;
+        return StoredSample.FromStored(sample.Timestamp.ToDateTime(), value, SampleMapper.ToStatus(sample));
+    }
+
+    /// <summary>Reads and evaluates one per-sample processed read.</summary>
+    /// <param name="client">The gateway client.</param>
+    /// <param name="tag">Historian tag name.</param>
+    /// <param name="aggregate">A per-sample aggregate (<see cref="Serves"/>).</param>
+    /// <param name="plan">The bucket plan.</param>
+    /// <param name="maxWindowRows">The in-window raw-row cap (&lt;= 0 = unbounded).</param>
+    /// <param name="now">Server timestamp, and the anchor of the look-ahead horizon.</param>
+    /// <param name="ct">Cancellation token.</param>
+    /// <returns>One value per interval, in the plan's order.</returns>
+    /// <exception cref="HistoryReadLimitExceededException">The window holds more than <paramref name="maxWindowRows"/> rows,
+    ///     or the stream could not be read to completion within the requested row budget.</exception>
+    public static async Task<IReadOnlyList<DataValueSnapshot>> ReadAsync(IHistorianGatewayClient client, string tag,
+        HistoryAggregateType aggregate, BucketPlan plan, int maxWindowRows, DateTime now, CancellationToken ct)
+    {
+        if (plan.BucketCount == 0) return [];
+        var needsRange = AggregateEvaluator.NeedsDataRange(aggregate, plan);
+        var requestStart = plan.StartUtc.Ticks > BoundaryMargin.Ticks ? plan.StartUtc - BoundaryMargin : plan.StartUtc;
+        var maxValues = MaxValuesFor(maxWindowRows);
+
+        var buckets = new RawBucketAccumulator(aggregate == HistoryAggregateType.MaximumActualTime);
+        DateTime? earliest = null, latest = null;
+        long windowRows = 0, received = 0;
+        var overCap = false;
+        // Consumed to COMPLETION (never broken out of): abandoning a gateway stream cancels the call server-side, and the
+        // gateway currently treats that cancellation as a connection fault and evicts its shared historian session. Past
+        // the cap the rest is drained (bounded by maxValues) without being evaluated.
+        await foreach (var raw in client.ReadRawAsync(tag, requestStart, plan.EndUtc, maxValues, ct).ConfigureAwait(false))
+        {
+            received++;
+            if (overCap || raw.Timestamp is null) continue;
+            var s = ToStored(raw);
+            var t = s.TimeUtc;
+            if (plan.IndexOf(t) is { } bucket)
+            {
+                if (maxWindowRows > 0 && ++windowRows > maxWindowRows)
+                {
+                    overCap = true;
+                    continue;
+                }
+                buckets.Add(bucket, s);
+            }
+            // The data range as the SQL backends define it: earliest stored sample at or before the window's later edge,
+            // latest at or after its earlier edge. A row before the window (the AVEVA boundary row, or a stored sample in
+            // the margin) proves earlier data only when its OPC quality is not Bad: AVEVA's boundary row for a tag with no
+            // earlier data is a NULL row, which arrives WITH a value (histsdk sets NumericValue = 0 / StringValue "") and
+            // OPC quality 0.
+            if (t >= plan.StartUtc || IsNotBadOnTheWire(raw))
+            {
+                if (t <= plan.EndUtc && (earliest is null || t < earliest)) earliest = t;
+                if (t >= plan.StartUtc && (latest is null || t > latest)) latest = t;
+            }
+        }
+        if (overCap || (maxWindowRows > 0 && received >= maxValues))
+            throw LimitExceeded(tag, maxWindowRows);
+
+        if (needsRange)
+        {
+            // Earlier data: no non-Bad pre-window row ⇒ ask explicitly (normally only when the tag's history really starts
+            // inside the window).
+            var earlierSeen = earliest is { } e && (e < plan.StartUtc || (plan.Reversed && e == plan.StartUtc));
+            if (!earlierSeen && plan.StartUtc.AddTicks(-1) > ProbeFloor)
+                if (await FirstSampleAsync(client, tag, ProbeFloor, plan.StartUtc.AddTicks(-1), ct).ConfigureAwait(false) is { } before)
+                    earliest = before;
+
+            // Later data: the main read covers a sample exactly at the window end (forward: that is already later data);
+            // otherwise a small separate read from the end to now + horizon.
+            var laterSeen = latest is { } l && (plan.Reversed ? l > plan.EndUtc : l >= plan.EndUtc);
+            var horizon = DateTime.MaxValue - now > LookAheadHorizon
+                ? DateTime.SpecifyKind(now + LookAheadHorizon, DateTimeKind.Utc)
+                : DateTime.MaxValue;
+            if (!laterSeen && horizon > plan.EndUtc)
+                if (await FirstSampleAsync(client, tag, plan.EndUtc, horizon, ct).ConfigureAwait(false) is { } after)
+                    latest = after;
+        }
+
+        return AggregateEvaluator.EvaluateAll(aggregate, plan, buckets.Rows(), earliest, latest, now);
+    }
+
+    /// <summary>
+    ///     The time of the first stored sample (any quality) strictly between <paramref name="startUtc"/> and
+    ///     <paramref name="endUtc"/>, read as a <see cref="ProbeRows"/>-row raw read consumed to completion. A row stamped
+    ///     exactly at the query start is the historian's boundary row (the value in effect there, or a NULL row when there is
+    ///     none) and a row exactly at the query end may be a terminal row — neither is a stored sample, whatever its value.
+    /// </summary>
+    private static async Task<DateTime?> FirstSampleAsync(IHistorianGatewayClient client, string tag, DateTime startUtc,
+        DateTime endUtc, CancellationToken ct)
+    {
+        DateTime? found = null;
+        await foreach (var raw in client.ReadRawAsync(tag, startUtc, endUtc, ProbeRows, ct).ConfigureAwait(false))
+        {
+            if (found is not null || raw.Timestamp is null) continue; // drain, never break (see ReadAsync)
+            var t = raw.Timestamp.ToDateTime();
+            if (t > startUtc && t < endUtc) found = t;
+        }
+        return found;
+    }
+
+    /// <summary>
+    ///     The row's OPC DA quality alone (<c>opc_quality</c> ≥ 64: Good or Uncertain) — deliberately without
+    ///     <see cref="SampleMapper.ToStatus"/>'s fallback to the AVEVA <c>quality</c> field when <c>opc_quality</c> is 0, which
+    ///     is also the OPC DA Bad value a NULL row carries.
+    /// </summary>
+    private static bool IsNotBadOnTheWire(HistorianSample raw) => raw.OpcQuality >= 64;
+
+    private static HistoryReadLimitExceededException LimitExceeded(string tag, int maxWindowRows) =>
+        new(maxWindowRows,
+            $"The processed read of '{tag}' covers more than {maxWindowRows} raw samples " +
+            "(ServerHistorian:MaxTimeWeightedRawRows); the Gateway backend evaluates per-sample aggregates from raw reads — " +
+            "narrow the window.");
+}
+
+/// <summary>
+///     Builds the per-bucket statistics (<see cref="BucketRow"/>) the SQL backends' aggregate queries produce, from a stream
+///     of raw samples — the C# statement of <c>SqlAggregateQueries</c>' columns: counts by class (usable = non-Bad with a
+///     value; good = usable with Good severity), Avg/Min/Max/Sum over the usable numeric values, first/last usable value by
+///     time, the extreme usable numeric value with its time (ties ⇒ earliest), and the first/last sample of any quality.
+/// </summary>
+/// <remarks>
+///     Keep in step with the SQL backends' per-sample value columns: a change to which samples feed a statistic there
+///     (e.g. Uncertain handling) must be repeated here, or the Gateway backend silently diverges.
+/// </remarks>
+internal sealed class RawBucketAccumulator
+{
+    private readonly bool _maximum;
+    private readonly Dictionary<long, State> _buckets = new();
+
+    /// <summary>Creates an accumulator.</summary>
+    /// <param name="maximum">Extreme = the maximum (MaximumActualTime); otherwise the minimum.</param>
+    public RawBucketAccumulator(bool maximum) => _maximum = maximum;
+
+    private sealed class State
+    {
+        public int Total, Usable, Good, NonNumeric, Numeric;
+        public double Sum, Min = double.PositiveInfinity, Max = double.NegativeInfinity;
+        public StoredSample? First, Last, FirstAny, LastAny, Extreme;
+    }
+
+    /// <summary>Adds one in-window sample to its bucket. Order-independent (picks compare by time).</summary>
+    /// <param name="bucket">Ascending bucket index (<see cref="BucketPlan.IndexOf"/>).</param>
+    /// <param name="s">The sample.</param>
+    public void Add(long bucket, StoredSample s)
+    {
+        if (!_buckets.TryGetValue(bucket, out var st)) _buckets[bucket] = st = new State();
+        st.Total++;
+        if (st.FirstAny is not { } fa || s.TimeUtc < fa.TimeUtc) st.FirstAny = s;
+        if (st.LastAny is not { } la || s.TimeUtc >= la.TimeUtc) st.LastAny = s;
+        if (!s.IsUsable) return;
+        st.Usable++;
+        if (s.IsGood) st.Good++;
+        if (st.First is not { } f || s.TimeUtc < f.TimeUtc) st.First = s;
+        if (st.Last is not { } l || s.TimeUtc >= l.TimeUtc) st.Last = s;
+        if (!s.Value.TryGetDouble(out var v))
+        {
+            st.NonNumeric++;
+            return;
+        }
+        st.Numeric++;
+        st.Sum += v;
+        if (v < st.Min) st.Min = v;
+        if (v > st.Max) st.Max = v;
+        if (st.Extreme is not { } e) st.Extreme = s;
+        else
+        {
+            var ev = e.Value.Number;
+            var better = _maximum ? v > ev : v < ev;
+            if (better || (v == ev && s.TimeUtc < e.TimeUtc)) st.Extreme = s;
+        }
+    }
+
+    /// <summary>The non-empty buckets' rows, ascending by bucket index.</summary>
+    /// <returns>The rows.</returns>
+    public IEnumerable<BucketRow> Rows() =>
+        _buckets.OrderBy(kv => kv.Key).Select(kv =>
+        {
+            var st = kv.Value;
+            var hasNumeric = st.Numeric > 0;
+            return new BucketRow
+            {
+                Bucket = kv.Key,
+                Total = st.Total,
+                Usable = st.Usable,
+                Good = st.Good,
+                Bad = st.Total - st.Usable,
+                NonNumeric = st.NonNumeric,
+                Avg = hasNumeric ? st.Sum / st.Numeric : null,
+                Min = hasNumeric ? st.Min : null,
+                Max = hasNumeric ? st.Max : null,
+                Sum = hasNumeric ? st.Sum : null,
+                First = st.First?.Value ?? HistorianValue.Empty,
+                Last = st.Last?.Value ?? HistorianValue.Empty,
+                Extreme = st.Extreme?.Value.Number,
+                ExtremeTime = st.Extreme?.TimeUtc,
+                FirstAny = st.FirstAny,
+                LastAny = st.LastAny,
+            };
+        });
+}
@@ -11,6 +11,11 @@
    <ItemGroup>
        <ProjectReference Include="..\..\Core\ZB.MOM.WW.OtOpcUa.Core.Abstractions\ZB.MOM.WW.OtOpcUa.Core.Abstractions.csproj"/>
        <ProjectReference Include="..\..\Core\ZB.MOM.WW.OtOpcUa.Core.AlarmHistorian\ZB.MOM.WW.OtOpcUa.Core.AlarmHistorian.csproj"/>
+        <!-- #681: the per-sample processed aggregates are evaluated from raw reads with the SAME database-independent code the
+             SQL backends use (BucketPlan, BucketRow, AggregateEvaluator), so Gateway matches SqlServer/QuestDB exactly.
+             Sql.Common references only Core.Abstractions + Core.AlarmHistorian (both already referenced here) and no
+             driver, so this adds no cycle. -->
+        <ProjectReference Include="..\ZB.MOM.WW.OtOpcUa.Driver.Historian.Sql.Common\ZB.MOM.WW.OtOpcUa.Driver.Historian.Sql.Common.csproj"/>
        <!-- Runtime owns ServerHistorianOptions (the bound ServerHistorian config). The read-cutover
             factory (GatewayHistorian.CreateDataSource / HistorianGatewayClientAdapter.Create) maps
             those options onto the package client, so the driver references it. Runtime references no