[CRCR] Implement on-call bot and allowlist functionality for downstream CI failures (#8183)

## Summary

Implement CRCR on-call bot and merge-blocking logic for downstream CI
failures, driven by a structured allowlist (L1-L4) that maps downstream
repos to severity levels and on-call contacts.

## Changes

### CRCR Allowlist (`lib/crcrAllowlist.ts`)
Parses `.github/allowlist.yml` from `pytorch/pytorch`, mapping
downstream repos to four levels:
- **L1/L2**: Tracked only, no on-call behavior.
- **L3**: Non-blocking failure — on-call is notified but merge is not
blocked.
- **L4**: Blocking failure — on-call is notified and merge is blocked.

Includes a 15-minute in-memory cache and strict YAML validation. Errors
fail open.

### On-call Bot (`lib/bot/crcrOncallBot.ts`)
A Probot bot on `check_run.completed`: when a CRCR check run fails,
looks up on-calls from the allowlist and posts a tagged comment on the
associated PR. Deduplicates via an HTML comment marker (theoretical
TOCTOU race exists but is negligible in practice since checks arrive
sequentially).

### Merge Blocking (`lib/bot/pytorchBotHandler.ts`)
`@pytorchbot merge` checks the GitHub Checks API for CRCR failures on
the PR head commit. L4 failures block the merge with a comment listing
failing repos. `-f` bypasses this, consistent with existing force-merge
semantics. Errors fail open.
- Introduced `ensureHeadSha()` helper to deduplicate the lazy-load
pattern for `headSha` across methods.

### Dr.CI Integration
- `fetchRecentWorkflows.ts`: New `fetchOotWorkflows()` queries
`oot_workflow_job` from ClickHouse. Uses a new `downstreamLevel` field
on `RecentWorkflowsData` instead of overloading `failure_context`.
- `drci.ts`: L4 failures appear as blocking; L3 failures appear in a new
"OUT OF TREE (non-blocking)" section.

### Tests
Unit tests for allowlist parsing and on-call bot behavior, plus updated
Dr.CI test helpers.

Author: KarhouTam

torchci/lib/bot/crcrOncallBot.ts

@@ -0,0 +1,195 @@
+import { Probot } from "probot";
+import { fetchCrcrAllowlist } from "../crcrAllowlist";
+import { isPyTorchbotSupportedOrg } from "./utils";
+
+export const FAILURE_CONCLUSIONS: ReadonlySet<string> = new Set([
+  "failure",
+  "cancelled",
+  "timed_out",
+]);
+
+function markerForRepo(downstreamRepo: string): string {
+  return `<!-- crcr-oncall:${downstreamRepo} -->`;
+}
+
+/**
+ * Parse the downstream ``owner/repo`` out of a CRCR check run name.
+ *
+ * Check runs are named ``crcr/<owner>/<repo>/<workflow_name>/<job_name>``
+ * (see the Python gh_helper.check_run_name), so the downstream repo is the
+ * first two path segments after the ``crcr/`` prefix.
+ */
+export function downstreamRepoFromCheckRunName(name: string): string | null {
+  const prefix = "crcr/";
+  if (!name.startsWith(prefix)) {
+    return null;
+  }
+  const parts = name.slice(prefix.length).split("/");
+  if (parts.length < 4 || !parts[0] || !parts[1]) {
+    return null;
+  }
+  return `${parts[0]}/${parts[1]}`;
+}
+
+/**
+ * Search existing PR comments for one with the CRCR on-call marker.
+ * Returns its ID (or 0 if none found).  Uses pagination so the marker
+ * is found even when a PR has more than 30 comments.
+ */
+async function findExistingComment(
+  octokit: any,
+  owner: string,
+  repo: string,
+  prNumber: number,
+  downstreamRepo: string
+): Promise<number> {
+  const marker = markerForRepo(downstreamRepo);
+  const comments = await octokit.paginate(octokit.rest.issues.listComments, {
+    owner,
+    repo,
+    issue_number: prNumber,
+  });
+  for (const comment of comments) {
+    if (comment.body?.includes(marker)) {
+      return comment.id;
+    }
+  }
+  return 0;
+}
+
+export default function crcrOncallBot(app: Probot): void {
+  app.on("check_run.completed", async (ctx) => {
+    const owner = ctx.payload.repository.owner.login;
+    if (!isPyTorchbotSupportedOrg(owner)) {
+      return;
+    }
+
+    const repo = ctx.payload.repository.name;
+    const checkRun = ctx.payload.check_run;
+
+    // Only act on CRCR-created check runs
+    const downstreamRepo = downstreamRepoFromCheckRunName(checkRun.name);
+    if (!downstreamRepo) {
+      return;
+    }
+
+    // Only comment on failures
+    const conclusion = checkRun.conclusion ?? "";
+    if (!FAILURE_CONCLUSIONS.has(conclusion)) {
+      return;
+    }
+
+    // Get the PRs this check run belongs to.
+    // checkRun.pull_requests is empty for cross-fork PRs (most pytorch
+    // contributions), so fall back to the commits API to resolve PRs
+    // from the head SHA — the same strategy used by the merge-blocking path.
+    let prNumbers: number[] = [];
+    if (checkRun.pull_requests && checkRun.pull_requests.length > 0) {
+      prNumbers = checkRun.pull_requests.map((pr) => pr.number);
+    } else if (checkRun.head_sha) {
+      try {
+        const result =
+          await ctx.octokit.rest.repos.listPullRequestsAssociatedWithCommit({
+            owner,
+            repo,
+            commit_sha: checkRun.head_sha,
+          });
+        prNumbers = result.data.map((pr: any) => pr.number);
+      } catch (err) {
+        ctx.log(
+          { err },
+          `crcrOncall: failed to resolve PRs for commit ${checkRun.head_sha}, skipping`
+        );
+        return;
+      }
+    }
+
+    if (prNumbers.length === 0) {
+      ctx.log(
+        `crcrOncall: no PR associated with check run ${checkRun.name}, skipping`
+      );
+      return;
+    }
+
+    const headSha = checkRun.head_sha;
+    const checkRunUrl = checkRun.html_url ?? "";
+
+    // Load oncalls from the allowlist
+    let oncalls: string[];
+    try {
+      const allowlist = await fetchCrcrAllowlist(ctx.octokit);
+      oncalls = allowlist.getOncallsForRepo(downstreamRepo);
+    } catch (err) {
+      ctx.log({ err }, "crcrOncall: failed to load allowlist, skipping");
+      return;
+    }
+
+    if (oncalls.length === 0) {
+      ctx.log(
+        `crcrOncall: no oncalls configured for ${downstreamRepo}, skipping`
+      );
+      return;
+    }
+
+    const mentions = oncalls.map((o) => `@${o}`).join(" ");
+
+    // Post a comment on each associated PR (typically just one)
+    for (const prNumber of prNumbers) {
+      try {
+        // Dedup by marker: only comment once per PR.
+        // NOTE: There is a theoretical TOCTOU race here — two concurrent
+        // check_run.completed events for the same PR could both pass the
+        // marker check and post duplicate comments. In practice, checks
+        // arrive sequentially, so this is unlikely to cause issues.
+        const existingId = await findExistingComment(
+          ctx.octokit,
+          owner,
+          repo,
+          prNumber,
+          downstreamRepo
+        );
+        if (existingId !== 0) {
+          ctx.log(
+            `crcrOncall: comment already exists on PR #${prNumber}, skipping`
+          );
+          continue;
+        }
+
+        const commentBody = `${markerForRepo(downstreamRepo)}
+## :x: CRCR downstream CI failure
+
+The downstream CI workflow in **${downstreamRepo}** has failed on commit \`${headSha.slice(
+          0,
+          7
+        )}\`.
+
+${mentions} please investigate.
+
+### Details
+
+| Field | Value |
+|---|---|
+| **Check Run** | [${checkRun.name}](${checkRunUrl}) |
+| **Commit** | \`${headSha}\` |
+| **Conclusion** | \`${conclusion}\` |
+`;
+
+        await ctx.octokit.issues.createComment({
+          owner,
+          repo,
+          issue_number: prNumber,
+          body: commentBody,
+        });
+
+        ctx.log(
+          `crcrOncall: commented on PR #${prNumber} for ${downstreamRepo} (${conclusion})`
+        );
+      } catch (err) {
+        ctx.log(
+          { err },
+          `crcrOncall: failed to comment on PR for ${downstreamRepo}`
+        );
+      }
+    }
+  });
+}

torchci/lib/bot/index.ts

@@ -6,6 +6,7 @@ import cancelWorkflowsOnCloseBot from "./cancelWorkflowsOnCloseBot";
 import checkLabelsBot from "./checkLabelsBot";
 import ciflowPushTrigger from "./ciflowPushTrigger";
 import codevNoWritePerm from "./codevNoWritePermBot";
+import crcrOncallBot from "./crcrOncallBot";
 import drciBot from "./drciBot";
 import nitpickBot from "./nitpickBot";
 import pytorchBot from "./pytorchBot";
@@ -22,6 +23,7 @@ export default function bot(app: Probot) {
   checkLabelsBot(app);
   ciflowPushTrigger(app);
   codevNoWritePerm(app);
+  crcrOncallBot(app);
   drciBot(app);
   nitpickBot(app);
   pytorchBot(app);

torchci/lib/bot/pytorchBotHandler.ts

@@ -3,8 +3,10 @@ import _ from "lodash";
 import { updateDrciComments } from "pages/api/drci/drci";
 import shlex from "shlex";
 import { queryClickhouseSaved } from "../clickhouse";
+import { fetchCrcrAllowlist } from "../crcrAllowlist";
 import { getHelp, getParser } from "./cliParser";
 import { cherryPickClassifications } from "./Constants";
+import { downstreamRepoFromCheckRunName } from "./crcrOncallBot";
 import PytorchBotLogger from "./pytorchbotLogger";
 import {
   hasWritePermissions as _hasWP,
@@ -319,7 +321,23 @@ The explanation needs to be clear on why this is needed. Here are some good exam
     }
 
     await this.logger.log("merge", extra_data);
+
+    // Check for L4 CRCR blocking failures (L3 failures are non-blocking).
+    // Force merge (-f) bypasses this check, consistent with the existing
+    // force-merge semantics for in-repo CI failures.
+    // Only applies to pytorch/pytorch — the CRCR allowlist and check runs
+    // are specific to that repo.
     if (!forceRequested && isPyTorchPyTorch(this.owner, this.repo)) {
+      const blockingRepos = await this.getCrcrBlockingFailures();
+      if (blockingRepos.length > 0) {
+        await this.addComment(
+          `The following L4 downstream CI workflows are blocking this merge (failed or still running):\n\n` +
+            blockingRepos.map((r) => `- \`${r}\``).join("\n") +
+            `\n\nPlease investigate or use \`@pytorchbot merge -f\` to bypass.`
+        );
+        return;
+      }
+
       let labels: string[] = this.ctx.payload?.issue?.labels.map(
         (e: any) => e["name"]
       );
@@ -401,10 +419,8 @@ The explanation needs to be clear on why this is needed. Here are some good exam
     );
   }
 
-  async hasWorkflowRunningPermissions(username: string): Promise<boolean> {
-    if (await _hasWP(this.ctx, username)) {
-      return true;
-    }
+  /** Lazy-load and cache the PR head SHA, then return it. */
+  private async ensureHeadSha(): Promise<string> {
     if (this.headSha === undefined) {
       const pullRequest = await this.ctx.octokit.pulls.get({
         owner: this.owner,
@@ -413,12 +429,19 @@ The explanation needs to be clear on why this is needed. Here are some good exam
       });
       this.headSha = pullRequest.data.head.sha;
     }
+    return this.headSha!;
+  }
+
+  async hasWorkflowRunningPermissions(username: string): Promise<boolean> {
+    if (await _hasWP(this.ctx, username)) {
+      return true;
+    }
 
     return await hasApprovedPullRuns(
       this.ctx.octokit,
       this.ctx.payload.repository.owner.login,
       this.ctx.payload.repository.name,
-      this.headSha!
+      await this.ensureHeadSha()
     );
   }
 
@@ -635,6 +658,81 @@ The explanation needs to be clear on why this is needed. Here are some good exam
       headSha: this.headSha,
     });
   }
+
+  /**
+   * Return the list of L4 downstream repos whose CRCR check runs are failing
+   * or still pending on this PR's head commit.  L3 workflows are intentionally
+   * omitted — they are non-blocking.
+   *
+   * A pending (not-yet-completed) L4 check run also blocks merge: a still-
+   * running check could fail, so merging before it completes would bypass
+   * the downstream gating.  Use ``@pytorchbot merge -f`` to override.
+   */
+  async getCrcrBlockingFailures(): Promise<string[]> {
+    // Only "failure" blocks merge — "cancelled" and "timed_out" are often
+    // superseded / infra-related and should not gate merge.
+    const BLOCKING_CONCLUSIONS = new Set(["failure"]);
+
+    // Query GitHub Check Runs API for all check runs on this commit.
+    // Use paginate to handle PRs with more than 100 check runs.
+    let checkRuns: any[] = [];
+    try {
+      const headSha = await this.ensureHeadSha();
+      checkRuns = await this.ctx.octokit.paginate(
+        this.ctx.octokit.checks.listForRef,
+        {
+          owner: this.owner,
+          repo: this.repo,
+          ref: headSha,
+          filter: "latest",
+          per_page: 100,
+        }
+      );
+    } catch {
+      // If we can't fetch check runs, fail open (don't block merge on
+      // an infrastructure error, including transient ensureHeadSha failure)
+      this.ctx.log("getCrcrBlockingFailures: failed to list check runs");
+      return [];
+    }
+
+    // Find all CRCR check runs (not just failures — we also need to
+    // gate on pending L4 checks so a still-running check isn't bypassed).
+    const crcrCheckRuns = checkRuns.filter((cr: any) =>
+      cr.name.startsWith("crcr/")
+    );
+
+    if (crcrCheckRuns.length === 0) {
+      return [];
+    }
+
+    // Load the allowlist to classify each repo as L3 or L4
+    let allowlist;
+    try {
+      allowlist = await fetchCrcrAllowlist(this.ctx.octokit);
+    } catch {
+      this.ctx.log("getCrcrBlockingFailures: failed to load allowlist");
+      return [];
+    }
+
+    const blocking = new Set<string>();
+    for (const cr of crcrCheckRuns) {
+      const downstreamRepo = downstreamRepoFromCheckRunName(cr.name);
+      if (!downstreamRepo || !allowlist.isBlocking(downstreamRepo)) {
+        continue; // Not L4
+      }
+
+      // Block when the L4 check has failed OR is still pending.
+      // Successful/completed L4 checks don't block.
+      const isFailure = BLOCKING_CONCLUSIONS.has(cr.conclusion ?? "");
+      const isPending = cr.status !== "completed";
+
+      if (isFailure || isPending) {
+        blocking.add(downstreamRepo);
+      }
+    }
+
+    return [...blocking];
+  }
 }
 
 export default PytorchBotHandler;