This is an automated email from the ASF dual-hosted git repository.
potiuk pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/airflow.git
The following commit(s) were added to refs/heads/main by this push:
new 8058c889b27 Harden the fork CI reconciler and protect the fork CI gate
(#74909)
8058c889b27 is described below
commit 8058c889b279ea66026d77a70177ade284e10f04
Author: Jarek Potiuk <[email protected]>
AuthorDate: Tue Oct 27 21:50:29 2026 +0100
Harden the fork CI reconciler and protect the fork CI gate (#74909)
* Retry the fork CI push-again nudge when posting it fails
The state comment is what the next reconciler run compares with. Recording
awaiting_push there before the nudge was posted meant a failed post was
never
retried, and the contributor got no prompt before the inactivity closure.
Generated-by: Claude Opus 5
* Ignore fork-only workflows that skipped every job in the fork CI
tampering check
A workflow or trigger the base branch does not have always counted as
project CI, so a pull request adding a fork-only workflow that its jobs
guard off in apache/airflow was closed for gate tampering although nothing
ran.
Generated-by: Claude Opus 5
* Keep fork CI pull requests sharing a head commit across bases for a
maintainer
The green-trust decision is stored as a commit status, which every pull
request with that head commit shares. Pull requests with different bases
have different changed files, so a decision made for one could mark the
other ready without checking it.
Generated-by: Claude Opus 5
* Attribute fork CI runs to the right pull request in the tampering check
Runs from forks carry no pull request, so they were matched by fork branch
alone. Legitimate project CI against a base branch without the gate, made
before a retarget, or a run of another pull request from the same branch,
could close a gated pull request for gate tampering.
Generated-by: Claude Opus 5
* Close fork CI pull requests that change the fork CI gate
A pull request whose CI runs in the contributor's fork could change the gate
that sent it there: its code, its configuration, the gate job or the
monitor. Such a change decides where the pull request's own CI runs, so the
monitor now closes it and explains why. Contributors can still add, change
or remove their own pilot entry, which is how they join and leave the pilot.
Generated-by: Claude Opus 5
* Count only started jobs of fork-only workflows as project CI work
A fork-only workflow run cancelled before a runner took its jobs reports
them
completed and cancelled, so it counted as project CI work. It now uses the
same started-job test as ci-amd.yml runs.
Generated-by: Claude Opus 5
* Name the cancelled-run CI passed job in the fork CI tampering docs
The tampering check now accepts the name a cancelled ci-amd.yml run gives
its CI passed job, so the docs list it with the other names of that job.
Generated-by: Claude Opus 5
---
.github/fork-ci-gate.toml | 5 +
.github/workflows/ci-amd.yml | 4 +-
.github/workflows/fork-ci-monitor.yml | 3 +
contributing-docs/18_contribution_workflow.rst | 5 +-
contributing-docs/25_maintainer_pr_triage.md | 6 +-
.../doc/ci/08_fork_ci_for_external_contributors.md | 97 ++++-
.../src/airflow_breeze/utils/fork_ci_gate.py | 5 +
.../src/airflow_breeze/utils/fork_ci_github.py | 8 +-
.../src/airflow_breeze/utils/fork_ci_messages.py | 27 ++
.../src/airflow_breeze/utils/fork_ci_reconcile.py | 384 +++++++++++++++--
dev/breeze/tests/test_fork_ci_messages.py | 5 +
dev/breeze/tests/test_fork_ci_reconcile.py | 476 ++++++++++++++++++++-
scripts/ci/prek/check_ci_workflows_in_sync.py | 4 +-
13 files changed, 961 insertions(+), 68 deletions(-)
diff --git a/.github/fork-ci-gate.toml b/.github/fork-ci-gate.toml
index 6fc7a0de1ce..f7a2a608fdc 100644
--- a/.github/fork-ci-gate.toml
+++ b/.github/fork-ci-gate.toml
@@ -16,6 +16,11 @@
# under the License.
# Fork CI gate (AIP-120). See
dev/breeze/doc/ci/08_fork_ci_for_external_contributors.md
+#
+# This file is part of the gate. A pull request whose CI runs in the
contributor's fork may only
+# add, change or remove its author's own [pilot] entry here; the fork CI
monitor closes it if it
+# changes anything else, because that change would decide where its own CI
runs. See
+# "Gate protection" in the document above.
# Master switch. When false the gate job lets every pull request run project
CI, except
# pull requests labelled `use fork ci` and pull requests of [pilot]
participants. Push
diff --git a/.github/workflows/ci-amd.yml b/.github/workflows/ci-amd.yml
index e56377e053b..f709ee36fe2 100644
--- a/.github/workflows/ci-amd.yml
+++ b/.github/workflows/ci-amd.yml
@@ -164,7 +164,9 @@ jobs:
gate:
# Fork CI gate (AIP-120): decides whether project CI runs for this event.
Configuration and
# code come from the base branch so a pull request cannot change its own
eligibility. See
- # dev/breeze/doc/ci/08_fork_ci_for_external_contributors.md.
+ # dev/breeze/doc/ci/08_fork_ci_for_external_contributors.md. A pull
request whose CI runs
+ # in the contributor's fork and changes this job, or how build-info
depends on it, is closed
+ # by the fork CI monitor ("Gate protection" in that document).
name: "Fork CI gate"
runs-on: ["ubuntu-slim"]
timeout-minutes: 5
diff --git a/.github/workflows/fork-ci-monitor.yml
b/.github/workflows/fork-ci-monitor.yml
index ec34b9615b0..e95becc9e27 100644
--- a/.github/workflows/fork-ci-monitor.yml
+++ b/.github/workflows/fork-ci-monitor.yml
@@ -21,6 +21,9 @@
# those pull requests, and deletes their project CI runs that only skipped
every job. It never checks
# out or runs pull request code. See
# dev/breeze/doc/ci/08_fork_ci_for_external_contributors.md.
+#
+# Part of the fork CI gate: a pull request whose CI runs in the contributor's
fork and changes
+# this file is closed by this monitor ("Gate protection" in the document
above).
name: "Fork CI monitor"
on: # yamllint disable-line rule:truthy
schedule:
diff --git a/contributing-docs/18_contribution_workflow.rst
b/contributing-docs/18_contribution_workflow.rst
index 7ebe12fb120..e4d4d9fa378 100644
--- a/contributing-docs/18_contribution_workflow.rst
+++ b/contributing-docs/18_contribution_workflow.rst
@@ -186,7 +186,10 @@ at 00:00 UTC:
"your-login" = YYYY-MM-DDT00:00:00Z
Every pull request you open after that instant uses fork CI; your earlier pull
requests keep project
-CI. To leave the pilot, open a pull request that removes the line. A pilot
entry gates your pull
+CI. To leave the pilot, open a pull request that removes the line. Your own
``[pilot]`` line is
+the only part of the fork CI gate a pull request using fork CI may change: one
that changes the
+gate's code, the rest of its configuration or the ``gate`` job of
``ci-amd.yml`` is closed with an
+explanation. A pilot entry gates your pull
requests whatever your role, so a committer in the pilot also enables the
workflow;
``breeze ci audit`` tells you which applies to you. The design is described in
`Fork CI for external contributors
<../dev/breeze/doc/ci/08_fork_ci_for_external_contributors.md>`__.
diff --git a/contributing-docs/25_maintainer_pr_triage.md
b/contributing-docs/25_maintainer_pr_triage.md
index e3f6909194c..e340f98275e 100644
--- a/contributing-docs/25_maintainer_pr_triage.md
+++ b/contributing-docs/25_maintainer_pr_triage.md
@@ -250,7 +250,7 @@ The triage workflow uses the following labels and states:
## Pull requests gated to fork CI
Pull requests labelled `fork ci required` run their CI in the contributor's
fork. The scheduled
-fork CI reconciler (`breeze ci fork-ci-reconcile`, run by
`fork-ci-monitor.yml`) owns four things
+fork CI reconciler (`breeze ci fork-ci-reconcile`, run by
`fork-ci-monitor.yml`) owns five things
for them, and the triage skill must treat those pull requests as out of scope
for each:
- **Draft state.** The reconciler keeps the pull request in draft until the
fork run for its
@@ -261,6 +261,10 @@ for them, and the triage skill must treat those pull
requests as out of scope fo
- **`needs project ci`.** The reconciler adds it when the fork run is green
but the pull request
changes CI-defining files, and removes it once the state is anything else.
These pull requests
stay in draft and wait for a maintainer to review the CI change and add `use
project ci`.
+- **Closing for changing the gate.** The reconciler closes a pull request that
changes the fork
+ CI gate (its code, configuration other than the author's own `[pilot]`
entry, the `gate` job
+ of `ci-amd.yml` or the monitor), and reports it on Slack. Add `use project
ci` to accept such
+ a change.
- **Closing after inactivity.** The reconciler closes a pull request whose
fork CI has not run for
`close_after_days`.
diff --git a/dev/breeze/doc/ci/08_fork_ci_for_external_contributors.md
b/dev/breeze/doc/ci/08_fork_ci_for_external_contributors.md
index ad06dbc54c5..10a96244cfe 100644
--- a/dev/breeze/doc/ci/08_fork_ci_for_external_contributors.md
+++ b/dev/breeze/doc/ci/08_fork_ci_for_external_contributors.md
@@ -510,6 +510,7 @@ pull requests without a fork run for their head commit, not
by the total.
| **green_unverified**: conclusion success, but the pull request changes
CI-defining files (see "Trusting a green fork run") | draft | remove |
`pending` (yellow), link to the run | update: a maintainer must run project CI;
labelled `needs project ci` |
| setup_required or awaiting_push for `close_after_days` | unchanged |
unchanged | unchanged | close with a new comment; reopening restarts the clock |
| **ungated**: the base branch has no gate, for example after retargeting to
an older release branch | undraft\* | unchanged | `success` (green), link to
the pull request's checks | update: project CI is in use; `fork ci required` is
removed |
+| **gate_modified**: the pull request changes the fork CI gate (see "Gate
protection") | draft | remove | `error` (red), link to the pull request's
changed files | post a **new** comment explaining why, then close; checked
after gate tampering, before the fork state |
| **gate_bypassed**: a recorded project CI run of the pull request's branch
bypassed the gate (see "Gate tampering") | draft | remove | `error` (red), link
to the project CI run that should not have happened | post a **new** comment
explaining why, then close; takes precedence over every row above |
\* only if the last convert-to-draft event on the pull request was by
`github-actions[bot]`;
@@ -545,6 +546,10 @@ Rules that sit under the table:
closing comment or close failed, and the next run finishes the closure. The
same holds for a
`gate_bypassed` marker, where only a reopen by someone other than the author
counts, so the
closure is finished even after the offending run has left the tampering
check's window.
+- **New comments go first.** A nudge, a "fork CI stopped running" notice, a
closing explanation
+ or a tampering notice is posted before the state comment records the new
state, so when
+ posting fails the next run sees the old state and posts it again; a repeat
is preferred to a
+ lost notice.
- **`fork ci required` goes last.** When a run removes the label (on release,
or for an
ungated base branch), it does so after every other change, so a failure on
the way leaves
the label in place and the next run retries the rest.
@@ -572,7 +577,10 @@ The decision is read back from the `Fork CI / Tests (AMD)`
commit status on the
which only `apache/airflow`'s own tokens can write, so it is made once per
commit and cannot be
forged through a comment. The changed files depend on the base branch as well,
so a decision
recorded before the pull request was last retargeted (a `BaseRefChangedEvent`
in the timeline
-newer than the status) is made again, and the status is written again to date
it.
+newer than the status) is made again, and the status is written again to date
it. A commit status
+belongs to the commit, so it cannot hold one decision per base: when another
gated pull
+request with a different base has the same head commit, both are
`green_unverified` and wait
+for a maintainer, reported to Slack once.
This is a guard against the cheapest way to fake a green run, not a guarantee.
Any change can
weaken the tests it ships with (a `conftest.py` that skips tests, for
example), exactly as it
@@ -608,28 +616,36 @@ ignored, because they ran legitimately while the label
was on. So are runs creat
most recent reopen of the pull request by someone other than its author, which
is how a
maintainer brings back a pull request closed for tampering; reopening one
whose gate is still
missing triggers a new run, which is caught again. A reopen by the author does
not count, so
-closing and reopening one's own pull request cannot hide a tampered run. A run
counts as a
-bypass when both signals hold:
+closing and reopening one's own pull request cannot hide a tampered run. Runs
are listed per
+fork branch, because GitHub records no pull request on runs from forks. A run
made while the
+pull request targeted a base branch without the gate, known from its retargets
in the
+timeline, is project CI by design and ignored; its changes are compared with
that base. When
+another pull request from the same branch was open within the window, a
matching run could be
+that one's, so it is reported to Slack as an attention item rather than
closing the pull
+request. A run counts as a bypass when both signals hold:
1. **The run did project CI work.** Either the default branch's version of the
run's workflow
is not triggered by the run's event, which covers an injected workflow and
a trigger the
pull request added to an existing one (for example `pull_request` on
`ci-arm.yml`, which
- has no gate), or it is a `ci-amd.yml` run that has run for more than 5
minutes, finished or
- still running, and whose jobs (`GET .../actions/runs/<id>/jobs`) show more
than the gate: a
- started job that was not skipped, other than `Fork CI gate` and the jobs
that run in every
- gated run (`Check if the run is needed`, which has no steps for pull
requests, and the
- `CI passed` job, named `CI passed (project CI not run)` when project CI was
skipped and
- `CI passed` when a newer push cancelled the run before `Build info`
started), a missing or
- duplicated gate job, one of those jobs appearing twice (either `CI passed`
name counts as
- the same job), or one of them running for more than 5 minutes. The last two
cases catch
- project CI work done in jobs renamed to one of their names. A job counts as
started
- only when a runner picked it up (a non-zero `runner_id`, or steps): a
queued job reports its
- creation as `started_at`, and a run cancelled by a newer push marks its
never-started jobs
- `completed` and `cancelled`, so a gate waiting for a runner would otherwise
look like project
- CI work. A running run is judged as it goes, so one
- tampered to outlast the 6-hour window is still caught. A gate-only run
finishes in about a
- minute, so jobs are listed only for the rare long runs. The triggers are
read from the
- monitor's own checkout of `main`, so no API call is needed; a pull request
into a release
+ has no gate), and one of its jobs started and was not skipped (judged as
for `ci-amd.yml`
+ below when it is that workflow, and counted as work when the jobs cannot be
listed), so a
+ fork-only workflow the pull request adds and guards off in `apache/airflow`
does not count;
+ or it is a `ci-amd.yml` run that has run for more than 5 minutes, finished
or still running,
+ and whose jobs (`GET .../actions/runs/<id>/jobs`) show more than the gate:
a started job that
+ was not skipped, other than `Fork CI gate` and the jobs that run in every
gated run
+ (`Check if the run is needed`, which has no steps for pull requests, and
the `CI passed` job,
+ named `CI passed (project CI not run)` when project CI was skipped and,
when a newer push
+ cancelled the run before `Build info` started, `CI passed (run cancelled)`
-- or `CI passed`
+ from a `ci-amd.yml` that does not use that name yet), a missing or
duplicated gate job, one
+ of those jobs appearing twice (every `CI passed` name counts as the same
job), or one of them
+ running for more than 5 minutes. The last two cases catch project CI work
done in jobs
+ renamed to one of their names. A job counts as started only when a runner
picked it up (a
+ non-zero `runner_id`, or steps): a queued job reports its creation as
`started_at`, and a run
+ cancelled by a newer push marks its never-started jobs `completed` and
`cancelled`, so a gate
+ waiting for a runner would otherwise look like project CI work. A running
run is judged as it
+ goes, so one tampered to outlast the 6-hour window is still caught. A
gate-only run finishes
+ in about a minute, so jobs are listed only for the rare long runs. The
triggers are read from
+ the monitor's own checkout of `main`, so no API call is needed; a pull
request into a release
branch is judged by `main`'s workflows.
2. **The run's own commit changes workflow files.** The compare of the base
branch with that
commit (`GET /repos/apache/airflow/compare/<base>...<fork-owner>:<sha>`,
first page only:
@@ -668,6 +684,46 @@ signal 1 no longer holds and the pull request re-enters
the fork state machine.
ci` releases it like any other gated pull request; a maintainer who wants to
keep an external
workflow change has to leave that label on.
+### Gate protection
+
+The gate decides where a pull request's CI runs, so a pull request whose CI
runs in a fork must
+not change it: the change would decide where its own CI runs, and a maintainer
reviewing it later
+would be reviewing a pull request that already escaped the gate. For every
pull request the
+tampering check applies to (`external` and `pilot`), the reconciler lists the
changed files once
+per head commit, counting a renamed file under both paths, and closes the pull
request
+with a comment explaining why when it changes any of:
+
+- `dev/breeze/src/airflow_breeze/utils/fork_ci_gate.py`, the reconciler modules
+ `fork_ci_reconcile.py`, `fork_ci_github.py` and `fork_ci_messages.py`, or
+ `.github/workflows/fork-ci-monitor.yml`;
+- `.github/fork-ci-gate.toml`, other than adding, changing or removing the
author's own
+ `[pilot]` entry, which is how contributors join and leave the pilot (the
configuration is
+ parsed from the head commit and compared with its version at the pull
request's merge base,
+ so comments and later changes on the base branch do not count);
+- the `gate` job of `.github/workflows/ci-amd.yml`, the `needs` and `if` that
make
+ `build-info` depend on it, the set of jobs that do not depend on the gate
(directly or through
+ other jobs), or the workflow-level `on`, `defaults`, `env`, `permissions`
and `concurrency` (the head and merge-base versions are parsed and those parts
+ compared, so other changes to the workflow are left to the green-trust
check; a head version
+ that does not parse as a workflow counts as a change).
+
+A changed-file listing at GitHub's 3000-file cap may leave out a gate file, so
such a pull
+request is closed as one whose gate change cannot be checked.
+
+The state comment's marker records `"gate_checked": true` with the head commit
only when this
+check ran for it, so a marker written before the check existed, or while `use
fork ci` exempted
+the pull request, does not count, and a retarget since the state began makes
it check again.
+The changed files are listed for the pull request as it is when listed, so the
head commit is
+read again afterwards: if it moved, nothing is recorded and the next run
checks the new one.
+
+The state is `gate_modified`: the pull request is drafted, loses `ready for
maintainer review`
+and `needs project ci`, gets an `error` status linking its changed files, and
the closure is
+reported to Slack as an attention item. Gate tampering is checked first and
wins. As for gate
+tampering, only a reopen by someone other than the author counts: an open pull
request whose
+marker says `gate_modified` and that was not reopened since is closed again,
and a maintainer's
+reopen without restoring the gate closes it again on the next run. A
maintainer who wants the
+change adds `use project ci`, which releases the pull request before any of
these checks. Each
+of these files carries a comment saying so.
+
### State comment
One comment per pull request, created on first contact and edited afterwards.
It starts with
@@ -1091,7 +1147,8 @@ as a failed check rather than a crash.
CI-defining files keeps the pull request a draft, decided once per commit)
and the gate
tampering check: a tampered run on an earlier commit is caught after a clean
push, each
signal alone does not close, short runs and gate-only jobs do not count, an
in-progress
- non-gate job, a removed, duplicated or long-running gate job and an injected
workflow do,
+ non-gate job, a removed, duplicated or long-running gate job and an injected
workflow with a
+ job that ran do, an added workflow whose jobs all skipped does not,
runs before the last `use project ci` removal or the last reopen are
ignored, an
uncheckable 300-file change set is reported, `label_fork_ci` pull requests
are exempt and
`pilot` pull requests are checked. Discovery labels a pilot participant's
diff --git a/dev/breeze/src/airflow_breeze/utils/fork_ci_gate.py
b/dev/breeze/src/airflow_breeze/utils/fork_ci_gate.py
index 10fcc9a106a..33c5dd97543 100644
--- a/dev/breeze/src/airflow_breeze/utils/fork_ci_gate.py
+++ b/dev/breeze/src/airflow_breeze/utils/fork_ci_gate.py
@@ -22,6 +22,11 @@ to CI in the contributor's fork. See
dev/breeze/doc/ci/08_fork_ci_for_external_c
The gate job in ``ci-amd.yml`` runs this file with the runner's ``python3``
from a sparse
checkout, without installing breeze, so it must only import from the standard
library.
+
+This file is part of the fork CI gate. A pull request whose CI runs in the
contributor's fork
+must not change it, because the change would decide where its own CI runs: the
fork CI
+monitor closes such a pull request with a comment explaining why. See "Gate
protection" in
+the document above.
"""
from __future__ import annotations
diff --git a/dev/breeze/src/airflow_breeze/utils/fork_ci_github.py
b/dev/breeze/src/airflow_breeze/utils/fork_ci_github.py
index 721ecc7873d..77a7ef91011 100644
--- a/dev/breeze/src/airflow_breeze/utils/fork_ci_github.py
+++ b/dev/breeze/src/airflow_breeze/utils/fork_ci_github.py
@@ -14,7 +14,13 @@
# KIND, either express or implied. See the License for the
# specific language governing permissions and limitations
# under the License.
-"""GitHub access for the fork CI reconciler (AIP-120); the only module of it
that calls ``gh``."""
+"""
+GitHub access for the fork CI reconciler (AIP-120); the only module of it that
calls ``gh``.
+
+Part of the fork CI gate: a pull request whose CI runs in the contributor's
fork and changes
+this file is closed by the fork CI monitor. See "Gate protection" in
+dev/breeze/doc/ci/08_fork_ci_for_external_contributors.md.
+"""
from __future__ import annotations
diff --git a/dev/breeze/src/airflow_breeze/utils/fork_ci_messages.py
b/dev/breeze/src/airflow_breeze/utils/fork_ci_messages.py
index 90f6b6cba51..d19162d448f 100644
--- a/dev/breeze/src/airflow_breeze/utils/fork_ci_messages.py
+++ b/dev/breeze/src/airflow_breeze/utils/fork_ci_messages.py
@@ -20,6 +20,10 @@ Texts the fork CI reconciler posts on pull requests
(AIP-120).
All wording lives here so it can be reviewed in one place. Texts must say that
CI is free for the
contributor and finite for the project, must not characterise the recipient,
and must give the
exact command to run.
+
+Part of the fork CI gate: a pull request whose CI runs in the contributor's
fork and changes
+this file is closed by the fork CI monitor. See "Gate protection" in
+dev/breeze/doc/ci/08_fork_ci_for_external_contributors.md.
"""
from __future__ import annotations
@@ -80,6 +84,9 @@ class Marker:
behind: int | None = None
# The completed fork run (``<run id>:<attempt>:<conclusion>``) whose jobs
are all reported.
jobs: str | None = None
+ # Set only by a reconciler that ran the gate protection check for this
head commit, so a
+ # marker written before that check existed, or while it did not apply,
never counts as one.
+ gate_checked: bool = False
def render_marker(marker: Marker) -> str:
@@ -88,6 +95,8 @@ def render_marker(marker: Marker) -> str:
data["behind"] = marker.behind
if marker.jobs is not None:
data["jobs"] = marker.jobs
+ if marker.gate_checked:
+ data["gate_checked"] = True
return f"{MARKER_PREFIX}{json.dumps(data, sort_keys=True)} -->"
@@ -104,6 +113,7 @@ def parse_marker(body: str) -> Marker | None:
since=datetime.fromisoformat(data["since"]),
behind=data.get("behind"),
jobs=data.get("jobs"),
+ gate_checked=data.get("gate_checked") is True,
)
except (json.JSONDecodeError, KeyError, TypeError, ValueError) as e:
raise ValueError(f"Malformed fork CI marker: {match.group(0)!r}") from
e
@@ -145,6 +155,7 @@ _STATE_TEXTS = {
"comment below for how to continue.",
"gate_bypassed": "**Fork CI: closed.** Project CI ran for this pull
request although its CI runs in "
"the contributor's fork; see the comment below.",
+ "gate_modified": "**Fork CI: closed.** This pull request changes the fork
CI gate; see the comment below.",
}
@@ -181,6 +192,20 @@ def released_comment(marker: Marker) -> str:
)
+def gate_modified_close_comment(*, change: str | None) -> str:
+ return (
+ f"Closing this pull request because it changes the fork CI gate{f':
{change}' if change else ''}.\n\n"
+ "The gate decides whether a pull request runs its CI in apache/airflow
or in the contributor's "
+ "fork. It is made of the `gate` job of `.github/workflows/ci-amd.yml`,
"
+ "`dev/breeze/src/airflow_breeze/utils/fork_ci_gate.py`,
`.github/fork-ci-gate.toml` and the fork CI "
+ "monitor, and a pull request whose CI runs in a fork must not change
it, because the change would "
+ "decide where its own CI runs. The one exception is adding, changing
or removing your own `[pilot]` "
+ "entry in `.github/fork-ci-gate.toml`.\n\n"
+ "If the change is wanted, a maintainer can reopen the pull request and
add the `use project ci` "
+ "label." + FOOTER
+ )
+
+
def ungated_comment(marker: Marker, *, base_ref: str) -> str:
return (
f"{render_marker(marker)}\n\n"
@@ -269,6 +294,7 @@ _STATUS_DESCRIPTIONS = {
"gate_bypassed": "Project CI ran for a gated pull request - see Details",
"released": "Project CI in use (label `use project ci`) - see the checks",
"ungated": "Project CI in use (base branch without fork CI) - see the
checks",
+ "gate_modified": "Pull request changes the fork CI gate - see the comment",
}
# (commit status state, description) of a fork job, by its status and
conclusion. A queued job is
@@ -308,6 +334,7 @@ STATUS_STATES = {
"gate_bypassed": "error",
"released": "success",
"ungated": "success",
+ "gate_modified": "error",
}
diff --git a/dev/breeze/src/airflow_breeze/utils/fork_ci_reconcile.py
b/dev/breeze/src/airflow_breeze/utils/fork_ci_reconcile.py
index cd9edf5a44c..79b5035ac7c 100644
--- a/dev/breeze/src/airflow_breeze/utils/fork_ci_reconcile.py
+++ b/dev/breeze/src/airflow_breeze/utils/fork_ci_reconcile.py
@@ -20,6 +20,9 @@ Scheduled reconciler for pull requests gated to fork CI
(AIP-120).
``classify`` is a pure function from what GitHub recorded about one pull
request to the mutations
that bring it to the desired state; ``run_reconcile`` gathers that data and
applies the result.
See "Reconciler" in dev/breeze/doc/ci/08_fork_ci_for_external_contributors.md.
+
+Part of the fork CI gate: a pull request whose CI runs in the contributor's
fork and changes
+this file is closed by the fork CI monitor, see "Gate protection" in the same
document.
"""
from __future__ import annotations
@@ -64,6 +67,7 @@ from airflow_breeze.utils.fork_ci_messages import (
STATUS_STATES,
Marker,
fork_actions_url,
+ gate_modified_close_comment,
inactivity_close_comment,
job_status_context,
nudge_comment,
@@ -80,6 +84,19 @@ CI_WORKFLOW_FILE = "ci-amd.yml"
# The files the gate job of ci-amd.yml reads from a pull request's base branch.
GATE_CONFIG_PATH = ".github/fork-ci-gate.toml"
GATE_SCRIPT_PATH = "dev/breeze/src/airflow_breeze/utils/fork_ci_gate.py"
+CI_WORKFLOW_PATH = f".github/workflows/{CI_WORKFLOW_FILE}"
+# Files of the fork CI gate that a pull request whose CI runs in a fork must
not change at all.
+# The gate configuration and the gate job in ci-amd.yml are checked
separately, since the first
+# may carry the author's own pilot entry and the second shares its file with
the CI pipeline.
+GATE_PROTECTED_PATHS = frozenset(
+ {
+ GATE_SCRIPT_PATH,
+ "dev/breeze/src/airflow_breeze/utils/fork_ci_github.py",
+ "dev/breeze/src/airflow_breeze/utils/fork_ci_messages.py",
+ "dev/breeze/src/airflow_breeze/utils/fork_ci_reconcile.py",
+ ".github/workflows/fork-ci-monitor.yml",
+ }
+)
GATE_JOB_NAME = "Fork CI gate"
# Jobs that also run, without needing the gate, in a run the gate left to fork
CI: main's
# check-run-needed (no steps outside schedule and push runs) and the "CI
passed" gate job of the
@@ -159,6 +176,7 @@ class ForkState(str, Enum):
CLOSED = "closed"
RELEASED = "released"
UNGATED = "ungated"
+ GATE_MODIFIED = "gate_modified"
CLOSABLE_STATES = frozenset({ForkState.SETUP_REQUIRED,
ForkState.AWAITING_PUSH})
@@ -212,6 +230,8 @@ class PullRequestSnapshot:
last_reopened_at: datetime | None = None
last_reopened_by_anyone_at: datetime | None = None
last_base_change_at: datetime | None = None
+ # (changed at, previous base, new base) for each retarget, oldest first.
+ base_changes: tuple[tuple[datetime, str, str], ...] = ()
current_status: tuple[str, str, str] | None = None
status_created_at: datetime | None = None
ci_passed_state: str | None = None
@@ -447,6 +467,8 @@ def classify(
changes_ci_files: Callable[[], bool | None],
count_behind: Callable[[], int | None] = lambda: None,
jobs_reported: str | None = None,
+ head_shared_across_bases: bool = False,
+ find_gate_change: Callable[[], str | None] = lambda: None,
) -> Plan:
"""
Decide the mutations for one gated pull request.
@@ -463,6 +485,11 @@ def classify(
unknown; called at the same time as ``changes_ci_files``.
:param jobs_reported: the completed fork run whose jobs are all reported
as statuses, recorded
in the marker so that later runs do not read its jobs again.
+ :param head_shared_across_bases: another gated pull request has the same
head commit and a
+ different base, so their changed files differ but they share one
commit status.
+ :param find_gate_change: describes how the pull request changes the fork
CI gate, or returns
+ None; only called for pull requests that the gate tampering check
applies to, once per
+ head commit.
"""
plan = Plan()
labels = set(pr.labels)
@@ -525,6 +552,7 @@ def classify(
reopened_since_closing = {
ForkState.CLOSED.value: pr.last_reopened_by_anyone_at,
ForkState.GATE_BYPASSED.value: pr.last_reopened_at,
+ ForkState.GATE_MODIFIED.value: pr.last_reopened_at,
}
if marker and marker.state in reopened_since_closing:
reopened = reopened_since_closing[marker.state]
@@ -533,6 +561,8 @@ def classify(
_set_label(plan, labels, LABEL_NEEDS_PROJECT_CI, present=False)
if plan.state == ForkState.CLOSED:
comment =
inactivity_close_comment(days=config.close_after_days, fork=pr.head_repo)
+ elif plan.state == ForkState.GATE_MODIFIED:
+ comment = gate_modified_close_comment(change=None)
else:
# The status written with the marker links the run that
bypassed the gate.
bypass_url = (
@@ -545,6 +575,44 @@ def classify(
plan.close = True
return plan
+ # A pull request whose CI runs in a fork must not change the gate that
sent it there. Checked
+ # once per head commit: the marker records a head commit as checked only
when the check ran
+ # for it (not while `use fork ci` exempted it), and a retarget since the
state began changes
+ # what the pull request changes. A
+ # maintainer's reopen of one closed for it leaves the marker at
gate_modified, so reopening
+ # without restoring the gate closes it again; `use project ci` is the way
to accept the change.
+ gate_checked = reason in TAMPER_CHECKED_REASONS
+ already_checked = (
+ marker is not None
+ and marker.gate_checked
+ and marker.sha == pr.head_sha
+ and (pr.last_base_change_at is None or pr.last_base_change_at <
marker.since)
+ )
+ if gate_checked and not already_checked:
+ change = find_gate_change()
+ if change is not None:
+ plan.state = ForkState.GATE_MODIFIED
+ if not pr.is_draft:
+ plan.draft = True
+ _set_label(plan, labels, LABEL_READY_FOR_REVIEW, present=False)
+ _set_label(plan, labels, LABEL_NEEDS_PROJECT_CI, present=False)
+ _set_status(
+ plan, pr, ForkState.GATE_MODIFIED.value,
f"https://github.com/{repo}/pull/{pr.number}/files";
+ )
+ _set_marker_comment(
+ plan,
+ pr,
+ state_comment(
+ Marker(ForkState.GATE_MODIFIED.value, pr.head_sha, now),
fork=pr.head_repo, run_url=None
+ ),
+ )
+
plan.new_comments.append(gate_modified_close_comment(change=change))
+ plan.close = True
+ plan.attention.append(
+ f"closed for changing the fork CI gate (author
@{pr.author_login}): {change}"
+ )
+ return plan
+
state, unknown_reason = observe_fork_state(fork, marker)
if state is None:
plan.unknown.append(unknown_reason or "unknown fork state")
@@ -555,14 +623,22 @@ def classify(
# The decision is read back from the commit status on the head commit,
which only
# apache/airflow's own tokens can write, never from a comment. The
changed files depend
# on the base too, so a status written before the pull request was
retargeted is stale.
- # The status is shared by every pull request with this head commit;
scoping it to one is
- # tracked at https://github.com/apache/airflow/issues/74907
previous = status_state(pr.current_status)
decided = previous in (ForkState.GREEN.value,
ForkState.GREEN_UNVERIFIED.value)
stale = pr.last_base_change_at is not None and (
pr.status_created_at is None or pr.status_created_at <=
pr.last_base_change_at
)
- if decided and not stale:
+ if head_shared_across_bases:
+ # One commit status cannot hold a decision per base, and trusting
the one written for
+ # the other pull request could skip the check; both wait for a
maintainer instead.
+ unverified = True
+ if previous != ForkState.GREEN_UNVERIFIED.value:
+ plan.attention.append(
+ "fork CI is green, but another gated pull request to a
different base has the "
+ "same head commit, so its CI-defining changes cannot be
told apart; review and add "
+ "`use project ci` to verify"
+ )
+ elif decided and not stale:
unverified = previous == ForkState.GREEN_UNVERIFIED.value
else:
# Writing the status again, even unchanged, records when this
decision was made.
@@ -578,13 +654,10 @@ def classify(
plan.state = state
since = marker.since if marker and marker.state == state.value else now
- # The marker records awaiting_push before the nudge is posted, so a failed
post is not retried;
- # tracked at https://github.com/apache/airflow/issues/74908
if marker and marker.state == ForkState.SETUP_REQUIRED.value and state ==
ForkState.AWAITING_PUSH:
plan.new_comments.append(nudge_comment(fork=pr.head_repo))
# A new comment, because the edited state comment would not notify the
author that fork CI they
- # had working stopped running. Like the nudge, it is not retried when
posting it fails; tracked at
- # https://github.com/apache/airflow/issues/74908
+ # had working stopped running.
if (
marker
and marker.state not in (ForkState.SETUP_REQUIRED.value,
ForkState.CLOSED.value)
@@ -600,7 +673,11 @@ def classify(
_set_marker_comment(
plan,
pr,
- state_comment(Marker(ForkState.CLOSED.value, pr.head_sha, now),
fork=pr.head_repo, run_url=None),
+ state_comment(
+ Marker(ForkState.CLOSED.value, pr.head_sha, now,
gate_checked=gate_checked),
+ fork=pr.head_repo,
+ run_url=None,
+ ),
)
plan.new_comments.append(inactivity_close_comment(days=config.close_after_days,
fork=pr.head_repo))
plan.close = True
@@ -629,7 +706,14 @@ def classify(
plan,
pr,
state_comment(
- Marker(state.value, pr.head_sha, since, behind=behind if green
else None, jobs=jobs_reported),
+ Marker(
+ state.value,
+ pr.head_sha,
+ since,
+ behind=behind if green else None,
+ jobs=jobs_reported,
+ gate_checked=gate_checked,
+ ),
fork=pr.head_repo,
run_url=fork.run_url,
base_ref=pr.base_ref,
@@ -674,7 +758,7 @@ query($q: String!, $cursor: String) {
... on ConvertToDraftEvent { createdAt actor { login } }
... on UnlabeledEvent { createdAt label { name } }
... on ReopenedEvent { createdAt actor { login } }
- ... on BaseRefChangedEvent { createdAt }
+ ... on BaseRefChangedEvent { createdAt previousRefName
currentRefName }
}
}
commits(last: 1) { nodes { commit {
@@ -730,22 +814,25 @@ def parse_pull_request(node: dict[str, Any]) ->
PullRequestSnapshot:
last_reopened = None
last_reopened_by_anyone = None
last_base_change = None
+ base_changes = []
for item in (node.get("timelineItems") or {}).get("nodes") or []:
if item["__typename"] == "ConvertToDraftEvent":
last_draft_actor = _login(item.get("actor"))
elif item["__typename"] == "BaseRefChangedEvent":
last_base_change = parse_github_datetime(item["createdAt"])
+ base_changes.append(
+ (last_base_change, item.get("previousRefName") or "",
item.get("currentRefName") or "")
+ )
elif item["__typename"] == "ReopenedEvent":
last_reopened_by_anyone = parse_github_datetime(item["createdAt"])
# Only a reopen by someone else counts for the gate tampering
check: an author closing
# and reopening their own pull request must not hide a tampered
run from it.
if _login(item.get("actor")).casefold() !=
_login(node.get("author")).casefold():
last_reopened = last_reopened_by_anyone
- elif (
- item["__typename"] == "UnlabeledEvent"
- and (item.get("label") or {}).get("name") == LABEL_USE_PROJECT_CI
- ):
- last_unlabel = parse_github_datetime(item["createdAt"])
+ elif item["__typename"] == "UnlabeledEvent":
+ label = (item.get("label") or {}).get("name")
+ if label == LABEL_USE_PROJECT_CI:
+ last_unlabel = parse_github_datetime(item["createdAt"])
current_status = None
status_created_at = None
@@ -791,6 +878,7 @@ def parse_pull_request(node: dict[str, Any]) ->
PullRequestSnapshot:
last_reopened_at=last_reopened,
last_reopened_by_anyone_at=last_reopened_by_anyone,
last_base_change_at=last_base_change,
+ base_changes=tuple(base_changes),
current_status=current_status,
status_created_at=status_created_at,
ci_passed_state=ci_passed_state,
@@ -817,6 +905,17 @@ def _is_not_found(error: GitHubError) -> bool:
return "404" in str(error) or "Not Found" in str(error)
+def read_file(client: GitHubClient, repo: str, path: str, ref: str) -> str |
None:
+ """Return a file's content at a ref, or None when it does not exist
there."""
+ try:
+ content =
client.rest_get(f"repos/{repo}/contents/{path}?ref={quote(ref, safe='')}")
+ except GitHubError as e:
+ if _is_not_found(e):
+ return None
+ raise
+ return base64.b64decode(content["content"]).decode()
+
+
class BaseGateConfigs:
"""
The gate configuration of each base branch, as the gate job of a pull
request against it reads it.
@@ -831,15 +930,7 @@ class BaseGateConfigs:
self._configs: dict[str, GateConfig | None] = {default_branch:
default_config}
def _read(self, path: str, base_ref: str) -> str | None:
- try:
- content = self._client.rest_get(
- f"repos/{self._repo}/contents/{path}?ref={quote(base_ref,
safe='')}"
- )
- except GitHubError as e:
- if _is_not_found(e):
- return None
- raise
- return base64.b64decode(content["content"]).decode()
+ return read_file(self._client, self._repo, path, base_ref)
def get(self, base_ref: str) -> GateConfig | None:
"""Return the base branch's configuration, or None when the base
branch has no gate."""
@@ -1102,6 +1193,15 @@ def load_workflow_triggers(workflows_dir: Path) ->
dict[str, frozenset[str]]:
return triggers
+def base_at(pr: PullRequestSnapshot, when: datetime) -> str:
+ """The base branch the pull request had at ``when``, from its retargets."""
+ base = pr.base_ref
+ for changed_at, previous, _ in reversed(pr.base_changes):
+ if when < changed_at:
+ base = previous
+ return base
+
+
class BypassFinder:
"""
Finds project CI runs that bypassed the gate, from the runs apache/airflow
recorded.
@@ -1112,17 +1212,28 @@ class BypassFinder:
signals of the design must hold for a run: it did project CI work, and its
own commit changes
`.github/workflows/`. A run did project CI work when the base branch's
version of its
workflow is not triggered by the run's event (an injected workflow, or a
trigger the pull
- request added to an existing one, such as `pull_request` on ci-arm.yml),
or when it is a long
- ci-amd.yml run whose jobs show more than the gate.
+ request added to an existing one, such as `pull_request` on ci-arm.yml)
and a job of it ran,
+ or when it is a long ci-amd.yml run whose jobs show more than the gate.
"""
def __init__(
- self, client: GitHubClient, *, repo: str, now: datetime,
base_triggers: dict[str, frozenset[str]]
+ self,
+ client: GitHubClient,
+ *,
+ repo: str,
+ now: datetime,
+ base_triggers: dict[str, frozenset[str]],
+ is_gated_base: Callable[[str], bool] = lambda base_ref: True,
):
+ """:param is_gated_base: whether a base branch has the gate; runs
against one without it are project CI by design."""
self._client = client
self._repo = repo
self._now = now
self._base_triggers = base_triggers
+ self._is_gated_base = is_gated_base
+ # Pull requests whose matching run could belong to another pull
request from the same
+ # branch; reported instead of closed.
+ self.ambiguous: list[tuple[int, str]] = []
since = (now - TAMPER_WINDOW).strftime("%Y-%m-%dT%H:%M:%SZ")
self._runs_by_branch: dict[tuple[str, str], list[WorkflowRun]] = {}
incomplete = []
@@ -1159,9 +1270,19 @@ class BypassFinder:
def _did_run_project_ci(self, run: WorkflowRun) -> bool:
if run.id not in self._did_project_ci:
if run.event not in self._base_triggers.get(run.workflow_file,
frozenset()):
- # Counts even when every job skipped; tracked at
- # https://github.com/apache/airflow/issues/74905
- result = True
+ # A workflow or trigger the base branch does not have, for
example a fork-only
+ # workflow the pull request adds, did no work in
apache/airflow if its jobs all
+ # skipped there. Jobs that cannot be listed count as work.
+ try:
+ jobs = self._get_jobs(run)
+ except GitHubError:
+ result = True
+ else:
+ result = (
+ jobs_show_project_ci(jobs, self._now)
+ if run.workflow_file == CI_WORKFLOW_FILE
+ else any(has_job_started(job) and
job.get("conclusion") != "skipped" for job in jobs)
+ )
elif run.workflow_file != CI_WORKFLOW_FILE or not
run.is_long_running(self._now):
result = False
else:
@@ -1208,11 +1329,26 @@ class BypassFinder:
self._changes_workflows[key] = None if len(files) >=
COMPARE_FILES_CAP else False
return self._changes_workflows[key]
+ def _other_pull_requests(self, pr: PullRequestSnapshot) -> list[int]:
+ """Other pull requests from the same branch that were open within the
tampering window."""
+ owner = pr.head_repo.split("/", 1)[0] # type: ignore[union-attr]
+ pulls = self._client.rest_get(
+ f"repos/{self._repo}/pulls?head={quote(owner,
safe='')}:{quote(pr.head_ref, safe='')}"
+ "&state=all&per_page=100",
+ paginate=True,
+ )
+ window_start = self._now - TAMPER_WINDOW
+ return [
+ pull["number"]
+ for pull in pulls
+ if pull["number"] != pr.number
+ and (pull.get("closed_at") is None or
parse_github_datetime(pull["closed_at"]) >= window_start)
+ ]
+
def find(self, pr: PullRequestSnapshot) -> str | None:
if not pr.head_repo:
return None
- # Matching by fork branch alone also picks up runs of another pull
request from that branch
- # or from before a retarget; tracked at
https://github.com/apache/airflow/issues/74906
+ # Runs are listed per fork branch: GitHub records no pull request on
runs from forks.
runs = self._runs_by_branch.get((pr.head_repo.casefold(),
pr.head_ref), [])
# Runs before `use project ci` was removed were legitimate. Runs
before a maintainer
# reopened the pull request were already acted on; reopening without
restoring the gate
@@ -1223,15 +1359,161 @@ class BypassFinder:
for run in sorted(runs, key=lambda r: r.created_at, reverse=True):
if not_before is not None and run.created_at <= not_before:
continue
+ base_ref = base_at(pr, run.created_at)
+ # Project CI against a base branch without the gate is what its
gate job decides.
+ if not self._is_gated_base(base_ref):
+ continue
if not self._did_run_project_ci(run):
continue
# A change set too large to list fails closed: padding a tampered
commit past the
# compare API's cap must not hide the workflow change.
- if self._commit_changes_workflows(run, pr.base_ref) is not False:
+ if self._commit_changes_workflows(run, base_ref) is not False:
+ others = self._other_pull_requests(pr)
+ if others:
+ self.ambiguous.append(
+ (
+ pr.number,
+ f"project CI run {run.html_url} may have bypassed
the gate, but it could also "
+ f"belong to {', '.join(f'#{n}' for n in others)}
from the same branch; not closed, "
+ "check it",
+ )
+ )
+ return None
return run.html_url
return None
+def get_gate_wiring(workflow_text: str | None) -> Any:
+ """
+ The parts of a ci-amd.yml that make up the gate: its job, how build-info
depends on it, the
+ jobs that do not depend on it (a job freed from the gate runs whatever it
decides), and the
+ workflow-level triggers and settings that every job, the gate included,
runs with.
+
+ A file that is not a workflow mapping yields an object equal to nothing,
so a pull request
+ breaking the workflow counts as changing the gate rather than escaping the
check.
+ """
+ if workflow_text is None:
+ return None
+ try:
+ workflow = yaml.safe_load(workflow_text)
+ except yaml.YAMLError:
+ return object()
+ jobs = workflow.get("jobs") if isinstance(workflow, dict) else None
+ if not isinstance(jobs, dict):
+ return object()
+ build_info = jobs.get("build-info")
+ build_info = build_info if isinstance(build_info, dict) else {}
+ return (
+ jobs.get("gate"),
+ build_info.get("needs"),
+ build_info.get("if"),
+ find_jobs_outside_gate(jobs),
+ workflow.get("on", workflow.get(True)),
+ workflow.get("defaults"),
+ workflow.get("env"),
+ workflow.get("permissions"),
+ workflow.get("concurrency"),
+ )
+
+
+def find_jobs_outside_gate(jobs: dict[str, Any]) -> frozenset[str]:
+ """Jobs that do not depend on the gate job, directly or through other
jobs."""
+
+ def get_needs(job_id: str) -> list[str]:
+ job = jobs.get(job_id)
+ needs = job.get("needs") if isinstance(job, dict) else None
+ if isinstance(needs, str):
+ return [needs]
+ return [need for need in needs if isinstance(need, str)] if
isinstance(needs, list) else []
+
+ def depends_on_gate(job_id: str, seen: frozenset[str] = frozenset()) ->
bool:
+ # `seen` stops a dependency cycle, which GitHub rejects anyway.
+ return any(
+ need == "gate" or (need not in seen and depends_on_gate(need, seen
| {job_id}))
+ for need in get_needs(job_id)
+ )
+
+ return frozenset(job_id for job_id in jobs if job_id != "gate" and not
depends_on_gate(job_id))
+
+
+def is_own_pilot_change(base_text: str | None, head_text: str | None,
author_login: str) -> bool:
+ """Whether a gate configuration differs from the base only in the author's
own pilot entry."""
+ if base_text is None or head_text is None:
+ return False
+ try:
+ base = parse_config(base_text)
+ head = parse_config(head_text)
+ except (ValueError, TypeError, AttributeError):
+ # Valid TOML of the wrong shape, such as `pilot = []`, is no permitted
change either.
+ return False
+ base_pilot = {login.casefold(): joined for login, joined in
base.pilot.items()}
+ head_pilot = {login.casefold(): joined for login, joined in
head.pilot.items()}
+ changed = {
+ login
+ for login in base_pilot.keys() | head_pilot.keys()
+ if base_pilot.get(login) != head_pilot.get(login)
+ }
+ return replace(head, pilot=base.pilot) == base and changed <=
{author_login.casefold()}
+
+
+class HeadMovedError(RuntimeError):
+ """The pull request's head moved while it was checked; the next run checks
the new head."""
+
+
+class GateChangeFinder:
+ """Finds how a pull request changes the fork CI gate, from its changed
files."""
+
+ def __init__(self, client: GitHubClient, *, repo: str):
+ self._client = client
+ self._repo = repo
+
+ def _merge_base(self, pr: PullRequestSnapshot) -> str:
+ # The pull request's changes are against the merge base, not the base
branch's tip, which
+ # may have moved on: comparing with the tip would count others' later
changes as its own.
+ owner = (pr.head_repo or "").split("/", 1)[0]
+ compare = self._client.rest_get(
+ f"repos/{self._repo}/compare/{quote(pr.base_ref,
safe='')}...{owner}:{pr.head_sha}?per_page=1"
+ )
+ return compare["merge_base_commit"]["sha"]
+
+ def find(self, pr: PullRequestSnapshot) -> str | None:
+ files = self._client.rest_get(
+ f"repos/{self._repo}/pulls/{pr.number}/files?per_page=100",
paginate=True
+ )
+ # The listing is of the pull request as it is now, while the head
commit recorded as checked
+ # is the one read at the start of the run: a push in between would
mark a commit checked
+ # by another one's files.
+ current_head =
self._client.rest_get(f"repos/{self._repo}/pulls/{pr.number}")["head"]["sha"]
+ if current_head != pr.head_sha:
+ raise HeadMovedError(
+ f"head moved from {pr.head_sha[:8]} to {current_head[:8]}
during the gate check"
+ )
+ # A listing at GitHub's cap may leave out a gate file, so it is not
trusted.
+ if len(files) >= PULL_REQUEST_FILES_CAP:
+ return f"it changes {len(files)} or more files, more than GitHub
lists, so the gate cannot be checked"
+ paths = set(changed_paths(files))
+ for path in sorted(GATE_PROTECTED_PATHS & paths):
+ return f"it changes `{path}`"
+ if not paths & {GATE_CONFIG_PATH, CI_WORKFLOW_PATH}:
+ return None
+ head_repo = pr.head_repo or ""
+ merge_base = self._merge_base(pr)
+ if GATE_CONFIG_PATH in paths and not is_own_pilot_change(
+ read_file(self._client, self._repo, GATE_CONFIG_PATH, merge_base),
+ read_file(self._client, head_repo, GATE_CONFIG_PATH, pr.head_sha),
+ pr.author_login,
+ ):
+ return f"it changes `{GATE_CONFIG_PATH}` beyond the author's own
`[pilot]` entry"
+ if CI_WORKFLOW_PATH in paths and get_gate_wiring(
+ read_file(self._client, self._repo, CI_WORKFLOW_PATH, merge_base)
+ ) != get_gate_wiring(read_file(self._client, head_repo,
CI_WORKFLOW_PATH, pr.head_sha)):
+ return (
+ f"it changes the `gate` job of `{CI_WORKFLOW_PATH}`, which
jobs depend on it, or the "
+ "workflow-level triggers or settings it runs with"
+ )
+ return None
+
+
COMMENTS_QUERY = """
query($owner: String!, $name: String!, $number: Int!, $cursor: String) {
repository(owner: $owner, name: $name) { pullRequest(number: $number) {
@@ -1319,6 +1601,14 @@ def count_commits_behind(client: GitHubClient, repo:
str, pr: PullRequestSnapsho
return compare.get("behind_by")
+def find_heads_shared_across_bases(prs: Iterable[PullRequestSnapshot]) ->
set[str]:
+ """Head commits of more than one pull request whose bases differ."""
+ bases_by_head: dict[str, set[str]] = {}
+ for pr in prs:
+ bases_by_head.setdefault(pr.head_sha, set()).add(pr.base_ref)
+ return {head for head, bases in bases_by_head.items() if len(bases) > 1}
+
+
@dataclass(frozen=True)
class DraftChange:
"""A draft state change, applied by the monitor's draft-state job rather
than the reconciler."""
@@ -1347,6 +1637,11 @@ def apply_plan(client: GitHubClient, pr:
PullRequestSnapshot, plan: Plan) -> Dra
for label in plan.remove_labels:
if label != LABEL_FORK_CI_REQUIRED:
client.remove_label(pr.number, label)
+ # New comments go before the state comment records the new state: the
state recorded there is
+ # what the next run compares with, so a failure on the way posts them
again rather than never.
+ # They carry no marker, so the state comment stays the earliest bot
comment with one.
+ for comment in plan.new_comments:
+ client.create_comment(pr.number, comment)
if plan.marker_comment is not None:
if pr.marker_comment_id is not None:
client.update_comment(pr.number, pr.marker_comment_id,
plan.marker_comment)
@@ -1356,8 +1651,6 @@ def apply_plan(client: GitHubClient, pr:
PullRequestSnapshot, plan: Plan) -> Dra
client.set_status(pr.number, pr.head_sha, *plan.status)
if plan.ci_passed is not None:
client.set_status(pr.number, pr.head_sha, *plan.ci_passed,
context=CI_PASSED_CONTEXT)
- for comment in plan.new_comments:
- client.create_comment(pr.number, comment)
if plan.close:
client.close(pr.number)
# Last, because the label is what brings the pull request back to the next
run: when an
@@ -1496,11 +1789,25 @@ def run_reconcile(
)
]
observations = fetch_fork_observations(client, prs)
+
+ def is_gated_base(base_ref: str) -> bool:
+ # A base branch whose configuration cannot be read is treated as gated.
+ try:
+ return configs.get(base_ref) is not None
+ except (GitHubError, ValueError):
+ return True
+
bypass_finder = BypassFinder(
- client, repo=repo, now=now,
base_triggers=load_workflow_triggers(workflows_dir)
+ client,
+ repo=repo,
+ now=now,
+ base_triggers=load_workflow_triggers(workflows_dir),
+ is_gated_base=is_gated_base,
)
job_reporter = JobStatusReporter(client, repo=repo)
+ shared_heads = find_heads_shared_across_bases(prs)
+ gate_change_finder = GateChangeFinder(client, repo=repo)
for listed in prs:
try:
pr = listed if listed.comments_cursor is None else
find_state_comment(client, repo, listed)
@@ -1533,6 +1840,8 @@ def run_reconcile(
),
count_behind=functools.partial(count_commits_behind, client,
repo, pr),
jobs_reported=jobs_reported,
+ head_shared_across_bases=pr.head_sha in shared_heads,
+ find_gate_change=functools.partial(gate_change_finder.find,
pr),
)
draft_change = apply_plan(client, pr, plan)
if draft_change is not None:
@@ -1545,10 +1854,13 @@ def run_reconcile(
client.delete_run(pr.number, run.id)
unknown.extend((pr.number, reason) for reason in plan.unknown)
attention.extend((pr.number, item) for item in plan.attention)
+ except HeadMovedError as e:
+ console_print(f"[info]#{listed.number}: {e}; checked again on the
next run[/]")
except Exception as e:
console_print(f"[error]#{listed.number}: {e}[/]")
unknown.append((listed.number, f"{type(e).__name__}: {e}"))
+ attention.extend(bypass_finder.ambiguous)
notes = [bypass_finder.incomplete] if bypass_finder.incomplete else []
if job_reporter.deferred:
console_print(
diff --git a/dev/breeze/tests/test_fork_ci_messages.py
b/dev/breeze/tests/test_fork_ci_messages.py
index 0c238cff949..f7c205c0973 100644
--- a/dev/breeze/tests/test_fork_ci_messages.py
+++ b/dev/breeze/tests/test_fork_ci_messages.py
@@ -34,6 +34,7 @@ from airflow_breeze.utils.fork_ci_messages import (
MAX_LISTED_CI_FILES,
Marker,
ci_change_approval_comment,
+ gate_modified_close_comment,
inactivity_close_comment,
job_status_context,
nudge_comment,
@@ -42,6 +43,7 @@ from airflow_breeze.utils.fork_ci_messages import (
render_marker,
state_comment,
tamper_close_comment,
+ ungated_comment,
workflow_disabled_comment,
)
@@ -54,6 +56,7 @@ MARKER = Marker(state="awaiting_push", sha="abc123",
since=datetime(2026, 10, 15
pytest.param(MARKER, id="without-behind"),
pytest.param(replace(MARKER, behind=12), id="behind"),
pytest.param(replace(MARKER, jobs="123:SUCCESS"), id="jobs"),
+ pytest.param(replace(MARKER, gate_checked=True), id="gate-checked"),
],
)
def test_marker_round_trip(marker):
@@ -118,6 +121,8 @@ def test_parse_marker_malformed(body):
pytest.param(
workflow_disabled_comment(fork="u/airflow",
workflow_state="disabled_manually"), id="disabled"
),
+ pytest.param(gate_modified_close_comment(change="it changes `x`"),
id="gate-modified"),
+ pytest.param(ungated_comment(MARKER, base_ref="v2-10-test"),
id="ungated"),
],
)
def test_every_comment_ends_with_footer(text):
diff --git a/dev/breeze/tests/test_fork_ci_reconcile.py
b/dev/breeze/tests/test_fork_ci_reconcile.py
index 38977df88b4..fc4294a1f83 100644
--- a/dev/breeze/tests/test_fork_ci_reconcile.py
+++ b/dev/breeze/tests/test_fork_ci_reconcile.py
@@ -44,6 +44,8 @@ from airflow_breeze.utils.fork_ci_reconcile import (
DraftChange,
ForkObservation,
ForkState,
+ GateChangeFinder,
+ HeadMovedError,
JobStatusReporter,
Plan,
PullRequestSnapshot,
@@ -57,8 +59,10 @@ from airflow_breeze.utils.fork_ci_reconcile import (
drop_reported,
fetch_fork_observations,
fetch_workflow_state,
+ find_heads_shared_across_bases,
find_state_comment,
is_gated_skip_run,
+ is_own_pilot_change,
jobs_show_project_ci,
load_workflow_triggers,
parse_pull_request,
@@ -96,11 +100,21 @@ def _pr(**overrides) -> PullRequestSnapshot:
return PullRequestSnapshot(**values)
-def _marker(state: str, *, days_ago: float = 0, sha: str = "abc123") -> Marker:
- return Marker(state=state, sha=sha, since=NOW - timedelta(days=days_ago))
+def _marker(state: str, *, days_ago: float = 0, sha: str = "abc123",
gate_checked: bool = False) -> Marker:
+ return Marker(state=state, sha=sha, since=NOW - timedelta(days=days_ago),
gate_checked=gate_checked)
-def _classify(pr, fork=None, *, bypass_run=None, changes_ci_files=False,
config=CONFIG, behind=None):
+def _classify(
+ pr,
+ fork=None,
+ *,
+ bypass_run=None,
+ changes_ci_files=False,
+ config=CONFIG,
+ behind=None,
+ head_shared_across_bases=False,
+ gate_change=None,
+):
find_bypass = mock.Mock(return_value=bypass_run)
changes = mock.Mock(return_value=changes_ci_files)
plan = classify(
@@ -112,6 +126,8 @@ def _classify(pr, fork=None, *, bypass_run=None,
changes_ci_files=False, config=
find_bypass_run=find_bypass,
changes_ci_files=changes,
count_behind=mock.Mock(return_value=behind),
+ head_shared_across_bases=head_shared_across_bases,
+ find_gate_change=lambda: gate_change,
)
return plan, find_bypass, changes
@@ -402,6 +418,28 @@ def
test_green_trust_is_decided_again_after_retargeting(status_created_at, lists
assert (plan.status == GREEN_STATUS) is reposts_status
+def test_green_is_not_trusted_for_a_head_shared_across_bases():
+ plan, _, changes = _classify(
+ _pr(current_status=GREEN_STATUS), GREEN, changes_ci_files=False,
head_shared_across_bases=True
+ )
+
+ assert plan.state == ForkState.GREEN_UNVERIFIED
+ assert plan.add_labels == ["needs project ci"]
+ assert len(plan.attention) == 1
+ changes.assert_not_called()
+
+
+def test_find_heads_shared_across_bases():
+ prs = [
+ _pr(number=1, head_sha="a", base_ref="main"),
+ _pr(number=2, head_sha="a", base_ref="v3-1-test"),
+ _pr(number=3, head_sha="b", base_ref="main"),
+ _pr(number=4, head_sha="b", base_ref="main"),
+ ]
+
+ assert find_heads_shared_across_bases(prs) == {"a"}
+
+
@pytest.mark.parametrize(
("files", "expected"),
[
@@ -557,7 +595,7 @@ def
test_unfinished_tampering_closure_is_retried(pr_overrides, expected_link):
def test_since_is_kept_while_state_is_unchanged():
- marker = Marker("setup_required", "abc123", NOW - timedelta(days=2))
+ marker = Marker("setup_required", "abc123", NOW - timedelta(days=2),
gate_checked=True)
plan, _, _ = _classify(_pr(marker=marker),
ForkObservation(workflow_state="disabled_fork"))
@@ -860,6 +898,8 @@ def _finder(
jobs=(),
compare_files=(".github/workflows/ci-amd.yml",),
total_count=None,
+ other_pulls=(),
+ is_gated_base=lambda base_ref: True,
):
def rest_get_pages(path):
event = path.split("event=", 1)[1].split("&", 1)[0]
@@ -873,11 +913,13 @@ def _finder(
if "/compare/" in path:
assert not paginate, "the compare API returns files only on its
first page"
return {"files": [{"filename": name} for name in compare_files]}
+ if "/pulls?head=" in path:
+ return [{"number": 42, "closed_at": None}, *other_pulls]
raise AssertionError(path)
client.rest_get_pages.side_effect = rest_get_pages
client.rest_get.side_effect = rest_get
- return BypassFinder(client, repo=REPO, now=NOW,
base_triggers=BASE_TRIGGERS)
+ return BypassFinder(client, repo=REPO, now=NOW,
base_triggers=BASE_TRIGGERS, is_gated_base=is_gated_base)
@pytest.fixture
@@ -980,7 +1022,7 @@ def test_finder_ignores_legitimate_runs(client, runs,
jobs, compare_files, pr_ov
],
)
def test_finder_catches_triggers_the_base_branch_does_not_have(client, run):
- finder = _finder(client, [run])
+ finder = _finder(client, [run], jobs=[_job("Tests", "success")])
assert finder.find(_pr()) ==
"https://github.com/apache/airflow/actions/runs/9";
@@ -1013,11 +1055,44 @@ def
test_finder_fails_closed_on_a_change_set_it_cannot_check(client):
assert finder.find(_pr()) ==
"https://github.com/apache/airflow/actions/runs/7";
+def test_finder_ignores_runs_made_while_the_base_had_no_gate(client):
+ finder = _finder(
+ client, [_run(7)], jobs=TAMPERED_JOBS, is_gated_base=lambda base_ref:
base_ref != "v2-10-test"
+ )
+ pr = _pr(base_changes=((NOW - timedelta(minutes=30), "v2-10-test",
"main"),))
+
+ assert finder.find(pr) is None
+
+
[email protected](
+ "other_pulls",
+ [
+ pytest.param([{"number": 43, "closed_at": None}], id="open"),
+ pytest.param(
+ [{"number": 43, "closed_at": (NOW -
timedelta(hours=2)).isoformat()}], id="closed-recently"
+ ),
+ ],
+)
+def test_finder_reports_a_run_another_pull_request_may_own(client,
other_pulls):
+ finder = _finder(client, [_run(7)], jobs=TAMPERED_JOBS,
other_pulls=other_pulls)
+
+ assert finder.find(_pr()) is None
+ assert finder.ambiguous == [
+ (
+ 42,
+ "project CI run https://github.com/apache/airflow/actions/runs/7
may have bypassed the gate, "
+ "but it could also belong to #43 from the same branch; not closed,
check it",
+ )
+ ]
+
+
def test_finder_catches_a_workflow_moved_away(client):
finder = _finder(client, [_run(7)], jobs=TAMPERED_JOBS)
client.rest_get.side_effect = lambda path, paginate=False, items_key=None:
(
TAMPERED_JOBS
if path.endswith("jobs?filter=latest&per_page=100")
+ else [{"number": 42, "closed_at": None}]
+ if "/pulls?head=" in path
else {"files": [{"filename": "docs/x.yml", "previous_filename":
".github/workflows/x.yml"}]}
)
@@ -1153,6 +1228,11 @@ def
test_parse_pull_request_reads_timeline_status_and_marker():
"createdAt": "2026-11-03T00:00:00Z",
"label": {"name": "use project ci"},
},
+ {
+ "__typename": "UnlabeledEvent",
+ "createdAt": "2026-11-03T12:00:00Z",
+ "label": {"name": "use fork ci"},
+ },
{
"__typename": "ConvertToDraftEvent",
"createdAt": "2026-11-04T00:00:00Z",
@@ -1168,7 +1248,12 @@ def
test_parse_pull_request_reads_timeline_status_and_marker():
"createdAt": "2026-11-06T00:00:00Z",
"actor": {"login": "someone"},
},
- {"__typename": "BaseRefChangedEvent", "createdAt":
"2026-11-07T00:00:00Z"},
+ {
+ "__typename": "BaseRefChangedEvent",
+ "createdAt": "2026-11-07T00:00:00Z",
+ "previousRefName": "v3-1-test",
+ "currentRefName": "main",
+ },
]
},
commits={
@@ -1209,6 +1294,7 @@ def
test_parse_pull_request_reads_timeline_status_and_marker():
assert pr.last_reopened_at == datetime(2026, 11, 5, tzinfo=UTC)
assert pr.last_reopened_by_anyone_at == datetime(2026, 11, 6, tzinfo=UTC)
assert pr.last_base_change_at == datetime(2026, 11, 7, tzinfo=UTC)
+ assert pr.base_changes == ((datetime(2026, 11, 7, tzinfo=UTC),
"v3-1-test", "main"),)
assert pr.current_status == ("pending", "u", "d")
assert pr.status_created_at == datetime(2026, 11, 8, tzinfo=UTC)
assert pr.ci_passed_state == "success"
@@ -1510,6 +1596,10 @@ def
test_run_reconcile_isolates_failures_and_reports_them(client, tmp_path):
if len(workflow_state_calls) == 1:
raise GitHubError("gh api failed: HTTP 500")
return {"state": "disabled_fork"}
+ if path.endswith("/files?per_page=100"):
+ return []
+ if path.startswith(f"repos/{REPO}/pulls/"):
+ return {"head": {"sha": f"sha{path.rsplit('/', 1)[1]}"}}
raise AssertionError(path)
client.rest_get.side_effect = rest_get
@@ -1556,6 +1646,10 @@ def
test_run_reconcile_rereads_workflow_state_while_awaiting_push(client, tmp_pa
def rest_get(path, paginate=False, items_key=None):
if path == "repos/someone/my-airflow/actions/workflows/ci-amd.yml":
return {"state": "disabled_manually"}
+ if path.endswith("/files?per_page=100"):
+ return []
+ if path.startswith(f"repos/{REPO}/pulls/"):
+ return {"head": {"sha": "sha1"}}
raise AssertionError(path)
client.rest_get.side_effect = rest_get
@@ -2145,3 +2239,371 @@ def
test_job_status_reporter_compares_contexts_case_insensitively(client):
assert JobStatusReporter(client, repo=REPO).report(_pr(), COMPLETED_RUN)
== "1:1:FAILURE"
client.set_status.assert_not_called()
+
+
[email protected](
+ ("marker", "workflow_state", "expected_text"),
+ [
+ pytest.param(_marker("setup_required"), "active", "Push this branch
again", id="nudge"),
+ pytest.param(
+ _marker("awaiting_push"), "disabled_manually", "Fork CI stopped
running", id="workflow-disabled"
+ ),
+ ],
+)
+def test_new_comment_is_posted_before_the_state_comment_records_the_state(
+ client, marker, workflow_state, expected_text
+):
+ pr = _pr(marker=marker, marker_comment_id=5)
+ plan, _, _ = _classify(pr, ForkObservation(workflow_state=workflow_state))
+ client.create_comment.side_effect = GitHubError("gh api failed: HTTP 500")
+
+ with pytest.raises(GitHubError):
+ apply_plan(client, pr, plan)
+
+ assert expected_text in client.create_comment.call_args.args[1]
+ client.update_comment.assert_not_called()
+
+
[email protected](
+ "jobs",
+ [
+ pytest.param([_job("Fork only"), _job("Also fork only")],
id="all-skipped"),
+ pytest.param([_job("Fork only", "cancelled", started=False)],
id="cancelled-before-a-runner-took-it"),
+ ],
+)
+def test_finder_ignores_an_added_workflow_whose_jobs_did_not_run(client, jobs):
+ run = _run(9, path=".github/workflows/fork-only.yml", minutes=1)
+ finder = _finder(client, [run], jobs=jobs)
+
+ assert finder.find(_pr()) is None
+
+
[email protected]("airflow_breeze.utils.fork_ci_reconcile.BypassFinder",
autospec=True)
+def test_run_reconcile_reports_ambiguous_runs(mock_finder, client, tmp_path):
+ output = tmp_path / "slack-message.json"
+ client.graphql.side_effect = [_search_result(), _search_result(),
_search_result()]
+ mock_finder.return_value.incomplete = None
+ mock_finder.return_value.ambiguous = [(42, "could also belong to #43")]
+
+ has_alerts = run_reconcile(
+ client, CONFIG, now=NOW, run_url="https://run";, output_file=output,
workflows_dir=tmp_path
+ )
+
+ assert has_alerts is True
+ assert "|#42>: could also belong to #43" in output.read_text()
+
+
[email protected](
+ ("base_ref", "rest_get", "expected"),
+ [
+ pytest.param("main", None, True, id="default-branch"),
+ pytest.param("v2-10-test", _contents({}), False, id="no-gate"),
+ pytest.param("v3-1-test", GitHubError("gh api failed: HTTP 500"),
True, id="unreadable"),
+ pytest.param(
+ "v3-1-test", _contents({(GATE_FILES[1], "v3-1-test"): "#
gate\n"}), True, id="half-gate"
+ ),
+ ],
+)
[email protected]("airflow_breeze.utils.fork_ci_reconcile.BypassFinder",
autospec=True)
+def test_run_reconcile_tells_the_finder_which_bases_are_gated(
+ mock_finder, client, tmp_path, base_ref, rest_get, expected
+):
+ client.graphql.side_effect = [_search_result(), _search_result(),
_search_result()]
+ mock_finder.return_value.incomplete = None
+ mock_finder.return_value.ambiguous = []
+ run_reconcile(
+ client,
+ CONFIG,
+ now=NOW,
+ run_url="https://run";,
+ output_file=tmp_path / "s.json",
+ workflows_dir=tmp_path,
+ )
+ client.rest_get.side_effect = rest_get
+
+ assert mock_finder.call_args.kwargs["is_gated_base"](base_ref) is expected
+
+
+def test_a_change_to_the_gate_closes():
+ pr = _pr(is_draft=False, labels=("fork ci required", "ready for maintainer
review"))
+
+ plan, _, _ = _classify(pr, GREEN, gate_change="it changes `x`")
+
+ assert plan.state == ForkState.GATE_MODIFIED
+ assert plan.close is True
+ assert plan.draft is True
+ assert plan.remove_labels == ["ready for maintainer review"]
+ assert plan.status == (
+ "error",
+ "https://github.com/apache/airflow/pull/42/files";,
+ "Pull request changes the fork CI gate - see the comment",
+ )
+ assert "changes the fork CI gate: it changes `x`" in plan.new_comments[0]
+ assert '"state": "gate_modified"' in plan.marker_comment
+ assert len(plan.attention) == 1
+
+
[email protected](
+ ("pr_overrides", "checked"),
+ [
+ pytest.param({}, True, id="new-pull-request"),
+ pytest.param({"marker": _marker("running", sha="older")}, True,
id="new-head-commit"),
+ pytest.param(
+ {"marker": _marker("running", gate_checked=True)}, False,
id="head-commit-already-checked"
+ ),
+ pytest.param({"marker": _marker("running")}, True,
id="marker-from-before-the-check"),
+ pytest.param(
+ {"marker": _marker("gate_modified", days_ago=1),
"last_reopened_at": NOW},
+ True,
+ id="reopened-by-maintainer",
+ ),
+ pytest.param(
+ {"author_association": "MEMBER", "labels": ("fork ci required",
"use fork ci")},
+ False,
+ id="committer-on-fork-ci",
+ ),
+ pytest.param({"marker": _marker("released")}, True,
id="released-without-a-check"),
+ pytest.param({"marker": _marker("ungated")}, True,
id="ungated-without-a-check"),
+ pytest.param(
+ {
+ "marker": _marker("running", days_ago=1, gate_checked=True),
+ "last_base_change_at": NOW - timedelta(hours=1),
+ },
+ True,
+ id="retargeted-since",
+ ),
+ ],
+)
+def test_gate_change_is_checked_once_per_head_commit(pr_overrides, checked):
+ find = mock.Mock(return_value=None)
+ classify(
+ _pr(**pr_overrides),
+ GREEN,
+ config=CONFIG,
+ now=NOW,
+ repo=REPO,
+ find_bypass_run=lambda: None,
+ changes_ci_files=lambda: False,
+ find_gate_change=find,
+ )
+
+ assert find.called is checked
+
+
+def test_unfinished_gate_change_closure_is_retried():
+ plan, _, _ = _classify(_pr(marker=_marker("gate_modified", days_ago=1)),
GREEN, gate_change=None)
+
+ assert plan.state == ForkState.GATE_MODIFIED
+ assert plan.close is True
+ assert "Closing this pull request because it changes the fork CI gate." in
plan.new_comments[0]
+
+
+PILOT_BASE = "enabled = true\n[pilot]\nother = 2026-10-01T00:00:00Z\n"
+
+
[email protected](
+ ("head_text", "expected"),
+ [
+ pytest.param(
+ "enabled = true\n[pilot]\nother = 2026-10-01T00:00:00Z\nSomeone =
2026-12-01T00:00:00Z\n",
+ True,
+ id="adds-own-entry",
+ ),
+ pytest.param("enabled = true\n", False, id="removes-another-entry"),
+ pytest.param(
+ "enabled = false\n[pilot]\nother = 2026-10-01T00:00:00Z\nsomeone =
2026-12-01T00:00:00Z\n",
+ False,
+ id="also-switches-the-gate-off",
+ ),
+ pytest.param("enabled = [", False, id="unparsable"),
+ pytest.param("pilot = []\n", False, id="wrong-shape"),
+ pytest.param(None, False, id="deleted"),
+ ],
+)
+def test_is_own_pilot_change(head_text, expected):
+ assert is_own_pilot_change(PILOT_BASE, head_text, "someone") is expected
+
+
+CI_AMD = """
+on: [push, pull_request]
+jobs:
+ gate:
+ steps: [{run: gate}]
+ build-info:
+ needs: [gate]
+ if: needs.gate.outputs.project-ci == 'true'
+ tests:
+ needs: [build-info]
+ steps: [{run: tests}]
+"""
+
+
+GATE_JOB_CHANGE = (
+ "it changes the `gate` job of `.github/workflows/ci-amd.yml`, which jobs
depend on it, "
+ "or the workflow-level triggers or settings it runs with"
+)
+
+
+def _gate_change_finder(client, changed, head_files, base_files=None,
head_sha="abc123"):
+ base_files = {
+ ".github/workflows/ci-amd.yml": CI_AMD,
+ ".github/fork-ci-gate.toml": PILOT_BASE,
+ **(base_files or {}),
+ }
+
+ def rest_get(path, paginate=False, items_key=None):
+ if path.endswith("/files?per_page=100"):
+ return changed
+ if "/compare/" in path:
+ return {"merge_base_commit": {"sha": "merge-base"}}
+ if path == f"repos/{REPO}/pulls/42":
+ return {"head": {"sha": head_sha}}
+ repo_name, rest = path.removeprefix("repos/").split("/contents/")
+ name, ref = rest.split("?ref=")
+ files = base_files if (repo_name, ref) == (REPO, "merge-base") else
head_files
+ if name not in files:
+ raise GitHubError("gh api failed: HTTP 404: Not Found")
+ return {"content": base64.b64encode(files[name].encode()).decode()}
+
+ client.rest_get.side_effect = rest_get
+ return GateChangeFinder(client, repo=REPO)
+
+
[email protected](
+ ("changed", "head_files", "expected"),
+ [
+ pytest.param(
+ [{"filename":
"dev/breeze/src/airflow_breeze/utils/fork_ci_gate.py"}],
+ {},
+ "it changes `dev/breeze/src/airflow_breeze/utils/fork_ci_gate.py`",
+ id="gate-code",
+ ),
+ pytest.param(
+ [{"filename": "docs/x.yml", "previous_filename":
".github/workflows/fork-ci-monitor.yml"}],
+ {},
+ "it changes `.github/workflows/fork-ci-monitor.yml`",
+ id="monitor-moved-away",
+ ),
+ pytest.param(
+ [{"filename": ".github/fork-ci-gate.toml"}],
+ {".github/fork-ci-gate.toml": "enabled = false\n"},
+ "it changes `.github/fork-ci-gate.toml` beyond the author's own
`[pilot]` entry",
+ id="gate-config",
+ ),
+ pytest.param(
+ [{"filename": ".github/workflows/ci-amd.yml"}],
+ {".github/workflows/ci-amd.yml": CI_AMD.replace("[gate]", "[]")},
+ GATE_JOB_CHANGE,
+ id="gate-unwired",
+ ),
+ pytest.param(
+ [{"filename": ".github/fork-ci-gate.toml"}, {"filename":
".github/workflows/ci-amd.yml"}],
+ {
+ ".github/fork-ci-gate.toml": (
+ "enabled = true\n[pilot]\nother =
2026-10-01T00:00:00Z\nsomeone = 2026-12-01T00:00:00Z\n"
+ ),
+ ".github/workflows/ci-amd.yml": CI_AMD.replace("run: tests",
"run: more-tests"),
+ },
+ None,
+ id="own-pilot-entry-and-other-ci-change",
+ ),
+ pytest.param(
+ [{"filename": ".github/workflows/ci-amd.yml"}],
+ {".github/workflows/ci-amd.yml": "jobs: [broken"},
+ GATE_JOB_CHANGE,
+ id="workflow-unparsable",
+ ),
+ pytest.param(
+ [{"filename": ".github/workflows/ci-amd.yml"}],
+ {".github/workflows/ci-amd.yml": "jobs: [one, two]\n"},
+ GATE_JOB_CHANGE,
+ id="jobs-not-a-mapping",
+ ),
+ pytest.param(
+ [{"filename": ".github/workflows/ci-amd.yml"}],
+ {".github/workflows/ci-amd.yml": "defaults:\n run:\n
working-directory: /tmp\n" + CI_AMD},
+ GATE_JOB_CHANGE,
+ id="workflow-defaults",
+ ),
+ pytest.param(
+ [{"filename": ".github/workflows/ci-amd.yml"}],
+ {".github/workflows/ci-amd.yml": CI_AMD.replace("needs:
[build-info]", "needs: []")},
+ GATE_JOB_CHANGE,
+ id="job-freed-from-the-gate",
+ ),
+ pytest.param(
+ [{"filename": ".github/workflows/ci-amd.yml"}],
+ {
+ ".github/workflows/ci-amd.yml": CI_AMD.replace(
+ "[push, pull_request]", "[push, pull_request,
issue_comment]"
+ )
+ },
+ GATE_JOB_CHANGE,
+ id="trigger-added",
+ ),
+ pytest.param(
+ [{"filename": f"airflow-core/f{i}.py"} for i in range(3000)],
+ {},
+ "it changes 3000 or more files, more than GitHub lists, so the
gate cannot be checked",
+ id="listing-at-cap",
+ ),
+ ],
+)
+def test_gate_change_finder(client, changed, head_files, expected):
+ finder = _gate_change_finder(client, changed, head_files)
+
+ assert finder.find(_pr()) == expected
+
+
[email protected](
+ ("pr_overrides", "gate_checked"),
+ [
+ pytest.param({}, True, id="checked"),
+ pytest.param(
+ {"author_association": "MEMBER", "labels": ("fork ci required",
"use fork ci")},
+ False,
+ id="exempt-with-use-fork-ci",
+ ),
+ ],
+)
+def test_marker_records_whether_the_gate_was_checked(pr_overrides,
gate_checked):
+ plan, _, _ = _classify(_pr(**pr_overrides),
ForkObservation(suite_status="IN_PROGRESS", run_url=RUN_URL))
+
+ assert ('"gate_checked": true' in plan.marker_comment) is gate_checked
+
+
+def test_gate_change_finder_aborts_when_the_head_moved(client):
+ finder = _gate_change_finder(client, [], {}, head_sha="newer")
+
+ with pytest.raises(HeadMovedError, match="head moved from abc123 to
newer"):
+ finder.find(_pr())
+
+
+def test_run_reconcile_retries_a_pull_request_whose_head_moved(client,
tmp_path):
+ output = tmp_path / "slack-message.json"
+ client.graphql.side_effect = [
+ _search_result(),
+ _search_result(),
+ _search_result(_pr_node(1)),
+ {"pr0": {"object": None}},
+ ]
+ client.rest_get_pages.return_value = [{"total_count": 0, "workflow_runs":
[]}]
+
+ def rest_get(path, paginate=False, items_key=None):
+ if path.endswith("/actions/workflows/ci-amd.yml"):
+ return {"state": "active"}
+ if path == f"repos/{REPO}/pulls/1/files?per_page=100":
+ return []
+ if path == f"repos/{REPO}/pulls/1":
+ return {"head": {"sha": "moved"}}
+ raise AssertionError(path)
+
+ client.rest_get.side_effect = rest_get
+
+ has_alerts = run_reconcile(
+ client, CONFIG, now=NOW, run_url="https://run";, output_file=output,
workflows_dir=tmp_path
+ )
+
+ assert has_alerts is False
+ client.set_status.assert_not_called()
+ client.create_comment.assert_not_called()
diff --git a/scripts/ci/prek/check_ci_workflows_in_sync.py
b/scripts/ci/prek/check_ci_workflows_in_sync.py
index 047c3f0dd7c..7043131d242 100755
--- a/scripts/ci/prek/check_ci_workflows_in_sync.py
+++ b/scripts/ci/prek/check_ci_workflows_in_sync.py
@@ -161,7 +161,9 @@ AMD_ONLY_BLOCK = """ schedule:
AMD_ONLY_GATE_BLOCK = """ gate:
# Fork CI gate (AIP-120): decides whether project CI runs for this event.
Configuration and
# code come from the base branch so a pull request cannot change its own
eligibility. See
- # dev/breeze/doc/ci/08_fork_ci_for_external_contributors.md.
+ # dev/breeze/doc/ci/08_fork_ci_for_external_contributors.md. A pull
request whose CI runs
+ # in the contributor's fork and changes this job, or how build-info
depends on it, is closed
+ # by the fork CI monitor ("Gate protection" in that document).
name: "Fork CI gate"
runs-on: ["ubuntu-slim"]
timeout-minutes: 5
