From abf7272747c69704f77228714ee10bb28242eb0c Mon Sep 17 00:00:00 2001
From: Jarek Potiuk <jarek@potiuk.com>
Date: Sat, 25 Apr 2026 21:21:18 +0200
Subject: [PATCH] [v3-2-test] CI: Notify open PRs that conflict on uv.lock
 after main merges (#65765)

* CI: Notify open PRs that conflict on uv.lock after main merges

When `uv.lock` changes on `main`, open PRs that touch `uv.lock` frequently
end up with rebase conflicts. This adds a workflow that, on each push to
`main` that modifies `uv.lock`, scans open non-draft PRs (updated in the
last 14 days), identifies those currently reporting conflicts, and posts
(or updates, deduped via an HTML-comment marker) a single comment with
rebase-and-`uv lock` instructions.

The source PR that caused the change is resolved from the squash-merge
commit headline's trailing `(#NNN)` and linked in the notice.

Implemented as a standalone PEP 723 Python script (`httpx` + GitHub's
GraphQL API) so logic stays testable; workflow YAML is a thin driver that
installs `uv` from `uv.lock` and runs the script. Unit tests cover the
regex, classifier, pagination/early-exit, UNKNOWN-mergeable retry loop,
and notice posting.

* Fix scripts-tests by stubbing httpx in unit-test module

The scripts-project CI venv doesn't have httpx installed, so importing
``notify_uv_lock_conflicts`` fails with ``ModuleNotFoundError: No module
named 'httpx'`` before any test runs. httpx is a PEP 723 runtime dep of
the script; the tests mock the client and never need the real library.

Stub ``httpx`` in ``sys.modules`` with ``setdefault`` before loading the
module under test: keeps the scripts distribution's declared deps
unchanged, and leaves real ``httpx`` alone when present (e.g. in the
workspace-wide dev environment).
(cherry picked from commit 4fae29c5f3adf6a814ffe25839b90175e8935ebe)

Co-authored-by: Jarek Potiuk <jarek@potiuk.com>
---
 .../workflows/notify-uv-lock-conflicts.yml    |  55 +++
 scripts/ci/notify_uv_lock_conflicts.py        | 461 ++++++++++++++++++
 .../tests/ci/test_notify_uv_lock_conflicts.py | 329 +++++++++++++
 3 files changed, 845 insertions(+)
 create mode 100644 .github/workflows/notify-uv-lock-conflicts.yml
 create mode 100644 scripts/ci/notify_uv_lock_conflicts.py
 create mode 100644 scripts/tests/ci/test_notify_uv_lock_conflicts.py

diff --git a/.github/workflows/notify-uv-lock-conflicts.yml b/.github/workflows/notify-uv-lock-conflicts.yml
new file mode 100644
index 0000000000000..08a4a7543efb0
--- /dev/null
+++ b/.github/workflows/notify-uv-lock-conflicts.yml
@@ -0,0 +1,55 @@
+# 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: "Notify uv.lock conflicts"
+on:  # yamllint disable-line rule:truthy
+  push:
+    branches: [main]
+    paths: [uv.lock]
+
+permissions:
+  contents: read
+  pull-requests: write
+
+jobs:
+  notify:
+    name: "Notify open PRs that conflict on uv.lock"
+    runs-on: ["ubuntu-22.04"]
+    timeout-minutes: 10
+    steps:
+      - name: "Checkout ${{ github.ref }} ( ${{ github.sha }} )"
+        uses: actions/checkout@de0fac2e4500dabe0009e67214ff5f5447ce83dd  # v6.0.2
+        with:
+          persist-credentials: false
+      - name: "Install uv"
+        # Extract uv version from uv.lock. The format is stable: the line
+        # immediately after `name = "uv"` is `version = "<X.Y.Z>"`.
+        run: |
+          UV_VERSION=$(sed -n '/^name = "uv"$/{n;s/^version = "\(.*\)"$/\1/p;}' uv.lock)
+          if [[ -z "${UV_VERSION}" ]]; then
+            echo "Failed to extract uv version from uv.lock" >&2
+            exit 1
+          fi
+          echo "Installing uv==${UV_VERSION}"
+          pip install "uv==${UV_VERSION}"
+      - name: "Notify open PRs"
+        env:
+          GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }}
+          GITHUB_REPOSITORY: ${{ github.repository }}
+          GITHUB_SHA: ${{ github.sha }}
+        run: uv run scripts/ci/notify_uv_lock_conflicts.py
diff --git a/scripts/ci/notify_uv_lock_conflicts.py b/scripts/ci/notify_uv_lock_conflicts.py
new file mode 100644
index 0000000000000..84002266544fa
--- /dev/null
+++ b/scripts/ci/notify_uv_lock_conflicts.py
@@ -0,0 +1,461 @@
+#!/usr/bin/env python3
+# 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.
+
+# /// script
+# requires-python = ">=3.10"
+# dependencies = [
+#     "httpx>=0.27",
+# ]
+# ///
+"""
+Notify open PRs that conflict on ``uv.lock`` after a push to ``main``.
+
+Triggered from ``.github/workflows/notify-uv-lock-conflicts.yml`` on pushes
+to ``main`` that modify ``uv.lock``. For every non-draft, non-stale open
+PR targeting ``main`` that also modifies ``uv.lock`` and currently
+conflicts, posts (or updates) a comment asking the author to rebase and
+re-run ``uv lock``.
+
+Squash-merge is the house convention, so the PR that caused the push is
+identified from the trailing ``(#NNN)`` of the commit headline; we fall
+back to just the commit URL if parsing fails.
+
+Environment variables (all provided by the Actions runner):
+  GITHUB_TOKEN         - token with ``pull-requests:write`` and ``contents:read``
+  GITHUB_REPOSITORY    - ``owner/repo`` (e.g. ``apache/airflow``)
+  GITHUB_SHA           - head commit of the push
+  GITHUB_STEP_SUMMARY  - (optional) path to append the job summary to
+"""
+
+from __future__ import annotations
+
+import os
+import re
+import sys
+import time
+from dataclasses import dataclass
+from datetime import datetime, timedelta, timezone
+from typing import Any
+
+import httpx
+
+MARKER = "<!-- uv-lock-rebase-notice -->"
+STALE_DAYS = 14
+PAGE_SIZE = 20
+INITIAL_DELAY_S = 30
+RETRY_DELAY_S = 30
+MAX_RETRIES = 5
+
+GRAPHQL_URL = "https://api.github.com/graphql"
+PR_NUMBER_IN_COMMIT_RE = re.compile(r"\(#(\d+)\)\s*$")
+
+LIST_QUERY = """
+query($owner: String!, $repo: String!, $cursor: String, $size: Int!) {
+  repository(owner: $owner, name: $repo) {
+    pullRequests(
+      first: $size, after: $cursor,
+      states: OPEN, baseRefName: "main",
+      orderBy: {field: UPDATED_AT, direction: DESC}
+    ) {
+      pageInfo { hasNextPage endCursor }
+      nodes {
+        id number isDraft updatedAt mergeable
+        files(first: 100) { nodes { path } totalCount }
+        comments(last: 100) { nodes { id body } }
+      }
+    }
+  }
+}
+"""
+
+SINGLE_QUERY = """
+query($owner: String!, $repo: String!, $number: Int!) {
+  repository(owner: $owner, name: $repo) {
+    pullRequest(number: $number) {
+      id number isDraft updatedAt mergeable
+      files(first: 100) { nodes { path } totalCount }
+      comments(last: 100) { nodes { id body } }
+    }
+  }
+}
+"""
+
+COMMIT_HEADLINE_QUERY = """
+query($owner: String!, $repo: String!, $oid: GitObjectID!) {
+  repository(owner: $owner, name: $repo) {
+    object(oid: $oid) {
+      ... on Commit { messageHeadline }
+    }
+  }
+}
+"""
+
+PR_SUMMARY_QUERY = """
+query($owner: String!, $repo: String!, $number: Int!) {
+  repository(owner: $owner, name: $repo) {
+    pullRequest(number: $number) { number url title }
+  }
+}
+"""
+
+CREATE_MUTATION = """
+mutation($subjectId: ID!, $body: String!) {
+  addComment(input: {subjectId: $subjectId, body: $body}) { clientMutationId }
+}
+"""
+
+UPDATE_MUTATION = """
+mutation($id: ID!, $body: String!) {
+  updateIssueComment(input: {id: $id, body: $body}) { clientMutationId }
+}
+"""
+
+
+@dataclass
+class Stats:
+    scanned: int = 0
+    pages: int = 0
+    drafts: int = 0
+    stale: int = 0
+    no_uv_lock: int = 0
+    already_notified: int = 0
+    mergeable: int = 0
+    conflicting: int = 0
+    unknown: int = 0
+    retries: int = 0
+    still_unknown: int = 0
+    posted: int = 0
+    updated: int = 0
+
+
+class GitHubGraphQL:
+    def __init__(self, token: str) -> None:
+        self._http = httpx.Client(
+            base_url=GRAPHQL_URL,
+            headers={
+                "Authorization": f"Bearer {token}",
+                "Accept": "application/vnd.github+json",
+            },
+            timeout=httpx.Timeout(30.0),
+        )
+
+    def __enter__(self) -> GitHubGraphQL:
+        return self
+
+    def __exit__(self, *_: object) -> None:
+        self._http.close()
+
+    def call(self, query: str, variables: dict[str, Any]) -> dict[str, Any]:
+        r = self._http.post("", json={"query": query, "variables": variables})
+        r.raise_for_status()
+        data = r.json()
+        if "errors" in data:
+            raise RuntimeError(f"GraphQL errors: {data['errors']}")
+        return data["data"]
+
+
+def log(msg: str) -> None:
+    print(msg, flush=True)
+
+
+def warn(msg: str) -> None:
+    print(f"::warning::{msg}", flush=True)
+
+
+def parse_updated_at(value: str) -> datetime:
+    return datetime.fromisoformat(value.replace("Z", "+00:00"))
+
+
+def classify(
+    pr: dict[str, Any],
+    stale_cut: datetime,
+    short_sha: str,
+) -> tuple[str, dict[str, Any] | None]:
+    """
+    Return (kind, entry). ``kind`` is one of:
+      - skip reasons: ``drafts``, ``stale``, ``no_uv_lock``, ``already_notified``
+      - action buckets: ``conflicting``, ``mergeable``, ``unknown``
+
+    For action buckets ``entry`` is ``{"pr": ..., "existing": ...}``; for
+    skip reasons it is ``None``.
+    """
+    if pr["isDraft"]:
+        return "drafts", None
+    if parse_updated_at(pr["updatedAt"]) < stale_cut:
+        return "stale", None
+    if not any(f["path"] == "uv.lock" for f in pr["files"]["nodes"]):
+        return "no_uv_lock", None
+    existing = next(
+        (c for c in pr["comments"]["nodes"] if c.get("body") and MARKER in c["body"]),
+        None,
+    )
+    if existing and short_sha in existing["body"]:
+        return "already_notified", None
+    entry = {"pr": pr, "existing": existing}
+    if pr["mergeable"] == "CONFLICTING":
+        return "conflicting", entry
+    if pr["mergeable"] == "MERGEABLE":
+        return "mergeable", entry
+    return "unknown", entry
+
+
+def resolve_source_pr(
+    client: GitHubGraphQL, owner: str, repo: str, sha: str, short_sha: str
+) -> dict[str, Any] | None:
+    """Extract ``(#NNN)`` from the squash-merge commit headline and hydrate PR info."""
+    try:
+        data = client.call(COMMIT_HEADLINE_QUERY, {"owner": owner, "repo": repo, "oid": sha})
+        commit = data.get("repository", {}).get("object") or {}
+        headline = commit.get("messageHeadline", "")
+        match = PR_NUMBER_IN_COMMIT_RE.search(headline)
+        if not match:
+            log(f"No PR number found in commit headline: {headline!r}")
+            return None
+        number = int(match.group(1))
+        data = client.call(PR_SUMMARY_QUERY, {"owner": owner, "repo": repo, "number": number})
+        return data.get("repository", {}).get("pullRequest")
+    except Exception as err:
+        warn(f"Could not resolve source PR for {short_sha}: {err}")
+        return None
+
+
+def build_body(source_ref_md: str) -> str:
+    return "\n".join(
+        [
+            MARKER,
+            f"`uv.lock` on `main` just moved via {source_ref_md} and this PR currently conflicts.",
+            "",
+            "Quickest fix:",
+            "```bash",
+            "git fetch upstream main && git rebase upstream/main",
+            "rm uv.lock && uv lock",
+            "git add uv.lock && git rebase --continue",
+            "git push --force-with-lease",
+            "```",
+            "",
+            "_Automated nudge — ignore if you're not ready to rebase. "
+            "This comment is updated in place on future `uv.lock` bumps._",
+        ]
+    )
+
+
+def scan_open_prs(
+    client: GitHubGraphQL,
+    owner: str,
+    repo: str,
+    stale_cut: datetime,
+    short_sha: str,
+    stats: Stats,
+) -> tuple[list[dict[str, Any]], list[dict[str, Any]]]:
+    confirmed: list[dict[str, Any]] = []
+    unknowns: list[dict[str, Any]] = []
+    cursor: str | None = None
+
+    log(f"Scanning open PRs targeting main (page size {PAGE_SIZE}, stale cutoff {STALE_DAYS}d)...")
+    while True:
+        data = client.call(
+            LIST_QUERY,
+            {"owner": owner, "repo": repo, "cursor": cursor, "size": PAGE_SIZE},
+        )
+        stats.pages += 1
+        conn = data["repository"]["pullRequests"]
+        nodes = conn["nodes"]
+        log(f"  page {stats.pages}: {len(nodes)} PR(s)")
+
+        hit_stale = False
+        for pr in nodes:
+            stats.scanned += 1
+            if parse_updated_at(pr["updatedAt"]) < stale_cut:
+                # List is sorted UPDATED_AT DESC — every later PR is also stale.
+                stats.stale += 1
+                hit_stale = True
+                break
+            kind, entry = classify(pr, stale_cut, short_sha)
+            setattr(stats, kind, getattr(stats, kind) + 1)
+            if kind == "conflicting" and entry is not None:
+                confirmed.append(entry)
+            elif kind == "unknown" and entry is not None:
+                unknowns.append(entry)
+
+        if hit_stale:
+            log("  hit stale cutoff — stopping pagination")
+            break
+        if not conn["pageInfo"]["hasNextPage"]:
+            break
+        cursor = conn["pageInfo"]["endCursor"]
+
+    log(
+        f"Scan complete: {stats.scanned} scanned across {stats.pages} page(s) — "
+        f"conflicting={stats.conflicting} unknown={stats.unknown} mergeable={stats.mergeable} "
+        f"drafts={stats.drafts} stale={stats.stale} noUvLock={stats.no_uv_lock} "
+        f"alreadyNotified={stats.already_notified}"
+    )
+    return confirmed, unknowns
+
+
+def retry_unknowns(
+    client: GitHubGraphQL,
+    owner: str,
+    repo: str,
+    stale_cut: datetime,
+    short_sha: str,
+    confirmed: list[dict[str, Any]],
+    unknowns: list[dict[str, Any]],
+    stats: Stats,
+) -> list[dict[str, Any]]:
+    attempt = 0
+    while unknowns and attempt < MAX_RETRIES:
+        attempt += 1
+        pr_ids = ", ".join(f"#{u['pr']['number']}" for u in unknowns)
+        log(
+            f"Retry {attempt}/{MAX_RETRIES}: {len(unknowns)} PR(s) still UNKNOWN "
+            f"({pr_ids}) — waiting {RETRY_DELAY_S}s..."
+        )
+        time.sleep(RETRY_DELAY_S)
+        stats.retries += 1
+
+        nxt: list[dict[str, Any]] = []
+        for old in unknowns:
+            number = old["pr"]["number"]
+            data = client.call(SINGLE_QUERY, {"owner": owner, "repo": repo, "number": number})
+            pr = data["repository"]["pullRequest"]
+            kind, entry = classify(pr, stale_cut, short_sha)
+            if kind in {"drafts", "stale", "no_uv_lock", "already_notified"}:
+                log(f"  #{number} → skip ({kind})")
+                continue
+            log(f"  #{number} → {kind}")
+            if kind == "conflicting" and entry is not None:
+                confirmed.append(entry)
+            elif kind == "unknown" and entry is not None:
+                nxt.append(entry)
+        unknowns = nxt
+
+    return unknowns
+
+
+def post_notices(
+    client: GitHubGraphQL,
+    confirmed: list[dict[str, Any]],
+    body: str,
+    stats: Stats,
+) -> None:
+    log(f"Notifying {len(confirmed)} PR(s)...")
+    for c in confirmed:
+        pr = c["pr"]
+        existing = c["existing"]
+        if existing:
+            client.call(UPDATE_MUTATION, {"id": existing["id"], "body": body})
+            stats.updated += 1
+            log(f"  #{pr['number']} — updated existing notice")
+        else:
+            client.call(CREATE_MUTATION, {"subjectId": pr["id"], "body": body})
+            stats.posted += 1
+            log(f"  #{pr['number']} — posted new notice")
+
+
+def write_summary(
+    summary_path: str | None,
+    short_sha: str,
+    source_ref_md: str,
+    stats: Stats,
+) -> None:
+    if not summary_path:
+        return
+    rows: list[tuple[str, int]] = [
+        ("Scanned", stats.scanned),
+        ("Pages", stats.pages),
+        ("Conflicting", stats.conflicting),
+        ("Unknown (initial)", stats.unknown),
+        ("Mergeable", stats.mergeable),
+        ("Drafts skipped", stats.drafts),
+        ("Stale skipped", stats.stale),
+        ("No uv.lock", stats.no_uv_lock),
+        ("Already notified", stats.already_notified),
+        ("Retry passes", stats.retries),
+        ("Still unknown", stats.still_unknown),
+        ("Notices posted", stats.posted),
+        ("Notices updated", stats.updated),
+    ]
+    lines = [
+        f"# uv.lock conflict notifier — {short_sha}",
+        "",
+        f"**Source of change:** {source_ref_md}",
+        "",
+        "| Metric | Count |",
+        "|---|---|",
+        *(f"| {label} | {value} |" for label, value in rows),
+        "",
+    ]
+    with open(summary_path, "a", encoding="utf-8") as fh:
+        fh.write("\n".join(lines))
+
+
+def main() -> int:
+    try:
+        token = os.environ["GITHUB_TOKEN"]
+        owner, repo = os.environ["GITHUB_REPOSITORY"].split("/", 1)
+        sha = os.environ["GITHUB_SHA"]
+    except KeyError as err:
+        print(f"Missing required environment variable: {err}", file=sys.stderr)
+        return 2
+
+    short_sha = sha[:7]
+    commit_url = f"https://github.com/{owner}/{repo}/commit/{sha}"
+    stale_cut = datetime.now(tz=timezone.utc) - timedelta(days=STALE_DAYS)
+
+    with GitHubGraphQL(token) as client:
+        source_pr = resolve_source_pr(client, owner, repo, sha, short_sha)
+        if source_pr:
+            source_ref_md = (
+                f"[#{source_pr['number']}]({source_pr['url']}) "
+                f'("{source_pr["title"]}"), commit [`{short_sha}`]({commit_url})'
+            )
+            source_ref_plain = f"#{source_pr['number']} ({source_pr['url']}) — commit {short_sha}"
+        else:
+            source_ref_md = f"commit [`{short_sha}`]({commit_url})"
+            source_ref_plain = f"commit {short_sha}"
+
+        log(f"Source of uv.lock change: {source_ref_plain}")
+        body = build_body(source_ref_md)
+
+        log(
+            f"Initial delay: waiting {INITIAL_DELAY_S}s for GitHub to compute "
+            f"mergeability after {source_ref_plain}..."
+        )
+        time.sleep(INITIAL_DELAY_S)
+
+        stats = Stats()
+        confirmed, unknowns = scan_open_prs(client, owner, repo, stale_cut, short_sha, stats)
+        unknowns = retry_unknowns(client, owner, repo, stale_cut, short_sha, confirmed, unknowns, stats)
+
+        if unknowns:
+            stats.still_unknown = len(unknowns)
+            pr_ids = ", ".join(f"#{u['pr']['number']}" for u in unknowns)
+            warn(
+                f"{len(unknowns)} PR(s) still UNKNOWN after {MAX_RETRIES} retries "
+                f"({pr_ids}) — they will be re-evaluated on the next uv.lock bump."
+            )
+
+        post_notices(client, confirmed, body, stats)
+        write_summary(os.environ.get("GITHUB_STEP_SUMMARY"), short_sha, source_ref_md, stats)
+
+    return 0
+
+
+if __name__ == "__main__":
+    sys.exit(main())
diff --git a/scripts/tests/ci/test_notify_uv_lock_conflicts.py b/scripts/tests/ci/test_notify_uv_lock_conflicts.py
new file mode 100644
index 0000000000000..eb9c41f322e56
--- /dev/null
+++ b/scripts/tests/ci/test_notify_uv_lock_conflicts.py
@@ -0,0 +1,329 @@
+# 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.
+from __future__ import annotations
+
+import importlib.util
+import sys
+import types
+from datetime import datetime, timedelta, timezone
+from pathlib import Path
+from typing import Any
+from unittest.mock import MagicMock
+
+import pytest
+
+# The script under test declares its runtime deps via PEP 723 inline metadata
+# and is executed with ``uv run`` in production. The unit tests never construct
+# the real HTTP client (they mock it), so we stub ``httpx`` in ``sys.modules``
+# before loading the module. This keeps the scripts-project test venv slim —
+# no need to add ``httpx`` to ``scripts/pyproject.toml`` just for tests.
+sys.modules.setdefault("httpx", types.ModuleType("httpx"))
+
+MODULE_PATH = Path(__file__).resolve().parents[3] / "scripts" / "ci" / "notify_uv_lock_conflicts.py"
+
+
+@pytest.fixture
+def mod():
+    module_name = "test_notify_uv_lock_conflicts_module"
+    sys.modules.pop(module_name, None)
+    spec = importlib.util.spec_from_file_location(module_name, MODULE_PATH)
+    assert spec is not None and spec.loader is not None
+    module = importlib.util.module_from_spec(spec)
+    # Register in sys.modules *before* exec_module so @dataclass (under
+    # ``from __future__ import annotations``) can resolve the defining module
+    # via sys.modules[cls.__module__] when parsing string annotations.
+    sys.modules[module_name] = module
+    try:
+        spec.loader.exec_module(module)
+        yield module
+    finally:
+        sys.modules.pop(module_name, None)
+
+
+# A recent timestamp — well inside the stale window — used by fixture PRs.
+FRESH_ISO = "2099-01-01T12:00:00Z"
+STALE_CUT = datetime(1970, 1, 1, tzinfo=timezone.utc)
+SHORT_SHA = "abc1234"
+
+
+def make_pr(
+    *,
+    number: int = 1,
+    is_draft: bool = False,
+    updated_at: str = FRESH_ISO,
+    mergeable: str = "CONFLICTING",
+    files: list[str] | None = None,
+    comments: list[dict[str, Any]] | None = None,
+    node_id: str = "PR_NODE",
+) -> dict[str, Any]:
+    return {
+        "id": node_id,
+        "number": number,
+        "isDraft": is_draft,
+        "updatedAt": updated_at,
+        "mergeable": mergeable,
+        "files": {
+            "nodes": [{"path": p} for p in (files if files is not None else ["uv.lock"])],
+            "totalCount": len(files) if files is not None else 1,
+        },
+        "comments": {"nodes": comments if comments is not None else []},
+    }
+
+
+class TestPrNumberRegex:
+    @pytest.mark.parametrize(
+        "headline,expected",
+        [
+            ("Fix bug in scheduler (#12345)", "12345"),
+            ("Refactor deps   (#9) ", "9"),  # trailing whitespace
+            ("Short (#1)\n", "1"),  # trailing newline
+        ],
+    )
+    def test_matches_trailing_pr_number(self, mod, headline, expected):
+        m = mod.PR_NUMBER_IN_COMMIT_RE.search(headline)
+        assert m is not None and m.group(1) == expected
+
+    @pytest.mark.parametrize(
+        "headline",
+        [
+            "Fix bug (see #12345)",  # not at end
+            "No PR ref",
+            "(#notanumber)",
+            "Reference (#1) somewhere in middle",
+        ],
+    )
+    def test_does_not_match_non_trailing(self, mod, headline):
+        assert mod.PR_NUMBER_IN_COMMIT_RE.search(headline) is None
+
+
+class TestParseUpdatedAt:
+    def test_handles_zulu_suffix(self, mod):
+        dt = mod.parse_updated_at("2025-04-24T10:20:30Z")
+        assert dt == datetime(2025, 4, 24, 10, 20, 30, tzinfo=timezone.utc)
+
+    def test_handles_explicit_offset(self, mod):
+        dt = mod.parse_updated_at("2025-04-24T10:20:30+00:00")
+        assert dt == datetime(2025, 4, 24, 10, 20, 30, tzinfo=timezone.utc)
+
+
+class TestClassify:
+    def test_draft(self, mod):
+        kind, entry = mod.classify(make_pr(is_draft=True), STALE_CUT, SHORT_SHA)
+        assert kind == "drafts" and entry is None
+
+    def test_stale(self, mod):
+        cutoff = datetime.now(tz=timezone.utc) - timedelta(days=14)
+        old_pr = make_pr(updated_at="2000-01-01T00:00:00Z")
+        kind, entry = mod.classify(old_pr, cutoff, SHORT_SHA)
+        assert kind == "stale" and entry is None
+
+    def test_no_uv_lock(self, mod):
+        kind, entry = mod.classify(make_pr(files=["other.txt"]), STALE_CUT, SHORT_SHA)
+        assert kind == "no_uv_lock" and entry is None
+
+    def test_already_notified_for_this_sha(self, mod):
+        comments = [{"id": "C1", "body": f"{mod.MARKER}\nmessage referencing {SHORT_SHA}"}]
+        kind, entry = mod.classify(make_pr(comments=comments), STALE_CUT, SHORT_SHA)
+        assert kind == "already_notified" and entry is None
+
+    def test_conflicting(self, mod):
+        kind, entry = mod.classify(make_pr(mergeable="CONFLICTING"), STALE_CUT, SHORT_SHA)
+        assert kind == "conflicting" and entry is not None and entry["existing"] is None
+
+    def test_mergeable(self, mod):
+        kind, entry = mod.classify(make_pr(mergeable="MERGEABLE"), STALE_CUT, SHORT_SHA)
+        assert kind == "mergeable" and entry is not None
+
+    def test_unknown(self, mod):
+        kind, entry = mod.classify(make_pr(mergeable="UNKNOWN"), STALE_CUT, SHORT_SHA)
+        assert kind == "unknown" and entry is not None
+
+    def test_previous_marker_with_different_sha_still_notifies(self, mod):
+        """An existing marker comment referencing an *older* sha must not short-circuit."""
+        comments = [{"id": "C1", "body": f"{mod.MARKER}\nolder notice for deadbee"}]
+        kind, entry = mod.classify(make_pr(comments=comments), STALE_CUT, SHORT_SHA)
+        assert kind == "conflicting"
+        assert entry is not None
+        assert entry["existing"] == {"id": "C1", "body": f"{mod.MARKER}\nolder notice for deadbee"}
+
+
+class TestBuildBody:
+    def test_includes_marker_source_and_instructions(self, mod):
+        body = mod.build_body("[#42](https://example/pr/42)")
+        assert body.startswith(mod.MARKER)
+        assert "[#42](https://example/pr/42)" in body
+        assert "git fetch upstream main" in body
+        assert "rm uv.lock && uv lock" in body
+        assert "Automated nudge" in body
+
+
+class TestResolveSourcePr:
+    def test_extracts_pr_number_from_headline(self, mod):
+        client = MagicMock()
+        client.call.side_effect = [
+            {"repository": {"object": {"messageHeadline": "Fix something (#987)"}}},
+            {"repository": {"pullRequest": {"number": 987, "url": "U", "title": "T"}}},
+        ]
+        pr = mod.resolve_source_pr(client, "o", "r", "deadbeef", "deadbee")
+        assert pr == {"number": 987, "url": "U", "title": "T"}
+        assert client.call.call_count == 2
+
+    def test_returns_none_when_no_pr_number(self, mod):
+        client = MagicMock()
+        client.call.return_value = {"repository": {"object": {"messageHeadline": "Direct push"}}}
+        assert mod.resolve_source_pr(client, "o", "r", "deadbeef", "deadbee") is None
+        assert client.call.call_count == 1  # second query skipped
+
+    def test_returns_none_on_error(self, mod):
+        client = MagicMock()
+        client.call.side_effect = RuntimeError("boom")
+        assert mod.resolve_source_pr(client, "o", "r", "deadbeef", "deadbee") is None
+
+
+class TestScanOpenPrs:
+    def _page(self, nodes, *, has_next=False, cursor=None):
+        return {
+            "repository": {
+                "pullRequests": {
+                    "pageInfo": {"hasNextPage": has_next, "endCursor": cursor},
+                    "nodes": nodes,
+                }
+            }
+        }
+
+    def test_early_exit_on_stale(self, mod):
+        """A stale PR stops pagination even if hasNextPage is True."""
+        client = MagicMock()
+        recent_cut = datetime.now(tz=timezone.utc) - timedelta(days=1)
+        fresh = (datetime.now(tz=timezone.utc)).isoformat().replace("+00:00", "Z")
+        stale = (datetime.now(tz=timezone.utc) - timedelta(days=30)).isoformat().replace("+00:00", "Z")
+        client.call.return_value = self._page(
+            [
+                make_pr(number=1, updated_at=fresh, mergeable="CONFLICTING"),
+                make_pr(number=2, updated_at=stale),  # triggers early exit
+            ],
+            has_next=True,
+            cursor="NEXT",
+        )
+        stats = mod.Stats()
+        confirmed, unknowns = mod.scan_open_prs(client, "o", "r", recent_cut, SHORT_SHA, stats)
+        assert len(confirmed) == 1 and confirmed[0]["pr"]["number"] == 1
+        assert unknowns == []
+        assert stats.stale == 1 and stats.conflicting == 1
+        assert client.call.call_count == 1  # did not fetch page 2
+
+    def test_buckets_prs_correctly(self, mod):
+        client = MagicMock()
+        client.call.return_value = self._page(
+            [
+                make_pr(number=1, mergeable="CONFLICTING"),
+                make_pr(number=2, mergeable="UNKNOWN"),
+                make_pr(number=3, mergeable="MERGEABLE"),
+                make_pr(number=4, is_draft=True),
+                make_pr(number=5, files=["other.txt"]),
+            ]
+        )
+        stats = mod.Stats()
+        confirmed, unknowns = mod.scan_open_prs(client, "o", "r", STALE_CUT, SHORT_SHA, stats)
+        assert [c["pr"]["number"] for c in confirmed] == [1]
+        assert [u["pr"]["number"] for u in unknowns] == [2]
+        assert stats.conflicting == 1
+        assert stats.unknown == 1
+        assert stats.mergeable == 1
+        assert stats.drafts == 1
+        assert stats.no_uv_lock == 1
+
+    def test_paginates_until_exhausted(self, mod):
+        client = MagicMock()
+        client.call.side_effect = [
+            self._page([make_pr(number=1)], has_next=True, cursor="C1"),
+            self._page([make_pr(number=2)], has_next=False),
+        ]
+        stats = mod.Stats()
+        confirmed, _ = mod.scan_open_prs(client, "o", "r", STALE_CUT, SHORT_SHA, stats)
+        assert [c["pr"]["number"] for c in confirmed] == [1, 2]
+        assert stats.pages == 2
+
+
+class TestRetryUnknowns:
+    @pytest.fixture(autouse=True)
+    def _patch_sleep(self, mod, monkeypatch):
+        """Skip real sleeps so the retry loop is instantaneous under test."""
+        monkeypatch.setattr(mod.time, "sleep", lambda _s: None)
+
+    def test_unknowns_resolve_on_retry(self, mod):
+        client = MagicMock()
+        initial_unknown = {"pr": make_pr(number=7, mergeable="UNKNOWN"), "existing": None}
+        client.call.return_value = {"repository": {"pullRequest": make_pr(number=7, mergeable="CONFLICTING")}}
+        stats = mod.Stats()
+        confirmed: list[dict[str, Any]] = []
+        remaining = mod.retry_unknowns(
+            client, "o", "r", STALE_CUT, SHORT_SHA, confirmed, [initial_unknown], stats
+        )
+        assert remaining == []
+        assert [c["pr"]["number"] for c in confirmed] == [7]
+        assert stats.retries == 1
+
+    def test_gives_up_after_max_retries(self, mod):
+        client = MagicMock()
+        # Always returns UNKNOWN — simulate a PR GitHub never resolves during the run.
+        client.call.return_value = {"repository": {"pullRequest": make_pr(number=8, mergeable="UNKNOWN")}}
+        stats = mod.Stats()
+        initial = [{"pr": make_pr(number=8, mergeable="UNKNOWN"), "existing": None}]
+        remaining = mod.retry_unknowns(client, "o", "r", STALE_CUT, SHORT_SHA, [], initial, stats)
+        assert [r["pr"]["number"] for r in remaining] == [8]
+        assert stats.retries == mod.MAX_RETRIES
+        assert client.call.call_count == mod.MAX_RETRIES
+
+
+class TestPostNotices:
+    def test_updates_when_marker_exists_and_creates_otherwise(self, mod):
+        client = MagicMock()
+        existing_pr = make_pr(number=1, node_id="PR_A")
+        new_pr = make_pr(number=2, node_id="PR_B")
+        confirmed = [
+            {"pr": existing_pr, "existing": {"id": "C_OLD", "body": f"{mod.MARKER}\nold"}},
+            {"pr": new_pr, "existing": None},
+        ]
+        stats = mod.Stats()
+        mod.post_notices(client, confirmed, "BODY", stats)
+        assert stats.updated == 1 and stats.posted == 1
+        assert client.call.call_count == 2
+        # First call must be the update mutation (existing marker); second must be create.
+        (update_query, update_vars), _ = client.call.call_args_list[0]
+        (create_query, create_vars), _ = client.call.call_args_list[1]
+        assert "updateIssueComment" in update_query
+        assert update_vars == {"id": "C_OLD", "body": "BODY"}
+        assert "addComment" in create_query
+        assert create_vars == {"subjectId": "PR_B", "body": "BODY"}
+
+
+class TestWriteSummary:
+    def test_noop_without_path(self, mod):
+        # Should silently do nothing; absence of an error is the assertion.
+        mod.write_summary(None, SHORT_SHA, "ref", mod.Stats())
+
+    def test_writes_table_when_path_given(self, mod, tmp_path):
+        path = tmp_path / "summary.md"
+        stats = mod.Stats(scanned=5, pages=1, conflicting=2, unknown=1, mergeable=1, posted=2, updated=0)
+        mod.write_summary(str(path), SHORT_SHA, "**PR ref**", stats)
+        text = path.read_text()
+        assert f"uv.lock conflict notifier — {SHORT_SHA}" in text
+        assert "**Source of change:** **PR ref**" in text
+        assert "| Scanned | 5 |" in text
+        assert "| Conflicting | 2 |" in text
+        assert "| Notices posted | 2 |" in text
