diff --git a/.github/actions/cache-extensions/action.yml b/.github/actions/cache-extensions/action.yml
index 7ddf65de8ce..3043621f180 100644
--- a/.github/actions/cache-extensions/action.yml
+++ b/.github/actions/cache-extensions/action.yml
@@ -1,24 +1,52 @@
 name: Cache extensions
-description: Cache and reuse JIT-compiled extensions (e.g. CUDA extensions) across runs.
+description: Restore JIT-compiled CUDA extensions and Triton kernels cached by earlier runs. Pair with cache-extensions/save after the tests.
 
 inputs:
   cache-key:
-    description: Environment discriminator for the cache key (e.g. GPU arch + container image).
+    description: Environment discriminator for the cache keys (e.g. GPU arch + lane + container image).
     required: true
 
+outputs:
+  ext-key:
+    description: Key under which cache-extensions/save stores the torch extensions cache.
+    value: ${{ steps.ext.outputs.cache-primary-key }}
+  ext-hit:
+    description: "'true' if the torch extensions cache was restored with an exact key match."
+    value: ${{ steps.ext.outputs.cache-hit }}
+  triton-key:
+    description: Key under which cache-extensions/save stores the Triton cache.
+    value: ${{ steps.triton.outputs.cache-primary-key }}
+  triton-hit:
+    description: "'true' if the Triton cache was restored with an exact key match."
+    value: ${{ steps.triton.outputs.cache-hit }}
+
 runs:
   using: composite
   steps:
     - shell: bash
-      run: echo "TORCH_EXTENSIONS_DIR=/root/.cache/torch_extensions" >> "$GITHUB_ENV"
-    - id: cache
-      uses: actions/cache@v6
+      run: |
+        echo "TORCH_EXTENSIONS_DIR=/root/.cache/torch_extensions" >> "$GITHUB_ENV"
+        echo "TRITON_CACHE_DIR=/root/.cache/triton" >> "$GITHUB_ENV"
+    # CUDA extensions: exact key only, because ninja decides what to rebuild from file timestamps.
+    # Only CUDA/C++ sources and the loaders are hashed, so Triton-only edits keep the entry valid.
+    - id: ext
+      uses: actions/cache/restore@v6
       with:
         path: /root/.cache/torch_extensions
-        key: torch-ext-${{ inputs.cache-key }}-${{ hashFiles('modelopt/torch/kernels/quantization/**', 'modelopt/torch/quantization/extensions.py', 'modelopt/torch/utils/cpp_extension.py')
-          }}
+        key: torch-ext-${{ inputs.cache-key }}-${{ hashFiles('modelopt/torch/kernels/quantization/**/*.cu', 'modelopt/torch/kernels/quantization/**/*.cuh',
+          'modelopt/torch/kernels/quantization/**/*.cpp', 'modelopt/torch/kernels/quantization/**/*.h', 'modelopt/torch/kernels/quantization/conv/implicit_gemm_cuda.py',
+          'modelopt/torch/quantization/extensions.py', 'modelopt/torch/utils/cpp_extension.py') }}
     # On a cache hit, backdate sources below the cached objects so ninja reuses them (touching
     # the objects instead would desync ninja's .ninja_deps and force a rebuild).
-    - if: steps.cache.outputs.cache-hit == 'true'
+    - if: steps.ext.outputs.cache-hit == 'true'
       shell: bash
       run: find modelopt/torch/kernels/quantization -exec touch -d '2000-01-01' {} +
+    # Triton: entries are content-addressed, so restoring a cache from an older commit is safe
+    # and only costs the kernels that changed.
+    - id: triton
+      uses: actions/cache/restore@v6
+      with:
+        path: /root/.cache/triton
+        key: triton-${{ inputs.cache-key }}-${{ hashFiles('modelopt/torch/kernels/**/*.py') }}
+        restore-keys: |
+          triton-${{ inputs.cache-key }}-
diff --git a/.github/actions/cache-extensions/save/action.yml b/.github/actions/cache-extensions/save/action.yml
new file mode 100644
index 00000000000..ed54b18853c
--- /dev/null
+++ b/.github/actions/cache-extensions/save/action.yml
@@ -0,0 +1,31 @@
+name: Save extension caches
+description: Save the JIT caches that cache-extensions did not restore with an exact key match. Run it under always() so a failed or cancelled run still
+  seeds the next one.
+
+inputs:
+  ext-key:
+    description: The ext-key output of cache-extensions.
+    required: true
+  ext-hit:
+    description: The ext-hit output of cache-extensions.
+    required: true
+  triton-key:
+    description: The triton-key output of cache-extensions.
+    required: true
+  triton-hit:
+    description: The triton-hit output of cache-extensions.
+    required: true
+
+runs:
+  using: composite
+  steps:
+    - if: inputs.ext-hit != 'true'
+      uses: actions/cache/save@v6
+      with:
+        path: /root/.cache/torch_extensions
+        key: ${{ inputs.ext-key }}
+    - if: inputs.triton-hit != 'true'
+      uses: actions/cache/save@v6
+      with:
+        path: /root/.cache/triton
+        key: ${{ inputs.triton-key }}
diff --git a/.github/workflows/_example_tests_runner.yml b/.github/workflows/_example_tests_runner.yml
index e3378865b65..22664c224d8 100644
--- a/.github/workflows/_example_tests_runner.yml
+++ b/.github/workflows/_example_tests_runner.yml
@@ -58,9 +58,12 @@ jobs:
         run: apt-get update && apt-get install -y git
       - uses: actions/checkout@v6
       - uses: nv-gha-runners/setup-proxy-cache@main
-      - uses: ./.github/actions/cache-extensions
+      - id: jit-cache
+        uses: ./.github/actions/cache-extensions
         with:
-          cache-key: rtxpro6000-${{ inputs.docker_image }}
+          # One entry per example: the GPU-tests lanes and the examples that share an image would
+          # otherwise share one immutable, partially filled entry.
+          cache-key: rtxpro6000-example-${{ inputs.example }}-${{ inputs.docker_image }}
       - name: Setup environment variables
         run: |
           echo "LD_LIBRARY_PATH=${LD_LIBRARY_PATH}:/usr/include:/usr/lib/x86_64-linux-gnu:/usr/local/tensorrt/targets/x86_64-linux-gnu/lib" >> $GITHUB_ENV
@@ -113,6 +116,15 @@ jobs:
             pytest_args+=(-n 4)
           fi
           python3 -m pytest tests/examples/${{ inputs.example }} --cov "${pytest_args[@]}"
+      - name: Save JIT caches
+        # Also after a failed or cancelled run, so the next run starts warm
+        if: always()
+        uses: ./.github/actions/cache-extensions/save
+        with:
+          ext-key: ${{ steps.jit-cache.outputs.ext-key }}
+          ext-hit: ${{ steps.jit-cache.outputs.ext-hit }}
+          triton-key: ${{ steps.jit-cache.outputs.triton-key }}
+          triton-hit: ${{ steps.jit-cache.outputs.triton-hit }}
       - name: Flag allowed failure
         if: ${{ inputs.allow_failure && steps.run_tests.outcome == 'failure' }}
         run: |
diff --git a/.github/workflows/gpu_tests.yml b/.github/workflows/gpu_tests.yml
index a0389da7b62..c9ae39ebc2a 100644
--- a/.github/workflows/gpu_tests.yml
+++ b/.github/workflows/gpu_tests.yml
@@ -70,6 +70,8 @@ jobs:
         HF_TOKEN: ${{ secrets.HF_TOKEN }}
         # Build CUDA kernels only for the runner's RTX PRO 6000 (sm_120), not the image's ~6 archs.
         TORCH_CUDA_ARCH_LIST: "12.0"
+        # Where the gpu lane keeps the CUDA wheels it would otherwise compile (see noxfile.py).
+        WHEELHOUSE: /root/.cache/wheelhouse
     steps:
       - name: Install git
         # The vllm container ships without git; needed for a real checkout (correct
@@ -78,19 +80,61 @@ jobs:
         run: apt-get update && apt-get install -y git
       - uses: actions/checkout@v6
       - uses: nv-gha-runners/setup-proxy-cache@main
-      - uses: ./.github/actions/cache-extensions
+      - id: jit-cache
+        uses: ./.github/actions/cache-extensions
         with:
-          cache-key: rtxpro6000-${{ matrix.container_image }}
+          # One entry per lane: lanes sharing an image (gpu_megatron and the megatron_bridge
+          # example) would otherwise share one immutable, partially filled entry.
+          cache-key: rtxpro6000-${{ matrix.example }}-${{ matrix.container_image }}
       - name: Setup environment variables
         run: |
           echo "LD_LIBRARY_PATH=${LD_LIBRARY_PATH}:/usr/include:/usr/lib/x86_64-linux-gnu" >> $GITHUB_ENV
+      # The gpu lane compiles mamba_ssm, causal-conv1d and fast-hadamard-transform, 6-8 min. The
+      # nightly run on main builds them once per container image and ISO week (so new releases
+      # are picked up) and saves the wheels; every other run restores them. Without a cached
+      # wheelhouse the gpu session compiles them itself.
+      - name: Get the ISO week
+        id: week
+        if: matrix.example == 'gpu'
+        run: echo "iso=$(date -u +%G-W%V)" >> "$GITHUB_OUTPUT"
+      - name: Restore CUDA wheels
+        id: wheelhouse
+        if: matrix.example == 'gpu'
+        uses: actions/cache/restore@v6
+        with:
+          path: /root/.cache/wheelhouse
+          key: wheelhouse-${{ matrix.container_image }}-${{ steps.week.outputs.iso }}-${{ hashFiles('noxfile.py') }}
+      - name: Build CUDA wheels
+        if: matrix.example == 'gpu' && github.ref == 'refs/heads/main' && steps.wheelhouse.outputs.cache-hit != 'true'
+        run: python3 -m pip install nox && nox -s gpu_wheels
+      - name: Save CUDA wheels
+        if: matrix.example == 'gpu' && github.ref == 'refs/heads/main' && steps.wheelhouse.outputs.cache-hit != 'true'
+        uses: actions/cache/save@v6
+        with:
+          path: /root/.cache/wheelhouse
+          key: ${{ steps.wheelhouse.outputs.cache-primary-key }}
       - name: Run gpu tests
         env:
           COVERAGE_PROCESS_START: ${{ github.workspace }}/pyproject.toml
           COVERAGE_FILE: ${{ github.workspace }}/.coverage
+          # sys.monitoring records the same lines as the default C tracer at a fraction of the
+          # overhead on import-heavy worker processes. Needs Python 3.12+.
+          COVERAGE_CORE: sysmon
+          # Tests marked `release` (real-weight downloads and the like) run on nightly and
+          # on-demand runs, not on pull requests.
+          PYTEST_ADDOPTS: ${{ !startsWith(github.ref, 'refs/heads/pull-request/') && '--run-release' || '' }}
         run: |
           # Use `python3` (the vllm image has no `python` on PATH)
           python3 -m pip install nox && nox -s ${{ matrix.example }}
+      - name: Save JIT caches
+        # Also after a failed or cancelled run, so the next run starts warm
+        if: always()
+        uses: ./.github/actions/cache-extensions/save
+        with:
+          ext-key: ${{ steps.jit-cache.outputs.ext-key }}
+          ext-hit: ${{ steps.jit-cache.outputs.ext-hit }}
+          triton-key: ${{ steps.jit-cache.outputs.triton-key }}
+          triton-hit: ${{ steps.jit-cache.outputs.triton-hit }}
       - name: Upload GPU coverage to Codecov
         uses: codecov/codecov-action@v7
         with:
diff --git a/.github/workflows/regression_tests.yml b/.github/workflows/regression_tests.yml
index 001c4e6af4a..37f3524afce 100644
--- a/.github/workflows/regression_tests.yml
+++ b/.github/workflows/regression_tests.yml
@@ -60,6 +60,9 @@ jobs:
         env:
           COVERAGE_PROCESS_START: ${{ github.workspace }}/pyproject.toml
           COVERAGE_FILE: ${{ github.workspace }}/.coverage
+          # sys.monitoring records the same lines as the default C tracer at a fraction of the
+          # overhead on import-heavy worker processes. Needs Python 3.12+.
+          COVERAGE_CORE: sysmon
         run: python -m pip install nox && nox -s regression
       - name: Upload regression coverage to Codecov
         uses: codecov/codecov-action@v7
diff --git a/.github/workflows/unit_tests.yml b/.github/workflows/unit_tests.yml
index 6fbca185d9f..3442e6147c1 100644
--- a/.github/workflows/unit_tests.yml
+++ b/.github/workflows/unit_tests.yml
@@ -69,6 +69,12 @@ jobs:
           COVERAGE_FILE: ${{ github.workspace }}/.coverage
           # Everything but PRs also runs the slow tests marked ``manual`` (e.g. AutoNAS).
           PYTEST_EXTRA: ${{ github.event_name != 'pull_request' && '--run-manual' || '' }}
+          # sys.monitoring records the same lines as the default C tracer at a fraction of the overhead
+          COVERAGE_CORE: sysmon
+          # One single-threaded xdist worker per vCPU (`-n auto` counts physical cores); torch's
+          # default threads per worker oversubscribe the runner and run slower than serial.
+          PYTEST_ADDOPTS: -n 4 --dist loadfile
+          OMP_NUM_THREADS: "1"
         run: pip install nox uv && nox -s "unit-3.12(torch_214, tf_latest)" -- $PYTEST_EXTRA
       - name: Upload coverage reports to Codecov
         uses: codecov/codecov-action@v7
@@ -146,6 +152,9 @@ jobs:
         with:
           python-version: ${{ matrix.python_version }}
       - name: Run unit tests
+        env:
+          PYTEST_ADDOPTS: -n 4 --dist loadfile
+          OMP_NUM_THREADS: "1"
         run: pip install nox uv && nox -s "${{ matrix.nox_session }}"
   partial-install:
     if: needs.check-file-changes.outputs.any_changed == 'true'
diff --git a/docs/source/deployment/3_unified_hf.rst b/docs/source/deployment/3_unified_hf.rst
index d3dc840cc62..c231169cd23 100644
--- a/docs/source/deployment/3_unified_hf.rst
+++ b/docs/source/deployment/3_unified_hf.rst
@@ -144,8 +144,8 @@ prompts, asserting that each returns non-empty output.
 Two limits are worth stating plainly, because they bound what any ✅ below can mean:
 
 * **These are declared cases, not PR-gated coverage.** The suite is marked ``release`` and collects
-  only when pytest is given ``--run-release``, which no workflow in ``.github/workflows/`` currently
-  passes. A green check on a pull request does not mean these cases ran.
+  only when pytest is given ``--run-release``, which no example workflow in ``.github/workflows/``
+  currently passes. A green check on a pull request does not mean these cases ran.
 * **Each case is a load-and-generate smoke check on the text path.** It does not verify accuracy,
   image or audio inputs, diffusion output, or that speculative decoding actually engages.
 
diff --git a/examples/torch_onnx/torch_quant_to_onnx.py b/examples/torch_onnx/torch_quant_to_onnx.py
index bd935146e0e..a9a441e1a6d 100644
--- a/examples/torch_onnx/torch_quant_to_onnx.py
+++ b/examples/torch_onnx/torch_quant_to_onnx.py
@@ -27,7 +27,6 @@
 
 import timm
 import torch
-import torch.multiprocessing as mp
 import torch.nn.functional as F
 from datasets import load_dataset
 from download_example_onnx import export_to_onnx
@@ -56,9 +55,6 @@
 """
 
 
-mp.set_start_method("spawn", force=True)  # Needed for data loader with multiple workers
-
-
 _FP8_CONV_OVERRIDE: list = [
     {
         "parent_class": "nn.Conv2d",
@@ -311,16 +307,17 @@ def load_calibration_data(model, data_size, batch_size, device, with_labels=Fals
     calib_tensor = [transforms(img) for img in images]
     calib_tensor = [t.to(device) for t in calib_tensor]
 
+    # The samples already sit on ``device``; worker processes would only add spawn and import cost.
     if with_labels:
         labels = dataset["train"][:data_size]["label"]
         labels = torch.tensor(labels, device=device)
         calib_dataset = [{"image": img, "label": lbl} for img, lbl in zip(calib_tensor, labels)]
         return torch.utils.data.DataLoader(
-            calib_dataset, batch_size=batch_size, shuffle=True, num_workers=4
+            calib_dataset, batch_size=batch_size, shuffle=True, num_workers=0
         )
     else:
         return torch.utils.data.DataLoader(
-            calib_tensor, batch_size=batch_size, shuffle=True, num_workers=4
+            calib_tensor, batch_size=batch_size, shuffle=True, num_workers=0
         )
 
 
@@ -761,6 +758,10 @@ def build_trt_engine(onnx_path, optimization_level=4):
         f"--onnx={onnx_path}",
         "--stronglyTyped",
         f"--builderOptimizationLevel={optimization_level}",
+        # One inference iteration shows the engine runs; the default 3 s timed benchmark is not shown.
+        "--warmUp=0",
+        "--duration=0",
+        "--iterations=1",
     ]
     print(f"\nBuilding TensorRT engine: {' '.join(cmd)}")
     try:
diff --git a/noxfile.py b/noxfile.py
index 27d65b26810..7e82803ed95 100644
--- a/noxfile.py
+++ b/noxfile.py
@@ -77,11 +77,12 @@ def unit(session, torch_ver, tf_ver):
     if run_manual:
         # A separate process: some manual tests (e.g. multi-threaded from_pretrained) leak global
         # state that breaks later tests. No coverage, so it cannot overwrite the main run's data.
+        # Serial too (CI's PYTEST_ADDOPTS asks for xdist): a worker would run them in some order.
         session.run(
             "python",
             # Caller args first: the last ``-m`` wins, and it must be ours.
             *("-m", "pytest", "tests/unit", *pytest_args, "--run-manual", "-m", "manual"),
-            env={**_CPU_ONLY_ENV, "COVERAGE_PROCESS_START": ""},
+            env={**_CPU_ONLY_ENV, "COVERAGE_PROCESS_START": "", "PYTEST_ADDOPTS": ""},
         )
 
 
@@ -113,6 +114,25 @@ def partial_unit(session, subset):
 # existing Python environment (e.g. /opt/venv in NeMo) instead of an isolated one.
 # Use `python -m pip/pytest` to ensure the container's active venv Python is used,
 # not a stale PATH entry (e.g. NeMo container has pip → /usr/local/bin/pip but python → /opt/venv/bin/python).
+
+# CUDA packages the gpu session compiles against the container's torch, which takes ~7 min together
+# (wheel file prefix -> requirement). CI builds them once with `nox -s gpu_wheels` into $WHEELHOUSE
+# (an Actions cache); the gpu session installs the wheels found there instead of compiling again.
+_WHEELHOUSE = os.environ.get("WHEELHOUSE")
+_CUDA_BUILDS = {
+    "fast_hadamard_transform": "git+https://github.com/Dao-AILab/fast-hadamard-transform.git",
+    # Latest *released* sdists
+    "mamba_ssm": "mamba_ssm",
+    "causal_conv1d": "causal-conv1d",
+}
+
+
+def _cuda_build(name):
+    """The wheel of ``name`` from the wheelhouse if it has one, else the requirement to build it from."""
+    wheels = sorted(glob.glob(f"{_WHEELHOUSE}/{name}-*.whl")) if _WHEELHOUSE else []
+    return wheels[-1] if wheels else _CUDA_BUILDS[name]
+
+
 # Container: nvcr.io/nvidia/pytorch:26.07-py3 or later with CUDA 13
 @nox.session(venv_backend="none")
 def gpu(session):
@@ -124,7 +144,7 @@ def gpu(session):
         "pip",
         "install",
         "--no-build-isolation",
-        "git+https://github.com/Dao-AILab/fast-hadamard-transform.git",
+        _cuda_build("fast_hadamard_transform"),
     )
     session.run("python", "-m", "pip", "install", "-e", ".[all,dev-test,dev-fla]")
     session.run(
@@ -133,13 +153,32 @@ def gpu(session):
         "pip",
         "install",
         "--no-build-isolation",
-        # Install the latest *released* sdists (built against the container torch)
-        "mamba_ssm",
-        "causal-conv1d",
+        _cuda_build("mamba_ssm"),
+        _cuda_build("causal_conv1d"),
     )
     session.run("python", "-m", "pytest", "tests/gpu", *_cov_args())
 
 
+# Container: same as gpu
+@nox.session(venv_backend="none")
+def gpu_wheels(session):
+    """Build the CUDA wheels of the gpu session into $WHEELHOUSE."""
+    if not _WHEELHOUSE:
+        session.error("Set WHEELHOUSE to the directory the wheels are built into")
+    for requirement in _CUDA_BUILDS.values():
+        session.run(
+            "python",
+            "-m",
+            "pip",
+            "wheel",
+            "--no-build-isolation",
+            "--no-deps",
+            "-w",
+            _WHEELHOUSE,
+            requirement,
+        )
+
+
 # Container: nvcr.io/nvidia/nemo:26.08 or later
 @nox.session(venv_backend="none")
 def gpu_megatron(session):
diff --git a/pyproject.toml b/pyproject.toml
index 8843bcfc23e..c98129247cb 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -362,8 +362,9 @@ skips = [
 [tool.pytest.ini_options]
 # Default additional options
 # Show a short test summary info for all except passed tests with -ra flag
-# print execution time for 50 slowest tests and generate coverage reports
-addopts = "-v -ra --instafail --cov-report=term-missing --cov-report=html --cov-report=xml:coverage.xml --cov-config=pyproject.toml --durations=50 --strict-markers"
+# print execution time for 50 slowest tests and write the coverage.xml that CI uploads
+# (add --cov-report=html or --cov-report=term-missing locally; each adds 5-20 s to the end of a run)
+addopts = "-v -ra --instafail --cov-report=xml:coverage.xml --cov-config=pyproject.toml --durations=50 --strict-markers"
 pythonpath = ["tests/"]
 # Apply per-test timeouts (see tests/conftest.py) to the test call only, not fixture setup/teardown
 timeout_func_only = true
diff --git a/tests/_test_utils/examples/example_runner.py b/tests/_test_utils/examples/example_runner.py
new file mode 100644
index 00000000000..5895d960b88
--- /dev/null
+++ b/tests/_test_utils/examples/example_runner.py
@@ -0,0 +1,442 @@
+# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+# SPDX-License-Identifier: Apache-2.0
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+"""Run ``python <script>.py`` example steps in one long-lived worker instead of one launch per step.
+
+A fresh interpreter spends 10-20 s importing torch, transformers and modelopt before an example does any
+work, and an example test runs several steps. The worker pays that once. Each step runs like
+``python script.py`` (``runpy`` as ``__main__``, ``sys.argv``, the example directory as cwd and first on
+``sys.path``); afterwards everything it changed in the process is put back, so the next step starts clean
+while the imports stay cached. Environment variables are applied to ``os.environ``, so those read at
+interpreter start-up or at import time keep the worker's values.
+
+The worker is a separate process, not the pytest process, so a hung step can be killed (that step fails and
+the next one gets a fresh worker) and pytest holds no CUDA context. The worker is also recycled after any
+failed step. ``run_command.run_example_command`` hands an installed ``ExampleRunner`` the steps it
+``accepts``; every other step, and every step when ``MODELOPT_EXAMPLE_RUNNER=subprocess``, still runs as a
+subprocess. ``MODELOPT_EXAMPLE_RUNNER=inprocess`` runs the steps in the calling process instead: no
+isolation, but ``pdb`` works.
+"""
+
+import contextlib
+import gc
+import importlib
+import logging
+import multiprocessing
+import os
+import random
+import runpy
+import shutil
+import signal
+import subprocess
+import sys
+import tempfile
+import traceback
+import warnings
+from collections.abc import Callable, Iterable
+from pathlib import Path
+
+EXAMPLES_DIR = Path(__file__).parents[3] / "examples"
+
+_SIGNALS = [getattr(signal, n) for n in ("SIGTERM", "SIGINT", "SIGHUP") if hasattr(signal, n)]
+# A clean worker exit lets its atexit handlers run, which is how coverage writes its data.
+_WORKER_EXIT_TIMEOUT_S = 30
+_KILL_TIMEOUT_S = 5
+
+
+class StepFailed(subprocess.CalledProcessError):
+    """A step exited non-zero, raised, or took its worker down; ``output`` is what it printed."""
+
+    def __init__(self, returncode: int, cmd: list[str], output: str):
+        super().__init__(returncode, cmd, output=output)
+        # What run_example_command scans for transient Hub errors.
+        self.captured_output = output
+
+
+class StepTimeout(subprocess.TimeoutExpired):
+    """A step outlived its timeout and its worker was killed; ``output`` is what it had printed."""
+
+    def __init__(self, cmd: list[str], timeout: float, output: str):
+        super().__init__(cmd, timeout, output=output)
+        self.captured_output = output
+
+
+def _loggers() -> list[logging.Logger]:
+    with logging._lock:  # a getLogger() on another thread mutates loggerDict
+        named = [
+            lg for lg in logging.root.manager.loggerDict.values() if isinstance(lg, logging.Logger)
+        ]
+    return [logging.root, *named]
+
+
+def _torch_flags(torch) -> list[tuple]:
+    """The (object, attribute) pairs a script may flip, e.g. onnx_ptq/evaluation.py sets cudnn's."""
+    backends = torch.backends
+    return [
+        (backends.cudnn, "deterministic"),
+        (backends.cudnn, "benchmark"),
+        (backends.cudnn, "allow_tf32"),
+        (backends.cuda.matmul, "allow_tf32"),
+    ]
+
+
+def _purge_example_modules(before: set[str], examples_dir: Path) -> set[str]:
+    """Drop modules the step imported from ``examples/``: ``utils``, ``config``, sibling example dirs.
+
+    The next step may bring a different module of the same name. Third-party and modelopt modules
+    stay cached, which is the saving.
+    """
+    root = examples_dir.resolve()
+    purged = set()
+    for name in set(sys.modules) - before:
+        file = getattr(sys.modules[name], "__file__", None)
+        if file and Path(file).resolve().is_relative_to(root):
+            purged.add(name)
+            del sys.modules[name]
+    return purged
+
+
+def _restore_loggers(snapshot: dict, purged: set[str]) -> None:
+    """Restore levels everywhere, handler lists only where the script owns them.
+
+    That is root, ``__main__`` and loggers named after purged modules: a library may configure its
+    own logger lazily during a step and must keep its handler.
+    """
+    owned = {"__main__", *(name.split(".")[0] for name in purged)}
+    for lg in _loggers():
+        is_owned = lg is logging.root or lg.name.split(".")[0] in owned
+        if lg in snapshot:
+            handlers, level, propagate, disabled = snapshot[lg]
+            lg.setLevel(level)
+            lg.propagate, lg.disabled = propagate, disabled
+            if is_owned:
+                lg.handlers[:] = handlers
+        elif is_owned:
+            lg.handlers[:] = []
+            lg.setLevel(logging.NOTSET)
+
+
+@contextlib.contextmanager
+def _restored_state(examples_dir: Path):
+    """Put back what a step changed in the process, which ``python script.py`` discards on exit."""
+    argv, path, cwd, environ = list(sys.argv), list(sys.path), os.getcwd(), dict(os.environ)
+    modules = set(sys.modules)
+    signal_handlers = {sig: signal.getsignal(sig) for sig in _SIGNALS}
+    loggers = {lg: (list(lg.handlers), lg.level, lg.propagate, lg.disabled) for lg in _loggers()}
+    log_disable = logging.root.manager.disable
+    start_method = multiprocessing.get_start_method(allow_none=True)
+    py_random = random.getstate()
+    numpy = sys.modules.get("numpy")
+    np_random = numpy.random.get_state() if numpy else None
+    torch = sys.modules.get("torch")
+    torch_rng = torch.get_rng_state() if torch else None
+    torch_flags = (
+        [(obj, name, getattr(obj, name)) for obj, name in _torch_flags(torch)] if torch else []
+    )
+    with warnings.catch_warnings():
+        try:
+            yield
+        finally:
+            sys.argv[:] = argv
+            sys.path[:] = path
+            os.chdir(cwd)
+            os.environ.clear()
+            os.environ.update(environ)
+            for sig, handler in signal_handlers.items():
+                # A handler installed from C reads back as None and cannot be restored.
+                if handler is not None:
+                    signal.signal(sig, handler)
+            logging.disable(log_disable)
+            if multiprocessing.get_start_method(allow_none=True) != start_method:
+                multiprocessing.set_start_method(start_method, force=True)
+            random.setstate(py_random)
+            if np_random is not None:
+                numpy.random.set_state(np_random)
+            if torch is not None:
+                torch.set_rng_state(torch_rng)
+                for obj, name, value in torch_flags:
+                    setattr(obj, name, value)
+            _restore_loggers(loggers, _purge_example_modules(modules, examples_dir))
+            gc.collect()
+            if torch is not None and torch.cuda.is_initialized():
+                torch.cuda.empty_cache()
+
+
+def _drain(sink) -> str:
+    sink.flush()
+    sink.seek(0)
+    return sink.read()
+
+
+def run_step(
+    cmd: list[str],
+    cwd: Path | str,
+    env: dict[str, str] | None,
+    sink,
+    hooks: Iterable[Callable] = (),
+    examples_dir: Path = EXAMPLES_DIR,
+) -> tuple[int, str]:
+    """Run ``cmd`` (``[python, script, *args]``) in this process as ``python script ...`` in ``cwd``.
+
+    Everything written to fds 1 and 2, by this process or its children, goes to ``sink``. ``env``
+    replaces ``os.environ`` for the step. Each of ``hooks`` is a callable returning a context
+    manager that wraps the step, for global state the generic restore does not know about.
+    Returns ``(returncode, output)``. Only a ``BaseException`` that is not a script failure, such
+    as ``KeyboardInterrupt`` or a pytest timeout, propagates, with ``captured_output`` set.
+    """
+    script = Path(cwd, cmd[1])
+    sink.seek(0)
+    sink.truncate()
+    sys.stdout.flush()
+    sys.stderr.flush()
+    saved_fds = (os.dup(1), os.dup(2))
+    saved_streams = (sys.stdout, sys.stderr)
+    returncode = 0
+    try:
+        with _restored_state(examples_dir), contextlib.ExitStack() as stack:
+            try:
+                # Child processes and C extensions write to the fds; pytest's capture replaces
+                # sys.stdout and sys.stderr with objects over its own files, so swap those too.
+                os.dup2(sink.fileno(), 1)
+                os.dup2(sink.fileno(), 2)
+                sys.stdout = sys.stderr = sink
+                os.chdir(cwd)
+                if env is not None:
+                    os.environ.clear()
+                    os.environ.update(env)
+                sys.argv[:] = [str(script), *cmd[2:]]
+                sys.path.insert(0, str(script.parent))
+                importlib.invalidate_caches()
+                for hook in hooks:
+                    stack.enter_context(hook())
+                runpy.run_path(str(script), run_name="__main__")
+            except SystemExit as e:  # argparse errors, ``sys.exit(1)`` after a logged failure
+                if e.code not in (0, None):
+                    returncode = e.code if isinstance(e.code, int) else 1
+                    if not isinstance(e.code, int):
+                        print(e.code, file=sys.stderr)
+            except Exception:
+                traceback.print_exc()
+                returncode = 1
+    except BaseException as e:
+        e.captured_output = _drain(sink)
+        raise
+    finally:
+        sys.stdout, sys.stderr = saved_streams
+        os.dup2(saved_fds[0], 1)
+        os.dup2(saved_fds[1], 2)
+        os.close(saved_fds[0])
+        os.close(saved_fds[1])
+    return returncode, _drain(sink)
+
+
+@contextlib.contextmanager
+def keep_post_conversion_plugins():
+    """Hook undoing a step's ``CUSTOM_POST_CONVERSION_PLUGINS.add`` calls.
+
+    torch_quant_to_onnx.py registers a ResNet plugin there, which would apply to every later step.
+    """
+    from modelopt.torch.quantization.plugins import custom
+
+    plugins = custom.CUSTOM_POST_CONVERSION_PLUGINS
+    saved = set(plugins)
+    try:
+        yield
+    finally:
+        plugins.clear()
+        plugins.update(saved)
+
+
+def _preload(modules: Iterable[str]) -> None:
+    """Import what ``_restored_state`` snapshots, so a first step cannot change it unnoticed."""
+    for name in modules:
+        with contextlib.suppress(ImportError):
+            importlib.import_module(name)
+
+
+def _open_sink(path: str):
+    return open(path, "w+", encoding="utf-8", errors="replace", buffering=1)
+
+
+def _worker_main(conn, sink_path, hooks, examples_dir, preload) -> None:
+    # Own session and process group: killing the worker takes a step's children along.
+    os.setsid()
+    # Steps may start DataLoader workers or Pools, which a daemonic process is not allowed to.
+    multiprocessing.current_process().daemon = False
+    _preload(preload)
+    sink = _open_sink(sink_path)
+    while True:
+        try:
+            request = conn.recv()
+        except EOFError:
+            return
+        if request is None:
+            return
+        cmd, cwd, env = request
+        try:
+            reply = run_step(cmd, cwd, env, sink, hooks, Path(examples_dir))
+        except BaseException:
+            reply = (1, traceback.format_exc())
+        conn.send(reply)
+
+
+class ExampleRunner:
+    """Run the example steps listed in ``scripts`` in a long-lived worker process.
+
+    Install it with ``run_command.set_in_process_runner``. ``scripts`` are the names of the scripts
+    (as in ``["python", name, ...]``) known to be safe to run repeatedly in one process. ``hooks``
+    wrap every step, see ``run_step``; the worker imports them, so they must be module-level.
+    ``timeout`` bounds one step; by default only pytest-timeout does, and it kills the worker too.
+    ``preload`` names modules imported before the first step.
+    """
+
+    def __init__(
+        self,
+        scripts: Iterable[str],
+        hooks: Iterable[Callable] = (),
+        examples_dir: Path | str = EXAMPLES_DIR,
+        timeout: float | None = None,
+        preload: Iterable[str] = ("torch",),
+        in_process: bool | None = None,
+    ):
+        self.scripts = frozenset(scripts)
+        self._hooks = tuple(hooks)
+        self._examples_dir = Path(examples_dir)
+        self._timeout = timeout
+        self._preload = tuple(preload)
+        if in_process is None:
+            in_process = os.environ.get("MODELOPT_EXAMPLE_RUNNER") == "inprocess"
+        self._in_process = in_process
+        self._sink_dir = tempfile.mkdtemp(prefix="example_runner_")
+        self._sink_path = os.path.join(self._sink_dir, "output.txt")
+        # In-process only, and never closed: a handler built during a step may still hold it.
+        self._sink = None
+        self._proc = None
+        self._conn = None
+
+    def accepts(self, cmd_parts: list[str]) -> bool:
+        """Whether the step is ``python <one of scripts> ...``."""
+        interpreters = {"python", "python3", Path(sys.executable).name}
+        return (
+            len(cmd_parts) > 1
+            and Path(str(cmd_parts[0])).name in interpreters
+            and str(cmd_parts[1]) in self.scripts
+        )
+
+    def __call__(
+        self,
+        cmd_parts: list[str],
+        example_path: str,
+        env: dict[str, str] | None = None,
+        timeout: float | None = None,
+    ) -> str:
+        """Run a step in ``examples/<example_path>`` and return its output.
+
+        Raises ``StepFailed`` or ``StepTimeout``. ``env`` is the step's whole environment; ``None``
+        keeps the worker's own.
+        """
+        cmd = [str(part) for part in cmd_parts]
+        cwd = self._examples_dir / example_path
+        timeout = self._timeout if timeout is None else timeout
+        output = ""
+        try:
+            if self._in_process:
+                output = self._run_in_process(cmd, cwd, env)
+            else:
+                output = self._run_in_worker(cmd, cwd, env, timeout)
+            return output
+        except BaseException as e:
+            output = getattr(e, "captured_output", "")
+            raise
+        finally:
+            print(output)  # keep the step's output in the test log
+
+    def close(self) -> None:
+        self._stop_worker()
+        shutil.rmtree(self._sink_dir, ignore_errors=True)
+
+    def _run_in_process(self, cmd: list[str], cwd: Path, env: dict[str, str] | None) -> str:
+        if self._sink is None:
+            _preload(self._preload)
+            self._sink = _open_sink(self._sink_path)
+        returncode, output = run_step(cmd, cwd, env, self._sink, self._hooks, self._examples_dir)
+        if returncode != 0:
+            raise StepFailed(returncode, cmd, output)
+        return output
+
+    def _run_in_worker(
+        self, cmd: list[str], cwd: Path, env: dict[str, str] | None, timeout: float | None
+    ) -> str:
+        self._start_worker()
+        try:
+            self._conn.send((cmd, str(cwd), env))
+            reply = self._conn.recv() if self._conn.poll(timeout) else None
+        except (EOFError, OSError):
+            # The worker died during the step: segfault, OOM kill.
+            output = self._partial_output()
+            raise StepFailed(self._kill_worker(), cmd, output) from None
+        except BaseException as e:
+            # Interrupted while waiting, e.g. by pytest-timeout: do not leave the step running.
+            e.captured_output = self._partial_output()
+            self._kill_worker()
+            raise
+        if reply is None:
+            # Hung: only this step fails, the next one gets a fresh worker.
+            output = self._partial_output()
+            self._kill_worker()
+            raise StepTimeout(cmd, timeout, output)
+        returncode, output = reply
+        if returncode != 0:
+            self._stop_worker()  # do not trust the state a failed step left behind
+            raise StepFailed(returncode, cmd, output)
+        return output
+
+    def _start_worker(self) -> None:
+        if self._proc is not None and self._proc.is_alive():
+            return
+        context = multiprocessing.get_context("spawn")
+        self._conn, child_conn = context.Pipe()
+        args = (child_conn, self._sink_path, self._hooks, str(self._examples_dir), self._preload)
+        self._proc = context.Process(target=_worker_main, args=args, daemon=True)
+        self._proc.start()
+        child_conn.close()  # only the worker holds this end, so its death shows up as EOF
+
+    def _partial_output(self) -> str:
+        try:
+            return Path(self._sink_path).read_text(encoding="utf-8", errors="replace")
+        except OSError:
+            return ""
+
+    def _kill_worker(self) -> int:
+        """SIGKILL the worker's process group and return the worker's exit code."""
+        proc, self._proc = self._proc, None
+        with contextlib.suppress(ProcessLookupError, PermissionError):
+            # Before the pid is reaped, so it cannot have been recycled.
+            os.killpg(proc.pid, signal.SIGKILL)
+        proc.join(_KILL_TIMEOUT_S)
+        return -1 if proc.exitcode is None else proc.exitcode
+
+    def _stop_worker(self) -> None:
+        """Let the worker exit on its own, so its coverage data is written, or kill it."""
+        proc = self._proc
+        if proc is None:
+            return
+        if proc.is_alive():
+            with contextlib.suppress(OSError):
+                self._conn.send(None)
+            proc.join(_WORKER_EXIT_TIMEOUT_S)
+        if proc.is_alive():
+            self._kill_worker()
+        else:
+            self._proc = None
diff --git a/tests/_test_utils/examples/run_command.py b/tests/_test_utils/examples/run_command.py
index 218106f9f89..f7aee9d95f8 100644
--- a/tests/_test_utils/examples/run_command.py
+++ b/tests/_test_utils/examples/run_command.py
@@ -137,13 +137,20 @@ def _drain():
     return returncode, "".join(chunks)
 
 
-# Set by a suite's conftest to run a command without spawning a subprocess (see
-# ``_test_utils.examples.megatron_example_runner``). Unset means always use a subprocess.
+# Set by a suite's conftest to run steps without a subprocess each. Unset means always use one, and so
+# does MODELOPT_EXAMPLE_RUNNER=subprocess (to bisect a suspected leak between steps).
 _in_process_runner = None
 
 
 def set_in_process_runner(runner) -> None:
-    """Register ``runner(cmd_parts, example_path) -> str``; it must not fall back to a subprocess."""
+    """Register either of two kinds of runner.
+
+    A plain callable ``runner(cmd_parts, example_path) -> str`` runs every step in the ambient
+    environment and must not fall back to a subprocess (``megatron_example_runner``). An object
+    with ``accepts(cmd_parts)`` and ``runner(cmd_parts, example_path, env=env) -> str`` takes the
+    step's environment and only the steps it accepts; the rest run as subprocesses
+    (``example_runner.ExampleRunner``).
+    """
     global _in_process_runner
     _in_process_runner = runner
 
@@ -158,24 +165,32 @@ def run_example_command(
 ) -> str | None:
     """Run an example command, retrying transient HuggingFace access errors."""
     print(f"[{example_path}] Running command: {cmd_parts}")
-    in_process = _in_process_runner if env is None else None
-    if _in_process_runner is not None and env is not None:
-        # The in-process runner uses the ambient environment, so a caller-supplied env would be
+    in_process = (
+        None if os.environ.get("MODELOPT_EXAMPLE_RUNNER") == "subprocess" else _in_process_runner
+    )
+    takes_env = hasattr(in_process, "accepts")
+    if takes_env and not in_process.accepts(cmd_parts):
+        in_process = None
+    elif in_process is not None and not takes_env and env is not None:
+        # A plain in-process runner uses the ambient environment, so a caller-supplied env would be
         # silently dropped. Fall back to a subprocess rather than run with the wrong environment.
         warnings.warn(f"[{example_path}] env= given; running this step as a subprocess")
+        in_process = None
     env = os.environ.copy() if env is None else env
     cwd = MODELOPT_ROOT / "examples" / example_path
 
     for attempt in range(hf_max_retries + 1):
         if setup_free_port:
-            # Subprocess steps only: an in-process runner picks its own free port, since env is a
-            # copy it never sees.
+            # A plain in-process runner picks its own free port, since env is a copy it never sees.
             env["MASTER_PORT"] = str(get_free_port())  # fresh port per attempt
         if in_process is not None:
             # Inside the loop so in-process steps get the same transient-HuggingFace retries;
             # these tests do hit the Hub (e.g. calib_dataset_name="cnn_dailymail").
             try:
-                result = in_process(cmd_parts, example_path)
+                if takes_env:
+                    result = in_process(cmd_parts, example_path, env=env)
+                else:
+                    result = in_process(cmd_parts, example_path)
             except Exception as e:
                 # Re-raise unless it looks transient, so a real failure keeps its traceback
                 # instead of being flattened into CalledProcessError.
diff --git a/tests/_test_utils/onnx/lib_test_models.py b/tests/_test_utils/onnx/lib_test_models.py
index 57a0621c5da..12dba85b20f 100644
--- a/tests/_test_utils/onnx/lib_test_models.py
+++ b/tests/_test_utils/onnx/lib_test_models.py
@@ -260,12 +260,12 @@ def find_init(onnx_model: onnx.ModelProto, init_name: str) -> np.ndarray:
 #           conv_4
 #             |
 #
-def build_conv_concat_model():
+def build_conv_concat_model(channels=128, height=240, width=64):
     # Define your model inputs and outputs
     input_names = ["input_0"]
     output_names = ["conv4_conv/Conv2D:0"]
-    input_shapes = [(1, 128, 240, 64)]
-    output_shapes = [(1, 256, 240, 64)]
+    input_shapes = [(1, channels, height, width)]
+    output_shapes = [(1, 2 * channels, height, width)]
 
     inputs = [
         helper.make_tensor_value_info(input_name, onnx.TensorProto.FLOAT, input_shape)
@@ -340,26 +340,26 @@ def build_conv_concat_model():
         helper.make_tensor(
             name="ReadVariableOp:0",
             data_type=onnx.TensorProto.FLOAT,
-            dims=(128, 128, 3, 3),
-            vals=np.random.uniform(low=0.5, high=1.0, size=128 * 128 * 3 * 3),
+            dims=(channels, channels, 3, 3),
+            vals=np.random.uniform(low=0.5, high=1.0, size=channels * channels * 3 * 3),
         ),
         helper.make_tensor(
             name="ReadVariableOp:1",
             data_type=onnx.TensorProto.FLOAT,
-            dims=(128, 128, 3, 3),
-            vals=np.random.uniform(low=0.5, high=1.0, size=128 * 128 * 3 * 3),
+            dims=(channels, channels, 3, 3),
+            vals=np.random.uniform(low=0.5, high=1.0, size=channels * channels * 3 * 3),
         ),
         helper.make_tensor(
             name="ReadVariableOp:2",
             data_type=onnx.TensorProto.FLOAT,
-            dims=(128, 128, 3, 3),
-            vals=np.random.uniform(low=0.5, high=1.0, size=128 * 128 * 3 * 3),
+            dims=(channels, channels, 3, 3),
+            vals=np.random.uniform(low=0.5, high=1.0, size=channels * channels * 3 * 3),
         ),
         helper.make_tensor(
             name="ReadVariableOp:3",
             data_type=onnx.TensorProto.FLOAT,
-            dims=(256, 384, 1, 1),
-            vals=np.random.uniform(low=0.5, high=1.0, size=256 * 384 * 1 * 1),
+            dims=(2 * channels, 3 * channels, 1, 1),
+            vals=np.random.uniform(low=0.5, high=1.0, size=2 * channels * 3 * channels),
         ),
     ]
 
diff --git a/tests/_test_utils/onnx/quantization/autotune/models.py b/tests/_test_utils/onnx/quantization/autotune/models.py
index 84a8b4ab82e..89721661b4a 100644
--- a/tests/_test_utils/onnx/quantization/autotune/models.py
+++ b/tests/_test_utils/onnx/quantization/autotune/models.py
@@ -25,13 +25,13 @@
 from onnx import helper
 
 
-def _create_simple_conv_onnx_model():
+def _create_simple_conv_onnx_model(batch_size=64, spatial_size=224):
     """Build ONNX model: Input -> Conv -> Relu -> Output (minimal for autotuner tests)."""
     input_tensor = helper.make_tensor_value_info(
-        "input", onnx.TensorProto.FLOAT, [64, 32, 224, 224]
+        "input", onnx.TensorProto.FLOAT, [batch_size, 32, spatial_size, spatial_size]
     )
     output_tensor = helper.make_tensor_value_info(
-        "output", onnx.TensorProto.FLOAT, [64, 64, 224, 224]
+        "output", onnx.TensorProto.FLOAT, [batch_size, 64, spatial_size, spatial_size]
     )
     conv_node = helper.make_node(
         "Conv",
diff --git a/tests/_test_utils/torch/calib_data.py b/tests/_test_utils/torch/calib_data.py
new file mode 100644
index 00000000000..03622986d82
--- /dev/null
+++ b/tests/_test_utils/torch/calib_data.py
@@ -0,0 +1,41 @@
+# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+# SPDX-License-Identifier: Apache-2.0
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+"""Local calibration text for tests that only need some text, not a particular Hub dataset."""
+
+import json
+import random
+from pathlib import Path
+
+__all__ = ["write_calib_jsonl"]
+
+_TEXT = (
+    "the of and to in is that for it as was with be by on not he this are or his from at which "
+    "but have an had they you were their one all we can her has there been if more when will "
+    "would who so no model layer weight scale range value sample batch token cache quantize"
+)
+_WORDS = _TEXT.split()
+
+
+def write_calib_jsonl(path: Path | str, num_samples: int = 64, words_per_sample: int = 150) -> str:
+    """Write ``num_samples`` random-word texts to ``path`` as ``{"text": ...}`` JSON lines.
+
+    Every text is long enough to fill the default 512-token calibration window with a character
+    level tokenizer. The texts differ from each other and are the same on every call.
+    """
+    rng = random.Random(0)
+    texts = [" ".join(rng.choices(_WORDS, k=words_per_sample)) for _ in range(num_samples)]
+    Path(path).write_text("".join(json.dumps({"text": t}) + "\n" for t in texts), encoding="utf-8")
+    return str(path)
diff --git a/tests/_test_utils/torch/distributed/utils.py b/tests/_test_utils/torch/distributed/utils.py
index ea954efbcf5..515b3207e54 100644
--- a/tests/_test_utils/torch/distributed/utils.py
+++ b/tests/_test_utils/torch/distributed/utils.py
@@ -15,6 +15,7 @@
 
 import contextlib
 import faulthandler
+import gc
 import os
 import queue
 import signal
@@ -96,6 +97,53 @@ def get_live_pool(pools, key, make_pool):
     return pool
 
 
+def destroy_extra_process_groups():
+    """Destroy every process group except the default one.
+
+    Libraries that create groups per model (e.g. Megatron's ``initialize_model_parallel``, ~20 NCCL
+    groups) drop their references on teardown but not the groups themselves. In a worker that
+    lives through hundreds of jobs they pile up (communicators, one watchdog thread each).
+    """
+    from torch.distributed.distributed_c10d import _get_default_group, _world
+
+    default = _get_default_group()
+    for group in [g for g in _world.pg_map if g is not default]:
+        try:
+            dist.destroy_process_group(group)
+        except Exception as e:  # noqa: PERF203
+            print(f"Error destroying process group: {e}")
+
+
+_BASELINE_TORCH_FLAGS = {}
+
+
+def _capture_baseline_torch_flags():
+    """Remember the torch settings a freshly spawned worker starts with."""
+    _BASELINE_TORCH_FLAGS.update(
+        cudnn_deterministic=torch.backends.cudnn.deterministic,
+        cudnn_benchmark=torch.backends.cudnn.benchmark,
+        cudnn_tf32=torch.backends.cudnn.allow_tf32,
+        matmul_tf32=torch.backends.cuda.matmul.allow_tf32,
+        deterministic_algorithms=torch.are_deterministic_algorithms_enabled(),
+    )
+
+
+def reset_worker_state(rank, world_size):
+    """Return a persistent worker to the state it had right after spawning.
+
+    ``set_seed`` and similar helpers change process-global torch settings; run this job at the
+    boundary between test modules that share workers.
+    """
+    torch.backends.cudnn.deterministic = _BASELINE_TORCH_FLAGS["cudnn_deterministic"]
+    torch.backends.cudnn.benchmark = _BASELINE_TORCH_FLAGS["cudnn_benchmark"]
+    torch.backends.cudnn.allow_tf32 = _BASELINE_TORCH_FLAGS["cudnn_tf32"]
+    torch.backends.cuda.matmul.allow_tf32 = _BASELINE_TORCH_FLAGS["matmul_tf32"]
+    torch.use_deterministic_algorithms(_BASELINE_TORCH_FLAGS["deterministic_algorithms"])
+    torch.manual_seed(1234)
+    gc.collect()
+    torch.cuda.empty_cache()
+
+
 def default_worker_teardown(rank, world_size):
     """Minimal cleanup between tests in persistent workers."""
     try:
@@ -131,7 +179,9 @@ def _test_fn(rank, size): ...
         pool.shutdown()
     """
 
-    def __init__(self, world_size, backend="nccl", teardown_fn=default_worker_teardown):
+    def __init__(
+        self, world_size, backend="nccl", teardown_fn=default_worker_teardown, ready_timeout=300
+    ):
         assert world_size > 0, "World size must be greater than 0"
         self.world_size = world_size
         # Set once a job was interrupted: the workers are then killed, see ``run``.
@@ -160,9 +210,37 @@ def __init__(self, world_size, backend="nccl", teardown_fn=default_worker_teardo
             p.start()
             self._processes.append(p)
 
-        for _ in range(world_size):
-            msg = self._result_queue.get(timeout=120)
-            assert msg == "ready", f"Worker failed to initialize: {msg}"
+        try:
+            for _ in range(world_size):
+                # Cold imports of the Megatron/TE/modelopt stack take 75-100 s under coverage in CI
+                msg = self._get(ready_timeout)
+                assert msg == "ready", f"Worker failed to initialize: {msg}"
+        except BaseException:
+            self._abort()  # do not leave workers behind when the start fails
+            raise
+
+    def _get(self, timeout):
+        """Return the next message from the workers; fail within a second if a worker died.
+
+        A worker that exits (an illegal memory access aborts the process, an OOM kill) would
+        otherwise only be noticed when ``timeout`` runs out.
+        """
+        deadline = time.monotonic() + timeout
+        while True:
+            try:
+                return self._result_queue.get(
+                    timeout=max(0.01, min(1.0, deadline - time.monotonic()))
+                )
+            except queue.Empty:  # noqa: PERF203
+                dead = [
+                    (rank, p.exitcode) for rank, p in enumerate(self._processes) if not p.is_alive()
+                ]
+                if dead:
+                    raise RuntimeError(
+                        f"Worker process(es) died (rank, exit code): {dead}"
+                    ) from None
+                if time.monotonic() >= deadline:
+                    raise
 
     @staticmethod
     def _worker_loop(rank, world_size, backend, port, cmd_queue, result_queue, teardown_fn):
@@ -178,6 +256,7 @@ def _worker_loop(rank, world_size, backend, port, cmd_queue, result_queue, teard
         if backend == "nccl" and torch.cuda.is_available():
             torch.cuda.set_device(rank)
         torch.manual_seed(1234)
+        _capture_baseline_torch_flags()
         result_queue.put("ready")
 
         while True:
@@ -226,7 +305,7 @@ def run(self, fn, *args, **kwargs):
                 q.put((fn, args, kwargs))
             for _ in range(self.world_size):
                 timeout = self.job_timeout if deadline is None else deadline - time.monotonic()
-                status, rank, tb = self._result_queue.get(timeout=max(timeout, 0))
+                status, rank, tb = self._get(max(timeout, 0))
                 if status == "error":
                     errors.append(f"--- Rank {rank} ---\n{tb}")
                     deadline = deadline or time.monotonic() + self.failure_grace
diff --git a/tests/_test_utils/torch/puzzletron/utils.py b/tests/_test_utils/torch/puzzletron/utils.py
index 2ce44858ef5..525b597518d 100644
--- a/tests/_test_utils/torch/puzzletron/utils.py
+++ b/tests/_test_utils/torch/puzzletron/utils.py
@@ -270,9 +270,9 @@ def save_dummy_dataset(dataset_path: Path | str):
         },
     ]
 
-    # Prepare train and val splits with sample repeated, 2500 samples are for
-    # 128 samples with block-size 8192 and LLama3 tokenizer
-    data = [{"conversation": sample}] * 2500
+    # Prepare train and val splits with the sample repeated. With the tiny tokenizer a sample is
+    # 1424 tokens, so 400 samples give 278 blocks of 2048 tokens; the pruning scores read at most 100.
+    data = [{"conversation": sample}] * 400
 
     # For train-val splits
     data_dict = DatasetDict({"train": Dataset.from_list(data), "valid": Dataset.from_list(data)})
diff --git a/tests/_test_utils/torch/quantization/onnx_export.py b/tests/_test_utils/torch/quantization/onnx_export.py
index e65e6f23500..b3041f780d1 100644
--- a/tests/_test_utils/torch/quantization/onnx_export.py
+++ b/tests/_test_utils/torch/quantization/onnx_export.py
@@ -23,7 +23,7 @@
 
 from .models import SimpleConv, SimpleConvLinear, SimpleLinear
 
-TEST_MODELS = {SimpleLinear, SimpleConv, SimpleConvLinear}
+TEST_MODELS = (SimpleLinear, SimpleConv, SimpleConvLinear)
 
 
 def onnx_export_tester(model, device, num_bits, per_channel_quantization, constant_folding, dtype):
diff --git a/tests/conftest.py b/tests/conftest.py
index af77546a244..b1685e3fe63 100644
--- a/tests/conftest.py
+++ b/tests/conftest.py
@@ -58,7 +58,10 @@ def pytest_addoption(parser):
     "gpu_trtllm": 60,
     "gpu_vllm": 60,
     "regression": 180,
-    "unit": 120 if platform.system() == "Windows" else 60,
+    # Tests that spawn processes take 2-3x longer under pytest-xdist: 4 workers and their children
+    # share the 4 vCPUs of a CI runner (test_dist.py::test_data_parallel: 8-15 s serial, 25-35 s with
+    # xdist, over 60 s on a slow runner).
+    "unit": 120,
 }
 
 
diff --git a/tests/examples/llm_eval/test_llm_eval.py b/tests/examples/llm_eval/test_llm_eval.py
index d65f497e320..c1ba2c29e95 100644
--- a/tests/examples/llm_eval/test_llm_eval.py
+++ b/tests/examples/llm_eval/test_llm_eval.py
@@ -32,9 +32,10 @@ def test_lm_eval_hf(tmp_path):
         ["python", "lm_eval_hf.py"],
         model="hf",
         model_args=f"pretrained={model_dir}",
-        tasks="mmlu",
+        # One subtask takes the same harness and gate path as the 57-subtask mmlu group, which
+        # spends minutes loading datasets on a model whose context is only 32 tokens
+        tasks="mmlu_abstract_algebra",
         num_fewshot=5,
-        limit=0.1,
         batch_size=8,
         # Exercise the accuracy gate: reads the results file and enforces the bound
         output_path=str(tmp_path / "results"),
@@ -62,9 +63,9 @@ def test_qwen3_eval_fp8(tmp_path):
             simple_eval_tasks="humaneval",
             lm_eval_limit=16,
             simple_eval_limit=16,
-            # MMLU has no inherent sample cap and otherwise evaluates the full ~14k-question
-            # test set; limit it like lm_eval/simple_eval to keep this a fast smoke test.
-            mmlu_limit=16,
+            # mmlu.py applies the limit per subject (57 of them), one TRT-LLM call per prompt:
+            # 1 still exercises every subject, 16 would be 912 sequential calls.
+            mmlu_limit=1,
             output=128,  # Cap generation length: gsm8k/humaneval otherwise generate up to 1024 tokens/sample
             batch=8,
         )
diff --git a/tests/examples/llm_qat/test_llm_qat.py b/tests/examples/llm_qat/test_llm_qat.py
index a7b610a807c..bec89bd1b97 100644
--- a/tests/examples/llm_qat/test_llm_qat.py
+++ b/tests/examples/llm_qat/test_llm_qat.py
@@ -19,6 +19,7 @@
 import pytest
 import torch
 from _test_utils.examples.run_command import run_example_command
+from _test_utils.fs_utils import assert_unmodified_tree
 from safetensors.torch import load_file
 
 # Mapping from backend name to accelerate config file
@@ -102,6 +103,38 @@ def _run_export(ckpt_dir: str, export_dir: str):
     )
 
 
+@pytest.fixture(scope="module")
+def dataset_cache_dir(tmp_path_factory):
+    """Tokenized-dataset cache for every quantize/train run below.
+
+    The cache key covers the tokenizer, max length and blend (not the command), so the dataset is
+    built once for quantize.py (8192) and once for train.py (128) instead of once per command.
+    """
+    return str(tmp_path_factory.mktemp("dataset_cache"))
+
+
+@pytest.fixture(scope="module")
+def nvfp4_ptq_dir(tiny_qwen3_path, tmp_path_factory, dataset_cache_dir):
+    """NVFP4 PTQ checkpoint shared by the QAT and QAD tests, which only read it.
+
+    quantize.py ignores the QAD-only config keys, so qat_nvfp4.yaml and qad_nvfp4.yaml give the
+    same arguments and one run serves both.
+    """
+    ptq_dir = tmp_path_factory.mktemp("ptq")
+    _run_quantize(
+        "configs/train/qat_nvfp4.yaml",
+        [
+            "--model_name_or_path", tiny_qwen3_path,
+            "--recipe", "general/ptq/nvfp4_default-kv_fp8",
+            "--calib_size", "64",
+            "--output_dir", str(ptq_dir),
+        ],
+        cache_dir=dataset_cache_dir,
+    )
+    with assert_unmodified_tree(ptq_dir) as path:
+        yield str(path)
+
+
 def test_dataset_utils_pretokenize(tiny_qwen3_path, tmp_path):
     """Test dataset_utils.py standalone CLI pre-tokenization."""
     cache_dir = tmp_path / "dataset_cache"
@@ -122,33 +155,18 @@ def test_dataset_utils_pretokenize(tiny_qwen3_path, tmp_path):
     "deepspeed",
     "ddp",
 ])
-def test_qwen3_qat_nvfp4(tiny_qwen3_path, tmp_path, backend):
-    ptq_output_dir = tmp_path / "ptq"
+def test_qwen3_qat_nvfp4(nvfp4_ptq_dir, dataset_cache_dir, tmp_path, backend):
     qat_output_dir = tmp_path / "qat"
-    cache_dir = str(tmp_path / "dataset_cache")
-
-    # Step 1: Quantize
-    _run_quantize(
-        "configs/train/qat_nvfp4.yaml",
-        [
-            "--model_name_or_path", tiny_qwen3_path,
-            "--recipe", "general/ptq/nvfp4_default-kv_fp8",
-            "--calib_size", "64",
-            "--output_dir", str(ptq_output_dir),
-        ],
-        cache_dir=cache_dir,
-    )
 
-    # Step 2: QAT
     _run_train(
         "configs/train/qat_nvfp4.yaml",
         [
-            "--model_name_or_path", str(ptq_output_dir),
+            "--model_name_or_path", nvfp4_ptq_dir,
             "--do_train", "True",
             "--output_dir", str(qat_output_dir),
         ],
         backend=backend,
-        cache_dir=cache_dir,
+        cache_dir=dataset_cache_dir,
     )
 
 @pytest.mark.skip(reason="FSDP2 LoRA checkpoint save omits adapter_model.safetensors")
@@ -211,43 +229,28 @@ def test_qwen3_lora_qat_nvfp4(tiny_qwen3_path, tmp_path):
     "fsdp2",
     "deepspeed",
 ])
-def test_qwen3_qad_nvfp4(tiny_qwen3_path, tmp_path, backend):
-    ptq_output_dir = tmp_path / "ptq"
+def test_qwen3_qad_nvfp4(tiny_qwen3_path, nvfp4_ptq_dir, dataset_cache_dir, tmp_path, backend):
     qad_output_dir = tmp_path / "qad"
-    cache_dir = str(tmp_path / "dataset_cache")
-
-    # Step 1: Quantize student
-    _run_quantize(
-        "configs/train/qad_nvfp4.yaml",
-        [
-            "--model_name_or_path", tiny_qwen3_path,
-            "--recipe", "general/ptq/nvfp4_default-kv_fp8",
-            "--calib_size", "64",
-            "--output_dir", str(ptq_output_dir),
-        ],
-        cache_dir=cache_dir,
-    )
 
-    # Step 2: QAD (quantization-aware distillation)
+    # QAD (quantization-aware distillation) of the shared quantized student
     _run_train(
         "configs/train/qad_nvfp4.yaml",
         [
-            "--model_name_or_path", str(ptq_output_dir),
+            "--model_name_or_path", nvfp4_ptq_dir,
             "--do_train", "True",
             "--output_dir", str(qad_output_dir),
             "--distill", "True",
             "--teacher_model", tiny_qwen3_path,
         ],
         backend=backend,
-        cache_dir=cache_dir,
+        cache_dir=dataset_cache_dir,
     )
 
 
-def test_qwen3_qlora_nvfp4(tiny_qwen3_path, tmp_path):
+def test_qwen3_qlora_nvfp4(tiny_qwen3_path, dataset_cache_dir, tmp_path):
     ptq_output_dir = tmp_path / "ptq"
-    cache_dir = str(tmp_path / "dataset_cache")
 
-    # Step 1: Quantize with compression for QLoRA
+    # Step 1: Quantize with compression for QLoRA (its own checkpoint: the others are not compressed)
     _run_quantize(
         "configs/train/qlora_nvfp4.yaml",
         [
@@ -257,7 +260,7 @@ def test_qwen3_qlora_nvfp4(tiny_qwen3_path, tmp_path):
             "--compress", "True",
             "--output_dir", str(ptq_output_dir),
         ],
-        cache_dir=cache_dir,
+        cache_dir=dataset_cache_dir,
     )
 
     # Step 2: QLoRA training
@@ -271,7 +274,7 @@ def test_qwen3_qlora_nvfp4(tiny_qwen3_path, tmp_path):
             "--output_dir", str(qlora_output_dir),
         ],
         backend="ddp",
-        cache_dir=cache_dir,
+        cache_dir=dataset_cache_dir,
     )
 
     # Step 3: Export the QLoRA checkpoint for deployment
diff --git a/tests/examples/megatron_bridge/conftest.py b/tests/examples/megatron_bridge/conftest.py
index 1093fb7b281..17043f6aebc 100644
--- a/tests/examples/megatron_bridge/conftest.py
+++ b/tests/examples/megatron_bridge/conftest.py
@@ -40,18 +40,22 @@ def _fast_example_runner():
 
 
 @pytest.fixture(autouse=True)
-def _isolate_megatron_global_state():
+def _isolate_megatron_global_state(request):
     """Reset shared state around every test so a failure cannot cascade into the next one.
 
     In-process steps share the interpreter. Besides Megatron's singletons, Transformer-Engine
     records its chosen attention backend in ``NVTE_*``, which failed a Mamba hybrid that ran after
     an attention model -- so the environment is restored wholesale rather than by naming variables.
     """
+    # Only tests that take ``num_gpus`` run example steps; the gc-heavy reset is wasted on the rest.
+    reset_state = "num_gpus" in request.fixturenames
     env_before = os.environ.copy()
-    reset_megatron_global_state()
+    if reset_state:
+        reset_megatron_global_state()
     try:
         yield
     finally:
-        reset_megatron_global_state()
+        if reset_state:
+            reset_megatron_global_state()
         os.environ.clear()
         os.environ.update(env_before)
diff --git a/tests/examples/speculative_decoding/test_eagle.py b/tests/examples/speculative_decoding/test_eagle.py
index 8d284d4ba1b..7325fe72e84 100644
--- a/tests/examples/speculative_decoding/test_eagle.py
+++ b/tests/examples/speculative_decoding/test_eagle.py
@@ -147,9 +147,17 @@ def test_llama_eagle3(tiny_llama_path,
         *_TINY_EAGLE_ARCH,
     ]
 
+    env = None
+    if cp_size == 1 and not mix_hidden_states:
+        # The 1-GPU PR smoke compiles with mode="max-autotune"; benchmarking Triton GEMM templates
+        # is a large share of that on a tiny model and the test only needs the compiled path to
+        # run. [2-False] (2 GPUs, nightly) keeps the full autotune.
+        env = {**os.environ, "TORCHINDUCTOR_MAX_AUTOTUNE_GEMM_BACKENDS": "ATEN"}
+
     run_example_command(
         ["./launch_train.sh", "--config", EAGLE3_YAML, *overrides],
         "speculative_decoding",
+        env=env,
     )
 
 
@@ -222,7 +230,6 @@ def test_convert_to_vllm_ckpt(tiny_llama_path, eagle_output_dir):
 @pytest.mark.parametrize(
     ("model_source", "use_fake_base"),
     [
-        (None, False),  # tiny_llama (from fixture), no FakeBase
         pytest.param("moonshotai/Kimi-K2.5", True),  # remote HF repo, FakeBaseModel
         pytest.param(  # remote HF repo, no FakeBaseModel
             "moonshotai/Kimi-K2-Thinking",
@@ -235,22 +242,23 @@ def test_convert_to_vllm_ckpt(tiny_llama_path, eagle_output_dir):
             marks=pytest.mark.manual(reason="skip redundand test, too slow"),
         ),
     ],
-    ids=["tinyllama", "kimi-k2.5", "kimi-k2-thinking", "minimax-m2.5"],
+    ids=["kimi-k2.5", "kimi-k2-thinking", "minimax-m2.5"],
 )
 def test_offline_eagle3_training(
-    tiny_llama_path,
     tiny_daring_anteater_path,
     tmp_path,
     eagle_output_dir,
     model_source,
     use_fake_base,
 ):
-    """Test Eagle3 training with pre-computed hidden states (offline mode / FakeBaseModel)."""
-    model_path = tiny_llama_path if model_source is None else model_source
-    model_id = "tinyllama" if model_source is None else model_source.split("/")[-1]
+    """Test Eagle3 training with pre-computed hidden states on a FakeBaseModel.
+
+    Offline training on a real base model is covered by test_eagle_offline_ptq.py.
+    """
+    model_id = model_source.split("/")[-1]
     output_subdir = eagle_output_dir / f"eagle-{model_id}-offline"
 
-    cfg = AutoConfig.from_pretrained(model_path, trust_remote_code=True)
+    cfg = AutoConfig.from_pretrained(model_source, trust_remote_code=True)
     if hasattr(cfg, "text_config"):  # vlm: get text_config
         cfg = cfg.text_config
 
@@ -262,7 +270,7 @@ def test_offline_eagle3_training(
     )
 
     overrides = [
-        f"model.model_name_or_path={model_path}",
+        f"model.model_name_or_path={model_source}",
         "model.trust_remote_code=true",
         f"model.use_fake_base_for_offline={str(use_fake_base).lower()}",
         f"data.data_path={tiny_daring_anteater_path}",
diff --git a/tests/examples/speculative_decoding/test_eagle_offline_ptq.py b/tests/examples/speculative_decoding/test_eagle_offline_ptq.py
index 5fca5a3d71c..3bd1be71dba 100644
--- a/tests/examples/speculative_decoding/test_eagle_offline_ptq.py
+++ b/tests/examples/speculative_decoding/test_eagle_offline_ptq.py
@@ -97,7 +97,8 @@ def test_offline_eagle_training(tiny_llama_path, tiny_daring_anteater_path, offl
         f"training.output_dir={output_dir}",
         "training.num_train_epochs=1",
         "training.learning_rate=1e-5",
-        "training.training_seq_len=64",
+        # Shorter than the dumped sequences, so the offline collator truncates them.
+        "training.training_seq_len=16",
         "training.save_steps=1",
         # torch.compile is smoke-tested once by test_llama_eagle3[1-False]; skip its warmup here.
         "eagle.eagle_use_torch_compile=false",
diff --git a/tests/examples/torch_onnx/conftest.py b/tests/examples/torch_onnx/conftest.py
new file mode 100644
index 00000000000..f226fbee9f9
--- /dev/null
+++ b/tests/examples/torch_onnx/conftest.py
@@ -0,0 +1,44 @@
+# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+# SPDX-License-Identifier: Apache-2.0
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+"""Run the torch_onnx steps in one long-lived worker; see ``_test_utils.examples.example_runner``."""
+
+import pytest
+from _test_utils.examples.example_runner import ExampleRunner, keep_post_conversion_plugins
+from _test_utils.examples.run_command import set_in_process_runner
+
+# hf_embedding_quant_to_onnx.py has not been checked for repeated runs in one process, so it keeps
+# running as a subprocess.
+_SCRIPTS = {"torch_quant_to_onnx.py"}
+
+
+@pytest.fixture(scope="session")
+def _example_runner():
+    runner = ExampleRunner(_SCRIPTS, hooks=(keep_post_conversion_plugins,))
+    yield runner
+    runner.close()
+
+
+@pytest.fixture(autouse=True)
+def _run_steps_in_the_worker(_example_runner):
+    """Install the runner per test, not per session.
+
+    The hook is a module global in ``run_command``, so leaving it installed would also reach any
+    other example suite collected later in the same session (``pytest tests/examples``).
+    """
+    set_in_process_runner(_example_runner)
+    try:
+        yield
+    finally:
+        set_in_process_runner(None)
diff --git a/tests/examples/torch_onnx/test_torch_quant_to_onnx.py b/tests/examples/torch_onnx/test_torch_quant_to_onnx.py
index 28de24e9429..9fa4aa958d1 100644
--- a/tests/examples/torch_onnx/test_torch_quant_to_onnx.py
+++ b/tests/examples/torch_onnx/test_torch_quant_to_onnx.py
@@ -15,6 +15,7 @@
 
 
 import importlib
+import os
 from collections import defaultdict
 from pathlib import Path
 
@@ -38,6 +39,10 @@
     "resnet50": ("resnet50", None),
 }
 
+# PRs run swinv2 for fp8 and nvfp4 only (swin_tiny covers every format); scheduled runs keep all.
+_PR_CI = os.environ.get("GITHUB_REF", "").startswith("refs/heads/pull-request/")
+_SKIPPED_ON_PR = {("swinv2_tiny", "int8"), ("swinv2_tiny", "mxfp8"), ("swinv2_tiny", "auto")}
+
 
 def _assert_residual_inputs_are_quantized(onnx_save_path):
     model = onnx.load(onnx_save_path)
@@ -74,6 +79,8 @@ def _assert_residual_inputs_are_quantized(onnx_save_path):
 def test_torch_onnx(tmp_path, model_key, qformat):
     if model_key == "resnet50" and qformat not in _RESNET_RECIPE_QFORMATS:
         pytest.skip("Only FP8 and INT8 quantization are supported for ResNet")
+    if _PR_CI and (model_key, qformat) in _SKIPPED_ON_PR:
+        pytest.skip("Covered by swin_tiny on PRs; runs in the scheduled CI")
 
     timm_model_name, model_kwargs = _MODELS[model_key]
     onnx_save_path = tmp_path / f"{model_key}.{qformat}.onnx"
diff --git a/tests/gpu/onnx/quantization/autotune/test_benchmark.py b/tests/gpu/onnx/quantization/autotune/test_benchmark.py
index 012bf495ab6..47311e25219 100644
--- a/tests/gpu/onnx/quantization/autotune/test_benchmark.py
+++ b/tests/gpu/onnx/quantization/autotune/test_benchmark.py
@@ -46,8 +46,8 @@
 
 @pytest.fixture
 def simple_conv_model_bytes():
-    """ONNX model bytes: Input -> Conv -> Relu -> Output (from _test_utils)."""
-    model = _create_simple_conv_onnx_model()
+    """ONNX model bytes: Input -> Conv -> Relu -> Output (from _test_utils), small input."""
+    model = _create_simple_conv_onnx_model(batch_size=1, spatial_size=32)
     return model.SerializeToString()
 
 
@@ -431,7 +431,12 @@ def test_run_with_path(self, simple_conv_model_path):
         with tempfile.NamedTemporaryFile(suffix=".cache", delete=False) as f:
             cache_path = f.name
         try:
-            benchmark = TrtExecBenchmark(timing_cache_file=cache_path, warmup_runs=1, timing_runs=2)
+            benchmark = TrtExecBenchmark(
+                timing_cache_file=cache_path,
+                warmup_runs=1,
+                timing_runs=2,
+                trtexec_args=["--duration=0"],
+            )
             latency_ms = benchmark.run(simple_conv_model_path)
             assert isinstance(latency_ms, float)
             assert latency_ms > 0
@@ -445,7 +450,12 @@ def test_run_with_bytes(self, simple_conv_model_bytes):
         with tempfile.NamedTemporaryFile(suffix=".cache", delete=False) as f:
             cache_path = f.name
         try:
-            benchmark = TrtExecBenchmark(timing_cache_file=cache_path, warmup_runs=1, timing_runs=2)
+            benchmark = TrtExecBenchmark(
+                timing_cache_file=cache_path,
+                warmup_runs=1,
+                timing_runs=2,
+                trtexec_args=["--duration=0"],
+            )
             latency_ms = benchmark.run(simple_conv_model_bytes)
             assert isinstance(latency_ms, float)
             assert latency_ms > 0
diff --git a/tests/gpu/onnx/quantization/autotune/test_workflow.py b/tests/gpu/onnx/quantization/autotune/test_workflow.py
index eebc42f6c33..b323402f1eb 100644
--- a/tests/gpu/onnx/quantization/autotune/test_workflow.py
+++ b/tests/gpu/onnx/quantization/autotune/test_workflow.py
@@ -59,7 +59,10 @@ def test_export_quantized_model(use_trtexec, simple_conv_model):
     output_path = output_dir + ".quant.onnx"
 
     try:
-        init_benchmark_instance(use_trtexec=use_trtexec, timing_runs=100)
+        # trtexec measures for at least 3 s by default; --duration=0 stops after timing_runs iterations
+        init_benchmark_instance(
+            use_trtexec=use_trtexec, timing_runs=100, trtexec_args=["--duration=0"]
+        )
         autotuner = region_pattern_autotuning_workflow(baseline_model_path, Path(output_dir))
 
         # Export model with Q/DQ insertion
diff --git a/tests/gpu/onnx/quantization/test_concat_elim.py b/tests/gpu/onnx/quantization/test_concat_elim.py
index 68179051909..de9feafe7e7 100644
--- a/tests/gpu/onnx/quantization/test_concat_elim.py
+++ b/tests/gpu/onnx/quantization/test_concat_elim.py
@@ -23,10 +23,19 @@
 
 from modelopt.onnx.quantization.quantize import quantize
 
+# Keep every Conv above the 16-channel limit for INT8/FP8 quantization (graph_selection.py) so the
+# same nodes are quantized as with the default 128-channel model, but calibrate a small input.
+SMALL_MODEL_SHAPE = {"channels": 32, "height": 16, "width": 16}
+
 
 def _check_concat_qdq_status(onnx_path, quantize_mode):
-    # Quantize the input model
-    quantize(onnx_path, quantize_mode=quantize_mode, passes="concat_elimination")
+    # Quantize the input model. Only the graph rewrite is checked, so calibrate on CPU.
+    quantize(
+        onnx_path,
+        quantize_mode=quantize_mode,
+        passes="concat_elimination",
+        calibration_eps=["cpu"],
+    )
 
     # Output model should be produced in the same tmpdir
     output_onnx_path = onnx_path.replace(".onnx", ".quant.onnx")
@@ -61,7 +70,7 @@ def _check_concat_qdq_status(onnx_path, quantize_mode):
 
 def test_concat_elim_int8(tmp_path):
     # Copy the test model to the tmp_path
-    model = build_conv_concat_model()
+    model = build_conv_concat_model(**SMALL_MODEL_SHAPE)
     this_function_name = sys._getframe().f_code.co_name
     onnx_path = os.path.join(tmp_path, f"{this_function_name}.onnx")
     onnx.save(model, onnx_path)
@@ -70,7 +79,7 @@ def test_concat_elim_int8(tmp_path):
 
 def test_concat_elim_fp8(tmp_path):
     # Copy the test model to the tmp_path
-    model = build_conv_concat_model()
+    model = build_conv_concat_model(**SMALL_MODEL_SHAPE)
     this_function_name = sys._getframe().f_code.co_name
     onnx_path = os.path.join(tmp_path, f"{this_function_name}.onnx")
     onnx.save(model, onnx_path)
diff --git a/tests/gpu/torch/export/test_unified_hf_export_and_check_safetensors.py b/tests/gpu/torch/export/test_unified_hf_export_and_check_safetensors.py
index b45943e6e5e..7a67c026b3b 100644
--- a/tests/gpu/torch/export/test_unified_hf_export_and_check_safetensors.py
+++ b/tests/gpu/torch/export/test_unified_hf_export_and_check_safetensors.py
@@ -17,6 +17,7 @@
 import pytest
 import torch
 from _test_utils.examples.run_command import extend_cmd_parts, run_example_command
+from _test_utils.torch.calib_data import write_calib_jsonl
 from _test_utils.torch.transformers_models import (
     create_tiny_gpt_oss_dir,
     create_tiny_llama_dir,
@@ -26,6 +27,12 @@
 from safetensors import safe_open
 
 
+@pytest.fixture(scope="module")
+def calib_jsonl(tmp_path_factory):
+    """Local calibration texts; the test checks the exported structure, so it needs no Hub dataset."""
+    return write_calib_jsonl(tmp_path_factory.mktemp("calib") / "calib.jsonl")
+
+
 # Here we map each qformat -> the suffix we expect in the generated safetensors directory
 @pytest.mark.parametrize(
     (
@@ -55,6 +62,7 @@
 )
 def test_unified_hf_export_and_check_safetensors(
     tmp_path,
+    calib_jsonl,
     qformat,
     expected_suffix,
     fuse_input_scale,
@@ -103,11 +111,15 @@ def test_unified_hf_export_and_check_safetensors(
         pyt_ckpt_path=tiny_model_dir,
         qformat=qformat,
         export_path=output_dir,
-        dataset="cnn_dailymail",
+        dataset=calib_jsonl,
         # This test only checks the exported safetensors structure (not accuracy), so a
         # small calibration set is enough. Avoids the 1024-sample default on a toy model.
         calib_size=64,
+        # One batch holds the whole set, so hf_ptq does not probe for the largest batch.
+        batch_size=64,
     )
+    # The generate() previews before and after quantization do not affect the export.
+    cmd_parts.append("--skip_generate")
 
     # Run the command
     # current transformers T5 model seem to be incompatible with multiple GPUs
diff --git a/tests/gpu/torch/puzzletron/resources/configs/Qwen/Qwen2.5-7B-Instruct/Qwen2.5-7B-Instruct.yaml b/tests/gpu/torch/puzzletron/resources/configs/Qwen/Qwen2.5-7B-Instruct/Qwen2.5-7B-Instruct.yaml
index 2843f0b97ac..6691a65d9d7 100644
--- a/tests/gpu/torch/puzzletron/resources/configs/Qwen/Qwen2.5-7B-Instruct/Qwen2.5-7B-Instruct.yaml
+++ b/tests/gpu/torch/puzzletron/resources/configs/Qwen/Qwen2.5-7B-Instruct/Qwen2.5-7B-Instruct.yaml
@@ -19,7 +19,7 @@ build_replacement_library:
   add_attention_no_ops: true
 
 calc_subblock_stats:
-  batch_sizes: [64, 96, 128]
+  batch_sizes: [96]  # must include the batch_size in mip.subblock_stats_args
   prefill_seq_len: 4096
   generation_seq_len: 4096
   num_active_tokens_override:       # Optional override for sequence lengths
diff --git a/tests/gpu/torch/puzzletron/resources/configs/Qwen/Qwen3-8B/Qwen3-8B.yaml b/tests/gpu/torch/puzzletron/resources/configs/Qwen/Qwen3-8B/Qwen3-8B.yaml
index cd82a47271a..59e69520a4b 100644
--- a/tests/gpu/torch/puzzletron/resources/configs/Qwen/Qwen3-8B/Qwen3-8B.yaml
+++ b/tests/gpu/torch/puzzletron/resources/configs/Qwen/Qwen3-8B/Qwen3-8B.yaml
@@ -19,7 +19,7 @@ build_replacement_library:
   add_attention_no_ops: true
 
 calc_subblock_stats:
-  batch_sizes: [64, 96, 128]
+  batch_sizes: [96]  # must include the batch_size in mip.subblock_stats_args
   prefill_seq_len: 4096
   generation_seq_len: 4096
   num_active_tokens_override:       # Optional override for sequence lengths
diff --git a/tests/gpu/torch/puzzletron/resources/configs/Qwen/Qwen3-VL-30B-A3B-Instruct/Qwen3-VL-30B-A3B-Instruct.yaml b/tests/gpu/torch/puzzletron/resources/configs/Qwen/Qwen3-VL-30B-A3B-Instruct/Qwen3-VL-30B-A3B-Instruct.yaml
index 00b21ea9798..77567a777f9 100644
--- a/tests/gpu/torch/puzzletron/resources/configs/Qwen/Qwen3-VL-30B-A3B-Instruct/Qwen3-VL-30B-A3B-Instruct.yaml
+++ b/tests/gpu/torch/puzzletron/resources/configs/Qwen/Qwen3-VL-30B-A3B-Instruct/Qwen3-VL-30B-A3B-Instruct.yaml
@@ -19,7 +19,7 @@ build_replacement_library:
   add_attention_no_ops: true
 
 calc_subblock_stats:
-  batch_sizes: [64, 96, 128]
+  batch_sizes: [96]  # must include the batch_size in mip.subblock_stats_args
   prefill_seq_len: 4096
   generation_seq_len: 4096
   num_active_tokens_override:       # Optional override for sequence lengths
diff --git a/tests/gpu/torch/puzzletron/resources/configs/meta-llama/Llama-3.1-8B-Instruct/Llama-3.1-8B-Instruct.yaml b/tests/gpu/torch/puzzletron/resources/configs/meta-llama/Llama-3.1-8B-Instruct/Llama-3.1-8B-Instruct.yaml
index 8e2e0786b33..880f630d83b 100644
--- a/tests/gpu/torch/puzzletron/resources/configs/meta-llama/Llama-3.1-8B-Instruct/Llama-3.1-8B-Instruct.yaml
+++ b/tests/gpu/torch/puzzletron/resources/configs/meta-llama/Llama-3.1-8B-Instruct/Llama-3.1-8B-Instruct.yaml
@@ -19,7 +19,7 @@ build_replacement_library:
   add_attention_no_ops: true
 
 calc_subblock_stats:
-  batch_sizes: [64, 96, 128]
+  batch_sizes: [96]  # must include the batch_size in mip.subblock_stats_args
   prefill_seq_len: 4096
   generation_seq_len: 4096
   num_active_tokens_override: # Optional override for sequence lengths
diff --git a/tests/gpu/torch/puzzletron/resources/configs/meta-llama/Llama-3.2-3B-Instruct/Llama-3.2-3B-Instruct.yaml b/tests/gpu/torch/puzzletron/resources/configs/meta-llama/Llama-3.2-3B-Instruct/Llama-3.2-3B-Instruct.yaml
index 78cb6bd73c4..ab45abe41f9 100644
--- a/tests/gpu/torch/puzzletron/resources/configs/meta-llama/Llama-3.2-3B-Instruct/Llama-3.2-3B-Instruct.yaml
+++ b/tests/gpu/torch/puzzletron/resources/configs/meta-llama/Llama-3.2-3B-Instruct/Llama-3.2-3B-Instruct.yaml
@@ -19,7 +19,7 @@ build_replacement_library:
   add_attention_no_ops: true
 
 calc_subblock_stats:
-  batch_sizes: [64, 96, 128]
+  batch_sizes: [96]  # must include the batch_size in mip.subblock_stats_args
   prefill_seq_len: 4096
   generation_seq_len: 4096
   num_active_tokens_override: # Optional override for sequence lengths
diff --git a/tests/gpu/torch/puzzletron/resources/configs/mistralai/Mistral-Small-24B-Instruct-2501/Mistral-Small-24B-Instruct-2501.yaml b/tests/gpu/torch/puzzletron/resources/configs/mistralai/Mistral-Small-24B-Instruct-2501/Mistral-Small-24B-Instruct-2501.yaml
index e042c4bb627..a6593200354 100644
--- a/tests/gpu/torch/puzzletron/resources/configs/mistralai/Mistral-Small-24B-Instruct-2501/Mistral-Small-24B-Instruct-2501.yaml
+++ b/tests/gpu/torch/puzzletron/resources/configs/mistralai/Mistral-Small-24B-Instruct-2501/Mistral-Small-24B-Instruct-2501.yaml
@@ -19,7 +19,7 @@ build_replacement_library:
   add_attention_no_ops: true
 
 calc_subblock_stats:
-  batch_sizes: [64, 96, 128]
+  batch_sizes: [96]  # must include the batch_size in mip.subblock_stats_args
   prefill_seq_len: 4096
   generation_seq_len: 4096
   num_active_tokens_override:       # Optional override for sequence lengths
diff --git a/tests/gpu/torch/puzzletron/resources/configs/nvidia/NVIDIA-Nemotron-3-Nano-30B-A3B-Base-BF16/NVIDIA-Nemotron-3-Nano-30B-A3B-Base-BF16.yaml b/tests/gpu/torch/puzzletron/resources/configs/nvidia/NVIDIA-Nemotron-3-Nano-30B-A3B-Base-BF16/NVIDIA-Nemotron-3-Nano-30B-A3B-Base-BF16.yaml
index ab2b09e6794..33a8318215d 100644
--- a/tests/gpu/torch/puzzletron/resources/configs/nvidia/NVIDIA-Nemotron-3-Nano-30B-A3B-Base-BF16/NVIDIA-Nemotron-3-Nano-30B-A3B-Base-BF16.yaml
+++ b/tests/gpu/torch/puzzletron/resources/configs/nvidia/NVIDIA-Nemotron-3-Nano-30B-A3B-Base-BF16/NVIDIA-Nemotron-3-Nano-30B-A3B-Base-BF16.yaml
@@ -20,7 +20,7 @@ build_replacement_library:
   add_attention_no_ops: true
 
 calc_subblock_stats:
-  batch_sizes: [64, 96, 128]
+  batch_sizes: [96]  # must include the batch_size in mip.subblock_stats_args
   prefill_seq_len: 4096
   generation_seq_len: 4096
   num_active_tokens_override:       # Optional override for sequence lengths
diff --git a/tests/gpu/torch/puzzletron/resources/configs/nvidia/NVIDIA-Nemotron-Nano-12B-v2/NVIDIA-Nemotron-Nano-12B-v2.yaml b/tests/gpu/torch/puzzletron/resources/configs/nvidia/NVIDIA-Nemotron-Nano-12B-v2/NVIDIA-Nemotron-Nano-12B-v2.yaml
index 906b7338d8c..05b22051392 100644
--- a/tests/gpu/torch/puzzletron/resources/configs/nvidia/NVIDIA-Nemotron-Nano-12B-v2/NVIDIA-Nemotron-Nano-12B-v2.yaml
+++ b/tests/gpu/torch/puzzletron/resources/configs/nvidia/NVIDIA-Nemotron-Nano-12B-v2/NVIDIA-Nemotron-Nano-12B-v2.yaml
@@ -19,7 +19,7 @@ build_replacement_library:
   add_attention_no_ops: true
 
 calc_subblock_stats:
-  batch_sizes: [64, 96, 128]
+  batch_sizes: [96]  # must include the batch_size in mip.subblock_stats_args
   prefill_seq_len: 4096
   generation_seq_len: 4096
   num_active_tokens_override:       # Optional override for sequence lengths
diff --git a/tests/gpu/torch/puzzletron/resources/configs/openai/gpt-oss-20b/gpt-oss-20b.yaml b/tests/gpu/torch/puzzletron/resources/configs/openai/gpt-oss-20b/gpt-oss-20b.yaml
index 60d6568df60..d57ff295796 100644
--- a/tests/gpu/torch/puzzletron/resources/configs/openai/gpt-oss-20b/gpt-oss-20b.yaml
+++ b/tests/gpu/torch/puzzletron/resources/configs/openai/gpt-oss-20b/gpt-oss-20b.yaml
@@ -20,7 +20,7 @@ build_replacement_library:
   add_attention_no_ops: true
 
 calc_subblock_stats:
-  batch_sizes: [64, 96, 128]
+  batch_sizes: [96]  # must include the batch_size in mip.subblock_stats_args
   prefill_seq_len: 4096
   generation_seq_len: 4096
   num_active_tokens_override: # Optional override for sequence lengths
diff --git a/tests/gpu/torch/puzzletron/test_nas_convert.py b/tests/gpu/torch/puzzletron/test_nas_convert.py
index 57239f1c0e4..7ddb0584fc7 100644
--- a/tests/gpu/torch/puzzletron/test_nas_convert.py
+++ b/tests/gpu/torch/puzzletron/test_nas_convert.py
@@ -27,62 +27,6 @@
 import modelopt.torch.utils.distributed as dist
 
 
-def test_nas_convert_ffn_pruning(project_root_path: Path, tmp_path: Path):
-    spawn_multiprocess_job(
-        size=torch.cuda.device_count(),
-        job=partial(_test_nas_convert_ffn_pruning_multiprocess_job, project_root_path, tmp_path),
-        backend="nccl",
-    )
-
-
-def _test_nas_convert_ffn_pruning_multiprocess_job(
-    project_root_path: Path, tmp_path: Path, rank: int, size: int
-):
-    dist.setup(timeout=timedelta(minutes=10))
-    # Setup the test model and data.
-    puzzle_dir, llama_checkpoint_path, dataset_path = setup_test_model_and_data(
-        tmp_path, rank, "meta-llama/Llama-3.1-8B-Instruct"
-    )
-    hydra_config_dir = project_root_path / "tests/gpu/torch/puzzletron/resources/configs"
-    hydra_config_name = "meta-llama/Llama-3.1-8B-Instruct/Llama-3.1-8B-Instruct"
-
-    #
-    # Run the mnt.convert() step
-    #
-    input_model = mtpz.puzzletron_nas_plugin.PuzzletronModel()
-    mtn.convert(
-        input_model,
-        mode=[
-            (
-                "puzzletron",
-                {
-                    "puzzle_dir": str(puzzle_dir),
-                    "input_model_path": str(llama_checkpoint_path),
-                    "hydra_config_dir": str(hydra_config_dir),
-                    "hydra_config_name": hydra_config_name,
-                    "dataset_path": str(dataset_path),
-                },
-            )
-        ],
-    )
-
-    #
-    # Check assertions
-    #
-    if rank == 0:
-        # assertions for the score_pruning_activations step
-        rank = int(os.environ["RANK"])
-        rank_filepath = (
-            f"pruning/pruning_scores/ffn_iterative/100samples_diverse_mini/rank_{rank}.pth"
-        )
-        assert (puzzle_dir / rank_filepath).is_file()
-
-        # assertions for the pruning_ckpts step
-        assert (puzzle_dir / "ckpts/ffn_256_attn_no_op").exists()
-
-    dist.cleanup()
-
-
 def test_nas_convert_attn_pruning(project_root_path: Path, tmp_path: Path):
     spawn_multiprocess_job(
         size=torch.cuda.device_count(),
diff --git a/tests/gpu/torch/puzzletron/test_nas_search.py b/tests/gpu/torch/puzzletron/test_nas_search.py
index ae0dff47be5..4cbd95ca2f9 100644
--- a/tests/gpu/torch/puzzletron/test_nas_search.py
+++ b/tests/gpu/torch/puzzletron/test_nas_search.py
@@ -76,9 +76,16 @@ def _test_nas_search_multiprocess_job(
     )
 
     #
-    # Check assertions for mtn.search() step
+    # Check assertions for the mtn.convert() and mtn.search() steps
     #
     if rank == 0:
+        # assertions for the score_pruning_activations step of mtn.convert()
+        rank_filepath = "pruning/pruning_scores/ffn_iterative/100samples_diverse_mini/rank_0.pth"
+        assert (puzzle_dir / rank_filepath).is_file()
+
+        # assertions for the pruning_ckpts step of mtn.convert()
+        assert (puzzle_dir / "ckpts/ffn_256_attn_no_op").exists()
+
         # assertions for the build_library_and_stats step
         assert (puzzle_dir / "replacement_library.json").is_file()
         assert (puzzle_dir / "subblock_stats.json").is_file()
diff --git a/tests/gpu/torch/quantization/test_gptq.py b/tests/gpu/torch/quantization/test_gptq.py
index 496db08578a..abe3386064a 100644
--- a/tests/gpu/torch/quantization/test_gptq.py
+++ b/tests/gpu/torch/quantization/test_gptq.py
@@ -18,6 +18,7 @@
 
 import pytest
 import torch
+from _test_utils.torch.calib_data import write_calib_jsonl
 from _test_utils.torch.transformers_models import get_tiny_llama
 from conftest import requires_triton
 
@@ -130,14 +131,14 @@ def test_gptq_export_roundtrip():
 @pytest.mark.parametrize(
     "quant_cfg", [mtq.NVFP4_DEFAULT_CFG, mtq.FP8_DEFAULT_CFG, mtq.INT4_BLOCKWISE_WEIGHT_ONLY_CFG]
 )
-def test_gptq_e2e_flow(quant_cfg, tiny_tokenizer):
+def test_gptq_e2e_flow(quant_cfg, tiny_tokenizer, tmp_path):
     model = get_tiny_llama(vocab_size=tiny_tokenizer.vocab_size).to("cuda")
     model.eval()
 
     quant_cfg = copy.deepcopy(quant_cfg)
     quant_cfg["algorithm"] = {"method": "gptq", "layerwise": {"enable": True}}
     calib_dataloader = get_dataset_dataloader(
-        dataset_name="cnn_dailymail",
+        dataset_name=write_calib_jsonl(tmp_path / "calib.jsonl", num_samples=8),
         tokenizer=tiny_tokenizer,
         batch_size=2,
         num_samples=8,
diff --git a/tests/gpu/torch/utils/test_dataset_utils.py b/tests/gpu/torch/utils/test_dataset_utils.py
index bc9451ca03a..f83266762b6 100644
--- a/tests/gpu/torch/utils/test_dataset_utils.py
+++ b/tests/gpu/torch/utils/test_dataset_utils.py
@@ -42,6 +42,9 @@ def qwen3_tokenizer():
     return tokenizer
 
 
+@pytest.mark.release(
+    reason="Streams the 7 gated Nemotron v3 datasets; runs on the nightly schedule."
+)
 def test_get_dataset_dataloader_nemotron_v3_chat_template(qwen3_tokenizer):
     if not get_token():
         pytest.skip(
diff --git a/tests/gpu_megatron/conftest.py b/tests/gpu_megatron/conftest.py
index 5760b9fe22d..8b4490e314e 100644
--- a/tests/gpu_megatron/conftest.py
+++ b/tests/gpu_megatron/conftest.py
@@ -13,11 +13,17 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 import contextlib
+import gc
 
 import pytest
 import torch
 from _test_utils.fs_utils import assert_unmodified_tree
-from _test_utils.torch.distributed.utils import DistributedWorkerPool, get_live_pool
+from _test_utils.torch.distributed.utils import (
+    DistributedWorkerPool,
+    destroy_extra_process_groups,
+    get_live_pool,
+    reset_worker_state,
+)
 from _test_utils.torch.transformers_models import get_tiny_tokenizer
 from megatron.core.parallel_state import destroy_model_parallel
 
@@ -68,6 +74,8 @@ def _prestart_preprocess_forkserver():
 
 def megatron_worker_teardown(rank, world_size):
     """Clean up model-parallel state between tests in persistent workers."""
+    # Surface asynchronous CUDA errors at the test that caused them, not at a later one
+    torch.cuda.synchronize()
     if dist.is_initialized():
         dist.barrier()
     try:
@@ -79,6 +87,9 @@ def megatron_worker_teardown(rank, world_size):
             apex_destroy()
         except Exception as e:
             print(f"Error destroying model parallel with Apex: {e}")
+    # destroy_model_parallel() drops Megatron's references but leaves the NCCL groups behind
+    destroy_extra_process_groups()
+    gc.collect()
     torch.cuda.empty_cache()
 
 
@@ -90,16 +101,17 @@ def _make_pool(world_size):
     )
 
 
-@pytest.fixture(scope="module")
+@pytest.fixture(scope="session")
 def _pool_cache():
-    """Module-scoped cache of worker pools keyed by world_size.
+    """Session-scoped cache of worker pools keyed by world_size.
 
     Spinning up a pool cold-imports the full torch/megatron/modelopt stack per worker
-    (~tens of seconds), so fixtures that request the same world_size share one pool
-    instead of spawning duplicates — e.g. on a 2-GPU runner ``dist_workers`` and
-    ``dist_workers_size_2`` are both size 2. The cache is module-scoped and torn down at
-    module end, so workers are never reused across modules (avoids cross-test
-    state contamination).
+    (75-100 s in CI), so every module that requests the same world_size shares one pool,
+    e.g. on a 2-GPU runner ``dist_workers`` and ``dist_workers_size_2`` are both size 2.
+    Isolation between tests comes from ``megatron_worker_teardown`` (model-parallel state and
+    process groups after every job), from ``_reset_shared_pools`` (process-global torch settings
+    at each module boundary) and from ``get_live_pool`` replacing a pool that a hung, timed-out
+    or crashed job killed.
     """
     pools: dict[int, DistributedWorkerPool] = {}
     yield pools
@@ -107,21 +119,29 @@ def _pool_cache():
         pool.shutdown()
 
 
+@pytest.fixture(scope="module", autouse=True)
+def _reset_shared_pools(_pool_cache):
+    """Workers outlive test modules: undo what the previous module did to process-global torch state."""
+    for pool in _pool_cache.values():
+        if not pool.broken:
+            pool.run(reset_worker_state)
+
+
 def _get_pool(cache, world_size):
     # A hung test kills its pool; a fresh one keeps a single hang from timing out every later
-    # test in the module (that cascade cost the nightly 20+ min).
+    # test (that cascade cost the nightly 20+ min).
     return get_live_pool(cache, world_size, lambda: _make_pool(world_size))
 
 
 @pytest.fixture
 def dist_workers(_pool_cache):
-    """Pool (cached per module) with world_size=torch.cuda.device_count()."""
+    """Shared pool with world_size=torch.cuda.device_count()."""
     return _get_pool(_pool_cache, torch.cuda.device_count())
 
 
 @pytest.fixture
 def dist_workers_size_1(_pool_cache):
-    """Pool (cached per module) with world_size=1 for tests that require a single process."""
+    """Shared pool with world_size=1 for tests that require a single process."""
     return _get_pool(_pool_cache, 1)
 
 
diff --git a/tests/gpu_megatron/torch/nas/plugins/test_megatron_hybrid_dynamic_modules.py b/tests/gpu_megatron/torch/nas/plugins/test_megatron_hybrid_dynamic_modules.py
index 8c98f7127fa..32df35a719c 100644
--- a/tests/gpu_megatron/torch/nas/plugins/test_megatron_hybrid_dynamic_modules.py
+++ b/tests/gpu_megatron/torch/nas/plugins/test_megatron_hybrid_dynamic_modules.py
@@ -14,6 +14,7 @@
 # limitations under the License.
 
 
+import pytest
 import torch
 from _test_utils.import_helper import skip_if_no_mamba
 
@@ -133,6 +134,7 @@ def _test_mamba_search_space(rank, size):
     assert not any(named_dynamic_modules(model))
 
 
+@pytest.mark.timeout(240)  # The first Mamba (Triton) kernel compile takes 75-100+ s on a cold cache
 def test_mamba_search_space(dist_workers):
     dist_workers.run(_test_mamba_search_space)
 
diff --git a/tests/gpu_megatron/torch/utils/plugins/test_megatron_preprocess_data.py b/tests/gpu_megatron/torch/utils/plugins/test_megatron_preprocess_data.py
index 5dd0815080c..be96e50e037 100644
--- a/tests/gpu_megatron/torch/utils/plugins/test_megatron_preprocess_data.py
+++ b/tests/gpu_megatron/torch/utils/plugins/test_megatron_preprocess_data.py
@@ -49,7 +49,7 @@ def test_worker_context_falls_back_to_spawn_without_forkserver(monkeypatch):
     strict=False,
     reason="Flaky: transient HF datasets-server 5xx on the /splits lookup",
 )
-def test_megatron_preprocess_data_with_jsonl_path(tmp_path):
+def test_megatron_preprocess_data_with_jsonl_path(tmp_path, tiny_tokenizer_path):
     input_jsonl = tmp_path / "minipile.jsonl"
     input_jsonl.write_text(
         "".join(json.dumps({"text": f"Sample document {index}."}) + "\n" for index in range(4)),
@@ -67,7 +67,7 @@ def test_megatron_preprocess_data_with_jsonl_path(tmp_path):
     prefixes = megatron_preprocess_data(
         jsonl_paths=input_jsonl,
         output_dir=tmp_path,
-        tokenizer_name_or_path="gpt2",
+        tokenizer_name_or_path=tiny_tokenizer_path,
         json_keys=["text"],
         workers=1,
     )
@@ -81,7 +81,7 @@ def test_megatron_preprocess_data_with_jsonl_path(tmp_path):
     assert Path(prefixes[0] + ".idx").stat().st_size > 0, "Index file should not be empty"
 
 
-def test_megatron_preprocess_data_jsonl_stops_at_max_tokens(tmp_path):
+def test_megatron_preprocess_data_jsonl_stops_at_max_tokens(tmp_path, tiny_tokenizer_path):
     input_path = tmp_path / "data.jsonl"
     input_path.write_text(
         "".join(json.dumps({"text": f"Document {index} " * 100}) + "\n" for index in range(10)),
@@ -91,7 +91,7 @@ def test_megatron_preprocess_data_jsonl_stops_at_max_tokens(tmp_path):
     common_args = {
         "jsonl_paths": input_path,
         "output_dir": tmp_path,
-        "tokenizer_name_or_path": "gpt2",
+        "tokenizer_name_or_path": tiny_tokenizer_path,
         "json_keys": "text",
         "workers": 2,
     }
@@ -115,13 +115,13 @@ def test_megatron_preprocess_data_jsonl_stops_at_max_tokens(tmp_path):
     strict=False,
     reason="Flaky: transient HF datasets-server 5xx on the /splits lookup",
 )
-def test_megatron_preprocess_data_hf_split_stops_at_max_tokens(tmp_path):
+def test_megatron_preprocess_data_hf_split_stops_at_max_tokens(tmp_path, tiny_tokenizer_path):
     common_args = {
         "hf_dataset": "nanotron/minipile_100_samples",
         "hf_split": "train",
         "hf_max_samples_per_split": 100,
         "hf_streaming": True,
-        "tokenizer_name_or_path": "gpt2",
+        "tokenizer_name_or_path": tiny_tokenizer_path,
         "json_keys": "text",
         "workers": 2,
     }
@@ -147,7 +147,9 @@ def test_megatron_preprocess_data_hf_split_stops_at_max_tokens(tmp_path):
     strict=False,
     reason="Flaky: transient HF datasets-server 5xx on the /splits lookup",
 )
-def test_megatron_preprocess_data_hf_split_resume_uses_cached_token_count(tmp_path):
+def test_megatron_preprocess_data_hf_split_resume_uses_cached_token_count(
+    tmp_path, tiny_tokenizer_path
+):
     args = {
         "hf_dataset": "nanotron/minipile_100_samples",
         "hf_split": "train",
@@ -155,7 +157,7 @@ def test_megatron_preprocess_data_hf_split_resume_uses_cached_token_count(tmp_pa
         "hf_streaming": True,
         "max_tokens": 100,
         "output_dir": tmp_path,
-        "tokenizer_name_or_path": "gpt2",
+        "tokenizer_name_or_path": tiny_tokenizer_path,
         "json_keys": "text",
         "max_sequence_length": 16,
         "workers": 2,
@@ -253,7 +255,7 @@ def test_megatron_preprocess_data_reasoning_content(tmp_path):
     )
 
 
-def test_megatron_preprocess_data_with_gzip_input(tmp_path):
+def test_megatron_preprocess_data_with_gzip_input(tmp_path, tiny_tokenizer_path):
     gz_path = tmp_path / "data.jsonl.gz"
     with gzip.open(gz_path, "wt", encoding="utf-8") as f:
         for i in range(10):
@@ -262,7 +264,7 @@ def test_megatron_preprocess_data_with_gzip_input(tmp_path):
     prefixes = megatron_preprocess_data(
         jsonl_paths=gz_path,
         output_dir=tmp_path,
-        tokenizer_name_or_path="gpt2",
+        tokenizer_name_or_path=tiny_tokenizer_path,
         json_keys=["text"],
         workers=1,
         strip_newlines=True,
@@ -324,7 +326,7 @@ def test_megatron_preprocess_data_tool_calls_arguments_normalized(tmp_path):
     strict=False,
     reason="Flaky: transient HF datasets-server 5xx on the /splits lookup",
 )
-def test_megatron_preprocess_data_hf_streaming_warning(tmp_path):
+def test_megatron_preprocess_data_hf_streaming_warning(tmp_path, tiny_tokenizer_path):
     # hf_streaming without hf_max_samples_per_split should warn and fall back to non-streaming
     with pytest.warns(UserWarning, match="hf_streaming"):
         megatron_preprocess_data(
@@ -332,7 +334,7 @@ def test_megatron_preprocess_data_hf_streaming_warning(tmp_path):
             hf_split="train",
             hf_streaming=True,
             output_dir=tmp_path,
-            tokenizer_name_or_path="gpt2",
+            tokenizer_name_or_path=tiny_tokenizer_path,
             json_keys=["text"],
             workers=1,
         )
diff --git a/tests/gpu_megatron/torch/utils/plugins/test_utils_megatron.py b/tests/gpu_megatron/torch/utils/plugins/test_utils_megatron.py
index 8d3af2d7f3c..286eea97a5b 100644
--- a/tests/gpu_megatron/torch/utils/plugins/test_utils_megatron.py
+++ b/tests/gpu_megatron/torch/utils/plugins/test_utils_megatron.py
@@ -64,6 +64,9 @@ def _test_megatron_generate_and_mmlu(rank, size, parallelism):
     assert 0.36 < megatron_mmlu(model, tokenizer, fraction=0.1, batch_size=16) < 0.39
 
 
+@pytest.mark.release(
+    reason="Real Qwen3-0.6B weights and a 1.2-1.5 GB download; run with --run-release."
+)
 @pytest.mark.parametrize("parallelism", ["tp", "pp", "cp", "dp"])
 def test_megatron_generate_and_mmlu(dist_workers, parallelism, num_gpus):
     if num_gpus == 1 and parallelism != "tp":
diff --git a/tests/gpu_vllm/torch/quantization/test_vllm_dynamic_modules.py b/tests/gpu_vllm/torch/quantization/test_vllm_dynamic_modules.py
index b8a0464b841..6e0fd19dc68 100644
--- a/tests/gpu_vllm/torch/quantization/test_vllm_dynamic_modules.py
+++ b/tests/gpu_vllm/torch/quantization/test_vllm_dynamic_modules.py
@@ -1083,7 +1083,13 @@ def _missing(module, name, slots):
     }
 
 
-def _boot_llm(model_dir, max_model_len=64, max_num_seqs=1, **extra):
+_MAX_MODEL_LEN = 64
+# One 64-token request needs a handful of KV blocks. The defaults size the KV pool from the memory
+# fraction (about 17 GB, millions of Python block objects) and the token budget to thousands.
+_SMALL_ENGINE = {"num_gpu_blocks_override": 32, "max_num_batched_tokens": _MAX_MODEL_LEN}
+
+
+def _boot_llm(model_dir, max_model_len=_MAX_MODEL_LEN, max_num_seqs=1, **extra):
     """Construct a vLLM engine on a tiny model.
 
     MoE fixtures override with ``moe_backend="triton"`` (pins the Triton
@@ -1154,7 +1160,9 @@ def _tiny_moe_llm(tmp_path_factory, model_type):
         num_experts_per_tok=2,
         **expert_config,
     )
-    llm = _boot_llm(model_dir, moe_backend="triton", enable_expert_parallel=True)
+    # GPT-OSS has sliding-window layers; the KV pool override is not verified on them.
+    small = _SMALL_ENGINE if model_type == "qwen3_moe" else {}
+    llm = _boot_llm(model_dir, moe_backend="triton", enable_expert_parallel=True, **small)
     try:
         yield llm
     finally:
@@ -1184,7 +1192,7 @@ def tiny_deepseek_llm(tmp_path_factory):
     model_dir = create_tiny_deepseek_v3_dir(
         tmp, kv_lora_rank=512, qk_nope_head_dim=128, qk_rope_head_dim=64, v_head_dim=128
     )
-    llm = _boot_llm(model_dir, moe_backend="triton", enable_expert_parallel=True)
+    llm = _boot_llm(model_dir, moe_backend="triton", enable_expert_parallel=True, **_SMALL_ENGINE)
     try:
         yield llm
     finally:
@@ -1782,7 +1790,11 @@ def tiny_deepseek_fp8_llm(tmp_path):
         tmp_path, kv_lora_rank=512, qk_nope_head_dim=128, qk_rope_head_dim=64, v_head_dim=128
     )
     llm = _boot_llm(
-        model_dir, quantization="fp8", moe_backend="triton", enable_expert_parallel=True
+        model_dir,
+        quantization="fp8",
+        moe_backend="triton",
+        enable_expert_parallel=True,
+        **_SMALL_ENGINE,
     )
     try:
         yield llm
diff --git a/tests/regression/torch/speculative/test_dflash.py b/tests/regression/torch/speculative/test_dflash.py
index 18e95148fcd..82a80a17153 100644
--- a/tests/regression/torch/speculative/test_dflash.py
+++ b/tests/regression/torch/speculative/test_dflash.py
@@ -61,6 +61,10 @@
     "dflash.dflash_architecture_config.num_hidden_layers=2",
 ]
 
+# Steps, not epochs: accelerate runs one process per GPU, so an epoch is 500 steps on 1 GPU, 250 on 2.
+_TRAIN_STEPS = 500
+_RESUME_STEPS = 20
+
 
 @pytest.fixture(scope="session")
 def qwen3_model_name():
@@ -80,8 +84,8 @@ def test_dflash_training(qwen3_model_name, dflash_output_dir):
     overrides = [
         f"model.model_name_or_path={qwen3_model_name}",
         f"training.output_dir={output_dir}",
-        "training.num_train_epochs=3",
-        "training.save_steps=500",
+        f"training.max_steps={_TRAIN_STEPS}",
+        f"training.save_steps={_TRAIN_STEPS}",
         *_DFLASH_OVERRIDES,
     ]
 
@@ -108,7 +112,7 @@ def test_dflash_training(qwen3_model_name, dflash_output_dir):
     first_loss = float(logs[0]["loss"])
     final_loss = float(logs[-1]["loss"])
     assert final_loss < first_loss, f"Loss did not decrease: {first_loss:.3f} -> {final_loss:.3f}"
-    # Sanity: final loss should be reasonable (baseline: ~1.1 on L40)
+    # Sanity: final loss should be reasonable (baseline: ~1.8 at step 500 on L40)
     assert final_loss < 3.0, f"Final loss {final_loss:.3f} too high (expected < 3.0)"
 
 
@@ -118,7 +122,7 @@ def test_dflash_resume(qwen3_model_name, dflash_output_dir):
     overrides = [
         f"model.model_name_or_path={qwen3_model_name}",
         f"training.output_dir={output_dir}",
-        "training.num_train_epochs=4",
+        f"training.max_steps={_TRAIN_STEPS + _RESUME_STEPS}",
         "training.save_steps=5000",
         *_DFLASH_OVERRIDES,
     ]
@@ -128,6 +132,11 @@ def test_dflash_resume(qwen3_model_name, dflash_output_dir):
         "speculative_decoding",
     )
 
+    with open(os.path.join(output_dir, "trainer_state.json")) as f:
+        state = json.load(f)
+    assert state["global_step"] == _TRAIN_STEPS + _RESUME_STEPS, state["global_step"]
+    assert any(h["step"] > _TRAIN_STEPS for h in state["log_history"]), "no log entry after resume"
+
 
 def test_dflash_export(dflash_output_dir):
     """Export DFlash checkpoint to deployment format."""
diff --git a/tests/gpu/onnx/quantization/test_dla_qdq_rules.py b/tests/unit/onnx/quantization/test_dla_qdq_rules.py
similarity index 100%
rename from tests/gpu/onnx/quantization/test_dla_qdq_rules.py
rename to tests/unit/onnx/quantization/test_dla_qdq_rules.py
diff --git a/tests/gpu/onnx/quantization/test_dla_scalar_constants.py b/tests/unit/onnx/quantization/test_dla_scalar_constants.py
similarity index 100%
rename from tests/gpu/onnx/quantization/test_dla_scalar_constants.py
rename to tests/unit/onnx/quantization/test_dla_scalar_constants.py
diff --git a/tests/gpu/onnx/quantization/test_quantization_initializer_names.py b/tests/unit/onnx/quantization/test_quantization_initializer_names.py
similarity index 100%
rename from tests/gpu/onnx/quantization/test_quantization_initializer_names.py
rename to tests/unit/onnx/quantization/test_quantization_initializer_names.py
diff --git a/tests/unit/test_distributed_worker_pool.py b/tests/unit/test_distributed_worker_pool.py
new file mode 100644
index 00000000000..e41bc8e7420
--- /dev/null
+++ b/tests/unit/test_distributed_worker_pool.py
@@ -0,0 +1,185 @@
+# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+# SPDX-License-Identifier: Apache-2.0
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+"""CPU (gloo) tests for what ``DistributedWorkerPool`` adds to its hang handling (see tests/gpu)."""
+
+import multiprocessing
+import os
+import signal
+import time
+from functools import partial
+from pathlib import Path
+
+import pytest
+import torch
+import torch.distributed as dist
+from _test_utils.torch.distributed.utils import (
+    DistributedWorkerPool,
+    destroy_extra_process_groups,
+    reset_worker_state,
+)
+
+pytestmark = [pytest.mark.usefixtures("skip_on_windows"), pytest.mark.timeout(120)]
+
+
+def _touch(rank, size, directory, name="ran"):
+    (Path(directory) / f"{name}{rank}").write_text(str(os.getpid()))
+
+
+def _raise_on_rank1(rank, size):
+    if rank == 1:
+        raise ValueError("boom")
+
+
+def _hang(rank, size):
+    time.sleep(3600)
+
+
+def _die_on_rank0(rank, size):
+    if rank == 0:
+        os._exit(13)
+    time.sleep(3600)
+
+
+def _extra_groups_are_destroyed(rank, size, directory):
+    from torch.distributed.distributed_c10d import _world
+
+    groups = [dist.new_group([0, 1]) for _ in range(3)]
+    dist.all_reduce(torch.ones(1), group=groups[0])
+    assert len(_world.pg_map) == 1 + 3
+
+    destroy_extra_process_groups()
+    assert len(_world.pg_map) == 1
+
+    # Groups can be created and used again afterwards, and the default group still works
+    dist.all_reduce(torch.ones(1), group=dist.new_group([0, 1]))
+    destroy_extra_process_groups()
+    dist.all_reduce(torch.ones(1))
+    (Path(directory) / f"groups{rank}").write_text("ok")
+
+
+def _flags_are_reset(rank, size, directory):
+    baseline = (
+        torch.backends.cudnn.deterministic,
+        torch.backends.cudnn.benchmark,
+        torch.backends.cudnn.allow_tf32,
+        torch.backends.cuda.matmul.allow_tf32,
+        torch.are_deterministic_algorithms_enabled(),
+    )
+    torch.backends.cudnn.deterministic = not baseline[0]
+    torch.backends.cudnn.benchmark = not baseline[1]
+    torch.backends.cudnn.allow_tf32 = not baseline[2]
+    torch.backends.cuda.matmul.allow_tf32 = not baseline[3]
+    torch.use_deterministic_algorithms(not baseline[4])
+    torch.manual_seed(5)
+
+    reset_worker_state(rank, size)
+
+    assert baseline == (
+        torch.backends.cudnn.deterministic,
+        torch.backends.cudnn.benchmark,
+        torch.backends.cudnn.allow_tf32,
+        torch.backends.cuda.matmul.allow_tf32,
+        torch.are_deterministic_algorithms_enabled(),
+    )
+    assert torch.initial_seed() == 1234
+    (Path(directory) / f"flags{rank}").write_text("ok")
+
+
+class _Interrupt(BaseException):
+    """Stands in for pytest-timeout's ``Failed``, which is also a ``BaseException``."""
+
+
+@pytest.fixture
+def pool():
+    pool = DistributedWorkerPool(world_size=2, backend="gloo", teardown_fn=None, ready_timeout=90)
+    yield pool
+    pool.shutdown()
+
+
+def _pids(pool):
+    return [p.pid for p in pool._processes]
+
+
+def _is_dead(pid):
+    try:
+        os.kill(pid, 0)
+    except ProcessLookupError:
+        return True
+    return False
+
+
+def test_job_error_keeps_the_workers(pool, tmp_path):
+    pids = _pids(pool)
+    pool.run(partial(_touch, directory=tmp_path))
+    with pytest.raises(RuntimeError, match="boom"):
+        pool.run(_raise_on_rank1)
+    pool.run(partial(_touch, directory=tmp_path, name="again"))
+
+    assert not pool.broken
+    assert _pids(pool) == pids
+    assert {int(p.read_text()) for p in tmp_path.glob("again*")} == set(pids)
+
+
+def test_interrupt_kills_the_workers_and_breaks_the_pool(pool):
+    # What pytest-timeout's signal method does: raise in the main thread while the run waits.
+    def interrupt(signum, frame):
+        raise _Interrupt
+
+    pids = _pids(pool)
+    previous = signal.signal(signal.SIGALRM, interrupt)
+    signal.setitimer(signal.ITIMER_REAL, 2)
+    try:
+        with pytest.raises(_Interrupt):
+            pool.run(_hang)
+    finally:
+        signal.setitimer(signal.ITIMER_REAL, 0)
+        signal.signal(signal.SIGALRM, previous)
+
+    assert pool.broken
+    assert all(_is_dead(pid) for pid in pids)
+    with pytest.raises(RuntimeError, match="killed"):
+        pool.run(_hang)
+
+
+def test_dead_worker_fails_fast_and_breaks_the_pool(pool):
+    pids = _pids(pool)
+    started = time.monotonic()
+    with pytest.raises(RuntimeError, match=r"died.*13"):
+        pool.run(_die_on_rank0)
+    assert time.monotonic() - started < 20  # not the 600 s job timeout
+
+    assert pool.broken
+    assert all(_is_dead(pid) for pid in pids)  # the survivor, stuck in the job, is killed too
+
+
+def test_failed_start_raises_and_leaves_no_workers():
+    before = {p.pid for p in multiprocessing.active_children()}
+    started = time.monotonic()
+    with pytest.raises(RuntimeError, match="died"):
+        DistributedWorkerPool(world_size=2, backend="not-a-backend", teardown_fn=None)
+    assert time.monotonic() - started < 60  # not the 300 s ready timeout
+    time.sleep(1)
+    assert {p.pid for p in multiprocessing.active_children()} <= before
+
+
+def test_extra_process_groups_are_destroyed(pool, tmp_path):
+    pool.run(partial(_extra_groups_are_destroyed, directory=tmp_path))
+    assert len(list(tmp_path.glob("groups*"))) == 2
+
+
+def test_reset_worker_state_restores_the_spawn_time_settings(pool, tmp_path):
+    pool.run(partial(_flags_are_reset, directory=tmp_path))
+    assert len(list(tmp_path.glob("flags*"))) == 2
diff --git a/tests/unit/test_example_runner.py b/tests/unit/test_example_runner.py
new file mode 100644
index 00000000000..7660676fbbf
--- /dev/null
+++ b/tests/unit/test_example_runner.py
@@ -0,0 +1,616 @@
+# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+# SPDX-License-Identifier: Apache-2.0
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+"""Tests for ``ExampleRunner``: example steps run in a long-lived worker and leave no state behind."""
+
+import contextlib
+import json
+import os
+import re
+import signal
+import subprocess
+import sys
+import textwrap
+import threading
+import time
+from pathlib import Path
+
+import pytest
+from _test_utils.examples import run_command
+from _test_utils.examples.example_runner import (
+    ExampleRunner,
+    StepFailed,
+    StepTimeout,
+    keep_post_conversion_plugins,
+)
+
+pytestmark = pytest.mark.usefixtures("skip_on_windows")
+
+# Decorators run when the module is imported, which also happens on Windows (the tests then skip),
+# and signal has no SIGKILL there.
+_SIGKILL = getattr(signal, "SIGKILL", 9)
+
+
+class _Lane:
+    """One example directory of a runner, with helpers to write and run its scripts."""
+
+    def __init__(self, runner, directory):
+        directory.mkdir(parents=True)
+        self.runner, self.dir = runner, directory
+
+    def write(self, name, body):
+        (self.dir / name).write_text(textwrap.dedent(body))
+
+    def run(self, script, *args, **kwargs):
+        return self.runner(["python", script, *args], self.dir.name, **kwargs)
+
+    def probe(self, script="probe.py"):
+        return json.loads(self.run(script).strip().splitlines()[-1])
+
+
+def _is_alive(pid):
+    try:
+        os.kill(pid, 0)
+    except ProcessLookupError:
+        return False
+    try:  # a zombie is dead; nothing reaps it where pid 1 does not
+        return Path(f"/proc/{pid}/stat").read_text().rsplit(")", 1)[1].split()[0] != "Z"
+    except OSError:
+        return True
+
+
+def _wait_until_dead(pid):
+    for _ in range(100):
+        if not _is_alive(pid):
+            return
+        time.sleep(0.1)
+    os.kill(pid, signal.SIGKILL)
+    pytest.fail(f"process {pid} outlived its step")
+
+
+@contextlib.contextmanager
+def _in_process_lane(tmp_path, hooks=()):
+    runner = ExampleRunner((), hooks, tmp_path, preload=(), in_process=True)
+    try:
+        yield _Lane(runner, tmp_path / "example")
+    finally:
+        runner.close()
+
+
+def _lane_factory(runner, examples_dir, request):
+    name = re.sub(r"\W+", "_", request.node.name)
+    return lambda suffix="": _Lane(runner, examples_dir / f"{name}{suffix}")
+
+
+@pytest.fixture(scope="module")
+def examples_dir(tmp_path_factory):
+    return tmp_path_factory.mktemp("examples")
+
+
+@pytest.fixture(scope="module")
+def worker_runner(examples_dir):
+    # No preload: every failed step recycles the worker, and importing torch takes seconds.
+    runner = ExampleRunner((), examples_dir=examples_dir, preload=(), in_process=False)
+    yield runner
+    runner.close()
+
+
+@pytest.fixture(scope="module", params=["worker", "in_process"])
+def backend(request):
+    return request.param
+
+
+@pytest.fixture(scope="module")
+def any_runner(backend, examples_dir, worker_runner):
+    if backend == "worker":
+        yield worker_runner
+        return
+    runner = ExampleRunner((), examples_dir=examples_dir, preload=(), in_process=True)
+    yield runner
+    runner.close()
+
+
+@pytest.fixture
+def make_lane(worker_runner, examples_dir, request):
+    """Lanes of the shared worker."""
+    return _lane_factory(worker_runner, examples_dir, request)
+
+
+@pytest.fixture
+def make_lane_on_both_backends(any_runner, examples_dir, request):
+    """Lanes of the shared worker, and of an in-process runner, as the test runs once per backend."""
+    return _lane_factory(any_runner, examples_dir, request)
+
+
+@pytest.fixture
+def worker_lane(tmp_path):
+    """A lane of a worker of its own, for tests that kill or stop it."""
+    runner = ExampleRunner((), examples_dir=tmp_path, preload=(), in_process=False)
+    yield _Lane(runner, tmp_path / "example")
+    runner.close()
+
+
+def test_step_runs_like_python_script_py(make_lane_on_both_backends):
+    lane = make_lane_on_both_backends()
+    lane.write(
+        "s.py",
+        """
+        import argparse, os, sys
+
+        def main():
+            parser = argparse.ArgumentParser()
+            parser.add_argument("--x", type=int)
+            cwd, argv0 = os.path.basename(os.getcwd()), os.path.basename(sys.argv[0])
+            print("x", parser.parse_args().x, "cwd", cwd, "argv0", argv0)
+
+        if __name__ == "__main__":
+            main()
+        """,
+    )
+    assert f"x 3 cwd {lane.dir.name} argv0 s.py" in lane.run("s.py", "--x", "3")
+
+
+def test_failed_step_raises_step_failed_and_the_runner_recovers(make_lane_on_both_backends):
+    lane = make_lane_on_both_backends()
+    lane.write("bad.py", "raise ValueError('bad')")
+    lane.write("ok.py", "print('still fine')")
+    with pytest.raises(StepFailed) as failure:
+        lane.run("bad.py")
+    assert isinstance(failure.value, subprocess.CalledProcessError)
+    assert failure.value.returncode == 1
+    assert "ValueError: bad" in failure.value.output
+    assert failure.value.captured_output == failure.value.output
+    assert "still fine" in lane.run("ok.py")
+
+
+@pytest.mark.parametrize(
+    ("body", "returncode", "text"),
+    [
+        ("import sys; sys.exit(3)", 3, ""),
+        ("import sys; sys.exit('boom')", 1, "boom"),
+        ("import argparse; argparse.ArgumentParser().parse_args(['--nope'])", 2, "unrecognized"),
+    ],
+)
+def test_exit_statuses_map_to_the_return_code(make_lane, body, returncode, text):
+    lane = make_lane()
+    lane.write("bad.py", body)
+    with pytest.raises(StepFailed) as failure:
+        lane.run("bad.py")
+    assert failure.value.returncode == returncode
+    assert text in failure.value.output
+
+
+def test_sys_exit_zero_is_success(make_lane):
+    lane = make_lane()
+    lane.write("z.py", "import sys; print('done'); sys.exit(0)")
+    assert "done" in lane.run("z.py")
+
+
+def test_process_state_changed_by_a_step_is_restored(make_lane_on_both_backends):
+    lane = make_lane_on_both_backends()
+    lane.write(
+        "leak.py",
+        """
+        import logging, os, signal, sys, warnings
+
+        os.environ["LEAK"] = "1"
+        os.chdir("/")
+        sys.path.insert(0, "/nonexistent-leak")
+        signal.signal(signal.SIGTERM, lambda *args: None)
+        logger = logging.getLogger("__main__")
+        logger.addHandler(logging.StreamHandler(sys.stdout))
+        logger.setLevel(10)
+        logging.getLogger().addHandler(logging.StreamHandler(sys.stdout))
+        logging.disable(logging.CRITICAL)
+        warnings.simplefilter("ignore")
+        sys.argv.append("leak")
+        """,
+    )
+    lane.write(
+        "probe.py",
+        """
+        import json, logging, os, signal, sys, warnings
+
+        print(json.dumps({
+            "leak": "LEAK" in os.environ,
+            "path": "/nonexistent-leak" in sys.path,
+            "argv": len(sys.argv),
+            "sigterm": str(signal.getsignal(signal.SIGTERM)),
+            "main_handlers": len(logging.getLogger("__main__").handlers),
+            "root_handlers": len(logging.getLogger().handlers),
+            "root_level": logging.getLogger().level,
+            "disabled": logging.root.manager.disable,
+            "filters": len(warnings.filters),
+        }))
+        """,
+    )
+    cwd, before = os.getcwd(), lane.probe()
+    lane.run("leak.py")
+    after = lane.probe()
+    assert after == before
+    assert after["leak"] is False
+    assert after["main_handlers"] == 0
+    assert os.getcwd() == cwd
+
+
+def test_random_state_start_method_and_backend_flags_are_restored(backend, examples_dir):
+    # Preload torch up front, as the real lanes do.
+    runner = ExampleRunner(
+        (), examples_dir=examples_dir, preload=("torch",), in_process=backend == "in_process"
+    )
+    lane = _Lane(runner, examples_dir / f"state_{backend}")
+    lane.write(
+        "mutate.py",
+        """
+        import multiprocessing, random
+        import numpy, torch
+
+        multiprocessing.set_start_method("forkserver", force=True)
+        random.seed(1)
+        numpy.random.seed(1)
+        torch.manual_seed(1)
+        torch.backends.cudnn.deterministic = True
+        torch.backends.cudnn.benchmark = True
+        """,
+    )
+    lane.write(
+        "probe.py",
+        """
+        import json, multiprocessing, random
+        import numpy, torch
+
+        print(json.dumps({
+            "start_method": multiprocessing.get_start_method(allow_none=True),
+            "python": random.random(),
+            "numpy": float(numpy.random.rand()),
+            "torch": float(torch.rand(1)),
+            "deterministic": torch.backends.cudnn.deterministic,
+            "benchmark": torch.backends.cudnn.benchmark,
+        }))
+        """,
+    )
+    try:
+        before = lane.probe()
+        lane.run("mutate.py")
+        assert lane.probe() == before
+    finally:
+        runner.close()
+
+
+def test_handlers_added_by_each_step_do_not_accumulate(make_lane):
+    # What diffusers' quantize.py does with its setup_logging().
+    lane = make_lane()
+    lane.write(
+        "logs.py",
+        """
+        import logging, sys
+
+        log = logging.getLogger(__name__)
+        log.setLevel(logging.INFO)
+        log.addHandler(logging.StreamHandler(sys.stdout))
+        log.info("LINE")
+        """,
+    )
+    for _ in range(3):
+        assert lane.run("logs.py").count("LINE") == 1
+
+
+def test_modules_of_the_example_dir_are_not_shared_but_imports_stay_cached(make_lane):
+    first, second = make_lane("_a"), make_lane("_b")
+    for lane, value in ((first, "A"), (second, "B")):
+        lane.write("utils.py", f"VALUE = '{value}'")
+        lane.write(
+            "main.py", "import sys, utils\nprint('got', utils.VALUE, 'colorsys' in sys.modules)"
+        )
+    first.write("warm.py", "import colorsys")
+    first.run("warm.py")
+    seen = [lane.run("main.py") for lane in (first, second, first)]
+    assert ["got A True" in seen[0], "got B True" in seen[1], "got A True" in seen[2]] == [True] * 3
+
+
+def test_env_is_the_whole_environment_of_the_step_and_is_restored(make_lane):
+    lane = make_lane()
+    lane.write("keys.py", "import os; print('KEYS', sorted(os.environ))")
+    assert "KEYS ['ONLY']" in lane.run("keys.py", env={"ONLY": "1"})
+    assert "ONLY" not in lane.run("keys.py")
+
+
+def test_output_of_fds_and_child_processes_is_captured(make_lane):
+    lane = make_lane()
+    lane.write(
+        "o.py",
+        """
+        import os, subprocess, sys
+
+        print("py")
+        print("err", file=sys.stderr)
+        os.write(1, b"fd1\\n")
+        subprocess.run(["echo", "child"])
+        """,
+    )
+    output = lane.run("o.py")
+    assert all(text in output for text in ("py", "err", "fd1", "child"))
+
+
+def test_step_can_start_child_processes(make_lane):
+    # DataLoader workers and Pools: a daemonic worker would refuse.
+    lane = make_lane()
+    lane.write(
+        "fork.py",
+        """
+        import multiprocessing
+
+        def child():
+            print("child ran")
+
+        if __name__ == "__main__":
+            process = multiprocessing.get_context("fork").Process(target=child)
+            process.start()
+            process.join()
+            print("exit", process.exitcode)
+        """,
+    )
+    output = lane.run("fork.py")
+    assert "child ran" in output
+    assert "exit 0" in output
+
+
+def test_hooks_wrap_every_step(tmp_path):
+    events = []
+
+    @contextlib.contextmanager
+    def hook():
+        events.append("enter")
+        try:
+            yield
+        finally:
+            events.append("exit")
+
+    with _in_process_lane(tmp_path, (hook,)) as lane:
+        lane.write("s.py", "print('ran')")
+        assert "ran" in lane.run("s.py")
+    assert events == ["enter", "exit"]
+
+
+def test_failing_hook_fails_the_step(tmp_path):
+    def hook():
+        raise RuntimeError("hook failed")
+
+    with _in_process_lane(tmp_path, (hook,)) as lane:
+        lane.write("s.py", "print('ran')")
+        with pytest.raises(StepFailed) as failure:
+            lane.run("s.py")
+    assert "hook failed" in failure.value.output
+    assert "ran" not in failure.value.output
+
+
+def test_keep_post_conversion_plugins_undoes_the_registrations_of_a_step(tmp_path):
+    from modelopt.torch.quantization.plugins.custom import CUSTOM_POST_CONVERSION_PLUGINS
+
+    before = set(CUSTOM_POST_CONVERSION_PLUGINS)
+    with _in_process_lane(tmp_path, (keep_post_conversion_plugins,)) as lane:
+        lane.write(
+            "s.py",
+            """
+            from modelopt.torch.quantization.plugins.custom import CUSTOM_POST_CONVERSION_PLUGINS
+
+            CUSTOM_POST_CONVERSION_PLUGINS.add(lambda model: None)
+            """,
+        )
+        lane.run("s.py")
+    assert before == CUSTOM_POST_CONVERSION_PLUGINS
+
+
+def test_worker_is_reused_until_a_step_fails(worker_lane):
+    worker_lane.write("pid.py", "import os; print('pid', os.getpid())")
+    worker_lane.write("bad.py", "raise SystemExit(1)")
+
+    def pid():
+        return worker_lane.run("pid.py").split()[-1]
+
+    first = pid()
+    assert pid() == first != str(os.getpid())
+    with pytest.raises(StepFailed):
+        worker_lane.run("bad.py")
+    assert pid() != first
+
+
+def test_timeout_kills_the_hung_step_with_its_children_and_the_next_step_gets_a_fresh_worker(
+    worker_lane,
+):
+    worker_lane.write("pid.py", "import os; print('pid', os.getpid())")
+    worker_lane.write(
+        "hang.py",
+        """
+        import subprocess, time
+
+        child = subprocess.Popen(["sleep", "60"])
+        print("child", child.pid, flush=True)
+        time.sleep(60)
+        """,
+    )
+    first = worker_lane.run("pid.py")
+    started = time.monotonic()
+    with pytest.raises(StepTimeout) as timeout:
+        worker_lane.run("hang.py", timeout=2)
+    assert time.monotonic() - started < 15
+    child = int(re.search(r"child (\d+)", timeout.value.captured_output).group(1))
+    _wait_until_dead(child)
+    assert worker_lane.run("pid.py") != first
+
+
+@pytest.mark.parametrize(
+    ("death", "returncode"),
+    [("os._exit(7)", 7), ("os.kill(os.getpid(), signal.SIGKILL)", -_SIGKILL)],
+)
+def test_worker_death_fails_the_step_not_the_session(worker_lane, death, returncode):
+    worker_lane.write("die.py", f"import os, signal\nprint('about to die', flush=True)\n{death}")
+    worker_lane.write("ok.py", "print('alive')")
+    with pytest.raises(StepFailed) as failure:
+        worker_lane.run("die.py")
+    assert failure.value.returncode == returncode
+    assert "about to die" in failure.value.output
+    assert "alive" in worker_lane.run("ok.py")
+
+
+def test_interrupt_while_waiting_kills_the_step(worker_lane):
+    # What pytest-timeout's signal method does: raise in the main thread while the runner waits.
+    class Interrupted(BaseException):
+        pass
+
+    def interrupt(*_):
+        raise Interrupted
+
+    worker_lane.write(
+        "hang.py",
+        """
+        import os, pathlib, time
+
+        pathlib.Path(__file__).with_name("pid").write_text(str(os.getpid()))
+        print("before hang", flush=True)
+        time.sleep(60)
+        """,
+    )
+    previous = signal.signal(signal.SIGUSR1, interrupt)
+    main_thread = threading.main_thread().ident
+    timer = threading.Timer(1.0, signal.pthread_kill, (main_thread, signal.SIGUSR1))
+    timer.start()
+    try:
+        with pytest.raises(Interrupted) as interrupted:
+            worker_lane.run("hang.py", timeout=30)
+    finally:
+        timer.cancel()
+        signal.signal(signal.SIGUSR1, previous)
+    assert "before hang" in interrupted.value.captured_output
+    _wait_until_dead(int((worker_lane.dir / "pid").read_text()))
+
+
+def test_close_lets_the_worker_exit_cleanly(worker_lane):
+    # A clean exit is what lets coverage write the worker's data.
+    worker_lane.write(
+        "marker.py",
+        """
+        import atexit, pathlib
+
+        marker = pathlib.Path(__file__).with_name("exited")
+        atexit.register(marker.write_text, "clean")
+        """,
+    )
+    worker_lane.run("marker.py")
+    marker = worker_lane.dir / "exited"
+    assert not marker.exists()
+    worker_lane.runner.close()
+    assert marker.read_text() == "clean"
+
+
+def test_accepts_only_listed_scripts_run_with_python():
+    runner = ExampleRunner({"a.py"}, preload=())
+    try:
+        assert runner.accepts(["python", "a.py", "--x", "1"])
+        assert runner.accepts(["python3", "a.py"])
+        assert runner.accepts([sys.executable, "a.py"])
+        assert not runner.accepts(["python", "b.py"])
+        assert not runner.accepts(["bash", "a.py"])
+        assert not runner.accepts(["python", "-m", "a"])
+        assert not runner.accepts(["python"])
+    finally:
+        runner.close()
+
+
+@pytest.fixture
+def installed_runner(tmp_path, monkeypatch):
+    """A worker for ``in_worker.py`` installed in ``run_command``; yields ``examples/demo``."""
+    monkeypatch.setattr(run_command, "MODELOPT_ROOT", tmp_path)
+    demo = tmp_path / "examples" / "demo"
+    demo.mkdir(parents=True)
+    runner = ExampleRunner({"in_worker.py"}, examples_dir=tmp_path / "examples", preload=())
+    run_command.set_in_process_runner(runner)
+    yield demo
+    run_command.set_in_process_runner(None)
+    runner.close()
+
+
+def _run_demo(script, **kwargs):
+    return run_command.run_example_command([sys.executable, script], "demo", **kwargs)
+
+
+def _demo_pid(script):
+    return int(_run_demo(script).split()[-1])
+
+
+def test_run_example_command_gives_listed_scripts_to_the_runner(installed_runner):
+    for name in ("in_worker.py", "elsewhere.py"):
+        (installed_runner / name).write_text("import os; print('pid', os.getpid())")
+    worker = _demo_pid("in_worker.py")
+    assert worker != os.getpid()
+    assert _demo_pid("in_worker.py") == worker
+    assert _demo_pid("elsewhere.py") not in (worker, os.getpid())
+
+
+def test_kill_switch_runs_listed_scripts_as_subprocesses(installed_runner, monkeypatch):
+    (installed_runner / "in_worker.py").write_text("import os; print('pid', os.getpid())")
+    monkeypatch.setenv("MODELOPT_EXAMPLE_RUNNER", "subprocess")
+    assert _demo_pid("in_worker.py") != _demo_pid("in_worker.py")
+
+
+def test_run_example_command_passes_env_to_the_runner(installed_runner):
+    (installed_runner / "in_worker.py").write_text("import os; print('KEYS', sorted(os.environ))")
+    assert "KEYS ['ONLY']" in _run_demo("in_worker.py", env={"ONLY": "1"})
+
+
+def test_failed_step_raises_called_process_error_through_run_example_command(installed_runner):
+    (installed_runner / "in_worker.py").write_text("raise ValueError('nope')")
+    with pytest.raises(subprocess.CalledProcessError) as failure:
+        _run_demo("in_worker.py")
+    assert "ValueError: nope" in failure.value.output
+
+
+def test_transient_hub_errors_are_retried_through_the_runner(installed_runner):
+    (installed_runner / "in_worker.py").write_text(
+        textwrap.dedent(
+            """
+            import pathlib
+
+            marker = pathlib.Path(__file__).with_name("tried")
+            if not marker.exists():
+                marker.write_text("1")
+                raise RuntimeError("ConnectionError: Max retries exceeded")
+            print("recovered")
+            """
+        )
+    )
+    with pytest.warns(UserWarning, match="transient HuggingFace access error"):
+        assert "recovered" in _run_demo("in_worker.py", hf_retry_delay_s=0)
+
+
+def test_plain_runners_keep_their_contract(tmp_path, monkeypatch):
+    monkeypatch.setattr(run_command, "MODELOPT_ROOT", tmp_path)
+    (tmp_path / "examples" / "demo").mkdir(parents=True)
+    calls = []
+
+    def plain_runner(cmd_parts, example_path):
+        calls.append((cmd_parts, example_path))
+        return "ran in process"
+
+    run_command.set_in_process_runner(plain_runner)
+    try:
+        assert run_command.run_example_command(["x.py"], "demo") == "ran in process"
+        with pytest.warns(UserWarning, match="env= given"):
+            output = run_command.run_example_command(
+                [sys.executable, "-c", "print('subprocess')"], "demo", env=os.environ.copy()
+            )
+    finally:
+        run_command.set_in_process_runner(None)
+    assert "subprocess" in output
+    assert calls == [(["x.py"], "demo")]
diff --git a/tests/unit/torch/nas/test_full_algorithms_vision.py b/tests/unit/torch/nas/test_full_algorithms_vision.py
index f6c30c1da52..b3d81881696 100644
--- a/tests/unit/torch/nas/test_full_algorithms_vision.py
+++ b/tests/unit/torch/nas/test_full_algorithms_vision.py
@@ -117,7 +117,7 @@ def run_search():
         pytest.param(False, False, marks=pytest.mark.manual),
     ],
 )
-def test_search_constraints(flops_only: bool, bounded_latency: bool):
+def test_search_constraints(flops_only: bool, bounded_latency: bool, monkeypatch):
     def _fake_latency(self, model, precomputed=None):
         coeff_params = 0.00 if flops_only else 0.75
         if not hasattr(_fake_latency, "call_counter"):
@@ -130,8 +130,9 @@ def _fake_latency(self, model, precomputed=None):
     dummy_input = (*args, kwargs)
     model = mtn.convert(model, mode="autonas")
 
-    # get fake latency and spoof ConstraintsFunc class
-    ConstraintsFunc._get_true_latency = _fake_latency
+    # get fake latency and spoof ConstraintsFunc class (undone after the test: a class-level patch
+    # that stays behind changes the latency of every later test in the same process)
+    monkeypatch.setattr(ConstraintsFunc, "_get_true_latency", _fake_latency)
 
     # now setup constraints functor
     constraints_func = ConstraintsFunc(model, {}, dummy_input, {"device": None})
diff --git a/tests/unit/torch/speculative/plugins/test_hf_dflash2.py b/tests/unit/torch/speculative/plugins/test_hf_dflash2.py
index 931a4a3ee46..a5c8410763c 100644
--- a/tests/unit/torch/speculative/plugins/test_hf_dflash2.py
+++ b/tests/unit/torch/speculative/plugins/test_hf_dflash2.py
@@ -349,6 +349,7 @@ def test_selector_loss_increases_total_loss(self):
             losses[alpha] = float(out.loss.detach())
         assert losses[1.0] > losses[0.0]
 
+    @pytest.mark.timeout(300)  # ~53 s with torch's default threads, longer single-threaded
     def test_overfits_a_single_batch(self):
         """A few steps on one batch drive backbone and selector accuracy up.