This is an automated email from the ASF dual-hosted git repository.
vatsrahul1001 pushed a commit to branch v3-4-test
in repository https://gitbox.apache.org/repos/asf/airflow.git
commit e4e26aa7e6c2c022cc5a29f19a072884e4522f88
Author: github-actions[bot]
<41898282+github-actions[bot]@users.noreply.github.com>
AuthorDate: Sat Oct 31 17:20:36 2026 +0100
[v3-4-test] Show fork CI progress sooner and gate rebased pilot pull
requests (#76148) (#76149)
Feedback from the first day of the contributor pilot:
Pilot participants expected their open pull requests to move to fork CI
once they rebased them after joining, but only pull requests created
after the join instant were gated.
A push soon after opening a pull request cancels the first project CI
run, and its cancelled checks stayed next to the green fork results,
giving the pull request a confusing red/green status.
Fork CI reports to the pull request only when the scheduled monitor
runs, so for a while a contributor saw nothing telling them their CI
runs in the fork. The gate cannot post a status itself (it runs with a
read-only token on the contributor's event), so a required "CI run
started" check now shows as expected and waiting from the moment of the
push, and the monitor fills it in with a link to the run. Doing that in
the scheduled monitor, rather than in a job of every CI run, keeps it
off the shared runner queue, and its writes come out of the monitor's
existing per-run budget so the repository's shared rate limit is not
spent any faster.
One full-matrix fork run needs more job statuses than that budget
allows in a single monitor run, so the Slack note about deferred job
statuses fired on ordinary pilot runs. It now waits until deferring
persists, which is what it is meant to signal.
(cherry picked from commit a3567e948eb7f3730433e47c730b5b7f131a75ec)
Generated-by: Claude Opus 5
Co-authored-by: Jarek Potiuk <[email protected]>
---
.asf.yaml | 10 +-
.github/workflows/ci-amd.yml | 4 +
contributing-docs/05_pull_requests.rst | 3 +-
contributing-docs/18_contribution_workflow.rst | 5 +-
dev/AGENTS.md | 3 +-
.../doc/ci/08_fork_ci_for_external_contributors.md | 81 +++-
.../src/airflow_breeze/utils/fork_ci_gate.py | 53 ++-
.../src/airflow_breeze/utils/fork_ci_messages.py | 9 +-
.../src/airflow_breeze/utils/fork_ci_reconcile.py | 342 +++++++++++++-
.../airflow_breeze/utils/workflow_run_approval.py | 42 +-
dev/breeze/tests/test_fork_ci_gate.py | 110 +++++
dev/breeze/tests/test_fork_ci_reconcile.py | 515 ++++++++++++++++++++-
dev/breeze/tests/test_workflow_run_approval.py | 62 +++
scripts/ci/prek/check_ci_workflows_in_sync.py | 4 +
14 files changed, 1186 insertions(+), 57 deletions(-)
diff --git a/.asf.yaml b/.asf.yaml
index 8c62e4183ba..af7067281b2 100644
--- a/.asf.yaml
+++ b/.asf.yaml
@@ -136,9 +136,13 @@ github:
strict_required_status_checks_policy: false
required_status_checks:
- context: "CI passed"
+ # Reported by the fork CI monitor once the head commit has a CI
run, in the
+ # contributor's fork for a gated pull request or in
apache/airflow otherwise; until
+ # then GitHub lists it on the pull request as expected and
waiting.
+ - context: "CI run started"
# The branches `ci-amd.yml` runs pull request CI for, apart from `main`. A
release branch gets
# the "CI passed" job only when it is backported there; until then its PRs
need a committer to
- # bypass the rule.
+ # bypass the rule. "CI run started" comes from the fork CI monitor on
`main`, for every base.
- name: "Require CI on release branches"
target: branch
enforcement: active
@@ -165,6 +169,10 @@ github:
strict_required_status_checks_policy: false
required_status_checks:
- context: "CI passed"
+ # Reported by the fork CI monitor once the head commit has a CI
run, in the
+ # contributor's fork for a gated pull request or in
apache/airflow otherwise; until
+ # then GitHub lists it on the pull request as expected and
waiting.
+ - context: "CI run started"
- name: "Branch protection"
type: branch
branches:
diff --git a/.github/workflows/ci-amd.yml b/.github/workflows/ci-amd.yml
index 31f81ea7df2..6acd387f3f8 100644
--- a/.github/workflows/ci-amd.yml
+++ b/.github/workflows/ci-amd.yml
@@ -171,6 +171,10 @@ jobs:
name: "Fork CI gate"
runs-on: ["ubuntu-slim"]
timeout-minutes: 5
+ permissions:
+ contents: read
+ # Reads the creation time of this run, which decides for a pilot's older
pull request.
+ actions: read
outputs:
project-ci: ${{ steps.gate.outputs.project-ci }}
reason: ${{ steps.gate.outputs.reason }}
diff --git a/contributing-docs/05_pull_requests.rst
b/contributing-docs/05_pull_requests.rst
index 0904333ed0e..2a4aaf1a04a 100644
--- a/contributing-docs/05_pull_requests.rst
+++ b/contributing-docs/05_pull_requests.rst
@@ -269,7 +269,8 @@ and approves them automatically when all of these hold:
at them first.
If your PR's CI runs in your fork (it carries the ``fork ci required`` label),
the CI run in
-``apache/airflow`` only checks that, so none of the rules above apply. Its
runs are approved
+``apache/airflow`` only checks that, so none of the rules above apply. The
``CI run started`` check
+waits until the fork CI monitor sees your fork's run, usually within 15
minutes, and then links it. Its runs are approved
automatically unless the PR changes files that define CI: anything under
``.github/``,
``dev/breeze/``, ``scripts/ci/``, ``scripts/docker/`` or
``scripts/in_container/``, a top-level
``Dockerfile*``, or a ``.pre-commit-config.yaml`` in any directory. Then the
runs wait for a maintainer and a comment on the PR explains
diff --git a/contributing-docs/18_contribution_workflow.rst
b/contributing-docs/18_contribution_workflow.rst
index 8b647dd16f1..d49d732c5d2 100644
--- a/contributing-docs/18_contribution_workflow.rst
+++ b/contributing-docs/18_contribution_workflow.rst
@@ -185,8 +185,9 @@ at 00:00 UTC:
[pilot]
"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. Your own
``[pilot]`` line is
+Every pull request you open after that instant uses fork CI. Your earlier pull
requests keep project
+CI until you push to them, rebase them or close and reopen them after that
instant, and use fork CI
+from then on. 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
diff --git a/dev/AGENTS.md b/dev/AGENTS.md
index 313c7860183..24267879fdb 100644
--- a/dev/AGENTS.md
+++ b/dev/AGENTS.md
@@ -54,7 +54,8 @@ workflow of `apache/airflow` shares (see "Write budget" in
If the budget is exceeded, turn the job statuses off: set `job_statuses =
false` in
`.github/fork-ci-gate.toml`. Signs that it is exceeded are the monitor's Slack
note that job
-statuses "did not fit in the budget" persisting across runs, rate-limit
retries ("GitHub rate limit,
+statuses "have not fit in the budget" (posted only after three runs in a row
deferred them)
+staying for hours, rate-limit retries ("GitHub rate limit,
retrying") or "still rate limited" errors in the monitor's run log or its
Slack messages, or other
workflows of the repository failing on secondary rate limits. The aggregate
`Fork CI / Tests (AMD)`
status and `CI passed` are posted either way, so turning job statuses off
loses detail, not gating.
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 c861467f958..122081de163 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
@@ -174,7 +174,7 @@ concurrency cap. A manual `workflow_dispatch` in a fork
still runs.
| Pull request carries `use project ci` | run | `label_project_ci` |
| Pull request head repository is `apache/airflow` | run | `head_is_upstream` |
| Pull request carries `use fork ci` | skip | `label_fork_ci` |
-| Author login is in `[pilot]` and the pull request was created at or after
that login's join instant | skip | `pilot` |
+| Author login is in `[pilot]` and the pull request was created, or its
`ci-amd.yml` run was created, at or after that login's join instant | skip |
`pilot` |
| `enabled` is false | run | `gate_disabled` |
| Pull request created before `enabled_since` | run | `before_enabled_since` |
| Author association is `MEMBER`, `OWNER` or `COLLABORATOR` (see below for
private membership) | run | `trusted_author` |
@@ -248,7 +248,21 @@ login and a join instant to `[pilot]`; any committer can
merge it. Once the entr
base branch, every pull request they open is gated from its first run, whether
or not
`enabled` is set, and goes through exactly the flow described in this
document: setup
comment, draft, fork run, commit status, undraft, close after inactivity, gate
tampering
-check. Pull requests they opened before the join instant keep project CI.
+check. Pull requests they opened before the join instant keep project CI until
their first
+`ci-amd.yml` run created at or after it -- the run of a push, a rebase, a
reopen or marking the
+pull request ready for review -- and are gated from then on. A re-run of an
earlier run keeps
+that run's creation time, so it does not switch the pull request.
+
+The gate reads the creation time of its own run (`GET
/repos/apache/airflow/actions/runs/<id>`,
+hence `actions: read` on the gate job) and fails, to be re-run, when it
cannot; the event's
+`updated_at` would not do, because discovery cannot read it back for the same
decision.
+Discovery reads the same time back for such a pull request: the latest
`ci-amd.yml`
+`pull_request` run of its head commit created at or after the join instant,
one REST call per
+pull request, made only for a `[pilot]` author's pull request created before
they joined. Once
+discovery labelled it, the reconciler takes the `fork ci required` label as
the evidence instead,
+because it deletes the gated run that discovery saw. The gate tampering check
applies to the pull
+request from then on and, when project CI was allowed for it before the join
instant (not, for
+example, when it was already gated as `external`), leaves out its runs from
before that instant.
The gate reads the table from the base branch, so the entry takes effect when
it merges,
while the join instant is compared with pull request creation. The join
instant must
@@ -353,8 +367,10 @@ gated run that never ran.
pull request's author, association, creation time, head repository and current
labels, the inputs
the gate job of the waiting run uses. When the reason is gated (`external`,
`pilot` or
`label_fork_ci`), the rules for pull requests that run project CI (estimated
job-minutes, lines of
-source code, dependency files, full tests) do not apply, because none of those
jobs run. One rule
-applies instead:
+source code, dependency files, full tests) do not apply, because none of those
jobs run. A pilot's
+pull request created before they joined counts as gated from the join instant:
its waiting runs
+created after it are approved this way, while earlier ones, whose gate ran
project CI, are left to a
+maintainer. One rule applies instead:
- The waiting runs of the head commit are approved when the pull request's
merge commit changes no
file that the reconciler counts as CI-defining: `.github/`, `dev/breeze/`,
`scripts/ci/`,
@@ -599,9 +615,9 @@ in `apache/airflow`.
The check applies only to pull requests whose `fork_ci_gate.decide` reason is
`external` or
`pilot`, recomputed from the pass 2 fields. A pilot participant's pull request
is gated from
-its creation, so it never ran project CI legitimately unless `use project ci`
was on it,
-which is accounted for below, or it was opened before the `[pilot]` entry
merged
-(see "Contributor pilot"). Committers piloting with `use fork ci` are excluded:
+its creation, or from its first run after the join instant, so it never ran
project CI
+legitimately since unless `use project ci` was on it, which is accounted for
below, or it was
+opened before the `[pilot]` entry merged (see "Contributor pilot"). Committers
piloting with `use fork ci` are excluded:
their pull request legitimately ran project CI before the label was added.
The check works from the runs `apache/airflow` recorded, not from the head
commit. Once per
@@ -798,9 +814,13 @@ suites); its jobs come from `GET
/repos/<fork>/actions/runs/<id>/jobs?filter=lat
why check runs do not fit), so each one is a REST call, and they count
towards GitHub's limits on
content-creating requests: a reconciler run posts at most
`MAX_JOB_STATUSES_PER_RUN` (60) of
them, each write charged as it is made. What does not fit is posted by the
next run, and once
- the budget is spent no further pull request's jobs or statuses are read in
that run. A run that
- defers job statuses adds a note to the Slack message, with fixed wording so
it is posted once
- while deferrals persist; the run log lists the pull requests.
+ the budget is spent no further pull request's jobs or statuses are read in
that run. One fork run
+ of the full matrix (about 130 jobs) needs more than a run's budget on its
own, which the next
+ run catches up on, so deferring alone is not a problem: the run log lists
the deferred pull
+ requests, and only when three runs in a row deferred, about 45 minutes of
the budget not
+ keeping up, is a note added to the Slack message. The count is kept in the
record the monitor
+ carries between runs (`--reported-file`); without a readable record the note
is added at once.
+ Its wording is fixed, so it is posted once while deferrals persist.
- Commit statuses cannot be deleted and have no neutral state. When a run has
completed, a job
context posted for an earlier run of the same commit that the completed run
does not have, or
skipped, is overwritten with `success` and the description "Not in the
latest fork run - see
@@ -851,7 +871,10 @@ requests a minute and 500 an hour. Assuming about 35
reported jobs for an averag
limit and more at peaks. The cap of 60 writes a run, four runs an hour,
holds job statuses to at
most 240 an hour: 48% of the content-creating limit and 24% of the REST
limit, plus two reads per
pull request with a run in progress. Job statuses then lag behind at busy
times, the aggregate
- status does not.
+ status does not. The `CI run started` statuses of pull requests whose CI
runs in
+ `apache/airflow` come out of the same 60 writes a run, first and at most 30
of them (see "The
+ `CI run started` check"), so they do not raise the total; at about 11 pull
request runs an hour
+ they take about 3 writes a run.
These are estimates: the share of gated runs, the jobs per push and the pushes
per pull request
are assumptions, not measurements. Watch the monitor's call counts in its run
log after enabling
@@ -888,6 +911,35 @@ is the source the ruleset accepts.
job must not report success there; it reports under another name when
`build-info` is
skipped, and the reconciler's status is the only `CI passed` the pull
request has.
+### The `CI run started` check
+
+Fork CI reports to the pull request only when the scheduled monitor runs,
about every 15 minutes
+and later when GitHub delays scheduled workflows, and until then a gated pull
request shows only
+skipped checks. The gate cannot say more itself: it runs on the contributor's
`pull_request`
+event with a read-only token. Instead both CI rulesets in `.asf.yaml` also
require a
+`CI run started` check, which nothing reports when a commit is pushed, so
GitHub lists it on every
+pull request as expected and waiting for its status. The reconciler reports
it, as a commit status
+with state `success` linking the run:
+
+- for a gated pull request, once the fork has a run of `ci-amd.yml` for the
head commit, whatever
+ its state; the fork run is linked, so the contributor can follow it before
the state comment
+ appears.
+- for every other open pull request, once project CI has a `ci-amd.yml`
`pull_request` run for the
+ head commit that is not waiting for a maintainer's approval. One GraphQL
search lists the open
+ pull requests without `fork ci required`, most recently updated first, with
the head commit's
+ `CI run started` status and its GitHub Actions check suites. Pull requests
discovery labelled in
+ the same run are skipped, since the search index can still list them without
the label. When the
+ last commit listed is not the head, as after a rebase that keeps old commit
dates, two REST calls
+ read the head commit's statuses and latest run instead.
+
+Folding the check into the scheduled monitor costs no runner per push, at the
price of a pull
+request waiting for the next monitor run before it can merge, which a CI run
takes longer than
+anyway. Each new head commit needs one write, taken first out of the per-run
budget of job
+statuses (see "Job statuses"), since merging waits for it; at most 30 such
writes per run, so job
+statuses keep at least half of the budget, and after the check became required
the open pull
+requests are caught up a batch at a time. A release branch has no monitor of
its own, but `main`'s monitor covers pull
+requests to every base branch.
+
### Gated project CI runs are deleted
A gated run of `ci-amd.yml` lists every project CI job on the pull request as
a skipped check,
@@ -902,6 +954,13 @@ name within its time limit, and its run is the evidence. A
re-run is kept too: d
deletes every attempt, and an earlier attempt may have run project CI. Nothing
is deleted for a
pull request in the `gate_bypassed` state. A dry run only reports the
deletions.
+A push soon after opening a pull request cancels its first run, often while
the gate is still
+deciding, and the run that replaces it is the one left to fork CI. The
cancelled run lists its
+jobs as cancelled checks next to the fork's green result, so it is deleted
too, under the same
+conditions: completed, first attempt, its commit does not change
`.github/workflows/`, and no job
+did project CI work by the tampering check's rules. Of its jobs only the gate
and the jobs that
+also run in a gated run started, none failed, and at least one was cancelled.
+
### Far behind the base branch
A fork run tests the pushed commit as it is, while project CI tests it merged
with the base
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 33c5dd97543..13d59238efb 100644
--- a/dev/breeze/src/airflow_breeze/utils/fork_ci_gate.py
+++ b/dev/breeze/src/airflow_breeze/utils/fork_ci_gate.py
@@ -146,6 +146,18 @@ class GateConfig:
return joined_at
return None
+ def find_pilot_join_after(self, login: str, created_at: datetime | None)
-> datetime | None:
+ """
+ The pilot join instant of an author whose pull request was created
before it, else None.
+
+ Such a pull request is gated from its first CI run at or after that
instant, so deciding
+ needs the creation time of that run.
+ """
+ joined_at = self.pilot_joined_at(login)
+ if joined_at is None or created_at is None or created_at >= joined_at:
+ return None
+ return joined_at
+
def _require_aware_datetime(value: Any, name: str) -> datetime:
if not isinstance(value, datetime) or value.tzinfo is None:
@@ -206,6 +218,11 @@ class GateEvent:
author_association: str = ""
created_at: datetime | None = None
labels: tuple[str, ...] = ()
+ # When the ci-amd.yml run deciding for the pull request was created: a
pilot's pull request
+ # created before they joined is gated from its first run at or after
joining, such as the run
+ # of a push, a rebase or a reopen. The run's time, unlike the event's
`updated_at`, is what the
+ # reconciler can read back for the same decision.
+ ci_run_at: datetime | None = None
@dataclass(frozen=True)
@@ -218,6 +235,10 @@ class LabelsUnavailableError(RuntimeError):
"""The current labels of a pull request could not be read."""
+class RunTimeUnavailableError(RuntimeError):
+ """The creation time of the gate's own workflow run could not be read."""
+
+
def _is_canonical_branch(ref: str) -> bool:
branch = ref.removeprefix("refs/heads/")
return any(pattern.match(branch) for pattern in
CANONICAL_PUSH_BRANCH_PATTERNS)
@@ -244,7 +265,9 @@ def decide(event: GateEvent, config: GateConfig) ->
Decision:
if LABEL_USE_FORK_CI in event.labels:
return Decision(run=False, reason=Reason.LABEL_FORK_CI)
joined_at = config.pilot_joined_at(event.author_login)
- if joined_at is not None and event.created_at is not None and
event.created_at >= joined_at:
+ if joined_at is not None and any(
+ moment is not None and moment >= joined_at for moment in
(event.created_at, event.ci_run_at)
+ ):
return Decision(run=False, reason=Reason.PILOT)
if not config.enabled:
return Decision(run=True, reason=Reason.GATE_DISABLED)
@@ -290,6 +313,18 @@ def fetch_pull_request_labels(repository: str, number:
str) -> tuple[str, ...] |
return tuple(labels)
+def fetch_run_created_at(repository: str, run_id: str) -> datetime | None:
+ """The creation time of a workflow run, which stays the same across its
re-runs."""
+ output = _run_gh("api", f"repos/{repository}/actions/runs/{run_id}",
"--jq", ".created_at")
+ if output is None:
+ return None
+ try:
+ return parse_github_datetime(output.strip())
+ except ValueError:
+ print(f"Unexpected created_at of run {run_id}: {output!r}",
file=sys.stderr)
+ return None
+
+
def _write_line(env_var: str, line: str, env: Mapping[str, str]) -> None:
path = env.get(env_var)
if path:
@@ -311,6 +346,16 @@ def evaluate_from_env(env: Mapping[str, str]) -> Decision:
f"Could not read the labels of pull request
#{env.get('PR_NUMBER', '')}; re-run the job"
)
labels = fetched
+ config = load_config(Path(env.get("GATE_CONFIG",
".github/fork-ci-gate.toml")))
+ created_at = parse_github_datetime(env.get("PR_CREATED_AT", ""))
+ ci_run_at = None
+ if event_name == "pull_request" and
config.find_pilot_join_after(env.get("PR_AUTHOR", ""), created_at):
+ ci_run_at = fetch_run_created_at(repository, env.get("GITHUB_RUN_ID",
""))
+ if ci_run_at is None:
+ # Project CI would run for a pull request the reconciler may treat
as gated.
+ raise RunTimeUnavailableError(
+ f"Could not read the creation time of run
{env.get('GITHUB_RUN_ID', '')}; re-run the job"
+ )
event = GateEvent(
event_name=event_name,
repository=repository,
@@ -318,10 +363,10 @@ def evaluate_from_env(env: Mapping[str, str]) -> Decision:
head_repository=env.get("PR_HEAD_REPO", ""),
author_login=env.get("PR_AUTHOR", ""),
author_association=env.get("PR_AUTHOR_ASSOCIATION", ""),
- created_at=parse_github_datetime(env.get("PR_CREATED_AT", "")),
+ created_at=created_at,
labels=labels,
+ ci_run_at=ci_run_at,
)
- config = load_config(Path(env.get("GATE_CONFIG",
".github/fork-ci-gate.toml")))
return decide(event, config)
@@ -329,7 +374,7 @@ def main(env: Mapping[str, str] | None = None) -> int:
env = os.environ if env is None else env
try:
decision = evaluate_from_env(env)
- except LabelsUnavailableError as e:
+ except (LabelsUnavailableError, RunTimeUnavailableError) as e:
print(e, file=sys.stderr)
return 1
project_ci = "true" if decision.run else "false"
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 d19162d448f..0814021495b 100644
--- a/dev/breeze/src/airflow_breeze/utils/fork_ci_messages.py
+++ b/dev/breeze/src/airflow_breeze/utils/fork_ci_messages.py
@@ -35,7 +35,8 @@ from datetime import datetime
STATUS_CONTEXT = "Fork CI / Tests (AMD)"
# Each job of the fork run is reported under this prefix. Job names come from
the contributor's
-# workflow, so the prefix keeps them from posting STATUS_CONTEXT or
CI_PASSED_CONTEXT.
+# workflow, so the prefix keeps them from posting STATUS_CONTEXT,
CI_PASSED_CONTEXT or
+# CI_RUN_STARTED_CONTEXT.
JOB_STATUS_CONTEXT_PREFIX = "Fork CI job / "
JOB_NAME_MAX_LENGTH = 100
# The check the "Require CI on main" ruleset requires before a pull request
can (auto-)merge. Project
@@ -44,6 +45,12 @@ JOB_NAME_MAX_LENGTH = 100
CI_PASSED_CONTEXT = "CI passed"
CI_PASSED_DESCRIPTION = "Fork CI passed for this commit"
CI_PASSED_REVOKED_DESCRIPTION = "Fork CI result no longer trusted for this
commit"
+# Also required by the merge rulesets, so GitHub lists it as expected and
waiting on every pull
+# request from the moment it is pushed. The reconciler reports it, linking the
run: for a gated pull
+# request once the fork has a run for the head commit, for any other once
project CI has one.
+CI_RUN_STARTED_CONTEXT = "CI run started"
+CI_RUN_STARTED_DESCRIPTION = "CI runs in the contributor's fork - details open
the run"
+CI_RUN_STARTED_PROJECT_DESCRIPTION = "Project CI runs in apache/airflow -
details open the run"
# A fork run tests the head commit as it is, not merged with the base branch.
Further behind than
# this, it says little about the result after merge.
BEHIND_WARNING_COMMITS = 200
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 78e86abdff2..c5470bf1b22 100644
--- a/dev/breeze/src/airflow_breeze/utils/fork_ci_reconcile.py
+++ b/dev/breeze/src/airflow_breeze/utils/fork_ci_reconcile.py
@@ -31,9 +31,9 @@ import base64
import functools
import json
import re
-from collections.abc import Callable, Iterable, Sequence
+from collections.abc import Callable, Collection, Iterable, Sequence
from dataclasses import asdict, dataclass, field, replace
-from datetime import datetime, timedelta
+from datetime import UTC, datetime, timedelta
from enum import Enum
from pathlib import Path
from typing import Any
@@ -61,6 +61,9 @@ from airflow_breeze.utils.fork_ci_messages import (
CI_PASSED_CONTEXT,
CI_PASSED_DESCRIPTION,
CI_PASSED_REVOKED_DESCRIPTION,
+ CI_RUN_STARTED_CONTEXT,
+ CI_RUN_STARTED_DESCRIPTION,
+ CI_RUN_STARTED_PROJECT_DESCRIPTION,
JOB_STATUS_CONTEXT_PREFIX,
JOB_STATUSES,
STALE_JOB_STATUS,
@@ -153,12 +156,23 @@ JOB_STATE_RANKS = {"failure": 0, "pending": 1, "success":
2}
# every workflow of the repository. At four runs an hour this keeps job
statuses to at most about
# half of it; the rest follow on the next run, failures first. See "Job
statuses" in the design doc.
MAX_JOB_STATUSES_PER_RUN = 60
+# `CI run started` statuses of pull requests whose CI runs in apache/airflow
come out of that budget
+# first: one write per new head commit, which merging waits for, while job
statuses are informational.
+# At most this many, so job statuses keep the rest; the first runs after the
check became required
+# catch up on the open pull requests a batch at a time, most recently updated
first.
+MAX_CI_RUN_STARTED_PER_RUN = 30
+# The GitHub Actions app, whose check suites are the workflow runs of a commit.
+GITHUB_ACTIONS_APP_ID = 15368
+# A single large fork run can need more writes than one run's budget, which
the next run catches up
+# on, so deferring is only reported when it lasts this many consecutive runs:
then the budget does
+# not keep up with the pull requests.
+DEFERRED_RUNS_BEFORE_ALERT = 3
# Fixed text, so the note is posted to Slack once while deferrals persist; the
run log lists the pull
# requests.
JOB_STATUSES_DEFERRED_NOTE = (
- f"job statuses did not fit in the budget of {MAX_JOB_STATUSES_PER_RUN}
writes per run and are "
- "posted by later runs; if this persists, set `job_statuses = false` in
.github/fork-ci-gate.toml "
- "(see dev/AGENTS.md)"
+ f"job statuses have not fit in the budget of {MAX_JOB_STATUSES_PER_RUN}
writes per run for "
+ f"{DEFERRED_RUNS_BEFORE_ALERT} runs in a row and are posted by later runs;
if this persists, set "
+ "`job_statuses = false` in .github/fork-ci-gate.toml (see dev/AGENTS.md)"
)
RUNNING_STATUSES = frozenset({"QUEUED", "IN_PROGRESS", "PENDING", "WAITING",
"REQUESTED"})
@@ -236,6 +250,13 @@ class PullRequestSnapshot:
current_status: tuple[str, str, str] | None = None
status_created_at: datetime | None = None
ci_passed_state: str | None = None
+ ci_run_started_state: str | None = None
+ # GateEvent.ci_run_at of a pilot's pull request created before they
joined; set from its
+ # `fork ci required` label by `mark_older_pilot_pull_request`.
+ ci_run_at: datetime | None = None
+ # The join instant of such a pull request when project CI was allowed for
it before then, so its
+ # runs from before it are not gate tampering.
+ pilot_joined_at: datetime | None = None
marker: Marker | None = None
marker_comment_id: int | None = None
marker_comment_body: str | None = None
@@ -269,6 +290,8 @@ class Plan:
status: tuple[str, str, str] | None = None
# (state, target URL, description) of the `CI passed` status.
ci_passed: tuple[str, str, str] | None = None
+ # (state, target URL, description) of the `CI run started` status.
+ ci_run_started: tuple[str, str, str] | None = None
marker_comment: str | None = None
new_comments: list[str] = field(default_factory=list)
close: bool = False
@@ -370,6 +393,22 @@ def is_gated_skip_run(jobs: Iterable[dict[str, Any]], now:
datetime) -> bool:
)
+def is_cancelled_gate_run(jobs: Iterable[dict[str, Any]], now: datetime) ->
bool:
+ """
+ Whether a completed ci-amd.yml run was cancelled before it did any project
CI work.
+
+ A push soon after opening a pull request cancels the first run, often
before its gate has
+ decided, and the run that replaces it is the one left to fork CI. The
cancelled run's jobs stay
+ on the pull request as cancelled checks next to the fork's result. Only
the gate and the jobs
+ that also run in a gated run started, none failed, and at least one was
cancelled.
+ """
+ jobs = list(jobs)
+ if jobs_show_project_ci(jobs, now):
+ return False
+ conclusions = {job.get("conclusion") for job in jobs}
+ return "cancelled" in conclusions and conclusions <= {"success",
"skipped", "cancelled"}
+
+
def touches(files: Iterable[str], prefixes: tuple[str, ...]) -> bool:
return any(path.startswith(prefixes) for path in files)
@@ -413,6 +452,7 @@ def _gate_event(pr: PullRequestSnapshot, repo: str) ->
GateEvent:
author_association=pr.author_association,
created_at=pr.created_at,
labels=pr.labels,
+ ci_run_at=pr.ci_run_at,
)
@@ -452,6 +492,12 @@ def _set_ci_passed(plan: Plan, pr: PullRequestSnapshot, *,
passed: bool, target_
plan.ci_passed = ("pending", target_url, CI_PASSED_REVOKED_DESCRIPTION)
+def _set_ci_run_started(plan: Plan, pr: PullRequestSnapshot, fork:
ForkObservation) -> None:
+ """Report the `CI run started` check the merge rulesets require once the
fork runs the head commit."""
+ if fork.run_url and pr.ci_run_started_state != "success":
+ plan.ci_run_started = ("success", fork.run_url,
CI_RUN_STARTED_DESCRIPTION)
+
+
def _set_marker_comment(plan: Plan, pr: PullRequestSnapshot, body: str) ->
None:
if body != pr.marker_comment_body:
plan.marker_comment = body
@@ -702,6 +748,7 @@ def classify(
target_url = fork.run_url or fork_actions_url(pr.head_repo)
_set_status(plan, pr, state.value, target_url, force=refresh_status)
_set_ci_passed(plan, pr, passed=state == ForkState.GREEN,
target_url=target_url)
+ _set_ci_run_started(plan, pr, fork)
green = state in (ForkState.GREEN, ForkState.GREEN_UNVERIFIED)
_set_marker_comment(
plan,
@@ -732,7 +779,7 @@ query($q: String!, $cursor: String) {
search(type: ISSUE, first: 100, after: $cursor, query: $q) {
pageInfo { hasNextPage endCursor }
nodes { ... on PullRequest {
- number createdAt authorAssociation baseRefName
+ number createdAt authorAssociation baseRefName headRefName headRefOid
author { login }
headRepository { nameWithOwner }
labels(first: 50) { nodes { name } }
@@ -767,6 +814,7 @@ query($q: String!, $cursor: String) {
status {
context(name: "%(context)s") { state targetUrl description createdAt
}
ciPassed: context(name: "%(ci_passed)s") { state }
+ ciRunStarted: context(name: "%(ci_run_started)s") { state }
}
} } }
comments(first: 100) {
@@ -838,12 +886,15 @@ def parse_pull_request(node: dict[str, Any]) ->
PullRequestSnapshot:
current_status = None
status_created_at = None
ci_passed_state = None
+ ci_run_started_state = None
commits = (node.get("commits") or {}).get("nodes") or []
head_status_unread = bool(commits) and commits[0]["commit"].get("oid") !=
node["headRefOid"]
if commits and not head_status_unread:
status = commits[0]["commit"].get("status") or {}
if status.get("ciPassed"):
ci_passed_state = status["ciPassed"]["state"].lower()
+ if status.get("ciRunStarted"):
+ ci_run_started_state = status["ciRunStarted"]["state"].lower()
context = status.get("context")
if context:
current_status = (
@@ -883,6 +934,7 @@ def parse_pull_request(node: dict[str, Any]) ->
PullRequestSnapshot:
current_status=current_status,
status_created_at=status_created_at,
ci_passed_state=ci_passed_state,
+ ci_run_started_state=ci_run_started_state,
marker=marker,
marker_comment_id=marker_id,
marker_comment_body=marker_body,
@@ -892,6 +944,116 @@ def parse_pull_request(node: dict[str, Any]) ->
PullRequestSnapshot:
)
+PROJECT_CI_RUNS_QUERY = """
+query($q: String!, $cursor: String) {
+ search(type: ISSUE, first: 50, after: $cursor, query: $q) {
+ pageInfo { hasNextPage endCursor }
+ nodes { ... on PullRequest {
+ number headRefOid
+ commits(last: 1) { nodes { commit {
+ oid
+ status { context(name: "%(context)s") { state } }
+ checkSuites(first: 30, filterBy: {appId: %(app_id)d}) { nodes {
+ createdAt conclusion
+ workflowRun { url event workflow { resourcePath } }
+ } }
+ } } }
+ } }
+ }
+}
+"""
+
+
+def find_project_ci_run_url(node: dict[str, Any]) -> str | None:
+ """
+ The latest ci-amd.yml pull_request run of a pull request's head commit
that still needs `CI run started`.
+
+ None when the commit already has the status, when its latest run has not
started because it waits
+ for a maintainer's approval or there is none yet, and when the last commit
listed is not the head:
+ GitHub orders the commit list by date, so after a rebase that keeps old
dates it can be another one,
+ and ``read_project_ci_run_url`` reads the head instead.
+ """
+ if not is_head_listed(node):
+ return None
+ commit = node["commits"]["nodes"][0]["commit"]
+ if ((commit.get("status") or {}).get("context") or {}).get("state") ==
"SUCCESS":
+ return None
+ runs = [
+ suite
+ for suite in (commit.get("checkSuites") or {}).get("nodes") or []
+ if (run := suite.get("workflowRun") or {}).get("event") ==
"pull_request"
+ and _workflow_file((run.get("workflow") or {}).get("resourcePath")) ==
CI_WORKFLOW_FILE
+ ]
+ if not runs:
+ return None
+ latest = max(runs, key=lambda suite: suite["createdAt"])
+ if latest.get("conclusion") == "ACTION_REQUIRED":
+ return None
+ return latest["workflowRun"]["url"]
+
+
+def is_head_listed(node: dict[str, Any]) -> bool:
+ commits = (node.get("commits") or {}).get("nodes") or []
+ return bool(commits) and commits[0]["commit"].get("oid") ==
node.get("headRefOid")
+
+
+def read_project_ci_run_url(client: GitHubClient, repo: str, head_sha: str) ->
str | None:
+ """``find_project_ci_run_url`` for a head commit the search did not list,
read from REST."""
+ statuses = client.rest_get(
+ f"repos/{repo}/commits/{head_sha}/status?per_page=100", paginate=True,
items_key="statuses"
+ )
+ if any(
+ status["context"] == CI_RUN_STARTED_CONTEXT and status["state"] ==
"success" for status in statuses
+ ):
+ return None
+ runs = client.rest_get(
+ f"repos/{repo}/actions/workflows/{CI_WORKFLOW_FILE}/runs"
+ f"?event=pull_request&head_sha={head_sha}&per_page=1"
+ ).get("workflow_runs", [])
+ if not runs or runs[0].get("conclusion") == "action_required":
+ return None
+ return runs[0]["html_url"]
+
+
+def report_project_ci_runs_started(
+ client: GitHubClient, *, repo: str, skip: Collection[int], budget: int =
MAX_CI_RUN_STARTED_PER_RUN
+) -> list[int]:
+ """
+ Report `CI run started` for open pull requests whose CI runs in
apache/airflow; return their numbers.
+
+ :param skip: the pull requests this run treated as gated, which get the
status from their fork
+ run instead; the search index can still list one without `fork ci
required`.
+ """
+ reported: list[int] = []
+ for node in _search(
+ client,
+ PROJECT_CI_RUNS_QUERY % {"context": CI_RUN_STARTED_CONTEXT, "app_id":
GITHUB_ACTIONS_APP_ID},
+ f'repo:{repo} is:pr is:open -label:"{LABEL_FORK_CI_REQUIRED}"
sort:updated-desc',
+ ):
+ if len(reported) >= budget:
+ console_print(f"[info]`{CI_RUN_STARTED_CONTEXT}` of more pull
requests left for the next run[/]")
+ break
+ if node["number"] in skip:
+ continue
+ run_url = (
+ find_project_ci_run_url(node)
+ if is_head_listed(node)
+ else read_project_ci_run_url(client, repo, node["headRefOid"])
+ )
+ if run_url is None:
+ continue
+ client.set_status(
+ node["number"],
+ node["headRefOid"],
+ "success",
+ run_url,
+ CI_RUN_STARTED_PROJECT_DESCRIPTION,
+ context=CI_RUN_STARTED_CONTEXT,
+ )
+ reported.append(node["number"])
+ return reported
+
+
def _search(client: GitHubClient, query: str, search_query: str) ->
Iterable[dict[str, Any]]:
cursor = None
while True:
@@ -950,6 +1112,68 @@ class BaseGateConfigs:
return self._configs[base_ref]
+def find_ci_run_since(
+ client: GitHubClient, repo: str, *, head_repo: str, head_ref: str,
head_sha: str, since: datetime
+) -> datetime | None:
+ """
+ When the latest ci-amd.yml pull_request run of a branch's commit created
at or after ``since`` was created.
+
+ The gate decides with the creation time of its own run
(GateEvent.ci_run_at); this reads the
+ same time back, so both decide alike for a pilot's pull request created
before they joined.
+ Runs are matched by branch and fork too: the same commit pushed to another
branch is another
+ pull request's.
+ """
+ created =
quote(f">={since.astimezone(UTC).strftime('%Y-%m-%dT%H:%M:%SZ')}", safe="")
+ runs = client.rest_get(
+
f"repos/{repo}/actions/workflows/{CI_WORKFLOW_FILE}/runs?event=pull_request"
+ f"&branch={quote(head_ref,
safe='')}&head_sha={head_sha}&created={created}&per_page=100"
+ ).get("workflow_runs", [])
+ times = [
+ parse_github_datetime(run["created_at"])
+ for run in runs
+ if ((run.get("head_repository") or {}).get("full_name") or
"").casefold() == head_repo.casefold()
+ ]
+ return max((time for time in times if time is not None), default=None)
+
+
+def read_ci_run_at(
+ client: GitHubClient,
+ repo: str,
+ config: GateConfig,
+ *,
+ author: str,
+ created_at: datetime | None,
+ head_repo: str,
+ head_ref: str,
+ head_sha: str,
+) -> datetime | None:
+ """The ``ci_run_at`` of a gate event; only read for a pilot's pull request
created before they joined."""
+ joined_at = config.find_pilot_join_after(author, created_at)
+ if joined_at is None or not (head_repo and head_ref and head_sha):
+ return None
+ return find_ci_run_since(
+ client, repo, head_repo=head_repo, head_ref=head_ref,
head_sha=head_sha, since=joined_at
+ )
+
+
+def mark_older_pilot_pull_request(
+ pr: PullRequestSnapshot, config: GateConfig, *, repo: str, now: datetime
+) -> PullRequestSnapshot:
+ """
+ Record that a pilot's pull request created before they joined is gated,
for the tampering check.
+
+ Its `fork ci required` label shows that discovery found a run created
after joining, which is
+ what gated it; that run is not read again, because the reconciler deletes
gated runs. Before the
+ join instant the label came from something else, such as `use fork ci`.
Its runs from before
+ joining are left out of the tampering check only when project CI was
allowed then.
+ """
+ joined_at = config.find_pilot_join_after(pr.author_login, pr.created_at)
+ if joined_at is None or joined_at > now:
+ return pr
+ allowed_before = decide(_gate_event(pr, repo), config).reason not in
GATED_REASONS
+ return replace(pr, ci_run_at=joined_at, pilot_joined_at=joined_at if
allowed_before else None)
+
+
def discover(
client: GitHubClient,
configs: BaseGateConfigs,
@@ -981,15 +1205,27 @@ def discover(
continue
if config is None:
continue
+ author = _login(node.get("author"))
+ created_at = parse_github_datetime(node["createdAt"])
event = GateEvent(
event_name="pull_request",
repository=repo,
ref="",
head_repository=(node.get("headRepository") or
{}).get("nameWithOwner") or "",
- author_login=_login(node.get("author")),
+ author_login=author,
author_association=node["authorAssociation"],
- created_at=parse_github_datetime(node["createdAt"]),
+ created_at=created_at,
labels=_labels(node),
+ ci_run_at=read_ci_run_at(
+ client,
+ repo,
+ config,
+ author=author,
+ created_at=created_at,
+ head_repo=(node.get("headRepository") or
{}).get("nameWithOwner") or "",
+ head_ref=node.get("headRefName") or "",
+ head_sha=node.get("headRefOid") or "",
+ ),
)
if decide(event, config).reason in GATED_REASONS:
client.add_labels(node["number"], [LABEL_FORK_CI_REQUIRED])
@@ -1293,7 +1529,8 @@ class BypassFinder:
def find_gated_runs(self, pr: PullRequestSnapshot) -> list[WorkflowRun]:
"""
- Completed ci-amd.yml pull_request runs of the pull request's branch
that only gated it.
+ Completed ci-amd.yml pull_request runs of the pull request's branch
that only gated it, or
+ that a newer push cancelled before they did any project CI work.
A commit that changes workflows could do project CI work under the
gate's name within its
time limit, so its runs are kept as evidence, as are runs whose change
set is too large
@@ -1310,7 +1547,10 @@ class BypassFinder:
and run.status == "completed"
and run.run_attempt == 1
and not self._did_run_project_ci(run)
- and is_gated_skip_run(self._get_jobs(run), self._now)
+ and (
+ is_gated_skip_run(self._get_jobs(run), self._now)
+ or is_cancelled_gate_run(self._get_jobs(run), self._now)
+ )
and self._commit_changes_workflows(run, pr.base_ref) is False
]
@@ -1360,6 +1600,9 @@ 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
+ # A pilot's pull request created before they joined was not gated
until then.
+ if pr.pilot_joined_at is not None and run.created_at <
pr.pilot_joined_at:
+ 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):
@@ -1634,6 +1877,7 @@ def read_head_status(client: GitHubClient, repo: str, pr:
PullRequestSnapshot) -
}
status = statuses.get(STATUS_CONTEXT)
ci_passed = statuses.get(CI_PASSED_CONTEXT)
+ ci_run_started = statuses.get(CI_RUN_STARTED_CONTEXT)
return replace(
pr,
current_status=(
@@ -1643,6 +1887,7 @@ def read_head_status(client: GitHubClient, repo: str, pr:
PullRequestSnapshot) -
),
status_created_at=parse_github_datetime(status.get("created_at") or
"") if status else None,
ci_passed_state=ci_passed["state"] if ci_passed else None,
+ ci_run_started_state=ci_run_started["state"] if ci_run_started else
None,
head_status_unread=False,
)
@@ -1724,6 +1969,8 @@ 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)
+ if plan.ci_run_started is not None:
+ client.set_status(pr.number, pr.head_sha, *plan.ci_run_started,
context=CI_RUN_STARTED_CONTEXT)
if plan.close:
client.close(pr.number)
# Last, because the label is what brings the pull request back to the next
run: when an
@@ -1791,17 +2038,21 @@ def drop_reported(
attention: list[tuple[int, str]],
notes: list[str],
record: bool = True,
+ deferred_runs: int = 0,
) -> tuple[list[tuple[int, str]], list[tuple[int, str]], list[str]]:
"""
Leave out the items the previous run already reported, and record this
run's items.
Only the previous run's items are kept, so an item that went away and
comes back is posted
again. A missing or unreadable file reports everything.
+
+ :param deferred_runs: the consecutive runs that deferred job statuses,
recorded for the next run.
"""
- current: dict[str, list[Any]] = {
+ current: dict[str, Any] = {
"unknown": [list(item) for item in unknown],
"attention": [list(item) for item in attention],
"notes": notes,
+ "deferred_runs": deferred_runs,
}
try:
previous = json.loads(reported_file.read_text())
@@ -1819,6 +2070,21 @@ def drop_reported(
return drop("unknown", unknown), drop("attention", attention),
drop("notes", notes)
+def read_deferred_runs(reported_file: Path | None) -> int | None:
+ """The consecutive runs before this one that deferred job statuses; None
without a readable record."""
+ if reported_file is None:
+ return None
+ try:
+ previous = json.loads(reported_file.read_text())
+ except (OSError, ValueError):
+ return None
+ if not isinstance(previous, dict):
+ return None
+ # A record written before the count was kept has none.
+ count = previous.get("deferred_runs", 0)
+ return count if isinstance(count, int) and not isinstance(count, bool)
else 0
+
+
def run_reconcile(
client: GitHubClient,
config: GateConfig,
@@ -1850,14 +2116,19 @@ def run_reconcile(
draft_changes: list[DraftChange] = []
configs = BaseGateConfigs(client, repo=repo,
default_branch=default_branch, default_config=config)
- discover(client, configs, repo=repo, now=now, unknown=unknown)
+ labelled = discover(client, configs, repo=repo, now=now, unknown=unknown)
prs = [
parse_pull_request(node)
for node in _search(
client,
PASS_TWO_QUERY
- % {"page": PASS_TWO_PAGE_SIZE, "context": STATUS_CONTEXT,
"ci_passed": CI_PASSED_CONTEXT},
+ % {
+ "page": PASS_TWO_PAGE_SIZE,
+ "context": STATUS_CONTEXT,
+ "ci_passed": CI_PASSED_CONTEXT,
+ "ci_run_started": CI_RUN_STARTED_CONTEXT,
+ },
f'repo:{repo} is:pr is:open label:"{LABEL_FORK_CI_REQUIRED}"',
)
]
@@ -1877,7 +2148,21 @@ def run_reconcile(
base_triggers=load_workflow_triggers(workflows_dir),
is_gated_base=is_gated_base,
)
- job_reporter = JobStatusReporter(client, repo=repo)
+ ci_run_started_note = None
+ try:
+ ci_run_started_writes = len(
+ report_project_ci_runs_started(client, repo=repo, skip={*labelled,
*(pr.number for pr in prs)})
+ )
+ except Exception as e:
+ console_print(f"[error]Reporting `{CI_RUN_STARTED_CONTEXT}` for
project CI failed: {e}[/]")
+ # Writes made before the failure are not known, so the whole allowance
is charged.
+ ci_run_started_writes = MAX_CI_RUN_STARTED_PER_RUN
+ ci_run_started_note = (
+ f"reporting `{CI_RUN_STARTED_CONTEXT}` for project CI failed:
{type(e).__name__}"
+ )
+ job_reporter = JobStatusReporter(
+ client, repo=repo, budget=max(MAX_JOB_STATUSES_PER_RUN -
ci_run_started_writes, 0)
+ )
shared_heads = find_heads_shared_across_bases(prs)
gate_change_finder = GateChangeFinder(client, repo=repo)
@@ -1901,10 +2186,13 @@ def run_reconcile(
console_print(f"[warning]#{pr.number}: reporting job
statuses failed: {e}[/]")
unknown.append((pr.number, f"reporting job statuses
failed: {type(e).__name__}: {e}"))
+ base_config = configs.get(pr.base_ref)
+ if base_config is not None:
+ pr = mark_older_pilot_pull_request(pr, base_config, repo=repo,
now=now)
plan = classify(
pr,
fork,
- config=configs.get(pr.base_ref),
+ config=base_config,
now=now,
repo=repo,
find_bypass_run=functools.partial(bypass_finder.find, pr),
@@ -1920,8 +2208,8 @@ def run_reconcile(
if draft_change is not None:
draft_changes.append(draft_change)
# The fork's result is in the `Fork CI / Tests (AMD)` status; the
gated run's skipped
- # checks only clutter the pull request. Runs with any project CI
work stay for the
- # tampering check.
+ # checks, and the cancelled ones of a run a newer push cancelled,
only clutter the pull
+ # request. Runs with any project CI work stay for the tampering
check.
if plan.state != ForkState.GATE_BYPASSED:
for run in bypass_finder.find_gated_runs(pr):
client.delete_run(pr.number, run.id)
@@ -1935,12 +2223,19 @@ def run_reconcile(
attention.extend(bypass_finder.ambiguous)
notes = [bypass_finder.incomplete] if bypass_finder.incomplete else []
+ if ci_run_started_note:
+ notes.append(ci_run_started_note)
+ previous_deferred_runs = read_deferred_runs(reported_file)
+ deferred_runs = (previous_deferred_runs or 0) + 1 if job_reporter.deferred
else 0
if job_reporter.deferred:
console_print(
- f"[info]Job statuses of {len(job_reporter.deferred)} pull requests
left for the next run: "
- f"{job_reporter.deferred}[/]"
+ f"[info]Job statuses of {len(job_reporter.deferred)} pull requests
left for the next run "
+ f"({deferred_runs} runs in a row): {job_reporter.deferred}[/]"
)
- notes.append(JOB_STATUSES_DEFERRED_NOTE)
+ # Without a record of earlier runs, deferring is reported at once, as
it cannot be told apart
+ # from a budget that does not keep up.
+ if previous_deferred_runs is None or deferred_runs >=
DEFERRED_RUNS_BEFORE_ALERT:
+ notes.append(JOB_STATUSES_DEFERRED_NOTE)
# The full list for the run log; Slack may show only part of a very long
one.
for number, reason in [*unknown, *attention]:
console_print(f"[warning]#{number}: {reason}[/]")
@@ -1949,7 +2244,12 @@ def run_reconcile(
draft_changes_file.write_text(json.dumps([asdict(change) for change in
draft_changes]))
if reported_file is not None:
unknown, attention, notes = drop_reported(
- reported_file, unknown=unknown, attention=attention, notes=notes,
record=not client.dry_run
+ reported_file,
+ unknown=unknown,
+ attention=attention,
+ notes=notes,
+ record=not client.dry_run,
+ deferred_runs=deferred_runs,
)
has_alerts = bool(unknown or attention or notes)
if has_alerts:
diff --git a/dev/breeze/src/airflow_breeze/utils/workflow_run_approval.py
b/dev/breeze/src/airflow_breeze/utils/workflow_run_approval.py
index ec30ba279a4..d32d0455f1b 100644
--- a/dev/breeze/src/airflow_breeze/utils/workflow_run_approval.py
+++ b/dev/breeze/src/airflow_breeze/utils/workflow_run_approval.py
@@ -43,7 +43,7 @@ import os
import re
import subprocess
from dataclasses import dataclass, field
-from datetime import datetime
+from datetime import UTC, datetime
from typing import Any
import requests
@@ -170,6 +170,8 @@ class ApprovalDecision:
comment: str | None = None
# The labels the decision depends on: runs started before one of them
changed are not approved.
decision_labels: frozenset[str] = TEST_SELECTION_LABELS
+ # Runs started before this are not approved either.
+ not_before: datetime | None = None
def count_source_lines(changed_files: list[ChangedFile]) -> int:
@@ -196,8 +198,14 @@ def find_new_provider_files(files: list[str]) -> list[str]:
]
-def is_gated_to_fork_ci(*, pr: dict[str, Any], github_repository: str, config:
GateConfig) -> bool:
- """Whether the PR's ``Tests (AMD)`` run only executes the fork CI gate
(AIP-120)."""
+def is_gated_to_fork_ci(
+ *, pr: dict[str, Any], github_repository: str, config: GateConfig,
ci_run_at: datetime | None = None
+) -> bool:
+ """
+ Whether the PR's ``Tests (AMD)`` run only executes the fork CI gate
(AIP-120).
+
+ :param ci_run_at: when the run was created, as the gate reads it; see
``GateEvent.ci_run_at``.
+ """
event = GateEvent(
event_name="pull_request",
repository=github_repository,
@@ -207,6 +215,7 @@ def is_gated_to_fork_ci(*, pr: dict[str, Any],
github_repository: str, config: G
author_association=pr.get("author_association") or "",
created_at=parse_github_datetime(pr["created_at"]),
labels=tuple(label["name"] for label in pr.get("labels", [])),
+ ci_run_at=ci_run_at,
)
return decide(event, config).reason in GATED_REASONS
@@ -663,7 +672,11 @@ def process_waiting_runs(api: GitHubApi, *, created_since:
datetime) -> list[Pul
# A trusted repository's runs do not depend on the base branch
or labels they started with.
if decision.approve and stakeholder is None:
pr_run_ids = find_runs_started_for_current_pr(
- api=api, pr=pr, runs=head_runs,
decision_labels=decision.decision_labels
+ api=api,
+ pr=pr,
+ runs=head_runs,
+ decision_labels=decision.decision_labels,
+ not_before=decision.not_before,
)
if not pr_run_ids:
decision = ApprovalDecision(
@@ -726,6 +739,7 @@ def find_runs_started_for_current_pr(
pr: dict[str, Any],
runs: list[dict[str, Any]],
decision_labels: frozenset[str] = TEST_SELECTION_LABELS,
+ not_before: datetime | None = None,
) -> list[int]:
"""
Return the runs started after the PR was opened and after its base branch
last changed.
@@ -744,11 +758,13 @@ def find_runs_started_for_current_pr(
api.get_last_base_change(pr["number"]),
api.get_last_label_change(pr["number"], decision_labels),
]
- started_after = max([started_after, *(change for change in changes if
change is not None)])
+ started_after = max([started_after, *(time for time in (*changes,
not_before) if time is not None)])
return [run["id"] for run in runs if parse_github_time(run["created_at"])
>= started_after]
-def decide_for_pull_request(*, api: GitHubApi, pr: dict[str, Any]) ->
ApprovalDecision:
+def decide_for_pull_request(
+ *, api: GitHubApi, pr: dict[str, Any], now: datetime | None = None
+) -> ApprovalDecision:
merge_sha = fetch_merge_commit(
github_repository=api.github_repository, pr_number=pr["number"],
head_sha=pr["head"]["sha"]
)
@@ -760,10 +776,20 @@ def decide_for_pull_request(*, api: GitHubApi, pr:
dict[str, Any]) -> ApprovalDe
changed_files = get_changed_files(merge_sha)
except ValueError:
return ApprovalDecision(approve=False, reasons=["its changed files
could not be listed"])
+ config = load_config(FORK_CI_GATE_CONFIG_PATH)
+ # A pilot's PR created before they joined is gated from its first run
created after joining; its
+ # runs from before, and every run until the join instant has passed, stay
with the rules below.
+ joined_at = config.find_pilot_join_after(
+ (pr.get("user") or {}).get("login") or "",
parse_github_time(pr["created_at"])
+ )
+ if joined_at is not None and joined_at > (now or datetime.now(UTC)):
+ joined_at = None
if pr["base"]["ref"] == AUTO_APPROVABLE_BASE_BRANCH and
is_gated_to_fork_ci(
- pr=pr, github_repository=api.github_repository,
config=load_config(FORK_CI_GATE_CONFIG_PATH)
+ pr=pr, github_repository=api.github_repository, config=config,
ci_run_at=joined_at
):
- return decide_gated_approval(changed_files=changed_files)
+ decision = decide_gated_approval(changed_files=changed_files)
+ decision.not_before = joined_at
+ return decision
pr_labels = tuple(label["name"] for label in pr.get("labels", []))
return decide_approval(
base_branch=pr["base"]["ref"],
diff --git a/dev/breeze/tests/test_fork_ci_gate.py
b/dev/breeze/tests/test_fork_ci_gate.py
index 51f351e8b03..29b4d5c820c 100644
--- a/dev/breeze/tests/test_fork_ci_gate.py
+++ b/dev/breeze/tests/test_fork_ci_gate.py
@@ -62,6 +62,7 @@ def _pr(
created_at: datetime = AFTER,
labels: tuple[str, ...] = (),
head_repository: str = "someone/airflow",
+ ci_run_at: datetime | None = None,
) -> GateEvent:
return GateEvent(
event_name="pull_request",
@@ -72,6 +73,7 @@ def _pr(
author_association=association,
created_at=created_at,
labels=labels,
+ ci_run_at=ci_run_at,
)
@@ -202,6 +204,28 @@ def _push(repository: str, branch: str) -> GateEvent:
Decision(run=True, reason=Reason.GATE_DISABLED),
id="pilot-before-join-instant",
),
+ pytest.param(
+ _pr(author="pilot-user", created_at=JOINED - timedelta(days=5),
ci_run_at=JOINED),
+ {"enabled": False},
+ Decision(run=False, reason=Reason.PILOT),
+ id="pilot-older-pr-run-at-join-instant",
+ ),
+ pytest.param(
+ _pr(
+ author="pilot-user",
+ created_at=JOINED - timedelta(days=5),
+ ci_run_at=JOINED - timedelta(seconds=1),
+ ),
+ {"enabled": False},
+ Decision(run=True, reason=Reason.GATE_DISABLED),
+ id="pilot-older-pr-run-before-join-instant",
+ ),
+ pytest.param(
+ _pr(created_at=SINCE - timedelta(days=5), ci_run_at=AFTER),
+ {},
+ Decision(run=True, reason=Reason.BEFORE_ENABLED_SINCE),
+ id="run-time-does-not-move-enabled-since",
+ ),
pytest.param(
_pr(author="pilot-user", created_at=AFTER, labels=("use project
ci",)),
{},
@@ -428,3 +452,89 @@ def test_main_fork_push_does_not_read_labels(mock_run,
gate_env):
assert _outputs(gate_env) == ["project-ci=true", "reason=fork_push"]
mock_run.assert_not_called()
+
+
+PILOT_CONFIG = '[pilot]\n"pilot-user" = 2026-10-20T00:00:00Z\n'
+
+
+def _gh_responses(labels: str, run: subprocess.CompletedProcess | Exception):
+ def run_gh(args, **kwargs):
+ if args[2].startswith("repos/apache/airflow/pulls/"):
+ return subprocess.CompletedProcess([], 0, stdout=labels, stderr="")
+ assert args[2] == "repos/apache/airflow/actions/runs/123"
+ if isinstance(run, Exception):
+ raise run
+ return run
+
+ return run_gh
+
+
+def _older_pilot_pull_request(env: dict[str, str]) -> None:
+ Path(env["GATE_CONFIG"]).write_text(PILOT_CONFIG)
+ env.update(PR_AUTHOR="pilot-user", PR_CREATED_AT="2026-10-15T00:00:00Z",
GITHUB_RUN_ID="123")
+
+
[email protected]("airflow_breeze.utils.fork_ci_gate.subprocess.run", autospec=True)
[email protected](
+ ("run_created_at", "expected"),
+ [
+ pytest.param("2026-10-20T00:00:00Z\n", ["project-ci=false",
"reason=pilot"], id="run-after-joining"),
+ pytest.param(
+ "2026-10-19T23:59:59Z\n", ["project-ci=true",
"reason=gate_disabled"], id="run-before-joining"
+ ),
+ ],
+)
+def test_main_decides_a_pilots_older_pull_request_by_its_run_time(
+ mock_run, gate_env, run_created_at, expected
+):
+ _older_pilot_pull_request(gate_env)
+ mock_run.side_effect = _gh_responses(
+ "[]\n", subprocess.CompletedProcess([], 0, stdout=run_created_at,
stderr="")
+ )
+
+ assert main(gate_env) == 0
+
+ assert _outputs(gate_env) == expected
+
+
[email protected]("airflow_breeze.utils.fork_ci_gate.subprocess.run", autospec=True)
[email protected](
+ "run",
+ [
+ pytest.param(subprocess.CompletedProcess([], 1, stdout="",
stderr="boom"), id="gh-fails"),
+ pytest.param(subprocess.CompletedProcess([], 0, stdout="not a date\n",
stderr=""), id="bad-output"),
+ pytest.param(FileNotFoundError("gh"), id="no-gh"),
+ ],
+)
+def
test_main_fails_when_the_run_time_of_a_pilots_older_pull_request_cannot_be_read(mock_run,
gate_env, run):
+ _older_pilot_pull_request(gate_env)
+ mock_run.side_effect = _gh_responses("[]\n", run)
+
+ assert main(gate_env) == 1
+
+ assert not Path(gate_env["GITHUB_OUTPUT"]).exists()
+
+
[email protected]("airflow_breeze.utils.fork_ci_gate.subprocess.run", autospec=True)
+def
test_main_reads_the_run_time_only_for_a_pilots_older_pull_request(mock_run,
gate_env):
+ Path(gate_env["GATE_CONFIG"]).write_text(PILOT_CONFIG)
+ gate_env.update(PR_AUTHOR="pilot-user", GITHUB_RUN_ID="123")
+ mock_run.return_value = subprocess.CompletedProcess([], 0, stdout="[]\n",
stderr="")
+
+ assert main(gate_env) == 0
+
+ assert _outputs(gate_env) == ["project-ci=false", "reason=pilot"]
+ mock_run.assert_called_once()
+
+
[email protected](
+ ("login", "created_at", "expected"),
+ [
+ pytest.param("Pilot-User", JOINED - timedelta(seconds=1), JOINED,
id="created-before"),
+ pytest.param("pilot-user", JOINED, None, id="created-at-join-instant"),
+ pytest.param("someone", JOINED - timedelta(days=1), None,
id="not-a-pilot"),
+ pytest.param("pilot-user", None, None, id="no-creation-time"),
+ ],
+)
+def test_find_pilot_join_after(login, created_at, expected):
+ assert _config().find_pilot_join_after(login, created_at) == expected
diff --git a/dev/breeze/tests/test_fork_ci_reconcile.py
b/dev/breeze/tests/test_fork_ci_reconcile.py
index 7d906d37023..d0cf3feb67d 100644
--- a/dev/breeze/tests/test_fork_ci_reconcile.py
+++ b/dev/breeze/tests/test_fork_ci_reconcile.py
@@ -19,7 +19,7 @@ from __future__ import annotations
import base64
import dataclasses
import json
-from datetime import UTC, datetime, timedelta
+from datetime import UTC, datetime, timedelta, timezone
from unittest import mock
import pytest
@@ -30,6 +30,9 @@ from airflow_breeze.utils.fork_ci_gate import GateConfig
from airflow_breeze.utils.fork_ci_github import GitHubClient, GitHubError
from airflow_breeze.utils.fork_ci_messages import (
CI_PASSED_CONTEXT,
+ CI_RUN_STARTED_CONTEXT,
+ CI_RUN_STARTED_DESCRIPTION,
+ CI_RUN_STARTED_PROJECT_DESCRIPTION,
JOB_STATUS_CONTEXT_PREFIX,
STALE_JOB_STATUS,
STATUS_CONTEXT,
@@ -59,15 +62,22 @@ from airflow_breeze.utils.fork_ci_reconcile import (
drop_reported,
fetch_fork_observations,
fetch_workflow_state,
+ find_ci_run_since,
find_heads_shared_across_bases,
+ find_project_ci_run_url,
find_state_comment,
+ is_cancelled_gate_run,
is_gated_skip_run,
is_own_pilot_change,
jobs_show_project_ci,
load_workflow_triggers,
+ mark_older_pilot_pull_request,
parse_pull_request,
pull_request_changes_ci_files,
+ read_ci_run_at,
read_head_status,
+ read_project_ci_run_url,
+ report_project_ci_runs_started,
run_reconcile,
)
from airflow_breeze.utils.path_utils import AIRFLOW_ROOT_PATH
@@ -75,6 +85,7 @@ from airflow_breeze.utils.path_utils import AIRFLOW_ROOT_PATH
REPO = "apache/airflow"
NOW = datetime(2026, 11, 10, 12, 0, tzinfo=UTC)
CREATED = datetime(2026, 11, 1, tzinfo=UTC)
+JOINED = CREATED + timedelta(days=2)
CONFIG = GateConfig(enabled=True, enabled_since=datetime(2026, 10, 1,
tzinfo=UTC), close_after_days=7)
RUN_URL = "https://github.com/someone/airflow/actions/runs/1";
PROJECT_RUN_URL = "https://github.com/apache/airflow/actions/runs/2";
@@ -940,6 +951,21 @@ def
test_finder_catches_a_tampered_run_still_in_progress(client):
assert finder.find(_pr()) ==
"https://github.com/apache/airflow/actions/runs/7";
[email protected](
+ ("joined_hours_ago", "expected"),
+ [
+ pytest.param(0.5, None, id="run-before-joining-was-legitimate"),
+ pytest.param(2, "https://github.com/apache/airflow/actions/runs/7";,
id="run-after-joining"),
+ ],
+)
+def test_finder_leaves_out_runs_of_an_older_pilot_pull_request_before_joining(
+ client, joined_hours_ago, expected
+):
+ finder = _finder(client, [_run(7, age_hours=1)], jobs=TAMPERED_JOBS)
+
+ assert finder.find(_pr(pilot_joined_at=NOW -
timedelta(hours=joined_hours_ago))) == expected
+
+
def test_finder_catches_a_tampered_run_on_an_earlier_commit(client):
finder = _finder(client, [_run(7, sha="tampered")], jobs=TAMPERED_JOBS)
@@ -1141,10 +1167,49 @@ def test_is_gated_skip_run(jobs, expected):
assert is_gated_skip_run(jobs, NOW) is expected
+CANCELLED_JOBS = [
+ _job("Check if the run is needed", "success"),
+ _job("Fork CI gate", "cancelled"),
+ _job("Build info", "cancelled", started=False),
+ _job("Tests", "cancelled", started=False),
+ _job("CI passed (run cancelled)", "success"),
+]
+
+
[email protected](
+ ("jobs", "expected"),
+ [
+ pytest.param(CANCELLED_JOBS, True, id="cancelled-while-the-gate-ran"),
+ pytest.param(
+ [*CANCELLED_JOBS[:2], _job("Build info", "cancelled"),
*CANCELLED_JOBS[3:]],
+ False,
+ id="cancelled-after-build-info-started",
+ ),
+ pytest.param([*CANCELLED_JOBS[:4], _job("CI passed", "failure")],
False, id="failed-ci-passed"),
+ pytest.param(GATED_JOBS, False, id="gated-not-cancelled"),
+ pytest.param(
+ [_job("Fork CI gate", "cancelled", minutes=30),
*CANCELLED_JOBS[2:]],
+ False,
+ id="work-under-the-gate-name",
+ ),
+ ],
+)
+def test_is_cancelled_gate_run(jobs, expected):
+ assert is_cancelled_gate_run(jobs, NOW) is expected
+
+
@pytest.mark.parametrize(
("runs", "jobs", "compare_files", "expected_ids"),
[
pytest.param([_run(7, minutes=1)], GATED_JOBS, ("README.md",), [7],
id="gated"),
+ pytest.param([_run(7, minutes=1)], CANCELLED_JOBS, ("README.md",),
[7], id="cancelled"),
+ pytest.param(
+ [_run(7, minutes=1)],
+ CANCELLED_JOBS,
+ (".github/workflows/ci-amd.yml",),
+ [],
+ id="cancelled-with-workflows-changed",
+ ),
pytest.param([_run(7, minutes=1)], TAMPERED_JOBS, ("README.md",), [],
id="project-ci-ran"),
pytest.param(
[_run(7, minutes=1, status="in_progress")], GATED_JOBS,
("README.md",), [], id="in-progress"
@@ -1380,6 +1445,256 @@ def test_read_head_status(client, statuses,
expected_status, expected_ci_passed)
assert read.head_status_unread is False
+def test_parse_pull_request_reads_ci_run_started():
+ node = _pr_node(
+ 1,
+ commits={
+ "nodes": [
+ {"commit": {"oid": "sha1", "status": {"context": None,
"ciRunStarted": {"state": "SUCCESS"}}}}
+ ]
+ },
+ )
+
+ assert parse_pull_request(node).ci_run_started_state == "success"
+
+
+def test_read_head_status_reads_ci_run_started(client):
+ client.rest_get.return_value = [{"context": CI_RUN_STARTED_CONTEXT,
"state": "success"}]
+ pr = dataclasses.replace(parse_pull_request(_pr_node(1)),
head_status_unread=True)
+
+ assert read_head_status(client, REPO, pr).ci_run_started_state == "success"
+
+
+def _branch_run(created_at: str, head_repo: str = "someone/airflow") -> dict:
+ return {"created_at": created_at, "head_repository": {"full_name":
head_repo}}
+
+
+def _find_ci_run_since(client, since=datetime(2026, 11, 1, tzinfo=UTC)):
+ return find_ci_run_since(
+ client, REPO, head_repo="Someone/airflow", head_ref="my/change",
head_sha="sha1", since=since
+ )
+
+
[email protected](
+ ("runs", "expected"),
+ [
+ pytest.param(
+ [_branch_run("2026-11-02T10:00:00Z"),
_branch_run("2026-11-03T10:00:00Z")],
+ datetime(2026, 11, 3, 10, tzinfo=UTC),
+ id="latest-run",
+ ),
+ pytest.param([_branch_run("2026-11-02T10:00:00Z", "other/airflow")],
None, id="other-fork"),
+ pytest.param([], None, id="no-run"),
+ ],
+)
+def test_find_ci_run_since(client, runs, expected):
+ client.rest_get.return_value = {"workflow_runs": runs}
+
+ assert _find_ci_run_since(client) == expected
+ client.rest_get.assert_called_once_with(
+
f"repos/{REPO}/actions/workflows/ci-amd.yml/runs?event=pull_request&branch=my%2Fchange"
+ "&head_sha=sha1&created=%3E%3D2026-11-01T00%3A00%3A00Z&per_page=100"
+ )
+
+
+def test_find_ci_run_since_converts_the_join_instant_to_utc(client):
+ client.rest_get.return_value = {"workflow_runs": []}
+
+ _find_ci_run_since(client, since=datetime(2026, 11, 1, 10,
tzinfo=timezone(timedelta(hours=2))))
+
+ assert "created=%3E%3D2026-11-01T08%3A00%3A00Z" in
client.rest_get.call_args.args[0]
+
+
[email protected](
+ ("author", "created_at"),
+ [
+ pytest.param("someone", datetime(2026, 10, 1, tzinfo=UTC),
id="not-a-pilot"),
+ pytest.param("pilot-user", datetime(2026, 11, 1, tzinfo=UTC),
id="created-after-joining"),
+ ],
+)
+def test_read_ci_run_at_only_reads_for_older_pilot_pull_requests(client,
author, created_at):
+ config = GateConfig(pilot={"pilot-user": datetime(2026, 11, 1,
tzinfo=UTC)})
+
+ assert (
+ read_ci_run_at(
+ client,
+ REPO,
+ config,
+ author=author,
+ created_at=created_at,
+ head_repo="someone/airflow",
+ head_ref="branch",
+ head_sha="sha1",
+ )
+ is None
+ )
+ client.rest_get.assert_not_called()
+
+
+def _project_node(number, *, status=None, suites=(), oid=None):
+ return {
+ "number": number,
+ "headRefOid": f"sha{number}",
+ "commits": {
+ "nodes": [
+ {
+ "commit": {
+ "oid": oid or f"sha{number}",
+ "status": {"context": {"state": status}} if status
else None,
+ "checkSuites": {"nodes": list(suites)},
+ }
+ }
+ ]
+ },
+ }
+
+
+def _suite(
+ run_id,
+ *,
+ created="2026-11-10T10:00:00Z",
+ event="pull_request",
+ path=".github/workflows/ci-amd.yml",
+ conclusion=None,
+):
+ return {
+ "createdAt": created,
+ "conclusion": conclusion,
+ "workflowRun": {
+ "url": f"https://github.com/apache/airflow/actions/runs/{run_id}";,
+ "event": event,
+ "workflow": {"resourcePath": path},
+ },
+ }
+
+
[email protected](
+ ("node", "expected"),
+ [
+ pytest.param(
+ _project_node(1, suites=[_suite(7), _suite(8,
created="2026-11-10T11:00:00Z")]),
+ "https://github.com/apache/airflow/actions/runs/8";,
+ id="latest-run",
+ ),
+ pytest.param(_project_node(1, status="SUCCESS", suites=[_suite(7)]),
None, id="already-reported"),
+ pytest.param(
+ _project_node(1, status="PENDING", suites=[_suite(7)]),
+ "https://github.com/apache/airflow/actions/runs/7";,
+ id="pending-is-replaced",
+ ),
+ pytest.param(_project_node(1), None, id="no-run-yet"),
+ pytest.param(
+ _project_node(
+ 1, suites=[_suite(7, event="push"), _suite(8,
path=".github/workflows/codeql.yml")]
+ ),
+ None,
+ id="other-runs-only",
+ ),
+ pytest.param(
+ _project_node(1, suites=[_suite(7, conclusion="ACTION_REQUIRED")]),
+ None,
+ id="waiting-for-approval",
+ ),
+ pytest.param(_project_node(1, suites=[_suite(7)], oid="other"), None,
id="not-the-head"),
+ ],
+)
+def test_find_project_ci_run_url(node, expected):
+ assert find_project_ci_run_url(node) == expected
+
+
[email protected](
+ ("statuses", "runs", "expected"),
+ [
+ pytest.param(
+ [],
+ [{"html_url": "https://github.com/apache/airflow/actions/runs/9";,
"conclusion": None}],
+ "https://github.com/apache/airflow/actions/runs/9";,
+ id="run-found",
+ ),
+ pytest.param(
+ [{"context": CI_RUN_STARTED_CONTEXT, "state": "success"}],
+ [{"html_url": "https://github.com/apache/airflow/actions/runs/9";,
"conclusion": None}],
+ None,
+ id="already-reported",
+ ),
+ pytest.param(
+ [],
+ [
+ {
+ "html_url":
"https://github.com/apache/airflow/actions/runs/9";,
+ "conclusion": "action_required",
+ }
+ ],
+ None,
+ id="waiting-for-approval",
+ ),
+ pytest.param([], [], None, id="no-run-yet"),
+ ],
+)
+def test_read_project_ci_run_url(client, statuses, runs, expected):
+ def rest_get(path, paginate=False, items_key=None):
+ if path == f"repos/{REPO}/commits/sha1/status?per_page=100":
+ assert (paginate, items_key) == (True, "statuses")
+ return statuses
+ assert (
+ path
+ ==
f"repos/{REPO}/actions/workflows/ci-amd.yml/runs?event=pull_request&head_sha=sha1&per_page=1"
+ )
+ return {"workflow_runs": runs}
+
+ client.rest_get.side_effect = rest_get
+
+ assert read_project_ci_run_url(client, REPO, "sha1") == expected
+
+
+def
test_report_project_ci_runs_started_reads_a_head_commit_the_search_did_not_list(client):
+ client.graphql.return_value = _search_result(_project_node(1,
suites=[_suite(7)], oid="older"))
+ client.rest_get.side_effect = [
+ [],
+ {
+ "workflow_runs": [
+ {"html_url":
"https://github.com/apache/airflow/actions/runs/9";, "conclusion": None}
+ ]
+ },
+ ]
+
+ assert report_project_ci_runs_started(client, repo=REPO, skip=set()) == [1]
+ assert client.set_status.call_args.args[:4] == (
+ 1,
+ "sha1",
+ "success",
+ "https://github.com/apache/airflow/actions/runs/9";,
+ )
+
+
+def
test_report_project_ci_runs_started_skips_gated_pull_requests_and_keeps_to_the_budget(client):
+ client.graphql.return_value = _search_result(
+ _project_node(1, suites=[_suite(1)]),
+ _project_node(2, suites=[_suite(2)]),
+ _project_node(3),
+ _project_node(4, suites=[_suite(4)]),
+ _project_node(5, suites=[_suite(5)]),
+ )
+
+ reported = report_project_ci_runs_started(client, repo=REPO, skip={2},
budget=2)
+
+ assert reported == [1, 4]
+ assert client.set_status.call_args_list == [
+ mock.call(
+ number,
+ f"sha{number}",
+ "success",
+ f"https://github.com/apache/airflow/actions/runs/{number}";,
+ CI_RUN_STARTED_PROJECT_DESCRIPTION,
+ context=CI_RUN_STARTED_CONTEXT,
+ )
+ for number in (1, 4)
+ ]
+ assert client.graphql.call_args.args[1]["q"] == (
+ f'repo:{REPO} is:pr is:open -label:"fork ci required"
sort:updated-desc'
+ )
+
+
def test_discover_labels_pilot_pull_requests_while_disabled(client):
config = GateConfig(enabled=False, pilot={"pilot-user": datetime(2026, 11,
1, tzinfo=UTC)})
new = {**_pr_node(1), "author": {"login": "pilot-user"}, "labels":
{"nodes": []}}
@@ -1387,12 +1702,37 @@ def
test_discover_labels_pilot_pull_requests_while_disabled(client):
other = {**_pr_node(3), "labels": {"nodes": []}}
client.graphql.side_effect = [_search_result(new, old, other),
_search_result()]
+ client.rest_get.return_value = {"workflow_runs": []}
configs = BaseGateConfigs(client, repo=REPO, default_branch="main",
default_config=config)
labelled = discover(client, configs, repo=REPO, now=NOW)
assert labelled == [1]
client.add_labels.assert_called_once_with(1, ["fork ci required"])
+ client.rest_get.assert_called_once_with(
+
f"repos/{REPO}/actions/workflows/ci-amd.yml/runs?event=pull_request&branch=branch-2"
+ "&head_sha=sha2&created=%3E%3D2026-11-01T00%3A00%3A00Z&per_page=100"
+ )
+
+
+def
test_discover_labels_an_older_pilot_pull_request_after_a_run_since_joining(client):
+ config = GateConfig(enabled=False, pilot={"pilot-user": datetime(2026, 11,
1, tzinfo=UTC)})
+ old = {
+ **_pr_node(2),
+ "author": {"login": "pilot-user"},
+ "createdAt": "2026-10-20T00:00:00Z",
+ "labels": {"nodes": []},
+ }
+ client.graphql.side_effect = [_search_result(old), _search_result()]
+ client.rest_get.return_value = {
+ "workflow_runs": [_branch_run("2026-11-02T10:00:00Z",
"someone/my-airflow")]
+ }
+ configs = BaseGateConfigs(client, repo=REPO, default_branch="main",
default_config=config)
+
+ labelled = discover(client, configs, repo=REPO, now=NOW)
+
+ assert labelled == [2]
+ client.add_labels.assert_called_once_with(2, ["fork ci required"])
def _contents(files: dict[tuple[str, str], str]):
@@ -1586,6 +1926,7 @@ def
test_run_reconcile_isolates_failures_and_reports_them(client, tmp_path):
_search_result(),
_search_result(_pr_node(1), _pr_node(2, isDraft=False)),
{"pr0": {"object": None}, "pr1": {"object": None}},
+ _search_result(),
]
workflow_state_calls = []
client.rest_get_pages.return_value = [{"total_count": 0, "workflow_runs":
[]}]
@@ -1640,6 +1981,7 @@ def
test_run_reconcile_rereads_workflow_state_while_awaiting_push(client, tmp_pa
_search_result(),
_search_result(_pr_node(1, comments={"nodes": [marker_comment]})),
{"pr0": {"object": None}},
+ _search_result(),
]
client.rest_get_pages.return_value = [{"total_count": 0, "workflow_runs":
[]}]
@@ -1684,6 +2026,7 @@ def test_run_reconcile_deletes_gated_runs(mock_classify,
client, tmp_path, state
_pr_node(1, headRefName="my-change",
headRepository={"nameWithOwner": "someone/airflow"})
),
{"pr0": {"object": None}},
+ _search_result(),
]
client.rest_get_pages.side_effect = lambda path: [
{"total_count": 1, "workflow_runs": [_run(7, minutes=1)] if
"event=pull_request&" in path else []}
@@ -1749,7 +2092,7 @@ def test_command_writes_has_alerts_output(mock_run,
tmp_path, monkeypatch):
)
def test_run_reconcile_without_gated_pull_requests(client, tmp_path,
total_count, has_alerts):
output = tmp_path / "slack-message.json"
- client.graphql.side_effect = [_search_result(), _search_result(),
_search_result()]
+ client.graphql.side_effect = [_search_result(), _search_result(),
_search_result(), _search_result()]
client.rest_get_pages.return_value = [{"total_count": total_count,
"workflow_runs": []}]
assert (
@@ -1833,7 +2176,12 @@ def
test_drop_reported_reports_everything_without_a_readable_previous_run(tmp_pa
reported_file.write_text(content)
assert drop_reported(reported_file, unknown=[(1, "a")], attention=[],
notes=[]) == ([(1, "a")], [], [])
- assert json.loads(reported_file.read_text()) == {"unknown": [[1, "a"]],
"attention": [], "notes": []}
+ assert json.loads(reported_file.read_text()) == {
+ "unknown": [[1, "a"]],
+ "attention": [],
+ "notes": [],
+ "deferred_runs": 0,
+ }
@pytest.mark.parametrize(
@@ -1849,7 +2197,7 @@ def
test_run_reconcile_posts_a_persisting_item_once(client, tmp_path, first_dry_
def reconcile(dry_run=False):
client.dry_run = dry_run
- client.graphql.side_effect = [_search_result(), _search_result(),
_search_result()]
+ client.graphql.side_effect = [_search_result(), _search_result(),
_search_result(), _search_result()]
return run_reconcile(
client,
CONFIG,
@@ -1881,6 +2229,54 @@ def test_slack_payload_stays_within_slack_limits():
assert "more sections are in the <https://run|run log>" in texts[-1]
[email protected](
+ ("fork", "ci_run_started_state", "expected"),
+ [
+ pytest.param(GREEN, None, ("success", RUN_URL,
CI_RUN_STARTED_DESCRIPTION), id="green"),
+ pytest.param(
+ ForkObservation(suite_status="IN_PROGRESS", run_url=RUN_URL),
+ "pending",
+ ("success", RUN_URL, CI_RUN_STARTED_DESCRIPTION),
+ id="running",
+ ),
+ pytest.param(GREEN, "success", None, id="already-reported"),
+ pytest.param(ForkObservation(workflow_state="active"), None, None,
id="no-fork-run"),
+ ],
+)
+def test_ci_run_started_is_reported_once_the_fork_runs_the_head_commit(fork,
ci_run_started_state, expected):
+ plan, _, _ = _classify(_pr(ci_run_started_state=ci_run_started_state),
fork)
+
+ assert plan.ci_run_started == expected
+
+
+def test_apply_plan_posts_ci_run_started_under_its_own_context(client):
+ plan, _, _ = _classify(_pr(), GREEN)
+
+ apply_plan(client, _pr(), plan)
+
+ assert (
+ mock.call(
+ 42, "abc123", "success", RUN_URL, CI_RUN_STARTED_DESCRIPTION,
context=CI_RUN_STARTED_CONTEXT
+ )
+ in client.set_status.call_args_list
+ )
+
+
[email protected](
+ ("ci_run_at", "checked"),
+ [
+ pytest.param(CREATED + timedelta(days=1), True,
id="run-since-joining"),
+ pytest.param(None, False, id="no-run-since-joining"),
+ ],
+)
+def
test_older_pilot_pull_request_is_checked_for_tampering_after_a_run_since_joining(ci_run_at,
checked):
+ config = dataclasses.replace(CONFIG, enabled=False, pilot={"someone":
CREATED + timedelta(hours=1)})
+
+ _, find_bypass, _ = _classify(_pr(ci_run_at=ci_run_at), GREEN,
config=config)
+
+ assert find_bypass.called is checked
+
+
def test_apply_plan_posts_ci_passed_under_its_own_context(client):
plan, _, _ = _classify(_pr(), GREEN)
@@ -2072,6 +2468,7 @@ def
test_run_reconcile_reports_job_statuses_before_classifying(
_search_result(),
_search_result(_pr_node(1, headRepository={"nameWithOwner":
"someone/airflow"})),
{"pr0": {"object": None}},
+ _search_result(),
]
client.rest_get_pages.return_value = [{"total_count": 0, "workflow_runs":
[]}]
client.rest_get.return_value = {"state": "active"}
@@ -2148,12 +2545,13 @@ def
test_job_status_reporter_charges_each_write_even_when_a_later_one_fails(clie
assert reporter.remaining == 9
-def _reconcile_with_reporter(client, tmp_path, config=CONFIG):
+def _reconcile_with_reporter(client, tmp_path, config=CONFIG,
reported_file=None):
client.graphql.side_effect = [
_search_result(),
_search_result(),
_search_result(_pr_node(1, headRepository={"nameWithOwner":
"someone/airflow"})),
{"pr0": {"object": None}},
+ _search_result(),
]
client.rest_get_pages.return_value = [{"total_count": 0, "workflow_runs":
[]}]
client.rest_get.return_value = {"state": "active"}
@@ -2164,6 +2562,7 @@ def _reconcile_with_reporter(client, tmp_path,
config=CONFIG):
run_url="https://run";,
output_file=tmp_path / "s.json",
workflows_dir=tmp_path,
+ reported_file=reported_file,
)
@@ -2218,6 +2617,26 @@ def
test_run_reconcile_reports_deferred_job_statuses_to_slack(
assert JOB_STATUSES_DEFERRED_NOTE in (tmp_path / "s.json").read_text()
[email protected]("airflow_breeze.utils.fork_ci_reconcile.JobStatusReporter",
autospec=True)
[email protected]("airflow_breeze.utils.fork_ci_reconcile.classify", autospec=True)
+def
test_run_reconcile_reports_deferred_job_statuses_only_when_deferring_persists(
+ mock_classify, mock_reporter, client, tmp_path
+):
+ mock_classify.return_value = Plan(state=ForkState.RUNNING)
+ mock_reporter.return_value.report.return_value = None
+ client.dry_run = False
+ reported_file = tmp_path / "reported.json"
+ reported_file.write_text("{}")
+ alerts = []
+ for deferred in ([1], [1], [], [1], [1], [1]):
+ mock_reporter.return_value.deferred = deferred
+ alerts.append(_reconcile_with_reporter(client, tmp_path,
reported_file=reported_file))
+
+ assert alerts == [False, False, False, False, False, True]
+ assert JOB_STATUSES_DEFERRED_NOTE in (tmp_path / "s.json").read_text()
+ assert json.loads(reported_file.read_text())["deferred_runs"] == 3
+
+
def test_build_job_statuses_treats_contexts_case_insensitively():
statuses = build_job_statuses(
[_fork_job("Tests", conclusion="failure", url="bad"),
_fork_job("tests", url="ok")]
@@ -2281,7 +2700,7 @@ def
test_finder_ignores_an_added_workflow_whose_jobs_did_not_run(client, jobs):
@mock.patch("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()]
+ client.graphql.side_effect = [_search_result(), _search_result(),
_search_result(), _search_result()]
mock_finder.return_value.incomplete = None
mock_finder.return_value.ambiguous = [(42, "could also belong to #43")]
@@ -2308,7 +2727,7 @@ def
test_run_reconcile_reports_ambiguous_runs(mock_finder, client, tmp_path):
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()]
+ client.graphql.side_effect = [_search_result(), _search_result(),
_search_result(), _search_result()]
mock_finder.return_value.incomplete = None
mock_finder.return_value.ambiguous = []
run_reconcile(
@@ -2698,6 +3117,7 @@ def
test_run_reconcile_retries_a_pull_request_whose_head_moved(client, tmp_path)
_search_result(),
_search_result(_pr_node(1)),
{"pr0": {"object": None}},
+ _search_result(),
]
client.rest_get_pages.return_value = [{"total_count": 0, "workflow_runs":
[]}]
@@ -2719,3 +3139,84 @@ def
test_run_reconcile_retries_a_pull_request_whose_head_moved(client, tmp_path)
assert has_alerts is False
client.set_status.assert_not_called()
client.create_comment.assert_not_called()
+
+
[email protected](
+ ("fails", "job_status_budget"),
+ [pytest.param(False, 58, id="reported"), pytest.param(True, 30,
id="fails")],
+)
[email protected]("airflow_breeze.utils.fork_ci_reconcile.JobStatusReporter",
autospec=True)
[email protected]("airflow_breeze.utils.fork_ci_reconcile.report_project_ci_runs_started",
autospec=True)
[email protected]("airflow_breeze.utils.fork_ci_reconcile.classify", autospec=True)
+def
test_run_reconcile_marks_older_pilot_pull_requests_and_reports_project_ci_runs(
+ mock_classify, mock_report, mock_reporter, client, tmp_path, fails,
job_status_budget
+):
+ config = dataclasses.replace(CONFIG, pilot={"someone": datetime(2026, 11,
2, tzinfo=UTC)})
+ mock_classify.return_value = Plan(state=ForkState.AWAITING_PUSH)
+ mock_report.side_effect = GitHubError("gh api failed: HTTP 502") if fails
else None
+ mock_report.return_value = [7, 8]
+ mock_reporter.return_value.report.return_value = None
+ mock_reporter.return_value.deferred = []
+ 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"}
+ raise AssertionError(path)
+
+ client.rest_get.side_effect = rest_get
+ output = tmp_path / "s.json"
+
+ has_alerts = run_reconcile(
+ client, config, now=NOW, run_url="https://run";, output_file=output,
workflows_dir=tmp_path
+ )
+
+ assert mock_classify.call_args.args[0].ci_run_at == datetime(2026, 11, 2,
tzinfo=UTC)
+ mock_report.assert_called_once_with(client, repo=REPO, skip={1})
+ mock_reporter.assert_called_once_with(client, repo=REPO,
budget=job_status_budget)
+ assert has_alerts is fails
+ if fails:
+ assert "reporting `CI run started` for project CI failed: GitHubError"
in output.read_text()
+
+
[email protected](
+ ("config_overrides", "author", "expected_ci_run_at",
"expected_pilot_joined_at"),
+ [
+ pytest.param(
+ {"enabled": False}, "pilot-user", JOINED, JOINED,
id="project-ci-allowed-before-joining"
+ ),
+ pytest.param({}, "pilot-user", JOINED, None,
id="already-gated-as-external"),
+ pytest.param({"enabled": False}, "someone", None, None,
id="not-a-pilot"),
+ pytest.param(
+ {"enabled": False, "pilot": {"pilot-user": NOW +
timedelta(days=1)}},
+ "pilot-user",
+ None,
+ None,
+ id="joining-later",
+ ),
+ ],
+)
+def test_mark_older_pilot_pull_request(
+ config_overrides, author, expected_ci_run_at, expected_pilot_joined_at
+):
+ config = dataclasses.replace(CONFIG, **{"pilot": {"pilot-user": JOINED},
**config_overrides})
+
+ marked = mark_older_pilot_pull_request(_pr(author_login=author), config,
repo=REPO, now=NOW)
+
+ assert (marked.ci_run_at, marked.pilot_joined_at) == (expected_ci_run_at,
expected_pilot_joined_at)
+
+
+def
test_older_pilot_pull_request_stays_checked_for_tampering_after_its_gated_run_is_deleted():
+ config = dataclasses.replace(CONFIG, enabled=False, pilot={"someone":
JOINED})
+
+ _, find_bypass, _ = _classify(
+ mark_older_pilot_pull_request(_pr(), config, repo=REPO, now=NOW),
GREEN, config=config
+ )
+
+ assert find_bypass.called
diff --git a/dev/breeze/tests/test_workflow_run_approval.py
b/dev/breeze/tests/test_workflow_run_approval.py
index 408aeca15f9..6aa5d85d5ee 100644
--- a/dev/breeze/tests/test_workflow_run_approval.py
+++ b/dev/breeze/tests/test_workflow_run_approval.py
@@ -1308,3 +1308,65 @@ def
test_process_waiting_runs_clears_a_failed_pull_request_whose_runs_were_appro
api.is_run_waiting_for_approval.assert_called_once_with(1)
assert api.set_approval_status.call_args_list[-1] == mock.call("aaa",
"success", "Workflow runs approved")
api.remove_label.assert_called_once_with(10, NEED_WORKFLOW_APPROVAL_LABEL)
+
+
[email protected](f"{MODULE}.decide_for_pull_request", autospec=True)
+def
test_process_waiting_runs_does_not_approve_runs_before_the_decisions_cutoff(mock_decide):
+ api = mock.MagicMock(spec=GitHubApi)
+ api.github_repository = "apache/airflow"
+ api.get_runs_waiting_for_approval.return_value = [
+ _run(2, "aaa", created_at="2026-10-22T10:00:00Z"),
+ _run(3, "aaa", created_at="2026-10-22T11:00:00Z"),
+ ]
+ api.get_open_pull_requests.return_value = [_pr(10, "aaa")]
+ api.get_last_base_change.return_value = None
+ api.get_last_label_change.return_value = None
+ mock_decide.return_value = ApprovalDecision(
+ approve=True, reasons=["gated"], not_before=datetime(2026, 10, 22, 10,
30, tzinfo=UTC)
+ )
+
+ process_waiting_runs(api, created_since=CREATED_SINCE)
+
+ assert api.approve_run.call_args_list == [mock.call(3)]
+
+
[email protected](
+ ("joined_at", "gated", "not_before"),
+ [
+ pytest.param(
+ datetime(2026, 10, 23, tzinfo=UTC),
+ True,
+ datetime(2026, 10, 23, tzinfo=UTC),
+ id="pilot-joined-after-opening",
+ ),
+ pytest.param(datetime(2026, 10, 21, tzinfo=UTC), True, None,
id="pilot-joined-before-opening"),
+ pytest.param(datetime(2026, 10, 31, tzinfo=UTC), False, None,
id="pilot-joining-later"),
+ ],
+)
[email protected](f"{MODULE}.decide_approval", autospec=True)
[email protected](f"{MODULE}.build_selective_checks", autospec=True)
[email protected](f"{MODULE}.load_config", autospec=True)
[email protected](f"{MODULE}.get_changed_files", autospec=True,
return_value=LARGE_SOURCE_CHANGE)
[email protected](f"{MODULE}.fetch_merge_commit", autospec=True,
return_value="merge")
+def
test_decide_for_pull_request_gates_an_older_pilot_pull_request_from_joining(
+ mock_fetch,
+ mock_get_changed_files,
+ mock_load_config,
+ mock_selective_checks,
+ mock_decide_approval,
+ joined_at,
+ gated,
+ not_before,
+):
+ mock_load_config.return_value = GateConfig(pilot={"contributor":
joined_at})
+ api = mock.MagicMock(spec=GitHubApi)
+ api.github_repository = "apache/airflow"
+
+ decision = decide_for_pull_request(api=api, pr=_gated_pr(),
now=datetime(2026, 10, 30, tzinfo=UTC))
+
+ if gated:
+ assert decision.approve is True
+ assert decision.not_before == not_before
+ mock_decide_approval.assert_not_called()
+ else:
+ assert decision is mock_decide_approval.return_value
diff --git a/scripts/ci/prek/check_ci_workflows_in_sync.py
b/scripts/ci/prek/check_ci_workflows_in_sync.py
index 15edf610d7f..d9676a81ade 100755
--- a/scripts/ci/prek/check_ci_workflows_in_sync.py
+++ b/scripts/ci/prek/check_ci_workflows_in_sync.py
@@ -168,6 +168,10 @@ AMD_ONLY_GATE_BLOCK = """ gate:
name: "Fork CI gate"
runs-on: ["ubuntu-slim"]
timeout-minutes: 5
+ permissions:
+ contents: read
+ # Reads the creation time of this run, which decides for a pilot's older
pull request.
+ actions: read
outputs:
project-ci: ${{ steps.gate.outputs.project-ci }}
reason: ${{ steps.gate.outputs.reason }}
