Reject stale dashboard delivery workers

.github/scripts/pull-request-dashboard/delivery.py

@@ -21,6 +21,7 @@
 from pr_status_comment import update_status_comments_from_state
 from state import (
     author_nudge_state_path,
+    claim_delivery_worker,
     copilot_review_request_state_path,
     notification_state_path,
     set_state_dir,
@@ -112,14 +113,26 @@ def deliver_with_state(
     repo: str,
     state_branch_name: str,
     state_dir: Path,
+    worker_run_id: int,
+    worker_sha: str,
+    github_output: Path | None = None,
 ) -> int:
     repo_key = repo_state_key(repo)
     author_retry = runner_temp_path("prior-author-nudge-state.json")
     copilot_retry = runner_temp_path("prior-copilot-review-request-state.json")
     notification_retry = runner_temp_path("prior-notification-state.json")
     errors: list[str] = []
+    active_worker = False
 
     def deliver() -> int:
+        nonlocal active_worker
+        active_worker = claim_delivery_worker(worker_run_id, worker_sha)
+        if not active_worker:
+            print(
+                f"delivery worker run {worker_run_id} does not match the active release; skipping",
+                file=sys.stderr,
+            )
+            return 0
         errors[:] = deliver_from_state(
             repo,
             author_retry,
@@ -142,6 +155,9 @@ def deliver() -> int:
     )
     if status != 0:
         return status
+    if github_output is not None:
+        with github_output.open("a", encoding="utf-8") as output:
+            output.write(f"active={'true' if active_worker else 'false'}\n")
     if not errors:
         return 0
     print("Dashboard delivery failed:", file=sys.stderr)
@@ -153,6 +169,9 @@ def main() -> int:
     parser = argparse.ArgumentParser(description=__doc__)
     parser.add_argument("--repo", help="target repository name")
     parser.add_argument("--state-branch", required=True, help="git branch used for workflow state")
+    parser.add_argument("--worker-run-id", required=True, type=int, help="GitHub Actions run ID")
+    parser.add_argument("--worker-sha", required=True, help="immutable shared-workflows commit SHA")
+    parser.add_argument("--github-output", type=Path, help="append the active worker result")
     args = parser.parse_args()
     repo = normalize_repo(args.repo) if args.repo else detect_repo()
     with state_branch.temporary_state_dir() as state_dir:
@@ -161,6 +180,9 @@ def main() -> int:
             repo,
             args.state_branch,
             state_dir,
+            args.worker_run_id,
+            args.worker_sha,
+            args.github_output,
         )
 
 

.github/scripts/pull-request-dashboard/state.py

@@ -15,6 +15,7 @@
 AUTHOR_NUDGE_STATE_FILE = "author-nudge-state.json"
 COPILOT_REVIEW_REQUEST_STATE_FILE = "copilot-review-request-state.json"
 STATUS_COMMENT_ROLLOUT_STATE_FILE = "status-comment-rollout-state.json"
+DELIVERY_WORKER_STATE_FILE = "delivery-worker-state.json"
 # State files are disposable workflow caches, not durable user data. Bump only
 # the version for the state shape whose meaning changed.
 DASHBOARD_STATE_VERSION = 5
@@ -23,6 +24,7 @@
 AUTHOR_NUDGE_STATE_VERSION = 2
 COPILOT_REVIEW_REQUEST_STATE_VERSION = 3
 STATUS_COMMENT_ROLLOUT_STATE_VERSION = 1
+DELIVERY_WORKER_STATE_VERSION = 1
 INITIAL_BACKFILL_COMPLETE_KEY = "initial_backfill_complete"
 _state_dir: Path | None = None
 
@@ -62,6 +64,10 @@ def status_comment_rollout_state_path() -> Path:
     return state_dir() / STATUS_COMMENT_ROLLOUT_STATE_FILE
 
 
+def delivery_worker_state_path() -> Path:
+    return state_dir() / DELIVERY_WORKER_STATE_FILE
+
+
 def dashboard_markdown_path() -> Path:
     return state_dir() / DASHBOARD_MARKDOWN_FILE
 
@@ -91,6 +97,14 @@ def empty_status_comment_rollout_state() -> dict[str, Any]:
     }
 
 
+def empty_delivery_worker_state() -> dict[str, Any]:
+    return {
+        "version": DELIVERY_WORKER_STATE_VERSION,
+        "active_run_id": 0,
+        "active_sha": "",
+    }
+
+
 def load_state_file(
     path: Path,
     current_version: int,
@@ -183,6 +197,50 @@ def save_status_comment_rollout_state(state: dict[str, Any]) -> None:
     )
 
 
+def load_delivery_worker_state() -> dict[str, Any]:
+    state = load_state_file(delivery_worker_state_path(), DELIVERY_WORKER_STATE_VERSION)
+    if state is None:
+        return empty_delivery_worker_state()
+    try:
+        active_run_id = max(int(state.get("active_run_id") or 0), 0)
+    except (TypeError, ValueError):
+        return empty_delivery_worker_state()
+    active_sha = state.get("active_sha")
+    return {
+        "version": DELIVERY_WORKER_STATE_VERSION,
+        "active_run_id": active_run_id,
+        "active_sha": active_sha if isinstance(active_sha, str) else "",
+    }
+
+
+def save_delivery_worker_state(state: dict[str, Any]) -> None:
+    save_state_file(
+        delivery_worker_state_path(),
+        {
+            "active_run_id": int(state.get("active_run_id") or 0),
+            "active_sha": str(state.get("active_sha") or ""),
+        },
+        DELIVERY_WORKER_STATE_VERSION,
+    )
+
+
+def claim_delivery_worker(worker_run_id: int, worker_sha: str) -> bool:
+    if worker_run_id <= 0:
+        raise ValueError("worker run ID must be positive")
+    if not worker_sha.strip():
+        raise ValueError("worker SHA must not be empty")
+
+    state = load_delivery_worker_state()
+    active_run_id = state["active_run_id"]
+    if state["active_sha"] == worker_sha:
+        return True
+    if active_run_id >= worker_run_id:
+        return False
+
+    save_delivery_worker_state({"active_run_id": worker_run_id, "active_sha": worker_sha})
+    return True
+
+
 def enqueue_status_comment_update(pr_number: int) -> None:
     state = load_status_comment_rollout_state()
     pending = set(state["pending_pr_numbers"])

.github/scripts/pull-request-dashboard/test_delivery.py

@@ -1,6 +1,7 @@
 from __future__ import annotations
 
 from pathlib import Path
+import tempfile
 import unittest
 from unittest.mock import ANY, Mock, call, patch
 
@@ -133,10 +134,12 @@ def test_open_pr_list_failure_skips_dependent_stages(self) -> None:
 
     @patch.object(delivery.sys, "stderr")
     @patch.object(delivery, "deliver_from_state", return_value=["status comments: boom"])
+    @patch.object(delivery, "claim_delivery_worker", return_value=True)
     @patch.object(delivery.state_branch, "push_state_changes")
     def test_reports_delivery_errors_after_state_push(
         self,
         push_state_changes,
+        _claim_delivery_worker,
         _deliver_from_state,
         _stderr,
     ) -> None:
@@ -153,10 +156,47 @@ def test_reports_delivery_errors_after_state_push(
                 "open-telemetry/example",
                 "dashboard-state",
                 Path("state"),
+                200,
+                "worker-sha",
             )
 
         self.assertEqual(1, status)
 
+    @patch.object(delivery, "deliver_from_state")
+    @patch.object(delivery, "claim_delivery_worker", return_value=False)
+    @patch.object(delivery.state_branch, "push_state_changes")
+    def test_stale_worker_skips_delivery_and_reports_inactive(
+        self,
+        push_state_changes,
+        claim_delivery_worker,
+        deliver_from_state,
+    ) -> None:
+        push_state_changes.side_effect = (
+            lambda _state_dir, _message, update_state, **_kwargs: update_state()
+        )
+
+        with (
+            tempfile.TemporaryDirectory() as temp_dir,
+            patch.object(delivery, "author_nudge_state_path", return_value=Path("author")),
+            patch.object(delivery, "copilot_review_request_state_path", return_value=Path("copilot")),
+            patch.object(delivery, "notification_state_path", return_value=Path("slack")),
+        ):
+            github_output = Path(temp_dir) / "github-output"
+            status = delivery.deliver_with_state(
+                "open-telemetry/example",
+                "dashboard-state",
+                Path("state"),
+                199,
+                "older-sha",
+                github_output,
+            )
+            github_output_text = github_output.read_text(encoding="utf-8")
+
+        self.assertEqual(0, status)
+        claim_delivery_worker.assert_called_once_with(199, "older-sha")
+        deliver_from_state.assert_not_called()
+        self.assertEqual("active=false\n", github_output_text)
+
 
 if __name__ == "__main__":
     unittest.main()

.github/scripts/pull-request-dashboard/test_state.py

@@ -13,11 +13,13 @@
     BACKFILL_STATE_VERSION,
     COPILOT_REVIEW_REQUEST_STATE_VERSION,
     DASHBOARD_STATE_VERSION,
+    DELIVERY_WORKER_STATE_VERSION,
     NOTIFICATION_STATE_VERSION,
     STATUS_COMMENT_ROLLOUT_STATE_VERSION,
     author_nudge_state_path,
     backfill_state_path,
     copilot_review_request_state_path,
+    claim_delivery_worker,
     dashboard_state_path,
     empty_state,
     enqueue_status_comment_update,
@@ -26,6 +28,7 @@
     load_backfill_state,
     load_copilot_review_requests,
     load_dashboard_state_cache,
+    load_delivery_worker_state,
     load_state_file,
     load_status_comment_rollout_state,
     load_notifications,
@@ -357,6 +360,26 @@ def test_enqueue_status_comment_update_is_deduplicated(self) -> None:
                 load_status_comment_rollout_state()["pending_pr_numbers"],
             )
 
+    def test_delivery_worker_identity_only_advances(self) -> None:
+        with tempfile.TemporaryDirectory() as temp_dir, patch("state._state_dir", Path(temp_dir)):
+            self.assertTrue(claim_delivery_worker(200, "newer-sha"))
+            self.assertTrue(claim_delivery_worker(200, "newer-sha"))
+            self.assertTrue(claim_delivery_worker(199, "newer-sha"))
+            self.assertFalse(claim_delivery_worker(200, "conflicting-sha"))
+            self.assertFalse(claim_delivery_worker(199, "older-sha"))
+
+            self.assertEqual(
+                {
+                    "version": DELIVERY_WORKER_STATE_VERSION,
+                    "active_run_id": 200,
+                    "active_sha": "newer-sha",
+                },
+                load_delivery_worker_state(),
+            )
+
+            self.assertTrue(claim_delivery_worker(201, "newest-sha"))
+            self.assertEqual("newest-sha", load_delivery_worker_state()["active_sha"])
+
     def test_backfill_state_preserves_version_three_cursor(self) -> None:
         with tempfile.TemporaryDirectory() as temp_dir, patch("state._state_dir", Path(temp_dir)):
             backfill_state_path().write_text(

.github/workflows/pull-request-dashboard-repo.yml

@@ -216,14 +216,25 @@ jobs:
           SLACK_CHANNEL: ${{ inputs.slack_channel }}
           SLACK_USER_MAP_JSON: ${{ inputs.slack_user_mapping_json }}
           REPO_NAME: ${{ inputs.repository }}
+          WORKER_RUN_ID: ${{ github.run_id }}
+          WORKER_SHA: ${{ github.sha }}
         run: |
           set -euo pipefail
           state_branch="${DASHBOARD_STATE_BRANCH_PREFIX}/${REPO_NAME}"
-          delivery_args=(--state-branch "$state_branch" --repo "$REPO_NAME")
+          delivery_args=(
+            --state-branch "$state_branch"
+            --repo "$REPO_NAME"
+            --worker-run-id "$WORKER_RUN_ID"
+            --worker-sha "$WORKER_SHA"
+            --github-output "$GITHUB_OUTPUT"
+          )
           python3 .github/scripts/pull-request-dashboard/delivery.py "${delivery_args[@]}"
 
       - name: Publish dashboard issue
-        if: always() && steps.dashboard-token.outcome == 'success'
+        if: >-
+          always() &&
+          steps.dashboard-token.outcome == 'success' &&
+          steps.delivery.outputs.active == 'true'
         env:
           GH_TOKEN: ${{ steps.dashboard-token.outputs.token }}
           REPO_NAME: ${{ inputs.repository }}