diff --git a/dev/breeze/src/airflow_breeze/commands/ci_commands.py b/dev/breeze/src/airflow_breeze/commands/ci_commands.py
index 08b82f07c2ebe..46c1588827dc6 100644
--- a/dev/breeze/src/airflow_breeze/commands/ci_commands.py
+++ b/dev/breeze/src/airflow_breeze/commands/ci_commands.py
@@ -64,6 +64,7 @@
 
 if TYPE_CHECKING:
     from github import Github
+    from github.IssueEvent import IssueEvent
     from github.Repository import Issue, Milestone, Repository
 
 
@@ -1008,6 +1009,49 @@ def upgrade(
         console_print("[info]PR creation skipped. Changes are committed locally.[/]")
 
 
+# ---------------------------------------------------------------------------
+# Milestone auto-tagging helpers (used by the ``set-milestone`` command and the
+# ``milestone-tag-assistant.yml`` workflow).
+#
+# Skip-decision pipeline
+# ----------------------
+# After fetching the PR (so the ``Issue`` is in scope), the ``set-milestone``
+# command runs these three checks IN ORDER and stops on the first one that
+# returns "skip":
+#
+# 1. **Milestone-already-set guard** (in ``set_milestone``): if
+#    ``issue.milestone`` is non-null, leave the existing milestone alone and
+#    exit.
+# 2. **GitHub events evaluation** (:func:`_should_skip_milestone_tagging`):
+#    if any ``unlabeled`` event in ``issue.get_events()`` removed a
+#    ``backport-to-*`` label AND no ``backport-to-*`` label remains on the
+#    PR, treat the removal as an explicit "don't auto-tag" signal and skip.
+#    GitHub repo settings restrict label changes to committers/triagers, so
+#    the actor's permission is NOT re-verified here — the change is
+#    implicitly authorised.
+# 3. **Static skip labels** (:func:`_should_skip_milestone_tagging`): if any
+#    ``MILESTONE_SKIP_LABELS`` label is currently on the PR, skip.
+#
+# The events check is intentionally ordered before the static-label check so
+# that an explicit maintainer "unbackport" overrides every other condition,
+# including the merge-to-version-branch heuristic that
+# :func:`_determine_milestone_version` would otherwise apply.
+#
+# Inputs the decision uses
+# ------------------------
+# - ``labels`` — the labels currently on the PR (live from ``issue.labels``,
+#   fetched fresh by the caller; the workflow's ``--pr-labels`` snapshot is
+#   no longer consulted for the decision).
+# - ``events`` — the issue events stream from ``issue.get_events()``,
+#   fetched fresh by the caller. Pass ``None`` to disable the events check
+#   (e.g. when the events fetch failed and only the static check should
+#   run).
+#
+# All milestone helpers log via :func:`console_print` with the workflow's
+# Rich console; callers do not need to add their own skip-related log lines.
+# ---------------------------------------------------------------------------
+
+
 VERSION_BRANCH_PATTERN = re.compile(r"^v(\d+)-(\d+)-test$")
 BACKPORT_LABEL_PATTERN = re.compile(r"^backport-to-v(\d+)-(\d+)-test$")
 
@@ -1139,9 +1183,85 @@ def _has_bug_fix_indicators(title: str, labels: list[str]) -> bool:
     return False
 
 
-def _should_skip_milestone_tagging(labels: list[str]) -> bool:
-    """Check if the PR should be skipped from milestone auto-tagging."""
-    return bool(set(labels) & MILESTONE_SKIP_LABELS)
+def _get_removed_backport_labels_from_events(events: Iterable[IssueEvent]) -> set[str]:
+    """Return ``backport-to-*`` labels that appear in any ``unlabeled`` event.
+
+    Scans the issue events stream for ``unlabeled`` events on ``backport-to-*``
+    labels and returns the set of label names that were removed at any point
+    in the PR's lifecycle. The actor is intentionally NOT checked: GitHub
+    repo settings restrict label changes to users with write or triage
+    access, so any ``unlabeled`` event is implicitly authorised.
+
+    Combined with the live ``backport-to-*`` labels still on the PR, this is
+    enough to distinguish:
+
+    - Full removal (event present, no current backport) → skip signal.
+    - Replacement (event present, different current backport) → no skip;
+      the caller's regular evaluation picks up the new label.
+    - No removal (no event) → no skip.
+    """
+    return {
+        event.label.name
+        for event in events
+        if getattr(event, "event", None) == "unlabeled"
+        and getattr(getattr(event, "label", None), "name", "").startswith("backport-to-")
+    }
+
+
+def _should_skip_milestone_tagging(
+    labels: list[str],
+    events: Iterable[IssueEvent] | None = None,
+) -> bool:
+    """Decide whether milestone auto-tagging should be skipped for the PR.
+
+    Single decision point for the ``set-milestone`` command, covering
+    checks 2 and 3 of the pipeline documented in the module-level comment
+    above. Check 1 (milestone-already-set guard) is handled separately in
+    :func:`set_milestone` because it operates on ``issue.milestone`` rather
+    than labels. Applies the checks IN ORDER and short-circuits on the
+    first hit; each skip outcome emits its own ``console_print`` info log
+    so callers only need to react to the boolean return.
+
+    1. **GitHub events evaluation** (when ``events`` is supplied). If any
+       ``unlabeled`` event in ``events`` removed a ``backport-to-*`` label
+       AND ``labels`` contains no ``backport-to-*`` label now, treat the
+       removal as an explicit "don't auto-tag" signal and skip. Pure label
+       replacement (e.g. ``backport-to-v3-1-test`` swapped for
+       ``backport-to-v3-2-test``) leaves a backport on the PR and does NOT
+       trigger the skip — the caller's regular evaluation picks up the new
+       label. The unlabel actor's permission is NOT re-verified: GitHub
+       repo settings already restrict label changes to committers/triagers.
+    2. **Static skip labels.** If any ``MILESTONE_SKIP_LABELS`` label is
+       in ``labels``, skip.
+    3. Otherwise, do not skip.
+
+    The events check is ordered first so that an explicit unbackport
+    overrides every other condition, including the merge-to-version-branch
+    heuristic that :func:`_determine_milestone_version` would otherwise
+    apply.
+
+    :param labels: Labels currently on the PR (live, from ``issue.labels``).
+    :param events: Issue events stream (live, from ``issue.get_events()``).
+        Pass ``None`` to disable the events check — e.g. when the events
+        fetch failed and only the static skip-label check should run.
+    """
+    if events is not None:
+        removed_backports = _get_removed_backport_labels_from_events(events)
+        current_backports = {label for label in labels if label.startswith("backport-to-")}
+        if removed_backports and not current_backports:
+            console_print(
+                f"[info]Skipping milestone tagging - backport labels were removed during the "
+                f"PR lifecycle and no replacement remains, treated as explicit "
+                f"maintainer/triager signal: {sorted(removed_backports)}[/]"
+            )
+            return True
+
+    skip_labels_present = set(labels) & MILESTONE_SKIP_LABELS
+    if skip_labels_present:
+        console_print(f"[info]Skipping milestone tagging - PR has skip label(s): {skip_labels_present}[/]")
+        return True
+
+    return False
 
 
 def _get_backport_version_from_labels(labels: list[str]) -> tuple[int, int] | None:
@@ -1240,29 +1360,18 @@ def set_milestone(
     console_print(f"[info]Base branch: {base_branch}[/]")
     console_print(f"[info]Merged by: {merged_by}[/]")
 
-    # Parse labels from JSON
+    # The workflow's ``--pr-labels`` snapshot is logged for diagnostic visibility
+    # but is no longer used in the decision: live labels and live events are
+    # fetched fresh below. See the milestone-helpers module-level comment for
+    # the full skip-decision pipeline.
     try:
-        labels = json.loads(pr_labels)
+        snapshot_labels = json.loads(pr_labels)
     except json.JSONDecodeError:
         console_print(f"[warning]Could not parse labels JSON: {pr_labels}[/]")
-        labels = []
-
-    console_print(f"[info]Labels: {labels}[/]")
-
-    # Check if we should skip
-    if _should_skip_milestone_tagging(labels):
-        console_print(
-            f"[info]Skipping milestone tagging - PR has skip label(s): {set(labels) & MILESTONE_SKIP_LABELS}[/]"
-        )
-        return
-
-    # Determine which milestone to use
-    version, reason = _determine_milestone_version(labels, pr_title, base_branch)
-    if version is None:
-        console_print(f"[info]No milestone to set: {reason}[/]")
-        return
+        snapshot_labels = []
+    console_print(f"[info]Workflow snapshot labels (diagnostic only): {snapshot_labels}[/]")
 
-    # Initialize GitHub client and get repository
+    # Initialize GitHub client and get repository.
     try:
         gh = _get_github_client(github_token)
         repo: Repository = gh.get_repo(github_repository)
@@ -1270,14 +1379,9 @@ def set_milestone(
         console_print(f"[error]Failed to connect to GitHub: {e}[/]")
         return
 
-    # Double check whether the PR already has a milestone set - if so, we don't want to override it
+    # Step 1: milestone-already-set guard — don't override an existing milestone.
     try:
         issue: Issue = repo.get_issue(pr_number)
-        if issue.milestone is not None:
-            console_print(
-                f"[info]PR #{pr_number} already has milestone '{issue.milestone.title}' set. Skipping.[/]"
-            )
-            return
     except UnknownObjectException:
         console_print(f"[error]PR #{pr_number} not found when checking existing milestone[/]")
         return
@@ -1285,39 +1389,40 @@ def set_milestone(
         console_print(f"[error]Failed to check existing milestone: {e}[/]")
         return
 
-    # Re-read labels from the freshly-fetched issue to close the race between
-    # the workflow's initial label snapshot (taken by the get-pr-info job a
-    # couple of minutes ago) and the actual milestone-set step. Maintainers
-    # sometimes add and remove a backport label inside that window; honour
-    # the latest state, not the stale snapshot.
+    if issue.milestone is not None:
+        console_print(
+            f"[info]PR #{pr_number} already has milestone '{issue.milestone.title}' set. Skipping.[/]"
+        )
+        return
+
+    # Pull live labels and events. Both feed the skip-decision function below
+    # (events → check 2, labels → check 3) and ``labels`` also feeds
+    # ``_determine_milestone_version`` once the skip pipeline lets the PR
+    # through.
     try:
-        current_labels = [label.name for label in issue.labels]
+        labels = [label.name for label in issue.labels]
     except Exception as e:
-        console_print(f"[warning]Could not re-read PR labels; falling back to snapshot decision: {e}[/]")
-        current_labels = labels
+        console_print(f"[warning]Could not read live PR labels; falling back to snapshot: {e}[/]")
+        labels = snapshot_labels
+    console_print(f"[info]Live labels: {sorted(labels)}[/]")
 
-    if set(current_labels) != set(labels):
-        console_print("[info]Labels changed since workflow snapshot; re-evaluating.[/]")
-        console_print(f"[info]Snapshot labels: {sorted(labels)}[/]")
-        console_print(f"[info]Current labels:  {sorted(current_labels)}[/]")
+    events: list | None
+    try:
+        events = list(issue.get_events())
+    except Exception as e:
+        console_print(f"[warning]Could not fetch issue events; skipping events check: {e}[/]")
+        events = None
 
-        if _should_skip_milestone_tagging(current_labels):
-            console_print(
-                f"[info]Skipping milestone tagging - PR now has skip label(s): "
-                f"{set(current_labels) & MILESTONE_SKIP_LABELS}[/]"
-            )
-            return
+    # Steps 2 and 3 of the pipeline: events-based unbackport signal, then
+    # static skip labels. The function logs its own reason when it returns True.
+    if _should_skip_milestone_tagging(labels, events=events):
+        return
 
-        new_version, new_reason = _determine_milestone_version(current_labels, pr_title, base_branch)
-        if (new_version, new_reason) != (version, reason):
-            console_print(
-                f"[info]Determination changed after re-read: was ({version}, {reason!r}); "
-                f"now ({new_version}, {new_reason!r}). Using current labels.[/]"
-            )
-            version, reason = new_version, new_reason
-            if version is None:
-                console_print(f"[info]No milestone to set after re-evaluation: {reason}[/]")
-                return
+    # Determine which milestone to use from the live labels.
+    version, reason = _determine_milestone_version(labels, pr_title, base_branch)
+    if version is None:
+        console_print(f"[info]No milestone to set: {reason}[/]")
+        return
 
     major, minor = version
     milestone_prefix = _get_milestone_prefix(major, minor)
diff --git a/dev/breeze/tests/test_set_milestone.py b/dev/breeze/tests/test_set_milestone.py
index 7c1f8b433ff65..82b16c7692e5f 100644
--- a/dev/breeze/tests/test_set_milestone.py
+++ b/dev/breeze/tests/test_set_milestone.py
@@ -17,6 +17,8 @@
 from __future__ import annotations
 
 import json
+import re
+from datetime import datetime, timezone
 from unittest.mock import MagicMock, patch
 
 import pytest
@@ -30,6 +32,7 @@
     _get_milestone_not_found_comment,
     _get_milestone_notification_comment,
     _get_milestone_prefix,
+    _get_removed_backport_labels_from_events,
     _has_bug_fix_indicators,
     _parse_milestone_version,
     _parse_version_from_backport_label,
@@ -37,6 +40,14 @@
     _should_skip_milestone_tagging,
 )
 
+_ANSI_ESCAPE_RE = re.compile(r"\x1b\[[0-9;]*m")
+
+
+def _plain_output(output: str) -> str:
+    """Strip ANSI color codes and collapse whitespace so wrap-tolerant substring
+    asserts don't trip over Rich's color escapes or soft line wraps."""
+    return " ".join(_ANSI_ESCAPE_RE.sub("", output).split())
+
 
 def _label(name: str) -> MagicMock:
     """Build a mock that quacks like a PyGithub ``Label`` for ``issue.labels``."""
@@ -45,6 +56,33 @@ def _label(name: str) -> MagicMock:
     return m
 
 
+def _unlabel_event(label_name: str, actor_login: str, when: datetime) -> MagicMock:
+    """Build a mock that quacks like a PyGithub IssueEvent for an ``unlabeled`` event."""
+    event = MagicMock()
+    event.event = "unlabeled"
+    event.label = _label(label_name)
+    event.actor = MagicMock()
+    event.actor.login = actor_login
+    event.created_at = when
+    return event
+
+
+def _issue_event(
+    event_name: str,
+    actor_login: str,
+    when: str,
+    label_name: str | None = None,
+) -> MagicMock:
+    """Build a mock shaped like a PyGithub ``IssueEvent`` for any event kind."""
+    event = MagicMock()
+    event.event = event_name
+    event.label = _label(label_name) if label_name else None
+    event.actor = MagicMock()
+    event.actor.login = actor_login
+    event.created_at = when
+    return event
+
+
 class TestParseVersionFromBranch:
     """Test cases for _parse_version_from_branch."""
 
@@ -156,11 +194,87 @@ class TestShouldSkipMilestoneTagging:
             ["area:CI"],
         ],
     )
-    def test_skip_with_skip_labels(self, labels):
-        assert _should_skip_milestone_tagging(labels)
+    def test_skip_with_static_skip_labels(self, labels):
+        assert _should_skip_milestone_tagging(labels, events=[])
+
+    def test_no_skip_without_static_skip_labels(self):
+        assert not _should_skip_milestone_tagging(["kind:feature", "area:scheduler"], events=[])
+
+    def test_skip_when_backport_unlabeled_with_no_replacement(self):
+        # Events show backport removal, no backport on PR now → skip (check 2).
+        events = [
+            _unlabel_event(
+                "backport-to-v3-1-test", "alice", datetime(2026, 5, 23, 12, 0, tzinfo=timezone.utc)
+            )
+        ]
+        assert _should_skip_milestone_tagging(["kind:bug"], events=events)
+
+    def test_no_skip_when_backport_replaced(self):
+        # Events show backport-to-v3-1 removal but v3-2 remains on the PR → no skip.
+        # The caller's regular evaluation will pick up the new label.
+        events = [
+            _unlabel_event(
+                "backport-to-v3-1-test", "alice", datetime(2026, 5, 23, 12, 0, tzinfo=timezone.utc)
+            )
+        ]
+        assert not _should_skip_milestone_tagging(["backport-to-v3-2-test", "kind:bug"], events=events)
+
+    def test_no_skip_when_no_removal_event(self):
+        # No unlabeled events at all — current backport label drives the decision elsewhere.
+        assert not _should_skip_milestone_tagging(["backport-to-v3-1-test", "kind:bug"], events=[])
+
+    def test_no_skip_when_unrelated_label_removed(self):
+        events = [
+            _unlabel_event("kind:documentation", "alice", datetime(2026, 5, 23, 12, 0, tzinfo=timezone.utc))
+        ]
+        assert not _should_skip_milestone_tagging(["kind:bug"], events=events)
+
+    def test_events_check_takes_precedence_over_static_labels(self):
+        # Both check 2 (events) and check 3 (static label) would fire here;
+        # the function returns True on the first one — events — and never logs
+        # the static-label reason.
+        events = [
+            _unlabel_event(
+                "backport-to-v3-1-test", "alice", datetime(2026, 5, 23, 12, 0, tzinfo=timezone.utc)
+            )
+        ]
+        assert _should_skip_milestone_tagging(["area:CI"], events=events)
+
+    def test_events_none_disables_events_check(self):
+        # When events is None (fetch failed), only the static label check runs.
+        assert not _should_skip_milestone_tagging(["kind:bug"], events=None)
+        assert _should_skip_milestone_tagging(["area:CI"], events=None)
+
+
+class TestGetRemovedBackportLabelsFromEvents:
+    """Test cases for _get_removed_backport_labels_from_events."""
+
+    def test_returns_all_unlabel_events_for_backports(self):
+        events = [
+            _unlabel_event(
+                "backport-to-v3-1-test", "alice", datetime(2026, 5, 23, 12, 0, tzinfo=timezone.utc)
+            ),
+            _unlabel_event("backport-to-v3-2-test", "bob", datetime(2026, 5, 23, 14, 0, tzinfo=timezone.utc)),
+        ]
+        assert _get_removed_backport_labels_from_events(events) == {
+            "backport-to-v3-1-test",
+            "backport-to-v3-2-test",
+        }
 
-    def test_no_skip_without_skip_labels(self):
-        assert not _should_skip_milestone_tagging(["kind:feature", "area:scheduler"])
+    def test_ignores_unrelated_label_unlabel_events(self):
+        events = [
+            _unlabel_event("kind:documentation", "alice", datetime(2026, 5, 23, 12, 0, tzinfo=timezone.utc))
+        ]
+        assert _get_removed_backport_labels_from_events(events) == set()
+
+    def test_ignores_non_unlabeled_events(self):
+        labeled = MagicMock()
+        labeled.event = "labeled"
+        labeled.label = _label("backport-to-v3-1-test")
+        assert _get_removed_backport_labels_from_events([labeled]) == set()
+
+    def test_empty_events(self):
+        assert _get_removed_backport_labels_from_events([]) == set()
 
 
 class TestGetBackportVersionFromLabels:
@@ -326,10 +440,18 @@ def mock_github_setup(self):
         ],
     )
     @patch("airflow_breeze.commands.ci_commands._get_github_client")
-    def test_skip_label_should_skip(self, mock_get_client, base_branch, skip_label, cli_runner):
+    def test_skip_label_should_skip(
+        self, mock_get_client, base_branch, skip_label, cli_runner, mock_github_setup
+    ):
         """When PR has a skip label, milestone tagging should be skipped."""
         from airflow_breeze.commands.ci_commands import ci_group
 
+        mock_gh, mock_repo, mock_issue = mock_github_setup
+        mock_issue.milestone = None
+        mock_issue.labels = [_label(skip_label)]
+        mock_issue.get_events.return_value = []
+        mock_get_client.return_value = mock_gh
+
         result = cli_runner.invoke(
             ci_group,
             [
@@ -349,14 +471,24 @@ def test_skip_label_should_skip(self, mock_get_client, base_branch, skip_label,
             ],
         )
 
-        mock_get_client.assert_not_called()
-        assert "Skipping milestone tagging" in result.output
+        mock_issue.edit.assert_not_called()
+        plain = _plain_output(result.output)
+        assert "Skipping milestone tagging" in plain
+        assert skip_label in plain
 
     @patch("airflow_breeze.commands.ci_commands._get_github_client")
-    def test_main_branch_without_backport_label_should_skip(self, mock_get_client, cli_runner):
+    def test_main_branch_without_backport_label_should_skip(
+        self, mock_get_client, cli_runner, mock_github_setup
+    ):
         """When PR is merged to main without backport label, milestone tagging should be skipped."""
         from airflow_breeze.commands.ci_commands import ci_group
 
+        mock_gh, mock_repo, mock_issue = mock_github_setup
+        mock_issue.milestone = None
+        mock_issue.labels = [_label("kind:feature")]
+        mock_issue.get_events.return_value = []
+        mock_get_client.return_value = mock_gh
+
         result = cli_runner.invoke(
             ci_group,
             [
@@ -376,7 +508,7 @@ def test_main_branch_without_backport_label_should_skip(self, mock_get_client, c
             ],
         )
 
-        mock_get_client.assert_not_called()
+        mock_issue.edit.assert_not_called()
         assert "No milestone to set" in result.output
 
     @pytest.mark.parametrize(
@@ -584,19 +716,26 @@ def test_not_find_milestone_should_comment_warning(
         assert "No open milestone found" in result.output
 
     @patch("airflow_breeze.commands.ci_commands._get_github_client")
-    def test_backport_label_removed_after_snapshot_should_skip(
+    def test_backport_unlabeled_with_no_replacement_should_skip(
         self, mock_get_client, cli_runner, mock_github_setup
     ):
-        """If a backport label is removed between the workflow snapshot and the action,
-        the action must re-read labels from the issue and honour the current state —
-        skip milestone-set when the only signal that triggered it (the backport label)
-        is gone. Regression test for PR #67301 race.
+        """If an ``unlabeled`` event for a ``backport-to-*`` label exists on the
+        PR and no ``backport-to-*`` label remains, the action must skip the
+        milestone-set. Regression test for PR #67301 race; the events stream
+        is now the single source of truth for the unbackport signal.
         """
         from airflow_breeze.commands.ci_commands import ci_group
 
         mock_gh, mock_repo, mock_issue = mock_github_setup
         mock_issue.milestone = None
         mock_issue.labels = [_label("kind:documentation")]
+        mock_issue.get_events.return_value = [
+            _unlabel_event(
+                "backport-to-v3-2-test",
+                "shahar1",
+                datetime(2026, 5, 23, 20, 32, 17, tzinfo=timezone.utc),
+            ),
+        ]
         mock_get_client.return_value = mock_gh
 
         result = cli_runner.invoke(
@@ -620,28 +759,84 @@ def test_backport_label_removed_after_snapshot_should_skip(
             ],
         )
 
-        # Snapshot still has the backport label, but the fresh issue.labels does not.
-        # The action must re-read, notice the change, and skip the milestone-set.
         mock_issue.edit.assert_not_called()
         mock_issue.create_comment.assert_not_called()
-        assert "Labels changed since workflow snapshot" in result.output
-        assert "No milestone to set after re-evaluation" in result.output
+        plain = _plain_output(result.output)
+        assert "Skipping milestone tagging" in plain
+        assert "backport labels were removed during the PR lifecycle" in plain
+        assert "backport-to-v3-2-test" in plain
         assert result.exit_code == 0
 
     @patch("airflow_breeze.commands.ci_commands._get_github_client")
-    def test_backport_label_changed_after_snapshot_should_use_current(
+    def test_backport_unlabeled_on_version_branch_should_skip(
         self, mock_get_client, cli_runner, mock_github_setup
     ):
-        """If the backport label is replaced with a different version between
-        snapshot and action (e.g. someone fixes the version target), the action
-        must re-determine the version using the current label, not the stale one.
+        """A backport-label removal recorded in the issue events must take
+        precedence over the merge-to-version-branch heuristic. Without this, a
+        PR merged to a version branch would still get that branch's milestone
+        even after a maintainer/triager explicitly removed the backport label.
+        """
+        from airflow_breeze.commands.ci_commands import ci_group
+
+        mock_gh, mock_repo, mock_issue = mock_github_setup
+        mock_issue.milestone = None
+        mock_issue.labels = [_label("kind:bug")]
+        mock_issue.get_events.return_value = [
+            _unlabel_event(
+                "backport-to-v3-2-test",
+                "testuser",
+                datetime(2026, 5, 23, 20, 32, 17, tzinfo=timezone.utc),
+            ),
+        ]
+        mock_get_client.return_value = mock_gh
+
+        result = cli_runner.invoke(
+            ci_group,
+            [
+                "set-milestone",
+                "--pr-number",
+                "12345",
+                "--pr-title",
+                "Fix: scheduler issue",
+                "--pr-labels",
+                json.dumps(["backport-to-v3-2-test", "kind:bug"]),
+                "--base-branch",
+                "v3-1-test",
+                "--merged-by",
+                "testuser",
+                "--github-token",
+                "fake-token",
+                "--github-repository",
+                "apache/airflow",
+            ],
+        )
+
+        mock_issue.edit.assert_not_called()
+        mock_issue.create_comment.assert_not_called()
+        plain = _plain_output(result.output)
+        assert "Skipping milestone tagging" in plain
+        assert "backport labels were removed during the PR lifecycle" in plain
+        assert "backport-to-v3-2-test" in plain
+        assert result.exit_code == 0
+
+    @patch("airflow_breeze.commands.ci_commands._get_github_client")
+    def test_backport_label_replaced_should_use_current(self, mock_get_client, cli_runner, mock_github_setup):
+        """When the events show one backport label removed but another
+        ``backport-to-*`` remains on the PR (e.g. someone swapped the version
+        target), the action must use the current label, not skip.
         """
         from airflow_breeze.commands.ci_commands import ci_group
 
         mock_gh, mock_repo, mock_issue = mock_github_setup
         mock_issue.milestone = None
-        # Fresh state: now targets v3-2-test, not v3-1-test.
         mock_issue.labels = [_label("backport-to-v3-2-test"), _label("kind:bug")]
+        mock_issue.get_events.return_value = [
+            _unlabel_event(
+                "backport-to-v3-1-test",
+                "testuser",
+                datetime(2026, 5, 23, 20, 30, 0, tzinfo=timezone.utc),
+            ),
+        ]
         mock_milestone = MagicMock()
         mock_milestone.title = "Airflow 3.2.3"
         mock_milestone.number = 140
@@ -673,23 +868,24 @@ def test_backport_label_changed_after_snapshot_should_use_current(
         )
 
         mock_issue.edit.assert_called_once_with(milestone=mock_milestone)
-        assert "Labels changed since workflow snapshot" in result.output
-        assert "Determination changed after re-read" in result.output
         assert "Airflow 3.2.3" in captured_comments[0]
         assert "backport label targeting v3-2-test" in captured_comments[0]
         assert result.exit_code == 0
 
     @patch("airflow_breeze.commands.ci_commands._get_github_client")
-    def test_skip_label_added_after_snapshot_should_skip(
+    def test_skip_label_present_on_live_labels_should_skip(
         self, mock_get_client, cli_runner, mock_github_setup
     ):
-        """A skip label added after the snapshot must also halt the action."""
+        """A skip label present on the live labels (regardless of what the
+        workflow snapshot had) must halt the action via check 3 of the
+        pipeline.
+        """
         from airflow_breeze.commands.ci_commands import ci_group
 
         mock_gh, mock_repo, mock_issue = mock_github_setup
         mock_issue.milestone = None
-        # Snapshot had no skip label; fresh state added area:CI.
         mock_issue.labels = [_label("backport-to-v3-1-test"), _label("area:CI")]
+        mock_issue.get_events.return_value = []
         mock_get_client.return_value = mock_gh
 
         result = cli_runner.invoke(
@@ -718,3 +914,77 @@ def test_skip_label_added_after_snapshot_should_skip(
         assert "Skipping milestone tagging" in result.output
         assert "area:CI" in result.output
         assert result.exit_code == 0
+
+    @patch("airflow_breeze.commands.ci_commands._get_github_client")
+    def test_pr_67301_real_events_should_skip(self, mock_get_client, cli_runner, mock_github_setup):
+        """End-to-end regression test against the real ``issue.get_events()``
+        stream from PR #67301 (the incident that motivated this change).
+
+        The events below are the actual events captured from
+        ``GET /repos/apache/airflow/issues/67301/events``, trimmed to those
+        that existed BEFORE the offending ``github-actions[bot] milestoned``
+        event — that ``milestoned`` event is exactly what the new
+        live-labels + events pipeline must prevent, so it is intentionally
+        omitted from this fixture. With the fix in place, ``set-milestone``
+        must notice shahar1's ``unlabeled backport-to-v3-2-test`` 92 seconds
+        earlier and skip.
+        """
+        from airflow_breeze.commands.ci_commands import ci_group
+
+        pr_67301_events = [
+            _issue_event("labeled", "boring-cyborg[bot]", "2026-05-21T19:42:28Z", "area:providers"),
+            _issue_event("labeled", "boring-cyborg[bot]", "2026-05-21T19:42:28Z", "kind:documentation"),
+            _issue_event("labeled", "boring-cyborg[bot]", "2026-05-21T19:42:28Z", "provider:standard"),
+            _issue_event("merged", "shahar1", "2026-05-21T20:31:18Z"),
+            _issue_event("closed", "shahar1", "2026-05-21T20:31:18Z"),
+            _issue_event("labeled", "shahar1", "2026-05-21T20:31:28Z", "backport-to-v3-2-test"),
+            _issue_event("unlabeled", "shahar1", "2026-05-21T20:32:17Z", "backport-to-v3-2-test"),
+        ]
+        # Live ``issue.labels`` at the moment set-milestone would have run:
+        # backport-to-v3-2-test had just been removed, leaving these three.
+        live_labels = ["area:providers", "kind:documentation", "provider:standard"]
+
+        mock_gh, mock_repo, mock_issue = mock_github_setup
+        mock_issue.milestone = None
+        mock_issue.labels = [_label(name) for name in live_labels]
+        mock_issue.get_events.return_value = pr_67301_events
+        mock_get_client.return_value = mock_gh
+
+        result = cli_runner.invoke(
+            ci_group,
+            [
+                "set-milestone",
+                "--pr-number",
+                "67301",
+                "--pr-title",
+                'fix: typo "@tash.bash" -> "@task.bash',
+                "--pr-labels",
+                # The workflow's stale snapshot from get-pr-info still saw the
+                # backport label; the new pipeline ignores this in favour of
+                # live state.
+                json.dumps(
+                    [
+                        "area:providers",
+                        "kind:documentation",
+                        "provider:standard",
+                        "backport-to-v3-2-test",
+                    ]
+                ),
+                "--base-branch",
+                "main",
+                "--merged-by",
+                "shahar1",
+                "--github-token",
+                "fake-token",
+                "--github-repository",
+                "apache/airflow",
+            ],
+        )
+
+        mock_issue.edit.assert_not_called()
+        mock_issue.create_comment.assert_not_called()
+        plain = _plain_output(result.output)
+        assert "Skipping milestone tagging" in plain
+        assert "backport labels were removed during the PR lifecycle" in plain
+        assert "backport-to-v3-2-test" in plain
+        assert result.exit_code == 0