Add REFRESH_DURATIONS mode to internal pipeline for pytest-split duration refresh (#22765)

## Summary & Motivation

Stacked on #22715 (pytest-split adoption). That PR introduces splitting
via `pytest-split` but, without `.test_durations*` files, splitting
falls back to count-based balancing. This PR wires up the machinery to
produce and consume per-suite duration data.

Rather than maintain a separate Buildkite pipeline whose suite list
duplicates the split config in `python_packages.py` / OSS `packages.py`,
refresh runs as a **mode** of the existing `internal` pipeline. The same
split config drives both normal sharded CI and the periodic refresh — no
separate list to keep in sync. Refresh is triggered manually (no
schedule); a Buildkite block step gates the final PR-creation step so a
human can inspect the collected artifacts before they're committed.

## How it works

**`REFRESH_DURATIONS=1` flag.** A new `BuildConfig.refresh_durations`
boolean lives in `buildkite-shared/context.py` and is settable via the
`REFRESH_DURATIONS` env var or `[REFRESH_DURATIONS]` magic string in the
build message.

**Step generation diverges in `PackageSpec.build_steps`.** For each tox
factor:

- **`splits > 1` in refresh mode:** collapse the parallel fan-out to a
single un-sharded step with `--store-durations --durations-path
.test_durations-<factor>[-<label_suffix>]`. Soft-fails so one broken
suite doesn't lose the rest. A `command_wrapper` does `set +e; <tox>;
status=$?; cd "$BUILDKITE_BUILD_CHECKOUT_PATH"; buildkite-agent artifact
upload "<dir>/.test_durations-..."; exit $status` — disables `set -e` so
the upload always runs, and `cd`s back to repo root so the artifact is
stored with its full repo-relative path (basenames would collide across
packages).
- **`splits <= 1` in refresh mode:** emit a normal step (which the
top-level filter drops). Not emitting nothing keeps OSS helpers like
`build_helm_steps` (`assert len(pkg_steps) == 1`) working.
- **Normal mode:** unchanged behavior plus `--durations-path
.test_durations-<factor>[-<label_suffix>]` so pytest-split reads exactly
the file refresh writes. Naming includes `label_suffix` to keep OSS
dagster-dbt's four `<deps>-core-main` ToxFactors (rest / test_resource /
test_asset_checks / cli) from clobbering each other's files.

**Top-level transform via `apply_transforms`.**
`InternalBuildkiteContext.apply_transforms` is overridden to run
`refresh_durations.apply` before delegating to the shared transform. The
transform:

1. Drops every step that doesn't carry `--store-durations` (i.e. lint,
kaniko, helm, dogfood, etc. — emitted normally during pipeline
construction, pruned here).
2. Scrubs dangling `depends_on` references on surviving steps (e.g. OSS
k8s-test-suite refresh leaves carry `depends_on: [test-project-3-12]`
pointing at an image-build step that was just filtered out — without
scrubbing, BK rejects the pipeline upload).
3. Appends a `BlockStepBuilder("approve-test-durations-pr")` step
depending on every refresh tox step's key, and a
`create-test-durations-pr` step depending on the block. The PR step runs
only after a reviewer clicks unblock.

In refresh mode the buildevents / AI-assessment steps get stripped too
(they don't carry `--store-durations`); acceptable for a manual refresh.

**`inject_flakes_clones` opt-out.** The quarantine flake-clone injector
early-returns in refresh mode — clones would inherit the
`--store-durations` flag, survive the filter, and race the original on
the same artifact path with `RUN_QUARANTINED=1`'s different test set.

**`get_python_package_step_skip_reason` left unchanged.** Refresh-mode
bypass of change-detection happens inside `_append_steps_for_cell`
(`skip_reason=None` only for the refresh-step emit), so external callers
like `skip_release_pipeline` and `_get_app_cloud_step_skip_reason` keep
their normal semantics.

**Aggregation step.** `scripts/buildkite/create_test_durations_pr.sh`:

- Downloads `**/.test_durations*` artifacts; treats "no artifacts" as a
clean no-op.
- Validates each file is non-empty + parseable JSON; skips invalid files
so a half-written durations file from a crashed tox run never lands in
the repo.
- Uses `find -print0 | xargs -0 git add` so any future
whitespace-in-path doesn't split.
- Pushes via `GIT_ASKPASS` helper with `GIT_TERMINAL_PROMPT=0`, so
`ELEMENTL_DEVTOOLS_PAT` never appears in argv or git's error output.
- Opens the PR via the GitHub REST API (slim image has no `gh` CLI).
Retries on HTTP `000`/`5xx`/`429`; treats `422` (PR already exists for
this branch) as success.

**Internal-pipeline conditional carves out
`auto/refresh-test-durations/*`.** The existing `(build.author.name !=
"elementl-devtools") && ...` filter would otherwise skip CI on the
auto-PR (it's authored by elementl-devtools). The carve-out runs
internal CI on the durations PR so the new files get validated by the
pipeline that consumes them before merge.

## Changes

- `dagster-oss/.buildkite/buildkite-shared/buildkite_shared/context.py`
— `BuildConfig.refresh_durations` field.
- `dagster-oss/.buildkite/buildkite-shared/buildkite_shared/packages.py`
— refresh-mode branch in `_append_steps_for_cell` + extracted
`_append_refresh_step` helper; `--durations-path` added to normal split
mode.
-
`.buildkite/dagster-internal-buildkite/dagster_internal_buildkite/context.py`
— `InternalBuildkiteContext.apply_transforms` override calling
`refresh_durations.apply` before `super`.
-
`.buildkite/dagster-internal-buildkite/dagster_internal_buildkite/steps/refresh_durations.py`
— filter + scrub-deps + block + PR step builder.
-
`.buildkite/dagster-internal-buildkite/dagster_internal_buildkite/steps/quarantine.py`
— early-return in refresh mode.
- `infra/terragrunt/modules/buildkite-pipelines/pipelines.tf` —
`auto/refresh-test-durations/*` carve-out on the internal pipeline's
author conditional.
- `scripts/buildkite/create_test_durations_pr.sh` — REST-API PR creation
+ HTTPS push with PAT via `GIT_ASKPASS` + JSON validation + retries.

### Removed (superseded by this PR's earlier iteration)

- The previously-added dedicated `refresh-test-durations` pipeline:
`pipelines/refresh_test_durations.py`, its `cli.py` entry, the
`pyproject.toml` script, the Terragrunt module, and the steps YAML — all
replaced by the in-pipeline mode above.

## Test Plan

- [ ] Merge #22715 first; this PR is a no-op on its own for normal CI
runs (files don't exist yet, pytest-split falls back to count-based
balancing).
- [ ] After merge, kick off a manual `internal` build with
`REFRESH_DURATIONS=1` in the build env (or include `[REFRESH_DURATIONS]`
in a commit message on a feature branch) and confirm:
- [ ] Only split tox suites run, each as one un-sharded job with
`--store-durations`.
- [ ] After the suites finish, the `approve-test-durations-pr` block
step appears for manual unblock.
- [ ] Clicking unblock runs `create-test-durations-pr`, which downloads
artifacts and opens an `auto/refresh-test-durations-<date>-<build>` PR.
- [ ] That PR's internal-pipeline CI build runs (carve-out works) and
the `.test_durations-*` files balance subsequent normal-mode shard
distributions after merge.

---------

Co-authored-by: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
Internal-RevId: 79790d5041821dd5ab3d083fd549eabb6574dce4

Author: 2 people

.buildkite/buildkite-shared/buildkite_shared/context.py

@@ -331,13 +331,21 @@ class BuildConfig:
     # when combined with the STEP_FILTER directive to run a subset of steps
     # multiple times, for example to surface flaky tests.
     repeat: int
+    # Collect pytest-split timing data instead of running tests normally. When
+    # set, split tox suites are emitted as a single un-sharded step with
+    # `--store-durations`, the resulting `.test_durations*` file is uploaded as
+    # a Buildkite artifact, and all other pipeline steps are pruned. The
+    # internal pipeline appends a final aggregation step that opens a PR with
+    # the collected files.
+    refresh_durations: bool
 
     @classmethod
     def from_raw(cls, raw: Mapping[str, str]) -> Self:
         return cls(
             no_skip=bool(raw.get("no_skip")),
             step_filter=raw.get("step_filter"),
             repeat=int(raw.get("repeat", "1")),
+            refresh_durations=bool(raw.get("refresh_durations")),
         )
 
     @classmethod

.buildkite/buildkite-shared/buildkite_shared/packages.py

@@ -361,14 +361,65 @@ def _append_steps_for_cell(
         label_suffix = f" {factor.label_suffix}" if factor and factor.label_suffix else ""
         resolved_image = factor.image if factor and factor.image else self.image
 
+        # Per-factor durations file under a single `.test_durations/` directory
+        # at the package root. pytest-split reads the file in normal mode and
+        # writes it via --store-durations in refresh mode. The nested layout
+        # keeps multi-factor packages from cluttering their root with one
+        # `.test_durations-<factor>` file per tox factor. The unfactored case
+        # uses the literal name `default` so the layout is uniform (directory
+        # everywhere, no mix of file-or-directory). The id includes
+        # label_suffix so two ToxFactors sharing a tox env but selecting
+        # different test subsets (e.g. OSS dagster-dbt's `<deps>-core-main` x
+        # [rest, test_resource, test_asset_checks, cli]) don't clobber each
+        # other in refresh mode.
+        if factor:
+            durations_id = factor.factor
+            if factor.label_suffix:
+                durations_id = f"{durations_id}-{factor.label_suffix}"
+        else:
+            durations_id = "default"
+        durations_file = f".test_durations/{durations_id}"
+
+        # Refresh mode: for factors that fan out, collapse to a single
+        # un-sharded run that writes a fresh durations file and uploads it
+        # as an artifact. Non-split factors fall through to the normal
+        # emit path — the caller's downstream filter
+        # (`steps/refresh_durations.py:keep_refresh_steps_only`) drops them
+        # by absence of `--store-durations` in the command. We don't return
+        # an empty list here because some callers (e.g. OSS
+        # `build_helm_steps`) assert on `len(pkg_steps) == 1`.
+        if ctx.config.refresh_durations and splits > 1:
+            # Refresh mode runs the full set of split tox suites unconditionally —
+            # change-detection isn't meaningful when collecting timing data
+            # against the current source tree, not validating a diff. Force the
+            # step to emit by suppressing skip_reason locally; we don't change
+            # `get_python_package_step_skip_reason` itself so external callers
+            # (release-pipeline gating, app-cloud gating) keep their semantics.
+            self._append_refresh_step(
+                tox_env=tox_env,
+                base_name=base_name,
+                label_suffix=label_suffix,
+                durations_file=durations_file,
+                factor_pytest_args=factor_pytest_args,
+                resolved_image=resolved_image,
+                py_version=py_version,
+                extra_pre=extra_pre,
+                extra_post_cd=extra_post_cd,
+                dependencies=dependencies,
+                skip_reason=None,
+                factor=factor,
+                out=out,
+            )
+            return
+
         # Splitting uses the pytest-split plugin. To use `splits > 1` with a
         # package, `pytest-split` must be declared in that package's test
         # dependencies (e.g. the `test`/`tests` extra in its pyproject.toml).
         for split_index in range(1, splits + 1):
             if splits > 1:
                 split_label = f"{base_name}{label_suffix} ({split_index}/{splits})"
                 pytest_args: list[str] | None = [
-                    f"--splits {splits} --group {split_index}",
+                    f"--splits {splits} --group {split_index} --durations-path {durations_file}",
                     *factor_pytest_args,
                 ]
             else:
@@ -409,6 +460,91 @@ def _append_steps_for_cell(
                 )
             )
 
+    def _append_refresh_step(
+        self,
+        *,
+        tox_env: str,
+        base_name: str,
+        label_suffix: str,
+        durations_file: str,
+        factor_pytest_args: list[str],
+        resolved_image: ToxImage,
+        py_version: AvailablePythonVersion,
+        extra_pre: list[str],
+        extra_post_cd: list[str] | None,
+        dependencies: list[str] | None,
+        skip_reason: str | None,
+        factor: ToxFactor | None,
+        out: list[GroupLeafStepConfiguration],
+    ) -> None:
+        split_label = f"{base_name}{label_suffix}"
+        key = slugify_label(f"{split_label} {_tox_env_to_label_suffix(tox_env)}")
+        pytest_args = [
+            f"--store-durations --durations-path {durations_file}",
+            *factor_pytest_args,
+        ]
+        # The tox command runs after `cd {directory}` so cwd is the package
+        # directory. We `cd` back to the build checkout root before the upload
+        # so the artifact is stored with its full repo-relative path —
+        # otherwise multiple packages sharing a factor name (e.g. `-default`)
+        # would all upload to the same basename and overwrite each other in
+        # BK's artifact store. pytest-split's --store-durations is a
+        # `pytest_sessionfinish` hook, so only a clean tox exit produces the
+        # file; the upload runs unconditionally (see _refresh_wrapper) and
+        # `buildkite-agent` exits non-zero when the file is missing, which is
+        # the signal we want — the create-test-durations-pr step then sees no
+        # artifact for that suite.
+        artifact_path = f"{self.directory}/{durations_file}"
+        upload_cmd = f'buildkite-agent artifact upload "{artifact_path}"'
+        original_wrapper = self.command_wrapper
+
+        def _refresh_wrapper(cmd: str) -> str:
+            # Disable `set -e` around the block so a failing tox doesn't abort
+            # before the upload runs (BK's `bash -e -c` would otherwise exit
+            # at the failing subshell). Preserve the inner exit code via
+            # $status so `soft_fail` still surfaces real test failures in BK
+            # UI; the upload's own exit code is intentionally ignored.
+            # `mkdir -p .test_durations` ensures the parent of `durations_file`
+            # exists before pytest-split's --store-durations writes — pytest-
+            # split uses a plain `open(path, "w")` and won't create parents.
+            inner = original_wrapper(cmd) if original_wrapper else cmd
+            return (
+                f"mkdir -p .test_durations; set +e; {inner}; status=$?;"
+                f' cd "$BUILDKITE_BUILD_CHECKOUT_PATH"; {upload_cmd}; exit $status'
+            )
+
+        out.append(
+            build_tox_step(
+                self.directory,
+                tox_env,
+                key=key,
+                label_emojis=self.label_emojis or [self._resolve_emoji()],
+                timeout_in_minutes=self.timeout_in_minutes,
+                tox_file=self.tox_file,
+                extra_commands_pre=extra_pre,
+                extra_commands_post_cd=extra_post_cd,
+                env=self.env_vars,
+                image=resolved_image,
+                python_version=py_version if resolved_image == "test" else None,
+                ecr_account_ids=self.ecr_account_ids,
+                queue=(factor.queue if factor and factor.queue else self.queue),
+                depends_on=dependencies,
+                skip_reason=skip_reason,
+                pytest_args=pytest_args,
+                section_header=self._section_header(tox_env),
+                concurrency=factor.concurrency if factor else None,
+                concurrency_group=(factor.concurrency_group if factor else None),
+                resources=(
+                    factor.resources if factor and factor.resources is not None else self.resources
+                ),
+                soft_fail=True,
+                with_docker=self.with_docker,
+                ecr_passthru=self.ecr_passthru,
+                command_wrapper=_refresh_wrapper,
+                mutator=self.mutator,
+            )
+        )
+
 
 def build_steps_from_package_specs(
     package_specs: Sequence[PackageSpec],