torchci: don't auto-dispatch the AI advisor on closed/draft PRs (#8197)

Follow-up to #8178. The Dr.CI auto-dispatch path fired the AI advisor on
**any** PR with recent NEW failures — including PRs that are
**closed/merged** (won't be worked on) and **draft** (work-in-progress).
Example: it dispatched on
[pytorch/pytorch#187528](pytorch/pytorch#187528)
right as it was closed.

## Fix
Add a PR-state gate in `autoDispatchAdvisorForNewFailures`:
- After dedup, once there's **fresh work** to dispatch, look up the PR
and **skip if it's not open** (closed/merged) or — by default
(`SKIP_DRAFT_PRS = true`) — **draft**.
- The lookup runs **only when there's something to dispatch**
(fully-deduped PRs are never looked up).
- **Fails closed**: a lookup error skips the pass (the next 15-min pass
retries) rather than dispatching on a possibly-closed PR.

## PR state from ClickHouse, not the GitHub API
PR state comes from the **`default.pull_request` CH mirror** via a new
`advisor_pr_state` saved query (`state` + `draft`), not
`octokit.pulls.get`. This is consistent with the rest of the advisor
path (all of its other reads are `queryClickhouseSaved`) and with
`getPRsWithPendingJobInComment`, which already reads
`pull_request.state` from CH — and it keeps the Dr.CI cron off the
GitHub rate limit. The mirror lags GitHub by ~1 min (verified: #187528
closed at 22:04:54Z, CH `updated_at` 22:05:58Z), well within the 15-min
cron cadence. A not-yet-mirrored PR (no row) is treated as **open** so
brand-new PRs aren't dropped.

## Draft PRs
`SKIP_DRAFT_PRS` defaults to **true** (skip drafts — WIP, tend to
churn). One-line flip if advisor feedback on drafts is desirable.

## Test plan
- `tsc --noEmit` clean; local lintrunner clean (incl. SQLFLUFF on the
new saved query).
- Verified the saved query live against CH: `{state: "closed", draft:
false}` for #187528.
- Unit tests: skips a closed PR (no dispatch, no marker write), skips a
draft PR, and does **not** look up PR state when every failure is
already deduped.

Co-authored-by: Ivan <izaitsevfb@meta.com>

Author: izaitsevfb

torchci/clickhouse_queries/advisor_pr_state/params.json

@@ -0,0 +1,7 @@
+{
+  "params": {
+    "prNumber": "Int64",
+    "htmlUrl": "String"
+  },
+  "tests": []
+}

torchci/clickhouse_queries/advisor_pr_state/query.sql

@@ -0,0 +1,12 @@
+-- Latest PR state (open/closed) + draft flag for a single PR, used to gate AI
+-- advisor auto-dispatch (skip closed/merged or draft PRs). Reads the
+-- default.pull_request mirror instead of the GitHub API. Filters on `number`
+-- (the table's primary/sorting key) for an indexed lookup; html_url pins the
+-- repo since PR numbers are not unique across repos.
+SELECT
+    state,
+    draft
+FROM default.pull_request FINAL
+WHERE
+    number = {prNumber: Int64}
+    AND html_url = {htmlUrl: String}

torchci/lib/advisor/advisorDispatch.ts

@@ -371,18 +371,56 @@ export async function recordDispatch(record: DispatchRecord): Promise<void> {
   });
 }
 
+// Auto-dispatch is skipped on draft PRs (work-in-progress, not ready for
+// review). Flip to false to also analyze drafts.
+const SKIP_DRAFT_PRS = true;
+
+/**
+ * Fetch the minimal PR state needed to gate auto-dispatch, from the
+ * default.pull_request ClickHouse mirror rather than the GitHub API. The rest
+ * of the advisor path already reads from CH (and getPRsWithPendingJobInComment
+ * in drci.ts already reads pull_request.state the same way), so this keeps the
+ * Dr.CI cron off the GitHub rate limit. The mirror lags GitHub by ~1 minute,
+ * well within the 15-minute cron cadence. A PR not yet mirrored (no row) is
+ * treated as open so we don't drop dispatches on brand-new PRs.
+ */
+export async function getPullRequestMeta(
+  owner: string,
+  repo: string,
+  prNumber: number
+): Promise<{ state: string; draft: boolean }> {
+  const rows = await queryClickhouseSaved("advisor_pr_state", {
+    prNumber,
+    htmlUrl: `https://github.com/${owner}/${repo}/pull/${prNumber}`,
+  });
+  if (rows.length === 0) return { state: "open", draft: false };
+  // Decode draft robustly: depending on the CH client/format a Bool can come
+  // back as a JS boolean, a number, or a string ("0"/"1"/"true"/"false"). A
+  // naive Boolean() would treat the string "0"/"false" as truthy and wrongly
+  // mark every non-draft PR as draft (suppressing all open PRs).
+  const draftRaw = rows[0].draft;
+  const draft =
+    draftRaw === true ||
+    draftRaw === 1 ||
+    draftRaw === "1" ||
+    draftRaw === "true";
+  return { state: rows[0].state as string, draft };
+}
+
 // Injectable dependencies so the orchestration logic is unit-testable without
 // touching ClickHouse or GitHub.
 export interface AutoDispatchDeps {
   readDispatchStates: typeof readDispatchStates;
   recordDispatch: typeof recordDispatch;
   dispatchAdvisorWorkflow: typeof dispatchAdvisorWorkflow;
+  getPullRequestMeta: typeof getPullRequestMeta;
 }
 
 const defaultDeps: AutoDispatchDeps = {
   readDispatchStates,
   recordDispatch,
   dispatchAdvisorWorkflow,
+  getPullRequestMeta,
 };
 
 export interface AutoDispatchArgs {
@@ -417,7 +455,9 @@ export function autoDispatchEnabled(owner: string, repo: string): boolean {
  *   - 'dispatched' is written on success; a 'failed' row supersedes the pre row
  *     when the dispatch throws, re-enabling retry up to MAX_DISPATCH_RETRIES,
  *   - bails entirely (dispatches nothing) if a PR has more NEW failures than the
- *     per-repo max (outage guard).
+ *     per-repo max (outage guard),
+ *   - skips PRs that are not open (closed/merged) or, by default, draft -- the
+ *     PR state is only looked up once there is fresh work to dispatch.
  */
 export async function autoDispatchAdvisorForNewFailures(
   args: AutoDispatchArgs,
@@ -471,8 +511,14 @@ export async function autoDispatchAdvisorForNewFailures(
     return;
   }
 
-  // The byKey set is already bounded by the outage guard above (<= the per-repo
-  // max), so dispatch every fresh failure; dedup state skips repeats.
+  // Which failures still need a dispatch (skip already dispatching/dispatched,
+  // or failed-and-out-of-retries). The byKey set is already bounded by the
+  // outage guard above.
+  const toDispatch: {
+    signalKey: string;
+    job: RecentWorkflowsData;
+    retryCount: number;
+  }[] = [];
   for (const [signalKey, job] of byKey) {
     const prev = states.get(signalKey);
     if (prev) {
@@ -484,7 +530,35 @@ export async function autoDispatchAdvisorForNewFailures(
       }
     }
     const retryCount = prev?.state === "failed" ? prev.retryCount + 1 : 0;
+    toDispatch.push({ signalKey, job, retryCount });
+  }
+  if (toDispatch.length === 0) return;
+
+  // Don't auto-dispatch on PRs that aren't open: a closed/merged PR won't be
+  // worked on, and (by default) draft PRs are work-in-progress. Looked up only
+  // now -- after dedup -- so the PR lookup is paid only when there is fresh
+  // work. Fail closed: a lookup error skips this pass (the next pass retries).
+  try {
+    const pr = await deps.getPullRequestMeta(owner, repo, prNumber);
+    if (pr.state !== "open") {
+      console.log(
+        `advisor auto-dispatch: PR ${prNumber} is ${pr.state}; skipping`
+      );
+      return;
+    }
+    if (pr.draft && SKIP_DRAFT_PRS) {
+      console.log(`advisor auto-dispatch: PR ${prNumber} is a draft; skipping`);
+      return;
+    }
+  } catch (e) {
+    console.error(
+      `advisor auto-dispatch: PR-state lookup failed for PR ${prNumber}, skipping pass`,
+      e
+    );
+    return;
+  }
 
+  for (const { signalKey, job, retryCount } of toDispatch) {
     const base = {
       owner,
       repo,

torchci/test/advisorDispatch.test.ts

@@ -26,6 +26,9 @@ function makeDeps(overrides: Partial<AutoDispatchDeps> = {}): AutoDispatchDeps {
     readDispatchStates: jest.fn().mockResolvedValue(new Map()),
     recordDispatch: jest.fn().mockResolvedValue(undefined),
     dispatchAdvisorWorkflow: jest.fn().mockResolvedValue(undefined),
+    getPullRequestMeta: jest
+      .fn()
+      .mockResolvedValue({ state: "open", draft: false }),
     ...overrides,
   };
 }
@@ -242,4 +245,50 @@ describe("autoDispatchAdvisorForNewFailures", () => {
       DEFAULT_MAX_NEW_FAILURES
     );
   });
+
+  it("does not dispatch on a closed PR", async () => {
+    const deps = makeDeps({
+      getPullRequestMeta: jest
+        .fn()
+        .mockResolvedValue({ state: "closed", draft: false }),
+    });
+    await autoDispatchAdvisorForNewFailures(
+      { ...baseArgs, newFailures: [job("wf / a")] },
+      deps
+    );
+    expect(deps.dispatchAdvisorWorkflow).not.toHaveBeenCalled();
+    // Skipped after dedup but before any marker write.
+    expect(deps.recordDispatch).not.toHaveBeenCalled();
+  });
+
+  it("does not dispatch on a draft PR", async () => {
+    const deps = makeDeps({
+      getPullRequestMeta: jest
+        .fn()
+        .mockResolvedValue({ state: "open", draft: true }),
+    });
+    await autoDispatchAdvisorForNewFailures(
+      { ...baseArgs, newFailures: [job("wf / a")] },
+      deps
+    );
+    expect(deps.dispatchAdvisorWorkflow).not.toHaveBeenCalled();
+  });
+
+  it("does not look up PR state when every failure is already deduped", async () => {
+    const states = new Map([
+      [
+        signalKeyForJob("wf / a"),
+        { state: "dispatched" as const, retryCount: 0 },
+      ],
+    ]);
+    const deps = makeDeps({
+      readDispatchStates: jest.fn().mockResolvedValue(states),
+    });
+    await autoDispatchAdvisorForNewFailures(
+      { ...baseArgs, newFailures: [job("wf / a")] },
+      deps
+    );
+    expect(deps.getPullRequestMeta).not.toHaveBeenCalled();
+    expect(deps.dispatchAdvisorWorkflow).not.toHaveBeenCalled();
+  });
 });