fix(e2e): paginate stale dispatch receipt rechecks

test/pr-e2e-dispatch-reconciliation.test.ts

@@ -10,6 +10,7 @@ import {
   DispatchReconciliationError,
   dispatchWorkflowWithReconciliation,
 } from "../tools/e2e/pr-e2e-dispatch-reconciliation.mts";
+import type { DispatchNotObservedReceipt } from "../tools/e2e/pr-e2e-retry-receipt.mts";
 
 const REPOSITORY = "NVIDIA/NemoClaw";
 const WORKFLOW_SHA = "d".repeat(40);
@@ -62,10 +63,50 @@ function workflowRun(runId = 23, overrides: Record<string, unknown> = {}) {
   };
 }
 
+function unrelatedWorkflowRun(runId: number) {
+  return workflowRun(runId, {
+    name: `E2E unrelated run ${runId}`,
+    display_title: `E2E unrelated run ${runId}`,
+  });
+}
+
 function inventory(runs: unknown[]) {
   return { total_count: runs.length, workflow_runs: runs };
 }
 
+function oldReceipt(
+  overrides: Partial<DispatchNotObservedReceipt> = {},
+): DispatchNotObservedReceipt {
+  return {
+    correlationId: CORRELATION_ID,
+    workflowSha: WORKFLOW_SHA,
+    sentAtMs: SENT_AT_MS,
+    deadlineAtMs: SENT_AT_MS + WINDOW_MS,
+    result: "not-observed",
+    failureKind: "transport",
+    ...overrides,
+  };
+}
+
+function oldReceiptOptions(overrides: Partial<DispatchNotObservedReceipt> = {}) {
+  return {
+    repository: REPOSITORY,
+    token: "token",
+    prNumber: 42,
+    receipt: oldReceipt(overrides),
+  };
+}
+
+function paginatedInventoryApi(runs: unknown[]) {
+  return vi.fn(async (apiPath: string) => {
+    const page = Number(new URL(`https://api.github.com/${apiPath}`).searchParams.get("page"));
+    return {
+      total_count: runs.length,
+      workflow_runs: runs.slice((page - 1) * 100, page * 100),
+    };
+  });
+}
+
 function reconciliationDeps(
   list: () => unknown | Promise<unknown>,
   getRun: (runId: number) => unknown | Promise<unknown> = (runId) => workflowRun(runId),
@@ -366,24 +407,98 @@ describe("PR E2E workflow dispatch reconciliation", () => {
     const { deps, api } = reconciliationDeps(() => inventory([]));
 
     await expect(
-      assertDispatchStillNotObserved(
-        {
-          repository: REPOSITORY,
-          token: "token",
-          prNumber: 42,
-          receipt: {
-            correlationId: CORRELATION_ID,
-            workflowSha: WORKFLOW_SHA,
-            sentAtMs: SENT_AT_MS,
-            deadlineAtMs: SENT_AT_MS + WINDOW_MS,
-            result: "not-observed",
-            failureKind: "http",
-            status: 500,
-          },
-        },
-        deps,
-      ),
+      assertDispatchStillNotObserved(oldReceiptOptions({ failureKind: "http", status: 500 }), deps),
+    ).resolves.toBeUndefined();
+    expect(api).toHaveBeenCalledOnce();
+  });
+
+  it("paginates a stale zero-match receipt before allowing replacement", async () => {
+    const runs = Array.from({ length: 101 }, (_value, index) => unrelatedWorkflowRun(index + 100));
+    const api = paginatedInventoryApi(runs);
+
+    await expect(
+      assertDispatchStillNotObserved(oldReceiptOptions(), {
+        api,
+        now: () => SENT_AT_MS + WINDOW_MS + 2_000,
+        clockSkewMs: 10,
+      }),
     ).resolves.toBeUndefined();
+    expect(api).toHaveBeenCalledTimes(2);
+  });
+
+  it("blocks replacement when the old correlation appears after the first inventory page", async () => {
+    const recheckTime = SENT_AT_MS + WINDOW_MS + 2_000;
+    const runs = [
+      ...Array.from({ length: 100 }, (_value, index) => unrelatedWorkflowRun(index + 100)),
+      workflowRun(501, {
+        created_at: new Date(SENT_AT_MS + WINDOW_MS + 1_000).toISOString(),
+      }),
+    ];
+    const api = paginatedInventoryApi(runs);
+
+    await expect(
+      assertDispatchStillNotObserved(oldReceiptOptions(), {
+        api,
+        now: () => recheckTime,
+        clockSkewMs: 10,
+      }),
+    ).rejects.toMatchObject({
+      name: "DispatchReconciliationError",
+      candidateRunIds: [501],
+    });
+    expect(api).toHaveBeenCalledTimes(2);
+  });
+
+  it.each([
+    {
+      label: "the total count changes",
+      secondPage: {
+        total_count: 102,
+        workflow_runs: [unrelatedWorkflowRun(500)],
+      },
+    },
+    {
+      label: "a run ID is duplicated",
+      secondPage: {
+        total_count: 101,
+        workflow_runs: [unrelatedWorkflowRun(100)],
+      },
+    },
+  ])("fails closed when paginated receipt inventory $label", async ({ secondPage }) => {
+    const firstPageRuns = Array.from({ length: 100 }, (_value, index) =>
+      unrelatedWorkflowRun(index + 100),
+    );
+    const pages = [{ total_count: 101, workflow_runs: firstPageRuns }, secondPage];
+    const api = vi.fn(async (apiPath: string) => {
+      const page = Number(new URL(`https://api.github.com/${apiPath}`).searchParams.get("page"));
+      return pages[page - 1];
+    });
+
+    await expect(
+      assertDispatchStillNotObserved(oldReceiptOptions(), {
+        api,
+        now: () => SENT_AT_MS + WINDOW_MS + 2_000,
+        clockSkewMs: 10,
+      }),
+    ).rejects.toThrow(/could not be rechecked/u);
+    expect(api).toHaveBeenCalledTimes(2);
+  });
+
+  it("fails closed when a stale receipt inventory exceeds GitHub's filtered-result cap", async () => {
+    const api = vi.fn().mockResolvedValue({
+      total_count: 1_001,
+      workflow_runs: Array.from({ length: 100 }, (_value, index) =>
+        unrelatedWorkflowRun(index + 100),
+      ),
+    });
+
+    await expect(
+      assertDispatchStillNotObserved(oldReceiptOptions(), {
+        api,
+        now: () => SENT_AT_MS + WINDOW_MS + 2_000,
+        clockSkewMs: 10,
+      }),
+    ).rejects.toThrow(/could not be rechecked/u);
     expect(api).toHaveBeenCalledOnce();
   });
 
@@ -395,22 +510,7 @@ describe("PR E2E workflow dispatch reconciliation", () => {
     const { deps, api } = reconciliationDeps(() => inventory([lateRun]));
 
     await expect(
-      assertDispatchStillNotObserved(
-        {
-          repository: REPOSITORY,
-          token: "token",
-          prNumber: 42,
-          receipt: {
-            correlationId: CORRELATION_ID,
-            workflowSha: WORKFLOW_SHA,
-            sentAtMs: SENT_AT_MS,
-            deadlineAtMs: SENT_AT_MS + WINDOW_MS,
-            result: "not-observed",
-            failureKind: "transport",
-          },
-        },
-        { ...deps, now: () => recheckTime },
-      ),
+      assertDispatchStillNotObserved(oldReceiptOptions(), { ...deps, now: () => recheckTime }),
     ).rejects.toMatchObject({
       name: "DispatchReconciliationError",
       candidateRunIds: [23],

tools/advisors/github.mts

@@ -25,6 +25,10 @@ export type GitHubApiFailureKind = "http" | "decode";
 const MAX_GITHUB_RESPONSE_EXCERPT_CHARS = 512;
 const GITHUB_REQUEST_ID_PATTERN = /^[A-Za-z0-9:-]{1,128}$/u;
 
+export function isValidGithubRequestId(value: unknown): value is string {
+  return typeof value === "string" && GITHUB_REQUEST_ID_PATTERN.test(value);
+}
+
 function responseExcerpt(text: string): string {
   const singleLine = text
     .replace(/[\r\n\t]+/gu, " ")
@@ -41,7 +45,7 @@ function responseRequestId(response: Response): string | undefined {
     headers && typeof headers.get === "function"
       ? headers.get("x-github-request-id")?.trim()
       : undefined;
-  return requestId && GITHUB_REQUEST_ID_PATTERN.test(requestId) ? requestId : undefined;
+  return isValidGithubRequestId(requestId) ? requestId : undefined;
 }
 
 export class GitHubApiError extends Error {

tools/e2e/pr-e2e-dispatch-reconciliation.mts

@@ -11,6 +11,7 @@ import {
   type DispatchFailureKind,
   type DispatchNotObservedReceipt,
   dispatchNotObservedReceiptMarker,
+  MAX_DISPATCH_RECONCILIATION_WINDOW_MS,
 } from "./pr-e2e-retry-receipt.mts";
 
 const E2E_WORKFLOW = "e2e.yaml";
@@ -21,11 +22,11 @@ const DEFAULT_POLL_INTERVAL_MS = 2_000;
 const DEFAULT_CLOCK_SKEW_MS = 10_000;
 const DEFAULT_DISPATCH_TIMEOUT_MS = 10_000;
 const DEFAULT_API_TIMEOUT_MS = 5_000;
-const MAX_RECONCILIATION_WINDOW_MS = 120_000;
 const MAX_DISPATCH_TIMEOUT_MS = 30_000;
 const MAX_API_TIMEOUT_MS = 10_000;
 const MAX_DISPATCH_AND_RECONCILIATION_BUDGET_MS = 65_000;
 const MAX_WORKFLOW_RUNS = 100;
+const MAX_WORKFLOW_RUN_PAGES = 10;
 const SHA_PATTERN = /^[a-f0-9]{40}$/u;
 const CORRELATION_PATTERN =
   /^[a-f0-9]{8}-[a-f0-9]{4}-4[a-f0-9]{3}-[89ab][a-f0-9]{3}-[a-f0-9]{12}$/u;
@@ -184,7 +185,7 @@ function assertTimingOptions(options: {
   if (
     !positiveSafeInteger(options.sentAtMs) ||
     !positiveSafeInteger(options.reconciliationWindowMs) ||
-    options.reconciliationWindowMs > MAX_RECONCILIATION_WINDOW_MS ||
+    options.reconciliationWindowMs > MAX_DISPATCH_RECONCILIATION_WINDOW_MS ||
     !positiveSafeInteger(options.pollIntervalMs) ||
     options.pollIntervalMs > options.reconciliationWindowMs ||
     !Number.isSafeInteger(options.clockSkewMs) ||
@@ -265,14 +266,14 @@ function validateWorkflowRun(value: unknown): WorkflowRun {
   };
 }
 
-function validateWorkflowRunInventory(value: unknown): WorkflowRunInventory {
+function validateWorkflowRunInventoryPage(value: unknown): WorkflowRunInventory {
   if (
     !isObjectRecord(value) ||
     !Number.isSafeInteger(value.total_count) ||
     (value.total_count as number) < 0 ||
     !Array.isArray(value.workflow_runs) ||
     value.workflow_runs.length > MAX_WORKFLOW_RUNS ||
-    value.workflow_runs.length !== value.total_count
+    value.workflow_runs.length > (value.total_count as number)
   ) {
     throw new Error("GitHub returned an invalid or incomplete workflow run listing");
   }
@@ -283,11 +284,20 @@ function validateWorkflowRunInventory(value: unknown): WorkflowRunInventory {
   return { totalCount: value.total_count as number, runs };
 }
 
+function validateWorkflowRunInventory(value: unknown): WorkflowRunInventory {
+  const inventory = validateWorkflowRunInventoryPage(value);
+  if (inventory.runs.length !== inventory.totalCount) {
+    throw new Error("GitHub returned an invalid or incomplete workflow run listing");
+  }
+  return inventory;
+}
+
 function inventoryApiPath(options: {
   repository: string;
   workflowSha: string;
   lowerBoundMs: number;
   upperBoundMs: number;
+  page?: number;
 }): string {
   const created = `${new Date(options.lowerBoundMs).toISOString()}..${new Date(options.upperBoundMs).toISOString()}`;
   const query = new URLSearchParams({
@@ -296,10 +306,56 @@ function inventoryApiPath(options: {
     head_sha: options.workflowSha,
     created,
     per_page: String(MAX_WORKFLOW_RUNS),
+    ...(options.page === undefined ? {} : { page: String(options.page) }),
   });
   return `repos/${options.repository}/actions/workflows/${E2E_WORKFLOW}/runs?${query}`;
 }
 
+async function readPaginatedInventory(
+  options: {
+    repository: string;
+    token: string;
+    workflowSha: string;
+    lowerBoundMs: number;
+    upperBoundMs: number;
+  },
+  api: GithubApi,
+  timeoutMs: number,
+): Promise<WorkflowRunInventory> {
+  return boundedOperation("Paginated workflow run inventory read", timeoutMs, async (signal) => {
+    const runs: WorkflowRun[] = [];
+    const runIds = new Set<number>();
+    let totalCount: number | undefined;
+    for (let page = 1; page <= MAX_WORKFLOW_RUN_PAGES; page += 1) {
+      const inventory = validateWorkflowRunInventoryPage(
+        await api(inventoryApiPath({ ...options, page }), options.token, {
+          userAgent: USER_AGENT,
+          signal,
+        }),
+      );
+      totalCount ??= inventory.totalCount;
+      if (
+        inventory.totalCount !== totalCount ||
+        totalCount > MAX_WORKFLOW_RUNS * MAX_WORKFLOW_RUN_PAGES
+      ) {
+        throw new Error("GitHub returned an unstable or oversized workflow run listing");
+      }
+      for (const run of inventory.runs) {
+        if (runIds.has(run.id)) {
+          throw new Error("GitHub returned duplicate workflow run IDs across pages");
+        }
+        runIds.add(run.id);
+        runs.push(run);
+      }
+      if (runs.length === totalCount) return { totalCount, runs };
+      if (runs.length > totalCount || inventory.runs.length < MAX_WORKFLOW_RUNS) {
+        throw new Error("GitHub returned an invalid or incomplete workflow run listing");
+      }
+    }
+    throw new Error("GitHub returned an incomplete workflow run listing after pagination");
+  });
+}
+
 async function readInventory(
   options: {
     repository: string;
@@ -704,7 +760,7 @@ export async function assertDispatchStillNotObserved(
   const correlatedRunIds = new Set<number>();
   let inventory: WorkflowRunInventory;
   try {
-    inventory = await readInventory(
+    inventory = await readPaginatedInventory(
       {
         repository: options.repository,
         token: options.token,