diff --git a/.github/workflows/pr-bot-new-prs.yml b/.github/workflows/pr-bot-new-prs.yml
index 9c924c423652..30bf436a9ead 100644
--- a/.github/workflows/pr-bot-new-prs.yml
+++ b/.github/workflows/pr-bot-new-prs.yml
@@ -47,4 +47,5 @@ jobs:
       - run: npm run processNewPrs
         env:
           GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }}
+          GEMINI_API_KEY: ${{ secrets.GEMINI_API_KEY }}
         working-directory: 'scripts/ci/pr-bot'
diff --git a/scripts/ci/pr-bot/dryRunAdvisor.ts b/scripts/ci/pr-bot/dryRunAdvisor.ts
new file mode 100644
index 000000000000..ff0cdce71e56
--- /dev/null
+++ b/scripts/ci/pr-bot/dryRunAdvisor.ts
@@ -0,0 +1,145 @@
+/*
+ * 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 { buildPrHistoryContext } from "./shared/gitHistory";
+import {
+  GeminiReviewerAdvisor,
+  GeminiClient,
+} from "./shared/geminiReviewerAdvisor";
+import { assignReviewersWithExpertise } from "./shared/commentStrings";
+
+/**
+ * Dry-run script to demonstrate and evaluate the Gemini Reviewer Assigner locally.
+ *
+ * Usage:
+ *   node lib/dryRunAdvisor.js [file1] [file2] ...
+ *
+ * If no files are specified, defaults to representative Beam files (e.g. KafkaIO).
+ */
+async function runDryRun() {
+  const customFiles = process.argv.slice(2);
+
+  const defaultFiles = [
+    {
+      filename:
+        "sdks/java/io/kafka/src/main/java/org/apache/beam/sdk/io/kafka/KafkaIO.java",
+      additions: 85,
+      deletions: 12,
+      changes: 97,
+      status: "modified",
+    },
+    {
+      filename:
+        "sdks/java/io/kafka/src/test/java/org/apache/beam/sdk/io/kafka/KafkaIOTest.java",
+      additions: 40,
+      deletions: 5,
+      changes: 45,
+      status: "modified",
+    },
+  ];
+
+  const filesToEvaluate =
+    customFiles.length > 0
+      ? customFiles.map((f) => ({
+          filename: f,
+          additions: 50,
+          deletions: 10,
+          changes: 60,
+          status: "modified",
+        }))
+      : defaultFiles;
+
+  console.log("=================================================");
+  console.log(" Beam LLM Review Assigner — Dry Run Prototype");
+  console.log("=================================================\n");
+
+  console.log("1. Extracting git history for touched files...");
+  for (const f of filesToEvaluate) {
+    console.log(`   - ${f.filename}`);
+  }
+
+  const prContext = buildPrHistoryContext(
+    39999,
+    "KafkaIO: Optimize consumer polling and watermark estimation",
+    "Refactors the reader loop to prevent deadlocks and improve dynamic backlog tracking.",
+    "sampleAuthor",
+    filesToEvaluate
+  );
+
+  console.log(
+    `\n2. Found ${prContext.candidates.length} candidate contributors in git history:`
+  );
+  for (const c of prContext.candidates.slice(0, 5)) {
+    console.log(
+      `   • @${c.login || c.email} (${c.name}): ${
+        c.commitCount
+      } commits, last active ${c.lastCommitDate}`
+    );
+  }
+
+  const apiKey = process.env.GEMINI_API_KEY || "";
+  const advisor = new GeminiReviewerAdvisor({
+    geminiClient: apiKey ? new GeminiClient(apiKey) : undefined,
+    committerCheck: async (login) =>
+      [
+        "kennknowles",
+        "chamikaramj",
+        "jrmccluskey",
+        "johnjcasey",
+        "damccorm",
+      ].includes(login.toLowerCase()),
+  });
+
+  console.log(
+    `\n3. Evaluating candidate expertise (${
+      apiKey ? "using Gemini API" : "using familiarity heuristic fallback"
+    })...\n`
+  );
+
+  const advice = await advisor.adviseReviewers(prContext);
+
+  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("==========================================================");
+}
+
+runDryRun().catch((err) => {
+  console.error("Dry run encountered error:", err);
+  process.exit(1);
+});
diff --git a/scripts/ci/pr-bot/package.json b/scripts/ci/pr-bot/package.json
index 5fc8d79c1dd1..49f228fe8285 100644
--- a/scripts/ci/pr-bot/package.json
+++ b/scripts/ci/pr-bot/package.json
@@ -11,7 +11,8 @@
     "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",
+    "dryRun": "npm run build && node lib/dryRunAdvisor.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..f3df8fec692a 100644
--- a/scripts/ci/pr-bot/processNewPrs.ts
+++ b/scripts/ci/pr-bot/processNewPrs.ts
@@ -29,6 +29,12 @@ const {
   REVIEWERS_ACTION,
 } = require("./shared/constants");
 import { CheckStatus } from "./shared/checks";
+import { buildPrHistoryContext } from "./shared/gitHistory";
+import {
+  GeminiReviewerAdvisor,
+  GeminiClient,
+  ReviewerAdviceResult,
+} from "./shared/geminiReviewerAdvisor";
 
 /*
  * Returns true if the pr needs to be processed or false otherwise.
@@ -167,7 +173,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 +202,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 +218,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 +247,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 +286,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 +335,82 @@ 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)
+  // 1. Attempt LLM / Git History based expert reviewer selection
+  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 apiKey = process.env.GEMINI_API_KEY || "";
+    const advisor = new GeminiReviewerAdvisor({
+      geminiClient: apiKey ? new GeminiClient(apiKey) : undefined,
+      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;
+  }
+
+  // Fallback: 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..3ab8481cc77c 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,40 @@ 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
+- \`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..e22b56b89299
--- /dev/null
+++ b/scripts/ci/pr-bot/shared/geminiReviewerAdvisor.ts
@@ -0,0 +1,426 @@
+/*
+ * 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 {
+  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: "gemini" | "heuristic-fallback";
+}
+
+/**
+ * Interface for LLM clients that can generate structured JSON.
+ */
+export interface IGeminiClient {
+  generateJson<T>(prompt: string): Promise<T>;
+}
+
+/**
+ * Standard HTTP Gemini client using global fetch.
+ */
+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("GEMINI_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(
+        `Gemini API request failed with status ${response.status}: ${errorText}`
+      );
+    }
+
+    const data: any = await response.json();
+    const candidateText = data?.candidates?.[0]?.content?.parts?.[0]?.text;
+
+    if (!candidateText) {
+      throw new Error("Empty or invalid candidate response from Gemini API.");
+    }
+
+    return JSON.parse(candidateText) as T;
+  }
+}
+
+/**
+ * Configuration options for the Gemini Reviewer Advisor.
+ */
+export interface ReviewerAdvisorOptions {
+  readonly geminiClient?: IGeminiClient;
+  readonly committerCheck?: (username: string) => Promise<boolean>;
+  readonly exclusionList?: readonly string[];
+  readonly maxReviewers?: number;
+}
+
+/**
+ * Advisor that analyzes PR git history and selects optimal reviewers using Gemini 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.geminiClient;
+    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 Gemini based on git history relevance.",
+        source: "gemini",
+      };
+    } catch (error) {
+      console.warn(
+        `Error during Gemini 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..145446634430 100644
--- a/scripts/ci/pr-bot/shared/userCommand.ts
+++ b/scripts/ci/pr-bot/shared/userCommand.ts
@@ -18,7 +18,7 @@
 
 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");
 
@@ -41,7 +41,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);
@@ -86,6 +86,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 +221,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;
diff --git a/scripts/ci/pr-bot/test/commentStringsTest.ts b/scripts/ci/pr-bot/test/commentStringsTest.ts
new file mode 100644
index 000000000000..0b3b0320c25c
--- /dev/null
+++ b/scripts/ci/pr-bot/test/commentStringsTest.ts
@@ -0,0 +1,104 @@
+/*
+ * 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);
+    });
+
+    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..962fca846501
--- /dev/null
+++ b/scripts/ci/pr-bot/test/geminiReviewerAdvisorTest.ts
@@ -0,0 +1,219 @@
+/*
+ * 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, "gemini");
+      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 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"]);
+    });
   });
 });