feat(site-call-audit): additive (UpdatedAtUtc, TrackedOperationId) keyset in the reconciliation pull contract

Site side: additive proto after_id field, ReadChangedSinceAsync gains an optional
afterId cursor and deterministic (UpdatedAtUtc, TrackedOperationId) ordering — a
batch fully inside one UpdatedAtUtc instant no longer re-reads the same page
forever. Absent/empty afterId preserves the exact legacy inclusive >= contract, so
an older central is unaffected. Regenerated the checked-in gRPC C#.

src/ZB.MOM.WW.ScadaBridge.Commons/Interfaces/Services/IOperationTrackingStore.cs

@@ -137,6 +137,19 @@ public interface IOperationTrackingStore
     /// the caller advance the cursor monotonically across follow-up pulls.
     /// </para>
     /// <para>
+    /// <b>Composite keyset (Task 15).</b> When <paramref name="afterId"/> is
+    /// supplied the read uses a <c>(UpdatedAtUtc, TrackedOperationId)</c> keyset:
+    /// it returns rows strictly after that cursor — <c>UpdatedAtUtc &gt; sinceUtc</c>,
+    /// or <c>UpdatedAtUtc = sinceUtc</c> with a <c>TrackedOperationId</c> greater
+    /// than <paramref name="afterId"/>. This un-pins a batch that would otherwise
+    /// stall: when more than <paramref name="batchSize"/> rows share one
+    /// <c>UpdatedAtUtc</c>, the plain inclusive <c>&gt;=</c> resume re-reads the
+    /// same page forever. When <paramref name="afterId"/> is <c>null</c>/empty the
+    /// legacy inclusive <c>&gt;=</c> contract is preserved exactly (an older
+    /// central that does not send the cursor is unaffected). Ordering is always
+    /// deterministic — <c>UpdatedAtUtc ASC, TrackedOperationId ASC</c>.
+    /// </para>
+    /// <para>
     /// <see cref="SiteCallOperational.SourceSite"/> is left as the empty string:
     /// the site id is not a tracking-store column, and the central client re-stamps
     /// it from the <c>siteId</c> it dialed (the only authority that knows which
@@ -148,10 +161,17 @@ public interface IOperationTrackingStore
     /// </remarks>
     /// <param name="sinceUtc">Inclusive lower bound on <c>UpdatedAtUtc</c>; <see cref="DateTime.MinValue"/> reads from the start.</param>
     /// <param name="batchSize">Maximum number of rows to return (oldest first).</param>
+    /// <param name="afterId">
+    /// Optional composite-keyset cursor — the <c>TrackedOperationId</c> of the last
+    /// row already consumed at <paramref name="sinceUtc"/>. When supplied the read
+    /// returns only rows strictly after <c>(sinceUtc, afterId)</c>; when
+    /// <c>null</c>/empty the legacy inclusive <c>&gt;= sinceUtc</c> behaviour applies.
+    /// </param>
     /// <param name="ct">Cancellation token.</param>
     /// <returns>The matching rows projected to <see cref="SiteCallOperational"/>, oldest-first, capped at <paramref name="batchSize"/>.</returns>
     Task<IReadOnlyList<SiteCallOperational>> ReadChangedSinceAsync(
         DateTime sinceUtc,
         int batchSize,
+        string? afterId = null,
         CancellationToken ct = default);
 }

src/ZB.MOM.WW.ScadaBridge.Communication/Grpc/SiteStreamGrpcServer.cs

@@ -561,19 +561,24 @@ public class SiteStreamGrpcServer : SiteStreamService.SiteStreamServiceBase
             ? DateTime.SpecifyKind(request.SinceUtc.ToDateTime(), DateTimeKind.Utc)
             : DateTime.MinValue;
 
+        // Composite-keyset cursor (Task 15): proto3 defaults an unset string to
+        // "", which the store treats as "no cursor" (legacy inclusive >= behaviour)
+        // — so an older central that never sets after_id is unaffected.
+        var afterId = string.IsNullOrEmpty(request.AfterId) ? null : request.AfterId;
+
         IReadOnlyList<SiteCallOperational> operationals;
         try
         {
             operationals = await store.ReadChangedSinceAsync(
-                since, request.BatchSize, context.CancellationToken);
+                since, request.BatchSize, afterId, context.CancellationToken);
         }
         catch (Exception ex)
         {
             // Best-effort, like PullAuditEvents: a read fault must never abort
             // the reconciliation tick — central retries on its next cycle.
             _logger.LogError(ex,
-                "ReadChangedSinceAsync failed for since={Since} batch={Batch}; returning empty response.",
+                "ReadChangedSinceAsync failed for since={Since} batch={Batch} afterId={AfterId}; returning empty response.",
-                since, request.BatchSize);
+                since, request.BatchSize, afterId);
             return new PullSiteCallsResponse();
         }
 

src/ZB.MOM.WW.ScadaBridge.Communication/Protos/sitestream.proto

@@ -171,6 +171,14 @@ message PullAuditEventsResponse {
 message PullSiteCallsRequest {
   google.protobuf.Timestamp since_utc = 1;
   int32 batch_size = 2;
+  // Composite-keyset cursor (Task 15): the TrackedOperationId ("D" GUID form) of
+  // the last row already consumed at since_utc. When set, the site returns only
+  // rows strictly after (since_utc, after_id) under a deterministic
+  // (UpdatedAtUtc, TrackedOperationId) order — un-pinning a batch that would
+  // otherwise stall when more than batch_size rows share one since_utc instant.
+  // Empty (the proto3 string default) preserves the legacy inclusive >= behaviour,
+  // so an older central that never sets it is unaffected. Additive-only.
+  string after_id = 3;
 }
 
 message PullSiteCallsResponse {

src/ZB.MOM.WW.ScadaBridge.Communication/SiteStreamGrpc/Sitestream.cs

@@ -83,29 +83,30 @@ namespace ZB.MOM.WW.ScadaBridge.Communication.Grpc {
             "GAEgASgLMhouZ29vZ2xlLnByb3RvYnVmLlRpbWVzdGFtcBISCgpiYXRjaF9z",
             "aXplGAIgASgFIlwKF1B1bGxBdWRpdEV2ZW50c1Jlc3BvbnNlEikKBmV2ZW50",
             "cxgBIAMoCzIZLnNpdGVzdHJlYW0uQXVkaXRFdmVudER0bxIWCg5tb3JlX2F2",
-            "YWlsYWJsZRgCIAEoCCJZChRQdWxsU2l0ZUNhbGxzUmVxdWVzdBItCglzaW5j",
+            "YWlsYWJsZRgCIAEoCCJrChRQdWxsU2l0ZUNhbGxzUmVxdWVzdBItCglzaW5j",
             "ZV91dGMYASABKAsyGi5nb29nbGUucHJvdG9idWYuVGltZXN0YW1wEhIKCmJh",
-            "dGNoX3NpemUYAiABKAUiaQoVUHVsbFNpdGVDYWxsc1Jlc3BvbnNlEjgKDG9w",
+            "dGNoX3NpemUYAiABKAUSEAoIYWZ0ZXJfaWQYAyABKAkiaQoVUHVsbFNpdGVD",
-            "ZXJhdGlvbmFscxgBIAMoCzIiLnNpdGVzdHJlYW0uU2l0ZUNhbGxPcGVyYXRp",
+            "YWxsc1Jlc3BvbnNlEjgKDG9wZXJhdGlvbmFscxgBIAMoCzIiLnNpdGVzdHJl",
-            "b25hbER0bxIWCg5tb3JlX2F2YWlsYWJsZRgCIAEoCCpcCgdRdWFsaXR5EhcK",
+            "YW0uU2l0ZUNhbGxPcGVyYXRpb25hbER0bxIWCg5tb3JlX2F2YWlsYWJsZRgC",
-            "E1FVQUxJVFlfVU5TUEVDSUZJRUQQABIQCgxRVUFMSVRZX0dPT0QQARIVChFR",
+            "IAEoCCpcCgdRdWFsaXR5EhcKE1FVQUxJVFlfVU5TUEVDSUZJRUQQABIQCgxR",
-            "VUFMSVRZX1VOQ0VSVEFJThACEg8KC1FVQUxJVFlfQkFEEAMqXQoOQWxhcm1T",
+            "VUFMSVRZX0dPT0QQARIVChFRVUFMSVRZX1VOQ0VSVEFJThACEg8KC1FVQUxJ",
-            "dGF0ZUVudW0SGwoXQUxBUk1fU1RBVEVfVU5TUEVDSUZJRUQQABIWChJBTEFS",
+            "VFlfQkFEEAMqXQoOQWxhcm1TdGF0ZUVudW0SGwoXQUxBUk1fU1RBVEVfVU5T",
-            "TV9TVEFURV9OT1JNQUwQARIWChJBTEFSTV9TVEFURV9BQ1RJVkUQAiqFAQoO",
+            "UEVDSUZJRUQQABIWChJBTEFSTV9TVEFURV9OT1JNQUwQARIWChJBTEFSTV9T",
-            "QWxhcm1MZXZlbEVudW0SFAoQQUxBUk1fTEVWRUxfTk9ORRAAEhMKD0FMQVJN",
+            "VEFURV9BQ1RJVkUQAiqFAQoOQWxhcm1MZXZlbEVudW0SFAoQQUxBUk1fTEVW",
-            "X0xFVkVMX0xPVxABEhcKE0FMQVJNX0xFVkVMX0xPV19MT1cQAhIUChBBTEFS",
+            "RUxfTk9ORRAAEhMKD0FMQVJNX0xFVkVMX0xPVxABEhcKE0FMQVJNX0xFVkVM",
-            "TV9MRVZFTF9ISUdIEAMSGQoVQUxBUk1fTEVWRUxfSElHSF9ISUdIEAQytwMK",
+            "X0xPV19MT1cQAhIUChBBTEFSTV9MRVZFTF9ISUdIEAMSGQoVQUxBUk1fTEVW",
-            "EVNpdGVTdHJlYW1TZXJ2aWNlElUKEVN1YnNjcmliZUluc3RhbmNlEiEuc2l0",
+            "RUxfSElHSF9ISUdIEAQytwMKEVNpdGVTdHJlYW1TZXJ2aWNlElUKEVN1YnNj",
-            "ZXN0cmVhbS5JbnN0YW5jZVN0cmVhbVJlcXVlc3QaGy5zaXRlc3RyZWFtLlNp",
+            "cmliZUluc3RhbmNlEiEuc2l0ZXN0cmVhbS5JbnN0YW5jZVN0cmVhbVJlcXVl",
-            "dGVTdHJlYW1FdmVudDABEkcKEUluZ2VzdEF1ZGl0RXZlbnRzEhsuc2l0ZXN0",
+            "c3QaGy5zaXRlc3RyZWFtLlNpdGVTdHJlYW1FdmVudDABEkcKEUluZ2VzdEF1",
-            "cmVhbS5BdWRpdEV2ZW50QmF0Y2gaFS5zaXRlc3RyZWFtLkluZ2VzdEFjaxJQ",
+            "ZGl0RXZlbnRzEhsuc2l0ZXN0cmVhbS5BdWRpdEV2ZW50QmF0Y2gaFS5zaXRl",
-            "ChVJbmdlc3RDYWNoZWRUZWxlbWV0cnkSIC5zaXRlc3RyZWFtLkNhY2hlZFRl",
+            "c3RyZWFtLkluZ2VzdEFjaxJQChVJbmdlc3RDYWNoZWRUZWxlbWV0cnkSIC5z",
-            "bGVtZXRyeUJhdGNoGhUuc2l0ZXN0cmVhbS5Jbmdlc3RBY2sSWgoPUHVsbEF1",
+            "aXRlc3RyZWFtLkNhY2hlZFRlbGVtZXRyeUJhdGNoGhUuc2l0ZXN0cmVhbS5J",
-            "ZGl0RXZlbnRzEiIuc2l0ZXN0cmVhbS5QdWxsQXVkaXRFdmVudHNSZXF1ZXN0",
+            "bmdlc3RBY2sSWgoPUHVsbEF1ZGl0RXZlbnRzEiIuc2l0ZXN0cmVhbS5QdWxs",
-            "GiMuc2l0ZXN0cmVhbS5QdWxsQXVkaXRFdmVudHNSZXNwb25zZRJUCg1QdWxs",
+            "QXVkaXRFdmVudHNSZXF1ZXN0GiMuc2l0ZXN0cmVhbS5QdWxsQXVkaXRFdmVu",
-            "U2l0ZUNhbGxzEiAuc2l0ZXN0cmVhbS5QdWxsU2l0ZUNhbGxzUmVxdWVzdBoh",
+            "dHNSZXNwb25zZRJUCg1QdWxsU2l0ZUNhbGxzEiAuc2l0ZXN0cmVhbS5QdWxs",
-            "LnNpdGVzdHJlYW0uUHVsbFNpdGVDYWxsc1Jlc3BvbnNlQiuqAihaQi5NT00u",
+            "U2l0ZUNhbGxzUmVxdWVzdBohLnNpdGVzdHJlYW0uUHVsbFNpdGVDYWxsc1Jl",
-            "V1cuU2NhZGFCcmlkZ2UuQ29tbXVuaWNhdGlvbi5HcnBjYgZwcm90bzM="));
+            "c3BvbnNlQiuqAihaQi5NT00uV1cuU2NhZGFCcmlkZ2UuQ29tbXVuaWNhdGlv",
+            "bi5HcnBjYgZwcm90bzM="));
       descriptor = pbr::FileDescriptor.FromGeneratedCode(descriptorData,
           new pbr::FileDescriptor[] { global::Google.Protobuf.WellKnownTypes.TimestampReflection.Descriptor, global::Google.Protobuf.WellKnownTypes.WrappersReflection.Descriptor, },
           new pbr::GeneratedClrTypeInfo(new[] {typeof(global::ZB.MOM.WW.ScadaBridge.Communication.Grpc.Quality), typeof(global::ZB.MOM.WW.ScadaBridge.Communication.Grpc.AlarmStateEnum), typeof(global::ZB.MOM.WW.ScadaBridge.Communication.Grpc.AlarmLevelEnum), }, null, new pbr::GeneratedClrTypeInfo[] {
@@ -121,7 +122,7 @@ namespace ZB.MOM.WW.ScadaBridge.Communication.Grpc {
             new pbr::GeneratedClrTypeInfo(typeof(global::ZB.MOM.WW.ScadaBridge.Communication.Grpc.CachedTelemetryBatch), global::ZB.MOM.WW.ScadaBridge.Communication.Grpc.CachedTelemetryBatch.Parser, new[]{ "Packets" }, null, null, null, null),
             new pbr::GeneratedClrTypeInfo(typeof(global::ZB.MOM.WW.ScadaBridge.Communication.Grpc.PullAuditEventsRequest), global::ZB.MOM.WW.ScadaBridge.Communication.Grpc.PullAuditEventsRequest.Parser, new[]{ "SinceUtc", "BatchSize" }, null, null, null, null),
             new pbr::GeneratedClrTypeInfo(typeof(global::ZB.MOM.WW.ScadaBridge.Communication.Grpc.PullAuditEventsResponse), global::ZB.MOM.WW.ScadaBridge.Communication.Grpc.PullAuditEventsResponse.Parser, new[]{ "Events", "MoreAvailable" }, null, null, null, null),
-            new pbr::GeneratedClrTypeInfo(typeof(global::ZB.MOM.WW.ScadaBridge.Communication.Grpc.PullSiteCallsRequest), global::ZB.MOM.WW.ScadaBridge.Communication.Grpc.PullSiteCallsRequest.Parser, new[]{ "SinceUtc", "BatchSize" }, null, null, null, null),
+            new pbr::GeneratedClrTypeInfo(typeof(global::ZB.MOM.WW.ScadaBridge.Communication.Grpc.PullSiteCallsRequest), global::ZB.MOM.WW.ScadaBridge.Communication.Grpc.PullSiteCallsRequest.Parser, new[]{ "SinceUtc", "BatchSize", "AfterId" }, null, null, null, null),
             new pbr::GeneratedClrTypeInfo(typeof(global::ZB.MOM.WW.ScadaBridge.Communication.Grpc.PullSiteCallsResponse), global::ZB.MOM.WW.ScadaBridge.Communication.Grpc.PullSiteCallsResponse.Parser, new[]{ "Operationals", "MoreAvailable" }, null, null, null, null)
           }));
     }
@@ -5200,6 +5201,7 @@ namespace ZB.MOM.WW.ScadaBridge.Communication.Grpc {
     public PullSiteCallsRequest(PullSiteCallsRequest other) : this() {
       sinceUtc_ = other.sinceUtc_ != null ? other.sinceUtc_.Clone() : null;
       batchSize_ = other.batchSize_;
+      afterId_ = other.afterId_;
       _unknownFields = pb::UnknownFieldSet.Clone(other._unknownFields);
     }
 
@@ -5233,6 +5235,27 @@ namespace ZB.MOM.WW.ScadaBridge.Communication.Grpc {
       }
     }
 
+    /// <summary>Field number for the "after_id" field.</summary>
+    public const int AfterIdFieldNumber = 3;
+    private string afterId_ = "";
+    /// <summary>
+    /// Composite-keyset cursor (Task 15): the TrackedOperationId ("D" GUID form) of
+    /// the last row already consumed at since_utc. When set, the site returns only
+    /// rows strictly after (since_utc, after_id) under a deterministic
+    /// (UpdatedAtUtc, TrackedOperationId) order — un-pinning a batch that would
+    /// otherwise stall when more than batch_size rows share one since_utc instant.
+    /// Empty (the proto3 string default) preserves the legacy inclusive >= behaviour,
+    /// so an older central that never sets it is unaffected. Additive-only.
+    /// </summary>
+    [global::System.Diagnostics.DebuggerNonUserCodeAttribute]
+    [global::System.CodeDom.Compiler.GeneratedCode("protoc", null)]
+    public string AfterId {
+      get { return afterId_; }
+      set {
+        afterId_ = pb::ProtoPreconditions.CheckNotNull(value, "value");
+      }
+    }
+
     [global::System.Diagnostics.DebuggerNonUserCodeAttribute]
     [global::System.CodeDom.Compiler.GeneratedCode("protoc", null)]
     public override bool Equals(object other) {
@@ -5250,6 +5273,7 @@ namespace ZB.MOM.WW.ScadaBridge.Communication.Grpc {
       }
       if (!object.Equals(SinceUtc, other.SinceUtc)) return false;
       if (BatchSize != other.BatchSize) return false;
+      if (AfterId != other.AfterId) return false;
       return Equals(_unknownFields, other._unknownFields);
     }
 
@@ -5259,6 +5283,7 @@ namespace ZB.MOM.WW.ScadaBridge.Communication.Grpc {
       int hash = 1;
       if (sinceUtc_ != null) hash ^= SinceUtc.GetHashCode();
       if (BatchSize != 0) hash ^= BatchSize.GetHashCode();
+      if (AfterId.Length != 0) hash ^= AfterId.GetHashCode();
       if (_unknownFields != null) {
         hash ^= _unknownFields.GetHashCode();
       }
@@ -5285,6 +5310,10 @@ namespace ZB.MOM.WW.ScadaBridge.Communication.Grpc {
         output.WriteRawTag(16);
         output.WriteInt32(BatchSize);
       }
+      if (AfterId.Length != 0) {
+        output.WriteRawTag(26);
+        output.WriteString(AfterId);
+      }
       if (_unknownFields != null) {
         _unknownFields.WriteTo(output);
       }
@@ -5303,6 +5332,10 @@ namespace ZB.MOM.WW.ScadaBridge.Communication.Grpc {
         output.WriteRawTag(16);
         output.WriteInt32(BatchSize);
       }
+      if (AfterId.Length != 0) {
+        output.WriteRawTag(26);
+        output.WriteString(AfterId);
+      }
       if (_unknownFields != null) {
         _unknownFields.WriteTo(ref output);
       }
@@ -5319,6 +5352,9 @@ namespace ZB.MOM.WW.ScadaBridge.Communication.Grpc {
       if (BatchSize != 0) {
         size += 1 + pb::CodedOutputStream.ComputeInt32Size(BatchSize);
       }
+      if (AfterId.Length != 0) {
+        size += 1 + pb::CodedOutputStream.ComputeStringSize(AfterId);
+      }
       if (_unknownFields != null) {
         size += _unknownFields.CalculateSize();
       }
@@ -5340,6 +5376,9 @@ namespace ZB.MOM.WW.ScadaBridge.Communication.Grpc {
       if (other.BatchSize != 0) {
         BatchSize = other.BatchSize;
       }
+      if (other.AfterId.Length != 0) {
+        AfterId = other.AfterId;
+      }
       _unknownFields = pb::UnknownFieldSet.MergeFrom(_unknownFields, other._unknownFields);
     }
 
@@ -5370,6 +5409,10 @@ namespace ZB.MOM.WW.ScadaBridge.Communication.Grpc {
             BatchSize = input.ReadInt32();
             break;
           }
+          case 26: {
+            AfterId = input.ReadString();
+            break;
+          }
         }
       }
     #endif
@@ -5400,6 +5443,10 @@ namespace ZB.MOM.WW.ScadaBridge.Communication.Grpc {
             BatchSize = input.ReadInt32();
             break;
           }
+          case 26: {
+            AfterId = input.ReadString();
+            break;
+          }
         }
       }
     }

src/ZB.MOM.WW.ScadaBridge.SiteRuntime/Tracking/OperationTrackingStore.cs

@@ -365,6 +365,7 @@ public class OperationTrackingStore : IOperationTrackingStore, IAsyncDisposable,
     public async Task<IReadOnlyList<SiteCallOperational>> ReadChangedSinceAsync(
         DateTime sinceUtc,
         int batchSize,
+        string? afterId = null,
         CancellationToken ct = default)
     {
         ObjectDisposedException.ThrowIf(Volatile.Read(ref _disposeState) != 0, this);
@@ -379,18 +380,28 @@ public class OperationTrackingStore : IOperationTrackingStore, IAsyncDisposable,
         await readConnection.OpenAsync(ct).ConfigureAwait(false);
 
         await using var cmd = readConnection.CreateCommand();
-        // Inclusive lower bound on UpdatedAtUtc (>=) so a caller resuming from
+
-        // the last returned timestamp does not skip a row sharing that instant;
+        // Composite (UpdatedAtUtc, TrackedOperationId) keyset (Task 15). Ordering
-        // central ingest is insert-if-not-exists + upsert-on-newer, so the
+        // is ALWAYS deterministic (UpdatedAtUtc ASC, TrackedOperationId ASC) so the
-        // boundary row re-read is a no-op. ORDER BY ... ASC + LIMIT yields the
+        // cursor is well-defined even when many rows share one UpdatedAtUtc.
-        // OLDEST matching rows so the cursor advances monotonically.
+        //   • afterId present → strict keyset: skip everything up to and including
-        cmd.CommandText = """
+        //     (sinceUtc, afterId). Without this a page fully inside one UpdatedAtUtc
+        //     instant re-reads forever (the plain >= resume never advances past the
+        //     tie). TrackedOperationId is the "D" GUID form on both sides, so the
+        //     TEXT comparison is a stable total order.
+        //   • afterId null/empty → legacy inclusive >= sinceUtc (an older central
+        //     that does not send a cursor is byte-for-byte unaffected).
+        var useKeyset = !string.IsNullOrEmpty(afterId);
+        var predicate = useKeyset
+            ? "UpdatedAtUtc > $since OR (UpdatedAtUtc = $since AND TrackedOperationId > $afterId)"
+            : "UpdatedAtUtc >= $since";
+        cmd.CommandText = $"""
             SELECT TrackedOperationId, Kind, TargetSummary, Status,
                    RetryCount, LastError, HttpStatus,
                    CreatedAtUtc, UpdatedAtUtc, TerminalAtUtc, SourceNode
             FROM   OperationTracking
-            WHERE  UpdatedAtUtc >= $since
+            WHERE  {predicate}
-            ORDER  BY UpdatedAtUtc ASC
+            ORDER  BY UpdatedAtUtc ASC, TrackedOperationId ASC
             LIMIT  $batchSize;
             """;
         // Force UTC kind before formatting so the cursor's "o" text matches the
@@ -403,6 +414,10 @@ public class OperationTrackingStore : IOperationTrackingStore, IAsyncDisposable,
             .ToString("o", CultureInfo.InvariantCulture);
         cmd.Parameters.AddWithValue("$since", sinceText);
         cmd.Parameters.AddWithValue("$batchSize", batchSize);
+        if (useKeyset)
+        {
+            cmd.Parameters.AddWithValue("$afterId", afterId!);
+        }
 
         var rows = new List<SiteCallOperational>();
         await using var reader = await cmd.ExecuteReaderAsync(ct).ConfigureAwait(false);

tests/ZB.MOM.WW.ScadaBridge.Communication.Tests/SiteStreamPullSiteCallsTests.cs

@@ -72,7 +72,7 @@ public class SiteStreamPullSiteCallsTests : TestKit
     {
         var store = Substitute.For<IOperationTrackingStore>();
         var rows = Enumerable.Range(0, 5).Select(_ => NewOperational()).ToList();
-        store.ReadChangedSinceAsync(Arg.Any<DateTime>(), Arg.Any<int>(), Arg.Any<CancellationToken>())
+        store.ReadChangedSinceAsync(Arg.Any<DateTime>(), Arg.Any<int>(), Arg.Any<string?>(), Arg.Any<CancellationToken>())
             .Returns((IReadOnlyList<SiteCallOperational>)rows);
 
         var server = CreateServer();
@@ -97,7 +97,7 @@ public class SiteStreamPullSiteCallsTests : TestKit
     {
         var store = Substitute.For<IOperationTrackingStore>();
         var capturedSince = DateTime.MinValue;
-        store.ReadChangedSinceAsync(Arg.Any<DateTime>(), Arg.Any<int>(), Arg.Any<CancellationToken>())
+        store.ReadChangedSinceAsync(Arg.Any<DateTime>(), Arg.Any<int>(), Arg.Any<string?>(), Arg.Any<CancellationToken>())
             .Returns(call =>
             {
                 capturedSince = call.ArgAt<DateTime>(0);
@@ -121,6 +121,65 @@ public class SiteStreamPullSiteCallsTests : TestKit
         Assert.Equal(since, capturedSince);
     }
 
+    [Fact]
+    public async Task PullSiteCalls_AfterIdSet_PassesCursorThroughToStore()
+    {
+        // Task 15: a non-empty after_id is forwarded verbatim to the store as the
+        // composite-keyset cursor.
+        var store = Substitute.For<IOperationTrackingStore>();
+        string? capturedAfterId = "SENTINEL";
+        store.ReadChangedSinceAsync(Arg.Any<DateTime>(), Arg.Any<int>(), Arg.Any<string?>(), Arg.Any<CancellationToken>())
+            .Returns(call =>
+            {
+                capturedAfterId = call.ArgAt<string?>(2);
+                return (IReadOnlyList<SiteCallOperational>)Array.Empty<SiteCallOperational>();
+            });
+
+        var server = CreateServer();
+        server.SetOperationTrackingStore(store);
+
+        var cursor = Guid.NewGuid().ToString("D");
+        var request = new PullSiteCallsRequest
+        {
+            SinceUtc = Timestamp.FromDateTime(DateTime.UtcNow.AddHours(-1)),
+            BatchSize = 50,
+            AfterId = cursor,
+        };
+
+        await server.PullSiteCalls(request, NewContext());
+
+        Assert.Equal(cursor, capturedAfterId);
+    }
+
+    [Fact]
+    public async Task PullSiteCalls_AfterIdUnset_PassesNullToStore()
+    {
+        // proto3 defaults an unset string to ""; the handler must normalise that to
+        // null so the store keeps its legacy inclusive >= behaviour (older central).
+        var store = Substitute.For<IOperationTrackingStore>();
+        string? capturedAfterId = "SENTINEL";
+        store.ReadChangedSinceAsync(Arg.Any<DateTime>(), Arg.Any<int>(), Arg.Any<string?>(), Arg.Any<CancellationToken>())
+            .Returns(call =>
+            {
+                capturedAfterId = call.ArgAt<string?>(2);
+                return (IReadOnlyList<SiteCallOperational>)Array.Empty<SiteCallOperational>();
+            });
+
+        var server = CreateServer();
+        server.SetOperationTrackingStore(store);
+
+        var request = new PullSiteCallsRequest
+        {
+            SinceUtc = Timestamp.FromDateTime(DateTime.UtcNow.AddHours(-1)),
+            BatchSize = 50,
+            // AfterId left unset → "" on the wire.
+        };
+
+        await server.PullSiteCalls(request, NewContext());
+
+        Assert.Null(capturedAfterId);
+    }
+
     [Fact]
     public async Task PullSiteCalls_SinceUtcUnset_PassesDateTimeMinValue()
     {
@@ -130,7 +189,7 @@ public class SiteStreamPullSiteCallsTests : TestKit
         // without a null-deref — this proves the very first cycle doesn't crash.
         var store = Substitute.For<IOperationTrackingStore>();
         var captured = new DateTime(2099, 1, 1, 0, 0, 0, DateTimeKind.Utc); // sentinel
-        store.ReadChangedSinceAsync(Arg.Any<DateTime>(), Arg.Any<int>(), Arg.Any<CancellationToken>())
+        store.ReadChangedSinceAsync(Arg.Any<DateTime>(), Arg.Any<int>(), Arg.Any<string?>(), Arg.Any<CancellationToken>())
             .Returns(call =>
             {
                 captured = call.ArgAt<DateTime>(0);
@@ -158,7 +217,7 @@ public class SiteStreamPullSiteCallsTests : TestKit
     {
         var store = Substitute.For<IOperationTrackingStore>();
         var rows = Enumerable.Range(0, 3).Select(_ => NewOperational()).ToList();
-        store.ReadChangedSinceAsync(Arg.Any<DateTime>(), Arg.Any<int>(), Arg.Any<CancellationToken>())
+        store.ReadChangedSinceAsync(Arg.Any<DateTime>(), Arg.Any<int>(), Arg.Any<string?>(), Arg.Any<CancellationToken>())
             .Returns((IReadOnlyList<SiteCallOperational>)rows);
 
         var server = CreateServer();
@@ -200,7 +259,7 @@ public class SiteStreamPullSiteCallsTests : TestKit
     {
         // Best-effort: a read fault must never abort the reconciliation tick.
         var store = Substitute.For<IOperationTrackingStore>();
-        store.ReadChangedSinceAsync(Arg.Any<DateTime>(), Arg.Any<int>(), Arg.Any<CancellationToken>())
+        store.ReadChangedSinceAsync(Arg.Any<DateTime>(), Arg.Any<int>(), Arg.Any<string?>(), Arg.Any<CancellationToken>())
             .ThrowsAsync(new InvalidOperationException("SQLite disposed mid-call"));
 
         var server = CreateServer();

tests/ZB.MOM.WW.ScadaBridge.SiteRuntime.Tests/Tracking/OperationTrackingStoreTests.cs

@@ -466,7 +466,7 @@ public class OperationTrackingStoreTests
 
         // Cursor at the middle row's UpdatedAtUtc: inclusive lower bound, so
         // middle + newer come back, older is excluded.
-        var result = await store.ReadChangedSinceAsync(t0.AddMinutes(10), batchSize: 100, CancellationToken.None);
+        var result = await store.ReadChangedSinceAsync(t0.AddMinutes(10), batchSize: 100, ct: CancellationToken.None);
 
         Assert.Equal(2, result.Count);
         Assert.Equal(middle, result[0].TrackedOperationId);
@@ -483,7 +483,7 @@ public class OperationTrackingStoreTests
         await store.RecordEnqueueAsync(TrackedOperationId.New(), nameof(AuditKind.ApiCallCached), "A", null, null, null);
         await store.RecordEnqueueAsync(TrackedOperationId.New(), nameof(AuditKind.ApiCallCached), "B", null, null, null);
 
-        var result = await store.ReadChangedSinceAsync(DateTime.MinValue, batchSize: 100, CancellationToken.None);
+        var result = await store.ReadChangedSinceAsync(DateTime.MinValue, batchSize: 100, ct: CancellationToken.None);
 
         Assert.Equal(2, result.Count);
     }
@@ -504,7 +504,7 @@ public class OperationTrackingStoreTests
             SetUpdatedAt(dataSource, id, t0.AddMinutes(i));
         }
 
-        var result = await store.ReadChangedSinceAsync(DateTime.MinValue, batchSize: 3, CancellationToken.None);
+        var result = await store.ReadChangedSinceAsync(DateTime.MinValue, batchSize: 3, ct: CancellationToken.None);
 
         // Capped to 3 — and the cap takes the OLDEST 3 (asc order) so the
         // caller can advance the cursor monotonically across follow-up pulls.
@@ -527,7 +527,7 @@ public class OperationTrackingStoreTests
         await store.RecordAttemptAsync(apiId, nameof(AuditStatus.Attempted), 2, "HTTP 503", 503);
         await store.RecordTerminalAsync(dbId, nameof(AuditStatus.Parked), "max retries", null);
 
-        var result = await store.ReadChangedSinceAsync(DateTime.MinValue, batchSize: 100, CancellationToken.None);
+        var result = await store.ReadChangedSinceAsync(DateTime.MinValue, batchSize: 100, ct: CancellationToken.None);
         var api = result.Single(r => r.TrackedOperationId == apiId);
         var db = result.Single(r => r.TrackedOperationId == dbId);
 
@@ -720,4 +720,92 @@ public class OperationTrackingStoreTests
 
         Assert.Equal(2, rows.Count);
     }
+
+    // ── Task 15: composite (UpdatedAtUtc, TrackedOperationId) keyset ──
+
+    /// <summary>
+    /// Seeds <c>batchSize + 2</c> rows that all share ONE <c>UpdatedAtUtc</c> and
+    /// pages through them with the composite keyset. Page 1 (no cursor) returns the
+    /// first <c>batchSize</c>; page 2 resumes from <c>(sinceUtc, lastId)</c> and
+    /// returns the REMAINING rows with no overlap — the stall the plain inclusive
+    /// <c>&gt;=</c> resume can never escape.
+    /// </summary>
+    [Fact]
+    public async Task ReadChangedSince_CompositeKeyset_AdvancesPastSharedTimestamp()
+    {
+        const int batchSize = 3;
+        var (store, dataSource) = CreateStore(nameof(ReadChangedSince_CompositeKeyset_AdvancesPastSharedTimestamp));
+        await using var _store = store;
+
+        for (var i = 0; i < batchSize + 2; i++)
+        {
+            await store.RecordEnqueueAsync(TrackedOperationId.New(),
+                kind: nameof(AuditKind.ApiCallCached), targetSummary: $"ERP.{i}",
+                sourceInstanceId: "I", sourceScript: "S", sourceNode: "node-a");
+        }
+
+        var boundary = ForceSharedUpdatedAt(dataSource);
+
+        // Page 1: no cursor → the first batchSize rows, ordered by TrackedOperationId.
+        var page1 = await store.ReadChangedSinceAsync(boundary, batchSize);
+        Assert.Equal(batchSize, page1.Count);
+
+        // Page 2: resume strictly after the last id of page 1.
+        var afterId = page1[^1].TrackedOperationId.ToString();
+        var page2 = await store.ReadChangedSinceAsync(boundary, batchSize, afterId);
+
+        // The remaining two rows, none seen on page 1, all strictly greater than the cursor.
+        Assert.Equal(2, page2.Count);
+        var page1Ids = page1.Select(r => r.TrackedOperationId).ToHashSet();
+        Assert.All(page2, r => Assert.DoesNotContain(r.TrackedOperationId, page1Ids));
+        Assert.All(page2, r => Assert.True(
+            string.CompareOrdinal(r.TrackedOperationId.ToString(), afterId) > 0));
+
+        // Union covers every seeded row exactly once — no gap, no dup.
+        Assert.Equal(batchSize + 2, page1Ids.Concat(page2.Select(r => r.TrackedOperationId)).Distinct().Count());
+    }
+
+    /// <summary>
+    /// Regression guard for the legacy contract: with NO cursor a resume at the
+    /// shared timestamp re-reads the same page (the exact stall the keyset fixes) —
+    /// an older central that never sends <c>after_id</c> is byte-for-byte unaffected.
+    /// </summary>
+    [Fact]
+    public async Task ReadChangedSince_WithoutKeyset_ReReadsSamePage_LegacyPreserved()
+    {
+        const int batchSize = 3;
+        var (store, dataSource) = CreateStore(nameof(ReadChangedSince_WithoutKeyset_ReReadsSamePage_LegacyPreserved));
+        await using var _store = store;
+
+        for (var i = 0; i < batchSize + 2; i++)
+        {
+            await store.RecordEnqueueAsync(TrackedOperationId.New(),
+                kind: nameof(AuditKind.ApiCallCached), targetSummary: $"ERP.{i}",
+                sourceInstanceId: "I", sourceScript: "S", sourceNode: "node-a");
+        }
+
+        var boundary = ForceSharedUpdatedAt(dataSource);
+
+        var page1 = await store.ReadChangedSinceAsync(boundary, batchSize);
+        // Legacy resume: same sinceUtc, no cursor → identical page (the stall).
+        var pageAgain = await store.ReadChangedSinceAsync(boundary, batchSize);
+
+        Assert.Equal(
+            page1.Select(r => r.TrackedOperationId).ToList(),
+            pageAgain.Select(r => r.TrackedOperationId).ToList());
+    }
+
+    /// <summary>Forces every OperationTracking row to one exact UpdatedAtUtc instant; returns it.</summary>
+    private DateTime ForceSharedUpdatedAt(string dataSource)
+    {
+        var boundary = DateTime.UtcNow;
+        var boundaryText = DateTime.SpecifyKind(boundary, DateTimeKind.Utc)
+            .ToString("o", System.Globalization.CultureInfo.InvariantCulture);
+        using var conn = OpenVerifierConnection(dataSource);
+        using var cmd = conn.CreateCommand();
+        cmd.CommandText = "UPDATE OperationTracking SET UpdatedAtUtc = $ts";
+        cmd.Parameters.AddWithValue("$ts", boundaryText);
+        cmd.ExecuteNonQuery();
+        return boundary;
+    }
 }