From 7949fd224dca154d6315fd446abc810b09214981 Mon Sep 17 00:00:00 2001
From: Kenneth Knowles <klk@google.com>
Date: Wed, 2 Sep 2026 18:07:54 +0000
Subject: [PATCH] Add experimental git history reviewer assigner

Introduce an experimental reviewer assigner that analyzes git commit
history and file churn to recommend domain-expert reviewers:

1. Feature flag: Keep automated assignment behind a feature flag that
   is disabled by default (ENABLE_LLM_REVIEW_ASSIGNER=false), leaving
   existing PR bot behavior unchanged.
2. Vertex AI integration: Authenticate using gcloud application-default
   credentials and Vertex AI, removing any reliance on API keys.
3. Interactive CLI tool: Provide a standalone command-line tool
   (npm run review-advisor) to inspect PRs or local files offline with
   optional recency-decayed familiarity heuristics.
4. PR bot command: Support explicit reviewer assignment via the command
   phrase "assign based on git history".
---
 scripts/ci/pr-bot/Commands.md                 |   3 +-
 scripts/ci/pr-bot/cli.ts                      | 319 +++++++++
 scripts/ci/pr-bot/package.json                |   4 +-
 scripts/ci/pr-bot/processNewPrs.ts            | 108 +++-
 scripts/ci/pr-bot/shared/commentStrings.ts    |  37 +-
 .../ci/pr-bot/shared/geminiReviewerAdvisor.ts | 605 ++++++++++++++++++
 scripts/ci/pr-bot/shared/gitHistory.ts        | 462 +++++++++++++
 scripts/ci/pr-bot/shared/pr.ts                |   5 +
 scripts/ci/pr-bot/shared/reviewerConfig.ts    |  14 +
 scripts/ci/pr-bot/shared/userCommand.ts       | 135 +++-
 scripts/ci/pr-bot/test/commentStringsTest.ts  | 105 +++
 .../pr-bot/test/geminiReviewerAdvisorTest.ts  | 232 +++++++
 scripts/ci/pr-bot/test/gitHistoryTest.ts      | 249 +++++++
 scripts/ci/pr-bot/test/prTest.ts              |   7 +
 14 files changed, 2269 insertions(+), 16 deletions(-)
 create mode 100644 scripts/ci/pr-bot/cli.ts
 create mode 100644 scripts/ci/pr-bot/shared/geminiReviewerAdvisor.ts
 create mode 100644 scripts/ci/pr-bot/shared/gitHistory.ts
 create mode 100644 scripts/ci/pr-bot/test/commentStringsTest.ts
 create mode 100644 scripts/ci/pr-bot/test/geminiReviewerAdvisorTest.ts
 create mode 100644 scripts/ci/pr-bot/test/gitHistoryTest.ts

diff --git a/scripts/ci/pr-bot/Commands.md b/scripts/ci/pr-bot/Commands.md
index 83d21cb7836d..984be635d5bf 100644
--- a/scripts/ci/pr-bot/Commands.md
+++ b/scripts/ci/pr-bot/Commands.md
@@ -29,4 +29,5 @@ All commands are case insensitive.
 | `stop reviewer notifications` | This will disable the bot for the PR. |
 | `remind me after tests pass` | This will comment after all checks complete and tag the person who commented the command. |
 | `waiting on author` | This shifts the attention set to the author. The author can shift the attention set back to the reviewer by commenting anywhere or pushing. |
-| `assign set of reviewers` | If the bot has not yet assigned a set of reviewers to the PR, this command will trigger that happening. |
\ No newline at end of file
+| `assign set of reviewers` | If the bot has not yet assigned a set of reviewers to the PR, this command will trigger that happening. |
+| `assign based on git history` | (Experimental) Analyzes git commit history and file churn for touched files to assign expert reviewers and backups with rationale. |
\ No newline at end of file
diff --git a/scripts/ci/pr-bot/cli.ts b/scripts/ci/pr-bot/cli.ts
new file mode 100644
index 000000000000..3468508cb8f1
--- /dev/null
+++ b/scripts/ci/pr-bot/cli.ts
@@ -0,0 +1,319 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+import * as childProcess from "child_process";
+import * as https from "https";
+import { buildPrHistoryContext } from "./shared/gitHistory";
+import {
+  GeminiReviewerAdvisor,
+  VertexAiClient,
+} from "./shared/geminiReviewerAdvisor";
+import { assignReviewersWithExpertise } from "./shared/commentStrings";
+
+function printHelp() {
+  console.log(`
+Beam Review Advisor CLI (Experimental)
+Analyzes git commit history and file churn to recommend expert reviewers.
+
+USAGE:
+  npm run review-advisor -- [options] [files...]
+  node lib/cli.js [options] [files...]
+
+OPTIONS:
+  --pr <number>         Fetch touched files and details for a GitHub pull request.
+  --local               Inspect uncommitted modified/staged files in local repository.
+  --branch <branch>     Inspect files modified relative to a branch (e.g. master).
+  --heuristic-only      Run offline recency-decayed familiarity scoring without LLM.
+  --project <project>   GCP project ID for Vertex AI (default: gcloud config or apache-beam-testing).
+  --json                Output raw JSON result instead of formatted markdown.
+  --help, -h            Show this help text.
+
+EXAMPLES:
+  # Inspect an open pull request:
+  npm run review-advisor -- --pr 39156
+
+  # Inspect specific files offline:
+  npm run review-advisor -- --heuristic-only sdks/java/io/kafka/src/main/java/.../KafkaIO.java
+
+  # Inspect local uncommitted changes:
+  npm run review-advisor -- --local
+`);
+}
+
+function fetchJson(url: string): Promise<any> {
+  return new Promise((resolve, reject) => {
+    https
+      .get(
+        url,
+        {
+          headers: {
+            "User-Agent": "beam-review-advisor-cli",
+            Accept: "application/vnd.github.v3+json",
+          },
+        },
+        (res) => {
+          let data = "";
+          res.on("data", (chunk) => (data += chunk));
+          res.on("end", () => {
+            try {
+              resolve(JSON.parse(data));
+            } catch (err) {
+              reject(new Error(`Failed to parse response from ${url}: ${err}`));
+            }
+          });
+        }
+      )
+      .on("error", reject);
+  });
+}
+
+async function runCli() {
+  const args = process.argv.slice(2);
+
+  if (args.includes("--help") || args.includes("-h")) {
+    printHelp();
+    return;
+  }
+
+  let prNumber: number | undefined;
+  let useHeuristicOnly = false;
+  let outputJson = false;
+  let localDiff = false;
+  let baseBranch: string | undefined;
+  let customProject: string | undefined;
+  const rawFileArgs: string[] = [];
+
+  for (let i = 0; i < args.length; i++) {
+    const arg = args[i];
+    if (arg === "--pr" && i + 1 < args.length) {
+      prNumber = parseInt(args[++i], 10);
+    } else if (arg === "--heuristic-only" || arg === "--offline") {
+      useHeuristicOnly = true;
+    } else if (arg === "--json") {
+      outputJson = true;
+    } else if (arg === "--local") {
+      localDiff = true;
+    } else if (arg === "--branch" && i + 1 < args.length) {
+      baseBranch = args[++i];
+    } else if (arg === "--project" && i + 1 < args.length) {
+      customProject = args[++i];
+    } else if (!arg.startsWith("--")) {
+      rawFileArgs.push(arg);
+    }
+  }
+
+  let prTitle = "Local Changes";
+  let prBody = "";
+  let prAuthor = "author";
+  let files: Array<{
+    filename: string;
+    additions: number;
+    deletions: number;
+    changes: number;
+    status: string;
+  }> = [];
+
+  if (prNumber) {
+    if (!outputJson) {
+      console.log(`Fetching PR #${prNumber} details from GitHub...`);
+    }
+    const prData = await fetchJson(
+      `https://api.github.com/repos/apache/beam/pulls/${prNumber}`
+    );
+    if (!prData || !prData.title) {
+      throw new Error(
+        `Failed to retrieve PR #${prNumber}: ${JSON.stringify(prData)}`
+      );
+    }
+    prTitle = prData.title;
+    prBody = prData.body || "";
+    prAuthor = prData.user?.login || "author";
+
+    const filesData = await fetchJson(
+      `https://api.github.com/repos/apache/beam/pulls/${prNumber}/files`
+    );
+    if (Array.isArray(filesData)) {
+      files = filesData.map((f: any) => ({
+        filename: f.filename,
+        additions: f.additions || 0,
+        deletions: f.deletions || 0,
+        changes: f.changes || (f.additions || 0) + (f.deletions || 0),
+        status: f.status || "modified",
+      }));
+    }
+  } else if (localDiff) {
+    const diffStat = childProcess
+      .execFileSync("git", ["diff", "--stat", "HEAD"], { encoding: "utf8" })
+      .trim();
+    const nameStatus = childProcess
+      .execFileSync("git", ["diff", "--name-status", "HEAD"], {
+        encoding: "utf8",
+      })
+      .trim();
+
+    const lines = nameStatus.split("\n").filter(Boolean);
+    for (const line of lines) {
+      const parts = line.split(/\s+/);
+      if (parts.length >= 2) {
+        files.push({
+          filename: parts[1],
+          additions: 10,
+          deletions: 5,
+          changes: 15,
+          status: parts[0].startsWith("A") ? "added" : "modified",
+        });
+      }
+    }
+  } else if (baseBranch) {
+    const nameStatus = childProcess
+      .execFileSync("git", ["diff", "--name-status", `${baseBranch}...HEAD`], {
+        encoding: "utf8",
+      })
+      .trim();
+
+    const lines = nameStatus.split("\n").filter(Boolean);
+    for (const line of lines) {
+      const parts = line.split(/\s+/);
+      if (parts.length >= 2) {
+        files.push({
+          filename: parts[1],
+          additions: 10,
+          deletions: 5,
+          changes: 15,
+          status: parts[0].startsWith("A") ? "added" : "modified",
+        });
+      }
+    }
+  } else if (rawFileArgs.length > 0) {
+    files = rawFileArgs.map((f) => ({
+      filename: f,
+      additions: 25,
+      deletions: 5,
+      changes: 30,
+      status: "modified",
+    }));
+  } else {
+    // Default demo mode
+    printHelp();
+    console.log("--- RUNNING SAMPLE DEMO ON KAFKAIO ---\n");
+    files = [
+      {
+        filename:
+          "sdks/java/io/kafka/src/main/java/org/apache/beam/sdk/io/kafka/KafkaIO.java",
+        additions: 80,
+        deletions: 15,
+        changes: 95,
+        status: "modified",
+      },
+    ];
+    prTitle = "KafkaIO: Reader loop stability and dynamic read improvements";
+    prBody =
+      "Optimize dynamic work rebalancing and consumer watermark tracking.";
+    prAuthor = "sampleContributor";
+  }
+
+  if (files.length === 0) {
+    console.log("No changed files detected.");
+    return;
+  }
+
+  if (!outputJson) {
+    console.log(`Analyzing ${files.length} file(s) across git history...`);
+    for (const f of files.slice(0, 5)) {
+      console.log(`  - ${f.filename} (+${f.additions}, -${f.deletions})`);
+    }
+    if (files.length > 5) {
+      console.log(`  ... and ${files.length - 5} more file(s)`);
+    }
+  }
+
+  const prContext = buildPrHistoryContext(
+    prNumber || 0,
+    prTitle,
+    prBody,
+    prAuthor,
+    files
+  );
+
+  const advisor = new GeminiReviewerAdvisor({
+    disableLlm: useHeuristicOnly,
+    vertexAiConfig: customProject ? { project: customProject } : undefined,
+    committerCheck: async (login) =>
+      [
+        "kennknowles",
+        "chamikaramj",
+        "jrmccluskey",
+        "johnjcasey",
+        "damccorm",
+        "ahmedabu98",
+        "abacn",
+      ].includes(login.toLowerCase()),
+  });
+
+  const advice = await advisor.adviseReviewers(prContext);
+
+  if (outputJson) {
+    console.log(
+      JSON.stringify(
+        {
+          prContext: {
+            number: prContext.prNumber,
+            title: prContext.title,
+            author: prContext.author,
+            candidatesCount: prContext.candidates.length,
+          },
+          advice,
+        },
+        null,
+        2
+      )
+    );
+    return;
+  }
+
+  console.log(`\nAdvisor Source: ${advice.source.toUpperCase()}`);
+  console.log(`Candidates in git history: ${prContext.candidates.length}`);
+  console.log("---------------- Selected Reviewers ----------------");
+  for (const reviewer of advice.selectedReviewers) {
+    console.log(
+      `Reviewer: @${
+        reviewer.username
+      } [${reviewer.role.toUpperCase()}] (Committer: ${reviewer.isCommitter})`
+    );
+    console.log(`Expertise: ${reviewer.expertise}`);
+    console.log(`Covered files: ${reviewer.coveredFiles.join(", ")}\n`);
+  }
+
+  if (advice.alternateReviewers.length > 0) {
+    console.log("---------------- Alternate Reviewers ---------------");
+    for (const alt of advice.alternateReviewers) {
+      console.log(`Backup: @${alt.username} — ${alt.expertise}`);
+    }
+    console.log();
+  }
+
+  console.log("Reasoning: " + advice.reasoning);
+  console.log("\n================ Generated GitHub Comment ================\n");
+  console.log(assignReviewersWithExpertise(advice));
+  console.log("==========================================================");
+}
+
+runCli().catch((err) => {
+  console.error("Review Advisor error:", err);
+  process.exit(1);
+});
diff --git a/scripts/ci/pr-bot/package.json b/scripts/ci/pr-bot/package.json
index 5fc8d79c1dd1..2246c79f17a4 100644
--- a/scripts/ci/pr-bot/package.json
+++ b/scripts/ci/pr-bot/package.json
@@ -11,7 +11,9 @@
     "processPrUpdate": "npm run build && node lib/processPrUpdate.js",
     "gatherMetrics": "npm run build && node lib/gatherMetrics.js",
     "updateReviewers": "npm run build && node lib/updateReviewers.js",
-    "findPrsNeedingAttention": "npm run build && node lib/findPrsNeedingAttention.js"
+    "findPrsNeedingAttention": "npm run build && node lib/findPrsNeedingAttention.js",
+    "review-advisor": "npm run build && node lib/cli.js",
+    "dryRun": "npm run build && node lib/cli.js"
   },
   "dependencies": {
     "@actions/exec": "^1.1.0",
diff --git a/scripts/ci/pr-bot/processNewPrs.ts b/scripts/ci/pr-bot/processNewPrs.ts
index dd20d4a98c2b..ca6da0c9a6a2 100644
--- a/scripts/ci/pr-bot/processNewPrs.ts
+++ b/scripts/ci/pr-bot/processNewPrs.ts
@@ -29,6 +29,11 @@ const {
   REVIEWERS_ACTION,
 } = require("./shared/constants");
 import { CheckStatus } from "./shared/checks";
+import { buildPrHistoryContext } from "./shared/gitHistory";
+import {
+  GeminiReviewerAdvisor,
+  ReviewerAdviceResult,
+} from "./shared/geminiReviewerAdvisor";
 
 /*
  * Returns true if the pr needs to be processed or false otherwise.
@@ -167,7 +172,9 @@ async function approvedBy(pull: any): Promise<string[]> {
 async function isAnyGithubReviewerCommitter(pull: any): Promise<boolean> {
   let reviewers: string[] = [];
   if (pull.requested_reviewers && pull.requested_reviewers.length > 0) {
-    reviewers = reviewers.concat(pull.requested_reviewers.map((r: any) => r.login));
+    reviewers = reviewers.concat(
+      pull.requested_reviewers.map((r: any) => r.login)
+    );
   }
   for (const reviewer of reviewers) {
     if (await github.checkIfCommitter(reviewer)) {
@@ -194,8 +201,8 @@ async function processPull(
       await github.addPrComment(
         pull.number,
         "Closing this PR because dependabot updates for container/** are not allowed due to generated files " +
-        "and excluded_paths is disabled due to dependabot/dependabot-core#14408. " +
-        "Once issue is resolved, please remove this step."
+          "and excluded_paths is disabled due to dependabot/dependabot-core#14408. " +
+          "Once issue is resolved, please remove this step."
       );
       await github.closePr(pull.number);
       return;
@@ -210,8 +217,10 @@ async function processPull(
   console.log(`Processing PR ${pull.number}`);
 
   // If reviewers are already assigned, we just need to check if we should assign a committer.
-  const hasReviewersAssignedForLabels = Object.keys(prState.reviewersAssignedForLabels).length > 0;
-  const hasGithubReviewers = pull.requested_reviewers && pull.requested_reviewers.length > 0;
+  const hasReviewersAssignedForLabels =
+    Object.keys(prState.reviewersAssignedForLabels).length > 0;
+  const hasGithubReviewers =
+    pull.requested_reviewers && pull.requested_reviewers.length > 0;
 
   if (hasReviewersAssignedForLabels || hasGithubReviewers) {
     if (prState.committerAssigned) {
@@ -237,7 +246,11 @@ async function processPull(
       // we can try to guess a label from the PR to assign a committer to.
       if (!labelOfReviewer) {
         let isGithubReviewer = false;
-        if (pull.requested_reviewers && pull.requested_reviewers.some((r: any) => r.login === approver)) isGithubReviewer = true;
+        if (
+          pull.requested_reviewers &&
+          pull.requested_reviewers.some((r: any) => r.login === approver)
+        )
+          isGithubReviewer = true;
 
         if (isGithubReviewer && pull.labels && pull.labels.length > 0) {
           const validLabels = reviewerConfig.getReviewersForAllLabels();
@@ -272,8 +285,7 @@ async function processPull(
         );
         const availableReviewers =
           reviewerConfig.getReviewersForLabel(labelOfReviewer);
-        const fallbackReviewers =
-          reviewerConfig.getFallbackReviewers();
+        const fallbackReviewers = reviewerConfig.getFallbackReviewers();
         const chosenCommitter = await reviewersState.assignNextCommitter(
           availableReviewers,
           fallbackReviewers
@@ -322,7 +334,85 @@ async function processPull(
   }
   prState.commentedAboutFailingChecks = false;
 
-  // Pick reviewers to assign. Store them in reviewerStateToUpdate and update the prState object with those reviewers (and their associated labels)
+  // Experimental LLM / Git History reviewer assigner is behind a feature flag (default: OFF)
+  const ENABLE_LLM_REVIEW_ASSIGNER =
+    process.env.ENABLE_LLM_REVIEW_ASSIGNER === "true";
+
+  if (ENABLE_LLM_REVIEW_ASSIGNER) {
+    let assignedViaAdvisor = false;
+    try {
+      const rawFiles = await github
+        .getGitHubClient()
+        .paginate(github.getGitHubClient().rest.pulls.listFiles, {
+          owner: REPO_OWNER,
+          repo: REPO,
+          pull_number: pull.number,
+        });
+
+      const prContext = buildPrHistoryContext(
+        pull.number,
+        pull.title,
+        pull.body || "",
+        pull.user.login,
+        rawFiles
+      );
+
+      const advisor = new GeminiReviewerAdvisor({
+        committerCheck: github.checkIfCommitter,
+        exclusionList: reviewerConfig.getAllExclusions(),
+      });
+
+      const advice: ReviewerAdviceResult = await advisor.adviseReviewers(
+        prContext
+      );
+
+      if (advice.selectedReviewers.length > 0) {
+        for (const reviewer of advice.selectedReviewers) {
+          prState.reviewersAssignedForLabels[reviewer.expertise] =
+            reviewer.username;
+        }
+        prState.alternateReviewers = advice.alternateReviewers.map(
+          (a) => a.username
+        );
+
+        console.log(
+          `Assigning reviewers with expertise for PR ${pull.number} via ${advice.source}`
+        );
+        await github.addPrComment(
+          pull.number,
+          commentStrings.assignReviewersWithExpertise(advice)
+        );
+
+        try {
+          await github.getGitHubClient().rest.pulls.requestReviewers({
+            owner: REPO_OWNER,
+            repo: REPO,
+            pull_number: pull.number,
+            reviewers: advice.selectedReviewers.map((r) => r.username),
+          });
+        } catch (reqErr) {
+          console.warn(
+            `Could not request reviewers via GitHub API for PR ${pull.number}: ${reqErr}`
+          );
+        }
+
+        github.nextActionReviewers(pull.number, pull.labels);
+        prState.nextAction = "Reviewers";
+        await stateClient.writePrState(pull.number, prState);
+        assignedViaAdvisor = true;
+      }
+    } catch (advisorErr) {
+      console.warn(
+        `Advisor selection failed for PR ${pull.number}: ${advisorErr}. Falling back to label rotation.`
+      );
+    }
+
+    if (assignedViaAdvisor) {
+      return;
+    }
+  }
+
+  // Default: Pick reviewers to assign using label rotation.
   let reviewerStateToUpdate: { [key: string]: typeof ReviewersForLabel } = {};
   const reviewersForLabels: { [key: string]: string[] } =
     reviewerConfig.getReviewersForLabels(pull.labels, [pull.user.login]);
diff --git a/scripts/ci/pr-bot/shared/commentStrings.ts b/scripts/ci/pr-bot/shared/commentStrings.ts
index b556deea4f75..be2186fa4d1b 100644
--- a/scripts/ci/pr-bot/shared/commentStrings.ts
+++ b/scripts/ci/pr-bot/shared/commentStrings.ts
@@ -17,6 +17,7 @@
  */
 
 const { NO_MATCHING_LABEL } = require("./constants");
+import { ReviewerAdviceResult } from "./geminiReviewerAdvisor";
 
 export function allChecksPassed(reviewersToNotify: string[]): string {
   return `All checks have passed: @${reviewersToNotify.join(" ")}`;
@@ -26,9 +27,41 @@ export function assignCommitter(committer: string): string {
   return `R: @${committer} for final approval`;
 }
 
+export function assignReviewersWithExpertise(
+  advice: ReviewerAdviceResult
+): string {
+  let commentString = "### 🧭 Reviewer Assignment\n\n";
+
+  for (const reviewer of advice.selectedReviewers) {
+    const roleBadge =
+      reviewer.role === "primary"
+        ? "**Primary Reviewer**"
+        : "**Secondary Reviewer**";
+    commentString += `- R: @${reviewer.username} (${roleBadge})\n  *Expertise:* ${reviewer.expertise}\n\n`;
+  }
+
+  if (advice.alternateReviewers && advice.alternateReviewers.length > 0) {
+    const alts = advice.alternateReviewers
+      .map((r) => `@${r.username}`)
+      .join(", ");
+    commentString += `*Selected a minimal reviewer set to keep review focused. Backup expert(s): ${alts}*\n\n`;
+  }
+
+  commentString += `Note: If you would like to opt out of this review, comment \`assign to next reviewer\`.
+
+Available commands:
+- \`assign to next reviewer\` - reassign to an alternate reviewer
+- \`assign based on git history\` - assign reviewers based on git history and file churn analysis
+- \`stop reviewer notifications\` - opt out of the automated review tooling
+- \`remind me after tests pass\` - tag the comment author after tests pass
+- \`waiting on author\` - shift the attention set back to the author (any comment or push by the author will return the attention set to the reviewers)
+
+The PR bot will only process comments in the main thread (not review comments).`;
+  return commentString;
+}
+
 export function assignReviewer(labelToReviewerMapping: any): string {
-  let commentString =
-    "Assigning reviewers:\n\n";
+  let commentString = "Assigning reviewers:\n\n";
 
   for (let label in labelToReviewerMapping) {
     let reviewer = labelToReviewerMapping[label];
diff --git a/scripts/ci/pr-bot/shared/geminiReviewerAdvisor.ts b/scripts/ci/pr-bot/shared/geminiReviewerAdvisor.ts
new file mode 100644
index 000000000000..a5dc27d1b2e9
--- /dev/null
+++ b/scripts/ci/pr-bot/shared/geminiReviewerAdvisor.ts
@@ -0,0 +1,605 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+import * as childProcess from "child_process";
+import {
+  PrHistoryContext,
+  CandidateContributor,
+  TouchedFileContext,
+} from "./gitHistory";
+
+/**
+ * Interface representing an individual recommended reviewer.
+ */
+export interface ReviewerRecommendation {
+  readonly username: string;
+  readonly role: "primary" | "secondary";
+  readonly isCommitter: boolean;
+  readonly expertise: string;
+  readonly coveredFiles: readonly string[];
+}
+
+/**
+ * Interface representing an alternate reviewer suggestion.
+ */
+export interface AlternateReviewer {
+  readonly username: string;
+  readonly expertise: string;
+}
+
+/**
+ * Result structure produced by the reviewer advisor.
+ */
+export interface ReviewerAdviceResult {
+  readonly selectedReviewers: readonly ReviewerRecommendation[];
+  readonly alternateReviewers: readonly AlternateReviewer[];
+  readonly reasoning: string;
+  readonly source: "vertex-ai" | "gemini" | "heuristic-fallback";
+}
+
+/**
+ * Interface for LLM clients that can generate structured JSON.
+ */
+export interface IGeminiClient {
+  generateJson<T>(prompt: string): Promise<T>;
+}
+
+/**
+ * Configuration options for Vertex AI.
+ */
+export interface VertexAiConfig {
+  readonly project?: string;
+  readonly location?: string;
+  readonly model?: string;
+  readonly token?: string;
+}
+
+/**
+ * Resolves an OAuth2 access token for Google Cloud.
+ * Checks environment variables first, then falls back to gcloud application-default
+ * credentials or gcloud auth print-access-token.
+ */
+export function getGcloudAccessToken(): string {
+  if (process.env.VERTEX_TOKEN) {
+    return process.env.VERTEX_TOKEN.trim();
+  }
+  if (process.env.CLOUD_ACCESS_TOKEN) {
+    return process.env.CLOUD_ACCESS_TOKEN.trim();
+  }
+  try {
+    const token = childProcess.execFileSync(
+      "gcloud",
+      ["auth", "application-default", "print-access-token"],
+      {
+        encoding: "utf8",
+        stdio: ["pipe", "pipe", "ignore"],
+      }
+    );
+    if (token && token.trim()) {
+      return token.trim();
+    }
+  } catch {}
+
+  try {
+    const token = childProcess.execFileSync(
+      "gcloud",
+      ["auth", "print-access-token"],
+      {
+        encoding: "utf8",
+        stdio: ["pipe", "pipe", "ignore"],
+      }
+    );
+    if (token && token.trim()) {
+      return token.trim();
+    }
+  } catch {}
+
+  return "";
+}
+
+/**
+ * Resolves the Google Cloud project ID for Vertex AI requests.
+ */
+export function getGcloudProject(explicitProject?: string): string {
+  if (explicitProject) return explicitProject;
+  if (process.env.VERTEX_PROJECT) return process.env.VERTEX_PROJECT;
+  if (process.env.GOOGLE_CLOUD_PROJECT) return process.env.GOOGLE_CLOUD_PROJECT;
+  if (process.env.CLOUDSDK_CORE_PROJECT)
+    return process.env.CLOUDSDK_CORE_PROJECT;
+  try {
+    const proj = childProcess.execFileSync(
+      "gcloud",
+      ["config", "get-value", "project"],
+      {
+        encoding: "utf8",
+        stdio: ["pipe", "pipe", "ignore"],
+      }
+    );
+    if (proj && proj.trim() && proj.trim() !== "(unset)") {
+      return proj.trim();
+    }
+  } catch {}
+  return "apache-beam-testing";
+}
+
+/**
+ * Vertex AI LLM client using gcloud OAuth2 authentication.
+ */
+export class VertexAiClient implements IGeminiClient {
+  private readonly project: string;
+  private readonly location: string;
+  private readonly model: string;
+  private readonly token?: string;
+
+  constructor(config: VertexAiConfig = {}) {
+    this.project = getGcloudProject(config.project);
+    this.location =
+      config.location || process.env.VERTEX_LOCATION || "us-central1";
+    this.model = config.model || process.env.VERTEX_MODEL || "gemini-2.5-flash";
+    this.token = config.token;
+  }
+
+  async generateJson<T>(prompt: string): Promise<T> {
+    const token = this.token || getGcloudAccessToken();
+    if (!token) {
+      throw new Error(
+        "No Google Cloud access token found. Please authenticate via `gcloud auth application-default login` or set VERTEX_TOKEN."
+      );
+    }
+
+    const url = `https://${encodeURIComponent(
+      this.location
+    )}-aiplatform.googleapis.com/v1/projects/${encodeURIComponent(
+      this.project
+    )}/locations/${encodeURIComponent(
+      this.location
+    )}/publishers/google/models/${encodeURIComponent(
+      this.model
+    )}:generateContent`;
+
+    const response = await fetch(url, {
+      method: "POST",
+      headers: {
+        Authorization: `Bearer ${token}`,
+        "Content-Type": "application/json",
+      },
+      body: JSON.stringify({
+        contents: [
+          {
+            role: "user",
+            parts: [{ text: prompt }],
+          },
+        ],
+        generationConfig: {
+          temperature: 0.1,
+          responseMimeType: "application/json",
+        },
+      }),
+    });
+
+    if (!response.ok) {
+      const errorText = await response.text();
+      throw new Error(
+        `Vertex AI request failed with status ${response.status}: ${errorText}`
+      );
+    }
+
+    const data: any = await response.json();
+    let candidateText = data?.candidates?.[0]?.content?.parts?.[0]?.text;
+
+    if (!candidateText) {
+      throw new Error("Empty or invalid candidate response from Vertex AI.");
+    }
+
+    candidateText = candidateText.trim();
+    if (candidateText.startsWith("```json")) {
+      candidateText = candidateText.slice(7);
+    } else if (candidateText.startsWith("```")) {
+      candidateText = candidateText.slice(3);
+    }
+    if (candidateText.endsWith("```")) {
+      candidateText = candidateText.slice(0, -3);
+    }
+
+    return JSON.parse(candidateText.trim()) as T;
+  }
+}
+
+/**
+ * Standard HTTP Gemini client using API key.
+ */
+export class GeminiClient implements IGeminiClient {
+  private readonly apiKey: string;
+  private readonly model: string;
+
+  constructor(apiKey: string, model: string = "gemini-2.5-flash") {
+    this.apiKey = apiKey;
+    this.model = model;
+  }
+
+  async generateJson<T>(prompt: string): Promise<T> {
+    if (!this.apiKey) {
+      throw new Error("API key is not configured.");
+    }
+
+    const url = `https://generativelanguage.googleapis.com/v1beta/models/${encodeURIComponent(
+      this.model
+    )}:generateContent?key=${encodeURIComponent(this.apiKey)}`;
+
+    const response = await fetch(url, {
+      method: "POST",
+      headers: {
+        "Content-Type": "application/json",
+      },
+      body: JSON.stringify({
+        contents: [
+          {
+            role: "user",
+            parts: [{ text: prompt }],
+          },
+        ],
+        generationConfig: {
+          temperature: 0.1,
+          responseMimeType: "application/json",
+        },
+      }),
+    });
+
+    if (!response.ok) {
+      const errorText = await response.text();
+      throw new Error(
+        `API request failed with status ${response.status}: ${errorText}`
+      );
+    }
+
+    const data: any = await response.json();
+    let candidateText = data?.candidates?.[0]?.content?.parts?.[0]?.text;
+
+    if (!candidateText) {
+      throw new Error("Empty or invalid candidate response.");
+    }
+
+    candidateText = candidateText.trim();
+    if (candidateText.startsWith("```json")) {
+      candidateText = candidateText.slice(7);
+    } else if (candidateText.startsWith("```")) {
+      candidateText = candidateText.slice(3);
+    }
+    if (candidateText.endsWith("```")) {
+      candidateText = candidateText.slice(0, -3);
+    }
+
+    return JSON.parse(candidateText.trim()) as T;
+  }
+}
+
+/**
+ * Configuration options for the Reviewer Advisor.
+ */
+export interface ReviewerAdvisorOptions {
+  readonly llmClient?: IGeminiClient;
+  readonly geminiClient?: IGeminiClient;
+  readonly vertexAiConfig?: VertexAiConfig;
+  readonly disableLlm?: boolean;
+  readonly committerCheck?: (username: string) => Promise<boolean>;
+  readonly exclusionList?: readonly string[];
+  readonly maxReviewers?: number;
+}
+
+/**
+ * Advisor that analyzes PR git history and selects optimal reviewers using Vertex AI or heuristic fallback.
+ */
+export class GeminiReviewerAdvisor {
+  private readonly client?: IGeminiClient;
+  private readonly committerCheck: (username: string) => Promise<boolean>;
+  private readonly exclusionList: readonly string[];
+  private readonly maxReviewers: number;
+
+  constructor(options: ReviewerAdvisorOptions = {}) {
+    this.client = options.disableLlm
+      ? undefined
+      : options.llmClient ||
+        options.geminiClient ||
+        new VertexAiClient(options.vertexAiConfig || {});
+    this.committerCheck = options.committerCheck ?? (async () => false);
+    this.exclusionList = options.exclusionList ?? [];
+    this.maxReviewers = options.maxReviewers ?? 2;
+  }
+
+  /**
+   * Constructs the prompt instructing Gemini on how to select reviewers.
+   *
+   * @param context Extracted git and PR history.
+   * @param committers Map of username to committer status.
+   * @returns Detailed prompt string.
+   */
+  public buildPrompt(
+    context: PrHistoryContext,
+    committers: Readonly<Record<string, boolean>>
+  ): string {
+    const fileSummaries = context.touchedFiles.map((file) => {
+      const commitSummaries = file.recentCommits
+        .slice(0, 5)
+        .map(
+          (c) =>
+            `    - [${c.date}] ${c.authorLogin || c.authorName}: ${c.subject}`
+        )
+        .join("\n");
+
+      return `- File: ${file.path} (+${file.additions}, -${
+        file.deletions
+      }, changes: ${file.changes}${
+        file.isNewFile ? " [NEW FILE]" : ""
+      })\n  Recent Commits:\n${commitSummaries || "    (No recent commits)"}`;
+    });
+
+    const candidateSummaries = context.candidates.map((c) => {
+      const isCommitter = committers[c.login] ?? false;
+      return `- @${c.login} (${c.name}): ${
+        c.commitCount
+      } commits, last active ${
+        c.lastCommitDate
+      }, committer=${isCommitter}. Files touched: ${c.touchedFilePaths.join(
+        ", "
+      )}`;
+    });
+
+    const exclusions =
+      this.exclusionList.map((e) => `@${e}`).join(", ") || "(none)";
+
+    return `You are the Apache Beam Code Review Assigner.
+Your goal is to choose a small, optimal set of expert reviewers for a pull request based on real git history and file churn.
+
+Pull Request Context:
+- PR Number: #${context.prNumber}
+- Title: "${context.title}"
+- Author: @${context.author}
+- Description: ${context.description || "(No description provided)"}
+
+Files Changed:
+${fileSummaries.join("\n\n")}
+
+Candidate Contributors from Git History:
+${candidateSummaries.join("\n") || "(No candidates found in history)"}
+
+Reviewer Exclusions (Do NOT assign):
+${exclusions}, and the PR author (@${context.author}).
+
+Assignment Guidelines:
+1. REVIEWER SET MINIMIZATION: Choose ideally ONE primary reviewer who can cover the core changes or the most critical subsystem. Only choose two reviewers if the PR touches two completely distinct, major subsystems with no overlapping expert.
+2. TECHNICAL RELEVANCE OVER CHURN: Differentiate between deep architectural contributions (e.g. state management, threading, runners, IO connectors) versus mechanical changes (spotless formatting, dependency bumps, docs).
+3. EXPLICIT EXPERTISE JUSTIFICATION: For each selected reviewer, state their specific technical expertise relevant to this PR in 1-2 concise sentences (e.g., "Authored core KafkaIO watermark estimation logic; directly familiar with reader loop").
+4. ALTERNATES: Suggest 1-2 alternate reviewers in case the primary reviewer is busy or opts out.
+
+Output Format:
+Respond ONLY with a JSON object conforming to this schema:
+{
+  "selectedReviewers": [
+    {
+      "username": "github_username",
+      "role": "primary" or "secondary",
+      "isCommitter": boolean,
+      "expertise": "Specific technical expertise rationale...",
+      "coveredFiles": ["file/path/1", "file/path/2"]
+    }
+  ],
+  "alternateReviewers": [
+    {
+      "username": "alternate_username",
+      "expertise": "Technical rationale..."
+    }
+  ],
+  "reasoning": "Brief explanation of why this set was selected and minimized."
+}`;
+  }
+
+  /**
+   * Deterministic recency-decayed code familiarity fallback when LLM is unavailable.
+   *
+   * @param context Extracted git history context.
+   * @param committers Map of candidate committer status.
+   * @returns ReviewerAdviceResult generated via heuristic familiarity.
+   */
+  public generateHeuristicFallback(
+    context: PrHistoryContext,
+    committers: Readonly<Record<string, boolean>>
+  ): ReviewerAdviceResult {
+    const excluded = new Set(
+      this.exclusionList
+        .concat([context.author])
+        .map((u) => u.toLowerCase().trim())
+    );
+
+    // Calculate familiarity score per candidate using recency time decay
+    const scores = new Map<string, number>();
+    const coveredFilesMap = new Map<string, Set<string>>();
+    const now = Date.now();
+
+    for (const file of context.touchedFiles) {
+      const fileWeight = Math.max(1, file.changes);
+      for (const commit of file.recentCommits) {
+        const login = commit.authorLogin || commit.authorEmail;
+        if (!login || excluded.has(login.toLowerCase())) {
+          continue;
+        }
+
+        const commitTime = new Date(commit.date).getTime();
+        const ageDays = Math.max(0, (now - commitTime) / (1000 * 60 * 60 * 24));
+        const recencyDecay = 1 / (1 + ageDays / 90);
+        const scoreInc = fileWeight * recencyDecay;
+
+        scores.set(login, (scores.get(login) ?? 0) + scoreInc);
+
+        if (!coveredFilesMap.has(login)) {
+          coveredFilesMap.set(login, new Set());
+        }
+        coveredFilesMap.get(login)!.add(file.path);
+      }
+    }
+
+    const sortedCandidates = Array.from(scores.entries())
+      .sort((a, b) => b[1] - a[1])
+      .map(([login]) => login);
+
+    if (sortedCandidates.length === 0) {
+      return {
+        selectedReviewers: [],
+        alternateReviewers: [],
+        reasoning: "No eligible reviewers found in git history.",
+        source: "heuristic-fallback",
+      };
+    }
+
+    const primaryLogin = sortedCandidates[0];
+    const primaryCoveredFiles = Array.from(
+      coveredFilesMap.get(primaryLogin) ?? []
+    );
+    const primaryCandidate = context.candidates.find(
+      (c) => c.login === primaryLogin
+    );
+
+    const primaryRecommendation: ReviewerRecommendation = {
+      username: primaryLogin,
+      role: "primary",
+      isCommitter: committers[primaryLogin] ?? false,
+      expertise: `Frequent contributor with ${
+        primaryCandidate?.commitCount ?? 1
+      } recent commit(s) touching modified files.`,
+      coveredFiles: Object.freeze(primaryCoveredFiles),
+    };
+
+    const alternates: AlternateReviewer[] = [];
+    for (let i = 1; i < Math.min(sortedCandidates.length, 3); i++) {
+      const altLogin = sortedCandidates[i];
+      const altCandidate = context.candidates.find((c) => c.login === altLogin);
+      alternates.push({
+        username: altLogin,
+        expertise: `Contributor with ${
+          altCandidate?.commitCount ?? 1
+        } recent commit(s) in affected files.`,
+      });
+    }
+
+    return {
+      selectedReviewers: Object.freeze([primaryRecommendation]),
+      alternateReviewers: Object.freeze(alternates),
+      reasoning:
+        "Selected top contributor based on recency-decayed git commit churn.",
+      source: "heuristic-fallback",
+    };
+  }
+
+  /**
+   * Evaluates the PR history and produces reviewer recommendations.
+   *
+   * @param context Extracted git history context.
+   * @returns ReviewerAdviceResult.
+   */
+  public async adviseReviewers(
+    context: PrHistoryContext
+  ): Promise<ReviewerAdviceResult> {
+    const committers: Record<string, boolean> = {};
+    for (const candidate of context.candidates) {
+      if (candidate.login) {
+        committers[candidate.login] = await this.committerCheck(
+          candidate.login
+        );
+      }
+    }
+
+    if (!this.client) {
+      return this.generateHeuristicFallback(context, committers);
+    }
+
+    try {
+      const prompt = this.buildPrompt(context, committers);
+      const rawResult = await this.client.generateJson<any>(prompt);
+
+      if (
+        !rawResult ||
+        !Array.isArray(rawResult.selectedReviewers) ||
+        rawResult.selectedReviewers.length === 0
+      ) {
+        console.warn(
+          "Gemini returned invalid or empty reviewer set, falling back to heuristics."
+        );
+        return this.generateHeuristicFallback(context, committers);
+      }
+
+      const excluded = new Set(
+        this.exclusionList
+          .concat([context.author])
+          .map((u) => u.toLowerCase().trim())
+      );
+
+      const filteredSelected: ReviewerRecommendation[] = [];
+      for (const rec of rawResult.selectedReviewers) {
+        const uname = (rec.username || "").replace(/^@/, "").trim();
+        if (uname && !excluded.has(uname.toLowerCase())) {
+          filteredSelected.push({
+            username: uname,
+            role: rec.role === "secondary" ? "secondary" : "primary",
+            isCommitter: committers[uname] ?? false,
+            expertise:
+              rec.expertise || "Selected for recent subsystem contributions.",
+            coveredFiles: Object.freeze(rec.coveredFiles || []),
+          });
+        }
+      }
+
+      if (filteredSelected.length === 0) {
+        return this.generateHeuristicFallback(context, committers);
+      }
+
+      const filteredAlternates: AlternateReviewer[] = [];
+      if (Array.isArray(rawResult.alternateReviewers)) {
+        for (const alt of rawResult.alternateReviewers) {
+          const uname = (alt.username || "").replace(/^@/, "").trim();
+          if (
+            uname &&
+            !excluded.has(uname.toLowerCase()) &&
+            !filteredSelected.some(
+              (s) => s.username.toLowerCase() === uname.toLowerCase()
+            )
+          ) {
+            filteredAlternates.push({
+              username: uname,
+              expertise: alt.expertise || "Contributor to related components.",
+            });
+          }
+        }
+      }
+
+      return {
+        selectedReviewers: Object.freeze(
+          filteredSelected.slice(0, this.maxReviewers)
+        ),
+        alternateReviewers: Object.freeze(filteredAlternates),
+        reasoning:
+          rawResult.reasoning ||
+          "Selected by model based on git history relevance.",
+        source: "vertex-ai",
+      };
+    } catch (error) {
+      console.warn(
+        `Error during reviewer advising: ${error}. Using heuristic fallback.`
+      );
+      return this.generateHeuristicFallback(context, committers);
+    }
+  }
+}
diff --git a/scripts/ci/pr-bot/shared/gitHistory.ts b/scripts/ci/pr-bot/shared/gitHistory.ts
new file mode 100644
index 000000000000..3f06d6eeace3
--- /dev/null
+++ b/scripts/ci/pr-bot/shared/gitHistory.ts
@@ -0,0 +1,462 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+import * as childProcess from "child_process";
+import * as path from "path";
+
+/**
+ * Information about a single git commit touching a file.
+ */
+export interface CommitInfo {
+  readonly hash: string;
+  readonly authorName: string;
+  readonly authorEmail: string;
+  readonly authorLogin: string;
+  readonly date: string;
+  readonly subject: string;
+}
+
+/**
+ * Contextual history and churn statistics for a touched file.
+ */
+export interface TouchedFileContext {
+  readonly path: string;
+  readonly additions: number;
+  readonly deletions: number;
+  readonly changes: number;
+  readonly isNewFile: boolean;
+  readonly recentCommits: readonly CommitInfo[];
+}
+
+/**
+ * Aggregated contributor profile derived from git history.
+ */
+export interface CandidateContributor {
+  readonly login: string;
+  readonly name: string;
+  readonly email: string;
+  readonly commitCount: number;
+  readonly lastCommitDate: string;
+  readonly touchedFilePaths: readonly string[];
+}
+
+/**
+ * Full PR context prepared for the LLM reviewer advisor.
+ */
+export interface PrHistoryContext {
+  readonly prNumber: number;
+  readonly title: string;
+  readonly description: string;
+  readonly author: string;
+  readonly touchedFiles: readonly TouchedFileContext[];
+  readonly candidates: readonly CandidateContributor[];
+}
+
+/**
+ * Known bot identifiers that should be filtered out from reviewer candidates.
+ */
+const BOT_IDENTIFIERS: readonly string[] = [
+  "dependabot",
+  "dependabot[bot]",
+  "github-actions",
+  "github-actions[bot]",
+  "beam-bot",
+  "codecov",
+  "codecov[bot]",
+  "asfgit",
+  "spotless",
+];
+
+/**
+ * File extensions and paths that are trivial or generated and should not dominate reviewer selection.
+ */
+const LOW_PRIORITY_FILE_PATTERNS: readonly RegExp[] = [
+  /\.lock$/,
+  /package-lock\.json$/,
+  /gradle\.lockfile$/,
+  /\.md$/,
+  /\.mailmap$/,
+  /buildSrc\/.*\.gradle$/,
+  /\.gitignore$/,
+];
+
+/**
+ * Determines whether a commit author represents an automated bot or service account.
+ *
+ * @param authorName The author display name.
+ * @param authorEmail The author email.
+ * @param authorLogin The resolved GitHub login if available.
+ * @returns True if the author is recognized as an automated bot.
+ */
+export function isBotAuthor(
+  authorName: string,
+  authorEmail: string,
+  authorLogin: string
+): boolean {
+  const lowerName = authorName.toLowerCase();
+  const lowerEmail = authorEmail.toLowerCase();
+  const lowerLogin = authorLogin.toLowerCase();
+
+  for (const bot of BOT_IDENTIFIERS) {
+    if (
+      lowerName.includes(bot) ||
+      lowerEmail.includes(bot) ||
+      lowerLogin.includes(bot)
+    ) {
+      return true;
+    }
+  }
+
+  if (lowerEmail.includes("noreply@github.com") && lowerName === "github") {
+    return true;
+  }
+
+  return false;
+}
+
+/**
+ * Resolves a GitHub username from an email, name, or commit subject.
+ *
+ * @param authorName The author display name.
+ * @param authorEmail The author email address.
+ * @param knownLogins Optional mapping of normalized email/name to GitHub usernames.
+ * @returns The resolved GitHub login, or empty string if unresolved.
+ */
+export function resolveAuthorLogin(
+  authorName: string,
+  authorEmail: string,
+  knownLogins: Readonly<Record<string, string>> = {}
+): string {
+  const lowerEmail = authorEmail.toLowerCase().trim();
+  const lowerName = authorName.toLowerCase().trim();
+
+  if (knownLogins[lowerEmail]) {
+    return knownLogins[lowerEmail];
+  }
+  if (knownLogins[lowerName]) {
+    return knownLogins[lowerName];
+  }
+
+  // GitHub noreply email format: [id+]login@users.noreply.github.com
+  const noreplyMatch = lowerEmail.match(
+    /^(?:\d+\+)?([a-z0-9](?:[a-z0-9-]*[a-z0-9])?)@users\.noreply\.github\.com$/
+  );
+  if (noreplyMatch && noreplyMatch[1]) {
+    return noreplyMatch[1];
+  }
+
+  return "";
+}
+
+/**
+ * Checks whether a file path is considered a low-priority or generated file.
+ *
+ * @param filePath Path of the file relative to repository root.
+ * @returns True if the file should be given lower weight in reviewer selection.
+ */
+export function isLowPriorityFile(filePath: string): boolean {
+  return LOW_PRIORITY_FILE_PATTERNS.some((pattern) => pattern.test(filePath));
+}
+
+let cachedRepoRoot = "";
+
+/**
+ * Returns the repository root directory using git rev-parse.
+ */
+export function getRepoRoot(): string {
+  if (!cachedRepoRoot) {
+    try {
+      const stdout = childProcess.execFileSync(
+        "git",
+        ["rev-parse", "--show-toplevel"],
+        {
+          encoding: "utf8",
+        }
+      );
+      cachedRepoRoot = stdout ? stdout.toString().trim() : process.cwd();
+    } catch {
+      cachedRepoRoot = process.cwd();
+    }
+  }
+  return cachedRepoRoot;
+}
+
+/**
+ * Extracts recent commits for a specific file or fallback directory using git log.
+ *
+ * @param filePath Relative path to the file.
+ * @param maxCommits Maximum number of recent commits to retrieve.
+ * @param workingDirectory Base git directory.
+ * @param knownLogins Optional map of author names/emails to GitHub logins.
+ * @returns Array of commit information objects.
+ */
+export function getRecentCommitsForFile(
+  filePath: string,
+  maxCommits: number = 10,
+  workingDirectory?: string,
+  knownLogins: Readonly<Record<string, string>> = {}
+): readonly CommitInfo[] {
+  const execOptions: childProcess.ExecFileSyncOptions = {
+    encoding: "utf8",
+    maxBuffer: 4 * 1024 * 1024,
+    cwd: workingDirectory ?? getRepoRoot(),
+  };
+
+  try {
+    let rawOutput = childProcess.execFileSync(
+      "git",
+      [
+        "log",
+        "-n",
+        String(maxCommits),
+        "--no-merges",
+        "--format=%H%x09%an%x09%ae%x09%as%x09%s",
+        "--",
+        filePath,
+      ],
+      execOptions
+    );
+    let stdout = rawOutput ? rawOutput.toString() : "";
+
+    // If file is new and has no history, fall back to parent directory history
+    if (!stdout.trim()) {
+      const parentDir = path.dirname(filePath);
+      if (parentDir && parentDir !== "." && parentDir !== "/") {
+        rawOutput = childProcess.execFileSync(
+          "git",
+          [
+            "log",
+            "-n",
+            String(maxCommits),
+            "--no-merges",
+            "--format=%H%x09%an%x09%ae%x09%as%x09%s",
+            "--",
+            parentDir,
+          ],
+          execOptions
+        );
+        stdout = rawOutput ? rawOutput.toString() : "";
+      }
+    }
+
+    const commits: CommitInfo[] = [];
+    const lines = stdout.trim().split("\n");
+
+    for (const line of lines) {
+      if (!line.trim()) {
+        continue;
+      }
+      const parts = line.split("\t");
+      if (parts.length < 5) {
+        continue;
+      }
+      const [hash, authorName, authorEmail, date, ...subjectParts] = parts;
+      const subject = subjectParts.join("\t");
+      const login = resolveAuthorLogin(authorName, authorEmail, knownLogins);
+
+      if (!isBotAuthor(authorName, authorEmail, login)) {
+        commits.push({
+          hash,
+          authorName,
+          authorEmail,
+          authorLogin: login,
+          date,
+          subject,
+        });
+      }
+    }
+
+    return Object.freeze(commits);
+  } catch (error) {
+    console.error(`Error reading git log for ${filePath}: ${error}`);
+    return Object.freeze([]);
+  }
+}
+
+/**
+ * Aggregates candidate contributors across all touched files, counting their commits and files touched.
+ *
+ * @param touchedFiles List of touched files with their commit history.
+ * @param prAuthor GitHub username of the PR author.
+ * @returns Array of candidate contributors sorted by commit count descending.
+ */
+export function aggregateCandidates(
+  touchedFiles: readonly TouchedFileContext[],
+  prAuthor: string
+): readonly CandidateContributor[] {
+  const candidateMap = new Map<
+    string,
+    {
+      login: string;
+      name: string;
+      email: string;
+      commitCount: number;
+      lastCommitDate: string;
+      touchedFilePaths: Set<string>;
+    }
+  >();
+
+  const normalizedPrAuthor = prAuthor.toLowerCase().trim();
+
+  for (const file of touchedFiles) {
+    for (const commit of file.recentCommits) {
+      const candidateKey = commit.authorLogin
+        ? commit.authorLogin.toLowerCase()
+        : commit.authorEmail.toLowerCase();
+
+      // Skip the PR author
+      if (
+        commit.authorLogin.toLowerCase() === normalizedPrAuthor ||
+        commit.authorName.toLowerCase() === normalizedPrAuthor
+      ) {
+        continue;
+      }
+
+      const existing = candidateMap.get(candidateKey);
+      if (existing) {
+        existing.commitCount += 1;
+        existing.touchedFilePaths.add(file.path);
+        if (commit.date > existing.lastCommitDate) {
+          existing.lastCommitDate = commit.date;
+        }
+        if (!existing.login && commit.authorLogin) {
+          existing.login = commit.authorLogin;
+        }
+      } else {
+        const touchedPaths = new Set<string>();
+        touchedPaths.add(file.path);
+        candidateMap.set(candidateKey, {
+          login: commit.authorLogin || commit.authorEmail,
+          name: commit.authorName,
+          email: commit.authorEmail,
+          commitCount: 1,
+          lastCommitDate: commit.date,
+          touchedFilePaths: touchedPaths,
+        });
+      }
+    }
+  }
+
+  const result: CandidateContributor[] = [];
+  for (const val of candidateMap.values()) {
+    result.push({
+      login: val.login,
+      name: val.name,
+      email: val.email,
+      commitCount: val.commitCount,
+      lastCommitDate: val.lastCommitDate,
+      touchedFilePaths: Object.freeze(Array.from(val.touchedFilePaths)),
+    });
+  }
+
+  // Sort by commit count descending, then by most recent commit date descending
+  result.sort((a, b) => {
+    if (b.commitCount !== a.commitCount) {
+      return b.commitCount - a.commitCount;
+    }
+    return b.lastCommitDate.localeCompare(a.lastCommitDate);
+  });
+
+  return Object.freeze(result);
+}
+
+/**
+ * Assembles the full PR history context from PR files and git history.
+ *
+ * @param prNumber GitHub pull request number.
+ * @param title PR title.
+ * @param description PR description body.
+ * @param author PR author username.
+ * @param files List of files changed in the PR with additions/deletions.
+ * @param options Configuration options.
+ * @returns Fully constructed PrHistoryContext.
+ */
+export function buildPrHistoryContext(
+  prNumber: number,
+  title: string,
+  description: string,
+  author: string,
+  files: readonly {
+    filename: string;
+    additions?: number;
+    deletions?: number;
+    changes?: number;
+    status?: string;
+  }[],
+  options: {
+    maxFiles?: number;
+    commitsPerFile?: number;
+    workingDirectory?: string;
+    knownLogins?: Readonly<Record<string, string>>;
+  } = {}
+): PrHistoryContext {
+  const maxFiles = options.maxFiles ?? 15;
+  const commitsPerFile = options.commitsPerFile ?? 8;
+  const workingDirectory = options.workingDirectory;
+  const knownLogins = options.knownLogins ?? {};
+
+  // Sort files so that substantive, high-churn source files appear before trivial/lock files
+  const sortedFiles = [...files].sort((a, b) => {
+    const aLow = isLowPriorityFile(a.filename) ? 1 : 0;
+    const bLow = isLowPriorityFile(b.filename) ? 1 : 0;
+    if (aLow !== bLow) {
+      return aLow - bLow;
+    }
+    const aChanges = a.changes ?? (a.additions ?? 0) + (a.deletions ?? 0);
+    const bChanges = b.changes ?? (b.additions ?? 0) + (b.deletions ?? 0);
+    return bChanges - aChanges;
+  });
+
+  const selectedFiles = sortedFiles.slice(0, maxFiles);
+  const touchedFileContexts: TouchedFileContext[] = [];
+
+  for (const file of selectedFiles) {
+    const additions = file.additions ?? 0;
+    const deletions = file.deletions ?? 0;
+    const changes = file.changes ?? additions + deletions;
+    const isNew = file.status === "added";
+
+    const commits = getRecentCommitsForFile(
+      file.filename,
+      commitsPerFile,
+      workingDirectory,
+      knownLogins
+    );
+
+    touchedFileContexts.push({
+      path: file.filename,
+      additions,
+      deletions,
+      changes,
+      isNewFile: isNew,
+      recentCommits: commits,
+    });
+  }
+
+  const frozenFiles = Object.freeze(touchedFileContexts);
+  const candidates = aggregateCandidates(frozenFiles, author);
+
+  return {
+    prNumber,
+    title,
+    description,
+    author,
+    touchedFiles: frozenFiles,
+    candidates,
+  };
+}
diff --git a/scripts/ci/pr-bot/shared/pr.ts b/scripts/ci/pr-bot/shared/pr.ts
index 2f64572068e5..5b6e48fa135f 100644
--- a/scripts/ci/pr-bot/shared/pr.ts
+++ b/scripts/ci/pr-bot/shared/pr.ts
@@ -25,6 +25,7 @@ export class Pr {
   public stopReviewerNotifications: boolean;
   public remindAfterTestsPass: string[];
   public committerAssigned: boolean;
+  public alternateReviewers: string[];
 
   constructor(propertyDictionary) {
     this.commentedAboutFailingChecks = false;
@@ -33,6 +34,7 @@ export class Pr {
     this.stopReviewerNotifications = false;
     this.remindAfterTestsPass = []; // List of handles
     this.committerAssigned = false;
+    this.alternateReviewers = [];
 
     if (!propertyDictionary) {
       return;
@@ -59,6 +61,9 @@ export class Pr {
       if ("committerAssigned" in propertyDictionary) {
         this.committerAssigned = propertyDictionary["committerAssigned"];
       }
+      if ("alternateReviewers" in propertyDictionary) {
+        this.alternateReviewers = propertyDictionary["alternateReviewers"];
+      }
     }
   }
 
diff --git a/scripts/ci/pr-bot/shared/reviewerConfig.ts b/scripts/ci/pr-bot/shared/reviewerConfig.ts
index 8fc0d32b014b..21a833a2f3d2 100644
--- a/scripts/ci/pr-bot/shared/reviewerConfig.ts
+++ b/scripts/ci/pr-bot/shared/reviewerConfig.ts
@@ -119,6 +119,20 @@ ${yaml.dump(this.config)}`;
     return labelObject?.exclusionList ?? [];
   }
 
+  // Returns all excluded reviewers configured across all labels.
+  getAllExclusions(): string[] {
+    const exclusions = new Set<string>();
+    const labelObjects = this.config.labels || [];
+    for (const labelObject of labelObjects) {
+      if (Array.isArray(labelObject.exclusionList)) {
+        for (const user of labelObject.exclusionList) {
+          exclusions.add(user);
+        }
+      }
+    }
+    return Array.from(exclusions);
+  }
+
   // Get fallback reviewers excluding the author.
   getFallbackReviewers(exclusionList: string[]): string[] {
     return this.excludeFromReviewers(
diff --git a/scripts/ci/pr-bot/shared/userCommand.ts b/scripts/ci/pr-bot/shared/userCommand.ts
index 6980468c3b19..ed9934899be2 100644
--- a/scripts/ci/pr-bot/shared/userCommand.ts
+++ b/scripts/ci/pr-bot/shared/userCommand.ts
@@ -18,9 +18,11 @@
 
 const github = require("./githubUtils");
 const commentStrings = require("./commentStrings");
-const { BOT_NAME } = require("./constants");
+const { BOT_NAME, REPO_OWNER, REPO } = require("./constants");
 const { StateClient } = require("./persistentState");
 const { ReviewerConfig } = require("./reviewerConfig");
+const { buildPrHistoryContext } = require("./gitHistory");
+const { GeminiReviewerAdvisor } = require("./geminiReviewerAdvisor");
 
 // Reads the comment and processes the command if one is contained in it.
 // Returns true if it runs a command, false otherwise.
@@ -41,7 +43,7 @@ export async function processCommand(
   commentText = commentText.toLowerCase();
 
   let prState = await stateClient.getPrState(pullNumber);
-  if(prState.stopReviewerNotifications) {
+  if (prState.stopReviewerNotifications) {
     // Notifications stopped, only "allow assign set of reviewers"
     if (commentText.indexOf("assign set of reviewers") > -1) {
       await assignReviewerSet(payload, pullNumber, stateClient, reviewerConfig);
@@ -51,6 +53,16 @@ export async function processCommand(
   } else {
     if (commentText.indexOf("r: @") > -1) {
       await manuallyAssignedToReviewer(pullNumber, stateClient);
+    } else if (
+      commentText.indexOf("assign based on git history") > -1 ||
+      commentText.indexOf("assign based on history") > -1
+    ) {
+      await assignBasedOnGitHistory(
+        payload,
+        pullNumber,
+        stateClient,
+        reviewerConfig
+      );
     } else if (commentText.indexOf("assign to next reviewer") > -1) {
       await assignToNextReviewer(
         payload,
@@ -86,6 +98,42 @@ async function assignToNextReviewer(
   reviewerConfig: typeof ReviewerConfig
 ) {
   let prState = await stateClient.getPrState(pullNumber);
+
+  // If the PR has alternate reviewers from the advisor, assign the next one
+  if (prState.alternateReviewers && prState.alternateReviewers.length > 0) {
+    const nextReviewer = prState.alternateReviewers.shift();
+    let labelOfReviewer = prState.getLabelForReviewer(payload.sender.login);
+    if (labelOfReviewer) {
+      prState.reviewersAssignedForLabels[labelOfReviewer] = nextReviewer;
+    } else {
+      prState.reviewersAssignedForLabels["expert"] = nextReviewer;
+    }
+
+    console.log(`Reassigning reviewer to alternate expert ${nextReviewer}`);
+    await github.addPrComment(
+      pullNumber,
+      `Reassigning reviewer to backup expert @${nextReviewer} per request.`
+    );
+    try {
+      await github.getGitHubClient().rest.pulls.requestReviewers({
+        owner: REPO_OWNER,
+        repo: REPO,
+        pull_number: pullNumber,
+        reviewers: [nextReviewer],
+      });
+    } catch (e) {
+      console.warn(`Failed to request reviewer via GitHub API: ${e}`);
+    }
+
+    const existingLabels =
+      payload.issue?.labels || payload.pull_request?.labels;
+    await github.nextActionReviewers(pullNumber, existingLabels);
+    prState.nextAction = "Reviewers";
+
+    await stateClient.writePrState(pullNumber, prState);
+    return;
+  }
+
   let labelOfReviewer = prState.getLabelForReviewer(payload.sender.login);
   if (labelOfReviewer) {
     let reviewersState = await stateClient.getReviewersForLabelState(
@@ -185,7 +233,7 @@ async function assignReviewerSet(
   reviewerConfig: typeof ReviewerConfig
 ) {
   let prState = await stateClient.getPrState(pullNumber);
-  if(prState.stopReviewerNotifications) {
+  if (prState.stopReviewerNotifications) {
     // Restore notifications, and clear any existing reviewer set to
     // allow new reviewers to be assigned.
     prState.stopReviewerNotifications = false;
@@ -243,3 +291,84 @@ async function assignReviewerSet(
     );
   }
 }
+
+async function assignBasedOnGitHistory(
+  payload: any,
+  pullNumber: number,
+  stateClient: typeof StateClient,
+  reviewerConfig: typeof ReviewerConfig
+) {
+  let prState = await stateClient.getPrState(pullNumber);
+  const client = github.getGitHubClient();
+
+  const pullResponse = await client.rest.pulls.get({
+    owner: REPO_OWNER,
+    repo: REPO,
+    pull_number: pullNumber,
+  });
+  const pull = pullResponse.data;
+
+  const rawFiles = await client.paginate(client.rest.pulls.listFiles, {
+    owner: REPO_OWNER,
+    repo: REPO,
+    pull_number: pullNumber,
+  });
+
+  const prContext = buildPrHistoryContext(
+    pullNumber,
+    pull.title,
+    pull.body || "",
+    pull.user.login,
+    rawFiles
+  );
+
+  const advisor = new GeminiReviewerAdvisor({
+    committerCheck: github.checkIfCommitter,
+    exclusionList: reviewerConfig.getAllExclusions(),
+  });
+
+  const advice = await advisor.adviseReviewers(prContext);
+
+  if (advice.selectedReviewers.length > 0) {
+    prState.reviewersAssignedForLabels = {};
+    for (const reviewer of advice.selectedReviewers) {
+      prState.reviewersAssignedForLabels[reviewer.expertise] =
+        reviewer.username;
+    }
+    prState.alternateReviewers = advice.alternateReviewers.map(
+      (a: any) => a.username
+    );
+
+    console.log(
+      `Assigning reviewers with expertise for PR ${pullNumber} via ${advice.source} per user command`
+    );
+    await github.addPrComment(
+      pullNumber,
+      commentStrings.assignReviewersWithExpertise(advice)
+    );
+
+    try {
+      await client.rest.pulls.requestReviewers({
+        owner: REPO_OWNER,
+        repo: REPO,
+        pull_number: pullNumber,
+        reviewers: advice.selectedReviewers.map((r: any) => r.username),
+      });
+    } catch (reqErr) {
+      console.warn(
+        `Could not request reviewers via GitHub API for PR ${pullNumber}: ${reqErr}`
+      );
+    }
+
+    const existingLabels =
+      payload.issue?.labels || payload.pull_request?.labels;
+    await github.nextActionReviewers(pullNumber, existingLabels);
+    prState.nextAction = "Reviewers";
+    await stateClient.writePrState(pullNumber, prState);
+  } else {
+    await github.addPrComment(
+      pullNumber,
+      "Unable to determine reviewers based on git history. Please assign reviewers manually using `R: @username`."
+    );
+  }
+}
diff --git a/scripts/ci/pr-bot/test/commentStringsTest.ts b/scripts/ci/pr-bot/test/commentStringsTest.ts
new file mode 100644
index 000000000000..e350a3023b37
--- /dev/null
+++ b/scripts/ci/pr-bot/test/commentStringsTest.ts
@@ -0,0 +1,105 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+import * as assert from "assert";
+import { assignReviewersWithExpertise } from "../shared/commentStrings";
+import { ReviewerAdviceResult } from "../shared/geminiReviewerAdvisor";
+
+describe("commentStrings", () => {
+  describe("assignReviewersWithExpertise()", () => {
+    it("should format primary reviewer and expertise correctly", () => {
+      const advice: ReviewerAdviceResult = {
+        selectedReviewers: [
+          {
+            username: "alice",
+            role: "primary",
+            isCommitter: true,
+            expertise:
+              "Authored KafkaIO dynamic reads and watermark estimation.",
+            coveredFiles: ["sdks/java/io/kafka/KafkaIO.java"],
+          },
+        ],
+        alternateReviewers: [
+          {
+            username: "bob",
+            expertise: "Active contributor to Kafka connector dependencies.",
+          },
+        ],
+        reasoning: "Alice is the most relevant expert on this subsystem.",
+        source: "gemini",
+      };
+
+      const comment = assignReviewersWithExpertise(advice);
+
+      assert.strictEqual(comment.includes("### 🧭 Reviewer Assignment"), true);
+      assert.strictEqual(
+        comment.includes("R: @alice (**Primary Reviewer**)"),
+        true
+      );
+      assert.strictEqual(
+        comment.includes(
+          "*Expertise:* Authored KafkaIO dynamic reads and watermark estimation."
+        ),
+        true
+      );
+      assert.strictEqual(comment.includes("Backup expert(s): @bob"), true);
+      assert.strictEqual(comment.includes("assign to next reviewer"), true);
+      assert.strictEqual(comment.includes("assign based on git history"), true);
+    });
+
+    it("should format multiple reviewers when selected", () => {
+      const advice: ReviewerAdviceResult = {
+        selectedReviewers: [
+          {
+            username: "alice",
+            role: "primary",
+            isCommitter: true,
+            expertise: "Flink runner engine specialist.",
+            coveredFiles: ["runners/flink/Runner.java"],
+          },
+          {
+            username: "charlie",
+            role: "secondary",
+            isCommitter: false,
+            expertise: "KafkaIO maintainer.",
+            coveredFiles: ["sdks/java/io/kafka/KafkaIO.java"],
+          },
+        ],
+        alternateReviewers: [],
+        reasoning: "Cross-cutting PR touching two distinct subsystems.",
+        source: "gemini",
+      };
+
+      const comment = assignReviewersWithExpertise(advice);
+
+      assert.strictEqual(
+        comment.includes("R: @alice (**Primary Reviewer**)"),
+        true
+      );
+      assert.strictEqual(
+        comment.includes("R: @charlie (**Secondary Reviewer**)"),
+        true
+      );
+      assert.strictEqual(
+        comment.includes("Flink runner engine specialist."),
+        true
+      );
+      assert.strictEqual(comment.includes("KafkaIO maintainer."), true);
+    });
+  });
+});
diff --git a/scripts/ci/pr-bot/test/geminiReviewerAdvisorTest.ts b/scripts/ci/pr-bot/test/geminiReviewerAdvisorTest.ts
new file mode 100644
index 000000000000..4709e7129e2d
--- /dev/null
+++ b/scripts/ci/pr-bot/test/geminiReviewerAdvisorTest.ts
@@ -0,0 +1,232 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+import * as assert from "assert";
+import {
+  GeminiReviewerAdvisor,
+  IGeminiClient,
+} from "../shared/geminiReviewerAdvisor";
+import { PrHistoryContext } from "../shared/gitHistory";
+
+/**
+ * Concrete test client that delivers a preconfigured JSON payload or throws an error.
+ */
+class StaticJsonGeminiClient implements IGeminiClient {
+  private readonly response: any;
+  private readonly shouldThrow: boolean;
+  public lastPrompt: string = "";
+
+  constructor(response: any, shouldThrow: boolean = false) {
+    this.response = response;
+    this.shouldThrow = shouldThrow;
+  }
+
+  async generateJson<T>(prompt: string): Promise<T> {
+    this.lastPrompt = prompt;
+    if (this.shouldThrow) {
+      throw new Error("Network timeout or invalid API key");
+    }
+    return this.response as T;
+  }
+}
+
+describe("GeminiReviewerAdvisor", () => {
+  const sampleContext: PrHistoryContext = {
+    prNumber: 9999,
+    title: "KafkaIO: Fix consumer poll deadlock",
+    description: "Fixes an issue where poll() blocks indefinitely",
+    author: "contributorA",
+    touchedFiles: [
+      {
+        path: "sdks/java/io/kafka/KafkaIO.java",
+        additions: 120,
+        deletions: 15,
+        changes: 135,
+        isNewFile: false,
+        recentCommits: [
+          {
+            hash: "h1",
+            authorName: "Alice",
+            authorEmail: "alice@example.com",
+            authorLogin: "alice",
+            date: "2026-08-15",
+            subject: "Refactor KafkaIO reader watermark estimation",
+          },
+          {
+            hash: "h2",
+            authorName: "Bob",
+            authorEmail: "bob@example.com",
+            authorLogin: "bob",
+            date: "2026-07-10",
+            subject: "Bump kafka clients version",
+          },
+        ],
+      },
+    ],
+    candidates: [
+      {
+        login: "alice",
+        name: "Alice",
+        email: "alice@example.com",
+        commitCount: 5,
+        lastCommitDate: "2026-08-15",
+        touchedFilePaths: ["sdks/java/io/kafka/KafkaIO.java"],
+      },
+      {
+        login: "bob",
+        name: "Bob",
+        email: "bob@example.com",
+        commitCount: 2,
+        lastCommitDate: "2026-07-10",
+        touchedFilePaths: ["sdks/java/io/kafka/KafkaIO.java"],
+      },
+    ],
+  };
+
+  describe("buildPrompt()", () => {
+    it("should construct prompt containing PR details, file changes, and instructions", () => {
+      const advisor = new GeminiReviewerAdvisor({
+        exclusionList: ["busyReviewer"],
+      });
+      const prompt = advisor.buildPrompt(sampleContext, {
+        alice: true,
+        bob: false,
+      });
+
+      assert.strictEqual(prompt.includes("PR Number: #9999"), true);
+      assert.strictEqual(
+        prompt.includes("KafkaIO: Fix consumer poll deadlock"),
+        true
+      );
+      assert.strictEqual(prompt.includes("KafkaIO.java"), true);
+      assert.strictEqual(prompt.includes("@alice"), true);
+      assert.strictEqual(prompt.includes("committer=true"), true);
+      assert.strictEqual(prompt.includes("@busyReviewer"), true);
+      assert.strictEqual(prompt.includes("@contributorA"), true);
+      assert.strictEqual(prompt.includes("REVIEWER SET MINIMIZATION"), true);
+    });
+  });
+
+  describe("generateHeuristicFallback()", () => {
+    it("should select the most active recent contributor using recency-decayed scoring", () => {
+      const advisor = new GeminiReviewerAdvisor();
+      const result = advisor.generateHeuristicFallback(sampleContext, {
+        alice: true,
+        bob: false,
+      });
+
+      assert.strictEqual(result.source, "heuristic-fallback");
+      assert.strictEqual(result.selectedReviewers.length, 1);
+      assert.strictEqual(result.selectedReviewers[0].username, "alice");
+      assert.strictEqual(result.selectedReviewers[0].role, "primary");
+      assert.strictEqual(result.selectedReviewers[0].isCommitter, true);
+      assert.strictEqual(result.alternateReviewers.length > 0, true);
+      assert.strictEqual(result.alternateReviewers[0].username, "bob");
+    });
+
+    it("should respect exclusion list in heuristic fallback", () => {
+      const advisor = new GeminiReviewerAdvisor({
+        exclusionList: ["alice"],
+      });
+      const result = advisor.generateHeuristicFallback(sampleContext, {
+        alice: true,
+        bob: false,
+      });
+
+      assert.strictEqual(result.selectedReviewers.length, 1);
+      assert.strictEqual(result.selectedReviewers[0].username, "bob");
+    });
+  });
+
+  describe("adviseReviewers() with LLM client", () => {
+    it("should parse and return structured recommendations from LLM", async () => {
+      const fakeGeminiResponse = {
+        selectedReviewers: [
+          {
+            username: "alice",
+            role: "primary",
+            isCommitter: true,
+            expertise:
+              "Authored core KafkaIO watermark estimation logic; directly familiar with reader loop.",
+            coveredFiles: ["sdks/java/io/kafka/KafkaIO.java"],
+          },
+        ],
+        alternateReviewers: [
+          {
+            username: "bob",
+            expertise:
+              "Updated Kafka clients and familiar with configuration dependencies.",
+          },
+        ],
+        reasoning:
+          "Alice's previous changes directly touched the watermark estimation loop being fixed here.",
+      };
+
+      const testClient = new StaticJsonGeminiClient(fakeGeminiResponse);
+      const advisor = new GeminiReviewerAdvisor({
+        geminiClient: testClient,
+        committerCheck: async (login) => login === "alice",
+      });
+
+      const result = await advisor.adviseReviewers(sampleContext);
+
+      assert.strictEqual(result.source, "vertex-ai");
+      assert.strictEqual(result.selectedReviewers.length, 1);
+      assert.strictEqual(result.selectedReviewers[0].username, "alice");
+      assert.strictEqual(result.selectedReviewers[0].role, "primary");
+      assert.strictEqual(result.selectedReviewers[0].isCommitter, true);
+      assert.strictEqual(
+        result.selectedReviewers[0].expertise,
+        "Authored core KafkaIO watermark estimation logic; directly familiar with reader loop."
+      );
+      assert.strictEqual(result.alternateReviewers.length, 1);
+      assert.strictEqual(result.alternateReviewers[0].username, "bob");
+      assert.strictEqual(
+        result.reasoning.includes("watermark estimation loop"),
+        true
+      );
+    });
+
+    it("should run offline heuristic when disableLlm is set to true", async () => {
+      const advisor = new GeminiReviewerAdvisor({
+        disableLlm: true,
+        committerCheck: async (login) => login === "alice",
+      });
+
+      const result = await advisor.adviseReviewers(sampleContext);
+
+      assert.strictEqual(result.source, "heuristic-fallback");
+      assert.strictEqual(result.selectedReviewers.length, 1);
+      assert.strictEqual(result.selectedReviewers[0].username, "alice");
+    });
+
+    it("should automatically fall back to heuristics if the LLM call fails", async () => {
+      const failingClient = new StaticJsonGeminiClient(null, true);
+      const advisor = new GeminiReviewerAdvisor({
+        geminiClient: failingClient,
+        committerCheck: async (login) => login === "alice",
+      });
+
+      const result = await advisor.adviseReviewers(sampleContext);
+
+      assert.strictEqual(result.source, "heuristic-fallback");
+      assert.strictEqual(result.selectedReviewers.length, 1);
+      assert.strictEqual(result.selectedReviewers[0].username, "alice");
+    });
+  });
+});
diff --git a/scripts/ci/pr-bot/test/gitHistoryTest.ts b/scripts/ci/pr-bot/test/gitHistoryTest.ts
new file mode 100644
index 000000000000..8e644cdcde9d
--- /dev/null
+++ b/scripts/ci/pr-bot/test/gitHistoryTest.ts
@@ -0,0 +1,249 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+import * as assert from "assert";
+import {
+  isBotAuthor,
+  resolveAuthorLogin,
+  isLowPriorityFile,
+  getRecentCommitsForFile,
+  aggregateCandidates,
+  buildPrHistoryContext,
+  TouchedFileContext,
+} from "../shared/gitHistory";
+
+describe("gitHistory", () => {
+  describe("isBotAuthor()", () => {
+    it("should identify known bots correctly", () => {
+      assert.strictEqual(
+        isBotAuthor(
+          "dependabot[bot]",
+          "dependabot@users.noreply.github.com",
+          "dependabot[bot]"
+        ),
+        true
+      );
+      assert.strictEqual(
+        isBotAuthor("github-actions", "actions@github.com", "github-actions"),
+        true
+      );
+      assert.strictEqual(
+        isBotAuthor("beam-bot", "beam-bot@apache.org", "beam-bot"),
+        true
+      );
+      assert.strictEqual(
+        isBotAuthor("Codecov", "support@codecov.io", "codecov"),
+        true
+      );
+    });
+
+    it("should not mark human contributors as bots", () => {
+      assert.strictEqual(
+        isBotAuthor("Kenn Knowles", "klk@google.com", "kennknowles"),
+        false
+      );
+      assert.strictEqual(
+        isBotAuthor("Alice", "alice@example.com", "alice"),
+        false
+      );
+    });
+  });
+
+  describe("resolveAuthorLogin()", () => {
+    it("should extract GitHub username from users.noreply.github.com email", () => {
+      assert.strictEqual(
+        resolveAuthorLogin("Alice", "123456+alice@users.noreply.github.com"),
+        "alice"
+      );
+      assert.strictEqual(
+        resolveAuthorLogin("Bob", "bob-dev@users.noreply.github.com"),
+        "bob-dev"
+      );
+    });
+
+    it("should use knownLogins map if email matches", () => {
+      const known = { "klk@google.com": "kennknowles" };
+      assert.strictEqual(
+        resolveAuthorLogin("Kenn Knowles", "klk@google.com", known),
+        "kennknowles"
+      );
+    });
+
+    it("should return empty string if username cannot be inferred", () => {
+      assert.strictEqual(
+        resolveAuthorLogin("Unknown", "unknown@example.com"),
+        ""
+      );
+    });
+  });
+
+  describe("isLowPriorityFile()", () => {
+    it("should recognize lockfiles and documentation as low priority", () => {
+      assert.strictEqual(isLowPriorityFile("package-lock.json"), true);
+      assert.strictEqual(isLowPriorityFile("gradle.lockfile"), true);
+      assert.strictEqual(isLowPriorityFile("README.md"), true);
+    });
+
+    it("should recognize source code files as normal priority", () => {
+      assert.strictEqual(
+        isLowPriorityFile("sdks/java/core/src/main/java/Foo.java"),
+        false
+      );
+      assert.strictEqual(
+        isLowPriorityFile("sdks/python/apache_beam/io/kafka.py"),
+        false
+      );
+    });
+  });
+
+  describe("getRecentCommitsForFile()", () => {
+    it("should read real commits from Beam git repository for KafkaIO.java", () => {
+      const kafkaFile =
+        "sdks/java/io/kafka/src/main/java/org/apache/beam/sdk/io/kafka/KafkaIO.java";
+      const commits = getRecentCommitsForFile(kafkaFile, 3);
+
+      assert.strictEqual(
+        commits.length > 0,
+        true,
+        "Expected commits to be found for KafkaIO.java"
+      );
+      const firstCommit = commits[0];
+      assert.strictEqual(typeof firstCommit.hash, "string");
+      assert.strictEqual(firstCommit.hash.length, 40);
+      assert.strictEqual(typeof firstCommit.authorName, "string");
+      assert.strictEqual(typeof firstCommit.subject, "string");
+    });
+  });
+
+  describe("aggregateCandidates()", () => {
+    it("should rank candidates by commit count and exclude the PR author", () => {
+      const touchedFiles: TouchedFileContext[] = [
+        {
+          path: "fileA.java",
+          additions: 10,
+          deletions: 2,
+          changes: 12,
+          isNewFile: false,
+          recentCommits: [
+            {
+              hash: "h1",
+              authorName: "Alice",
+              authorEmail: "alice@example.com",
+              authorLogin: "alice",
+              date: "2026-08-01",
+              subject: "Commit 1",
+            },
+            {
+              hash: "h2",
+              authorName: "Bob",
+              authorEmail: "bob@example.com",
+              authorLogin: "bob",
+              date: "2026-08-02",
+              subject: "Commit 2",
+            },
+            {
+              hash: "h3",
+              authorName: "Alice",
+              authorEmail: "alice@example.com",
+              authorLogin: "alice",
+              date: "2026-08-03",
+              subject: "Commit 3",
+            },
+          ],
+        },
+        {
+          path: "fileB.java",
+          additions: 5,
+          deletions: 1,
+          changes: 6,
+          isNewFile: false,
+          recentCommits: [
+            {
+              hash: "h4",
+              authorName: "PrAuthor",
+              authorEmail: "prauthor@example.com",
+              authorLogin: "prauthor",
+              date: "2026-08-04",
+              subject: "Commit 4",
+            },
+            {
+              hash: "h5",
+              authorName: "Bob",
+              authorEmail: "bob@example.com",
+              authorLogin: "bob",
+              date: "2026-08-05",
+              subject: "Commit 5",
+            },
+          ],
+        },
+      ];
+
+      const candidates = aggregateCandidates(touchedFiles, "prauthor");
+
+      // PR author must be excluded
+      assert.strictEqual(
+        candidates.some((c) => c.login === "prauthor"),
+        false
+      );
+
+      // Alice has 2 commits, Bob has 2 commits
+      assert.strictEqual(candidates.length, 2);
+      assert.strictEqual(candidates[0].commitCount, 2);
+      assert.strictEqual(candidates[1].commitCount, 2);
+    });
+  });
+
+  describe("buildPrHistoryContext()", () => {
+    it("should build structured context from changed files", () => {
+      const kafkaFile =
+        "sdks/java/io/kafka/src/main/java/org/apache/beam/sdk/io/kafka/KafkaIO.java";
+      const context = buildPrHistoryContext(
+        12345,
+        "Fix KafkaIO poll issue",
+        "Resolves deadlock in consumer polling",
+        "contributorX",
+        [
+          {
+            filename: kafkaFile,
+            additions: 50,
+            deletions: 10,
+            changes: 60,
+            status: "modified",
+          },
+          {
+            filename: "README.md",
+            additions: 2,
+            deletions: 1,
+            changes: 3,
+            status: "modified",
+          },
+        ],
+        { maxFiles: 5, commitsPerFile: 3 }
+      );
+
+      assert.strictEqual(context.prNumber, 12345);
+      assert.strictEqual(context.author, "contributorX");
+      // KafkaIO should be prioritized above README.md
+      assert.strictEqual(context.touchedFiles[0].path, kafkaFile);
+      assert.strictEqual(
+        context.touchedFiles[0].recentCommits.length > 0,
+        true
+      );
+    });
+  });
+});
diff --git a/scripts/ci/pr-bot/test/prTest.ts b/scripts/ci/pr-bot/test/prTest.ts
index b771852584a3..3be12174c6c0 100644
--- a/scripts/ci/pr-bot/test/prTest.ts
+++ b/scripts/ci/pr-bot/test/prTest.ts
@@ -41,5 +41,12 @@ describe("Pr", function () {
       };
       assert.equal("", testPr.getLabelForReviewer("testReviewer4"));
     });
+
+    it("should preserve alternateReviewers when initialized", function () {
+      let testPr = new Pr({
+        alternateReviewers: ["backup1", "backup2"],
+      });
+      assert.deepStrictEqual(testPr.alternateReviewers, ["backup1", "backup2"]);
+    });
   });
 });