fix(webapp): use composite keyset cursor for run pagination

listRunIds/listRuns order by the composite key (created_at, run_id) but
the cursor predicate cut on run_id alone. That is only sound when run_id
lexicographic order matches created_at order. When a burst of runs is
created such that the two diverge, keyset pagination both re-includes
already-returned runs (duplicates) and drops runs it should return
(skips). For bulk replay this produced duplicate runs; for the dashboard
and runs.list it could silently skip or repeat runs at page boundaries.

- Encode cursors as the composite (created_at, run_id) key
  (v2_<createdAtMs>_<runId>) and cut on the matching tuple predicate
  ((created_at, run_id) < / > (...)). The ORDER BY is unchanged, so the
  table's primary-key alignment (and query performance) is preserved.
- Cursors are server-issued opaque tokens (the SDK just echoes
  pagination.next/previous back), so this needs no client update. Legacy
  bare-run_id cursors decode to the old run_id-only predicate for
  backwards compatibility with in-flight cursors.
- Add listRunIdsWithCursor for forward-only batch iteration (bulk
  actions) so the created_at component is sourced from the same query
  that orders the rows.
- ClickHouse getTaskRunsQueryBuilder now also selects
  toUnixTimestamp64Milli(created_at) AS created_at_ms.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

Author: matt-aitken

.server-changes/bulk-action-cursor-pagination.md

@@ -0,0 +1,17 @@
+---
+area: webapp
+type: fix
+---
+
+Fix unsound keyset pagination in `ClickHouseRunsRepository` that could duplicate
+or skip runs. The list query orders by `(created_at, run_id)` but the cursor
+predicate cut on `run_id` alone, which is only sound when run_id order matches
+created_at order. When a burst of runs is created such that the two diverge,
+bulk replay could re-process already-replayed runs, and dashboard / `runs.list`
+pagination could silently skip or repeat runs at page boundaries.
+
+Cursors now encode the composite `(created_at, run_id)` key as an opaque
+URL-safe base64 token, and the query cuts on the matching tuple. Cursors are
+server-issued opaque tokens, so this needs no SDK update; legacy bare-run_id
+cursors decode to the old `run_id`-only predicate for backwards compatibility
+with in-flight cursors.

apps/webapp/app/services/runsRepository/clickhouseRunsRepository.server.ts

@@ -10,6 +10,9 @@ import {
   convertRunListInputOptionsToFilterRunsOptions,
 } from "./runsRepository.server";
 import parseDuration from "parse-duration";
+import { decodeRunsCursor, encodeRunsCursor } from "./runsCursor.server";
+
+type RunCursorRow = { runId: string; createdAt: number };
 
 export class ClickHouseRunsRepository implements IRunsRepository {
   constructor(private readonly options: RunsRepositoryOptions) {}
@@ -18,25 +21,52 @@ export class ClickHouseRunsRepository implements IRunsRepository {
     return "clickhouse";
   }
 
-  async listRunIds(options: ListRunsOptions) {
+  /**
+   * Runs the keyset-paginated query and returns `{ runId, createdAt }` rows
+   * (one extra beyond `page.size` to signal "has more"). The ordering is always
+   * the composite `(created_at, run_id)`; the cursor predicate must match it.
+   *
+   * Composite cursors carry both components, so we cut on the
+   * `(created_at, run_id)` tuple — sound regardless of how run_id order relates
+   * to created_at order. Legacy bare-run_id cursors fall back to the old
+   * `run_id`-only predicate (knowingly unsound) for backwards compatibility
+   * with in-flight cursors.
+   */
+  private async listRunRows(options: ListRunsOptions): Promise<RunCursorRow[]> {
     const queryBuilder = this.options.clickhouse.taskRuns.queryBuilder();
     applyRunFiltersToQueryBuilder(
       queryBuilder,
       await convertRunListInputOptionsToFilterRunsOptions(options, this.options.prisma)
     );
 
+    const forward = options.page.direction === "forward" || !options.page.direction;
+
     if (options.page.cursor) {
-      if (options.page.direction === "forward" || !options.page.direction) {
-        queryBuilder
-          .where("run_id < {runId: String}", { runId: options.page.cursor })
-          .orderBy("created_at DESC, run_id DESC")
-          .limit(options.page.size + 1);
+      const decoded = decodeRunsCursor(options.page.cursor);
+
+      if (forward) {
+        if (decoded.kind === "composite") {
+          queryBuilder.where(
+            "(created_at, run_id) < (fromUnixTimestamp64Milli({cursorCreatedAt: Int64}), {runId: String})",
+            { cursorCreatedAt: decoded.createdAt, runId: decoded.runId }
+          );
+        } else {
+          queryBuilder.where("run_id < {runId: String}", { runId: decoded.runId });
+        }
+        queryBuilder.orderBy("created_at DESC, run_id DESC");
       } else {
-        queryBuilder
-          .where("run_id > {runId: String}", { runId: options.page.cursor })
-          .orderBy("created_at ASC, run_id ASC")
-          .limit(options.page.size + 1);
+        if (decoded.kind === "composite") {
+          queryBuilder.where(
+            "(created_at, run_id) > (fromUnixTimestamp64Milli({cursorCreatedAt: Int64}), {runId: String})",
+            { cursorCreatedAt: decoded.createdAt, runId: decoded.runId }
+          );
+        } else {
+          queryBuilder.where("run_id > {runId: String}", { runId: decoded.runId });
+        }
+        queryBuilder.orderBy("created_at ASC, run_id ASC");
       }
+
+      queryBuilder.limit(options.page.size + 1);
     } else {
       // Initial page - no cursor provided
       queryBuilder.orderBy("created_at DESC, run_id DESC").limit(options.page.size + 1);
@@ -48,8 +78,31 @@ export class ClickHouseRunsRepository implements IRunsRepository {
       throw queryError;
     }
 
-    const runIds = result.map((row) => row.run_id);
-    return runIds;
+    return result.map((row) => ({ runId: row.run_id, createdAt: row.created_at_ms }));
+  }
+
+  async listRunIds(options: ListRunsOptions) {
+    const rows = await this.listRunRows(options);
+    return rows.map((row) => row.runId);
+  }
+
+  /**
+   * Forward-only batch iteration (bulk actions). Returns up to `page.size` run
+   * ids plus the composite cursor for the next batch (null when this batch is
+   * empty). The `created_at` component comes from the same query that orders the
+   * rows, so the next batch's tuple predicate is always consistent.
+   */
+  async listRunIdsWithCursor(
+    options: ListRunsOptions
+  ): Promise<{ runIds: string[]; nextCursor: string | null }> {
+    const rows = await this.listRunRows(options);
+    const batch = rows.slice(0, options.page.size);
+    const last = batch.at(-1);
+
+    return {
+      runIds: batch.map((row) => row.runId),
+      nextCursor: last ? encodeRunsCursor(last.createdAt, last.runId) : null,
+    };
   }
 
   async listFriendlyRunIds(options: ListRunsOptions) {
@@ -76,10 +129,15 @@ export class ClickHouseRunsRepository implements IRunsRepository {
   }
 
   async listRuns(options: ListRunsOptions) {
-    const runIds = await this.listRunIds(options);
+    const rows = await this.listRunRows(options);
 
     // If there are more runs than the page size, we need to fetch the next page
-    const hasMore = runIds.length > options.page.size;
+    const hasMore = rows.length > options.page.size;
+
+    // Cursors carry both (created_at, run_id) so the next/prev page predicate
+    // matches the composite ordering — see runsCursor.server.ts.
+    const cursorFor = (row: RunCursorRow | undefined): string | null =>
+      row ? encodeRunsCursor(row.createdAt, row.runId) : null;
 
     let nextCursor: string | null = null;
     let previousCursor: string | null = null;
@@ -88,30 +146,31 @@ export class ClickHouseRunsRepository implements IRunsRepository {
     const direction = options.page.direction ?? "forward";
     switch (direction) {
       case "forward": {
-        previousCursor = options.page.cursor ? runIds.at(0) ?? null : null;
+        previousCursor = options.page.cursor ? cursorFor(rows.at(0)) : null;
         if (hasMore) {
-          // The next cursor should be the last run ID from this page
-          nextCursor = runIds[options.page.size - 1];
+          // The next cursor should be the last run from this page
+          nextCursor = cursorFor(rows[options.page.size - 1]);
         }
         break;
       }
       case "backward": {
-        const reversedRunIds = [...runIds].reverse();
+        const reversedRows = [...rows].reverse();
         if (hasMore) {
-          previousCursor = reversedRunIds.at(1) ?? null;
-          nextCursor = reversedRunIds.at(options.page.size) ?? null;
+          previousCursor = cursorFor(reversedRows.at(1));
+          nextCursor = cursorFor(reversedRows.at(options.page.size));
         } else {
-          nextCursor = reversedRunIds.at(options.page.size - 1) ?? null;
+          nextCursor = cursorFor(reversedRows.at(options.page.size - 1));
         }
 
         break;
       }
     }
 
-    const runIdsToReturn =
+    const runIdsToReturn = (
       options.page.direction === "backward" && hasMore
-        ? runIds.slice(1, options.page.size + 1)
-        : runIds.slice(0, options.page.size);
+        ? rows.slice(1, options.page.size + 1)
+        : rows.slice(0, options.page.size)
+    ).map((row) => row.runId);
 
     let runs = await this.options.prisma.taskRun.findMany({
       where: {

apps/webapp/app/services/runsRepository/runsCursor.server.ts

@@ -0,0 +1,47 @@
+/**
+ * Cursor encoding for keyset pagination over `(created_at, run_id)`.
+ *
+ * The list query orders by the composite key `(created_at, run_id)`, so a sound
+ * cursor must carry BOTH components — cutting on `run_id` alone re-includes and
+ * skips rows whenever `run_id` order diverges from `created_at` order.
+ *
+ * A cursor is an opaque URL-safe base64 token wrapping `{ c: createdAtMs, r:
+ * runId }`. Cursors are server-issued (the SDK just echoes
+ * `pagination.next`/`previous` back), so this format needs no client update.
+ *
+ * Legacy cursors were the bare internal run_id (a cuid). They are detected by
+ * decode failure: a cuid base64-decodes to non-JSON bytes, so it falls through
+ * to `{ kind: "legacy" }` and the old (knowingly unsound) `run_id`-only
+ * predicate. In-flight legacy cursors keep working and drain naturally.
+ */
+
+export type DecodedRunsCursor =
+  | { kind: "composite"; createdAt: number; runId: string }
+  | { kind: "legacy"; runId: string };
+
+export function encodeRunsCursor(createdAtMs: number, runId: string): string {
+  return Buffer.from(JSON.stringify({ c: createdAtMs, r: runId })).toString("base64url");
+}
+
+export function decodeRunsCursor(cursor: string): DecodedRunsCursor {
+  try {
+    const parsed = JSON.parse(Buffer.from(cursor, "base64url").toString("utf8")) as unknown;
+    if (
+      parsed &&
+      typeof parsed === "object" &&
+      "c" in parsed &&
+      "r" in parsed &&
+      typeof (parsed as { c: unknown }).c === "number" &&
+      Number.isFinite((parsed as { c: number }).c) &&
+      typeof (parsed as { r: unknown }).r === "string" &&
+      (parsed as { r: string }).r.length > 0
+    ) {
+      const { c, r } = parsed as { c: number; r: string };
+      return { kind: "composite", createdAt: c, runId: r };
+    }
+  } catch {
+    // Not a v2 cursor — fall through to the legacy bare-run_id path.
+  }
+
+  return { kind: "legacy", runId: cursor };
+}

apps/webapp/app/services/runsRepository/runsRepository.server.ts

@@ -130,6 +130,15 @@ export type TagList = {
 export interface IRunsRepository {
   name: string;
   listRunIds(options: ListRunsOptions): Promise<string[]>;
+  /**
+   * Forward-only batch iteration (bulk actions). Returns up to `page.size` run
+   * ids and the composite cursor for the next batch (null when the batch is
+   * empty). Keeping cursor construction here ensures the `created_at` component
+   * comes from the same source as the ordering.
+   */
+  listRunIdsWithCursor(
+    options: ListRunsOptions
+  ): Promise<{ runIds: string[]; nextCursor: string | null }>;
   /** Returns friendly IDs (e.g., run_xxx) instead of internal UUIDs. Used for ClickHouse task_events queries. */
   listFriendlyRunIds(options: ListRunsOptions): Promise<string[]>;
   listRuns(options: ListRunsOptions): Promise<{
@@ -169,6 +178,23 @@ export class RunsRepository implements IRunsRepository {
     );
   }
 
+  async listRunIdsWithCursor(
+    options: ListRunsOptions
+  ): Promise<{ runIds: string[]; nextCursor: string | null }> {
+    return startActiveSpan(
+      "runsRepository.listRunIdsWithCursor",
+      async () => this.clickHouseRunsRepository.listRunIdsWithCursor(options),
+      {
+        attributes: {
+          "repository.name": "clickhouse",
+          organizationId: options.organizationId,
+          projectId: options.projectId,
+          environmentId: options.environmentId,
+        },
+      }
+    );
+  }
+
   async listFriendlyRunIds(options: ListRunsOptions): Promise<string[]> {
     return startActiveSpan(
       "runsRepository.listFriendlyRunIds",

apps/webapp/app/v3/services/bulk/BulkActionV2.server.ts

@@ -159,8 +159,10 @@ export class BulkActionService extends BaseService {
       throw new Error(`Bulk action group has invalid query name: ${group.queryName}`);
     }
 
-    // 2. Get the runs to process in this batch
-    const runIds = await runsRepository.listRunIds({
+    // 2. Get the runs to process in this batch, plus the cursor for the next
+    // batch. The cursor is a composite (created_at, run_id) keyset cursor so the
+    // next batch can't re-include or skip runs.
+    const { runIds: runIdsToProcess, nextCursor } = await runsRepository.listRunIdsWithCursor({
       ...filters,
       page: {
         size: env.BULK_ACTION_BATCH_SIZE,
@@ -172,8 +174,6 @@ export class BulkActionService extends BaseService {
     // 3. Process the runs
     let successCount = 0;
     let failureCount = 0;
-    // Slice because we fetch an extra for the cursor
-    const runIdsToProcess = runIds.slice(0, env.BULK_ACTION_BATCH_SIZE);
 
     switch (group.type) {
       case BulkActionType.CANCEL: {
@@ -292,7 +292,7 @@ export class BulkActionService extends BaseService {
     const updatedGroup = await this._prisma.bulkActionGroup.update({
       where: { id: bulkActionId },
       data: {
-        cursor: runIdsToProcess.at(runIdsToProcess.length - 1),
+        cursor: nextCursor,
         successCount: {
           increment: successCount,
         },