diff --git a/.github/workflows/dev.yml b/.github/workflows/dev.yml
index cd6d52094468e..56bcabeb9006d 100644
--- a/.github/workflows/dev.yml
+++ b/.github/workflows/dev.yml
@@ -88,7 +88,7 @@ jobs:
       - uses: actions/setup-node@949feb2413d6458794dcd2491c4babbbce0c15c1  # v7.1.0
         with:
           node-version: "24"
-      - run: node --test ci/scripts/test_fork_ci_status.cjs
+      - run: node --test ci/scripts/test_fork_ci_status.cjs ci/scripts/test_cancel_merge_queue_runs.cjs
 
   typos:
     name: Spell Check with Typos
diff --git a/.github/workflows/merge_queue_cleanup.yml b/.github/workflows/merge_queue_cleanup.yml
new file mode 100644
index 0000000000000..1d6b2fc9115e6
--- /dev/null
+++ b/.github/workflows/merge_queue_cleanup.yml
@@ -0,0 +1,75 @@
+# 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.
+
+name: Merge queue cleanup
+
+on:
+  # New groups trigger an early scan. Ref deletion can lag this event; the
+  # sweep also catches groups removed without a replacement or missed earlier.
+  merge_group:
+    types: [checks_requested]
+  schedule:
+    - cron: "*/15 * * * *"
+  workflow_dispatch:
+    inputs:
+      dry_run:
+        description: "Report obsolete runs without cancelling them"
+        type: boolean
+        default: true
+
+permissions:
+  contents: read
+  actions: write
+
+# Merge-group scans attach checks to commits that can land on main, so each
+# group SHA gets its own group: a replaced pending scan would show as cancelled.
+# Scheduled and manual scans share one group per mode, so a dry run cannot
+# replace a pending cancellation scan. Do not interrupt running scans.
+concurrency:
+  group: merge-queue-cleanup-${{ github.event_name == 'merge_group' && github.event.merge_group.head_sha || (vars.MERGE_QUEUE_CLEANUP_ENABLED != 'true' || (github.event_name == 'workflow_dispatch' && inputs.dry_run)) }}
+  cancel-in-progress: false
+
+jobs:
+  cleanup:
+    if: github.repository == 'apache/datafusion'
+    runs-on: ubuntu-slim
+    timeout-minutes: 10
+    steps:
+      - uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1
+        with:
+          ref: ${{ github.event.repository.default_branch }}
+          persist-credentials: false
+      - name: Cancel obsolete merge-group runs
+        # During the PR that introduces this workflow, main has no script yet.
+        # Execute main's copy of the script.
+        if: hashFiles('ci/scripts/cancel_merge_queue_runs.cjs') != ''
+        uses: actions/github-script@3a2844b7e9c422d3c10d287c895573f7108da1b3 # v9.0.0
+        env:
+          # Observe candidates for at least a week before enabling cancellation.
+          DRY_RUN: ${{ vars.MERGE_QUEUE_CLEANUP_ENABLED != 'true' || (github.event_name == 'workflow_dispatch' && inputs.dry_run) }}
+        with:
+          script: |
+            const { run } = require('./ci/scripts/cancel_merge_queue_runs.cjs');
+            const dryRun = process.env.DRY_RUN !== 'false';
+            const results = await run({ github, context, core, dryRun });
+            await core.summary
+              .addHeading(dryRun ? 'Would cancel merge-group runs' : 'Requested merge-group cancellations')
+              .addTable([
+                ['Run', 'SHA', 'Attempt', 'Observed at (UTC)', 'Reason'].map(data => ({ data, header: true })),
+                ...results.map(({ id, sha, attempt, observed_at, reason }) => [String(id), sha, String(attempt), observed_at, reason]),
+              ])
+              .write();
diff --git a/ci/scripts/cancel_merge_queue_runs.cjs b/ci/scripts/cancel_merge_queue_runs.cjs
new file mode 100644
index 0000000000000..9ebec8942a496
--- /dev/null
+++ b/ci/scripts/cancel_merge_queue_runs.cjs
@@ -0,0 +1,238 @@
+// 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.
+
+// Called on each new merge group and on a schedule. The queue's exact Git
+// refs, rather than PR state or failing checks, determine which runs are stale.
+const ACTIVE = new Set([
+  "queued",
+  "in_progress",
+  "waiting",
+  "pending",
+  "requested",
+]);
+const CLEANUP_PATH = ".github/workflows/merge_queue_cleanup.yml";
+const QUEUE_REF = /^gh-readonly-queue\/(.+)\/pr-[1-9][0-9]*-[0-9a-f]{40}$/;
+
+function eligible(run, repo, currentRunId) {
+  return (
+    run.id !== currentRunId &&
+    run.event === "merge_group" &&
+    run.run_attempt === 1 &&
+    ACTIVE.has(run.status) &&
+    run.path?.split("@")[0] !== CLEANUP_PATH &&
+    run.repository?.full_name?.toLowerCase() === repo.toLowerCase() &&
+    run.head_repository?.full_name?.toLowerCase() === repo.toLowerCase() &&
+    QUEUE_REF.test(run.head_branch) &&
+    /^[0-9a-f]{40}$/.test(run.head_sha)
+  );
+}
+
+async function refSha(github, repo, branch) {
+  try {
+    const { data } = await github.rest.git.getRef({
+      ...repo,
+      ref: `heads/${branch}`,
+    });
+    // Only a missing ref or a known, different commit proves obsolescence.
+    // Unexpected responses and permission/server errors must not cancel CI.
+    if (
+      data.object?.type !== "commit" ||
+      !/^[0-9a-f]{40}$/.test(data.object.sha)
+    ) {
+      throw new Error(`Invalid Git ref response for ${branch}`);
+    }
+    return data.object.sha;
+  } catch (error) {
+    if (error.status === 404) return null;
+    throw error;
+  }
+}
+
+// Filtered Actions listings are capped at 1,000 results even with pagination.
+// Active merge-group runs stay far below the cap, so a full listing is an error.
+async function listRuns(github, repo, status) {
+  const params = { ...repo, event: "merge_group", status, per_page: 100 };
+  const runs = [];
+  for (let page = 1; ; page++) {
+    const { data } = await github.rest.actions.listWorkflowRunsForRepo({
+      ...params,
+      page,
+    });
+    if (
+      !Number.isInteger(data.total_count) ||
+      data.total_count < 0 ||
+      !Array.isArray(data.workflow_runs)
+    ) {
+      throw new Error("Invalid workflow run listing");
+    }
+    if (data.total_count >= 1000) {
+      throw new Error(
+        `${data.total_count} ${status} runs exceed the listing cap`
+      );
+    }
+    runs.push(...data.workflow_runs);
+    if (data.workflow_runs.length < 100) return runs;
+  }
+}
+
+function reportError(core, context, message) {
+  core.warning(message);
+  // Maintenance errors must not mark a merge-group commit as failing CI.
+  // Scheduled/manual scans still expose broken cleanup as workflow failures.
+  if (context.eventName !== "merge_group") core.setFailed(message);
+}
+
+async function run({ github, context, core, dryRun = true }) {
+  const repo = { owner: context.repo.owner, repo: context.repo.repo };
+  const fullName = `${repo.owner}/${repo.repo}`;
+  if (fullName !== "apache/datafusion") return [];
+
+  // Finish pagination before cancelling: otherwise mutations shrink the active
+  // result sets and can cause later pages to skip runs.
+  const listed = new Map();
+  try {
+    for (const status of ACTIVE) {
+      const runs = await listRuns(github, repo, status);
+      for (const candidate of runs) listed.set(candidate.id, candidate);
+    }
+  } catch (error) {
+    reportError(
+      core,
+      context,
+      `Cannot list merge-group runs: ${error.message}`
+    );
+    return [];
+  }
+
+  const refs = new Map();
+  const comparisons = new Map();
+  const results = [];
+  const errors = [];
+  for (const candidate of listed.values()) {
+    if (!eligible(candidate, fullName, context.runId)) continue;
+    try {
+      // Share the initial lookup among workflows for the same merge group.
+      // Cache failures too: an outage must not multiply requests for this ref.
+      if (!refs.has(candidate.head_branch)) {
+        refs.set(
+          candidate.head_branch,
+          refSha(github, repo, candidate.head_branch)
+        );
+      }
+      if ((await refs.get(candidate.head_branch)) === candidate.head_sha) {
+        core.info(
+          `Kept active group ${JSON.stringify({
+            id: candidate.id,
+            sha: candidate.head_sha,
+            branch: candidate.head_branch,
+            observed_at: new Date().toISOString(),
+            trigger: context.eventName,
+          })}`
+        );
+        continue;
+      }
+
+      const { data: current } = await github.rest.actions.getWorkflowRun({
+        ...repo,
+        run_id: candidate.id,
+      });
+      // A completed run or a newly requested rerun is no longer this candidate.
+      if (
+        !eligible(current, fullName, context.runId) ||
+        current.head_branch !== candidate.head_branch ||
+        current.head_sha !== candidate.head_sha ||
+        current.run_attempt !== candidate.run_attempt
+      )
+        continue;
+
+      const head = await refSha(github, repo, current.head_branch);
+      if (head === current.head_sha) continue;
+      // A successful merge removes the queue ref too. Preserve checks on
+      // commits already in the target branch, including optional checks.
+      // Workflows of one group share the comparison, failures included.
+      const base = QUEUE_REF.exec(current.head_branch)[1];
+      const basehead = `${base}...${current.head_sha}`;
+      if (!comparisons.has(basehead)) {
+        comparisons.set(
+          basehead,
+          github.rest.repos.compareCommitsWithBasehead({
+            ...repo,
+            basehead,
+            per_page: 1,
+          })
+        );
+      }
+      const { data: comparison } = await comparisons.get(basehead);
+      if (["identical", "behind"].includes(comparison.status)) continue;
+      if (!["ahead", "diverged"].includes(comparison.status)) {
+        throw new Error(`Invalid comparison status for ${current.head_sha}`);
+      }
+      const reason = head === null ? "queue ref removed" : "queue ref replaced";
+      const result = {
+        id: current.id,
+        sha: current.head_sha,
+        branch: current.head_branch,
+        attempt: current.run_attempt,
+        reason,
+        observed_at: new Date().toISOString(),
+        trigger: context.eventName,
+      };
+      const description = JSON.stringify(result);
+      if (dryRun) {
+        core.info(`Would cancel ${description}`);
+      } else {
+        try {
+          await github.rest.actions.cancelWorkflowRun({
+            ...repo,
+            run_id: current.id,
+          });
+        } catch (error) {
+          // A 409 can mean a completion race, but also an active run that
+          // GitHub cannot cancel. Only suppress it after confirming the race.
+          if (error.status === 409) {
+            const { data: latest } = await github.rest.actions.getWorkflowRun({
+              ...repo,
+              run_id: current.id,
+            });
+            if (
+              latest.status === "completed" ||
+              latest.run_attempt !== current.run_attempt
+            )
+              continue;
+          }
+          throw error;
+        }
+        core.info(`Requested cancellation of ${description}`);
+      }
+      results.push(result);
+    } catch (error) {
+      // One failed read/cancellation must not leave other obsolete groups running.
+      const message = `Run ${candidate.id}: ${error.message}`;
+      core.warning(message);
+      errors.push(message);
+    }
+  }
+  if (errors.length)
+    reportError(
+      core,
+      context,
+      `${errors.length} merge queue cleanup errors; see warnings.`
+    );
+  return results;
+}
+
+module.exports = { run };
diff --git a/ci/scripts/test_cancel_merge_queue_runs.cjs b/ci/scripts/test_cancel_merge_queue_runs.cjs
new file mode 100755
index 0000000000000..64d91ed26f266
--- /dev/null
+++ b/ci/scripts/test_cancel_merge_queue_runs.cjs
@@ -0,0 +1,484 @@
+#!/usr/bin/env node
+// 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.
+
+const assert = require("node:assert/strict");
+const { test } = require("node:test");
+const { run } = require("./cancel_merge_queue_runs.cjs");
+
+const sha = "a".repeat(40);
+const otherSha = "b".repeat(40);
+const branch = `gh-readonly-queue/main/pr-123-${"c".repeat(40)}`;
+const fullName = "apache/datafusion";
+function workflow(id = 1, overrides = {}) {
+  return {
+    id,
+    event: "merge_group",
+    status: "in_progress",
+    path: ".github/workflows/rust.yml",
+    repository: { full_name: fullName },
+    head_repository: { full_name: fullName },
+    head_branch: branch,
+    head_sha: sha,
+    run_attempt: 1,
+    ...overrides,
+  };
+}
+const apiError = (status) =>
+  Object.assign(new Error(`HTTP ${status}`), { status });
+
+async function execute({
+  runs = [workflow()],
+  refs = {},
+  current = {},
+  cancelErrors = {},
+  dryRun = false,
+  omitDryRun = false,
+  owner = "apache",
+  listError,
+  listResponse,
+  comparisons = {},
+  eventName = "schedule",
+} = {}) {
+  const calls = [];
+  const cancellations = [];
+  const failures = [];
+  const warnings = [];
+  const logs = [];
+  const refReads = new Map();
+  const runReads = new Map();
+  const actions = {
+    listWorkflowRunsForRepo: async (params) => {
+      assert.equal(params.event, "merge_group");
+      assert.equal(params.per_page, 100);
+      assert.equal(`${params.owner}/${params.repo}`, fullName);
+      calls.push(`list:${params.status}`);
+      if (params.status === listError) throw apiError(503);
+      if (listResponse) return { data: await listResponse(params) };
+      assert.equal(params.created, undefined);
+      const filtered = runs.filter((r) => r.status === params.status);
+      return {
+        data: {
+          total_count: filtered.length,
+          workflow_runs: filtered.slice(
+            (params.page - 1) * 100,
+            params.page * 100
+          ),
+        },
+      };
+    },
+    getWorkflowRun: async ({ run_id }) => {
+      calls.push(`get:${run_id}`);
+      const index = runReads.get(run_id) ?? 0;
+      runReads.set(run_id, index + 1);
+      let value = current[run_id] ?? runs.find((r) => r.id === run_id);
+      if (Array.isArray(value))
+        value = value[Math.min(index, value.length - 1)];
+      if (value instanceof Error) throw value;
+      return { data: value };
+    },
+    cancelWorkflowRun: async ({ owner, repo, run_id }) => {
+      assert.equal(`${owner}/${repo}`, fullName);
+      calls.push(`cancel:${run_id}`);
+      if (cancelErrors[run_id]) throw cancelErrors[run_id];
+      cancellations.push(run_id);
+    },
+  };
+  const github = {
+    rest: {
+      actions,
+      repos: {
+        compareCommitsWithBasehead: async ({ basehead, owner, repo }) => {
+          assert.equal(`${owner}/${repo}`, fullName);
+          calls.push(`compare:${basehead}`);
+          const response = comparisons[basehead] ?? "diverged";
+          if (response instanceof Error) throw response;
+          return { data: { status: response } };
+        },
+      },
+      git: {
+        getRef: async ({ ref, owner, repo }) => {
+          assert.equal(`${owner}/${repo}`, fullName);
+          calls.push(`ref:${ref}`);
+          const index = refReads.get(ref) ?? 0;
+          refReads.set(ref, index + 1);
+          let value = refs[ref];
+          if (Array.isArray(value))
+            value = value[Math.min(index, value.length - 1)];
+          if (value instanceof Error) throw value;
+          if (value === undefined || value === null) throw apiError(404);
+          return {
+            data:
+              typeof value === "string"
+                ? { object: { type: "commit", sha: value } }
+                : value,
+          };
+        },
+      },
+    },
+  };
+  const results = await run({
+    github,
+    context: { repo: { owner, repo: "datafusion" }, runId: 999, eventName },
+    core: {
+      info: (s) => logs.push(s),
+      warning: (s) => warnings.push(s),
+      setFailed: (s) => failures.push(s),
+    },
+    ...(omitDryRun ? {} : { dryRun }),
+  });
+  return { results, cancellations, calls, failures, warnings, logs };
+}
+
+test("keeps every workflow of an active group, including failed-check siblings", async () => {
+  const result = await execute({
+    runs: [
+      workflow(1),
+      workflow(2, { path: ".github/workflows/extended.yml" }),
+    ],
+    refs: { [`heads/${branch}`]: sha },
+  });
+  assert.deepEqual(result.cancellations, []);
+  assert.equal(result.calls.filter((c) => c.startsWith("ref:")).length, 1);
+});
+
+test("cancels all workflows of a removed group after collecting every active status", async () => {
+  const statuses = ["queued", "in_progress", "waiting", "pending", "requested"];
+  const result = await execute({
+    runs: statuses.map((status, i) => workflow(i + 1, { status })),
+  });
+  assert.deepEqual(result.cancellations.sort(), [1, 2, 3, 4, 5]);
+  assert.deepEqual(
+    result.calls.slice(0, 5),
+    statuses.map((s) => `list:${s}`)
+  );
+  assert.ok(result.results.every((r) => r.reason === "queue ref removed"));
+});
+
+test("cancels an old SHA when its queue ref points at a replacement", async () => {
+  const result = await execute({
+    runs: [workflow(1), workflow(2, { head_sha: otherSha })],
+    refs: { [`heads/${branch}`]: otherSha },
+  });
+  assert.deepEqual(result.cancellations, [1]);
+  assert.equal(result.results[0].reason, "queue ref replaced");
+});
+
+test("new base ref for one PR does not cancel its replacement or another PR", async () => {
+  const replacement = `gh-readonly-queue/main/pr-123-${"d".repeat(40)}`;
+  const unrelated = `gh-readonly-queue/main/pr-456-${"e".repeat(40)}`;
+  const result = await execute({
+    runs: [
+      workflow(1),
+      workflow(2, { head_branch: replacement }),
+      workflow(3, { head_branch: unrelated }),
+    ],
+    refs: { [`heads/${replacement}`]: sha, [`heads/${unrelated}`]: sha },
+  });
+  assert.deepEqual(result.cancellations, [1]);
+});
+
+test("checks the full queue ref, including a base branch containing slashes", async () => {
+  const release = `gh-readonly-queue/release/56/pr-123-${"c".repeat(40)}`;
+  const result = await execute({
+    runs: [workflow(1), workflow(2, { head_branch: release })],
+    refs: { [`heads/${release}`]: sha },
+  });
+  assert.deepEqual(result.cancellations, [1]);
+});
+
+test("ignores non-queue events, foreign repos, malformed refs, and cleanup runs", async () => {
+  const overrides = [
+    { event: "push" },
+    { event: "pull_request" },
+    { event: "workflow_dispatch" },
+    { head_repository: { full_name: "contributor/datafusion" } },
+    { repository: { full_name: "contributor/datafusion" } },
+    { head_repository: null },
+    { head_branch: "main" },
+    { head_branch: "gh-readonly-queue/main/pr-123-invalid" },
+    { head_sha: "invalid" },
+    { path: ".github/workflows/merge_queue_cleanup.yml" },
+    { path: ".github/workflows/merge_queue_cleanup.yml@refs/heads/main" },
+    { id: 999 },
+    { status: "completed" },
+  ];
+  const result = await execute({
+    runs: overrides.map((value, i) => workflow(i + 1, value)),
+  });
+  assert.deepEqual(result.cancellations, []);
+  assert.equal(result.calls.length, 5);
+});
+
+test("does nothing on a fork", async () => {
+  const result = await execute({ owner: "contributor" });
+  assert.deepEqual(result.calls, []);
+});
+
+test("dry run performs the same checks without mutations", async () => {
+  const result = await execute({ dryRun: true });
+  assert.deepEqual(result.cancellations, []);
+  assert.equal(result.results[0].id, 1);
+  assert.equal(result.results[0].sha, sha);
+  assert.equal(result.results[0].attempt, 1);
+  assert.equal(result.results[0].reason, "queue ref removed");
+  assert.ok(Number.isFinite(Date.parse(result.results[0].observed_at)));
+  assert.match(result.logs[0], /Would cancel/);
+});
+
+test("defaults to a dry run when the caller does not specify a mode", async () => {
+  const result = await execute({ omitDryRun: true });
+  assert.deepEqual(result.cancellations, []);
+  assert.equal(result.results.length, 1);
+  assert.match(result.logs[0], /Would cancel/);
+});
+
+for (const status of [401, 403, 429, 500]) {
+  test(`a ${status} ref lookup cannot be mistaken for a removed ref`, async () => {
+    const result = await execute({
+      refs: { [`heads/${branch}`]: apiError(status) },
+    });
+    assert.deepEqual(result.cancellations, []);
+    assert.equal(result.failures.length, 1);
+  });
+}
+
+test("malformed ref responses cannot cancel a run", async () => {
+  for (const response of [
+    {},
+    { object: { type: "tag", sha: otherSha } },
+    { object: { type: "commit", sha: "" } },
+  ]) {
+    const result = await execute({ refs: { [`heads/${branch}`]: response } });
+    assert.deepEqual(result.cancellations, []);
+    assert.equal(result.failures.length, 1);
+  }
+});
+
+test("rechecks refs and keeps a group that is active again", async () => {
+  const result = await execute({ refs: { [`heads/${branch}`]: [null, sha] } });
+  assert.deepEqual(result.cancellations, []);
+});
+
+test("a failed second ref read cannot cancel a run", async () => {
+  const result = await execute({
+    refs: { [`heads/${branch}`]: [null, apiError(503)] },
+  });
+  assert.deepEqual(result.cancellations, []);
+  assert.equal(result.failures.length, 1);
+});
+
+test("does not cancel completed runs, newer attempts, or changed run identities", async () => {
+  for (const update of [
+    { status: "completed" },
+    { run_attempt: 2 },
+    { head_sha: otherSha },
+    { head_branch: `gh-readonly-queue/main/pr-456-${"c".repeat(40)}` },
+    { event: "push" },
+  ]) {
+    const result = await execute({ current: { 1: workflow(1, update) } });
+    assert.deepEqual(result.cancellations, []);
+  }
+});
+
+test("409 on cancellation is a harmless completion race", async () => {
+  const result = await execute({
+    current: { 1: [workflow(), workflow(1, { status: "completed" })] },
+    cancelErrors: { 1: apiError(409) },
+  });
+  assert.deepEqual(result.failures, []);
+  assert.deepEqual(result.results, []);
+});
+
+test("409 for an unchanged active run reports a failed cancellation", async () => {
+  const result = await execute({ cancelErrors: { 1: apiError(409) } });
+  assert.equal(result.failures.length, 1);
+  assert.match(result.warnings[0], /HTTP 409/);
+});
+
+test("409 after a new rerun starts leaves that attempt alone", async () => {
+  const result = await execute({
+    current: { 1: [workflow(), workflow(1, { run_attempt: 2 })] },
+    cancelErrors: { 1: apiError(409) },
+  });
+  assert.deepEqual(result.failures, []);
+  assert.deepEqual(result.results, []);
+});
+
+test("409 on a read is still an error", async () => {
+  const result = await execute({ current: { 1: apiError(409) } });
+  assert.deepEqual(result.cancellations, []);
+  assert.equal(result.failures.length, 1);
+});
+
+test("one cancellation failure does not stop cleanup of other runs", async () => {
+  const result = await execute({
+    runs: [workflow(1), workflow(2)],
+    cancelErrors: { 1: apiError(403) },
+  });
+  assert.deepEqual(result.cancellations, [2]);
+  assert.equal(result.failures.length, 1);
+});
+
+test("one failed group lookup does not stop cleanup of another group", async () => {
+  const result = await execute({
+    runs: [
+      workflow(1),
+      workflow(2, {
+        head_branch: `gh-readonly-queue/main/pr-456-${"c".repeat(40)}`,
+      }),
+    ],
+    refs: { [`heads/${branch}`]: apiError(503) },
+  });
+  assert.deepEqual(result.cancellations, [2]);
+  assert.equal(result.failures.length, 1);
+});
+
+test("a listing failure aborts before any cancellation", async () => {
+  const result = await execute({ listError: "requested" });
+  assert.deepEqual(result.cancellations, []);
+  assert.equal(result.failures.length, 1);
+  assert.match(result.warnings[0], /HTTP 503/);
+});
+
+test("uses all paginated results and deduplicates repeated run IDs", async () => {
+  const runs = Array.from({ length: 205 }, (_, i) => workflow(i + 1));
+  runs.push(runs[0]);
+  const result = await execute({ runs });
+  assert.equal(result.cancellations.length, 205);
+  assert.equal(new Set(result.cancellations).size, 205);
+});
+
+test("keeps merged commits even when their queue ref has disappeared or changed", async () => {
+  for (const status of ["identical", "behind"]) {
+    for (const ref of [null, otherSha]) {
+      const result = await execute({
+        comparisons: { [`main...${sha}`]: status },
+        refs: { [`heads/${branch}`]: ref },
+      });
+      assert.deepEqual(result.cancellations, []);
+      assert.deepEqual(result.failures, []);
+    }
+  }
+});
+
+test("compares against the actual target branch, including slashes", async () => {
+  const result = await execute({
+    runs: [
+      workflow(1, {
+        head_branch: `gh-readonly-queue/release/56/pr-123-${sha}`,
+      }),
+    ],
+    comparisons: { [`release/56...${sha}`]: "behind" },
+  });
+  assert.deepEqual(result.cancellations, []);
+  assert.ok(result.calls.includes(`compare:release/56...${sha}`));
+});
+
+test("only ahead or diverged comparisons permit cancellation", async () => {
+  for (const status of ["ahead", "diverged"]) {
+    const result = await execute({
+      comparisons: { [`main...${sha}`]: status },
+    });
+    assert.deepEqual(result.cancellations, [1]);
+  }
+  for (const status of [
+    "unknown",
+    "",
+    apiError(404),
+    apiError(403),
+    apiError(503),
+  ]) {
+    const result = await execute({
+      comparisons: { [`main...${sha}`]: status },
+    });
+    assert.deepEqual(result.cancellations, []);
+    assert.equal(result.failures.length, 1);
+  }
+});
+
+test("reruns are exempt across successive scans, not just on attempt changes", async () => {
+  for (let scan = 0; scan < 2; scan++) {
+    const result = await execute({
+      runs: [workflow(1, { run_attempt: 2 }), workflow(2)],
+    });
+    assert.deepEqual(result.cancellations, [2]);
+    assert.ok(!result.calls.includes("get:1"));
+  }
+});
+
+test("merge-group maintenance errors warn without failing the commit", async () => {
+  for (const options of [
+    { listError: "requested" },
+    { refs: { [`heads/${branch}`]: apiError(403) } },
+    { comparisons: { [`main...${sha}`]: apiError(503) } },
+    { cancelErrors: { 1: apiError(409) } },
+  ]) {
+    const result = await execute({ eventName: "merge_group", ...options });
+    assert.deepEqual(result.failures, []);
+    assert.ok(result.warnings.length > 0);
+    assert.deepEqual(result.cancellations, []);
+  }
+});
+
+test("manual cleanup failures remain visible", async () => {
+  const result = await execute({
+    eventName: "workflow_dispatch",
+    cancelErrors: { 1: apiError(409) },
+  });
+  assert.equal(result.failures.length, 1);
+});
+
+test("an unresolved 409 is retried by later scans and is never reported as cancelled", async () => {
+  const first = await execute({ cancelErrors: { 1: apiError(409) } });
+  assert.deepEqual(first.results, []);
+  assert.equal(first.failures.length, 1);
+  const next = await execute();
+  assert.deepEqual(next.cancellations, [1]);
+});
+
+test("a capped listing aborts without cancelling", async () => {
+  const result = await execute({
+    runs: Array.from({ length: 1000 }, (_, i) => workflow(i + 1)),
+  });
+  assert.deepEqual(result.cancellations, []);
+  assert.deepEqual(
+    result.calls.filter((c) => !c.startsWith("list:")),
+    []
+  );
+  assert.equal(result.failures.length, 1);
+  assert.match(result.warnings[0], /exceed the listing cap/);
+});
+
+test("workflows of one group share the target-branch comparison", async () => {
+  const result = await execute({
+    runs: [
+      workflow(1),
+      workflow(2, { path: ".github/workflows/extended.yml" }),
+    ],
+  });
+  assert.deepEqual(result.cancellations, [1, 2]);
+  assert.equal(result.calls.filter((c) => c.startsWith("compare:")).length, 1);
+});
+
+test("malformed listing metadata aborts without cancelling", async () => {
+  const result = await execute({
+    listResponse: () => ({ workflow_runs: [workflow()] }),
+  });
+  assert.deepEqual(result.cancellations, []);
+  assert.equal(result.failures.length, 1);
+});
diff --git a/dev/rust_lint.sh b/dev/rust_lint.sh
index b1dddaac3b65e..116bbf0b756f9 100755
--- a/dev/rust_lint.sh
+++ b/dev/rust_lint.sh
@@ -139,6 +139,7 @@ declare -a READONLY_STEPS=(
   "ci/scripts/check_rust_workflow_skips.py|false"
   "ci/scripts/test_rust_workflow_skips.py|false"
   "ci/scripts/test_fork_ci_status.cjs|false"
+  "ci/scripts/test_cancel_merge_queue_runs.cjs|false"
   "ci/scripts/check_large_files.sh|false"
   "ci/scripts/markdown_link_check.sh|false"
   "ci/scripts/security_audit.sh|false"
diff --git a/docs/source/contributor-guide/index.md b/docs/source/contributor-guide/index.md
index a9fc48b4bac46..5df074618de08 100644
--- a/docs/source/contributor-guide/index.md
+++ b/docs/source/contributor-guide/index.md
@@ -284,3 +284,17 @@ To make this work:
    synced fork builds much faster.
 3. Rebase your branch before you push. The fork run tests your branch as it
    is, while the merge queue tests it merged with `main`.
+
+#### Obsolete Merge Queue Runs
+
+The **Merge queue cleanup** workflow finds unfinished merge queue CI runs whose
+queue branch was removed or now points to a different commit. It scans when a
+new merge group is created and every 15 minutes.
+
+Cleanup keeps runs for active groups, commits already in their target branch,
+and rerun attempts. Contributors can rerun an obsolete job to investigate a
+failure. A failed check alone does not cause cancellation.
+
+The workflow only reports candidates until a maintainer sets the repository
+Actions variable `MERGE_QUEUE_CLEANUP_ENABLED` to `true`. Manual runs default
+to `dry_run`. To stop cancellation, remove the variable or set it to `false`.