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

## Problem

`ClickHouseRunsRepository.listRunIds` / `listRuns` order results by the
composite key `(created_at, run_id)`, but the cursor predicate cut on
`run_id` **alone**:

```ts
.where("run_id < {runId: String}", { runId: cursor })
.orderBy("created_at DESC, run_id DESC")
```

This is only sound when `run_id` lexicographic order matches
`created_at` order. `run_id`s are cuids — only coarsely time-sortable —
so when a burst of runs is created within a sub-second window, the two
orders can diverge. When they do, the next-page predicate (`run_id <
cursor`, where `cursor` is the *last* page element = the smallest
`created_at`, not necessarily the smallest `run_id`):

- **re-includes** rows already returned on a previous page (duplicates),
and
- **skips** rows it should have returned (silent data loss).

For bulk **replay** this caused runs to be replayed more than once
(replay has no idempotency guard). For the dashboard and the `runs.list`
API it could silently repeat or skip runs at page boundaries.

## Fix

Make the cursor predicate match the composite ordering:

- Cursors now encode the full `(created_at, run_id)` key as an **opaque
URL-safe base64 token**
(`base64url({"c":<createdAtMs>,"r":"<runId>"})`), and the query cuts on
the matching tuple — `(created_at, run_id) < (…)` forward / `> (…)`
backward.
- The `ORDER BY` is unchanged, so the query stays aligned with the
table's primary key — no performance regression (the tuple range
predicate is actually more index-friendly than `run_id <` alone).
- Cursors are **server-issued opaque tokens** (the SDK only echoes
`pagination.next` / `pagination.previous` back), so this needs **no
client/SDK update**. Legacy cursors were the bare internal `run_id`;
they're detected by decode failure (a cuid isn't a valid base64-wrapped
JSON payload) and fall back to the old `run_id`-only predicate, so
in-flight cursors keep working and drain naturally. New cursors also no
longer expose a bare internal run id.
- `listRunIds` is now the single cursor-aware list primitive: it returns
`{ runIds, pagination: { nextCursor, previousCursor } }`, and `listRuns`
builds on it (one place constructs cursors). Bulk actions consume the
same method and advance by `pagination.nextCursor`, finishing when it's
`null`.
- `getTaskRunsQueryBuilder` now also selects
`toUnixTimestamp64Milli(created_at) AS created_at_ms`, using a dedicated
`TaskRunListQueryResult` schema. The shared `TaskRunV2QueryResult` stays
`run_id`-only so the run-engine pending-version lookup
(`getPendingVersionIdsQueryBuilder`, which selects only `run_id`)
doesn't fail validation on a column it doesn't query.

## Tests

New `runsRepositoryCursor.test.ts` (testcontainer-backed, real
Postgres→ClickHouse replication):

- **forward** pagination returns every run exactly once when `run_id`
order is the reverse of `created_at` order (reproduces the
duplicate/skip bug — fails on `main`; this
walk-until-`nextCursor`-null-and-assert-complete is exactly the bulk
action's iteration),
- **backward** pagination round-trips to the previous page across a
boundary,
- **legacy** bare-`run_id` cursor still uses the old predicate
(backwards compatibility).

The existing `runsRepository` suites (part1–4) still pass; `part4`'s
`count new runs with listRunIds` test was updated for the new `{ runIds,
pagination }` return shape, and the `clickhouse` `taskRuns`
query-builder snapshots were regenerated for the added `created_at_ms`
column.

## Notes

- Separate, pre-existing issue (out of scope, not introduced here):
`listRuns`' backward display-slicing (`rows.slice(1, size+1)` when
`hasMore`) has an off-by-one that can return a straddled page. Tracked
separately.

🤖 Generated with [Claude Code](https://claude.com/claude-code)

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,6 @@
+---
+area: webapp
+type: fix
+---
+
+Fix run pagination that could duplicate or skip runs: the query orders by `(created_at, run_id)` but the cursor cut on `run_id` alone, which diverges when run_id order doesn't match created_at order (e.g. bulk replay re-processing runs). Cursors now encode the composite key as an opaque token and cut on the matching tuple; legacy bare-run_id cursors stay supported for in-flight pagination.

apps/webapp/app/presenters/v3/TestTaskPresenter.server.ts

@@ -203,7 +203,7 @@ export class TestTaskPresenter {
       prisma: this.replica as PrismaClient,
     });
 
-    const runIds = await runsRepository.listRunIds({
+    const { runIds } = await runsRepository.listRunIds({
       organizationId: environment.organizationId,
       environmentId: environment.id,
       projectId: environment.projectId,

apps/webapp/app/routes/resources.orgs.$organizationSlug.projects.$projectParam.env.$envParam.runs.live.ts

@@ -58,7 +58,7 @@ export async function loader({ request, params }: LoaderFunctionArgs) {
             return { count: 0, since: newRunsSince };
           }
 
-          const newRunIds = await runsRepository.listRunIds({
+          const { runIds: newRunIds } = await runsRepository.listRunIds({
             organizationId: project.organizationId,
             projectId: project.id,
             environmentId: environment.id,

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

@@ -4,12 +4,16 @@ import {
   type FilterRunsOptions,
   type IRunsRepository,
   type ListRunsOptions,
+  type RunIdsPage,
   type RunListInputOptions,
   type RunsRepositoryOptions,
   type TagListOptions,
   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 +22,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,75 +79,89 @@ 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 listFriendlyRunIds(options: ListRunsOptions) {
-    // First get internal IDs from ClickHouse
-    const internalIds = await this.listRunIds(options);
+  /**
+   * A keyset-paginated page of run ids ordered by `(created_at, run_id)`, plus
+   * the cursors to page forward/backward. Cursors are composite tokens that
+   * match the ordering, so pagination can't duplicate or skip runs even when
+   * run_id order diverges from created_at order. This is the single source of
+   * cursor construction — `listRuns` and bulk actions both build on it.
+   */
+  async listRunIds(options: ListRunsOptions): Promise<RunIdsPage> {
+    const rows = await this.listRunRows(options);
 
-    if (internalIds.length === 0) {
-      return [];
-    }
-
-    // Then get friendly IDs from Prisma
-    const runs = await this.options.prisma.taskRun.findMany({
-      where: {
-        id: {
-          in: internalIds,
-        },
-      },
-      select: {
-        friendlyId: true,
-      },
-    });
-
-    return runs.map((run) => run.friendlyId);
-  }
-
-  async listRuns(options: ListRunsOptions) {
-    const runIds = await this.listRunIds(options);
+    // listRunRows fetches one extra row beyond page.size to detect "has more".
+    const hasMore = rows.length > options.page.size;
 
-    // If there are more runs than the page size, we need to fetch the next page
-    const hasMore = runIds.length > options.page.size;
+    const cursorFor = (row: RunCursorRow | undefined): string | null =>
+      row ? encodeRunsCursor(row.createdAt, row.runId) : null;
 
     let nextCursor: string | null = null;
     let previousCursor: string | null = null;
 
-    //get cursors for next and previous pages
     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 is the last run on 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 =
-      options.page.direction === "backward" && hasMore
-        ? runIds.slice(1, options.page.size + 1)
-        : runIds.slice(0, options.page.size);
+    const runIds = (
+      direction === "backward" && hasMore
+        ? rows.slice(1, options.page.size + 1)
+        : rows.slice(0, options.page.size)
+    ).map((row) => row.runId);
+
+    return { runIds, pagination: { nextCursor, previousCursor } };
+  }
+
+  async listFriendlyRunIds(options: ListRunsOptions) {
+    // First get internal IDs from ClickHouse
+    const { runIds } = await this.listRunIds(options);
+
+    if (runIds.length === 0) {
+      return [];
+    }
+
+    // Then get friendly IDs from Prisma
+    const runs = await this.options.prisma.taskRun.findMany({
+      where: {
+        id: {
+          in: runIds,
+        },
+      },
+      select: {
+        friendlyId: true,
+      },
+    });
+
+    return runs.map((run) => run.friendlyId);
+  }
+
+  async listRuns(options: ListRunsOptions) {
+    const { runIds, pagination } = await this.listRunIds(options);
 
     let runs = await this.options.prisma.taskRun.findMany({
       where: {
         id: {
-          in: runIdsToReturn,
+          in: runIds,
         },
       },
       orderBy: {
@@ -163,10 +208,7 @@ export class ClickHouseRunsRepository implements IRunsRepository {
 
     return {
       runs,
-      pagination: {
-        nextCursor,
-        previousCursor,
-      },
+      pagination,
     };
   }
 

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.
+ */
+
+import { z } from "zod";
+
+export type DecodedRunsCursor =
+  | { kind: "composite"; createdAt: number; runId: string }
+  | { kind: "legacy"; runId: string };
+
+// `c` = created_at (ms since epoch), `r` = run_id. Short keys keep the token small.
+const CompositeCursor = z.object({
+  c: z.number().int(),
+  r: z.string().min(1),
+});
+
+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 = CompositeCursor.safeParse(
+      JSON.parse(Buffer.from(cursor, "base64url").toString("utf8"))
+    );
+    if (parsed.success) {
+      return { kind: "composite", createdAt: parsed.data.c, runId: parsed.data.r };
+    }
+  } catch {
+    // JSON.parse threw — not a composite cursor.
+  }
+
+  return { kind: "legacy", runId: cursor };
+}

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

@@ -127,9 +127,25 @@ export type TagList = {
   tags: string[];
 };
 
+export type CursorPagination = {
+  nextCursor: string | null;
+  previousCursor: string | null;
+};
+
+export type RunIdsPage = {
+  runIds: string[];
+  pagination: CursorPagination;
+};
+
 export interface IRunsRepository {
   name: string;
-  listRunIds(options: ListRunsOptions): Promise<string[]>;
+  /**
+   * A keyset-paginated page of run ids plus the cursors to navigate
+   * forward/backward. The cursors are opaque composite `(created_at, run_id)`
+   * tokens, so pagination can't duplicate or skip runs. This is the single
+   * cursor-aware list primitive — `listRuns` and bulk actions build on it.
+   */
+  listRunIds(options: ListRunsOptions): Promise<RunIdsPage>;
   /** 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<{
@@ -154,7 +170,7 @@ export class RunsRepository implements IRunsRepository {
     return "runsRepository";
   }
 
-  async listRunIds(options: ListRunsOptions): Promise<string[]> {
+  async listRunIds(options: ListRunsOptions): Promise<RunIdsPage> {
     return startActiveSpan(
       "runsRepository.listRunIds",
       async () => this.clickHouseRunsRepository.listRunIds(options),

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

@@ -159,8 +159,13 @@ 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,
+      pagination: { nextCursor },
+    } = await runsRepository.listRunIds({
       ...filters,
       page: {
         size: env.BULK_ACTION_BATCH_SIZE,
@@ -172,8 +177,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: {
@@ -274,7 +277,10 @@ export class BulkActionService extends BaseService {
       }
     }
 
-    const isFinished = runIdsToProcess.length === 0;
+    // A null nextCursor means there is no further page — this batch was the
+    // last (or there were no runs at all), so the action is complete. (An empty
+    // batch also yields a null cursor.)
+    const isFinished = nextCursor === null;
 
     logger.debug("Bulk action group processed batch", {
       bulkActionId,
@@ -292,7 +298,8 @@ export class BulkActionService extends BaseService {
     const updatedGroup = await this._prisma.bulkActionGroup.update({
       where: { id: bulkActionId },
       data: {
-        cursor: runIdsToProcess.at(runIdsToProcess.length - 1),
+        // Json column: leave unchanged when there's no next cursor (finished).
+        cursor: nextCursor ?? undefined,
         successCount: {
           increment: successCount,
         },