fix(site-call-audit): composite keyset cursor eliminates the single-timestamp reconciliation pin; pinned state now a published event

Reconciliation cursor becomes composite (UpdatedAtUtc, TrackedOperationId);
IPullSiteCallsClient/GrpcPullSiteCallsClient forward the Task-15 after_id keyset
(additive param, null preserves the legacy inclusive >= contract). A burst
sharing one exact UpdatedAtUtc now drains via the id tiebreak instead of pinning
forever. A legacy site that ignores after_id is latched + published as
SiteCallReconciliationPinnedChanged on the EventStream (transition-only),
replacing the prior silent log line.

src/ZB.MOM.WW.ScadaBridge.AuditLog/Central/GrpcPullSiteCallsClient.cs

@@ -80,6 +80,7 @@ public sealed class GrpcPullSiteCallsClient : IPullSiteCallsClient
     public async Task<PullSiteCallsResponse> PullAsync(
         string siteId,
         DateTime sinceUtc,
+        string? afterId,
         int batchSize,
         CancellationToken ct)
     {
@@ -100,6 +101,10 @@ public sealed class GrpcPullSiteCallsClient : IPullSiteCallsClient
             // EnsureUtc keeps Timestamp.FromDateTime happy (it requires UTC kind).
             SinceUtc = Timestamp.FromDateTime(EnsureUtc(sinceUtc)),
             BatchSize = batchSize,
+            // Composite-keyset tiebreak (Task 16). proto3 has no nullable string —
+            // an unset/empty AfterId is the site's signal to keep the legacy
+            // inclusive-timestamp contract (also what a first pull sends).
+            AfterId = afterId ?? string.Empty,
         };
 
         ProtoPullResponse reply;

src/ZB.MOM.WW.ScadaBridge.AuditLog/Central/IPullSiteCallsClient.cs

@@ -46,12 +46,22 @@ public interface IPullSiteCallsClient
     /// </summary>
     /// <param name="siteId">The identifier of the site to pull cached-call operational rows from.</param>
     /// <param name="sinceUtc">Only rows with an <c>UpdatedAtUtc</c> at or after this cursor time are returned.</param>
+    /// <param name="afterId">
+    /// The composite-keyset tiebreak cursor (Task 16). When non-null it is the
+    /// <c>TrackedOperationId</c> of the last row already consumed at
+    /// <paramref name="sinceUtc"/>; the site returns only rows strictly greater
+    /// than the composite <c>(UpdatedAtUtc, TrackedOperationId)</c> pair, so a
+    /// burst sharing one exact <c>UpdatedAtUtc</c> drains via the id tiebreak
+    /// instead of pinning the inclusive-timestamp cursor. Null on the first pull
+    /// (or against a legacy site) preserves the inclusive <c>&gt;=</c> contract.
+    /// </param>
     /// <param name="batchSize">Maximum number of rows to return per call.</param>
     /// <param name="ct">Cancellation token.</param>
     /// <returns>A task that resolves to the next reconciliation batch with a <c>MoreAvailable</c> flag.</returns>
     Task<PullSiteCallsResponse> PullAsync(
         string siteId,
         DateTime sinceUtc,
+        string? afterId,
         int batchSize,
         CancellationToken ct);
 }

src/ZB.MOM.WW.ScadaBridge.SiteCallAudit/SiteCallAuditActor.cs

@@ -131,17 +131,32 @@ public class SiteCallAuditActor : ReceiveActor
     private readonly bool _backgroundTimersEnabled;
 
     /// <summary>
-    /// Per-site reconciliation watermark — the highest
-    /// <see cref="SiteCall.UpdatedAtUtc"/> seen for that site on a previous
-    /// tick. The next tick asks for rows at or after this cursor; idempotent
-    /// monotonic <see cref="ISiteCallAuditRepository.UpsertAsync"/> swallows any
-    /// duplicate-with-same-timestamp rows. In-memory for the singleton's
-    /// lifetime — a failover / restart resets every cursor to
-    /// <see cref="DateTime.MinValue"/>, which is conservative but correct
-    /// (the next tick re-pulls and idempotent upsert dedupes). Mirrors
-    /// <c>SiteAuditReconciliationActor</c>.
+    /// Per-site reconciliation watermark — the composite
+    /// <c>(UpdatedAtUtc, TrackedOperationId)</c> keyset of the highest row seen
+    /// for that site on a previous tick (Task 16). The next tick asks for rows
+    /// strictly greater than this pair, so a burst sharing one exact
+    /// <see cref="SiteCall.UpdatedAtUtc"/> drains via the id tiebreak rather than
+    /// pinning the timestamp forever; idempotent monotonic
+    /// <see cref="ISiteCallAuditRepository.UpsertAsync"/> still swallows any
+    /// boundary duplicate. In-memory for the singleton's lifetime — a
+    /// failover / restart resets every cursor to
+    /// <c>(DateTime.MinValue, null)</c>, which is conservative but correct (the
+    /// next tick re-pulls from the start and idempotent upsert dedupes). Mirrors
+    /// <c>SiteAuditReconciliationActor</c>'s cursor (now composite).
     /// </summary>
-    private readonly Dictionary<string, DateTime> _reconciliationCursors = new();
+    private readonly Dictionary<string, (DateTime Since, string? AfterId)> _reconciliationCursors = new();
+
+    /// <summary>
+    /// Per-site "pinned" latch (Task 16) — <c>true</c> once a site's composite
+    /// cursor stopped advancing while the site still reported
+    /// <c>MoreAvailable=true</c> (in practice only a legacy site that ignores the
+    /// <c>after_id</c> keyset field). Tracks the latch so only <em>transitions</em>
+    /// publish <see cref="SiteCallReconciliationPinnedChanged"/> on the EventStream
+    /// (pinned on latch, unpinned when progress resumes), mirroring the sibling
+    /// <c>SiteAuditReconciliationActor</c>'s stalled-state machine — no flood of
+    /// redundant notifications.
+    /// </summary>
+    private readonly Dictionary<string, bool> _reconciliationPinned = new();
 
     private ICancelable? _reconciliationTimer;
     private ICancelable? _purgeTimer;
@@ -581,52 +596,56 @@ public class SiteCallAuditActor : ReceiveActor
     /// not the source of truth, so a slow site simply lags rather than corrupts.
     /// </para>
     /// <para>
-    /// <b>Inclusive cursor boundary.</b> The cursor is advanced to the maximum
-    /// <see cref="SiteCall.UpdatedAtUtc"/> seen, and the pull asks for rows at or
-    /// after it (<c>since</c> is <c>&gt;=</c>, not <c>&gt;</c>). The row whose
-    /// timestamp equals the cursor is therefore re-pulled on the next tick and
-    /// deduplicated by the idempotent monotonic upsert — the same inclusive-boundary
-    /// contract as <c>SiteAuditReconciliationActor</c>'s cursor.
+    /// <b>Composite keyset cursor (Task 16).</b> The cursor is the
+    /// <c>(UpdatedAtUtc, TrackedOperationId)</c> pair of the maximum row seen, and
+    /// the pull asks for rows strictly greater than it. The boundary row is NOT
+    /// re-pulled (unlike the prior inclusive-timestamp cursor), and — critically —
+    /// a burst of more than <see cref="SiteCallAuditOptions.ReconciliationBatchSize"/>
+    /// rows sharing one exact <see cref="SiteCall.UpdatedAtUtc"/> now drains via the
+    /// id tiebreak instead of pinning the timestamp forever. The idempotent monotonic
+    /// upsert still dedupes any overlap. Matches the composite keyset the site's
+    /// <c>ReadChangedSinceAsync</c> honours (Task 15).
     /// </para>
     /// <para>
     /// <b>Consumes <see cref="PullSiteCallsResponse.MoreAvailable"/>
-    /// to guarantee forward progress.</b> Whereas the prior implementation ignored
-    /// the flag entirely and relied solely on the tick cadence, this method now
-    /// continues pulling within the same tick while the site reports
-    /// <c>MoreAvailable=true</c>, bounded by
-    /// <see cref="MaxReconciliationPagesPerTick"/>. This closes the
-    /// single-timestamp-saturation edge: if a backlog larger than
-    /// <see cref="SiteCallAuditOptions.ReconciliationBatchSize"/> all shares one
-    /// exact <see cref="SiteCall.UpdatedAtUtc"/>, the inclusive max-timestamp cursor
-    /// cannot advance, so the previous code re-pulled the identical window forever
-    /// across ticks and never drained the tail. Here, a saturated batch whose
-    /// observed max timestamp did NOT advance past <c>since</c> is detected as a
-    /// no-progress pin: the loop stops and logs a Warning (the same observability
-    /// intent as the sibling's stalled signal, without its EventStream state
-    /// machine), so the pathological site surfaces rather than spinning silently.
-    /// This diverges from <c>SiteAuditReconciliationActor</c>, which reads
-    /// <c>MoreAvailable</c> to drive a <c>SiteAuditTelemetryStalledChanged</c>
-    /// stalled-detection state machine instead of a within-tick continuation drain.
+    /// to guarantee forward progress.</b> The method continues pulling within the
+    /// same tick while the site reports <c>MoreAvailable=true</c>, bounded by
+    /// <see cref="MaxReconciliationPagesPerTick"/> so a misbehaving site can never
+    /// spin the dispatcher. Each page advances the in-flight composite cursor. If a
+    /// page fails to advance the cursor yet the site still reports more available,
+    /// the site is NOT honouring the <c>after_id</c> keyset (a legacy pre-Task-15
+    /// site): continuing would re-pull the identical window forever, so the loop
+    /// latches the site as <em>pinned</em> and publishes
+    /// <see cref="SiteCallReconciliationPinnedChanged"/> on the EventStream — the
+    /// dead-end becomes a health-observable condition rather than a silent log line.
+    /// When a later tick makes progress the latch clears with <c>Pinned=false</c>.
+    /// This aligns with <c>SiteAuditReconciliationActor</c>'s
+    /// <c>SiteAuditTelemetryStalledChanged</c> transition-only publication.
     /// </para>
     /// </remarks>
     private async Task ReconcileSiteAsync(
         SiteEntry site, IPullSiteCallsClient client, ISiteCallAuditRepository repository)
     {
-        var cursor = _reconciliationCursors.TryGetValue(site.SiteId, out var c) ? c : DateTime.MinValue;
+        var cursor = _reconciliationCursors.TryGetValue(site.SiteId, out var c)
+            ? c
+            : (Since: DateTime.MinValue, AfterId: (string?)null);
 
         // Drain within the tick while the site keeps reporting
         // MoreAvailable, bounded by MaxReconciliationPagesPerTick so a misbehaving
         // site can never spin the dispatcher. Each page advances the in-flight
-        // cursor; a saturated page that fails to advance the cursor is the
-        // single-timestamp no-progress pin — break and surface it.
+        // composite cursor; a page that fails to advance it while MoreAvailable is
+        // still true means the site ignores after_id (legacy) — latch + publish
+        // the pinned state and break.
         for (var page = 0; page < MaxReconciliationPagesPerTick; page++)
         {
-            var since = cursor;
+            var since = cursor.Since;
+            var afterId = cursor.AfterId;
             var response = await client
-                .PullAsync(site.SiteId, since, _options.ReconciliationBatchSize, CancellationToken.None)
+                .PullAsync(site.SiteId, since, afterId, _options.ReconciliationBatchSize, CancellationToken.None)
                 .ConfigureAwait(false);
 
             var maxUpdated = since;
+            var maxAfterId = afterId;
             var nowUtc = DateTime.UtcNow;
             foreach (var row in response.SiteCalls)
             {
@@ -637,40 +656,57 @@ public class SiteCallAuditActor : ReceiveActor
                 var siteCall = row with { IngestedAtUtc = nowUtc };
                 await repository.UpsertAsync(siteCall).ConfigureAwait(false);
 
-                if (row.UpdatedAtUtc > maxUpdated)
+                // Advance the composite max: greater timestamp wins; on a tie the
+                // greater TrackedOperationId (ordinal, matching the site's SQLite
+                // BINARY text collation) wins.
+                var rowId = row.TrackedOperationId.ToString();
+                if (row.UpdatedAtUtc > maxUpdated ||
+                    (row.UpdatedAtUtc == maxUpdated && string.CompareOrdinal(rowId, maxAfterId) > 0))
                 {
                     maxUpdated = row.UpdatedAtUtc;
+                    maxAfterId = rowId;
                 }
             }
 
+            // Did the composite cursor move? (timestamp advanced OR — at the same
+            // timestamp — the id tiebreak advanced). CompareOrdinal handles nulls
+            // (CompareOrdinal(x, null) > 0 for any non-null x).
+            var advanced = maxUpdated > since || string.CompareOrdinal(maxAfterId, afterId) != 0;
+
             // Persist the advanced cursor after every page so a fault on a later
             // page (caught per-site upstream) still keeps the rows already drained.
-            cursor = maxUpdated;
+            cursor = (maxUpdated, maxAfterId);
             _reconciliationCursors[site.SiteId] = cursor;
 
+            // Progress resumed → clear any prior pinned latch (publishes the
+            // Pinned=false transition exactly once).
+            if (advanced)
+            {
+                SetReconciliationPinned(site.SiteId, pinned: false);
+            }
+
             if (!response.MoreAvailable)
             {
                 // Backlog fully drained for this site this tick.
                 return;
             }
 
-            if (maxUpdated <= since)
+            if (!advanced)
             {
-                // No-progress pin: the site saturated the batch yet the max
-                // observed UpdatedAtUtc did not advance past the inclusive cursor
-                // (a burst of > batch-size rows sharing one exact timestamp).
-                // Continuing would re-pull the identical window forever, so stop
-                // and surface it — the inclusive max-timestamp cursor cannot make
-                // progress on this input without a composite (timestamp,id)
-                // keyset, which the pull contract does not yet support.
+                // The composite cursor did not advance yet the site still reports
+                // MoreAvailable — the site ignored after_id (a legacy pre-keyset
+                // site), so a continuation would re-pull the identical window
+                // forever. Latch + publish the pinned state and stop; the next tick
+                // re-checks and unpins if the site catches up.
+                SetReconciliationPinned(site.SiteId, pinned: true);
                 _logger.LogWarning(
-                    "SiteCallAudit reconciliation for site {SiteId} cannot make progress: a saturated "
-                    + "batch of more than {BatchSize} rows shares a single UpdatedAtUtc ({CursorUtc:o}), "
-                    + "so the inclusive cursor is pinned. The backlog tail beyond the batch ceiling will "
-                    + "not reconcile until those rows' timestamps differ.",
+                    "SiteCallAudit reconciliation for site {SiteId} is pinned: the composite "
+                    + "(UpdatedAtUtc, TrackedOperationId) cursor did not advance past ({CursorUtc:o}, {AfterId}) "
+                    + "yet the site reports more rows available — the site does not honour the after_id keyset "
+                    + "(legacy). The backlog tail will not reconcile until the site is upgraded.",
                     site.SiteId,
-                    _options.ReconciliationBatchSize,
-                    since);
+                    since,
+                    afterId ?? "(none)");
                 return;
             }
         }
@@ -687,6 +723,25 @@ public class SiteCallAuditActor : ReceiveActor
             MaxReconciliationPagesPerTick);
     }
 
+    /// <summary>
+    /// Latches the per-site reconciliation pinned state and publishes
+    /// <see cref="SiteCallReconciliationPinnedChanged"/> on the actor system
+    /// EventStream ONLY on a transition — mirroring
+    /// <c>SiteAuditReconciliationActor.UpdateStalledState</c> so a downstream
+    /// health subscriber never sees a flood of same-state notifications.
+    /// </summary>
+    private void SetReconciliationPinned(string siteId, bool pinned)
+    {
+        var was = _reconciliationPinned.TryGetValue(siteId, out var prior) && prior;
+        if (was == pinned)
+        {
+            return;
+        }
+
+        _reconciliationPinned[siteId] = pinned;
+        Context.System.EventStream.Publish(new SiteCallReconciliationPinnedChanged(siteId, pinned));
+    }
+
     // ── Piece B: daily terminal-row purge scheduler ──
 
     /// <summary>

src/ZB.MOM.WW.ScadaBridge.SiteCallAudit/SiteCallReconciliationPinnedChanged.cs

@@ -0,0 +1,19 @@
+namespace ZB.MOM.WW.ScadaBridge.SiteCallAudit;
+
+/// <summary>
+/// Published on the actor system EventStream when a site's Site Call Audit
+/// reconciliation puller transitions into or out of the "pinned" state — the
+/// composite <c>(UpdatedAtUtc, TrackedOperationId)</c> keyset cursor could not
+/// advance yet the site still reports <c>MoreAvailable=true</c>, so the backlog
+/// tail cannot be drained. In practice this only happens against a legacy site
+/// that ignores the <c>after_id</c> keyset field (a modern site drains a
+/// single-timestamp burst via the id tiebreak). Mirrors
+/// <c>SiteAuditTelemetryStalledChanged</c>: only transitions are published
+/// (<c>Pinned=true</c> on latch, <c>Pinned=false</c> when progress resumes), so
+/// a downstream health subscriber never sees a flood of redundant notifications.
+/// The dead-end that the previous implementation only logged is now a
+/// health-observable condition.
+/// </summary>
+/// <param name="SiteId">The site whose reconciliation cursor pinned (or unpinned).</param>
+/// <param name="Pinned">True on entering the pinned state; false when the cursor makes progress again.</param>
+public sealed record SiteCallReconciliationPinnedChanged(string SiteId, bool Pinned);