diff --git a/changelog/789.trivial.md b/changelog/789.trivial.md
new file mode 100644
index 000000000..6c5c9b460
--- /dev/null
+++ b/changelog/789.trivial.md
@@ -0,0 +1,3 @@
+Consolidated the diagnostic test-case execution path so the provider integration tests re-run each diagnostic and verify its committed regression bundle has not drifted, rather than only asserting the run succeeded.
+The per-provider integration test modules now share a single implementation via `climate_ref.testing.create_no_drift_test`,
+and the unused `run_test_case` pytest fixture was removed (use `TestCaseRunner` directly in custom tests).
diff --git a/docs/api-surface.md b/docs/api-surface.md
index ae5bca382..213a10583 100644
--- a/docs/api-surface.md
+++ b/docs/api-surface.md
@@ -113,7 +113,6 @@ The primary module providers interact with.
 | `TestCase`                 | Class (attrs) | A single test case definition                   |
 | `TestDataSpecification`    | Class (attrs) | Collection of test cases for a diagnostic       |
 | `TestCasePaths`            | Class (attrs) | Path resolver for test case data                |
-| `RegressionValidator`      | Class (attrs) | Validate outputs from stored regression data    |
 | `validate_cmec_bundles`    | Function      | Validate CMEC metric/output bundles             |
 | `collect_test_case_params` | Function      | Collect pytest parametrize params from provider |
 | `load_datasets_from_yaml`  | Function      | Load ExecutionDatasetCollection from YAML       |
diff --git a/docs/how-to-guides/testing-diagnostics.md b/docs/how-to-guides/testing-diagnostics.md
index 11b48743d..a9f1eda36 100644
--- a/docs/how-to-guides/testing-diagnostics.md
+++ b/docs/how-to-guides/testing-diagnostics.md
@@ -334,42 +334,84 @@ ref test-cases run --provider my-provider --diagnostic my-diagnostic --output-di
 ref test-cases run --provider my-provider --diagnostic my-diagnostic --force-regen
 ```
 
-## Writing Pytest Tests
+## Pytest Integration
 
-The CLI provides a standard harness for running simple test cases.
-Sometimes additional custom tests maybe required which require writing tests via `pytest`.
+### The generated per-provider test
 
-### Using the `run_test_case` Fixture
+Every provider package wires its test cases into pytest with a single line in
+`tests/integration/test_diagnostics.py`:
 
-The `run_test_case` fixture automatically handles missing data by skipping tests:
+```python
+from climate_ref_example import provider
+
+from climate_ref.testing import create_no_drift_test
+
+test_run_test_cases = create_no_drift_test(provider)
+```
+
+`create_no_drift_test` generates one parameterized pytest case per test case,
+identified as `{diagnostic}/{test_case}` and marked `slow` and `test_cases`.
+Each case re-executes the diagnostic against the fetched catalog
+using the same execute/build stages as `ref test-cases run`,
+then asserts the freshly rebuilt committed bundle
+matches the tracked `regression/` baseline within tolerance.
+Cases are skipped when the fetched catalog or the committed baseline is missing,
+so the suite stays green for contributors who haven't fetched data for that provider.
+
+Run them with the `--slow` flag:
+
+```bash
+# All test cases for a provider
+uv run pytest packages/climate-ref-example/tests/integration/test_diagnostics.py --slow
+
+# A single test case, selected by its id
+uv run pytest --slow \
+    "packages/climate-ref-example/tests/integration/test_diagnostics.py::test_run_test_cases[global-mean-timeseries/default]"
+```
+
+### Writing custom tests
+
+For behaviour the standard drift check doesn't cover,
+run a test case directly with `TestCaseRunner`.
+The `config` fixture (an isolated per-test configuration)
+comes from the `climate_ref.conftest_plugin` pytest plugin,
+which is auto-discovered when `climate-ref` is installed.
 
 ```python
 import pytest
+from my_provider import provider
+
+from climate_ref.testing import TestCaseRunner
+from climate_ref_core.testing import TestCasePaths, load_datasets_from_yaml
+
+
+@pytest.mark.slow
+def test_my_diagnostic(config, tmp_path):
+    diagnostic = provider.get("my-diagnostic")
+    diagnostic.provider.configure(config)
 
-def test_my_diagnostic(run_test_case):
-    from my_provider import MyDiagnostic
+    paths = TestCasePaths.from_diagnostic(diagnostic, "default")
+    if paths is None or not paths.catalog.exists():
+        pytest.skip("Test data not fetched; run `ref test-cases fetch` first")
 
-    diagnostic = MyDiagnostic()
-    result = run_test_case.run(diagnostic, "default")
+    runner = TestCaseRunner(config=config, datasets=load_datasets_from_yaml(paths.catalog))
+    result = runner.run(diagnostic, "default", output_dir=tmp_path / "output")
 
     assert result.successful
-    assert result.metric_bundle_filename is not None
 ```
 
-In future, tests will be generated for each of the test cases.
+### Markers
 
-### Marking Tests
-
-Use pytest markers for test categorization:
+Mark custom tests so they run in the right CI tier:
 
 ```python
 @pytest.mark.slow
-def test_full_resolution(run_test_case):
+def test_full_resolution(config):
     """Test with full-resolution ESGF data (slow)."""
     ...
 
 @pytest.mark.requires_esgf_data
-def test_requires_fetched_data(run_test_case):
+def test_requires_fetched_data(config):
     """Test that requires fetched ESGF data."""
     ...
 ```
@@ -380,6 +422,9 @@ Run specific test categories:
 # Include slow tests
 pytest --slow
 
+# Only the generated test-case drift tests
+pytest --slow -m test_cases
+
 # Skip tests requiring ESGF data
 pytest -m "not requires_esgf_data"
 ```
@@ -526,6 +571,62 @@ If you hold credentials locally, `ref test-cases mint` does the same from your m
    - The derived test data directory path
    - Whether catalogs/regression directories are being created
 
+### Debugging a failing test case
+
+When a pytest drift test (or the CI replay gate) reports that a committed bundle has drifted,
+the fastest loop is usually through the CLI rather than pytest,
+because the outputs stay in a predictable place.
+
+1. **Re-run the test case via the CLI** and inspect what it produced:
+
+   ```bash
+   ref --verbose test-cases run --provider my-provider --diagnostic my-diagnostic
+   ```
+
+   Every run repopulates the gitignored `output/latest/` slot in the test case directory
+   with the executed native files and, under `output/latest/regression/`,
+   the rebuilt committed bundle.
+   Pass `--output-directory ./scratch` to also keep the raw execution directory
+   (logs, intermediate files) somewhere convenient.
+
+2. **Diff the rebuilt bundle against the tracked baseline**:
+
+   ```bash
+   cd packages/climate-ref-my-provider/tests/test-data/my-diagnostic/default
+   git diff --no-index regression/ output/latest/regression/
+   ```
+
+   The drift assertion also prints a per-file summary of what moved beyond tolerance.
+
+3. **Compare two runs side by side** using named output slots —
+   `latest` is overwritten on every run, but a named slot persists:
+
+   ```bash
+   ref test-cases run ... --label before
+   # ... change the code ...
+   ref test-cases run ... --label after
+   diff -r .../output/before/regression .../output/after/regression
+   ```
+
+4. **Separate "my code changed the results" from "the baseline is stale"** with
+   `ref test-cases replay --provider my-provider`.
+
+   Replay rebuilds the bundle from the stored native baseline *without executing the diagnostic*,
+   so if replay passes but `run` drifts, your change moved the results;
+   if replay also fails, the committed baseline itself is inconsistent
+   (regenerate with `run --force-regen` and bump `test_case_version`).
+
+5. **Debug through pytest** when you need a debugger at the point of failure:
+
+   ```bash
+   uv run pytest --slow --pdb \
+       "packages/climate-ref-my-provider/tests/integration/test_diagnostics.py::test_run_test_cases[my-diagnostic/default]"
+   ```
+
+   The pytest run works in the test's `tmp_path`
+   (retained for the most recent runs under `/tmp/pytest-of-<user>/`):
+   the execution directory is `exec/` and the rebuilt bundle is `slot/regression/`.
+
 ## Other useful links
 
 - [Adding Custom Diagnostics](adding_custom_diagnostics.md)
diff --git a/packages/climate-ref-core/src/climate_ref_core/diagnostics.py b/packages/climate-ref-core/src/climate_ref_core/diagnostics.py
index b403d0278..05aa8e331 100644
--- a/packages/climate-ref-core/src/climate_ref_core/diagnostics.py
+++ b/packages/climate-ref-core/src/climate_ref_core/diagnostics.py
@@ -610,7 +610,7 @@ def provider(self) -> DiagnosticProvider:
     def provider(self, value: DiagnosticProvider) -> None:
         self._provider = value
 
-    def run(self, definition: ExecutionDefinition) -> ExecutionResult:
+    def run(self, definition: ExecutionDefinition, *, capture_regression: bool = False) -> ExecutionResult:
         """
         Run the diagnostic on the given configuration.
 
@@ -620,11 +620,19 @@ def run(self, definition: ExecutionDefinition) -> ExecutionResult:
         ----------
         definition
             The configuration to run the diagnostic on.
+        capture_regression
+            When True, call :meth:`prepare_regression_output` between execute
+            and build so the native output is normalised for deterministic regression capture.
+
+            Used only by the test-case runner.
         """
         # Execute the diagnostic
         # This may be run in a separate process (or python environment)
         self.execute(definition)
 
+        if capture_regression:
+            self.prepare_regression_output(definition)
+
         # Build the result from the output bundle
         return self.build_execution_result(definition)
 
@@ -632,12 +640,12 @@ def prepare_regression_output(self, definition: ExecutionDefinition) -> None:
         """
         Normalise native output for deterministic regression capture.
 
-        Hook called by the test-case regression runner *between*
+        Hook called by :meth:`run` (only when ``capture_regression=True``) *between*
         :meth:`execute` and :meth:`build_execution_result`,
         allowing a provider to remove avoidable non-determinism (e.g. timestamped directory names)
         before the output bundle is built and captured as a baseline.
 
-        This is **not** called during normal execution (:meth:`run`),
+        Normal execution (``capture_regression=False``, the default) does not call it,
         so it must only be used for regression-capture concerns.
 
         The default implementation does nothing.
diff --git a/packages/climate-ref-core/src/climate_ref_core/testing.py b/packages/climate-ref-core/src/climate_ref_core/testing.py
index c29a317d5..99167aafa 100644
--- a/packages/climate-ref-core/src/climate_ref_core/testing.py
+++ b/packages/climate-ref-core/src/climate_ref_core/testing.py
@@ -4,14 +4,12 @@
 This module provides:
 - TestCase and TestDataSpecification for defining test scenarios
 - YAML serialization for dataset catalogs (with paths stored separately)
-- RegressionValidator for validating pre-stored outputs
-- Utilities for CMEC bundle validation
+- Utilities for CMEC bundle and series regression validation
 """
 
 from __future__ import annotations
 
 import json
-import shutil
 import sys
 from pathlib import Path
 from typing import TYPE_CHECKING, Any
@@ -27,10 +25,10 @@
     Selector,
     SourceDatasetType,
 )
-from climate_ref_core.diagnostics import ExecutionDefinition, ExecutionResult
+from climate_ref_core.diagnostics import ExecutionResult
 from climate_ref_core.esgf.base import ESGFRequest
 from climate_ref_core.metric_values.typing import SeriesMetricValue
-from climate_ref_core.output_files import PlaceholderMap, ordered_replacements
+from climate_ref_core.output_files import ordered_replacements
 from climate_ref_core.paths import safe_path
 from climate_ref_core.pycmec.metric import CMECMetric
 from climate_ref_core.pycmec.output import CMECOutput
@@ -651,93 +649,6 @@ def _sorted_dump(series: list[SeriesMetricValue]) -> list[dict[str, Any]]:
     )
 
 
-@frozen
-class RegressionValidator:
-    """
-    Validate diagnostic outputs from pre-stored regression data.
-
-    Loads regression outputs and validates CMEC bundles without
-    running the diagnostic. Suitable for fast CI validation.
-
-    The regression data is expected at:
-    test_data_dir/{diagnostic}/{test_case}/regression/
-    """
-
-    diagnostic: Diagnostic
-    test_case_name: str
-    test_data_dir: Path
-
-    @property
-    def paths(self) -> TestCasePaths:
-        """Get paths for this test case."""
-        return TestCasePaths.from_test_data_dir(self.test_data_dir, self.diagnostic.slug, self.test_case_name)
-
-    @property
-    def _baseline_placeholders(self) -> PlaceholderMap:
-        """
-        Base placeholder map for this verification context.
-
-        Declared once and reused by both :meth:`load_regression_definition` (hydrate) and
-        :meth:`validate` (series replacements) so the two sides cannot drift to different token sets.
-        This context does not know the shared-software root, so the map omits ``<SOFTWARE_ROOT_DIR>``;
-        each caller binds the per-execution ``<OUTPUT_DIR>`` via :meth:`PlaceholderMap.with_output`.
-        """
-        return PlaceholderMap.for_baseline(test_data_dir=self.test_data_dir)
-
-    def has_regression_data(self) -> bool:
-        """Check if regression data exists for this test case."""
-        regression_path = self.paths.regression
-        return regression_path.exists() and (regression_path / "diagnostic.json").exists()
-
-    def load_regression_definition(self, tmp_dir: Path) -> ExecutionDefinition:
-        """
-        Load regression data and create an ExecutionDefinition.
-
-        Copies regression data to tmp_dir and replaces path placeholders.
-        """
-        regression_path = self.paths.regression
-        catalog_path = self.paths.catalog
-
-        if not catalog_path.exists():
-            raise FileNotFoundError(
-                f"No catalog file at {catalog_path} for test case datasets. Run `ref test-cases fetch` first."
-            )
-        if not regression_path.exists():
-            raise FileNotFoundError(
-                f"No regression data at {regression_path}. Run 'ref test-cases run --force-regen' first."
-            )
-
-        output_dir = tmp_dir / "output"
-        output_dir.mkdir(parents=True, exist_ok=True)
-        shutil.copytree(regression_path, output_dir, dirs_exist_ok=True)
-
-        self._baseline_placeholders.with_output(output_dir).hydrate(output_dir)
-
-        datasets: ExecutionDatasetCollection = load_datasets_from_yaml(catalog_path)
-
-        return ExecutionDefinition(
-            diagnostic=self.diagnostic,
-            key=f"test-{self.test_case_name}",
-            datasets=datasets,
-            output_directory=output_dir,
-            root_directory=tmp_dir,
-        )
-
-    def validate(self, definition: ExecutionDefinition) -> None:
-        """Validate CMEC bundles and series in the regression output."""
-        result = self.diagnostic.build_execution_result(definition)
-        result.to_output_path("out.log").touch()  # Log file not tracked in regression
-        validate_cmec_bundles(self.diagnostic, result)
-        validate_series_regression(
-            expected_path=self.paths.regression / "series.json",
-            actual_path=definition.output_directory / "series.json",
-            slug=self.diagnostic.slug,
-            replacements=self._baseline_placeholders.with_output(
-                definition.output_directory
-            ).as_replacements(),
-        )
-
-
 def collect_test_case_params(provider: DiagnosticProvider) -> list[ParameterSet]:
     """
     Collect all diagnostic/test_case pairs from a provider for parameterized testing.
diff --git a/packages/climate-ref-core/tests/unit/test_diagnostics.py b/packages/climate-ref-core/tests/unit/test_diagnostics.py
index 49be8deed..799f41b09 100644
--- a/packages/climate-ref-core/tests/unit/test_diagnostics.py
+++ b/packages/climate-ref-core/tests/unit/test_diagnostics.py
@@ -198,6 +198,60 @@ def build_execution_result(self, definition):
         assert result == diagnostic_result
 
 
+class TestDiagnosticRun:
+    """Tests for ``Diagnostic.run`` and its ``capture_regression`` hook."""
+
+    def _make_diagnostic(self):
+        class RunDiagnostic(Diagnostic):
+            name = "run-diagnostic"
+            slug = "run-diagnostic"
+            data_requirements = (
+                DataRequirement(source_type=SourceDatasetType.CMIP6, filters=(), group_by=None),
+            )
+
+            def execute(self, definition):
+                """No-op execute."""
+
+            def build_execution_result(self, definition):
+                """Unused; patched per-test."""
+
+        return RunDiagnostic()
+
+    def test_run_default_skips_regression_hook(self, mocker):
+        """The default ``run`` executes then builds, without the regression hook."""
+        diagnostic = self._make_diagnostic()
+        execute = mocker.patch.object(diagnostic, "execute")
+        prepare = mocker.patch.object(diagnostic, "prepare_regression_output")
+        build = mocker.patch.object(diagnostic, "build_execution_result", return_value=mocker.sentinel.result)
+        definition = mocker.sentinel.definition
+
+        result = diagnostic.run(definition)
+
+        execute.assert_called_once_with(definition)
+        prepare.assert_not_called()
+        build.assert_called_once_with(definition)
+        assert result is mocker.sentinel.result
+
+    def test_run_capture_regression_calls_hook_between_execute_and_build(self, mocker):
+        """With ``capture_regression=True`` the hook runs after execute and before build."""
+        diagnostic = self._make_diagnostic()
+        manager = mocker.Mock()
+        manager.attach_mock(mocker.patch.object(diagnostic, "execute"), "execute")
+        manager.attach_mock(mocker.patch.object(diagnostic, "prepare_regression_output"), "prepare")
+        manager.attach_mock(
+            mocker.patch.object(diagnostic, "build_execution_result", return_value=mocker.sentinel.result),
+            "build",
+        )
+        definition = mocker.sentinel.definition
+
+        result = diagnostic.run(definition, capture_regression=True)
+
+        assert result is mocker.sentinel.result
+        # Ordering matters: normalise the native output only after execute, before build.
+        assert [call[0] for call in manager.mock_calls] == ["execute", "prepare", "build"]
+        manager.prepare.assert_called_once_with(definition)
+
+
 class TestExecutionResult:
     def test_build_from_output_bundle(
         self,
diff --git a/packages/climate-ref-core/tests/unit/test_testing.py b/packages/climate-ref-core/tests/unit/test_testing.py
index 414ce76ba..68cdbe423 100644
--- a/packages/climate-ref-core/tests/unit/test_testing.py
+++ b/packages/climate-ref-core/tests/unit/test_testing.py
@@ -11,7 +11,6 @@
 from climate_ref_core.metric_values.typing import SeriesMetricValue
 from climate_ref_core.regression.manifest import Manifest
 from climate_ref_core.testing import (
-    RegressionValidator,
     TestCase,
     TestCasePaths,
     TestDataSpecification,
@@ -784,162 +783,6 @@ def test_validate_cmec_bundles_fails_on_dimension_mismatch(self, tmp_path):
             validate_cmec_bundles(mock_diagnostic, mock_result)
 
 
-class TestRegressionValidator:
-    """Tests for RegressionValidator class."""
-
-    def test_paths_property(self, tmp_path):
-        """Test paths property returns correct TestCasePaths."""
-        mock_diagnostic = MagicMock()
-        mock_diagnostic.slug = "my-diag"
-
-        validator = RegressionValidator(
-            diagnostic=mock_diagnostic,
-            test_case_name="default",
-            test_data_dir=tmp_path,
-        )
-
-        paths = validator.paths
-        assert paths.root == tmp_path / "my-diag" / "default"
-
-    def test_has_regression_data_false_when_missing(self, tmp_path):
-        """Test has_regression_data returns False when no regression data."""
-        mock_diagnostic = MagicMock()
-        mock_diagnostic.slug = "my-diag"
-
-        validator = RegressionValidator(
-            diagnostic=mock_diagnostic,
-            test_case_name="default",
-            test_data_dir=tmp_path,
-        )
-
-        assert validator.has_regression_data() is False
-
-    def test_has_regression_data_true_when_present(self, tmp_path):
-        """Test has_regression_data returns True when regression data exists."""
-        mock_diagnostic = MagicMock()
-        mock_diagnostic.slug = "my-diag"
-
-        # Create regression directory with diagnostic.json
-        regression_dir = tmp_path / "my-diag" / "default" / "regression"
-        regression_dir.mkdir(parents=True)
-        (regression_dir / "diagnostic.json").write_text("{}")
-
-        validator = RegressionValidator(
-            diagnostic=mock_diagnostic,
-            test_case_name="default",
-            test_data_dir=tmp_path,
-        )
-
-        assert validator.has_regression_data() is True
-
-    def test_load_regression_definition_missing_data(self, tmp_path):
-        """Test load_regression_definition raises FileNotFoundError when no data."""
-        mock_diagnostic = MagicMock()
-        mock_diagnostic.slug = "my-diag"
-
-        validator = RegressionValidator(
-            diagnostic=mock_diagnostic,
-            test_case_name="default",
-            test_data_dir=tmp_path,
-        )
-
-        with pytest.raises(FileNotFoundError, match="No catalog file"):
-            validator.load_regression_definition(tmp_path / "tmp")
-
-    def test_load_regression_definition_success(self, tmp_path):
-        """Test load_regression_definition copies data and replaces placeholders."""
-        mock_diagnostic = MagicMock()
-        mock_diagnostic.slug = "my-diag"
-
-        metric_bundle = {
-            "DIMENSIONS": {"json_structure": []},
-            "RESULTS": {},
-            "path": "<OUTPUT_DIR>/result.nc",
-            "data_path": "<TEST_DATA_DIR>/input.nc",
-        }
-
-        validator = RegressionValidator(
-            diagnostic=mock_diagnostic,
-            test_case_name="default",
-            test_data_dir=tmp_path,
-        )
-
-        # Create regression data
-        validator.paths.create()
-        validator.paths.catalog.write_text("{}")
-        regression_dir = validator.paths.regression
-        regression_dir.mkdir(parents=True)
-        (regression_dir / "diagnostic.json").write_text(json.dumps(metric_bundle))
-        (regression_dir / "output.json").write_text('{"html": {}}')
-
-        work_dir = tmp_path / "work"
-        definition = validator.load_regression_definition(work_dir)
-
-        assert definition.diagnostic == mock_diagnostic
-        assert definition.key == "test-default"
-        assert definition.output_directory == work_dir / "output"
-
-        # Check placeholders were replaced
-        loaded_content = (work_dir / "output" / "diagnostic.json").read_text()
-        assert "<OUTPUT_DIR>" not in loaded_content
-        assert "<TEST_DATA_DIR>" not in loaded_content
-        assert str(work_dir / "output") in loaded_content
-        assert str(tmp_path) in loaded_content
-
-    def test_validate_calls_validate_cmec_bundles(self, tmp_path):
-        """Test validate method calls validate_cmec_bundles."""
-        mock_diagnostic = MagicMock()
-        mock_diagnostic.slug = "my-diag"
-        mock_diagnostic.facets = ("source_id", "metric", "statistic")
-
-        # Create output directory with valid bundles
-        output_dir = tmp_path / "output"
-        output_dir.mkdir()
-
-        # Create mock result with spec and proper attributes
-        mock_result = MagicMock(spec=ExecutionResult)
-        mock_result.successful = True
-        mock_result.metric_bundle_filename = Path("diagnostic.json")
-        mock_result.output_bundle_filename = Path("output.json")
-        mock_result.to_output_path = lambda f: output_dir / f
-        mock_diagnostic.build_execution_result.return_value = mock_result
-
-        # Create mock definition
-        mock_definition = MagicMock()
-        mock_definition.to_output_path = lambda f: output_dir / f if f else output_dir
-
-        # Create valid CMEC bundles (RESULTS needs nested dicts with scalars at leaf)
-        metric_bundle = {
-            "DIMENSIONS": {
-                "json_structure": ["source_id", "metric", "statistic"],
-                "source_id": {"M1": {}},
-                "metric": {"rmse": {}},
-                "statistic": {"value": {}},
-            },
-            "RESULTS": {"M1": {"rmse": {"value": 0.5}}},
-        }
-        output_bundle = {
-            "data": {},
-            "plots": {},
-            "metrics": {},
-            "provenance": {"environment": {}, "modeldata": "", "obsdata": "", "log": ""},
-        }
-        (output_dir / "diagnostic.json").write_text(json.dumps(metric_bundle))
-        (output_dir / "output.json").write_text(json.dumps(output_bundle))
-        (output_dir / "out.log").touch()  # Create log file that validate touches
-
-        validator = RegressionValidator(
-            diagnostic=mock_diagnostic,
-            test_case_name="default",
-            test_data_dir=tmp_path,
-        )
-
-        validator.validate(mock_definition)
-
-        # Verify build_execution_result was called
-        mock_diagnostic.build_execution_result.assert_called_once_with(mock_definition)
-
-
 def _series(value: float, *, dim: str = "M1", index_name: str = "year", attrs=None) -> SeriesMetricValue:
     """Build a small SeriesMetricValue for testing."""
     return SeriesMetricValue(
diff --git a/packages/climate-ref-esmvaltool/tests/integration/test_diagnostics.py b/packages/climate-ref-esmvaltool/tests/integration/test_diagnostics.py
index 89c99f416..694e8d80f 100644
--- a/packages/climate-ref-esmvaltool/tests/integration/test_diagnostics.py
+++ b/packages/climate-ref-esmvaltool/tests/integration/test_diagnostics.py
@@ -1,102 +1,5 @@
-from pathlib import Path
-
-import pytest
 from climate_ref_esmvaltool import provider
 
-from climate_ref.testing import TestCaseRunner, validate_result
-from climate_ref_core.diagnostics import Diagnostic
-from climate_ref_core.testing import (
-    RegressionValidator,
-    TestCasePaths,
-    collect_test_case_params,
-    load_datasets_from_yaml,
-)
-
-
-@pytest.fixture(scope="session")
-def provider_test_data_dir() -> Path:
-    """Path to the package-local test data directory."""
-    return Path(__file__).parent.parent / "test-data"
-
-
-# Test case params for parameterized test_case tests
-test_case_params = collect_test_case_params(provider)
-
-
-@pytest.mark.skip(
-    reason="Parked: RegressionValidator replays the committed bundle offline, but native baselines "
-    "now live in the object store (Framework B), so build_execution_result cannot rebuild from the "
-    "repo alone. `ref test-cases replay` provides regression coverage via the store; this offline "
-    "path is retained (body intact) for local step-through debugging."
-)
-@pytest.mark.parametrize("diagnostic,test_case_name", test_case_params)
-def test_validate_test_case_regression(
-    diagnostic: Diagnostic,
-    test_case_name: str,
-    provider_test_data_dir: Path,
-    config,
-    tmp_path: Path,
-):
-    """
-    Validate pre-stored test case regression outputs as CMEC bundles.
-
-    Each diagnostic/test_case is a separate parameterized test.
-    """
-    diagnostic.provider.configure(config)
-
-    paths = TestCasePaths.from_test_data_dir(
-        provider_test_data_dir,
-        diagnostic.slug,
-        test_case_name,
-    )
-
-    if not paths.catalog.exists():
-        pytest.skip(f"No catalog file for {diagnostic.slug}/{test_case_name}")
-    if not paths.regression.exists():
-        pytest.skip(f"No regression data for {diagnostic.slug}/{test_case_name}")
-
-    validator = RegressionValidator(
-        diagnostic=diagnostic,
-        test_case_name=test_case_name,
-        test_data_dir=provider_test_data_dir,
-    )
-
-    definition = validator.load_regression_definition(tmp_path / diagnostic.slug / test_case_name)
-    validator.validate(definition)
-
-
-@pytest.mark.slow
-@pytest.mark.test_cases
-@pytest.mark.parametrize("diagnostic,test_case_name", test_case_params)
-def test_run_test_cases(
-    diagnostic: Diagnostic,
-    test_case_name: str,
-    provider_test_data_dir: Path,
-    config,
-    tmp_path: Path,
-):
-    """
-    Run diagnostic test cases end-to-end with ESGF data.
-
-    Requires: `ref test-cases fetch --provider esmvaltool` to have been run first.
-    """
-    diagnostic.provider.configure(config)
-
-    paths = TestCasePaths.from_test_data_dir(
-        provider_test_data_dir,
-        diagnostic.slug,
-        test_case_name,
-    )
-
-    if not paths.catalog.exists():
-        pytest.skip(f"No catalog file for {diagnostic.slug}/{test_case_name}")
-
-    datasets = load_datasets_from_yaml(paths.catalog)
-
-    runner = TestCaseRunner(config=config, datasets=datasets)
-    output_dir = tmp_path / diagnostic.slug / test_case_name
-
-    result = runner.run(diagnostic, test_case_name, output_dir)
+from climate_ref.testing import create_no_drift_test
 
-    assert result.successful, f"Diagnostic {diagnostic.slug} failed"
-    validate_result(diagnostic, config, result)
+test_run_test_cases = create_no_drift_test(provider)
diff --git a/packages/climate-ref-example/tests/integration/test_diagnostics.py b/packages/climate-ref-example/tests/integration/test_diagnostics.py
index fa3b1a788..10f171b96 100644
--- a/packages/climate-ref-example/tests/integration/test_diagnostics.py
+++ b/packages/climate-ref-example/tests/integration/test_diagnostics.py
@@ -1,105 +1,5 @@
-from pathlib import Path
-
-import pytest
 from climate_ref_example import provider
 
-from climate_ref.testing import TestCaseRunner, validate_result
-from climate_ref_core.diagnostics import Diagnostic
-from climate_ref_core.testing import (
-    RegressionValidator,
-    TestCasePaths,
-    collect_test_case_params,
-    load_datasets_from_yaml,
-)
-
-
-@pytest.fixture(scope="session")
-def provider_test_data_dir() -> Path:
-    """Path to the package-local test data directory."""
-    return Path(__file__).parent.parent / "test-data"
-
-
-test_case_params = collect_test_case_params(provider)
-
-
-@pytest.mark.skip(
-    reason="Parked: RegressionValidator replays the committed bundle offline, but native baselines "
-    "now live in the object store (Framework B), so build_execution_result cannot rebuild from the "
-    "repo alone. `ref test-cases replay` provides regression coverage via the store; this offline "
-    "path is retained (body intact) for local step-through debugging."
-)
-@pytest.mark.parametrize("diagnostic,test_case_name", test_case_params)
-def test_validate_test_case_regression(
-    diagnostic: Diagnostic,
-    test_case_name: str,
-    provider_test_data_dir: Path,
-    config,
-    tmp_path: Path,
-):
-    """
-    Validate pre-stored test case regression outputs as CMEC bundles.
-
-    Each diagnostic/test_case is a separate parameterized test.
-    """
-    diagnostic.provider.configure(config)
-
-    paths = TestCasePaths.from_test_data_dir(
-        provider_test_data_dir,
-        diagnostic.slug,
-        test_case_name,
-    )
-
-    validator = RegressionValidator(
-        diagnostic=diagnostic,
-        test_case_name=test_case_name,
-        test_data_dir=provider_test_data_dir,
-    )
-
-    if not paths.catalog.exists():
-        pytest.skip(f"No catalog file for {diagnostic.slug}/{test_case_name}")
-    if not validator.has_regression_data():
-        pytest.skip(f"No regression data for {diagnostic.slug}/{test_case_name}")
-
-    # TODO: remove this once we have migrated test cases to use the committed bundles.
-    if not any(paths.regression.glob("*.nc")):
-        pytest.skip(f"No committed NetCDF baseline for {diagnostic.slug}/{test_case_name}")
-
-    definition = validator.load_regression_definition(tmp_path / diagnostic.slug / test_case_name)
-    validator.validate(definition)
-
-
-@pytest.mark.slow
-@pytest.mark.test_cases
-@pytest.mark.parametrize("diagnostic,test_case_name", test_case_params)
-def test_run_test_cases(
-    diagnostic: Diagnostic,
-    test_case_name: str,
-    provider_test_data_dir: Path,
-    config,
-    tmp_path: Path,
-):
-    """
-    Run diagnostic test cases end-to-end with ESGF data.
-
-    Requires: `ref test-cases fetch --provider example` to have been run first.
-    """
-    diagnostic.provider.configure(config)
-
-    paths = TestCasePaths.from_test_data_dir(
-        provider_test_data_dir,
-        diagnostic.slug,
-        test_case_name,
-    )
-
-    if not paths.catalog.exists():
-        pytest.skip(f"No catalog file for {diagnostic.slug}/{test_case_name}")
-
-    datasets = load_datasets_from_yaml(paths.catalog)
-
-    runner = TestCaseRunner(config=config, datasets=datasets)
-    output_dir = tmp_path / diagnostic.slug / test_case_name
-
-    result = runner.run(diagnostic, test_case_name, output_dir)
+from climate_ref.testing import create_no_drift_test
 
-    assert result.successful, f"Diagnostic {diagnostic.slug} failed"
-    validate_result(diagnostic, config, result)
+test_run_test_cases = create_no_drift_test(provider)
diff --git a/packages/climate-ref-ilamb/tests/integration/test_diagnostics.py b/packages/climate-ref-ilamb/tests/integration/test_diagnostics.py
index 748ab547d..4a852da09 100644
--- a/packages/climate-ref-ilamb/tests/integration/test_diagnostics.py
+++ b/packages/climate-ref-ilamb/tests/integration/test_diagnostics.py
@@ -1,110 +1,5 @@
-from pathlib import Path
-
-import pytest
 from climate_ref_ilamb import provider
 
-from climate_ref.testing import TestCaseRunner, validate_result
-from climate_ref_core.diagnostics import Diagnostic
-from climate_ref_core.testing import (
-    RegressionValidator,
-    TestCasePaths,
-    collect_test_case_params,
-    load_datasets_from_yaml,
-)
-
-
-@pytest.fixture(scope="session")
-def provider_test_data_dir() -> Path:
-    """Path to the package-local test data directory."""
-    return Path(__file__).parent.parent / "test-data"
-
-
-# Test case params for parameterized test_case tests
-test_case_params = collect_test_case_params(provider)
-
-
-@pytest.mark.skip(
-    reason="Parked: RegressionValidator replays the committed bundle offline, but native baselines "
-    "now live in the object store (Framework B), so build_execution_result cannot rebuild from the "
-    "repo alone. `ref test-cases replay` provides regression coverage via the store; this offline "
-    "path is retained (body intact) for local step-through debugging."
-)
-@pytest.mark.parametrize("diagnostic,test_case_name", test_case_params)
-def test_validate_test_case_regression(
-    diagnostic: Diagnostic,
-    test_case_name: str,
-    provider_test_data_dir: Path,
-    config,
-    tmp_path: Path,
-):
-    """
-    Validate pre-stored test case regression outputs as CMEC bundles.
-
-    Each diagnostic/test_case is a separate parameterized test.
-    """
-    diagnostic.provider.configure(config)
-
-    paths = TestCasePaths.from_test_data_dir(
-        provider_test_data_dir,
-        diagnostic.slug,
-        test_case_name,
-    )
-
-    if not paths.catalog.exists():
-        pytest.skip(f"No catalog file for {diagnostic.slug}/{test_case_name}")
-    if not paths.regression.exists():
-        pytest.skip(f"No regression data for {diagnostic.slug}/{test_case_name}")
-
-    validator = RegressionValidator(
-        diagnostic=diagnostic,
-        test_case_name=test_case_name,
-        test_data_dir=provider_test_data_dir,
-    )
-
-    # TODO: remove this once we have migrated test cases to use the committed bundles.
-    # Under Framework B the native baselines live in the object store, not the repo, so
-    # build_execution_result cannot be replayed from the committed bundle alone. The
-    # online equivalent (`ref test-cases replay`) materialises native blobs from the
-    # store; this offline test skips until that path is shared here.
-    if not any(paths.regression.glob("*.nc")):
-        pytest.skip(f"No committed native baseline for {diagnostic.slug}/{test_case_name}")
-
-    definition = validator.load_regression_definition(tmp_path / diagnostic.slug / test_case_name)
-    validator.validate(definition)
-
-
-@pytest.mark.slow
-@pytest.mark.test_cases
-@pytest.mark.parametrize("diagnostic,test_case_name", test_case_params)
-def test_run_test_cases(
-    diagnostic: Diagnostic,
-    test_case_name: str,
-    provider_test_data_dir: Path,
-    config,
-    tmp_path: Path,
-):
-    """
-    Run diagnostic test cases end-to-end with ESGF data.
-
-    Requires: `ref test-cases fetch --provider ilamb` to have been run first.
-    """
-    diagnostic.provider.configure(config)
-
-    paths = TestCasePaths.from_test_data_dir(
-        provider_test_data_dir,
-        diagnostic.slug,
-        test_case_name,
-    )
-
-    if not paths.catalog.exists():
-        pytest.skip(f"No catalog file for {diagnostic.slug}/{test_case_name}")
-
-    datasets = load_datasets_from_yaml(paths.catalog)
-
-    runner = TestCaseRunner(config=config, datasets=datasets)
-    output_dir = tmp_path / diagnostic.slug / test_case_name
-
-    result = runner.run(diagnostic, test_case_name, output_dir)
+from climate_ref.testing import create_no_drift_test
 
-    assert result.successful, f"Diagnostic {diagnostic.slug} failed"
-    validate_result(diagnostic, config, result)
+test_run_test_cases = create_no_drift_test(provider)
diff --git a/packages/climate-ref-pmp/tests/integration/test_diagnostics.py b/packages/climate-ref-pmp/tests/integration/test_diagnostics.py
index e5bceebb0..180bcaebe 100644
--- a/packages/climate-ref-pmp/tests/integration/test_diagnostics.py
+++ b/packages/climate-ref-pmp/tests/integration/test_diagnostics.py
@@ -1,102 +1,5 @@
-from pathlib import Path
-
-import pytest
 from climate_ref_pmp import provider
 
-from climate_ref.testing import TestCaseRunner, validate_result
-from climate_ref_core.diagnostics import Diagnostic
-from climate_ref_core.testing import (
-    RegressionValidator,
-    TestCasePaths,
-    collect_test_case_params,
-    load_datasets_from_yaml,
-)
-
-
-@pytest.fixture(scope="session")
-def provider_test_data_dir() -> Path:
-    """Path to the package-local test data directory."""
-    return Path(__file__).parent.parent / "test-data"
-
-
-# Test case params for parameterized test_case tests
-test_case_params = collect_test_case_params(provider)
-
-
-@pytest.mark.skip(
-    reason="Parked: RegressionValidator replays the committed bundle offline, but native baselines "
-    "now live in the object store (Framework B), so build_execution_result cannot rebuild from the "
-    "repo alone. `ref test-cases replay` provides regression coverage via the store; this offline "
-    "path is retained (body intact) for local step-through debugging."
-)
-@pytest.mark.parametrize("diagnostic,test_case_name", test_case_params)
-def test_validate_test_case_regression(
-    diagnostic: Diagnostic,
-    test_case_name: str,
-    provider_test_data_dir: Path,
-    config,
-    tmp_path: Path,
-):
-    """
-    Validate pre-stored test case regression outputs as CMEC bundles.
-
-    Each diagnostic/test_case is a separate parameterized test.
-    """
-    diagnostic.provider.configure(config)
-
-    paths = TestCasePaths.from_test_data_dir(
-        provider_test_data_dir,
-        diagnostic.slug,
-        test_case_name,
-    )
-
-    if not paths.catalog.exists():
-        pytest.skip(f"No catalog file for {diagnostic.slug}/{test_case_name}")
-    if not paths.regression.exists():
-        pytest.skip(f"No regression data for {diagnostic.slug}/{test_case_name}")
-
-    validator = RegressionValidator(
-        diagnostic=diagnostic,
-        test_case_name=test_case_name,
-        test_data_dir=provider_test_data_dir,
-    )
-
-    definition = validator.load_regression_definition(tmp_path / diagnostic.slug / test_case_name)
-    validator.validate(definition)
-
-
-@pytest.mark.slow
-@pytest.mark.test_cases
-@pytest.mark.parametrize("diagnostic,test_case_name", test_case_params)
-def test_run_test_cases(
-    diagnostic: Diagnostic,
-    test_case_name: str,
-    provider_test_data_dir: Path,
-    config,
-    tmp_path: Path,
-):
-    """
-    Run diagnostic test cases end-to-end with ESGF data.
-
-    Requires: `ref test-cases fetch --provider pmp` to have been run first.
-    """
-    diagnostic.provider.configure(config)
-
-    paths = TestCasePaths.from_test_data_dir(
-        provider_test_data_dir,
-        diagnostic.slug,
-        test_case_name,
-    )
-
-    if not paths.catalog.exists():
-        pytest.skip(f"No catalog file for {diagnostic.slug}/{test_case_name}")
-
-    datasets = load_datasets_from_yaml(paths.catalog)
-
-    runner = TestCaseRunner(config=config, datasets=datasets)
-    output_dir = tmp_path / diagnostic.slug / test_case_name
-
-    result = runner.run(diagnostic, test_case_name, output_dir)
+from climate_ref.testing import create_no_drift_test
 
-    assert result.successful, f"Diagnostic {diagnostic.slug} failed"
-    validate_result(diagnostic, config, result)
+test_run_test_cases = create_no_drift_test(provider)
diff --git a/packages/climate-ref/src/climate_ref/conftest_plugin.py b/packages/climate-ref/src/climate_ref/conftest_plugin.py
index 5ebf6ec55..d1396728d 100644
--- a/packages/climate-ref/src/climate_ref/conftest_plugin.py
+++ b/packages/climate-ref/src/climate_ref/conftest_plugin.py
@@ -18,7 +18,6 @@
 - ``test_data_dir`` / ``sample_data_dir`` -- data paths
 - ``sample_data`` -- session-scoped sample data fetch
 - ``cmip6_data_catalog`` / ``obs4mips_data_catalog`` / ``data_catalog`` -- data catalogs
-- ``run_test_case`` -- ``TestCaseRunner`` wrapper that converts errors to ``pytest.skip``
 - ``definition_factory`` -- create ``ExecutionDefinition`` instances
 - ``provider`` / ``mock_diagnostic`` -- mock diagnostic provider
 - ``solved_definition_factory`` -- build an ``ExecutionDefinition`` from the sample-data catalog
@@ -50,12 +49,10 @@
 from climate_ref.solver import solve_executions
 from climate_ref.testing import (
     TEST_DATA_DIR,
-    TestCaseRunner,
     fetch_sample_data,
 )
 from climate_ref_core.datasets import DatasetCollection, ExecutionDatasetCollection, SourceDatasetType
 from climate_ref_core.diagnostics import DataRequirement, Diagnostic, ExecutionDefinition, ExecutionResult
-from climate_ref_core.exceptions import TestCaseError
 from climate_ref_core.logging import add_log_handler, remove_log_handler
 from climate_ref_core.providers import DiagnosticProvider
 
@@ -176,31 +173,6 @@ def esgf_data_catalog_trimmed(
     return result
 
 
-@pytest.fixture
-def run_test_case(config: Config) -> object:
-    """
-    Fixture for running diagnostic test cases.
-
-    Wraps ``TestCaseRunner`` to convert ``TestCaseError`` into ``pytest.skip``.
-    """
-    runner = TestCaseRunner(config=config, datasets=None)
-
-    class PytestTestCaseRunner:
-        def run(
-            self,
-            diagnostic: Diagnostic,
-            test_case_name: str = "default",
-            output_dir: Path | None = None,
-        ) -> ExecutionResult:
-            try:
-                return runner.run(diagnostic, test_case_name, output_dir)
-            except TestCaseError as e:
-                pytest.skip(str(e))
-                raise  # unreachable, but keeps type checkers happy
-
-    return PytestTestCaseRunner()
-
-
 @pytest.fixture(scope="session")
 def sample_data() -> None:
     """Download sample data if not already present."""
@@ -318,7 +290,7 @@ class MockDiagnostic(Diagnostic):
     slug = "mock"
     data_requirements = (DataRequirement(source_type=SourceDatasetType.CMIP6, filters=(), group_by=None),)
 
-    def run(self, definition: ExecutionDefinition) -> ExecutionResult:
+    def run(self, definition: ExecutionDefinition, *, capture_regression: bool = False) -> ExecutionResult:
         """Run a no-op diagnostic that always succeeds."""
         return ExecutionResult(
             output_bundle_filename=definition.output_directory / "output.json",
@@ -335,7 +307,7 @@ class FailedDiagnostic(Diagnostic):
     slug = "failed"
     data_requirements = (DataRequirement(source_type=SourceDatasetType.CMIP6, filters=(), group_by=None),)
 
-    def run(self, definition: ExecutionDefinition) -> ExecutionResult:
+    def run(self, definition: ExecutionDefinition, *, capture_regression: bool = False) -> ExecutionResult:
         """Run a diagnostic that always returns a failure result."""
         return ExecutionResult.build_from_failure(definition)
 
diff --git a/packages/climate-ref/src/climate_ref/testing.py b/packages/climate-ref/src/climate_ref/testing.py
index a177f6cc9..a35189a0a 100644
--- a/packages/climate-ref/src/climate_ref/testing.py
+++ b/packages/climate-ref/src/climate_ref/testing.py
@@ -5,10 +5,13 @@
 - Path resolution for package-local test data (catalogs, regression data)
 - Sample data fetching utilities
 - TestCaseRunner for executing diagnostics with test data
-- Result validation helpers
+- Drift-checking helpers for test-case regression baselines
+  (``assert_test_case_no_drift`` and the ``create_no_drift_test`` factory
+  used by every provider's integration test module)
 """
 
 import shutil
+from collections.abc import Callable
 from pathlib import Path
 
 from attrs import define
@@ -16,17 +19,14 @@
 
 from climate_ref import SAMPLE_DATA_VERSION
 from climate_ref.config import Config
-from climate_ref.database import Database
-from climate_ref.executor.fragment import PLACEHOLDER_FRAGMENT, assign_execution_fragment
-from climate_ref.models import Execution, ExecutionGroup
 from climate_ref_core.dataset_registry import dataset_registry_manager, fetch_all_files
 from climate_ref_core.datasets import ExecutionDatasetCollection
 from climate_ref_core.diagnostics import Diagnostic, ExecutionDefinition, ExecutionResult
 from climate_ref_core.env import env
 from climate_ref_core.exceptions import DatasetResolutionError, NoTestDataSpecError, TestCaseNotFoundError
-from climate_ref_core.testing import (
-    validate_cmec_bundles,
-)
+from climate_ref_core.providers import DiagnosticProvider
+from climate_ref_core.regression import Manifest
+from climate_ref_core.testing import TestCasePaths, collect_test_case_params, load_datasets_from_yaml
 
 
 def _determine_test_directory() -> Path | None:
@@ -88,55 +88,150 @@ def fetch_sample_data(force_cleanup: bool = False, symlink: bool = False) -> Non
         fh.write(SAMPLE_DATA_VERSION)
 
 
-def validate_result(
-    diagnostic: Diagnostic, config: Config, result: ExecutionResult
-) -> None:  # pragma: no cover
+def assert_test_case_no_drift(
+    config: Config,
+    diagnostic: Diagnostic,
+    test_case_name: str,
+    paths: TestCasePaths,
+    work_dir: Path,
+) -> None:
     """
-    Asserts the correctness of the result of a diagnostic execution
+    Execute a diagnostic test case and assert its committed bundle has not drifted.
 
-    This should only be used by the test suite as it will create a fake
-    database entry for the diagnostic execution result.
+    Runs the diagnostic against the fetched catalog using the same execute/build stages as
+    ``ref test-cases run`` (see :mod:`climate_ref.cli.test_cases._stages`),
+    rebuilds the committed bundle from the fresh execution,
+    and compares it to the tracked ``regression/`` baseline within tolerance.
+
+    Unlike ``ref test-cases replay`` this re-executes the diagnostic rather than replaying,
+    so it also proves the diagnostic still runs and emits a valid bundle.
+
+    Ingesting a committed bundle into the database is a separate concern,
+    covered by the executor result-handling tests
+    (``packages/climate-ref/tests/unit/executor/test_result_handling.py`` and friends).
+    It is intentionally not re-checked here.
+
+    Parameters
+    ----------
+    config
+        The active configuration (provides the software root and results dir).
+    diagnostic
+        The diagnostic to run.
+    test_case_name
+        Name of the test case to run.
+    paths
+        Resolved paths for the test case (catalog + tracked regression baseline).
+    work_dir
+        A per-test scratch directory for the output slot and execution outputs.
+
+    Raises
+    ------
+    NoTestDataSpecError
+        If the diagnostic has no test_data_spec.
+    TestCaseNotFoundError
+        If the test case doesn't exist.
+    AssertionError
+        If the rebuilt committed bundle drifts from the tracked baseline.
     """
-    # TODO: Remove this function once we have moved to using RegressionValidator
-    # Add a fake execution/execution group in the Database
-    database = Database.from_config(config)
-    execution_group = ExecutionGroup(
-        diagnostic_id=1, key=result.definition.key, dirty=True, selectors=result.definition.datasets.selectors
+    from climate_ref.cli.test_cases._stages import (  # noqa: PLC0415
+        baseline_placeholders,
+        stage_build,
+        stage_compare,
+        stage_execute,
     )
-    database.session.add(execution_group)
-    database.session.flush()
 
-    execution = Execution(
-        execution_group_id=execution_group.id,
-        dataset_hash=result.definition.datasets.hash,
-        output_fragment=PLACEHOLDER_FRAGMENT,
+    if diagnostic.test_data_spec is None:
+        raise NoTestDataSpecError(f"Diagnostic {diagnostic.slug} has no test_data_spec")
+
+    if not diagnostic.test_data_spec.has_case(test_case_name):
+        raise TestCaseNotFoundError(
+            f"Test case {test_case_name!r} not found. Available: {diagnostic.test_data_spec.case_names}"
+        )
+
+    tc = diagnostic.test_data_spec.get_case(test_case_name)
+    datasets = load_datasets_from_yaml(paths.catalog)
+
+    slot = work_dir / "slot"
+    slot.mkdir(parents=True, exist_ok=True)
+    placeholders = baseline_placeholders(paths, config)
+
+    # stage_execute runs the diagnostic (raising if it is not successful)
+    # and copies the curated native set into the slot
+    # stage_build rebuilds the committed bundle from it.
+    source = stage_execute(
+        config=config,
+        diag=diagnostic,
+        tc=tc,
+        datasets=datasets,
+        slot=slot,
+        execution_dir=work_dir / "exec",
+        clean=True,
     )
-    database.session.add(execution)
-
-    assign_execution_fragment(
-        database.session,
-        execution,
-        provider_slug=diagnostic.provider.slug,
-        diagnostic_slug=diagnostic.slug,
-        selectors=result.definition.datasets.selectors,
-        group_id=execution_group.id,
+    stage_build(slot=slot, source=source, placeholders=placeholders)
+
+    expected = Manifest.load(paths.manifest).committed
+    failures, _ = stage_compare(slot=slot, paths=paths, slug=diagnostic.slug, expected=expected)
+    assert not failures, (
+        f"{diagnostic.provider.slug}/{diagnostic.slug}/{test_case_name}: committed bundle drift:\n"
+        + "\n".join(failures)
     )
 
-    assert result.successful
 
-    # Validate CMEC bundles
-    validate_cmec_bundles(diagnostic, result)
+def create_no_drift_test(provider: DiagnosticProvider) -> Callable[..., None]:
+    """
+    Build the standard per-provider integration test for committed-bundle drift.
+
+    Returns a pytest function parameterized with one case per diagnostic test case
+    (via :func:`climate_ref_core.testing.collect_test_case_params`),
+    marked ``slow`` and ``test_cases``.
+    Each case configures the provider,
+    resolves the package-local test-case paths,
+    skips when the fetched catalog or committed baseline is missing,
+    and delegates to :func:`assert_test_case_no_drift`.
+
+    Requires ``ref test-cases fetch --provider <slug>`` to have been run first.
+
+    Usage in a provider's ``tests/integration/test_diagnostics.py``::
+
+        from climate_ref_example import provider
 
-    # Create a fake log file if one doesn't exist
-    if not result.to_output_path("out.log").exists():
-        result.to_output_path("out.log").touch()
+        from climate_ref.testing import create_no_drift_test
+
+        test_run_test_cases = create_no_drift_test(provider)
+
+    Parameters
+    ----------
+    provider
+        The diagnostic provider whose test cases should be exercised.
+    """
+    import pytest  # noqa: PLC0415
+
+    @pytest.mark.slow
+    @pytest.mark.test_cases
+    @pytest.mark.parametrize("diagnostic,test_case_name", collect_test_case_params(provider))
+    def test_run_test_cases(
+        diagnostic: Diagnostic,
+        test_case_name: str,
+        config: Config,
+        tmp_path: Path,
+    ) -> None:
+        """Execute the test case end-to-end and assert the committed bundle has not drifted."""
+        diagnostic.provider.configure(config)
+
+        paths = TestCasePaths.from_diagnostic(diagnostic, test_case_name)
+        if paths is None:
+            pytest.skip(f"No test-data directory for {diagnostic.slug} (not a development checkout)")
+        if not paths.catalog.exists():
+            pytest.skip(
+                f"No catalog file for {diagnostic.slug}/{test_case_name}. "
+                f"Run `ref test-cases fetch --provider {provider.slug}` first."
+            )
+        if not paths.manifest.exists() or not paths.regression.exists():
+            pytest.skip(f"No committed baseline for {diagnostic.slug}/{test_case_name}")
 
-    # Import late to avoid importing executors,
-    # some of which have on-import side effects, at package load time
-    from climate_ref.executor import handle_execution_result  # noqa: PLC0415
+        assert_test_case_no_drift(config, diagnostic, test_case_name, paths, tmp_path)
 
-    # Process and store the result
-    handle_execution_result(config, database=database, execution=execution, result=result)
+    return test_run_test_cases
 
 
 @define
@@ -231,8 +326,6 @@ def run(
             root_directory=output_dir.parent,
         )
 
-        # Run the diagnostic.
-        # This mirrors ``Diagnostic.run`` but inserts the regression-capture hook before bundling.
-        diagnostic.execute(definition)
-        diagnostic.prepare_regression_output(definition)
-        return diagnostic.build_execution_result(definition)
+        # Run the diagnostic with the regression-capture hook enabled, so the native output is
+        # normalised (prepare_regression_output) before the bundle is built and captured.
+        return diagnostic.run(definition, capture_regression=True)
diff --git a/packages/climate-ref/tests/unit/test_testing.py b/packages/climate-ref/tests/unit/test_testing.py
index ece406f47..e4f04b850 100644
--- a/packages/climate-ref/tests/unit/test_testing.py
+++ b/packages/climate-ref/tests/unit/test_testing.py
@@ -9,6 +9,8 @@
 
 from climate_ref.testing import (
     TestCaseRunner,
+    assert_test_case_no_drift,
+    create_no_drift_test,
 )
 from climate_ref_core.datasets import DatasetCollection, ExecutionDatasetCollection
 from climate_ref_core.exceptions import (
@@ -121,13 +123,13 @@ def test_run_uses_default_output_dir(self, config):
             test_cases=(TestCase(name="default", description="Default"),)
         )
         mock_result = MagicMock(successful=True)
-        mock_diagnostic.build_execution_result.return_value = mock_result
+        mock_diagnostic.run.return_value = mock_result
 
         result = runner.run(mock_diagnostic, "default")
 
         assert result.successful
-        # Verify the diagnostic was executed with an ExecutionDefinition
-        call_args = mock_diagnostic.execute.call_args[0][0]
+        # The runner delegates to Diagnostic.run with an ExecutionDefinition.
+        call_args = mock_diagnostic.run.call_args[0][0]
         assert "test-cases" in str(call_args.output_directory)
         assert "test-provider" in str(call_args.output_directory)
         assert "test-diag" in str(call_args.output_directory)
@@ -150,19 +152,18 @@ def test_run_with_explicit_datasets(self, config, tmp_path):
             )
         )
 
-        # Mock the execution phases to avoid actual execution
+        # Mock the execution to avoid actual execution
         mock_result = MagicMock()
         mock_result.successful = True
-        mock_diagnostic.build_execution_result.return_value = mock_result
+        mock_diagnostic.run.return_value = mock_result
 
         result = runner.run(mock_diagnostic, "default", output_dir=tmp_path)
 
         assert result.successful
-        # The runner drives the phases itself, inserting the regression-capture hook
-        mock_diagnostic.execute.assert_called_once()
-        mock_diagnostic.prepare_regression_output.assert_called_once()
-        mock_diagnostic.build_execution_result.assert_called_once()
-        execution_definition = mock_diagnostic.execute.call_args[0][0]
+        # The runner delegates to Diagnostic.run with the regression-capture hook enabled.
+        mock_diagnostic.run.assert_called_once()
+        execution_definition = mock_diagnostic.run.call_args[0][0]
+        assert mock_diagnostic.run.call_args.kwargs["capture_regression"] is True
         assert execution_definition.datasets == mock_datasets
 
     def test_run_without_paths(self, config, tmp_path):
@@ -182,15 +183,176 @@ def test_run_without_paths(self, config, tmp_path):
         with pytest.raises(DatasetResolutionError, match="missing the required 'path' column"):
             runner.run(mock_diagnostic, "default", output_dir=tmp_path)
 
+    def test_run_clean_removes_existing_output_dir(self, config, tmp_path):
+        """``clean=True`` wipes a pre-existing output directory before running."""
+        mock_datasets = MagicMock(spec=ExecutionDatasetCollection)
+        mock_datasets.items.return_value = []
+        runner = TestCaseRunner(config=config, datasets=mock_datasets)
 
-class TestTestCaseRunnerPytestFixture:
-    """Tests for the pytest fixture wrapper."""
-
-    def test_run_no_test_data_spec_skips(self, run_test_case):
-        """Test that the fixture skips when diagnostic has no test_data_spec."""
         mock_diagnostic = MagicMock()
-        mock_diagnostic.test_data_spec = None
         mock_diagnostic.slug = "test-diag"
+        mock_diagnostic.provider.slug = "test-provider"
+        mock_diagnostic.test_data_spec = TestDataSpecification(
+            test_cases=(TestCase(name="default", description="Default"),)
+        )
+        mock_diagnostic.run.return_value = MagicMock(successful=True)
+
+        output_dir = tmp_path / "out"
+        output_dir.mkdir()
+        stale = output_dir / "stale.txt"
+        stale.write_text("stale")
+
+        runner.run(mock_diagnostic, "default", output_dir=output_dir, clean=True)
+
+        # The stale artefact from a prior run is gone, but the directory is recreated.
+        assert output_dir.is_dir()
+        assert not stale.exists()
+
+
+class TestCreateNoDriftTest:
+    """Tests for the per-provider drift-test factory."""
+
+    def test_returns_marked_test(self, provider):
+        """The factory returns a test function carrying the standard marks."""
+        test_fn = create_no_drift_test(provider)
+
+        marks = {mark.name for mark in test_fn.pytestmark}
+        assert marks == {"slow", "test_cases", "parametrize"}
+
+    def test_collects_provider_cases(self, provider, mock_diagnostic):
+        """The parametrization contains one case per diagnostic test case."""
+        mock_diagnostic.test_data_spec = TestDataSpecification(
+            test_cases=(TestCase(name="default", description="Default case"),)
+        )
+
+        test_fn = create_no_drift_test(provider)
+
+        parametrize = next(mark for mark in test_fn.pytestmark if mark.name == "parametrize")
+        assert [param.id for param in parametrize.args[1]] == ["mock/default"]
+
+    def _make_paths(self, *, catalog=True, manifest=True, regression=True):
+        """Build a TestCasePaths mock with configurable file existence."""
+        paths = MagicMock()
+        paths.catalog.exists.return_value = catalog
+        paths.manifest.exists.return_value = manifest
+        paths.regression.exists.return_value = regression
+        return paths
+
+    def test_body_skips_when_no_test_data_dir(self, provider, config, tmp_path):
+        """The generated test skips when the provider is not a development checkout."""
+        test_fn = create_no_drift_test(provider)
+        diagnostic = MagicMock()
+        diagnostic.slug = "my-diag"
+
+        with (
+            patch("climate_ref.testing.TestCasePaths.from_diagnostic", return_value=None),
+            pytest.raises(pytest.skip.Exception, match="not a development checkout"),
+        ):
+            test_fn(diagnostic, "default", config, tmp_path)
+
+    def test_body_skips_when_catalog_missing(self, provider, config, tmp_path):
+        """The generated test skips (pointing at fetch) when the catalog is absent."""
+        test_fn = create_no_drift_test(provider)
+        diagnostic = MagicMock()
+        diagnostic.slug = "my-diag"
+        paths = self._make_paths(catalog=False)
+
+        with (
+            patch("climate_ref.testing.TestCasePaths.from_diagnostic", return_value=paths),
+            pytest.raises(pytest.skip.Exception, match="Run `ref test-cases fetch"),
+        ):
+            test_fn(diagnostic, "default", config, tmp_path)
+
+    def test_body_skips_when_baseline_missing(self, provider, config, tmp_path):
+        """The generated test skips when the committed baseline has not been minted."""
+        test_fn = create_no_drift_test(provider)
+        diagnostic = MagicMock()
+        diagnostic.slug = "my-diag"
+        paths = self._make_paths(manifest=False)
+
+        with (
+            patch("climate_ref.testing.TestCasePaths.from_diagnostic", return_value=paths),
+            pytest.raises(pytest.skip.Exception, match="No committed baseline"),
+        ):
+            test_fn(diagnostic, "default", config, tmp_path)
+
+    def test_body_delegates_when_everything_present(self, provider, config, tmp_path):
+        """When catalog and baseline exist, the body delegates to assert_test_case_no_drift."""
+        test_fn = create_no_drift_test(provider)
+        diagnostic = MagicMock()
+        diagnostic.slug = "my-diag"
+        paths = self._make_paths()
+
+        with (
+            patch("climate_ref.testing.TestCasePaths.from_diagnostic", return_value=paths),
+            patch("climate_ref.testing.assert_test_case_no_drift") as assert_no_drift,
+        ):
+            test_fn(diagnostic, "default", config, tmp_path)
+
+        diagnostic.provider.configure.assert_called_once_with(config)
+        assert_no_drift.assert_called_once_with(config, diagnostic, "default", paths, tmp_path)
+
+
+class TestAssertTestCaseNoDrift:
+    """Tests for the execute/build/compare drift assertion."""
+
+    def test_raises_without_test_data_spec(self, config, tmp_path):
+        """A diagnostic with no test_data_spec cannot be drift-checked."""
+        diagnostic = MagicMock()
+        diagnostic.slug = "my-diag"
+        diagnostic.test_data_spec = None
+
+        with pytest.raises(NoTestDataSpecError, match="no test_data_spec"):
+            assert_test_case_no_drift(config, diagnostic, "default", MagicMock(), tmp_path)
+
+    def test_raises_for_unknown_test_case(self, config, tmp_path):
+        """An unknown test case name raises a friendly error, not a bare StopIteration."""
+        diagnostic = MagicMock()
+        diagnostic.slug = "my-diag"
+        diagnostic.test_data_spec.has_case.return_value = False
+        diagnostic.test_data_spec.case_names = ["default"]
+
+        with pytest.raises(TestCaseNotFoundError, match="not found"):
+            assert_test_case_no_drift(config, diagnostic, "missing", MagicMock(), tmp_path)
+
+    def _patch_stages(self, failures):
+        """Patch the stage helpers and the manifest loader used by the drift assertion."""
+        stages = "climate_ref.cli.test_cases._stages"
+        return (
+            patch(f"{stages}.baseline_placeholders"),
+            patch(f"{stages}.stage_execute"),
+            patch(f"{stages}.stage_build"),
+            patch(f"{stages}.stage_compare", return_value=(failures, [])),
+            patch("climate_ref.testing.load_datasets_from_yaml"),
+            patch("climate_ref.testing.Manifest"),
+        )
 
-        with pytest.raises(pytest.skip.Exception):
-            run_test_case.run(mock_diagnostic)
+    def test_passes_when_no_failures(self, config, tmp_path):
+        """The happy path wires execute -> build -> compare and returns quietly."""
+        diagnostic = MagicMock()
+        diagnostic.slug = "my-diag"
+        diagnostic.provider.slug = "my-provider"
+        paths = MagicMock()
+
+        baseline_ph, execute, build, compare, load_yaml, manifest = self._patch_stages(failures=[])
+        with baseline_ph, execute as execute_m, build as build_m, compare as compare_m, load_yaml, manifest:
+            assert_test_case_no_drift(config, diagnostic, "default", paths, tmp_path)
+
+        execute_m.assert_called_once()
+        build_m.assert_called_once()
+        compare_m.assert_called_once()
+        # The output slot is created under the per-test work directory.
+        assert (tmp_path / "slot").is_dir()
+
+    def test_raises_on_drift(self, config, tmp_path):
+        """Compare failures surface as an AssertionError naming the provider/diagnostic/case."""
+        diagnostic = MagicMock()
+        diagnostic.slug = "my-diag"
+        diagnostic.provider.slug = "my-provider"
+        paths = MagicMock()
+
+        failures = ["diagnostic.json: value drift beyond tolerance"]
+        baseline_ph, execute, build, compare, load_yaml, manifest = self._patch_stages(failures)
+        with baseline_ph, execute, build, compare, load_yaml, manifest:
+            with pytest.raises(AssertionError, match="my-provider/my-diag/default: committed bundle drift"):
+                assert_test_case_no_drift(config, diagnostic, "default", paths, tmp_path)