Show fork CI progress sooner and gate rebased pilot pull requests (#76148)

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.

Generated-by: Claude Opus 5

.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:

.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 }}

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

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

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.

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

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"

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

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:

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"],

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")
+
+
+@mock.patch("airflow_breeze.utils.fork_ci_gate.subprocess.run", autospec=True)
+@pytest.mark.parametrize(
+    ("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
+
+
+@mock.patch("airflow_breeze.utils.fork_ci_gate.subprocess.run", autospec=True)
+@pytest.mark.parametrize(
+    "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()
+
+
+@mock.patch("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()
+
+
+@pytest.mark.parametrize(
+    ("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

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"
 
 
+@pytest.mark.parametrize(
+    ("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"),
+]
+
+
+@pytest.mark.parametrize(
+    ("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
+    )
+
+
+@pytest.mark.parametrize(
+    ("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]
+
+
+@pytest.mark.parametrize(
+    ("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},
+        },
+    }
+
+
+@pytest.mark.parametrize(
+    ("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
+
+
+@pytest.mark.parametrize(
+    ("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]
 
 
+@pytest.mark.parametrize(
+    ("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
+    )
+
+
+@pytest.mark.parametrize(
+    ("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()
 
 
+@mock.patch("airflow_breeze.utils.fork_ci_reconcile.JobStatusReporter", autospec=True)
+@mock.patch("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()
+
+
+@pytest.mark.parametrize(
+    ("fails", "job_status_budget"),
+    [pytest.param(False, 58, id="reported"), pytest.param(True, 30, id="fails")],
+)
+@mock.patch("airflow_breeze.utils.fork_ci_reconcile.JobStatusReporter", autospec=True)
+@mock.patch("airflow_breeze.utils.fork_ci_reconcile.report_project_ci_runs_started", autospec=True)
+@mock.patch("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()
+
+
+@pytest.mark.parametrize(
+    ("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

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)
+
+
+@mock.patch(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)]
+
+
+@pytest.mark.parametrize(
+    ("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"),
+    ],
+)
+@mock.patch(f"{MODULE}.decide_approval", autospec=True)
+@mock.patch(f"{MODULE}.build_selective_checks", autospec=True)
+@mock.patch(f"{MODULE}.load_config", autospec=True)
+@mock.patch(f"{MODULE}.get_changed_files", autospec=True, return_value=LARGE_SOURCE_CHANGE)
+@mock.patch(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

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 }}