diff --git a/ci/tools/run-tests b/ci/tools/run-tests
index c5c0cc12336..ea3006ca102 100755
--- a/ci/tools/run-tests
+++ b/ci/tools/run-tests
@@ -1,6 +1,6 @@
 #!/usr/bin/env bash
 
-# SPDX-FileCopyrightText: Copyright (c) 2025 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+# SPDX-FileCopyrightText: Copyright (c) 2025-2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
 #
 # SPDX-License-Identifier: Apache-2.0
 
@@ -20,6 +20,14 @@ fi
 
 test_module=${1}
 
+FREE_THREADING=""
+PYTEST_PARALLEL_ARGS=()
+if python -c 'import sys; assert not sys._is_gil_enabled()' 2> /dev/null; then
+  FREE_THREADING="-ft"
+  PYTEST_PARALLEL_ARGS=(--parallel-threads=4)
+  pip install pytest-run-parallel
+fi
+
 # For standard modes, install pathfinder up front (it is a direct dependency
 # of bindings, and a transitive dependency of core).  Nightly modes install
 # all wheels together in a single pip call further below.
@@ -36,7 +44,7 @@ if [[ "${test_module}" == "pathfinder" ]]; then
     "LD:${CUDA_PATHFINDER_TEST_LOAD_NVIDIA_DYNAMIC_LIB_STRICTNESS} " \
     "FH:${CUDA_PATHFINDER_TEST_FIND_NVIDIA_HEADERS_STRICTNESS} " \
     "BC:${CUDA_PATHFINDER_TEST_FIND_NVIDIA_BITCODE_LIB_STRICTNESS}"
-  pytest -ra -s -v --durations=0 tests/ |& tee /tmp/pathfinder_test_log.txt
+  pytest -ra -s -v --durations=0 "${PYTEST_PARALLEL_ARGS[@]}" tests/ |& tee /tmp/pathfinder_test_log.txt
   # Report the number of "INFO test_" lines (including zero)
   # to support quick validations based on GHA log archives.
   line_count=$(awk '/^INFO test_/ {count++} END {print count+0}' /tmp/pathfinder_test_log.txt)
@@ -51,9 +59,9 @@ elif [[ "${test_module}" == "bindings" ]]; then
     pip install $(ls "${CUDA_BINDINGS_ARTIFACTS_DIR}"/*.whl)[all] --group test
   fi
   echo "Running bindings tests"
-  ${SANITIZER_CMD} pytest -rxXs -v --durations=0 --randomly-dont-reorganize tests/
+  ${SANITIZER_CMD} pytest -rxXs -v --durations=0 --randomly-dont-reorganize "${PYTEST_PARALLEL_ARGS[@]}" tests/
   if [[ "${SKIP_CYTHON_TEST}" == 0 ]]; then
-    ${SANITIZER_CMD} pytest -rxXs -v --durations=0 --randomly-dont-reorganize tests/cython
+    ${SANITIZER_CMD} pytest -rxXs -v --durations=0 --randomly-dont-reorganize "${PYTEST_PARALLEL_ARGS[@]}" tests/cython
   fi
   popd
 elif [[ "${test_module}" == "core" || "${test_module}" == nightly-* ]]; then
@@ -61,11 +69,6 @@ elif [[ "${test_module}" == "core" || "${test_module}" == nightly-* ]]; then
   TEST_CUDA_MAJOR="$(cut -d '.' -f 1 <<< ${CUDA_VER})"
   CUDA_VER_MINOR="$(cut -d '.' -f 1-2 <<< "${CUDA_VER}")"
 
-  FREE_THREADING=""
-  if python -c 'import sys; assert not sys._is_gil_enabled()' 2> /dev/null; then
-    FREE_THREADING+="-ft"
-  fi
-
   # Resolve bindings based on BINDINGS_SOURCE (set by env-vars):
   #   main/backport → local wheel from artifacts dir
   #   published     → install from PyPI by version
@@ -105,11 +108,11 @@ elif [[ "${test_module}" == "core" || "${test_module}" == nightly-* ]]; then
     echo "Installed packages before core tests:"
     pip list
     echo "Running core tests"
-    ${SANITIZER_CMD} pytest -rxXs -v --durations=0 --randomly-dont-reorganize tests/
+    ${SANITIZER_CMD} pytest -rxXs -v --durations=0 --randomly-dont-reorganize "${PYTEST_PARALLEL_ARGS[@]}" tests/
     # Currently our CI always installs the latest bindings (from either major version).
     # This is not compatible with the test requirements.
     if [[ "${SKIP_CYTHON_TEST}" == 0 ]]; then
-      ${SANITIZER_CMD} pytest -rxXs -v --durations=0 --randomly-dont-reorganize tests/cython
+      ${SANITIZER_CMD} pytest -rxXs -v --durations=0 --randomly-dont-reorganize "${PYTEST_PARALLEL_ARGS[@]}" tests/cython
     fi
     popd
   elif [[ "${test_module}" == "nightly-cuda-core" ]]; then
diff --git a/cuda_bindings/tests/test_cufile.py b/cuda_bindings/tests/test_cufile.py
index 46bd8429a62..7de4ceb2e20 100644
--- a/cuda_bindings/tests/test_cufile.py
+++ b/cuda_bindings/tests/test_cufile.py
@@ -138,7 +138,7 @@ def ctx():
     (err,) = cuda.cuCtxSetCurrent(ctx)
     assert err == cuda.CUresult.CUDA_SUCCESS
 
-    yield
+    yield ctx
 
     cuda.cuDevicePrimaryCtxRelease(device)
 
diff --git a/cuda_bindings/tests/test_examples.py b/cuda_bindings/tests/test_examples.py
index 63a56c78fb7..6ffce3e7fe7 100644
--- a/cuda_bindings/tests/test_examples.py
+++ b/cuda_bindings/tests/test_examples.py
@@ -20,6 +20,7 @@ def test_example(example):
 
     env = os.environ.copy()
     env["CUDA_BINDINGS_SKIP_EXAMPLE"] = "100"
+    env["MPLBACKEND"] = "Agg"  # avoid plt.show() from blocking
 
     process = subprocess.run([sys.executable, example], capture_output=True, env=env)  # noqa: S603
     # returncode is a special value used in the examples to indicate that system requirements are not met.
diff --git a/cuda_core/cuda/core/_device_resources.pxd b/cuda_core/cuda/core/_device_resources.pxd
index 98f91ab4733..d618c24cf10 100644
--- a/cuda_core/cuda/core/_device_resources.pxd
+++ b/cuda_core/cuda/core/_device_resources.pxd
@@ -2,8 +2,6 @@
 #
 # SPDX-License-Identifier: Apache-2.0
 
-cimport cython
-
 from cuda.bindings cimport cydriver
 from cuda.core._resource_handles cimport ContextHandle, GreenCtxHandle
 
@@ -17,7 +15,6 @@ cdef class SMResource:
         unsigned int _flags
         bint _is_usable
         object __weakref__
-        cython.pymutex _split_mutex
 
     @staticmethod
     cdef SMResource _from_dev_resource(cydriver.CUdevResource res, int device_id)
diff --git a/cuda_core/cuda/core/_device_resources.pyx b/cuda_core/cuda/core/_device_resources.pyx
index bafc462c936..ecd9e00bf05 100644
--- a/cuda_core/cuda/core/_device_resources.pyx
+++ b/cuda_core/cuda/core/_device_resources.pyx
@@ -498,11 +498,10 @@ cdef class SMResource:
         )
         _resolve_group_count(opts)
         _check_green_ctx_support()
-        with self._split_mutex:
-            if _can_use_structured_sm_split():
-                return _split_with_general_api(self, opts, dry_run)
-            # SplitByCount requires the same 12.4+ as green ctx support (already checked above)
-            return _split_with_count_api(self, opts, dry_run)
+        if _can_use_structured_sm_split():
+            return _split_with_general_api(self, opts, dry_run)
+        # SplitByCount requires the same 12.4+ as green ctx support (already checked above)
+        return _split_with_count_api(self, opts, dry_run)
 
 
 cdef class WorkqueueResource:
diff --git a/cuda_core/tests/conftest.py b/cuda_core/tests/conftest.py
index 7106b1e31f6..8e4bfb7ff4b 100644
--- a/cuda_core/tests/conftest.py
+++ b/cuda_core/tests/conftest.py
@@ -131,6 +131,18 @@ def wrapper(*args, **kwargs):
                 kwargs["mempool_device_x2"] = _mempool_device_impl(2)
             if "mempool_device_x3" in kwargs:
                 kwargs["mempool_device_x3"] = _mempool_device_impl(3)
+
+            # These are used by test_green_context.py.  The original fixtures include
+            # pytest.skip() but that should have correctly fired by this time.
+            if "sm_resource" in kwargs:
+                kwargs["sm_resource"] = device.resources.sm
+            if "wq_resource" in kwargs:
+                kwargs["wq_resource"] = device.resources.workqueue
+            if "green_ctx" in kwargs:
+                from cuda.core import ContextOptions, SMResourceOptions
+
+                groups, _ = device.resources.sm.split(SMResourceOptions(count=None))
+                kwargs["green_ctx"] = device.create_context(ContextOptions(resources=[groups[0]]))
             return func(*args, **kwargs)
 
     wrapper._cuda_core_worker_cuda_wrapped = True
diff --git a/cuda_core/tests/graph/test_graph_definition.py b/cuda_core/tests/graph/test_graph_definition.py
index da78bea577f..b18b2ac9bda 100644
--- a/cuda_core/tests/graph/test_graph_definition.py
+++ b/cuda_core/tests/graph/test_graph_definition.py
@@ -575,20 +575,6 @@ def node_spec(request, init_cuda):
 # =============================================================================
 
 
-@pytest.fixture
-def sample_graphdef(init_cuda):
-    """A sample GraphDefinition for standalone tests."""
-    return GraphDefinition()
-
-
-@pytest.fixture
-def dot_file(tmp_path):
-    """Temporary DOT file path, cleaned up after test."""
-    path = tmp_path / "graph.dot"
-    yield path
-    path.unlink(missing_ok=True)
-
-
 # =============================================================================
 # Topology tests (parameterized over graph specs)
 # =============================================================================
@@ -775,14 +761,16 @@ def registered(node):
 # =============================================================================
 
 
-def test_graphdef_handle_valid(sample_graphdef):
+def test_graphdef_handle_valid(init_cuda):
     """GraphDefinition has a valid non-null handle."""
+    sample_graphdef = GraphDefinition()
     assert sample_graphdef.handle is not None
     assert int(sample_graphdef.handle) != 0
 
 
-def test_graphdef_entry_is_virtual(sample_graphdef):
+def test_graphdef_entry_is_virtual(init_cuda):
     """Internal entry node is virtual (no pred/succ, type is None)."""
+    sample_graphdef = GraphDefinition()
     entry = sample_graphdef._entry
     assert isinstance(entry, GraphNode)
     assert entry.pred == set()
@@ -795,8 +783,9 @@ def test_graphdef_entry_is_virtual(sample_graphdef):
 # =============================================================================
 
 
-def test_alloc_zero_size_fails(sample_graphdef):
+def test_alloc_zero_size_fails(init_cuda):
     """Alloc with zero size raises error (CUDA limitation)."""
+    sample_graphdef = GraphDefinition()
     _skip_if_no_mempool()
     from cuda.core._utils.cuda_utils import CUDAError
 
@@ -804,8 +793,9 @@ def test_alloc_zero_size_fails(sample_graphdef):
         sample_graphdef.allocate(0)
 
 
-def test_free_creates_dependency(sample_graphdef):
+def test_free_creates_dependency(init_cuda):
     """Free node depends on its predecessor."""
+    sample_graphdef = GraphDefinition()
     _skip_if_no_mempool()
     with xfail_on_graph_mempool_oom():
         alloc = sample_graphdef.allocate(ALLOC_SIZE)
@@ -813,8 +803,9 @@ def test_free_creates_dependency(sample_graphdef):
     assert alloc in free.pred
 
 
-def test_alloc_free_chain(sample_graphdef):
+def test_alloc_free_chain(init_cuda):
     """Alloc and free can be chained."""
+    sample_graphdef = GraphDefinition()
     _skip_if_no_mempool()
     with xfail_on_graph_mempool_oom():
         a1 = sample_graphdef.allocate(ALLOC_SIZE)
@@ -831,8 +822,9 @@ def test_alloc_free_chain(sample_graphdef):
 # =============================================================================
 
 
-def test_alloc_memory_type_invalid(sample_graphdef):
+def test_alloc_memory_type_invalid(init_cuda):
     """Invalid memory type raises ValueError."""
+    sample_graphdef = GraphDefinition()
     with pytest.raises(ValueError, match="Invalid memory_type"):
         sample_graphdef.allocate(ALLOC_SIZE, memory_type="invalid")
 
@@ -844,8 +836,9 @@ def test_alloc_memory_type_invalid(sample_graphdef):
         pytest.param(lambda d: d, id="Device_object"),
     ],
 )
-def test_alloc_device_option(sample_graphdef, device_spec):
+def test_alloc_device_option(init_cuda, device_spec):
     """Device can be specified as int or Device object."""
+    sample_graphdef = GraphDefinition()
     _skip_if_no_mempool()
     device = Device()
     with xfail_on_graph_mempool_oom(device):
@@ -868,8 +861,9 @@ def test_alloc_peer_access(mempool_device_x2):
 
 
 @pytest.mark.parametrize("num_branches", [2, 3, 5])
-def test_join_merges_branches(sample_graphdef, num_branches):
+def test_join_merges_branches(init_cuda, num_branches):
     """join() with multiple branches creates correct dependencies."""
+    sample_graphdef = GraphDefinition()
     _skip_if_no_mempool()
     with xfail_on_graph_mempool_oom():
         branches = [sample_graphdef.allocate(ALLOC_SIZE) for _ in range(num_branches)]
@@ -883,8 +877,9 @@ def test_join_merges_branches(sample_graphdef, num_branches):
 # =============================================================================
 
 
-def test_launch_creates_node(sample_graphdef):
+def test_launch_creates_node(init_cuda):
     """launch() creates a KernelNode."""
+    sample_graphdef = GraphDefinition()
     mod = compile_common_kernels()
     kernel = mod.get_kernel("empty_kernel")
     config = LaunchConfig(grid=1, block=1)
@@ -892,8 +887,9 @@ def test_launch_creates_node(sample_graphdef):
     assert isinstance(node, KernelNode)
 
 
-def test_launch_chain_dependencies(sample_graphdef):
+def test_launch_chain_dependencies(init_cuda):
     """Chained launches create correct dependencies."""
+    sample_graphdef = GraphDefinition()
     mod = compile_common_kernels()
     kernel = mod.get_kernel("empty_kernel")
     config = LaunchConfig(grid=1, block=1)
@@ -955,15 +951,17 @@ def _instantiate_and_upload(graph_definition, kwargs, stream):
 
 
 @pytest.mark.parametrize("inst_kwargs", _INSTANTIATE_ONLY_OPTIONS)
-def test_instantiate_empty_graph(sample_graphdef, inst_kwargs):
+def test_instantiate_empty_graph(init_cuda, inst_kwargs):
     """Empty graph can be instantiated."""
+    sample_graphdef = GraphDefinition()
     graph = _instantiate(sample_graphdef, inst_kwargs)
     assert graph is not None
 
 
 @pytest.mark.parametrize("inst_kwargs", _INSTANTIATE_ONLY_OPTIONS)
-def test_instantiate_with_nodes(sample_graphdef, inst_kwargs):
+def test_instantiate_with_nodes(init_cuda, inst_kwargs):
     """Graph with nodes can be instantiated."""
+    sample_graphdef = GraphDefinition()
     _skip_if_no_mempool()
     with xfail_on_graph_mempool_oom():
         sample_graphdef.allocate(ALLOC_SIZE)
@@ -973,8 +971,9 @@ def test_instantiate_with_nodes(sample_graphdef, inst_kwargs):
 
 
 @pytest.mark.skipif(not Device(0).properties.unified_addressing, reason="requires unified addressing")
-def test_instantiate_and_execute_kernel_device_launch(sample_graphdef):
+def test_instantiate_and_execute_kernel_device_launch(init_cuda):
     """Kernel-only graph can be instantiated with device_launch flag."""
+    sample_graphdef = GraphDefinition()
     mod = compile_common_kernels()
     kernel = mod.get_kernel("empty_kernel")
     config = LaunchConfig(grid=1, block=1)
@@ -990,8 +989,9 @@ def test_instantiate_and_execute_kernel_device_launch(sample_graphdef):
 
 
 @pytest.mark.parametrize("inst_kwargs", _EXECUTE_OPTIONS)
-def test_instantiate_and_execute_kernel(sample_graphdef, inst_kwargs):
+def test_instantiate_and_execute_kernel(init_cuda, inst_kwargs):
     """Graph with kernel can be instantiated and executed."""
+    sample_graphdef = GraphDefinition()
     mod = compile_common_kernels()
     kernel = mod.get_kernel("empty_kernel")
     config = LaunchConfig(grid=1, block=1)
@@ -1004,8 +1004,9 @@ def test_instantiate_and_execute_kernel(sample_graphdef, inst_kwargs):
 
 
 @pytest.mark.parametrize("inst_kwargs", _EXECUTE_OPTIONS)
-def test_instantiate_and_execute_alloc_free(sample_graphdef, inst_kwargs):
+def test_instantiate_and_execute_alloc_free(init_cuda, inst_kwargs):
     """Graph with alloc/free can be executed."""
+    sample_graphdef = GraphDefinition()
     _skip_if_no_mempool()
     with xfail_on_graph_mempool_oom():
         alloc = sample_graphdef.allocate(ALLOC_SIZE)
@@ -1018,8 +1019,9 @@ def test_instantiate_and_execute_alloc_free(sample_graphdef, inst_kwargs):
 
 
 @pytest.mark.parametrize("inst_kwargs", _EXECUTE_OPTIONS)
-def test_instantiate_and_execute_memset(sample_graphdef, inst_kwargs):
+def test_instantiate_and_execute_memset(init_cuda, inst_kwargs):
     """Graph with alloc/memset/free can be executed."""
+    sample_graphdef = GraphDefinition()
     _skip_if_no_mempool()
     with xfail_on_graph_mempool_oom():
         alloc = sample_graphdef.allocate(ALLOC_SIZE)
@@ -1033,8 +1035,9 @@ def test_instantiate_and_execute_memset(sample_graphdef, inst_kwargs):
 
 
 @pytest.mark.parametrize("inst_kwargs", _EXECUTE_OPTIONS)
-def test_instantiate_and_execute_memcpy(sample_graphdef, inst_kwargs):
+def test_instantiate_and_execute_memcpy(init_cuda, inst_kwargs):
     """Graph with alloc/memset/memcpy/free can be executed and data is copied."""
+    sample_graphdef = GraphDefinition()
     _skip_if_no_mempool()
     import ctypes
 
@@ -1058,8 +1061,9 @@ def test_instantiate_and_execute_memcpy(sample_graphdef, inst_kwargs):
     assert all(b == 0xAB for b in host_buf)
 
 
-def test_instantiate_and_execute_child_graph(sample_graphdef):
+def test_instantiate_and_execute_child_graph(init_cuda):
     """Graph with embedded child graph can be executed."""
+    sample_graphdef = GraphDefinition()
     child = GraphDefinition()
     mod = compile_common_kernels()
     kernel = mod.get_kernel("empty_kernel")
@@ -1075,8 +1079,9 @@ def test_instantiate_and_execute_child_graph(sample_graphdef):
     stream.sync()
 
 
-def test_instantiate_and_execute_host_callback(sample_graphdef):
+def test_instantiate_and_execute_host_callback(init_cuda):
     """Graph with host callback can be executed and callback is invoked."""
+    sample_graphdef = GraphDefinition()
     results = []
 
     def my_callback():
@@ -1093,8 +1098,9 @@ def my_callback():
     assert results == [42]
 
 
-def test_instantiate_and_execute_host_callback_cfunc(sample_graphdef):
+def test_instantiate_and_execute_host_callback_cfunc(init_cuda):
     """Graph with ctypes function pointer callback can be executed."""
+    sample_graphdef = GraphDefinition()
     import ctypes
 
     CALLBACK = ctypes.CFUNCTYPE(None, ctypes.c_void_p)
@@ -1115,8 +1121,9 @@ def raw_fn(data):
     assert called[0]
 
 
-def test_host_callback_cfunc_with_user_data(sample_graphdef):
+def test_host_callback_cfunc_with_user_data(init_cuda):
     """Host callback with bytes user_data passes data to C function."""
+    sample_graphdef = GraphDefinition()
     import ctypes
 
     CALLBACK = ctypes.CFUNCTYPE(None, ctypes.c_void_p)
@@ -1137,14 +1144,16 @@ def read_byte(data):
     assert result[0] == 0xAB
 
 
-def test_host_callback_user_data_rejected_for_python_callable(sample_graphdef):
+def test_host_callback_user_data_rejected_for_python_callable(init_cuda):
     """user_data is rejected for Python callables."""
+    sample_graphdef = GraphDefinition()
     with pytest.raises(ValueError, match="user_data is only supported"):
         sample_graphdef.callback(lambda: None, user_data=b"hello")
 
 
-def test_instantiate_and_execute_event_record_wait(sample_graphdef):
+def test_instantiate_and_execute_event_record_wait(init_cuda):
     """Graph with event record and wait nodes can be executed."""
+    sample_graphdef = GraphDefinition()
     event = Device().create_event()
     rec = sample_graphdef.record(event)
     rec.wait(event)
@@ -1166,8 +1175,9 @@ def _skip_unless_cc_90():
         pytest.skip("Conditional node execution requires CC >= 9.0 (Hopper)")
 
 
-def test_instantiate_and_execute_if_then(sample_graphdef):
+def test_instantiate_and_execute_if_then(init_cuda):
     """If-conditional node: body executes only when condition is non-zero."""
+    sample_graphdef = GraphDefinition()
     _skip_unless_cc_90()
     _skip_if_no_mempool()
     import ctypes
@@ -1199,8 +1209,9 @@ def test_instantiate_and_execute_if_then(sample_graphdef):
     assert result[0] == 1
 
 
-def test_instantiate_and_execute_if_else(sample_graphdef):
+def test_instantiate_and_execute_if_else(init_cuda):
     """If-else node: then or else branch executes based on condition."""
+    sample_graphdef = GraphDefinition()
     _skip_unless_cc_90()
     _skip_if_no_mempool()
     import ctypes
@@ -1234,8 +1245,9 @@ def test_instantiate_and_execute_if_else(sample_graphdef):
     assert result[0] == 2
 
 
-def test_instantiate_and_execute_switch(sample_graphdef):
+def test_instantiate_and_execute_switch(init_cuda):
     """Switch node: selected branch executes based on condition value."""
+    sample_graphdef = GraphDefinition()
     _skip_unless_cc_90()
     _skip_if_no_mempool()
     import ctypes
@@ -1268,8 +1280,9 @@ def test_instantiate_and_execute_switch(sample_graphdef):
     assert result[0] == 1
 
 
-def test_conditional_node_type_preserved_by_nodes(sample_graphdef):
+def test_conditional_node_type_preserved_by_nodes(init_cuda):
     """Conditional nodes appear as ConditionalNode base when read back from graph."""
+    sample_graphdef = GraphDefinition()
     condition = try_create_condition(sample_graphdef)
     if_node = sample_graphdef.if_then(condition)
     assert isinstance(if_node, IfNode)
@@ -1285,8 +1298,10 @@ def test_conditional_node_type_preserved_by_nodes(sample_graphdef):
 # =============================================================================
 
 
-def test_debug_dot_print_creates_file(sample_graphdef, dot_file):
+def test_debug_dot_print_creates_file(init_cuda, tmp_path):
     """debug_dot_print writes a DOT file."""
+    sample_graphdef = GraphDefinition()
+    dot_file = tmp_path / "graph.dot"
     _skip_if_no_mempool()
     with xfail_on_graph_mempool_oom():
         sample_graphdef.allocate(ALLOC_SIZE)
@@ -1296,8 +1311,10 @@ def test_debug_dot_print_creates_file(sample_graphdef, dot_file):
     assert "digraph" in content
 
 
-def test_debug_dot_print_with_options(sample_graphdef, dot_file):
+def test_debug_dot_print_with_options(init_cuda, tmp_path):
     """debug_dot_print accepts GraphDebugPrintOptions."""
+    sample_graphdef = GraphDefinition()
+    dot_file = tmp_path / "graph.dot"
     _skip_if_no_mempool()
     with xfail_on_graph_mempool_oom():
         sample_graphdef.allocate(ALLOC_SIZE)
@@ -1306,8 +1323,10 @@ def test_debug_dot_print_with_options(sample_graphdef, dot_file):
     assert dot_file.exists()
 
 
-def test_debug_dot_print_invalid_options(sample_graphdef, dot_file):
+def test_debug_dot_print_invalid_options(init_cuda, tmp_path):
     """debug_dot_print rejects invalid options type."""
+    sample_graphdef = GraphDefinition()
+    dot_file = tmp_path / "graph.dot"
     _skip_if_no_mempool()
     with xfail_on_graph_mempool_oom():
         sample_graphdef.allocate(ALLOC_SIZE)
diff --git a/cuda_core/tests/graph/test_graph_memory_resource.py b/cuda_core/tests/graph/test_graph_memory_resource.py
index 9fc794f4cca..482e8fe1c57 100644
--- a/cuda_core/tests/graph/test_graph_memory_resource.py
+++ b/cuda_core/tests/graph/test_graph_memory_resource.py
@@ -21,6 +21,13 @@
 from cuda.core._utils.cuda_utils import CUDAError
 from cuda.core.graph import GraphCompleteOptions
 
+# NOTE(seberg): "global" mode seems thread-unsafe even when working on stream
+_GRAPH_MODES = [
+    pytest.param("global", marks=pytest.mark.thread_unsafe(reason="gb instances share stream unsafely")),
+    "thread_local",
+    "relaxed",
+]
+
 
 def _common_kernels_alloc():
     code = """
@@ -80,7 +87,7 @@ def free(self, buffers):
         self.stream.sync()
 
 
-@pytest.mark.parametrize("mode", ["no_graph", "global", "thread_local", "relaxed"])
+@pytest.mark.parametrize("mode", ["no_graph"] + _GRAPH_MODES)
 @pytest.mark.parametrize("action", ["incr", "fill"])
 def test_graph_alloc(mempool_device, mode, action):
     """Test basic graph capture with memory allocated and deallocated by
@@ -130,7 +137,7 @@ def apply_kernels(mr, stream, out):
         assert compare_buffer_to_constant(out, 3)
     else:
         # Capture work, then upload and launch.
-        gb = device.create_graph_builder().begin_building(mode)
+        gb = stream.create_graph_builder().begin_building(mode)
         with xfail_on_graph_mempool_oom(device):
             apply_kernels(mr=gmr, stream=gb, out=out)
             graph = gb.end_building().complete()
@@ -150,7 +157,7 @@ def apply_kernels(mr, stream, out):
 
 
 @pytest.mark.skipif(IS_WINDOWS or IS_WSL, reason="auto_free_on_launch not supported on Windows")
-@pytest.mark.parametrize("mode", ["global", "thread_local", "relaxed"])
+@pytest.mark.parametrize("mode", _GRAPH_MODES)
 def test_graph_alloc_with_output(mempool_device, mode):
     """Test for memory allocated in a graph being used outside the graph."""
     NBYTES = 64
@@ -168,7 +175,7 @@ def test_graph_alloc_with_output(mempool_device, mode):
     # Construct a graph to copy and increment the input. It returns a new
     # buffer allocated within the graph.  The auto_free_on_launch option
     # is required to properly use the output buffer.
-    gb = device.create_graph_builder().begin_building(mode)
+    gb = stream.create_graph_builder().begin_building(mode)
     with xfail_on_graph_mempool_oom(device):
         out = gmr.allocate(NBYTES, stream=gb)
         out.copy_from(in_, stream=gb)
@@ -195,7 +202,8 @@ def test_graph_alloc_with_output(mempool_device, mode):
     assert compare_buffer_to_constant(out, 6)
 
 
-@pytest.mark.parametrize("mode", ["global", "thread_local", "relaxed"])
+@pytest.mark.parametrize("mode", _GRAPH_MODES)
+@pytest.mark.thread_unsafe(reason="gb instances share default stream")
 def test_graph_mem_alloc_zero(mempool_device, mode):
     device = mempool_device
     gb = device.create_graph_builder().begin_building(mode)
@@ -213,7 +221,8 @@ def test_graph_mem_alloc_zero(mempool_device, mode):
     assert buffer.device_id == int(device)
 
 
-@pytest.mark.parametrize("mode", ["global", "thread_local", "relaxed"])
+@pytest.mark.parametrize("mode", _GRAPH_MODES)
+@pytest.mark.thread_unsafe(reason="GMR is shared, so high mark is global")
 def test_graph_mem_set_attributes(mempool_device, mode):
     device = mempool_device
     stream = device.create_stream()
@@ -265,7 +274,7 @@ def test_graph_mem_set_attributes(mempool_device, mode):
     mman.reset()
 
 
-@pytest.mark.parametrize("mode", ["global", "thread_local", "relaxed"])
+@pytest.mark.parametrize("mode", _GRAPH_MODES)
 def test_gmr_check_capture_state(mempool_device, mode):
     """
     Test expected errors (and non-errors) using GraphMemoryResource with graph
@@ -284,7 +293,7 @@ def test_gmr_check_capture_state(mempool_device, mode):
         gmr.allocate(1, stream=stream)
 
     # Capturing
-    gb = device.create_graph_builder().begin_building(mode=mode)
+    gb = stream.create_graph_builder().begin_building(mode=mode)
     with xfail_on_graph_mempool_oom(device):
         gmr.allocate(1, stream=gb)  # no error
         gb.end_building().complete()
@@ -320,7 +329,7 @@ def test_graph_memory_resource_attributes_repr(mempool_device):
     assert "used_mem_high=" in r
 
 
-@pytest.mark.parametrize("mode", ["global", "thread_local", "relaxed"])
+@pytest.mark.parametrize("mode", _GRAPH_MODES)
 def test_dmr_check_capture_state(mempool_device, mode):
     """
     Test expected errors (and non-errors) using DeviceMemoryResource with graph
@@ -334,7 +343,7 @@ def test_dmr_check_capture_state(mempool_device, mode):
     dmr.allocate(1, stream=stream).close()  # no error
 
     # Capturing
-    gb = device.create_graph_builder().begin_building(mode=mode)
+    gb = stream.create_graph_builder().begin_building(mode=mode)
     with pytest.raises(
         RuntimeError,
         match=r"cannot perform memory operations on a capturing "
diff --git a/cuda_core/tests/helpers/latch.py b/cuda_core/tests/helpers/latch.py
index c28fb222641..978e2dbdf18 100644
--- a/cuda_core/tests/helpers/latch.py
+++ b/cuda_core/tests/helpers/latch.py
@@ -1,7 +1,8 @@
-# SPDX-FileCopyrightText: Copyright (c) 2025 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+# SPDX-FileCopyrightText: Copyright (c) 2025-2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
 # SPDX-License-Identifier: Apache-2.0
 
 import ctypes
+import threading
 
 import pytest
 
@@ -20,9 +21,15 @@ class LatchKernel:
     Manages a kernel that blocks stream progress until released.
     """
 
-    def __init__(self, device, timeout_sec=60):
-        if helpers.CUDA_INCLUDE_PATH is None:
-            pytest.skip("need CUDA header")
+    _latch_kernel_lock = threading.Lock()
+    _latch_kernels = {}
+
+    @classmethod
+    def _get_kernel(cls, device):
+        kernel = cls._latch_kernels.get(device.uuid)
+        if kernel is not None:
+            return kernel
+
         code = """
                #include <cuda/atomic>
 
@@ -41,6 +48,7 @@ def __init__(self, device, timeout_sec=60):
 
                        // Check for timeout
                        if (clock64() - start >= timeout_cycles) {
+                           signal.store(-1, cuda::memory_order_relaxed);
                            break;  // Timeout reached
                        }
 
@@ -56,14 +64,25 @@ def __init__(self, device, timeout_sec=60):
         )
         prog = Program(code, code_type="c++", options=program_options)
         mod = prog.compile(target_type="cubin")
-        self.kernel = mod.get_kernel("latch")
+        kernel = mod.get_kernel("latch")
+
+        return cls._latch_kernels.setdefault(device.uuid, kernel)
+
+    def __init__(self, device, timeout_sec=60):
+        if helpers.CUDA_INCLUDE_PATH is None:
+            pytest.skip("need CUDA header")
+
+        with self._latch_kernel_lock:
+            self.kernel = self._get_kernel(device)
 
         mr = LegacyPinnedMemoryResource()
         self.buffer = mr.allocate(4)
-        self.busy_wait_flag[0] = 0
+        self.busy_wait_flag[0] = 1
         clock_rate_hz = device.properties.clock_rate * 1000
         self.timeout_cycles = int(timeout_sec * clock_rate_hz)
 
+        self.busy_wait_flag[0] = 0
+
     def launch(self, stream):
         """Launch the latch kernel, blocking stream progress via busy waiting."""
         config = LaunchConfig(grid=1, block=1)
diff --git a/cuda_core/tests/memory/test_managed_ops.py b/cuda_core/tests/memory/test_managed_ops.py
index 33def77935f..d9305399de4 100644
--- a/cuda_core/tests/memory/test_managed_ops.py
+++ b/cuda_core/tests/memory/test_managed_ops.py
@@ -364,6 +364,7 @@ def test_from_handle(self, init_cuda):
         finally:
             plain.close()
 
+    @pytest.mark.thread_unsafe(reason="external_managed_buffer is shared between threads")
     def test_read_mostly_roundtrip(self, external_managed_buffer):
         buf = external_managed_buffer
         assert buf.read_mostly is False
@@ -372,6 +373,7 @@ def test_read_mostly_roundtrip(self, external_managed_buffer):
         buf.read_mostly = False
         assert buf.read_mostly is False
 
+    @pytest.mark.thread_unsafe(reason="external_managed_buffer is shared between threads")
     def test_preferred_location_roundtrip(self, location_ops_device, external_managed_buffer):
         device = location_ops_device
         buf = external_managed_buffer
@@ -386,6 +388,7 @@ def test_preferred_location_roundtrip(self, location_ops_device, external_manage
         buf.preferred_location = None
         assert buf.preferred_location is None
 
+    @pytest.mark.thread_unsafe(reason="external_managed_buffer is shared between threads")
     def test_preferred_location_roundtrip_host_numa(self, location_ops_device):
         """Host(numa_id=N) round-trips correctly on CUDA 13 builds."""
         from cuda.core._utils.version import binding_version
@@ -406,6 +409,7 @@ def test_preferred_location_roundtrip_host_numa(self, location_ops_device):
         finally:
             plain.close()
 
+    @pytest.mark.thread_unsafe(reason="external_managed_buffer is shared between threads")
     def test_accessed_by_add_discard(self, location_ops_device, external_managed_buffer):
         device = location_ops_device
         buf = external_managed_buffer
@@ -417,6 +421,7 @@ def test_accessed_by_add_discard(self, location_ops_device, external_managed_buf
         buf.accessed_by.discard(device)
         assert device not in buf.accessed_by
 
+    @pytest.mark.thread_unsafe(reason="external_managed_buffer is shared between threads")
     def test_accessed_by_mutable_set_interface(self, location_ops_device, external_managed_buffer):
         """Full MutableSet conformance pass on AccessedBySetProxy.
 
@@ -436,6 +441,7 @@ def test_accessed_by_mutable_set_interface(self, location_ops_device, external_m
             non_member=Host(numa_id=0),
         )
 
+    @pytest.mark.thread_unsafe(reason="external_managed_buffer is shared between threads")
     def test_accessed_by_set_assignment(self, location_ops_device, external_managed_buffer):
         device = location_ops_device
         buf = external_managed_buffer
diff --git a/cuda_core/tests/system/test_system_device.py b/cuda_core/tests/system/test_system_device.py
index b5fe8cccbfa..83ff4b58a87 100644
--- a/cuda_core/tests/system/test_system_device.py
+++ b/cuda_core/tests/system/test_system_device.py
@@ -349,6 +349,7 @@ def test_c2c_mode_enabled():
 
 
 @pytest.mark.skipif(helpers.IS_WSL or helpers.IS_WINDOWS, reason="Persistence mode not supported on WSL or Windows")
+@pytest.mark.thread_unsafe(reason="device persistence mode is global state")
 def test_persistence_mode_enabled():
     for device in system.Device.get_all_devices():
         is_enabled = device.is_persistence_mode_enabled
diff --git a/cuda_core/tests/test_event.py b/cuda_core/tests/test_event.py
index 79f0090ace6..c0eb4654752 100644
--- a/cuda_core/tests/test_event.py
+++ b/cuda_core/tests/test_event.py
@@ -119,6 +119,7 @@ def test_error_timing_recorded():
 
 
 @pytest.mark.skipif(Device().compute_capability.major < 7, reason="__nanosleep is only available starting Volta (sm70)")
+@pytest.mark.parallel_threads_limit(8)  # Very many threads may cause latch to time out
 def test_error_timing_incomplete():
     device = Device()
     device.set_current()
diff --git a/cuda_core/tests/test_green_context.py b/cuda_core/tests/test_green_context.py
index 4078b166b7b..b525bc2a9ed 100644
--- a/cuda_core/tests/test_green_context.py
+++ b/cuda_core/tests/test_green_context.py
@@ -40,6 +40,8 @@
 # ---------------------------------------------------------------------------
 # Fixtures
 # ---------------------------------------------------------------------------
+# Note that the following fixtures (except fill_kernel) require per-thread setup
+# and are currently special cased to work with pytest-run-parallel in conftest.
 
 
 @pytest.fixture
diff --git a/cuda_core/tests/test_memory.py b/cuda_core/tests/test_memory.py
index f53e8244328..1c90d75fd75 100644
--- a/cuda_core/tests/test_memory.py
+++ b/cuda_core/tests/test_memory.py
@@ -251,9 +251,8 @@ def _pattern_bytes(value) -> bytes:
 
 
 @pytest.fixture(params=["device", "unified", "pinned"])
-def fill_env(request):
-    device = Device()
-    device.set_current()
+def fill_env(request, init_cuda):
+    device = init_cuda
     if request.param == "device":
         mr = DummyDeviceMemoryResource(device)
     elif request.param == "unified":
diff --git a/cuda_core/tests/test_module.py b/cuda_core/tests/test_module.py
index a0600966619..3ea18202ce6 100644
--- a/cuda_core/tests/test_module.py
+++ b/cuda_core/tests/test_module.py
@@ -5,6 +5,7 @@
 import os
 import pickle
 import subprocess
+import types
 import warnings
 from pathlib import Path
 
@@ -177,20 +178,20 @@ def get_saxpy_fatbin(init_cuda):
     return bytes(fatbin), sym_map
 
 
-def _read_saxpy_rdc(kind: str) -> bytes:
+@pytest.fixture(params=["object", "library"], ids=["object", "library"])
+def saxpy_rdc(request):
     """Read a pre-built saxpy RDC object or library.
 
     In CI: produced by the build stage.
     In local dev: auto-built on demand if nvcc is available; if you edit
     saxpy.cu, remove stale RDC files (i.e. saxpy.o, saxpy.a, or saxpy.lib) to force a rebuild.
     """
+    kind = request.param
+
     binaries_dir = Path(__file__).parent / "test_binaries"
-    if kind == "object":
-        rdc_path = binaries_dir / "saxpy.o"
-    elif kind == "library":
-        rdc_path = binaries_dir / ("saxpy.lib" if os.name == "nt" else "saxpy.a")
-    else:
-        raise ValueError(f"unknown saxpy RDC kind: {kind!r}")
+    suffix = ".o" if kind == "object" else ".lib" if os.name == "nt" else ".a"
+
+    rdc_path = binaries_dir / f"saxpy{suffix}"
 
     if not rdc_path.is_file():
         nvcc_path = find_nvidia_binary_utility("nvcc")
@@ -206,7 +207,13 @@ def _read_saxpy_rdc(kind: str) -> bytes:
             check=True,
             env=env,
         )
-    return rdc_path.read_bytes()
+
+    return types.SimpleNamespace(
+        kind=kind,
+        data=rdc_path.read_bytes(),
+        from_fn=ObjectCode.from_object if kind == "object" else ObjectCode.from_library,
+        suffix=suffix,
+    )
 
 
 def test_get_kernel(init_cuda):
@@ -367,47 +374,26 @@ def test_object_code_load_fatbin_from_file(get_saxpy_fatbin, tmp_path, convert_p
     mod_obj.get_kernel("saxpy<double>")  # force loading
 
 
-@pytest.mark.parametrize(
-    ("kind", "from_fn"),
-    [
-        ("object", ObjectCode.from_object),
-        ("library", ObjectCode.from_library),
-    ],
-)
-def test_object_code_load_rdc(kind, from_fn):
-    data = _read_saxpy_rdc(kind)
+def test_object_code_load_rdc(saxpy_rdc):
+    data = saxpy_rdc.data
     assert isinstance(data, bytes)
-    mod_obj = from_fn(data)
+    mod_obj = saxpy_rdc.from_fn(data)
     assert mod_obj.code == data
-    assert mod_obj.code_type == kind
-    with pytest.raises(RuntimeError, match=rf'Unsupported code type "{kind}"'):
+    assert mod_obj.code_type == saxpy_rdc.kind
+    with pytest.raises(RuntimeError, match=rf'Unsupported code type "{saxpy_rdc.kind}"'):
         mod_obj.get_kernel("saxpy<float>")
 
 
-@pytest.mark.parametrize(
-    ("kind", "from_fn", "suffix"),
-    [
-        ("object", ObjectCode.from_object, ".o"),
-        ("library", ObjectCode.from_library, ".lib" if os.name == "nt" else ".a"),
-    ],
-)
-def test_object_code_load_rdc_from_file(kind, from_fn, suffix, tmp_path):
-    rdc_file = tmp_path / f"test{suffix}"
-    rdc_file.write_bytes(_read_saxpy_rdc(kind))
+def test_object_code_load_rdc_from_file(saxpy_rdc, tmp_path):
+    rdc_file = tmp_path / f"test{saxpy_rdc.suffix}"
+    rdc_file.write_bytes(saxpy_rdc.data)
     arg = str(rdc_file)
-    mod_obj = from_fn(arg)
+    mod_obj = saxpy_rdc.from_fn(arg)
     assert mod_obj.code == arg
-    assert mod_obj.code_type == kind
+    assert mod_obj.code_type == saxpy_rdc.kind
 
 
-@pytest.mark.parametrize(
-    ("kind", "from_fn"),
-    [
-        ("object", ObjectCode.from_object),
-        ("library", ObjectCode.from_library),
-    ],
-)
-def test_object_code_load_rdc_with_linker(kind, from_fn, init_cuda):
+def test_object_code_load_rdc_with_linker(init_cuda, saxpy_rdc):
     arch = f"sm_{init_cuda.arch}"
     kernel_code = Program(
         r"""
@@ -421,7 +407,7 @@ def test_object_code_load_rdc_with_linker(kind, from_fn, init_cuda):
     ).compile("cubin")
     linked = Linker(
         kernel_code,
-        from_fn(_read_saxpy_rdc(kind)),
+        saxpy_rdc.from_fn(saxpy_rdc.data),
         options=LinkerOptions(arch=arch),
     ).link("cubin")
     kernel = linked.get_kernel("linked_kernel")
diff --git a/cuda_core/tests/test_multiprocessing_warning.py b/cuda_core/tests/test_multiprocessing_warning.py
index 0f96e0abfbc..94a671ff2f8 100644
--- a/cuda_core/tests/test_multiprocessing_warning.py
+++ b/cuda_core/tests/test_multiprocessing_warning.py
@@ -12,12 +12,17 @@
 import warnings
 from unittest.mock import patch
 
+import pytest
+
 from cuda.core import DeviceMemoryResource, DeviceMemoryResourceOptions, EventOptions
 from cuda.core._event import _reduce_event
 from cuda.core._memory._device_memory_resource import _deep_reduce_device_memory_resource
 from cuda.core._memory._ipc import _reduce_allocation_handle
 from cuda.core._utils.cuda_utils import check_multiprocessing_start_method, reset_fork_warning
 
+# We could move these to a (session) fixtures
+pytestmark = pytest.mark.thread_unsafe(reason="all tests use unittest.mock.patch")
+
 
 def test_warn_on_fork_method_device_memory_resource(ipc_device):
     """Test that warning is emitted when DeviceMemoryResource is pickled with fork method."""
diff --git a/cuda_core/tests/test_object_protocols.py b/cuda_core/tests/test_object_protocols.py
index e8391c75678..e4e8ee21c9b 100644
--- a/cuda_core/tests/test_object_protocols.py
+++ b/cuda_core/tests/test_object_protocols.py
@@ -527,6 +527,26 @@ def sample_switch_node_alt(sample_graphdef):
     return sample_graphdef.switch(condition, 3)
 
 
+# Indirect-parametrize helpers: request.getfixturevalue() runs here, in the
+# fixture (main thread), so the resolved object is already available when the
+# test function runs in a worker thread.
+
+
+@pytest.fixture
+def sample_object(request):
+    return request.getfixturevalue(request.param)
+
+
+@pytest.fixture
+def sample_object_a(request):
+    return request.getfixturevalue(request.param)
+
+
+@pytest.fixture
+def sample_object_b(request):
+    return request.getfixturevalue(request.param)
+
+
 # =============================================================================
 # Type groupings
 # =============================================================================
@@ -722,12 +742,11 @@ def sample_switch_node_alt(sample_graphdef):
 # =============================================================================
 
 
-@pytest.mark.parametrize("fixture_name", WEAKREF_TYPES)
-def test_weakref_supported(fixture_name, request):
+@pytest.mark.parametrize("sample_object", WEAKREF_TYPES, indirect=True)
+def test_weakref_supported(sample_object):
     """Object supports weak references."""
-    obj = request.getfixturevalue(fixture_name)
-    ref = weakref.ref(obj)
-    assert ref() is obj
+    ref = weakref.ref(sample_object)
+    assert ref() is sample_object
 
 
 # =============================================================================
@@ -735,27 +754,22 @@ def test_weakref_supported(fixture_name, request):
 # =============================================================================
 
 
-@pytest.mark.parametrize("fixture_name", HASH_TYPES)
-def test_hash_consistency(fixture_name, request):
+@pytest.mark.parametrize("sample_object", HASH_TYPES, indirect=True)
+def test_hash_consistency(sample_object):
     """Hash is consistent across multiple calls."""
-    obj = request.getfixturevalue(fixture_name)
-    assert hash(obj) == hash(obj)
+    assert hash(sample_object) == hash(sample_object)
 
 
-@pytest.mark.parametrize("a_name,b_name", SAME_TYPE_PAIRS)
-def test_hash_distinct_same_type(a_name, b_name, request):
+@pytest.mark.parametrize("sample_object_a,sample_object_b", SAME_TYPE_PAIRS, indirect=True)
+def test_hash_distinct_same_type(sample_object_a, sample_object_b):
     """Distinct objects of the same type have different hashes."""
-    obj_a = request.getfixturevalue(a_name)
-    obj_b = request.getfixturevalue(b_name)
-    assert hash(obj_a) != hash(obj_b)  # extremely unlikely
+    assert hash(sample_object_a) != hash(sample_object_b)  # extremely unlikely
 
 
-@pytest.mark.parametrize("a_name,b_name", itertools.combinations(HASH_TYPES, 2))
-def test_hash_distinct_cross_type(a_name, b_name, request):
+@pytest.mark.parametrize("sample_object_a,sample_object_b", itertools.combinations(HASH_TYPES, 2), indirect=True)
+def test_hash_distinct_cross_type(sample_object_a, sample_object_b):
     """Distinct objects of different types have different hashes."""
-    obj_a = request.getfixturevalue(a_name)
-    obj_b = request.getfixturevalue(b_name)
-    assert hash(obj_a) != hash(obj_b)  # extremely unlikely
+    assert hash(sample_object_a) != hash(sample_object_b)  # extremely unlikely
 
 
 # =============================================================================
@@ -763,41 +777,35 @@ def test_hash_distinct_cross_type(a_name, b_name, request):
 # =============================================================================
 
 
-@pytest.mark.parametrize("fixture_name", EQ_TYPES)
-def test_equality_basic(fixture_name, request):
+@pytest.mark.parametrize("sample_object", EQ_TYPES, indirect=True)
+def test_equality_basic(sample_object):
     """Object equality: reflexive, not equal to None or other types."""
-    obj = request.getfixturevalue(fixture_name)
-    assert obj == obj
-    assert obj is not None
-    assert obj != "string"
-    if hasattr(obj, "handle"):
-        assert obj != obj.handle
+    assert sample_object == sample_object
+    assert sample_object is not None
+    assert sample_object != "string"
+    if hasattr(sample_object, "handle"):
+        assert sample_object != sample_object.handle
 
 
-@pytest.mark.parametrize("a_name,b_name", itertools.combinations(EQ_TYPES, 2))
-def test_no_cross_type_equality(a_name, b_name, request):
+@pytest.mark.parametrize("sample_object_a,sample_object_b", itertools.combinations(EQ_TYPES, 2), indirect=True)
+def test_no_cross_type_equality(sample_object_a, sample_object_b):
     """No two distinct objects of different types should compare equal."""
-    obj_a = request.getfixturevalue(a_name)
-    obj_b = request.getfixturevalue(b_name)
-    assert obj_a != obj_b
+    assert sample_object_a != sample_object_b
 
 
-@pytest.mark.parametrize("a_name,b_name", SAME_TYPE_PAIRS)
-def test_same_type_inequality(a_name, b_name, request):
+@pytest.mark.parametrize("sample_object_a,sample_object_b", SAME_TYPE_PAIRS, indirect=True)
+def test_same_type_inequality(sample_object_a, sample_object_b):
     """Two distinct objects of the same type should not compare equal."""
-    obj_a = request.getfixturevalue(a_name)
-    obj_b = request.getfixturevalue(b_name)
-    assert obj_a is not obj_b
-    assert obj_a != obj_b
+    assert sample_object_a is not sample_object_b
+    assert sample_object_a != sample_object_b
 
 
-@pytest.mark.parametrize("fixture_name,copy_fn", FROM_HANDLE_COPIES)
-def test_equality_same_handle(fixture_name, copy_fn, request):
+@pytest.mark.parametrize("sample_object,copy_fn", FROM_HANDLE_COPIES, indirect=["sample_object"])
+def test_equality_same_handle(sample_object, copy_fn):
     """Two wrappers around the same handle should compare equal."""
-    obj = request.getfixturevalue(fixture_name)
-    obj2 = copy_fn(obj)
-    assert obj == obj2
-    assert hash(obj) == hash(obj2)
+    obj2 = copy_fn(sample_object)
+    assert sample_object == obj2
+    assert hash(sample_object) == hash(obj2)
 
 
 # =============================================================================
@@ -805,48 +813,43 @@ def test_equality_same_handle(fixture_name, copy_fn, request):
 # =============================================================================
 
 
-@pytest.mark.parametrize("fixture_name", DICT_KEY_TYPES)
-def test_usable_as_dict_key(fixture_name, request):
+@pytest.mark.parametrize("sample_object", DICT_KEY_TYPES, indirect=True)
+def test_usable_as_dict_key(sample_object):
     """Object can be used as a dictionary key."""
-    obj = request.getfixturevalue(fixture_name)
-    d = {obj: "value"}
-    assert d[obj] == "value"
-    assert obj in d
+    d = {sample_object: "value"}
+    assert d[sample_object] == "value"
+    assert sample_object in d
 
 
-@pytest.mark.parametrize("fixture_name", DICT_KEY_TYPES)
-def test_usable_in_set(fixture_name, request):
+@pytest.mark.parametrize("sample_object", DICT_KEY_TYPES, indirect=True)
+def test_usable_in_set(sample_object):
     """Object can be added to a set."""
-    obj = request.getfixturevalue(fixture_name)
-    s = {obj}
-    assert obj in s
+    s = {sample_object}
+    assert sample_object in s
 
 
-@pytest.mark.parametrize("fixture_name", WEAKREF_TYPES)
-def test_usable_in_weak_value_dict(fixture_name, request):
+@pytest.mark.parametrize("sample_object", WEAKREF_TYPES, indirect=True)
+def test_usable_in_weak_value_dict(sample_object):
     """Object can be used as a WeakValueDictionary value."""
-    obj = request.getfixturevalue(fixture_name)
     wvd = weakref.WeakValueDictionary()
-    wvd["key"] = obj
-    assert wvd["key"] is obj
+    wvd["key"] = sample_object
+    assert wvd["key"] is sample_object
 
 
-@pytest.mark.parametrize("fixture_name", WEAK_KEY_TYPES)
-def test_usable_in_weak_key_dict(fixture_name, request):
+@pytest.mark.parametrize("sample_object", WEAK_KEY_TYPES, indirect=True)
+def test_usable_in_weak_key_dict(sample_object):
     """Object can be used as a WeakKeyDictionary key."""
-    obj = request.getfixturevalue(fixture_name)
     wkd = weakref.WeakKeyDictionary()
-    wkd[obj] = "value"
-    assert wkd[obj] == "value"
+    wkd[sample_object] = "value"
+    assert wkd[sample_object] == "value"
 
 
-@pytest.mark.parametrize("fixture_name", WEAK_KEY_TYPES)
-def test_usable_in_weak_set(fixture_name, request):
+@pytest.mark.parametrize("sample_object", WEAK_KEY_TYPES, indirect=True)
+def test_usable_in_weak_set(sample_object):
     """Object can be added to a WeakSet."""
-    obj = request.getfixturevalue(fixture_name)
     ws = weakref.WeakSet()
-    ws.add(obj)
-    assert obj in ws
+    ws.add(sample_object)
+    assert sample_object in ws
 
 
 # =============================================================================
@@ -854,12 +857,10 @@ def test_usable_in_weak_set(fixture_name, request):
 # =============================================================================
 
 
-@pytest.mark.parametrize("fixture_name,pattern", REPR_PATTERNS)
-def test_repr_format(fixture_name, pattern, request):
+@pytest.mark.parametrize("sample_object,pattern", REPR_PATTERNS, indirect=["sample_object"])
+def test_repr_format(sample_object, pattern):
     """repr() returns a properly formatted string."""
-    obj = request.getfixturevalue(fixture_name)
-    result = repr(obj)
-    assert re.fullmatch(pattern, result)
+    assert re.fullmatch(pattern, repr(sample_object))
 
 
 # =============================================================================
@@ -868,10 +869,9 @@ def test_repr_format(fixture_name, pattern, request):
 
 
 @pytest.mark.parametrize("pickle_module", PICKLE_MODULES)
-@pytest.mark.parametrize("fixture_name", PICKLE_TYPES)
-def test_pickle_roundtrip(fixture_name, pickle_module, request):
+@pytest.mark.parametrize("sample_object", PICKLE_TYPES, indirect=True)
+def test_pickle_roundtrip(sample_object, pickle_module):
     """Object survives a pickle/cloudpickle roundtrip."""
     mod = pytest.importorskip(pickle_module)
-    obj = request.getfixturevalue(fixture_name)
-    result = mod.loads(mod.dumps(obj))
-    assert type(result) is type(obj)
+    result = mod.loads(mod.dumps(sample_object))
+    assert type(result) is type(sample_object)