From 68f832c1d146673eed9915b9bd55e32279627805 Mon Sep 17 00:00:00 2001
From: Ralf Juengling <rjuengling@nvidia.com>
Date: Thu, 8 Oct 2026 10:14:07 -0700
Subject: [PATCH 01/10] test(cuda.core): request init_cuda for memory resource
 tests

Tests that allocate through Device.memory_resource /
Device.allocate or construct any memory resource without init_cuda did
Device() + set_current() by hand, leaving a stale cached
default MemoryResource on the thread-local Device: no teardown
gc.collect(), no cuCtxSynchronize, and the context never popped.
The stale slot is the aliasing route for the planned
live-resource-count gc.collect() gate, and the tests violate the
tests/AGENTS.md rule that constructing a memory resource counts as
touching CUDA.

Route both groups onto init_cuda (using the yielded device) so the
fixture teardown clears the cached slot via del _tls.devices:

  - device.memory_resource allocators: 6 test_from_buffer* functions in
    test_utils.py and test_buffer_external_device in test_memory.py.
  - resource constructors: 28 functions in test_memory.py (buffers,
    dlpack, vmm, pinned/managed ipc, legacy pinned, abstract stubs)
    and 4 launch-scalar tests in test_launcher.py.

test_buffer_external_device and test_buffer_external_pinned_alloc still
select Device(n - 1) as before; init_cuda only supplies the context
and teardown. The no-context tests (test_close_with_default_stream_requires_context,
test_from_handle_mr_default_stream_requires_context) keep their no_current_context()
body. Assertions are unchanged.

Add a sentence to tests/AGENTS.md that allocating through
device.memory_resource or Device.allocate, or constructing any memory
resource, counts as touching CUDA and requires init_cuda.
---
 cuda_core/tests/AGENTS.md        |   4 +-
 cuda_core/tests/test_launcher.py |  22 +++---
 cuda_core/tests/test_memory.py   | 119 +++++++++++++------------------
 cuda_core/tests/test_utils.py    |  30 ++++----
 4 files changed, 75 insertions(+), 100 deletions(-)

diff --git a/cuda_core/tests/AGENTS.md b/cuda_core/tests/AGENTS.md
index 0bb6a1a3928..a8ff9eb9227 100644
--- a/cuda_core/tests/AGENTS.md
+++ b/cuda_core/tests/AGENTS.md
@@ -239,7 +239,9 @@ that calls a CUDA API without `init_cuda` (or an explicit
 to leave current on the thread — possibly none. With `pytest-randomly` that
 makes the pass/fail outcome depend on test order, so it moves seed to seed and
 looks like flakiness. Request `init_cuda` for any test that calls into the
-driver, or set up and tear down a context yourself.
+driver, or set up and tear down a context yourself. Allocating through
+`device.memory_resource` or `Device.allocate`, or constructing any memory
+resource, counts as touching CUDA and requires `init_cuda`.
 
 ## Assert on behavior, not implementation
 
diff --git a/cuda_core/tests/test_launcher.py b/cuda_core/tests/test_launcher.py
index 75cea063615..2466d966439 100644
--- a/cuda_core/tests/test_launcher.py
+++ b/cuda_core/tests/test_launcher.py
@@ -244,7 +244,7 @@ def test_to_native_launch_config_priority(init_cuda, priority, expected_num_attr
 
 
 @skipif_need_cuda_headers
-def test_pdl_primary_secondary_overlap_same_stream():
+def test_pdl_primary_secondary_overlap_same_stream(init_cuda):
     """Primary + secondary PDL launch on one stream can overlap on Hopper+.
 
     Secondary is launched with ``programmatic_stream_serialization=True``. After
@@ -255,10 +255,9 @@ def test_pdl_primary_secondary_overlap_same_stream():
     Note concurrency is opportunistic, so a missing overlap execution is reported as
     an expected failure.
     """
-    dev = Device()
+    dev = init_cuda
     if dev.compute_capability < (9, 0):
         pytest.skip("Programmatic Dependent Launch requires compute capability >= 9.0")
-    dev.set_current()
     stream = dev.create_stream(options=StreamOptions(nonblocking=True))
 
     # clock64 budgets are in GPU cycles; keep the post-trigger window long enough
@@ -592,9 +591,8 @@ def write_scalar_module():
 
 @pytest.mark.parametrize("python_type, cpp_type, init_value", PARAMS)
 @requires_module(np, "2.2.5", reason="need numpy 2.2.5+ (numpy GH #28632)")
-def test_launch_scalar_argument(write_scalar_module, python_type, cpp_type, init_value):
-    dev = Device(0)
-    dev.set_current()
+def test_launch_scalar_argument(init_cuda, write_scalar_module, python_type, cpp_type, init_value):
+    dev = init_cuda
 
     # Prepare pinned host array
     mr = LegacyPinnedMemoryResource()
@@ -825,7 +823,7 @@ class MyBool(ctypes.c_bool):
         "numpy_float32",
     ],
 )
-def test_launch_scalar_argument_subclass_fallback(base_type, np_dtype, cpp_type, raw_value):
+def test_launch_scalar_argument_subclass_fallback(init_cuda, base_type, np_dtype, cpp_type, raw_value):
     """Subclassed scalar arguments survive fallback handling and reach the kernel
     with the correct width/sign. The readback value (not just ptr != 0) guards each
     fallback branch against marshalling the wrong C type, e.g. uint64 -> uint32_t."""
@@ -836,8 +834,7 @@ class Subclassed(base_type):
     scalar = Subclassed(raw_value)
     expected = np_dtype(raw_value)
 
-    dev = Device()
-    dev.set_current()
+    dev = init_cuda
 
     mr = LegacyPinnedMemoryResource()
     b = mr.allocate(np.dtype(np_dtype).itemsize)
@@ -939,7 +936,9 @@ class MyComplex(complex):
     ids=_NUMPY_SUBCLASS_FALLBACK_IDS,
 )
 @pytest.mark.agent_authored(model="gpt-5.6-sol")
-def test_launch_numpy_scalar_subclass_fallback(write_scalar_module, base_type, np_dtype, cpp_type, raw_value):
+def test_launch_numpy_scalar_subclass_fallback(
+    init_cuda, write_scalar_module, base_type, np_dtype, cpp_type, raw_value
+):
     """Subclassed numpy scalars take prepare_numpy_arg's isinstance fallback and reach the kernel (readback)."""
 
     class Subclassed(base_type):
@@ -948,8 +947,7 @@ class Subclassed(base_type):
     scalar = Subclassed(raw_value)
     expected = np_dtype(raw_value)
 
-    dev = Device(0)
-    dev.set_current()
+    dev = init_cuda
 
     mr = LegacyPinnedMemoryResource()
     b = mr.allocate(np.dtype(np_dtype).itemsize)
diff --git a/cuda_core/tests/test_memory.py b/cuda_core/tests/test_memory.py
index f33e17ddccf..d2cdc6de459 100644
--- a/cuda_core/tests/test_memory.py
+++ b/cuda_core/tests/test_memory.py
@@ -158,9 +158,8 @@ def buffer_initialization(dummy_mr: MemoryResource):
     buffer.close()
 
 
-def test_buffer_initialization():
-    device = Device()
-    device.set_current()
+def test_buffer_initialization(init_cuda):
+    device = init_cuda
     buffer_initialization(DummyDeviceMemoryResource(device))
     buffer_initialization(DummyHostMemoryResource())
     buffer_initialization(DummyUnifiedMemoryResource(device))
@@ -177,10 +176,9 @@ def test_buffer_direct_init_forbidden():
 
 
 @pytest.mark.agent_authored(model="gpt-5.6-sol")
-def test_buffer_context_manager_closes_on_exit():
+def test_buffer_context_manager_closes_on_exit(init_cuda):
     """``with buf`` yields the buffer, closes it on exit, and does not swallow inner exceptions."""
-    device = Device()
-    device.set_current()
+    device = init_cuda
     mr = DummyDeviceMemoryResource(device)
     buf = mr.allocate(size=64, stream=device.default_stream)
     with buf as entered:
@@ -196,10 +194,9 @@ def test_buffer_context_manager_closes_on_exit():
 
 
 @pytest.mark.agent_authored(model="gpt-5.6-sol")
-def test_memory_resource_abstract_stubs():
+def test_memory_resource_abstract_stubs(init_cuda):
     """Every abstract MemoryResource member reports itself as unimplemented."""
-    device = Device()
-    device.set_current()
+    device = init_cuda
     mr = MemoryResource()
     stream = device.default_stream
     with pytest.raises(TypeError, match=r"^MemoryResource\.allocate must be implemented"):
@@ -237,9 +234,8 @@ def buffer_copy_to(dummy_mr: MemoryResource, device: Device, check=False):
     src_buffer.close()
 
 
-def test_buffer_copy_to():
-    device = Device()
-    device.set_current()
+def test_buffer_copy_to(init_cuda):
+    device = init_cuda
     buffer_copy_to(DummyDeviceMemoryResource(device), device)
     buffer_copy_to(DummyUnifiedMemoryResource(device), device)
     buffer_copy_to(DummyPinnedMemoryResource(device), device, check=True)
@@ -268,17 +264,15 @@ def buffer_copy_from(dummy_mr: MemoryResource, device, check=False):
     src_buffer.close()
 
 
-def test_buffer_copy_from():
-    device = Device()
-    device.set_current()
+def test_buffer_copy_from(init_cuda):
+    device = init_cuda
     buffer_copy_from(DummyDeviceMemoryResource(device), device)
     buffer_copy_from(DummyUnifiedMemoryResource(device), device)
     buffer_copy_from(DummyPinnedMemoryResource(device), device, check=True)
 
 
-def test_buffer_copy_to_size_mismatch_raises():
-    device = Device()
-    device.set_current()
+def test_buffer_copy_to_size_mismatch_raises(init_cuda):
+    device = init_cuda
     mr = DummyDeviceMemoryResource(device)
     stream = device.create_stream()
     src_buffer = mr.allocate(size=1024)
@@ -291,9 +285,8 @@ def test_buffer_copy_to_size_mismatch_raises():
     src_buffer.close()
 
 
-def test_buffer_copy_from_size_mismatch_raises():
-    device = Device()
-    device.set_current()
+def test_buffer_copy_from_size_mismatch_raises(init_cuda):
+    device = init_cuda
     mr = DummyDeviceMemoryResource(device)
     stream = device.create_stream()
     src_buffer = mr.allocate(size=1024)
@@ -428,9 +421,8 @@ def buffer_close(dummy_mr: MemoryResource):
     assert buffer.memory_resource is None
 
 
-def test_buffer_close():
-    device = Device()
-    device.set_current()
+def test_buffer_close(init_cuda):
+    device = init_cuda
     buffer_close(DummyDeviceMemoryResource(device))
     buffer_close(DummyHostMemoryResource())
     buffer_close(DummyUnifiedMemoryResource(device))
@@ -449,7 +441,7 @@ def test_buffer_external_host():
 
 
 @pytest.mark.parametrize("change_device", [True, False])
-def test_buffer_external_device(change_device):
+def test_buffer_external_device(init_cuda, change_device):
     n = len(Device.get_all_devices())
     if n < 1:
         pytest.skip("No devices found")
@@ -473,7 +465,7 @@ def test_buffer_external_device(change_device):
 
 
 @pytest.mark.parametrize("change_device", [True, False])
-def test_buffer_external_pinned_alloc(change_device):
+def test_buffer_external_pinned_alloc(init_cuda, change_device):
     n = len(Device.get_all_devices())
     if n < 1:
         pytest.skip("No devices found")
@@ -635,9 +627,8 @@ def test_buffer_deallocation_stream_configuration_paths(configuration, destructi
 
 
 @pytest.mark.agent_authored(model="gpt-5.6")
-def test_set_deallocation_stream_rejects_none_and_closed_buffer():
-    device = Device()
-    device.set_current()
+def test_set_deallocation_stream_rejects_none_and_closed_buffer(init_cuda):
+    device = init_cuda
     stream = device.create_stream()
     mr = StubMemoryResource(device)
     buf = Buffer.from_handle(1, 1024, mr=mr, stream=stream)
@@ -775,15 +766,14 @@ def test_from_handle_stream_requires_mr(buffer_type):
 
 
 @pytest.mark.agent_authored(model="claude-sonnet-4-6")
-def test_close_with_default_stream_requires_context():
+def test_close_with_default_stream_requires_context(init_cuda):
     """Buffer.close(stream=default_stream()) raises when no context is current.
 
     ``default_stream()`` has no bound context, so the close path must find
     a current context to anchor the free.  Without one it should raise rather
     than silently record an unusable stream handle.
     """
-    device = Device()
-    device.set_current()
+    device = init_cuda
     stream = device.create_stream()
     mr = StubMemoryResource(device)
     # Use a real stream at creation so _init succeeds without a current context later.
@@ -799,10 +789,9 @@ def test_close_with_default_stream_requires_context():
 
 @pytest.mark.agent_authored(model="cursor-grok-4.5")
 @pytest.mark.parametrize("buffer_type", [Buffer, ManagedBuffer])
-def test_from_handle_mr_default_stream_requires_context(buffer_type):
+def test_from_handle_mr_default_stream_requires_context(init_cuda, buffer_type):
     """Owning from_handle with the default stream needs a current context."""
-    device = Device()
-    device.set_current()
+    device = init_cuda
     mr = StubMemoryResource(device)
     with no_current_context():
         assert current_context_handle() == 0
@@ -1041,11 +1030,11 @@ def test_pool_buffer_deallocates_with_foreign_context(mempool_device_x2):
         alloc_dev.set_current()
 
 
-def test_memory_resource_and_owner_disallowed():
+def test_memory_resource_and_owner_disallowed(init_cuda):
     with pytest.raises(ValueError, match="cannot be both specified together"):
         a = (ctypes.c_byte * 20)()
         ptr = ctypes.addressof(a)
-        Buffer.from_handle(ptr, 20, mr=DummyDeviceMemoryResource(Device()), owner=a)
+        Buffer.from_handle(ptr, 20, mr=DummyDeviceMemoryResource(init_cuda), owner=a)
 
 
 def test_owner_close():
@@ -1059,9 +1048,8 @@ def test_owner_close():
     assert after == before
 
 
-def test_buffer_dunder_dlpack():
-    device = Device()
-    device.set_current()
+def test_buffer_dunder_dlpack(init_cuda):
+    device = init_cuda
     dummy_mr = DummyDeviceMemoryResource(device)
     buffer = dummy_mr.allocate(size=1024)
     capsule = buffer.__dlpack__()
@@ -1095,18 +1083,18 @@ def test_buffer_dunder_dlpack_device_success(DummyMR, expected):
     assert buffer.__dlpack_device__() == expected
 
 
-def test_buffer_dunder_dlpack_device_failure():
+def test_buffer_dunder_dlpack_device_failure(init_cuda):
     # avoids an error capturing the default stream with no context
-    Device().set_current()
+    # (init_cuda has already set a current context)
     dummy_mr = NullMemoryResource()
     buffer = dummy_mr.allocate(size=1024)
     with pytest.raises(BufferError, match=r"^buffer is neither device-accessible nor host-accessible$"):
         buffer.__dlpack_device__()
 
 
-def test_buffer_dlpack_failure_clean_up():
+def test_buffer_dlpack_failure_clean_up(init_cuda):
     # avoids an error capturing the default stream with no context
-    Device().set_current()
+    # (init_cuda has already set a current context)
     dummy_mr = NullMemoryResource()
     buffer = dummy_mr.allocate(size=1024)
     before = sys.getrefcount(buffer)
@@ -1119,10 +1107,9 @@ def test_buffer_dlpack_failure_clean_up():
     assert after == before
 
 
-def test_managed_buffer_dlpack_roundtrip_device_type():
+def test_managed_buffer_dlpack_roundtrip_device_type(init_cuda):
     """Verify that a managed Buffer round-trips through DLPack with kDLCUDAManaged."""
-    device = Device()
-    device.set_current()
+    device = init_cuda
     skip_if_managed_memory_unsupported(device)
     mr = DummyUnifiedMemoryResource(device)
     buf = mr.allocate(size=1024)
@@ -1136,7 +1123,7 @@ def test_managed_buffer_dlpack_roundtrip_device_type():
     assert view.__dlpack_device__() == (int(DLDeviceType.kDLCUDAManaged), 0)
 
 
-def test_managed_memory_resource_buffer_dlpack_device_type():
+def test_managed_memory_resource_buffer_dlpack_device_type(init_cuda):
     """Verify that pool-allocated managed memory reports kDLCUDAManaged.
 
     Allocations from ManagedMemoryResource go through cuMemAllocFromPoolAsync
@@ -1145,8 +1132,7 @@ def test_managed_memory_resource_buffer_dlpack_device_type():
     managed via its memory resource so that DLPack consumers (e.g. CCCL's
     make_tma_descriptor) accept the buffer.
     """
-    device = Device()
-    device.set_current()
+    device = init_cuda
     skip_if_managed_memory_unsupported(device)
     mr = create_managed_memory_resource_or_skip(ManagedMemoryResourceOptions(preferred_location=device.device_id))
     buf = mr.allocate(1024, stream=device.default_stream)
@@ -1160,10 +1146,9 @@ def test_managed_memory_resource_buffer_dlpack_device_type():
 
 
 @pytest.mark.parametrize("mr_kind", ["device", "pinned"])
-def test_non_managed_resources_report_not_managed(mr_kind):
+def test_non_managed_resources_report_not_managed(init_cuda, mr_kind):
     """Non-managed memory resources must report is_managed=False."""
-    device = Device()
-    device.set_current()
+    device = init_cuda
     if not device.properties.memory_pools_supported:
         pytest.skip("Device does not support mempool operations")
     if mr_kind == "device":
@@ -1182,19 +1167,17 @@ def test_non_managed_resources_report_not_managed(mr_kind):
 
 
 @pytest.mark.parametrize("use_device_object", [True, False])
-def test_device_memory_resource_initialization(use_device_object):
+def test_device_memory_resource_initialization(init_cuda, use_device_object):
     """Test that DeviceMemoryResource can be initialized successfully.
 
     This test verifies that the DeviceMemoryResource initializes properly,
     including the release threshold configuration for performance optimization.
     """
-    device = Device()
+    device = init_cuda
 
     if not device.properties.memory_pools_supported:
         pytest.skip("Device does not support mempool operations")
 
-    device.set_current()
-
     # This should succeed and configure the memory pool release threshold.
     # The resource can be constructed from either a device or device ordinal.
     device_arg = device if use_device_object else device.device_id
@@ -1366,7 +1349,7 @@ def _vmm_allocate_in_free_hole(device, size, extra):
 
 @pytest.mark.parametrize("use_device_object", [True, False])
 @pytest.mark.parametrize("handle_type", get_handle_type())
-def test_vmm_allocator_basic_allocation(use_device_object, handle_type):
+def test_vmm_allocator_basic_allocation(init_cuda, use_device_object, handle_type):
     """Test basic VMM allocation functionality.
 
     This test verifies that VirtualMemoryResource can allocate memory
@@ -1409,7 +1392,7 @@ def test_vmm_allocator_basic_allocation(use_device_object, handle_type):
 
 
 @pytest.mark.agent_authored(model="claude-fable-5-1")
-def test_vmm_allocator_policy_configuration():
+def test_vmm_allocator_policy_configuration(init_cuda):
     """Test VMM allocator with different policy configurations.
 
     The resource applies its configuration to every allocation. A
@@ -1502,7 +1485,7 @@ def test_vmm_allocator_policy_configuration():
 
 @pytest.mark.agent_authored(model="claude-fable-5-1")
 @pytest.mark.parametrize("handle_type", get_handle_type())
-def test_vmm_allocator_grow_allocation(handle_type):
+def test_vmm_allocator_grow_allocation(init_cuda, handle_type):
     """Test VMM allocator's ability to grow existing allocations.
 
     ``modify_allocation`` returns a new buffer that aliases the input, which
@@ -2124,7 +2107,7 @@ def test_vmm_allocate_close_does_not_leak(init_cuda, mode):
     _vmm_assert_released(reservations)
 
 
-def test_vmm_allocator_rdma_unsupported_exception():
+def test_vmm_allocator_rdma_unsupported_exception(init_cuda):
     """Test that VirtualMemoryResource throws an exception when RDMA is requested but device doesn't support it.
 
     This test verifies that the VirtualMemoryResource constructor throws a RuntimeError
@@ -2523,10 +2506,9 @@ def test_mempool_ipc_errors(mempool_device):
     device.sync()
 
 
-def test_pinned_mempool_ipc_basic():
+def test_pinned_mempool_ipc_basic(init_cuda):
     """Test basic IPC functionality for PinnedMemoryResource."""
-    device = Device()
-    device.set_current()
+    device = init_cuda
 
     skip_if_pinned_memory_unsupported(device)
 
@@ -2566,10 +2548,9 @@ def test_pinned_mempool_ipc_basic():
     mr.close()
 
 
-def test_pinned_mempool_ipc_errors():
+def test_pinned_mempool_ipc_errors(init_cuda):
     """Test error cases when IPC operations are disabled for PinnedMemoryResource."""
-    device = Device()
-    device.set_current()
+    device = init_cuda
 
     skip_if_pinned_memory_unsupported(device)
 
@@ -2932,7 +2913,7 @@ def test_legacy_pinned_allocate_zero_size(init_cuda):
     assert int(buf.handle) == 0
 
 
-def test_legacy_pinned_device_id_is_not_applicable():
+def test_legacy_pinned_device_id_is_not_applicable(init_cuda):
     """LegacyPinnedMemoryResource.device_id is -1, as documented for memory not bound to a device."""
     mr = LegacyPinnedMemoryResource()
     assert mr.device_id == -1
@@ -3103,7 +3084,7 @@ def test_vmm_options_handle_type_win32_raises():
 
 @pytest.mark.agent_authored(model="claude-fable-5-1")
 @pytest.mark.parametrize("location_type", ["host", "host_numa", "host_numa_current"])
-def test_vmm_host_location_types_report_host_accessible(location_type):
+def test_vmm_host_location_types_report_host_accessible(init_cuda, location_type):
     """Every host-backed location type reports is_host_accessible; only "host" needs handle_type=None.
 
     __init__ classifies "host", "host_numa" and "host_numa_current" alike when
diff --git a/cuda_core/tests/test_utils.py b/cuda_core/tests/test_utils.py
index 0aaec2e04b9..a54e8378cf9 100644
--- a/cuda_core/tests/test_utils.py
+++ b/cuda_core/tests/test_utils.py
@@ -388,9 +388,8 @@ def _dense_strides(shape, stride_order):
 @pytest.mark.parametrize("dtype", [np.dtype(np.int8), np.dtype(np.uint32)], ids=str)
 @pytest.mark.parametrize("stride_order", ["C", "F"])
 @pytest.mark.parametrize("readonly", [True, False])
-def test_from_buffer(shape, dtype, stride_order, readonly):
-    dev = Device()
-    dev.set_current()
+def test_from_buffer(init_cuda, shape, dtype, stride_order, readonly):
+    dev = init_cuda
     layout = _StridedLayout.dense(shape=shape, itemsize=dtype.itemsize, stride_order=stride_order)
     required_size = layout.required_size_in_bytes()
     assert required_size == math.prod(shape) * dtype.itemsize
@@ -414,20 +413,18 @@ def test_from_buffer(shape, dtype, stride_order, readonly):
         (None, None, "itemsize or dtype must be specified"),
     ],
 )
-def test_from_buffer_incompatible_dtype_and_itemsize(dtype, itemsize, msg):
+def test_from_buffer_incompatible_dtype_and_itemsize(init_cuda, dtype, itemsize, msg):
     layout = _StridedLayout.dense((5,), 2)
-    device = Device()
-    device.set_current()
+    device = init_cuda
     buffer = device.memory_resource.allocate(layout.required_size_in_bytes(), stream=device.default_stream)
     with pytest.raises(ValueError, match=msg):
         StridedMemoryView.from_buffer(buffer, (5,), dtype=dtype, itemsize=itemsize)
 
 
 @pytest.mark.parametrize("stride_order", ["C", "F"])
-def test_from_buffer_sliced(stride_order):
+def test_from_buffer_sliced(init_cuda, stride_order):
     layout = _StridedLayout.dense((5, 7), 2, stride_order=stride_order)
-    device = Device()
-    device.set_current()
+    device = init_cuda
     buffer = device.memory_resource.allocate(layout.required_size_in_bytes(), stream=device.default_stream)
     view = StridedMemoryView.from_buffer(buffer, (5, 7), dtype=np.dtype(np.int16))
     assert view.shape == (5, 7)
@@ -442,10 +439,9 @@ def test_from_buffer_sliced(stride_order):
     assert int(buffer.handle) + expected_offset * 2 == sliced_view.ptr
 
 
-def test_from_buffer_too_small():
+def test_from_buffer_too_small(init_cuda):
     layout = _StridedLayout.dense((5, 4), 2)
-    d = Device()
-    d.set_current()
+    d = init_cuda
     buffer = d.memory_resource.allocate(20, stream=d.default_stream)
     with pytest.raises(ValueError, match="Expected at least 40 bytes, got 20 bytes."):
         StridedMemoryView.from_buffer(
@@ -456,10 +452,9 @@ def test_from_buffer_too_small():
         )
 
 
-def test_from_buffer_disallowed_negative_offset():
+def test_from_buffer_disallowed_negative_offset(init_cuda):
     layout = _StridedLayout((5, 4), (-4, 1), 1)
-    d = Device()
-    d.set_current()
+    d = init_cuda
     buffer = d.memory_resource.allocate(20, stream=d.default_stream)
     with pytest.raises(ValueError):
         StridedMemoryView.from_buffer(
@@ -584,9 +579,8 @@ def test_view_zero_size_array(init_cuda, api, shape, dtype):
     assert smv.dtype == np.dtype(dtype)
 
 
-def test_from_buffer_with_non_power_of_two_itemsize():
-    dev = Device()
-    dev.set_current()
+def test_from_buffer_with_non_power_of_two_itemsize(init_cuda):
+    dev = init_cuda
     dtype = np.dtype([("a", "int32"), ("b", "int8")])
     shape = (1,)
     layout = _StridedLayout(shape=shape, strides=None, itemsize=dtype.itemsize)

From 480dddbb041bb013e5803adda6c5cca96bce3c69 Mon Sep 17 00:00:00 2001
From: Ralf Juengling <rjuengling@nvidia.com>
Date: Thu, 8 Oct 2026 12:01:37 -0700
Subject: [PATCH 02/10] test(cuda.core): release baseline-retained memory
 resources at test boundaries

The snapshot gc.collect() gate is sound only if no older
MemoryResource is released during a test, otherwise a mid-test
baseline release could mask a leak. Three tests retained a
closed (but not destroyed) memory resource past their teardown,
which would ratchet the baseline and later cancel a new leak:

  - tests/memory_ipc/test_errors.py: ChildErrorHarness keeps the
    fixture's buffer/mr/device references on the pytest-retained
    item._instance for the session; drop them from __dict__ so the closed
    MR objects do not persist.
  - tests/test_error_handling.py::test_escalated_cudawarning_from_cleanup_is_not_a_crash:
    the unraisable hook args hold exc_traceback whose frame holds
    ``mr`` (never closed); clear the list, drop the locals, and
    close the pool.
  - tests/test_memory.py::fill_env: the fixture returns the MR to
    the test via a tuple; convert it to a generator that dels ``mr`` in
    its teardown so the closed MR does not persist in the
    pytest fixture cache.

These are the baseline-retention fixes prerequisite for the planned
live-resource-count gc.collect() gate (see plan
mr-instance-count-gc-gate, "Soundness conditions of the snapshot
gate").
---
 cuda_core/tests/memory_ipc/test_errors.py |  8 ++++++++
 cuda_core/tests/test_error_handling.py    |  6 ++++++
 cuda_core/tests/test_memory.py            | 19 ++++++++++++-------
 3 files changed, 26 insertions(+), 7 deletions(-)

diff --git a/cuda_core/tests/memory_ipc/test_errors.py b/cuda_core/tests/memory_ipc/test_errors.py
index a09c4f44ac9..aceb00332ab 100644
--- a/cuda_core/tests/memory_ipc/test_errors.py
+++ b/cuda_core/tests/memory_ipc/test_errors.py
@@ -162,6 +162,14 @@ def test_main(self, ipc_device, ipc_memory_resource):
         finally:
             for mr in self._extra_mrs:
                 mr.close()
+            # mr-baseline-retention-fix (H1): drop the fixture MR and buffer
+            # references stored on this class instance, pytest retains
+            # ``item._instance`` for the session, so the closed (but not
+            # destroyed) MR objects would otherwise ratchet the baseline.
+            self.__dict__.pop("buffer", None)
+            self.__dict__.pop("mr", None)
+            self.__dict__.pop("device", None)
+            self._extra_mrs.clear()
 
     def child_main(self, pipe, device, mr):
         """Child process that pushes IPC errors to a shared pipe for testing."""
diff --git a/cuda_core/tests/test_error_handling.py b/cuda_core/tests/test_error_handling.py
index 1ff3ea2b461..56ebc764da7 100644
--- a/cuda_core/tests/test_error_handling.py
+++ b/cuda_core/tests/test_error_handling.py
@@ -230,6 +230,12 @@ def test_escalated_cudawarning_from_cleanup_is_not_a_crash(mempool_device):
     assert len(unraisable) == 1
     assert issubclass(unraisable[0].exc_type, CUDAWarning)
     assert "restoring the caller's context" in str(unraisable[0].exc_value)
+    # mr-baseline-retention-fix (H3): the unraisable hook args hold
+    # exc_traceback whose frame holds ``mr`` (never closed). Clear the
+    # list and drop the locals so gc can reclaim mr; close the pool too.
+    unraisable.clear()
+    mr.close()
+    del mr, buf
 
 
 @thread_unsafe_context_fault
diff --git a/cuda_core/tests/test_memory.py b/cuda_core/tests/test_memory.py
index d2cdc6de459..9a5c0738e5b 100644
--- a/cuda_core/tests/test_memory.py
+++ b/cuda_core/tests/test_memory.py
@@ -334,7 +334,14 @@ def fill_env(request, init_cuda):
         mr = DummyUnifiedMemoryResource(device)
     else:
         mr = DummyPinnedMemoryResource(device)
-    return device, mr
+    try:
+        yield device, mr
+    finally:
+        # mr-baseline-retention-fix (H4): drop the fixture's MR
+        # reference so the (closed-by-dealloc) object does not persist
+        # in the pytest fixture cache past this teardown; without this the
+        # baseline ratchets/oscillates with fragile release timing.
+        del mr
 
 
 _FILL_SIZE = 64  # Keep small; divisible by 1/2/4.
@@ -871,10 +878,9 @@ def test_from_handle_host_only_mr_without_cuda_init(mr_cls):
 
 @pytest.mark.thread_unsafe(reason="records process-global warnings and mutates the context stack")
 @pytest.mark.agent_authored(model="gpt-5.6")
-def test_mr_deallocation_failure_warns():
+def test_mr_deallocation_failure_warns(init_cuda):
     """Destructor-path MR failures are contained and reported as CUDAWarning."""
-    device = Device()
-    device.set_current()
+    device = init_cuda
     FailingMR, _ = make_instrumented_memory_resource(deallocate_error=RuntimeError("expected deallocation failure"))
     buf = Buffer.from_handle(1, 1024, mr=FailingMR(device))
 
@@ -887,7 +893,7 @@ def test_mr_deallocation_failure_warns():
 
 @pytest.mark.thread_unsafe(reason="records process-global warnings and mutates the context stack")
 @pytest.mark.agent_authored(model="claude-fable-5-1")
-def test_mr_deallocation_during_exception_keeps_the_exception():
+def test_mr_deallocation_during_exception_keeps_the_exception(init_cuda):
     """A Buffer released while an exception propagates leaves that exception in place.
 
     The deleter calls ``mr.deallocate`` on the Python thread. It must start
@@ -896,8 +902,7 @@ def test_mr_deallocation_during_exception_keeps_the_exception():
     exception, and the caller returned an error with no exception set, which
     Python reports as ``SystemError`` (found in review of #2917).
     """
-    device = Device()
-    device.set_current()
+    device = init_cuda
     FailingMR, _ = make_instrumented_memory_resource(deallocate_error=RuntimeError("expected deallocation failure"))
     mr = FailingMR(device)
 

From 47dfed47d59f496538cac0324d7cf348e642dcbd Mon Sep 17 00:00:00 2001
From: Ralf Juengling <rjuengling@nvidia.com>
Date: Thu, 8 Oct 2026 13:28:14 -0700
Subject: [PATCH 03/10] test(cuda.core): gate the init_cuda teardown
 gc.collect() on live MemoryResource count

Replace the always-on gc.collect() in the init_cuda teardown with a
runtime gate. MemoryResource keeps a process-wide atomic count of live
instances (incremented in __cinit__, decremented in __dealloc__) exposed
through a private accessor. init_cuda snapshots the count at setup and
collects at teardown only when the count has grown.

Also warn when a test leaves a MemoryResource alive past teardown
(UnreleasedMemoryResourceWarning) or releases one that was alive at
setup (BaselineMemoryResourceReleasedWarning), add tests for the
counter, and update tests/AGENTS.md.
---
 cuda_core/cuda/core/_memory/_buffer.pyi       |   4 +
 cuda_core/cuda/core/_memory/_buffer.pyx       |  16 ++-
 cuda_core/tests/AGENTS.md                     |  19 +--
 cuda_core/tests/conftest.py                   | 111 +++++++++++++++--
 cuda_core/tests/helpers/memory.py             |   8 ++
 .../tests/memory/test_resource_live_count.py  | 115 ++++++++++++++++++
 6 files changed, 255 insertions(+), 18 deletions(-)
 create mode 100644 cuda_core/tests/memory/test_resource_live_count.py

diff --git a/cuda_core/cuda/core/_memory/_buffer.pyi b/cuda_core/cuda/core/_memory/_buffer.pyi
index c2a03808968..3426f836861 100644
--- a/cuda_core/cuda/core/_memory/_buffer.pyi
+++ b/cuda_core/cuda/core/_memory/_buffer.pyi
@@ -301,6 +301,8 @@ class MemoryResource:
     buffer properties are retrieved simply by looking up the underlying memory
     resource's respective property.)
     """
+    def __init__(self): ...
+    def __dealloc__(self): ...
     def allocate(self, size: int, *, stream: Stream | GraphBuilder) -> Buffer:
         """Allocate a buffer of the requested size.
 
@@ -356,3 +358,5 @@ class _MemAttrs(TypedDict):
     is_device_accessible: bool
     is_host_accessible: bool
     is_managed: bool
+
+def _live_memory_resource_count() -> int: ...
diff --git a/cuda_core/cuda/core/_memory/_buffer.pyx b/cuda_core/cuda/core/_memory/_buffer.pyx
index b1ee356c245..d013be89d27 100644
--- a/cuda_core/cuda/core/_memory/_buffer.pyx
+++ b/cuda_core/cuda/core/_memory/_buffer.pyx
@@ -6,7 +6,7 @@ from __future__ import annotations
 
 cimport cython
 from libc.stdint cimport uintptr_t
-from libcpp.atomic cimport memory_order_acquire, memory_order_release
+from libcpp.atomic cimport memory_order_acquire, memory_order_release, atomic as std_atomic
 
 from cuda.bindings cimport cydriver
 from cuda.core._memory._device_memory_resource import DeviceMemoryResource
@@ -33,6 +33,14 @@ from cuda.core._memory._copy_attributes cimport _to_cu_memcpy_attributes  # no-c
 from cuda.core._stream cimport Stream, Stream_accept, Stream_is_legacy_default_token, default_stream
 from cuda.core._utils.cuda_utils cimport HANDLE_RETURN, _parse_fill_value
 
+
+# Live MemoryResource count, read by the test suite; atomic for free-threaded builds.
+cdef std_atomic[long long] _mr_live_count
+
+
+def _live_memory_resource_count() -> int:
+    return _mr_live_count.load()
+
 import warnings
 from collections.abc import Sequence
 from typing import TYPE_CHECKING
@@ -776,6 +784,12 @@ cdef class MemoryResource:
     resource's respective property.)
     """
 
+    def __cinit__(self):
+        _mr_live_count.fetch_add(1)
+
+    def __dealloc__(self):
+        _mr_live_count.fetch_sub(1)
+
     def allocate(self, size_t size, *, stream: Stream | GraphBuilder) -> Buffer:
         """Allocate a buffer of the requested size.
 
diff --git a/cuda_core/tests/AGENTS.md b/cuda_core/tests/AGENTS.md
index a8ff9eb9227..6cb1be6f732 100644
--- a/cuda_core/tests/AGENTS.md
+++ b/cuda_core/tests/AGENTS.md
@@ -60,13 +60,18 @@ with pytest.raises(RuntimeError, match="IPC is not available"):
 
 ## Release resources at test boundaries
 
-The `init_cuda` fixture in `conftest.py` runs `gc.collect()` followed
-by `cuCtxSynchronize()` before popping the context. Tests should not rely on
-that as a substitute for cleaning up explicitly: prefer context managers for
-resources whose lifetime fits a single scope, and keep pool lifetimes inside
-the test that creates them. The collected items and imported modules are
-frozen after collection, so the per-test collect covers only objects created
-during the run.
+The `init_cuda` fixture in `conftest.py` runs `gc.collect()` and
+`cuCtxSynchronize()` before popping the context. The collect runs only when the
+live `MemoryResource` count has changed since setup, so do not rely on it as a
+substitute for cleaning up explicitly: prefer context managers for resources
+whose lifetime fits a single scope, and keep pool lifetimes inside the test
+that creates them. The collected items and imported modules are frozen after
+collection, so the collect covers only objects created during the run.
+
+A resource that outlives a test without `init_cuda` makes the next `init_cuda`
+test emit `UnreleasedMemoryResourceWarning`. Release what you construct, and
+request `init_cuda` in any test that constructs one (see "Tests that touch CUDA
+must establish their own context").
 
 ## Shared test support
 
diff --git a/cuda_core/tests/conftest.py b/cuda_core/tests/conftest.py
index f7d0b5857e8..204070e5da0 100644
--- a/cuda_core/tests/conftest.py
+++ b/cuda_core/tests/conftest.py
@@ -8,6 +8,7 @@
 import os
 import pathlib
 import sys
+import warnings
 
 import pytest
 
@@ -29,7 +30,11 @@
 pytest_plugins = ["cuda_python_test_helpers._pytest_plugin"]
 
 from helpers.constants import POOL_SIZE
-from helpers.memory import skip_if_pinned_memory_unsupported
+from helpers.memory import (
+    BaselineMemoryResourceReleasedWarning,
+    UnreleasedMemoryResourceWarning,
+    skip_if_pinned_memory_unsupported,
+)
 
 import cuda.core
 from cuda.bindings import driver
@@ -43,6 +48,7 @@
     PinnedMemoryResourceOptions,
     _device,
 )
+from cuda.core._memory._buffer import _live_memory_resource_count
 from cuda.core._utils.cuda_utils import handle_return
 
 
@@ -125,6 +131,55 @@ def _item_uses_init_cuda(item):
     return "init_cuda" in getattr(item, "fixturenames", ())
 
 
+# init_cuda collects at teardown only if the process-wide live MemoryResource
+# count differs from its value at setup. That is sound only if no older
+# resource is released during a test (it could cancel a new leak), so tests
+# that construct resources must use init_cuda, and the leftover check below
+# reports resources that outlive a test without it.
+
+# Live count after the last init_cuda test; None until one has run.
+_mr_expected_live = None
+# Live count at the start of the current item, before its fixtures set up.
+_mr_live_at_item_start = 0
+_mr_last_init_cuda_nodeid = None
+
+
+def _device_default_memory_resource_count():
+    """Number of main-thread Devices that currently cache a default MR."""
+    return sum(d._memory_resource is not None for d in getattr(_device._tls, "devices", ()))
+
+
+def _release_device_default_memory_resources():
+    """Drop each main-thread Device's lazily created default memory resource.
+
+    Otherwise they are still counted when init_cuda compares counts at
+    teardown and force a collect in every test that allocates through the
+    device default. The setter rejects None,
+    hence the private access.
+    """
+    for d in getattr(_device._tls, "devices", ()):
+        d._memory_resource = None
+
+
+@pytest.hookimpl(wrapper=True)
+def pytest_runtest_protocol(item, nextitem):
+    # Leftover check: compare with the count after the previous init_cuda test,
+    # not the session start, because the count legitimately rises over a run
+    # (interned GraphMemoryResource, pytest-retained class instances).
+    global _mr_expected_live, _mr_live_at_item_start, _mr_last_init_cuda_nodeid
+    _mr_live_at_item_start = _live_memory_resource_count()
+    try:
+        return (yield)
+    finally:
+        live = _live_memory_resource_count()
+        if _item_uses_init_cuda(item):
+            _mr_expected_live = live
+            _mr_last_init_cuda_nodeid = item.nodeid
+        elif _mr_expected_live is not None:
+            # Only lower it: a released resource must not hide a later leak.
+            _mr_expected_live = min(_mr_expected_live, live)
+
+
 class _CudaCoreParallelPlugin:
     """A mini pytest plugin used only for pytest-run-parallel testing.
     pytest-run-parallel spawns new threads for each test and we need to
@@ -172,7 +227,7 @@ def session_setup():
 
 
 @pytest.fixture
-def init_cuda():
+def init_cuda(request):
     # TODO: rename this to e.g. init_context
     device = Device(0)
     device.set_current()
@@ -183,23 +238,59 @@ def init_cuda():
             driver.cuDevicePrimaryCtxSetFlags(device.device_id, driver.CUctx_flags.CU_CTX_SCHED_BLOCKING_SYNC)
         )
 
+    # Leftover check: growth since the previous init_cuda test means something
+    # in between left a resource alive. Collect first to sweep a cyclic
+    # leftover, and before the count at setup is recorded so it is reflected there.
+    if _mr_expected_live is not None and _mr_live_at_item_start > _mr_expected_live:
+        gc.collect()
+        held = _live_memory_resource_count() - _mr_expected_live
+        seed = getattr(request.config.option, "randomly_seed", None)
+        seed_msg = f" (pytest-randomly seed: {seed})" if seed is not None else ""
+        warnings.warn(
+            f"{_mr_live_at_item_start - _mr_expected_live} memory resource(s) were left alive by "
+            f"tests that ran since {_mr_last_init_cuda_nodeid} ({max(held, 0)} still alive after a "
+            f"collect). A test or fixture that constructs a memory resource must request "
+            f"init_cuda.{seed_msg}",
+            UnreleasedMemoryResourceWarning,
+            stacklevel=2,
+        )
+
+    # Live count at setup, excluding cached Device default resources (they are
+    # released at teardown before the count is compared).
+    live_at_setup = _live_memory_resource_count() - _device_default_memory_resource_count()
     try:
         yield device
     finally:
+        # pytest holds fixture values in item.funcargs until all teardown has
+        # run; drop them so already torn-down fixtures are not counted below.
+        request.node.funcargs.clear()
+        # Only a resource in a reference cycle outlives the test frame, so
+        # collect only if the count differs from its value at setup. A lower
+        # count means an older resource was released, which could mask a leak
+        # in this test: collect in that case too.
+        _release_device_default_memory_resources()
+        live_after_release = _live_memory_resource_count()
+        if live_after_release != live_at_setup:
+            gc.collect()
         # Force any pool/allocation whose only remaining reference was a local
         # in this test's frame to actually get destroyed now, then drain the
         # context so the stream-ordered frees that destruction enqueues retire
-        # before the next test runs. Without this, a memory pool's VA
-        # reservation is not returned until both have happened, and per-test
-        # leftovers accumulate across the run -- which is how full-suite runs
-        # can exhaust address space and hit CUDA_ERROR_OUT_OF_MEMORY on a
-        # device with plenty of free physical memory (issue #2381). gc.collect()
-        # must run first: cuCtxSynchronize alone cannot drain frees that were
-        # never enqueued because their owning object had not been collected yet.
+        # before the next test runs (issue #2381). gc.collect() above must run
+        # first: cuCtxSynchronize alone cannot drain frees that were never
+        # enqueued because their owning object had not been collected yet.
         # With pytest-run-parallel this runs after worker join.
-        gc.collect()
         driver.cuCtxSynchronize()
         _ = _device_unset_current()
+        # After the cleanup, so an escalated warning cannot skip it.
+        if live_after_release < live_at_setup:
+            warnings.warn(
+                f"{request.node.nodeid} released {live_at_setup - live_after_release} memory resource(s) "
+                f"that existed before it started (count {live_after_release}, snapshot {live_at_setup}). "
+                "A leak in this test could have been masked by that release. Find what held the "
+                "released resource.",
+                BaselineMemoryResourceReleasedWarning,
+                stacklevel=2,
+            )
 
 
 def _device_unset_current() -> bool:
diff --git a/cuda_core/tests/helpers/memory.py b/cuda_core/tests/helpers/memory.py
index da263ab81bc..5908314f62b 100644
--- a/cuda_core/tests/helpers/memory.py
+++ b/cuda_core/tests/helpers/memory.py
@@ -9,6 +9,14 @@
 from cuda.core._utils.cuda_utils import CUDAError
 
 
+class UnreleasedMemoryResourceWarning(Warning):
+    """A test or fixture left a ``MemoryResource`` alive without ``init_cuda``."""
+
+
+class BaselineMemoryResourceReleasedWarning(Warning):
+    """An ``init_cuda`` test released a resource that existed before it started."""
+
+
 def skip_if_pinned_memory_unsupported(device):
     try:
         if not device.properties.host_memory_pools_supported:
diff --git a/cuda_core/tests/memory/test_resource_live_count.py b/cuda_core/tests/memory/test_resource_live_count.py
new file mode 100644
index 00000000000..015c8947c2e
--- /dev/null
+++ b/cuda_core/tests/memory/test_resource_live_count.py
@@ -0,0 +1,115 @@
+# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+# SPDX-License-Identifier: Apache-2.0
+
+"""Tests for the process-wide live ``MemoryResource`` counter.
+
+``conftest.py`` uses the counter to decide whether the ``init_cuda`` teardown
+needs a ``gc.collect()`` and to detect resources left alive by other tests.
+Tests compare against a count read at their start, because resources retained
+from earlier tests make it nonzero. Each test requests ``init_cuda`` and frees
+what it constructs. They are ``thread_unsafe`` because the count is
+process-global.
+"""
+
+import gc
+import threading
+
+import pytest
+
+from cuda.core import DeviceMemoryResource, MemoryResource
+from cuda.core._memory._buffer import _live_memory_resource_count
+
+
+class _CountedMR(MemoryResource):
+    pass
+
+
+class _CountedMRWithArgs(MemoryResource):
+    def __init__(self, a, b):
+        self.a = a
+        self.b = b
+
+
+_THREADSAFE_MARKER = pytest.mark.thread_unsafe(reason="asserts on the process-global resource count")
+
+
+@_THREADSAFE_MARKER
+@pytest.mark.agent_authored(model="glm-5.2")
+def test_new_resource_is_counted_until_freed(init_cuda):
+    before = _live_memory_resource_count()
+    mr = _CountedMR()
+    assert _live_memory_resource_count() == before + 1
+    del mr
+    assert _live_memory_resource_count() == before
+
+
+@_THREADSAFE_MARKER
+@pytest.mark.agent_authored(model="glm-5.2")
+def test_resource_in_cycle_stays_counted_until_collect(init_cuda):
+    before = _live_memory_resource_count()
+    mr = _CountedMR()
+    mr.self_ref = mr  # create a reference cycle
+    del mr
+    # The cycle keeps the resource alive, so it is still counted.
+    assert _live_memory_resource_count() == before + 1
+    # Disable automatic collection so a background gc cannot free the cycle
+    # before the assertion is checked.
+    enabled = gc.isenabled()
+    gc.disable()
+    try:
+        gc.collect()
+        assert _live_memory_resource_count() == before
+    finally:
+        if enabled:
+            gc.enable()
+
+
+@_THREADSAFE_MARKER
+@pytest.mark.agent_authored(model="glm-5.2")
+def test_subclass_with_init_args_is_counted(init_cuda):
+    # Guards the no-argument base __cinit__: a Python subclass whose __init__
+    # takes extra arguments still constructs (and is counted) via the base.
+    before = _live_memory_resource_count()
+    mr = _CountedMRWithArgs(1, 2)
+    assert _live_memory_resource_count() == before + 1
+    assert mr.a == 1 and mr.b == 2
+    del mr
+    assert _live_memory_resource_count() == before
+
+
+@_THREADSAFE_MARKER
+@pytest.mark.agent_authored(model="glm-5.2")
+def test_cython_subclass_wrapper_is_counted(init_cuda):
+    # Covers a Cython subclass with its own __cinit__: a non-owning
+    # DeviceMemoryResource(device) wrapper (no options) wraps the default pool
+    # and is counted and uncounted like any resource.
+    before = _live_memory_resource_count()
+    mr = DeviceMemoryResource(init_cuda)
+    assert _live_memory_resource_count() == before + 1
+    del mr
+    assert _live_memory_resource_count() == before
+
+
+@_THREADSAFE_MARKER
+@pytest.mark.agent_authored(model="glm-5.2")
+def test_concurrent_construct_and_drop_returns_to_start(init_cuda):
+    # On a free-threaded build this guards against a non-atomic counter; on a
+    # GIL build it passes trivially. Every thread constructs and drops its own
+    # resources; after join the count is back at the starting value.
+    before = _live_memory_resource_count()
+    n_threads = 8
+    n_per_thread = 200
+
+    def worker():
+        for _ in range(n_per_thread):
+            mr = _CountedMR()
+            del mr
+
+    threads = [threading.Thread(target=worker) for _ in range(n_threads)]
+    for t in threads:
+        t.start()
+    for t in threads:
+        t.join()
+    # Let any deferred deallocation settle.
+    gc.collect()
+    assert _live_memory_resource_count() == before

From c4f065b2943aec4233179c9613484672cf3fbde2 Mon Sep 17 00:00:00 2001
From: Ralf Juengling <rjuengling@nvidia.com>
Date: Thu, 8 Oct 2026 14:10:30 -0700
Subject: [PATCH 04/10] address review feedback

---
 cuda_core/pytest.ini                          |   9 +
 cuda_core/tests/AGENTS.md                     |   3 +-
 cuda_core/tests/conftest.py                   |  12 +-
 .../memory/test_init_cuda_collect_gate.py     | 198 ++++++++++++++++++
 .../tests/memory/test_resource_live_count.py  |  12 +-
 cuda_core/tests/memory_ipc/test_errors.py     |   8 +-
 cuda_core/tests/test_error_handling.py        |   6 +-
 cuda_core/tests/test_memory.py                |   8 +-
 8 files changed, 235 insertions(+), 21 deletions(-)
 create mode 100644 cuda_core/tests/memory/test_init_cuda_collect_gate.py

diff --git a/cuda_core/pytest.ini b/cuda_core/pytest.ini
index 7c6ec1c3273..68867621bf4 100644
--- a/cuda_core/pytest.ini
+++ b/cuda_core/pytest.ini
@@ -6,6 +6,15 @@
 addopts = --showlocals --durations=20
 pythonpath = tests
 norecursedirs = cython
+filterwarnings =
+    # A test left a memory resource alive; see "Release resources at test boundaries" in tests/AGENTS.md.
+    # conftest.py makes helpers importable, which happens after pytest first
+    # applies these filters (and warns); the filter is applied again per test.
+    # The message regex below is pinned to pytest's wording (as of pytest 9.x);
+    # if a pytest upgrade rewords "Failed to import filter module" this stops
+    # suppressing the config warning (noisy, not fatal) and must be updated.
+    ignore:Failed to import filter module.*UnreleasedMemoryResourceWarning:pytest.PytestConfigWarning
+    error::helpers.memory.UnreleasedMemoryResourceWarning
 markers =
     samples: tests for cuda.core standalone samples
     # Keep this authorship marker registry in sync across all pytest config roots.
diff --git a/cuda_core/tests/AGENTS.md b/cuda_core/tests/AGENTS.md
index 6cb1be6f732..37380a29a4f 100644
--- a/cuda_core/tests/AGENTS.md
+++ b/cuda_core/tests/AGENTS.md
@@ -69,7 +69,8 @@ that creates them. The collected items and imported modules are frozen after
 collection, so the collect covers only objects created during the run.
 
 A resource that outlives a test without `init_cuda` makes the next `init_cuda`
-test emit `UnreleasedMemoryResourceWarning`. Release what you construct, and
+test fail with `UnreleasedMemoryResourceWarning` (escalated to an error in
+`pytest.ini`). Release what you construct, and
 request `init_cuda` in any test that constructs one (see "Tests that touch CUDA
 must establish their own context").
 
diff --git a/cuda_core/tests/conftest.py b/cuda_core/tests/conftest.py
index 204070e5da0..62cd076626b 100644
--- a/cuda_core/tests/conftest.py
+++ b/cuda_core/tests/conftest.py
@@ -145,7 +145,11 @@ def _item_uses_init_cuda(item):
 
 
 def _device_default_memory_resource_count():
-    """Number of main-thread Devices that currently cache a default MR."""
+    """Number of main-thread Devices that currently cache a default MR.
+
+    Reaches into Device thread-local state (``_device._tls``) on purpose; update
+    this and :func:`_release_device_default_memory_resources` if that moves.
+    """
     return sum(d._memory_resource is not None for d in getattr(_device._tls, "devices", ()))
 
 
@@ -261,8 +265,10 @@ def init_cuda(request):
     try:
         yield device
     finally:
-        # pytest holds fixture values in item.funcargs until all teardown has
-        # run; drop them so already torn-down fixtures are not counted below.
+        # pytest holds fixture values in item.funcargs (a pytest-internal dict)
+        # until all teardown has run; we deliberately drop them so already
+        # torn-down fixtures are not counted below. Revisit if pytest changes
+        # how it retains fixture values.
         request.node.funcargs.clear()
         # Only a resource in a reference cycle outlives the test frame, so
         # collect only if the count differs from its value at setup. A lower
diff --git a/cuda_core/tests/memory/test_init_cuda_collect_gate.py b/cuda_core/tests/memory/test_init_cuda_collect_gate.py
new file mode 100644
index 00000000000..160a94f2f29
--- /dev/null
+++ b/cuda_core/tests/memory/test_init_cuda_collect_gate.py
@@ -0,0 +1,198 @@
+# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+# SPDX-License-Identifier: Apache-2.0
+
+"""Tests for the ``init_cuda`` teardown gc.collect() gate and its warnings.
+
+``conftest.py`` snapshots the live ``MemoryResource`` count at ``init_cuda``
+setup and, at teardown, collects only when the count differs from that
+snapshot. It also warns when a test leaves a resource alive without
+``init_cuda`` (``UnreleasedMemoryResourceWarning``, raised at setup) or releases
+a resource that existed before it started (``BaselineMemoryResourceReleasedWarning``,
+raised at teardown).
+
+These behaviors live in the fixture's setup/teardown, so they cannot be observed
+from an ordinary test body. Each test here drives the *real* ``init_cuda``
+generator by hand with a minimal fake ``request`` and (where needed) a
+``gc.collect`` spy, so the collect decision and both warnings can be asserted
+directly. The fixture function and its module globals are reached through
+pytest's fixture registry (``FixtureDef.func``) so the monkeypatched globals are
+the ones the running code reads. Tests are ``thread_unsafe`` because they patch
+process-global state (``gc.collect`` and the conftest module globals) and assert
+on the process-global resource count; each cleans up every resource it creates
+so the real session's leftover tracking is not perturbed.
+"""
+
+import gc
+import inspect
+
+import pytest
+from helpers.memory import (
+    BaselineMemoryResourceReleasedWarning,
+    UnreleasedMemoryResourceWarning,
+)
+
+from cuda.core import MemoryResource
+from cuda.core._memory._buffer import _live_memory_resource_count
+
+
+class _LocalMR(MemoryResource):
+    """A trivial MemoryResource subclass that holds no CUDA pool.
+
+    Construction/destruction only moves the live counter, so it is a clean way
+    to make the gate's snapshot comparison grow or shrink without allocating.
+    """
+
+
+_THREAD_UNSAFE_MARKER = pytest.mark.thread_unsafe(
+    reason="patches gc.collect and conftest module globals, and asserts on the process-global resource count"
+)
+
+
+class _FakeOption:
+    randomly_seed = None
+
+
+class _FakeConfig:
+    def __init__(self):
+        self.option = _FakeOption()
+
+
+class _FakeNode:
+    def __init__(self, nodeid):
+        self.nodeid = nodeid
+        self.funcargs = {}
+
+
+class _FakeRequest:
+    """Minimal stand-in for the pytest request that ``init_cuda`` reads."""
+
+    def __init__(self, nodeid="manual::init_cuda_gate_probe"):
+        self.node = _FakeNode(nodeid)
+        self.config = _FakeConfig()
+
+
+def _init_cuda_func(request):
+    """Return the raw ``init_cuda`` generator function from the fixture registry.
+
+    ``FixtureDef.func`` is the undecorated fixture function, and its
+    ``__globals__`` is the live conftest namespace, so monkeypatching keys there
+    affects the code these tests drive. Both are pytest internals; the asserts
+    below fail loudly if a pytest change breaks the assumption.
+    """
+    fixturedefs = request._fixturemanager.getfixturedefs("init_cuda", request.node)
+    assert fixturedefs, "init_cuda fixture not found in the registry"
+    func = fixturedefs[-1].func
+    assert inspect.isgeneratorfunction(func), "init_cuda is expected to be a generator fixture"
+    return func
+
+
+def _spy_on_gc_collect(monkeypatch):
+    """Patch ``gc.collect`` to record each explicit call; returns the call list.
+
+    Automatic generational collections do not route through ``gc.collect``, so
+    the list counts only the fixture's explicit collects.
+    """
+    calls = []
+    real_collect = gc.collect
+
+    def spy(*args, **kwargs):
+        calls.append(1)
+        return real_collect(*args, **kwargs)
+
+    monkeypatch.setattr(gc, "collect", spy)
+    return calls
+
+
+# --- 2. the collect gate -------------------------------------------------------
+
+
+@_THREAD_UNSAFE_MARKER
+@pytest.mark.agent_authored(model="claude-opus-4.8")
+def test_teardown_skips_collect_when_mr_count_unchanged(request, monkeypatch):
+    func = _init_cuda_func(request)
+    # Disable the leftover check so only the teardown gate can call gc.collect.
+    monkeypatch.setitem(func.__globals__, "_mr_expected_live", None)
+    collects = _spy_on_gc_collect(monkeypatch)
+
+    gen = func(_FakeRequest())
+    next(gen)  # setup: records the snapshot
+    before = len(collects)
+    # A resource created and freed by refcount leaves the count at the snapshot.
+    mr = _LocalMR()
+    del mr
+    with pytest.raises(StopIteration):
+        next(gen)  # teardown runs the gate
+
+    assert len(collects) == before, "gate collected despite an unchanged count"
+
+
+@_THREAD_UNSAFE_MARKER
+@pytest.mark.agent_authored(model="claude-opus-4.8")
+def test_teardown_collects_when_mr_count_grows(request, monkeypatch):
+    func = _init_cuda_func(request)
+    monkeypatch.setitem(func.__globals__, "_mr_expected_live", None)
+    collects = _spy_on_gc_collect(monkeypatch)
+
+    gen = func(_FakeRequest())
+    next(gen)  # setup: records the snapshot
+    before = len(collects)
+    # A resource still alive at teardown lifts the count above the snapshot.
+    mr = _LocalMR()
+    try:
+        with pytest.raises(StopIteration):
+            next(gen)  # teardown runs the gate
+        assert len(collects) == before + 1, "gate skipped the collect despite a grown count"
+    finally:
+        # Restore the baseline so the session's leftover tracking is unperturbed.
+        del mr
+        gc.collect()
+
+
+# --- 3. the two warnings -------------------------------------------------------
+
+
+@_THREAD_UNSAFE_MARKER
+@pytest.mark.agent_authored(model="claude-opus-4.8")
+def test_setup_warns_when_resources_left_alive_since_last_init_cuda(request, monkeypatch):
+    func = _init_cuda_func(request)
+    g = func.__globals__
+    # Force the leftover-check precondition: the count at item start exceeds the
+    # baseline recorded after the previous init_cuda test, i.e. something in
+    # between left a resource alive. Keep a real resource referenced so the
+    # "still alive after a collect" count is nonzero.
+    base = _live_memory_resource_count()
+    leaked = _LocalMR()
+    monkeypatch.setitem(g, "_mr_expected_live", base)
+    monkeypatch.setitem(g, "_mr_live_at_item_start", base + 1)
+    monkeypatch.setitem(g, "_mr_last_init_cuda_nodeid", "tests/prev::test_prev")
+
+    gen = func(_FakeRequest())
+    try:
+        with pytest.warns(UnreleasedMemoryResourceWarning, match="left alive by"):
+            next(gen)  # setup emits the leftover warning
+        with pytest.raises(StopIteration):
+            next(gen)  # finish teardown
+    finally:
+        del leaked
+        gc.collect()
+
+
+@_THREAD_UNSAFE_MARKER
+@pytest.mark.agent_authored(model="claude-opus-4.8")
+def test_teardown_warns_on_baseline_release(request, monkeypatch):
+    func = _init_cuda_func(request)
+    # Disable the leftover check; this test exercises the teardown-side warning.
+    monkeypatch.setitem(func.__globals__, "_mr_expected_live", None)
+
+    # A resource that exists before setup is part of the snapshot; releasing it
+    # during the "test" drops the count below the snapshot at teardown.
+    baseline_mr = _LocalMR()
+    gen = func(_FakeRequest())
+    next(gen)  # setup snapshots the count, including baseline_mr
+    del baseline_mr  # release the pre-existing resource mid-"test"
+
+    with (
+        pytest.warns(BaselineMemoryResourceReleasedWarning, match="existed before it started"),
+        pytest.raises(StopIteration),
+    ):
+        next(gen)  # teardown sees a count below the snapshot
diff --git a/cuda_core/tests/memory/test_resource_live_count.py b/cuda_core/tests/memory/test_resource_live_count.py
index 015c8947c2e..9dbe2788cc0 100644
--- a/cuda_core/tests/memory/test_resource_live_count.py
+++ b/cuda_core/tests/memory/test_resource_live_count.py
@@ -30,10 +30,10 @@ def __init__(self, a, b):
         self.b = b
 
 
-_THREADSAFE_MARKER = pytest.mark.thread_unsafe(reason="asserts on the process-global resource count")
+_THREAD_UNSAFE_MARKER = pytest.mark.thread_unsafe(reason="asserts on the process-global resource count")
 
 
-@_THREADSAFE_MARKER
+@_THREAD_UNSAFE_MARKER
 @pytest.mark.agent_authored(model="glm-5.2")
 def test_new_resource_is_counted_until_freed(init_cuda):
     before = _live_memory_resource_count()
@@ -43,7 +43,7 @@ def test_new_resource_is_counted_until_freed(init_cuda):
     assert _live_memory_resource_count() == before
 
 
-@_THREADSAFE_MARKER
+@_THREAD_UNSAFE_MARKER
 @pytest.mark.agent_authored(model="glm-5.2")
 def test_resource_in_cycle_stays_counted_until_collect(init_cuda):
     before = _live_memory_resource_count()
@@ -64,7 +64,7 @@ def test_resource_in_cycle_stays_counted_until_collect(init_cuda):
             gc.enable()
 
 
-@_THREADSAFE_MARKER
+@_THREAD_UNSAFE_MARKER
 @pytest.mark.agent_authored(model="glm-5.2")
 def test_subclass_with_init_args_is_counted(init_cuda):
     # Guards the no-argument base __cinit__: a Python subclass whose __init__
@@ -77,7 +77,7 @@ def test_subclass_with_init_args_is_counted(init_cuda):
     assert _live_memory_resource_count() == before
 
 
-@_THREADSAFE_MARKER
+@_THREAD_UNSAFE_MARKER
 @pytest.mark.agent_authored(model="glm-5.2")
 def test_cython_subclass_wrapper_is_counted(init_cuda):
     # Covers a Cython subclass with its own __cinit__: a non-owning
@@ -90,7 +90,7 @@ def test_cython_subclass_wrapper_is_counted(init_cuda):
     assert _live_memory_resource_count() == before
 
 
-@_THREADSAFE_MARKER
+@_THREAD_UNSAFE_MARKER
 @pytest.mark.agent_authored(model="glm-5.2")
 def test_concurrent_construct_and_drop_returns_to_start(init_cuda):
     # On a free-threaded build this guards against a non-atomic counter; on a
diff --git a/cuda_core/tests/memory_ipc/test_errors.py b/cuda_core/tests/memory_ipc/test_errors.py
index aceb00332ab..9d1a3fef591 100644
--- a/cuda_core/tests/memory_ipc/test_errors.py
+++ b/cuda_core/tests/memory_ipc/test_errors.py
@@ -162,10 +162,10 @@ def test_main(self, ipc_device, ipc_memory_resource):
         finally:
             for mr in self._extra_mrs:
                 mr.close()
-            # mr-baseline-retention-fix (H1): drop the fixture MR and buffer
-            # references stored on this class instance, pytest retains
-            # ``item._instance`` for the session, so the closed (but not
-            # destroyed) MR objects would otherwise ratchet the baseline.
+            # Drop the fixture MR and buffer references stored on this class
+            # instance: pytest retains ``item._instance`` for the session, so the
+            # closed (but not destroyed) MR objects would otherwise ratchet the
+            # live-MemoryResource baseline the init_cuda teardown compares against.
             self.__dict__.pop("buffer", None)
             self.__dict__.pop("mr", None)
             self.__dict__.pop("device", None)
diff --git a/cuda_core/tests/test_error_handling.py b/cuda_core/tests/test_error_handling.py
index 56ebc764da7..48fbc9d37ce 100644
--- a/cuda_core/tests/test_error_handling.py
+++ b/cuda_core/tests/test_error_handling.py
@@ -230,9 +230,9 @@ def test_escalated_cudawarning_from_cleanup_is_not_a_crash(mempool_device):
     assert len(unraisable) == 1
     assert issubclass(unraisable[0].exc_type, CUDAWarning)
     assert "restoring the caller's context" in str(unraisable[0].exc_value)
-    # mr-baseline-retention-fix (H3): the unraisable hook args hold
-    # exc_traceback whose frame holds ``mr`` (never closed). Clear the
-    # list and drop the locals so gc can reclaim mr; close the pool too.
+    # The unraisable hook args hold exc_traceback whose frame holds ``mr``
+    # (never closed). Clear the list and drop the locals so gc can reclaim mr;
+    # close the pool too.
     unraisable.clear()
     mr.close()
     del mr, buf
diff --git a/cuda_core/tests/test_memory.py b/cuda_core/tests/test_memory.py
index 9a5c0738e5b..7019bed6c8b 100644
--- a/cuda_core/tests/test_memory.py
+++ b/cuda_core/tests/test_memory.py
@@ -337,10 +337,10 @@ def fill_env(request, init_cuda):
     try:
         yield device, mr
     finally:
-        # mr-baseline-retention-fix (H4): drop the fixture's MR
-        # reference so the (closed-by-dealloc) object does not persist
-        # in the pytest fixture cache past this teardown; without this the
-        # baseline ratchets/oscillates with fragile release timing.
+        # Defensive: drop the fixture-local MR reference promptly so the
+        # closed-by-dealloc object does not linger in this generator's frame.
+        # init_cuda's teardown also clears item.funcargs, so the test's own
+        # reference is released there regardless.
         del mr
 
 

From f50ec59a6d3fef3ecc7e8da1bd1e31f5f0429691 Mon Sep 17 00:00:00 2001
From: Ralf Juengling <rjuengling@nvidia.com>
Date: Thu, 8 Oct 2026 14:56:08 -0700
Subject: [PATCH 05/10] address TCC failure

---
 .../tests/memory/test_resource_live_count.py  |  6 +++-
 cuda_core/tests/test_green_context.py         | 30 +++++++++++++------
 2 files changed, 26 insertions(+), 10 deletions(-)

diff --git a/cuda_core/tests/memory/test_resource_live_count.py b/cuda_core/tests/memory/test_resource_live_count.py
index 9dbe2788cc0..ef13bc4e6b3 100644
--- a/cuda_core/tests/memory/test_resource_live_count.py
+++ b/cuda_core/tests/memory/test_resource_live_count.py
@@ -82,7 +82,11 @@ def test_subclass_with_init_args_is_counted(init_cuda):
 def test_cython_subclass_wrapper_is_counted(init_cuda):
     # Covers a Cython subclass with its own __cinit__: a non-owning
     # DeviceMemoryResource(device) wrapper (no options) wraps the default pool
-    # and is counted and uncounted like any resource.
+    # and is counted and uncounted like any resource. The no-options path
+    # needs the current device's memory pool, so skip where it is unsupported
+    # (e.g. Windows TCC); the counter behavior itself is platform-independent.
+    if not init_cuda.properties.memory_pools_supported:
+        pytest.skip("Device does not support memory pools")
     before = _live_memory_resource_count()
     mr = DeviceMemoryResource(init_cuda)
     assert _live_memory_resource_count() == before + 1
diff --git a/cuda_core/tests/test_green_context.py b/cuda_core/tests/test_green_context.py
index 773e0f4f42a..05946432c55 100644
--- a/cuda_core/tests/test_green_context.py
+++ b/cuda_core/tests/test_green_context.py
@@ -180,6 +180,9 @@ def test_memory_node_updates_preserve_green_context(
     memcpy_node.destroy()
     src.close()
     dst.close()
+    # Drop the pinned MR so it frees by refcount now rather than lingering
+    # for gc (see "Release resources at test boundaries" in tests/AGENTS.md).
+    del memory_resource
 
 
 # ---------------------------------------------------------------------------
@@ -641,19 +644,28 @@ def test_primary_context_stream_sm_resources(self, init_cuda, sm_resource):
 def _launch_fill_and_verify(dev, stream, kernel, n, value):
     """Launch the fill kernel and verify results on host."""
     dev_buf = dev.allocate(n * np.dtype(np.int32).itemsize, stream=stream)
-
-    config = LaunchConfig(grid=(n + 31) // 32, block=32)
-    launch(stream, config, kernel, dev_buf, np.int32(value), np.int32(n))
-
     host_mr = LegacyPinnedMemoryResource()
     host_buf = host_mr.allocate(n * np.dtype(np.int32).itemsize)
-    host_arr = np.from_dlpack(host_buf).view(np.int32)
-    host_arr[:] = 0
+    host_arr = None
+    try:
+        config = LaunchConfig(grid=(n + 31) // 32, block=32)
+        launch(stream, config, kernel, dev_buf, np.int32(value), np.int32(n))
 
-    dev_buf.copy_to(host_buf, stream=stream)
-    stream.sync()
+        host_arr = np.from_dlpack(host_buf).view(np.int32)
+        host_arr[:] = 0
+
+        dev_buf.copy_to(host_buf, stream=stream)
+        stream.sync()
 
-    np.testing.assert_array_equal(host_arr, np.full(n, value, dtype=np.int32))
+        np.testing.assert_array_equal(host_arr, np.full(n, value, dtype=np.int32))
+    finally:
+        # Release the DLPack view and close both buffers so the pinned MR
+        # frees by refcount at return instead of lingering in a cycle for gc
+        # (see "Release resources at test boundaries" in tests/AGENTS.md).
+        del host_arr
+        host_buf.close(stream)
+        dev_buf.close(stream)
+        del host_mr
 
 
 class TestGreenContextKernelLaunch:

From b102d260ba4e19344f3b60965a1d488ea892b427 Mon Sep 17 00:00:00 2001
From: Ralf Juengling <rjuengling@nvidia.com>
Date: Thu, 8 Oct 2026 15:39:01 -0700
Subject: [PATCH 06/10] test(cuda.core): request init_cuda for test_latchkernel

test_latchkernel constructed 3 DummyUnifiedMemoryResource instances (via
make_scratch_buffer) plus a LegacyPinnedMemoryResource (via LatchKernel)
without init_cuda. On Windows and aarch64 the ctypes-pointer/Buffer/MR
chains form cycles that survive refcounting, so the live-MemoryResource
count stays elevated past the test and the next init_cuda test
(test_oom_diagnostics_probe_basics_is_live_and_cheap) is blamed by the
leftover check (UnreleasedMemoryResourceWarning, escalated to an error).

Migrating to init_cuda lets the teardown gate collect the cyclic leftovers,
matching the pattern from 68f832c1d14 for the other MR-constructing tests.
---
 cuda_core/tests/test_helpers.py | 5 ++---
 1 file changed, 2 insertions(+), 3 deletions(-)

diff --git a/cuda_core/tests/test_helpers.py b/cuda_core/tests/test_helpers.py
index 35873da5231..5e27c89c353 100644
--- a/cuda_core/tests/test_helpers.py
+++ b/cuda_core/tests/test_helpers.py
@@ -33,12 +33,11 @@
 
 @pytest.mark.skipif(Device().compute_capability.major < 7, reason="__nanosleep is only available starting Volta (sm70)")
 @pytest.mark.thread_unsafe(reason="requires a barrier wait to avoid overlapping pinned latch allocations")
-def test_latchkernel():
+def test_latchkernel(init_cuda):
     """Test LatchKernel."""
     log = TimestampedLogger(enabled=ENABLE_LOGGING)
     log("begin")
-    device = Device()
-    device.set_current()
+    device = init_cuda
     stream = device.create_stream()
     target = make_scratch_buffer(device, 0, NBYTES)
     zeros = make_scratch_buffer(device, 0, NBYTES)

From 0651f45b0d338d8e05f51d01602b2f7b784fb3e6 Mon Sep 17 00:00:00 2001
From: Ralf Juengling <rjuengling@nvidia.com>
Date: Fri, 9 Oct 2026 14:06:31 -0700
Subject: [PATCH 07/10] test(cuda.core): count live owned pools and VA
 reservations in C++

Move the init_cuda teardown gc.collect() gate's instrumentation out of a
Python MemoryResource object counter and into the C++ RAII layer, split into
two process-wide atomic counts: owned memory pools (cuMemPoolCreate or
imported, not borrowed references) and VA reservations (cuMemAddressReserve).
These are the resources issue #2381 is about; borrowed pool references (the
device default pool) and MemoryResource subclasses that hold no pool or
reservation are no longer counted.

- Add std::atomic counts incremented at the owning RAII create sites
  (wrap_mempool_owned in memory.cpp, create_va_reservation_handle in
  virtual_memory.cpp) and decremented in their deleters, with
  live_owned_mempool_count / live_va_reservation_count accessors declared in
  api.hpp and bridged through _rt.pyx / _rt.pxd.
- _buffer.pyx: drop _mr_live_count and MemoryResource.__cinit__/__dealloc__;
  expose _live_owned_mempool_count / _live_va_reservation_count.
- conftest.py: snapshot a (owned_pools, va_reservations) tuple; collect when
  either component differs; warn per-component. Drop the device-default
  bookkeeping, since the default pool is a borrowed reference and is no longer
  counted.
- Rewrite the counter tests for the two counters (owned pool, borrowed pool
  not counted, cycle, VA reservation, concurrency); the gate tests script
  _live_resource_counts for deterministic skip/collect/warning coverage.
---
 cuda_core/cuda/core/_cpp/rt/api.hpp           |   9 ++
 cuda_core/cuda/core/_cpp/rt/memory.cpp        |  20 ++-
 .../cuda/core/_cpp/rt/virtual_memory.cpp      |  18 ++-
 cuda_core/cuda/core/_memory/_buffer.pyi       |   5 +-
 cuda_core/cuda/core/_memory/_buffer.pyx       |  22 +--
 cuda_core/cuda/core/_rt.pxd                   |   5 +
 cuda_core/cuda/core/_rt.pyx                   |   2 +
 cuda_core/tests/AGENTS.md                     |   3 +-
 cuda_core/tests/conftest.py                   |  94 +++++------
 .../memory/test_init_cuda_collect_gate.py     | 139 ++++++++--------
 .../tests/memory/test_resource_live_count.py  | 151 ++++++++++--------
 11 files changed, 265 insertions(+), 203 deletions(-)

diff --git a/cuda_core/cuda/core/_cpp/rt/api.hpp b/cuda_core/cuda/core/_cpp/rt/api.hpp
index d77e57c86e5..53c338f2eaf 100644
--- a/cuda_core/cuda/core/_cpp/rt/api.hpp
+++ b/cuda_core/cuda/core/_cpp/rt/api.hpp
@@ -270,6 +270,15 @@ VaReservationHandle create_va_reservation_handle(size_t size, size_t alignment,
 // Size of the reservation.
 size_t va_reservation_size(const VaReservationHandle& h) noexcept;
 
+// Test-suite instrumentation: process-wide counts of live *owned* driver
+// resources. Owned memory pools (cuMemPoolCreate or imported) and VA
+// reservations (cuMemAddressReserve) are each counted at their RAII create
+// site and uncounted in their deleter. Borrowed pool references and
+// MemoryResource subclasses that hold no pool or reservation are not counted.
+// conftest.py reads these to gate the per-test gc.collect().
+long long live_owned_mempool_count() noexcept;
+long long live_va_reservation_count() noexcept;
+
 // Map the whole allocation at ptr inside the reservation via cuMemMap and
 // apply the allocation's access descriptors. The mapping structurally depends
 // on both handles. When the last reference is released, cuMemUnmap is called
diff --git a/cuda_core/cuda/core/_cpp/rt/memory.cpp b/cuda_core/cuda/core/_cpp/rt/memory.cpp
index e53135924a6..47bdd89e387 100644
--- a/cuda_core/cuda/core/_cpp/rt/memory.cpp
+++ b/cuda_core/cuda/core/_cpp/rt/memory.cpp
@@ -8,6 +8,7 @@
 #include "driver_api.hpp"
 #include "error.hpp"
 #include "internal.hpp"
+#include <atomic>
 #include <cstddef>
 #include <cstring>
 #include <memory>
@@ -30,6 +31,13 @@ namespace {
 struct MemoryPoolBox {
     CUmemoryPool resource;
 };
+
+// Process-wide count of live *owned* memory pools -- those this layer created
+// with cuMemPoolCreate or imported (and will cuMemPoolDestroy). Borrowed pool
+// references (create_mempool_handle_ref, get_device_mempool) are not counted.
+// Read by the test suite to gate the per-test gc.collect(); atomic for
+// free-threaded builds.
+std::atomic<long long> g_live_owned_mempools{0};
 }  // namespace
 
 // Helper to clear peer access before destroying a memory pool.
@@ -65,18 +73,28 @@ static void clear_mempool_peer_access(CUmemoryPool pool, int owner_device) noexc
 // owner_device is the ordinal of a device-located pool, or -1 when there is no
 // device to skip (host pools) or the location is unknown (imported pools).
 static MemoryPoolHandle wrap_mempool_owned(CUmemoryPool pool, int owner_device) {
+    auto* raw = new MemoryPoolBox{pool};
+    // Count before constructing the shared_ptr: if its control-block allocation
+    // throws, it runs the deleter on raw, whose decrement then balances this.
+    g_live_owned_mempools.fetch_add(1, std::memory_order_relaxed);
     auto box = std::shared_ptr<const MemoryPoolBox>(
-        new MemoryPoolBox{pool},
+        raw,
         [owner_device](const MemoryPoolBox* b) {
             GILReleaseGuard gil;
             clear_mempool_peer_access(b->resource, owner_device);
             pw_cuMemPoolDestroy(b->resource);
+            g_live_owned_mempools.fetch_sub(1, std::memory_order_relaxed);
             delete b;
         }
     );
     return MemoryPoolHandle(box, &box->resource);
 }
 
+// Test-suite instrumentation; see g_live_owned_mempools.
+long long live_owned_mempool_count() noexcept {
+    return g_live_owned_mempools.load(std::memory_order_relaxed);
+}
+
 MemoryPoolHandle create_mempool_handle(const CUmemPoolProps& props) {
     GILReleaseGuard gil;
     CUmemoryPool pool;
diff --git a/cuda_core/cuda/core/_cpp/rt/virtual_memory.cpp b/cuda_core/cuda/core/_cpp/rt/virtual_memory.cpp
index 8a946180794..5f34d706deb 100644
--- a/cuda_core/cuda/core/_cpp/rt/virtual_memory.cpp
+++ b/cuda_core/cuda/core/_cpp/rt/virtual_memory.cpp
@@ -12,6 +12,7 @@
 #include "driver_api.hpp"
 #include "error.hpp"
 #include "internal.hpp"
+#include <atomic>
 #include <cstddef>
 #include <memory>
 #include <utility>
@@ -38,6 +39,11 @@ struct VaReservationBox {
     size_t size;                          // exact reserved size; cuMemAddressFree needs both
 };
 
+// Process-wide count of live VA reservations (cuMemAddressReserve), the leak
+// quantity issue #2381 is about; read by the test suite to gate the per-test
+// gc.collect(). Atomic for free-threaded builds.
+std::atomic<long long> g_live_va_reservations{0};
+
 struct VaMappingBox {
     VaMappingValue resource;              // mapped address
     size_t size;                          // == the allocation's size
@@ -93,17 +99,27 @@ VaReservationHandle create_va_reservation_handle(size_t size, size_t alignment,
     if (CUDA_SUCCESS != (err = DRIVER_CALL(cuMemAddressReserve, &ptr, size, alignment, hint, 0))) {
         return {};
     }
+    auto* raw = new VaReservationBox{{ptr}, size};
+    // Count before constructing the shared_ptr: if its control-block allocation
+    // throws, it runs the deleter on raw, whose decrement then balances this.
+    g_live_va_reservations.fetch_add(1, std::memory_order_relaxed);
     auto box = std::shared_ptr<const VaReservationBox>(
-        new VaReservationBox{{ptr}, size},
+        raw,
         [](const VaReservationBox* b) {
             GILReleaseGuard gil;
             pw_cuMemAddressFree(b->resource.raw, b->size);
+            g_live_va_reservations.fetch_sub(1, std::memory_order_relaxed);
             delete b;
         }
     );
     return VaReservationHandle(box, &box->resource);
 }
 
+// Test-suite instrumentation; see g_live_va_reservations.
+long long live_va_reservation_count() noexcept {
+    return g_live_va_reservations.load(std::memory_order_relaxed);
+}
+
 size_t va_reservation_size(const VaReservationHandle& h) noexcept {
     return h ? box_of<VaReservationBox>(h)->size : 0;
 }
diff --git a/cuda_core/cuda/core/_memory/_buffer.pyi b/cuda_core/cuda/core/_memory/_buffer.pyi
index 3426f836861..7cf76b05c6e 100644
--- a/cuda_core/cuda/core/_memory/_buffer.pyi
+++ b/cuda_core/cuda/core/_memory/_buffer.pyi
@@ -301,8 +301,6 @@ class MemoryResource:
     buffer properties are retrieved simply by looking up the underlying memory
     resource's respective property.)
     """
-    def __init__(self): ...
-    def __dealloc__(self): ...
     def allocate(self, size: int, *, stream: Stream | GraphBuilder) -> Buffer:
         """Allocate a buffer of the requested size.
 
@@ -359,4 +357,5 @@ class _MemAttrs(TypedDict):
     is_host_accessible: bool
     is_managed: bool
 
-def _live_memory_resource_count() -> int: ...
+def _live_owned_mempool_count() -> int: ...
+def _live_va_reservation_count() -> int: ...
diff --git a/cuda_core/cuda/core/_memory/_buffer.pyx b/cuda_core/cuda/core/_memory/_buffer.pyx
index d013be89d27..9a1ab411bdc 100644
--- a/cuda_core/cuda/core/_memory/_buffer.pyx
+++ b/cuda_core/cuda/core/_memory/_buffer.pyx
@@ -6,7 +6,7 @@ from __future__ import annotations
 
 cimport cython
 from libc.stdint cimport uintptr_t
-from libcpp.atomic cimport memory_order_acquire, memory_order_release, atomic as std_atomic
+from libcpp.atomic cimport memory_order_acquire, memory_order_release
 
 from cuda.bindings cimport cydriver
 from cuda.core._memory._device_memory_resource import DeviceMemoryResource
@@ -20,6 +20,8 @@ from cuda.core._rt cimport (
     deviceptr_create_with_owner,
     deviceptr_create_with_mr,
     register_mr_dealloc_callback,
+    live_owned_mempool_count,
+    live_va_reservation_count,
     as_intptr,
     as_cu,
     get_current_context,
@@ -34,12 +36,16 @@ from cuda.core._stream cimport Stream, Stream_accept, Stream_is_legacy_default_t
 from cuda.core._utils.cuda_utils cimport HANDLE_RETURN, _parse_fill_value
 
 
-# Live MemoryResource count, read by the test suite; atomic for free-threaded builds.
-cdef std_atomic[long long] _mr_live_count
+# Process-wide counts of live *owned* driver resources, read by the test suite.
+# Counting lives in the C++ RAII layer (owned CUmemoryPool and VA reservations),
+# so borrowed pool references and MemoryResource subclasses that hold no pool or
+# reservation are excluded. conftest.py reads these to gate the per-test collect.
+def _live_owned_mempool_count() -> int:
+    return live_owned_mempool_count()
 
 
-def _live_memory_resource_count() -> int:
-    return _mr_live_count.load()
+def _live_va_reservation_count() -> int:
+    return live_va_reservation_count()
 
 import warnings
 from collections.abc import Sequence
@@ -784,12 +790,6 @@ cdef class MemoryResource:
     resource's respective property.)
     """
 
-    def __cinit__(self):
-        _mr_live_count.fetch_add(1)
-
-    def __dealloc__(self):
-        _mr_live_count.fetch_sub(1)
-
     def allocate(self, size_t size, *, stream: Stream | GraphBuilder) -> Buffer:
         """Allocate a buffer of the requested size.
 
diff --git a/cuda_core/cuda/core/_rt.pxd b/cuda_core/cuda/core/_rt.pxd
index 30709187f38..96071d67a0f 100644
--- a/cuda_core/cuda/core/_rt.pxd
+++ b/cuda_core/cuda/core/_rt.pxd
@@ -302,6 +302,11 @@ cdef size_t mem_allocation_size(const MemAllocationHandle& h) noexcept nogil
 cdef VaReservationHandle create_va_reservation_handle(
     size_t size, size_t alignment, cydriver.CUdeviceptr hint) except+ nogil
 cdef size_t va_reservation_size(const VaReservationHandle& h) noexcept nogil
+
+# Test-suite instrumentation: process-wide counts of live owned pools and VA
+# reservations (see api.hpp). Read by conftest.py to gate the per-test collect.
+cdef long long live_owned_mempool_count() noexcept nogil
+cdef long long live_va_reservation_count() noexcept nogil
 cdef VaMappingHandle create_va_mapping_handle(
     cydriver.CUdeviceptr ptr, const MemAllocationHandle& h_alloc,
     const VaReservationHandle& h_res) except+ nogil
diff --git a/cuda_core/cuda/core/_rt.pyx b/cuda_core/cuda/core/_rt.pyx
index 8eef0d5877c..47318ba83ae 100644
--- a/cuda_core/cuda/core/_rt.pyx
+++ b/cuda_core/cuda/core/_rt.pyx
@@ -184,6 +184,8 @@ cdef extern from "_cpp/rt/rt.hpp" namespace "cuda_core::rt":
         size_t size, size_t alignment, cydriver.CUdeviceptr hint) except+ nogil
     size_t va_reservation_size "cuda_core::rt::va_reservation_size" (
         const VaReservationHandle& h) noexcept nogil
+    long long live_owned_mempool_count "cuda_core::rt::live_owned_mempool_count" () noexcept nogil
+    long long live_va_reservation_count "cuda_core::rt::live_va_reservation_count" () noexcept nogil
     VaMappingHandle create_va_mapping_handle "cuda_core::rt::create_va_mapping_handle" (
         cydriver.CUdeviceptr ptr, const MemAllocationHandle& h_alloc,
         const VaReservationHandle& h_res) except+ nogil
diff --git a/cuda_core/tests/AGENTS.md b/cuda_core/tests/AGENTS.md
index 37380a29a4f..daac7556de6 100644
--- a/cuda_core/tests/AGENTS.md
+++ b/cuda_core/tests/AGENTS.md
@@ -62,7 +62,8 @@ with pytest.raises(RuntimeError, match="IPC is not available"):
 
 The `init_cuda` fixture in `conftest.py` runs `gc.collect()` and
 `cuCtxSynchronize()` before popping the context. The collect runs only when the
-live `MemoryResource` count has changed since setup, so do not rely on it as a
+count of live owned driver resources (owned memory pools or VA reservations,
+tracked in the C++ layer) has changed since setup, so do not rely on it as a
 substitute for cleaning up explicitly: prefer context managers for resources
 whose lifetime fits a single scope, and keep pool lifetimes inside the test
 that creates them. The collected items and imported modules are frozen after
diff --git a/cuda_core/tests/conftest.py b/cuda_core/tests/conftest.py
index 62cd076626b..3ece2288693 100644
--- a/cuda_core/tests/conftest.py
+++ b/cuda_core/tests/conftest.py
@@ -48,7 +48,7 @@
     PinnedMemoryResourceOptions,
     _device,
 )
-from cuda.core._memory._buffer import _live_memory_resource_count
+from cuda.core._memory._buffer import _live_owned_mempool_count, _live_va_reservation_count
 from cuda.core._utils.cuda_utils import handle_return
 
 
@@ -131,38 +131,30 @@ def _item_uses_init_cuda(item):
     return "init_cuda" in getattr(item, "fixturenames", ())
 
 
-# init_cuda collects at teardown only if the process-wide live MemoryResource
-# count differs from its value at setup. That is sound only if no older
-# resource is released during a test (it could cancel a new leak), so tests
-# that construct resources must use init_cuda, and the leftover check below
-# reports resources that outlive a test without it.
+# init_cuda collects at teardown only if the process-wide count of live *owned*
+# driver resources (owned memory pools and VA reservations) differs from its
+# value at setup. The counts come from the C++ RAII layer, so a borrowed pool
+# reference (the device default pool) and MemoryResource subclasses that hold no
+# pool or reservation do not move them. That makes the gate sound only if no
+# older owned resource is released during a test (it could cancel a new leak),
+# so tests that construct owned resources must use init_cuda, and the leftover
+# check below reports owned resources that outlive a test without it.
 
-# Live count after the last init_cuda test; None until one has run.
-_mr_expected_live = None
-# Live count at the start of the current item, before its fixtures set up.
-_mr_live_at_item_start = 0
-_mr_last_init_cuda_nodeid = None
 
+def _live_resource_counts():
+    """(owned memory pools, VA reservations) live process-wide, as a tuple.
 
-def _device_default_memory_resource_count():
-    """Number of main-thread Devices that currently cache a default MR.
-
-    Reaches into Device thread-local state (``_device._tls``) on purpose; update
-    this and :func:`_release_device_default_memory_resources` if that moves.
+    Only *owned* driver resources are counted (see the C++ accessors), so this
+    excludes borrowed pool references and non-owning MemoryResource subclasses.
     """
-    return sum(d._memory_resource is not None for d in getattr(_device._tls, "devices", ()))
-
+    return (_live_owned_mempool_count(), _live_va_reservation_count())
 
-def _release_device_default_memory_resources():
-    """Drop each main-thread Device's lazily created default memory resource.
 
-    Otherwise they are still counted when init_cuda compares counts at
-    teardown and force a collect in every test that allocates through the
-    device default. The setter rejects None,
-    hence the private access.
-    """
-    for d in getattr(_device._tls, "devices", ()):
-        d._memory_resource = None
+# Counts after the last init_cuda test; None until one has run.
+_mr_expected_live = None
+# Counts at the start of the current item, before its fixtures set up.
+_mr_live_at_item_start = (0, 0)
+_mr_last_init_cuda_nodeid = None
 
 
 @pytest.hookimpl(wrapper=True)
@@ -171,17 +163,17 @@ def pytest_runtest_protocol(item, nextitem):
     # not the session start, because the count legitimately rises over a run
     # (interned GraphMemoryResource, pytest-retained class instances).
     global _mr_expected_live, _mr_live_at_item_start, _mr_last_init_cuda_nodeid
-    _mr_live_at_item_start = _live_memory_resource_count()
+    _mr_live_at_item_start = _live_resource_counts()
     try:
         return (yield)
     finally:
-        live = _live_memory_resource_count()
+        live = _live_resource_counts()
         if _item_uses_init_cuda(item):
             _mr_expected_live = live
             _mr_last_init_cuda_nodeid = item.nodeid
         elif _mr_expected_live is not None:
-            # Only lower it: a released resource must not hide a later leak.
-            _mr_expected_live = min(_mr_expected_live, live)
+            # Only lower each component: a released resource must not hide a later leak.
+            _mr_expected_live = tuple(min(a, b) for a, b in zip(_mr_expected_live, live))
 
 
 class _CudaCoreParallelPlugin:
@@ -243,25 +235,27 @@ def init_cuda(request):
         )
 
     # Leftover check: growth since the previous init_cuda test means something
-    # in between left a resource alive. Collect first to sweep a cyclic
-    # leftover, and before the count at setup is recorded so it is reflected there.
-    if _mr_expected_live is not None and _mr_live_at_item_start > _mr_expected_live:
+    # in between left an owned resource alive. Collect first to sweep a cyclic
+    # leftover, and before the counts at setup are recorded so they are reflected there.
+    if _mr_expected_live is not None and any(now > exp for now, exp in zip(_mr_live_at_item_start, _mr_expected_live)):
         gc.collect()
-        held = _live_memory_resource_count() - _mr_expected_live
+        now_counts = _live_resource_counts()
+        held = tuple(max(n - e, 0) for n, e in zip(now_counts, _mr_expected_live))
+        excess = tuple(s - e for s, e in zip(_mr_live_at_item_start, _mr_expected_live))
         seed = getattr(request.config.option, "randomly_seed", None)
         seed_msg = f" (pytest-randomly seed: {seed})" if seed is not None else ""
         warnings.warn(
-            f"{_mr_live_at_item_start - _mr_expected_live} memory resource(s) were left alive by "
-            f"tests that ran since {_mr_last_init_cuda_nodeid} ({max(held, 0)} still alive after a "
+            f"{excess} (owned_pools, va_reservations) memory resource(s) were left alive by "
+            f"tests that ran since {_mr_last_init_cuda_nodeid} ({held} still alive after a "
             f"collect). A test or fixture that constructs a memory resource must request "
             f"init_cuda.{seed_msg}",
             UnreleasedMemoryResourceWarning,
             stacklevel=2,
         )
 
-    # Live count at setup, excluding cached Device default resources (they are
-    # released at teardown before the count is compared).
-    live_at_setup = _live_memory_resource_count() - _device_default_memory_resource_count()
+    # Counts at setup. The device default pool is a borrowed reference and is
+    # not counted, so no cached-default bookkeeping is needed.
+    live_at_setup = _live_resource_counts()
     try:
         yield device
     finally:
@@ -270,13 +264,12 @@ def init_cuda(request):
         # torn-down fixtures are not counted below. Revisit if pytest changes
         # how it retains fixture values.
         request.node.funcargs.clear()
-        # Only a resource in a reference cycle outlives the test frame, so
-        # collect only if the count differs from its value at setup. A lower
+        # Only an owned resource in a reference cycle outlives the test frame, so
+        # collect only if the counts differ from their values at setup. A lower
         # count means an older resource was released, which could mask a leak
         # in this test: collect in that case too.
-        _release_device_default_memory_resources()
-        live_after_release = _live_memory_resource_count()
-        if live_after_release != live_at_setup:
+        live_after = _live_resource_counts()
+        if live_after != live_at_setup:
             gc.collect()
         # Force any pool/allocation whose only remaining reference was a local
         # in this test's frame to actually get destroyed now, then drain the
@@ -288,12 +281,13 @@ def init_cuda(request):
         driver.cuCtxSynchronize()
         _ = _device_unset_current()
         # After the cleanup, so an escalated warning cannot skip it.
-        if live_after_release < live_at_setup:
+        if any(now < snap for now, snap in zip(live_after, live_at_setup)):
+            released = tuple(s - n for s, n in zip(live_at_setup, live_after))
             warnings.warn(
-                f"{request.node.nodeid} released {live_at_setup - live_after_release} memory resource(s) "
-                f"that existed before it started (count {live_after_release}, snapshot {live_at_setup}). "
-                "A leak in this test could have been masked by that release. Find what held the "
-                "released resource.",
+                f"{request.node.nodeid} released {released} (owned_pools, va_reservations) memory "
+                f"resource(s) that existed before it started (counts {live_after}, snapshot "
+                f"{live_at_setup}). A leak in this test could have been masked by that release. "
+                "Find what held the released resource.",
                 BaselineMemoryResourceReleasedWarning,
                 stacklevel=2,
             )
diff --git a/cuda_core/tests/memory/test_init_cuda_collect_gate.py b/cuda_core/tests/memory/test_init_cuda_collect_gate.py
index 160a94f2f29..2a525525a29 100644
--- a/cuda_core/tests/memory/test_init_cuda_collect_gate.py
+++ b/cuda_core/tests/memory/test_init_cuda_collect_gate.py
@@ -3,23 +3,23 @@
 
 """Tests for the ``init_cuda`` teardown gc.collect() gate and its warnings.
 
-``conftest.py`` snapshots the live ``MemoryResource`` count at ``init_cuda``
-setup and, at teardown, collects only when the count differs from that
-snapshot. It also warns when a test leaves a resource alive without
-``init_cuda`` (``UnreleasedMemoryResourceWarning``, raised at setup) or releases
-a resource that existed before it started (``BaselineMemoryResourceReleasedWarning``,
-raised at teardown).
+``conftest.py`` snapshots the live owned-resource counts (owned memory pools and
+VA reservations) at ``init_cuda`` setup and, at teardown, collects only when a
+count differs from that snapshot. It also warns when a test leaves a resource
+alive without ``init_cuda`` (``UnreleasedMemoryResourceWarning``, raised at
+setup) or releases a resource that existed before it started
+(``BaselineMemoryResourceReleasedWarning``, raised at teardown).
 
 These behaviors live in the fixture's setup/teardown, so they cannot be observed
-from an ordinary test body. Each test here drives the *real* ``init_cuda``
-generator by hand with a minimal fake ``request`` and (where needed) a
-``gc.collect`` spy, so the collect decision and both warnings can be asserted
-directly. The fixture function and its module globals are reached through
-pytest's fixture registry (``FixtureDef.func``) so the monkeypatched globals are
-the ones the running code reads. Tests are ``thread_unsafe`` because they patch
-process-global state (``gc.collect`` and the conftest module globals) and assert
-on the process-global resource count; each cleans up every resource it creates
-so the real session's leftover tracking is not perturbed.
+from an ordinary test body. Each test drives the *real* ``init_cuda`` generator
+by hand with a minimal fake ``request``, and scripts ``conftest._live_resource_counts``
+so the gate's decision is exercised deterministically without allocating driver
+resources (that the counts move correctly is covered by
+``test_resource_live_count.py``). The fixture function and its module globals are
+reached through pytest's fixture registry (``FixtureDef.func``) so the
+monkeypatched globals are the ones the running code reads. Tests are
+``thread_unsafe`` because they patch process-global state (``gc.collect`` and the
+conftest module globals).
 """
 
 import gc
@@ -31,20 +31,27 @@
     UnreleasedMemoryResourceWarning,
 )
 
-from cuda.core import MemoryResource
-from cuda.core._memory._buffer import _live_memory_resource_count
 
+class _ScriptedCounts:
+    """Callable stand-in for ``conftest._live_resource_counts``.
 
-class _LocalMR(MemoryResource):
-    """A trivial MemoryResource subclass that holds no CUDA pool.
-
-    Construction/destruction only moves the live counter, so it is a clean way
-    to make the gate's snapshot comparison grow or shrink without allocating.
+    Returns the queued ``(owned_pools, va_reservations)`` tuples in order and
+    sticks on the last one, so the gate may read the counts more times than
+    scripted without raising.
     """
 
+    def __init__(self, *values):
+        self._values = list(values)
+        self._i = 0
+
+    def __call__(self):
+        value = self._values[min(self._i, len(self._values) - 1)]
+        self._i += 1
+        return value
+
 
 _THREAD_UNSAFE_MARKER = pytest.mark.thread_unsafe(
-    reason="patches gc.collect and conftest module globals, and asserts on the process-global resource count"
+    reason="patches gc.collect and conftest module globals (the resource-count hook and snapshots)"
 )
 
 
@@ -103,49 +110,49 @@ def spy(*args, **kwargs):
     return calls
 
 
+def _finish_teardown(gen):
+    with pytest.raises(StopIteration):
+        next(gen)
+
+
 # --- 2. the collect gate -------------------------------------------------------
 
 
 @_THREAD_UNSAFE_MARKER
 @pytest.mark.agent_authored(model="claude-opus-4.8")
-def test_teardown_skips_collect_when_mr_count_unchanged(request, monkeypatch):
+def test_teardown_skips_collect_when_counts_unchanged(request, monkeypatch):
     func = _init_cuda_func(request)
-    # Disable the leftover check so only the teardown gate can call gc.collect.
-    monkeypatch.setitem(func.__globals__, "_mr_expected_live", None)
+    g = func.__globals__
+    # Disable the leftover check so only the teardown gate can call gc.collect,
+    # and script equal setup/teardown counts.
+    monkeypatch.setitem(g, "_mr_expected_live", None)
+    monkeypatch.setitem(g, "_live_resource_counts", _ScriptedCounts((2, 1)))
     collects = _spy_on_gc_collect(monkeypatch)
 
     gen = func(_FakeRequest())
-    next(gen)  # setup: records the snapshot
+    next(gen)  # setup: live_at_setup == (2, 1)
     before = len(collects)
-    # A resource created and freed by refcount leaves the count at the snapshot.
-    mr = _LocalMR()
-    del mr
-    with pytest.raises(StopIteration):
-        next(gen)  # teardown runs the gate
+    _finish_teardown(gen)  # teardown: live_after == (2, 1) -> no collect
 
-    assert len(collects) == before, "gate collected despite an unchanged count"
+    assert len(collects) == before, "gate collected despite unchanged counts"
 
 
 @_THREAD_UNSAFE_MARKER
 @pytest.mark.agent_authored(model="claude-opus-4.8")
-def test_teardown_collects_when_mr_count_grows(request, monkeypatch):
+def test_teardown_collects_when_a_count_grows(request, monkeypatch):
     func = _init_cuda_func(request)
-    monkeypatch.setitem(func.__globals__, "_mr_expected_live", None)
+    g = func.__globals__
+    monkeypatch.setitem(g, "_mr_expected_live", None)
+    # Owned-pool count grows from 2 to 3 across the test.
+    monkeypatch.setitem(g, "_live_resource_counts", _ScriptedCounts((2, 1), (3, 1)))
     collects = _spy_on_gc_collect(monkeypatch)
 
     gen = func(_FakeRequest())
-    next(gen)  # setup: records the snapshot
+    next(gen)  # setup: live_at_setup == (2, 1)
     before = len(collects)
-    # A resource still alive at teardown lifts the count above the snapshot.
-    mr = _LocalMR()
-    try:
-        with pytest.raises(StopIteration):
-            next(gen)  # teardown runs the gate
-        assert len(collects) == before + 1, "gate skipped the collect despite a grown count"
-    finally:
-        # Restore the baseline so the session's leftover tracking is unperturbed.
-        del mr
-        gc.collect()
+    _finish_teardown(gen)  # teardown: live_after == (3, 1) -> collect
+
+    assert len(collects) == before + 1, "gate skipped the collect despite a grown count"
 
 
 # --- 3. the two warnings -------------------------------------------------------
@@ -156,43 +163,33 @@ def test_teardown_collects_when_mr_count_grows(request, monkeypatch):
 def test_setup_warns_when_resources_left_alive_since_last_init_cuda(request, monkeypatch):
     func = _init_cuda_func(request)
     g = func.__globals__
-    # Force the leftover-check precondition: the count at item start exceeds the
-    # baseline recorded after the previous init_cuda test, i.e. something in
-    # between left a resource alive. Keep a real resource referenced so the
-    # "still alive after a collect" count is nonzero.
-    base = _live_memory_resource_count()
-    leaked = _LocalMR()
-    monkeypatch.setitem(g, "_mr_expected_live", base)
-    monkeypatch.setitem(g, "_mr_live_at_item_start", base + 1)
+    # Force the leftover-check precondition: the item started with one more owned
+    # pool than the baseline recorded after the previous init_cuda test.
+    monkeypatch.setitem(g, "_mr_expected_live", (0, 0))
+    monkeypatch.setitem(g, "_mr_live_at_item_start", (1, 0))
     monkeypatch.setitem(g, "_mr_last_init_cuda_nodeid", "tests/prev::test_prev")
+    monkeypatch.setitem(g, "_live_resource_counts", _ScriptedCounts((1, 0)))
 
     gen = func(_FakeRequest())
-    try:
-        with pytest.warns(UnreleasedMemoryResourceWarning, match="left alive by"):
-            next(gen)  # setup emits the leftover warning
-        with pytest.raises(StopIteration):
-            next(gen)  # finish teardown
-    finally:
-        del leaked
-        gc.collect()
+    with pytest.warns(UnreleasedMemoryResourceWarning, match="left alive"):
+        next(gen)  # setup emits the leftover warning
+    _finish_teardown(gen)  # live_after == live_at_setup -> no baseline warning
 
 
 @_THREAD_UNSAFE_MARKER
 @pytest.mark.agent_authored(model="claude-opus-4.8")
 def test_teardown_warns_on_baseline_release(request, monkeypatch):
     func = _init_cuda_func(request)
-    # Disable the leftover check; this test exercises the teardown-side warning.
-    monkeypatch.setitem(func.__globals__, "_mr_expected_live", None)
+    g = func.__globals__
+    monkeypatch.setitem(g, "_mr_expected_live", None)  # disable the leftover check
+    # Script a snapshot of (1, 0) at setup and (0, 0) at teardown: an owned pool
+    # that existed at setup was released during the test.
+    monkeypatch.setitem(g, "_live_resource_counts", _ScriptedCounts((1, 0), (0, 0)))
 
-    # A resource that exists before setup is part of the snapshot; releasing it
-    # during the "test" drops the count below the snapshot at teardown.
-    baseline_mr = _LocalMR()
     gen = func(_FakeRequest())
-    next(gen)  # setup snapshots the count, including baseline_mr
-    del baseline_mr  # release the pre-existing resource mid-"test"
-
+    next(gen)  # setup: live_at_setup == (1, 0)
     with (
         pytest.warns(BaselineMemoryResourceReleasedWarning, match="existed before it started"),
         pytest.raises(StopIteration),
     ):
-        next(gen)  # teardown sees a count below the snapshot
+        next(gen)  # teardown: live_after == (0, 0) -> baseline-release warning
diff --git a/cuda_core/tests/memory/test_resource_live_count.py b/cuda_core/tests/memory/test_resource_live_count.py
index ef13bc4e6b3..086f882f427 100644
--- a/cuda_core/tests/memory/test_resource_live_count.py
+++ b/cuda_core/tests/memory/test_resource_live_count.py
@@ -1,112 +1,133 @@
 # SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
 # SPDX-License-Identifier: Apache-2.0
 
-"""Tests for the process-wide live ``MemoryResource`` counter.
+"""Tests for the process-wide live owned-resource counters.
+
+The counting lives in the C++ RAII layer: ``_live_owned_mempool_count`` tracks
+memory pools this layer owns (created with ``cuMemPoolCreate`` or imported), and
+``_live_va_reservation_count`` tracks VA reservations (``cuMemAddressReserve``).
+``conftest.py`` reads them to decide whether the ``init_cuda`` teardown needs a
+``gc.collect()``. Borrowed pool references (the device default pool) and
+``MemoryResource`` subclasses that hold no pool or reservation move neither
+counter, which these tests pin down.
 
-``conftest.py`` uses the counter to decide whether the ``init_cuda`` teardown
-needs a ``gc.collect()`` and to detect resources left alive by other tests.
 Tests compare against a count read at their start, because resources retained
-from earlier tests make it nonzero. Each test requests ``init_cuda`` and frees
-what it constructs. They are ``thread_unsafe`` because the count is
-process-global.
+from earlier tests make it nonzero. Each requests ``init_cuda`` and frees what
+it constructs. They are ``thread_unsafe`` because the counters are process-global.
 """
 
 import gc
+import platform
 import threading
 
 import pytest
+from helpers.constants import POOL_SIZE
 
-from cuda.core import DeviceMemoryResource, MemoryResource
-from cuda.core._memory._buffer import _live_memory_resource_count
+from cuda.core import (
+    DeviceMemoryResource,
+    DeviceMemoryResourceOptions,
+    VirtualMemoryResource,
+    VirtualMemoryResourceOptions,
+)
+from cuda.core._memory._buffer import _live_owned_mempool_count, _live_va_reservation_count
 
+# Matches the VMM handle type helper in test_memory.py.
+VMM_HANDLE_TYPE = "win32_kmt" if platform.system() == "Windows" else "posix_fd"
 
-class _CountedMR(MemoryResource):
-    pass
+_THREAD_UNSAFE_MARKER = pytest.mark.thread_unsafe(reason="asserts on the process-global owned-resource counts")
 
 
-class _CountedMRWithArgs(MemoryResource):
-    def __init__(self, a, b):
-        self.a = a
-        self.b = b
+def _skip_without_mempools(device):
+    if not device.properties.memory_pools_supported:
+        pytest.skip("Device does not support memory pools")
 
 
-_THREAD_UNSAFE_MARKER = pytest.mark.thread_unsafe(reason="asserts on the process-global resource count")
+def _owned_pool(device):
+    """An owning DeviceMemoryResource: passing options creates its own pool."""
+    return DeviceMemoryResource(device, DeviceMemoryResourceOptions(max_size=POOL_SIZE))
 
 
 @_THREAD_UNSAFE_MARKER
-@pytest.mark.agent_authored(model="glm-5.2")
-def test_new_resource_is_counted_until_freed(init_cuda):
-    before = _live_memory_resource_count()
-    mr = _CountedMR()
-    assert _live_memory_resource_count() == before + 1
+@pytest.mark.agent_authored(model="claude-opus-4.8")
+def test_owned_pool_is_counted_until_freed(init_cuda):
+    _skip_without_mempools(init_cuda)
+    before = _live_owned_mempool_count()
+    mr = _owned_pool(init_cuda)
+    assert _live_owned_mempool_count() == before + 1
     del mr
-    assert _live_memory_resource_count() == before
+    assert _live_owned_mempool_count() == before
 
 
 @_THREAD_UNSAFE_MARKER
-@pytest.mark.agent_authored(model="glm-5.2")
-def test_resource_in_cycle_stays_counted_until_collect(init_cuda):
-    before = _live_memory_resource_count()
-    mr = _CountedMR()
-    mr.self_ref = mr  # create a reference cycle
+@pytest.mark.agent_authored(model="claude-opus-4.8")
+def test_borrowed_default_pool_is_not_counted(init_cuda):
+    # A no-options DeviceMemoryResource borrows the device default pool
+    # (cuDeviceGetMemPool), which this layer does not own or destroy, so it must
+    # not move the owned-pool counter.
+    _skip_without_mempools(init_cuda)
+    before = _live_owned_mempool_count()
+    mr = DeviceMemoryResource(init_cuda)
+    assert _live_owned_mempool_count() == before
     del mr
-    # The cycle keeps the resource alive, so it is still counted.
-    assert _live_memory_resource_count() == before + 1
+    assert _live_owned_mempool_count() == before
+
+
+@_THREAD_UNSAFE_MARKER
+@pytest.mark.agent_authored(model="claude-opus-4.8")
+def test_owned_pool_in_cycle_stays_counted_until_collect(init_cuda):
+    _skip_without_mempools(init_cuda)
+    before = _live_owned_mempool_count()
+    mr = _owned_pool(init_cuda)
+    # A dict cycle holds the only reference to mr (the Cython resource cannot
+    # hold a self-attribute), so refcounting alone will not free it.
+    holder = {"mr": mr}
+    holder["self"] = holder
+    del mr, holder
+    assert _live_owned_mempool_count() == before + 1
     # Disable automatic collection so a background gc cannot free the cycle
     # before the assertion is checked.
     enabled = gc.isenabled()
     gc.disable()
     try:
         gc.collect()
-        assert _live_memory_resource_count() == before
+        assert _live_owned_mempool_count() == before
     finally:
         if enabled:
             gc.enable()
 
 
 @_THREAD_UNSAFE_MARKER
-@pytest.mark.agent_authored(model="glm-5.2")
-def test_subclass_with_init_args_is_counted(init_cuda):
-    # Guards the no-argument base __cinit__: a Python subclass whose __init__
-    # takes extra arguments still constructs (and is counted) via the base.
-    before = _live_memory_resource_count()
-    mr = _CountedMRWithArgs(1, 2)
-    assert _live_memory_resource_count() == before + 1
-    assert mr.a == 1 and mr.b == 2
-    del mr
-    assert _live_memory_resource_count() == before
-
-
-@_THREAD_UNSAFE_MARKER
-@pytest.mark.agent_authored(model="glm-5.2")
-def test_cython_subclass_wrapper_is_counted(init_cuda):
-    # Covers a Cython subclass with its own __cinit__: a non-owning
-    # DeviceMemoryResource(device) wrapper (no options) wraps the default pool
-    # and is counted and uncounted like any resource. The no-options path
-    # needs the current device's memory pool, so skip where it is unsupported
-    # (e.g. Windows TCC); the counter behavior itself is platform-independent.
-    if not init_cuda.properties.memory_pools_supported:
-        pytest.skip("Device does not support memory pools")
-    before = _live_memory_resource_count()
-    mr = DeviceMemoryResource(init_cuda)
-    assert _live_memory_resource_count() == before + 1
-    del mr
-    assert _live_memory_resource_count() == before
+@pytest.mark.agent_authored(model="claude-opus-4.8")
+def test_va_reservation_is_counted_until_freed(init_cuda):
+    if not init_cuda.properties.virtual_memory_management_supported:
+        pytest.skip("Device does not support virtual memory management")
+    vmm = VirtualMemoryResource(init_cuda, config=VirtualMemoryResourceOptions(handle_type=VMM_HANDLE_TYPE))
+    before = _live_va_reservation_count()
+    try:
+        buf = vmm.allocate(4096)
+    except NotImplementedError:
+        pytest.skip("VMM handle type not implemented on this platform")
+    # An allocation reserves at least one address range.
+    assert _live_va_reservation_count() >= before + 1
+    buf.close()
+    assert _live_va_reservation_count() == before
 
 
 @_THREAD_UNSAFE_MARKER
-@pytest.mark.agent_authored(model="glm-5.2")
-def test_concurrent_construct_and_drop_returns_to_start(init_cuda):
+@pytest.mark.agent_authored(model="claude-opus-4.8")
+def test_concurrent_owned_pool_construct_and_drop_returns_to_start(init_cuda):
     # On a free-threaded build this guards against a non-atomic counter; on a
-    # GIL build it passes trivially. Every thread constructs and drops its own
-    # resources; after join the count is back at the starting value.
-    before = _live_memory_resource_count()
-    n_threads = 8
-    n_per_thread = 200
+    # GIL build it passes trivially. Every thread sets the device current, then
+    # creates and drops owning pools; after join the count is back at the start.
+    _skip_without_mempools(init_cuda)
+    before = _live_owned_mempool_count()
+    n_threads = 4
+    n_per_thread = 10
 
     def worker():
+        init_cuda.set_current()
         for _ in range(n_per_thread):
-            mr = _CountedMR()
+            mr = _owned_pool(init_cuda)
             del mr
 
     threads = [threading.Thread(target=worker) for _ in range(n_threads)]
@@ -116,4 +137,4 @@ def worker():
         t.join()
     # Let any deferred deallocation settle.
     gc.collect()
-    assert _live_memory_resource_count() == before
+    assert _live_owned_mempool_count() == before

From 774ddcc62e3816c97f27fc283d1bb0b644ecea8e Mon Sep 17 00:00:00 2001
From: Ralf Juengling <rjuengling@nvidia.com>
Date: Mon, 12 Oct 2026 08:15:56 -0700
Subject: [PATCH 08/10] test(cuda.core): harden the live owned-pool counter
 tests

- test_owned_pool_in_cycle_stays_counted_until_collect: create the cyclic
  resource and assert it is still counted inside the gc.disable() block, so an
  automatic collection cannot break the cycle before the "+1" assertion.
- test_concurrent_owned_pool_construct_and_drop_returns_to_start: record
  worker-thread exceptions and fail on them; a raising worker only kills its
  own thread, leaving the count balanced and the test passing silently.
---
 .../tests/memory/test_resource_live_count.py  | 33 +++++++++++--------
 1 file changed, 20 insertions(+), 13 deletions(-)

diff --git a/cuda_core/tests/memory/test_resource_live_count.py b/cuda_core/tests/memory/test_resource_live_count.py
index 086f882f427..d4cdfd21fe2 100644
--- a/cuda_core/tests/memory/test_resource_live_count.py
+++ b/cuda_core/tests/memory/test_resource_live_count.py
@@ -76,19 +76,19 @@ def test_borrowed_default_pool_is_not_counted(init_cuda):
 @pytest.mark.agent_authored(model="claude-opus-4.8")
 def test_owned_pool_in_cycle_stays_counted_until_collect(init_cuda):
     _skip_without_mempools(init_cuda)
-    before = _live_owned_mempool_count()
-    mr = _owned_pool(init_cuda)
-    # A dict cycle holds the only reference to mr (the Cython resource cannot
-    # hold a self-attribute), so refcounting alone will not free it.
-    holder = {"mr": mr}
-    holder["self"] = holder
-    del mr, holder
-    assert _live_owned_mempool_count() == before + 1
     # Disable automatic collection so a background gc cannot free the cycle
-    # before the assertion is checked.
+    # before the still-counted assertion is checked.
     enabled = gc.isenabled()
     gc.disable()
     try:
+        before = _live_owned_mempool_count()
+        mr = _owned_pool(init_cuda)
+        # A dict cycle holds the only reference to mr (the Cython resource cannot
+        # hold a self-attribute), so refcounting alone will not free it.
+        holder = {"mr": mr}
+        holder["self"] = holder
+        del mr, holder
+        assert _live_owned_mempool_count() == before + 1
         gc.collect()
         assert _live_owned_mempool_count() == before
     finally:
@@ -123,18 +123,25 @@ def test_concurrent_owned_pool_construct_and_drop_returns_to_start(init_cuda):
     before = _live_owned_mempool_count()
     n_threads = 4
     n_per_thread = 10
+    # An exception in a worker only kills that thread, which would leave the
+    # count balanced and the test passing; collect them and fail explicitly.
+    errors = []
 
     def worker():
-        init_cuda.set_current()
-        for _ in range(n_per_thread):
-            mr = _owned_pool(init_cuda)
-            del mr
+        try:
+            init_cuda.set_current()
+            for _ in range(n_per_thread):
+                mr = _owned_pool(init_cuda)
+                del mr
+        except Exception as e:
+            errors.append(e)
 
     threads = [threading.Thread(target=worker) for _ in range(n_threads)]
     for t in threads:
         t.start()
     for t in threads:
         t.join()
+    assert not errors, f"worker thread(s) raised: {errors!r}"
     # Let any deferred deallocation settle.
     gc.collect()
     assert _live_owned_mempool_count() == before

From 4b4e76967b392626dbaa9bc371bc22a13598d16e Mon Sep 17 00:00:00 2001
From: Ralf Juengling <rjuengling@nvidia.com>
Date: Mon, 12 Oct 2026 09:06:09 -0700
Subject: [PATCH 09/10] test(cuda.core): share VMM_HANDLE_TYPE and tighten
 gate-test patching

- helpers/memory.py: define VMM_HANDLE_TYPE once from IS_WINDOWS and use it
  in test_memory.py, graph/test_graph_definition_lifetime.py and
  memory/test_resource_live_count.py, replacing four local copies (one of
  which used platform.system()).
- test_init_cuda_collect_gate.py: patch conftest globals through
  _patch_conftest_global, which asserts the name exists. monkeypatch.setitem
  silently adds a missing key, so a rename in conftest would have left the
  gate unpatched and a test passing without exercising it.
- tests/AGENTS.md: state that the leftover and baseline-release checks track
  only owned memory pools and VA reservations, name both warnings escalated
  in pytest.ini, and note that default-pool wrappers, pool-less
  MemoryResource subclasses and their buffers are not counted.
---
 cuda_core/tests/AGENTS.md                     | 14 +++++---
 .../graph/test_graph_definition_lifetime.py   |  6 ++--
 cuda_core/tests/helpers/memory.py             |  4 +++
 .../memory/test_init_cuda_collect_gate.py     | 35 +++++++++++--------
 .../tests/memory/test_resource_live_count.py  |  5 +--
 cuda_core/tests/test_memory.py                |  4 +--
 6 files changed, 40 insertions(+), 28 deletions(-)

diff --git a/cuda_core/tests/AGENTS.md b/cuda_core/tests/AGENTS.md
index daac7556de6..4cd50de44a2 100644
--- a/cuda_core/tests/AGENTS.md
+++ b/cuda_core/tests/AGENTS.md
@@ -69,11 +69,15 @@ whose lifetime fits a single scope, and keep pool lifetimes inside the test
 that creates them. The collected items and imported modules are frozen after
 collection, so the collect covers only objects created during the run.
 
-A resource that outlives a test without `init_cuda` makes the next `init_cuda`
-test fail with `UnreleasedMemoryResourceWarning` (escalated to an error in
-`pytest.ini`). Release what you construct, and
-request `init_cuda` in any test that constructs one (see "Tests that touch CUDA
-must establish their own context").
+An owned memory pool or VA reservation that outlives a test without `init_cuda`
+makes the next `init_cuda` test fail with `UnreleasedMemoryResourceWarning`, and
+an `init_cuda` test that releases one created before it started fails with
+`BaselineMemoryResourceReleasedWarning` (both escalated to errors in
+`pytest.ini`). Other resources -- wrappers of the device default pool,
+`MemoryResource` subclasses that hold no pool, and their buffers -- are not
+counted, so leaking one goes undetected. Release what you construct regardless,
+and request `init_cuda` in any test that constructs a memory resource (see
+"Tests that touch CUDA must establish their own context").
 
 ## Shared test support
 
diff --git a/cuda_core/tests/graph/test_graph_definition_lifetime.py b/cuda_core/tests/graph/test_graph_definition_lifetime.py
index a33b5497ae8..43f0d627781 100644
--- a/cuda_core/tests/graph/test_graph_definition_lifetime.py
+++ b/cuda_core/tests/graph/test_graph_definition_lifetime.py
@@ -13,9 +13,10 @@
 import pytest
 from cuda_python_test_helpers.subprocess_runner import run_python_snippet
 from helpers.graph_kernels import compile_common_kernels
+from helpers.memory import VMM_HANDLE_TYPE
 from helpers.misc import try_create_condition
 
-from cuda_python_test_helpers import IS_WINDOWS, under_compute_sanitizer
+from cuda_python_test_helpers import under_compute_sanitizer
 
 # Resource finalization triggered by graph destruction is not synchronous. A
 # CUDA user-object callback transfers each node attachment bundle to a
@@ -112,8 +113,7 @@ def _device_memory_resource(dev):
 def _virtual_memory_resource(dev):
     if not dev.properties.virtual_memory_management_supported:
         pytest.skip("Device does not support virtual memory management")
-    handle_type = "win32_kmt" if IS_WINDOWS else "posix_fd"
-    return VirtualMemoryResource(dev, config=VirtualMemoryResourceOptions(handle_type=handle_type))
+    return VirtualMemoryResource(dev, config=VirtualMemoryResourceOptions(handle_type=VMM_HANDLE_TYPE))
 
 
 # Memory resources whose buffers a graph node can retain through its
diff --git a/cuda_core/tests/helpers/memory.py b/cuda_core/tests/helpers/memory.py
index 5908314f62b..69d4e895fe2 100644
--- a/cuda_core/tests/helpers/memory.py
+++ b/cuda_core/tests/helpers/memory.py
@@ -7,6 +7,10 @@
 
 from cuda.core import ManagedMemoryResource, PinnedMemoryResource
 from cuda.core._utils.cuda_utils import CUDAError
+from cuda_python_test_helpers import IS_WINDOWS
+
+# Shareable handle type for VirtualMemoryResource allocations on this platform.
+VMM_HANDLE_TYPE = "win32_kmt" if IS_WINDOWS else "posix_fd"
 
 
 class UnreleasedMemoryResourceWarning(Warning):
diff --git a/cuda_core/tests/memory/test_init_cuda_collect_gate.py b/cuda_core/tests/memory/test_init_cuda_collect_gate.py
index 2a525525a29..7a1fe1bcf6d 100644
--- a/cuda_core/tests/memory/test_init_cuda_collect_gate.py
+++ b/cuda_core/tests/memory/test_init_cuda_collect_gate.py
@@ -93,6 +93,17 @@ def _init_cuda_func(request):
     return func
 
 
+def _patch_conftest_global(monkeypatch, func, name, value):
+    """Monkeypatch a conftest module global that ``func`` reads.
+
+    ``monkeypatch.setitem`` silently adds a missing key, so after a rename in
+    conftest the patch would no longer reach the gate; fail loudly instead.
+    """
+    g = func.__globals__
+    assert name in g, f"conftest no longer defines {name!r}"
+    monkeypatch.setitem(g, name, value)
+
+
 def _spy_on_gc_collect(monkeypatch):
     """Patch ``gc.collect`` to record each explicit call; returns the call list.
 
@@ -122,11 +133,10 @@ def _finish_teardown(gen):
 @pytest.mark.agent_authored(model="claude-opus-4.8")
 def test_teardown_skips_collect_when_counts_unchanged(request, monkeypatch):
     func = _init_cuda_func(request)
-    g = func.__globals__
     # Disable the leftover check so only the teardown gate can call gc.collect,
     # and script equal setup/teardown counts.
-    monkeypatch.setitem(g, "_mr_expected_live", None)
-    monkeypatch.setitem(g, "_live_resource_counts", _ScriptedCounts((2, 1)))
+    _patch_conftest_global(monkeypatch, func, "_mr_expected_live", None)
+    _patch_conftest_global(monkeypatch, func, "_live_resource_counts", _ScriptedCounts((2, 1)))
     collects = _spy_on_gc_collect(monkeypatch)
 
     gen = func(_FakeRequest())
@@ -141,10 +151,9 @@ def test_teardown_skips_collect_when_counts_unchanged(request, monkeypatch):
 @pytest.mark.agent_authored(model="claude-opus-4.8")
 def test_teardown_collects_when_a_count_grows(request, monkeypatch):
     func = _init_cuda_func(request)
-    g = func.__globals__
-    monkeypatch.setitem(g, "_mr_expected_live", None)
+    _patch_conftest_global(monkeypatch, func, "_mr_expected_live", None)
     # Owned-pool count grows from 2 to 3 across the test.
-    monkeypatch.setitem(g, "_live_resource_counts", _ScriptedCounts((2, 1), (3, 1)))
+    _patch_conftest_global(monkeypatch, func, "_live_resource_counts", _ScriptedCounts((2, 1), (3, 1)))
     collects = _spy_on_gc_collect(monkeypatch)
 
     gen = func(_FakeRequest())
@@ -162,13 +171,12 @@ def test_teardown_collects_when_a_count_grows(request, monkeypatch):
 @pytest.mark.agent_authored(model="claude-opus-4.8")
 def test_setup_warns_when_resources_left_alive_since_last_init_cuda(request, monkeypatch):
     func = _init_cuda_func(request)
-    g = func.__globals__
     # Force the leftover-check precondition: the item started with one more owned
     # pool than the baseline recorded after the previous init_cuda test.
-    monkeypatch.setitem(g, "_mr_expected_live", (0, 0))
-    monkeypatch.setitem(g, "_mr_live_at_item_start", (1, 0))
-    monkeypatch.setitem(g, "_mr_last_init_cuda_nodeid", "tests/prev::test_prev")
-    monkeypatch.setitem(g, "_live_resource_counts", _ScriptedCounts((1, 0)))
+    _patch_conftest_global(monkeypatch, func, "_mr_expected_live", (0, 0))
+    _patch_conftest_global(monkeypatch, func, "_mr_live_at_item_start", (1, 0))
+    _patch_conftest_global(monkeypatch, func, "_mr_last_init_cuda_nodeid", "tests/prev::test_prev")
+    _patch_conftest_global(monkeypatch, func, "_live_resource_counts", _ScriptedCounts((1, 0)))
 
     gen = func(_FakeRequest())
     with pytest.warns(UnreleasedMemoryResourceWarning, match="left alive"):
@@ -180,11 +188,10 @@ def test_setup_warns_when_resources_left_alive_since_last_init_cuda(request, mon
 @pytest.mark.agent_authored(model="claude-opus-4.8")
 def test_teardown_warns_on_baseline_release(request, monkeypatch):
     func = _init_cuda_func(request)
-    g = func.__globals__
-    monkeypatch.setitem(g, "_mr_expected_live", None)  # disable the leftover check
+    _patch_conftest_global(monkeypatch, func, "_mr_expected_live", None)  # disable the leftover check
     # Script a snapshot of (1, 0) at setup and (0, 0) at teardown: an owned pool
     # that existed at setup was released during the test.
-    monkeypatch.setitem(g, "_live_resource_counts", _ScriptedCounts((1, 0), (0, 0)))
+    _patch_conftest_global(monkeypatch, func, "_live_resource_counts", _ScriptedCounts((1, 0), (0, 0)))
 
     gen = func(_FakeRequest())
     next(gen)  # setup: live_at_setup == (1, 0)
diff --git a/cuda_core/tests/memory/test_resource_live_count.py b/cuda_core/tests/memory/test_resource_live_count.py
index d4cdfd21fe2..b299abf17b6 100644
--- a/cuda_core/tests/memory/test_resource_live_count.py
+++ b/cuda_core/tests/memory/test_resource_live_count.py
@@ -17,11 +17,11 @@
 """
 
 import gc
-import platform
 import threading
 
 import pytest
 from helpers.constants import POOL_SIZE
+from helpers.memory import VMM_HANDLE_TYPE
 
 from cuda.core import (
     DeviceMemoryResource,
@@ -31,9 +31,6 @@
 )
 from cuda.core._memory._buffer import _live_owned_mempool_count, _live_va_reservation_count
 
-# Matches the VMM handle type helper in test_memory.py.
-VMM_HANDLE_TYPE = "win32_kmt" if platform.system() == "Windows" else "posix_fd"
-
 _THREAD_UNSAFE_MARKER = pytest.mark.thread_unsafe(reason="asserts on the process-global owned-resource counts")
 
 
diff --git a/cuda_core/tests/test_memory.py b/cuda_core/tests/test_memory.py
index 7019bed6c8b..e03be0e86fa 100644
--- a/cuda_core/tests/test_memory.py
+++ b/cuda_core/tests/test_memory.py
@@ -34,6 +34,7 @@
 from helpers.constants import POOL_SIZE
 from helpers.contexts import assert_no_cuda_warning, current_context_handle, no_current_context
 from helpers.memory import (
+    VMM_HANDLE_TYPE,
     create_managed_memory_resource_or_skip,
     create_pinned_memory_resource_or_xfail,
     skip_if_managed_memory_unsupported,
@@ -1270,7 +1271,6 @@ def get_handle_type():
         return (("posix_fd", None),)
 
 
-VMM_HANDLE_TYPE = "win32_kmt" if IS_WINDOWS else "posix_fd"
 MIB = 1024 * 1024
 
 
@@ -1446,7 +1446,7 @@ def test_vmm_allocator_policy_configuration(init_cuda):
         location_type="device",
         granularity="recommended",
         gpu_direct_rdma=False,
-        handle_type="posix_fd" if not IS_WINDOWS else "win32_kmt",
+        handle_type=VMM_HANDLE_TYPE,
         peers=(),
         self_access="r",  # Read-only access
         peer_access="r",

From bc94ed42f938915cb9fe832a00e7cc3addaf5aff Mon Sep 17 00:00:00 2001
From: Ralf Juengling <rjuengling@nvidia.com>
Date: Mon, 12 Oct 2026 09:17:13 -0700
Subject: [PATCH 10/10] test(cuda.core): escalate the baseline-release warning
 and tidy gate wording

- pytest.ini: escalate BaselineMemoryResourceReleasedWarning to an error
  alongside UnreleasedMemoryResourceWarning, matching tests/AGENTS.md.
- conftest.py: print labelled counts (owned_pools=..., va_reservations=...)
  in both warnings; replace the stale GraphMemoryResource example in the
  leftover-check comment.
- helpers/memory.py: describe both warnings in terms of owned memory pools
  and VA reservations.
- test_resource_live_count.py: release the VA test buffer with a context
  manager so a failed assertion does not leak the reservation.
- test_init_cuda_collect_gate.py: assert on the warning class only.
---
 cuda_core/pytest.ini                          | 10 ++++---
 cuda_core/tests/conftest.py                   | 26 +++++++++++--------
 cuda_core/tests/helpers/memory.py             |  4 +--
 .../memory/test_init_cuda_collect_gate.py     |  4 +--
 .../tests/memory/test_resource_live_count.py  |  6 ++---
 5 files changed, 29 insertions(+), 21 deletions(-)

diff --git a/cuda_core/pytest.ini b/cuda_core/pytest.ini
index 68867621bf4..6c510783748 100644
--- a/cuda_core/pytest.ini
+++ b/cuda_core/pytest.ini
@@ -7,14 +7,18 @@ addopts = --showlocals --durations=20
 pythonpath = tests
 norecursedirs = cython
 filterwarnings =
-    # A test left a memory resource alive; see "Release resources at test boundaries" in tests/AGENTS.md.
+    # A test left an owned memory pool or VA reservation alive, or released one
+    # that existed before it started; see "Release resources at test boundaries"
+    # in tests/AGENTS.md.
     # conftest.py makes helpers importable, which happens after pytest first
-    # applies these filters (and warns); the filter is applied again per test.
-    # The message regex below is pinned to pytest's wording (as of pytest 9.x);
+    # applies these filters (and warns); the filters are applied again per test.
+    # The message regexes below are pinned to pytest's wording (as of pytest 9.x);
     # if a pytest upgrade rewords "Failed to import filter module" this stops
     # suppressing the config warning (noisy, not fatal) and must be updated.
     ignore:Failed to import filter module.*UnreleasedMemoryResourceWarning:pytest.PytestConfigWarning
+    ignore:Failed to import filter module.*BaselineMemoryResourceReleasedWarning:pytest.PytestConfigWarning
     error::helpers.memory.UnreleasedMemoryResourceWarning
+    error::helpers.memory.BaselineMemoryResourceReleasedWarning
 markers =
     samples: tests for cuda.core standalone samples
     # Keep this authorship marker registry in sync across all pytest config roots.
diff --git a/cuda_core/tests/conftest.py b/cuda_core/tests/conftest.py
index 3ece2288693..82dea0a8919 100644
--- a/cuda_core/tests/conftest.py
+++ b/cuda_core/tests/conftest.py
@@ -150,6 +150,11 @@ def _live_resource_counts():
     return (_live_owned_mempool_count(), _live_va_reservation_count())
 
 
+def _format_counts(counts):
+    owned_pools, va_reservations = counts
+    return f"owned_pools={owned_pools}, va_reservations={va_reservations}"
+
+
 # Counts after the last init_cuda test; None until one has run.
 _mr_expected_live = None
 # Counts at the start of the current item, before its fixtures set up.
@@ -159,9 +164,9 @@ def _live_resource_counts():
 
 @pytest.hookimpl(wrapper=True)
 def pytest_runtest_protocol(item, nextitem):
-    # Leftover check: compare with the count after the previous init_cuda test,
-    # not the session start, because the count legitimately rises over a run
-    # (interned GraphMemoryResource, pytest-retained class instances).
+    # Leftover check: compare with the counts after the previous init_cuda test,
+    # not the session start, because they legitimately rise over a run (pools
+    # owned by module- or class-scoped fixtures).
     global _mr_expected_live, _mr_live_at_item_start, _mr_last_init_cuda_nodeid
     _mr_live_at_item_start = _live_resource_counts()
     try:
@@ -245,10 +250,9 @@ def init_cuda(request):
         seed = getattr(request.config.option, "randomly_seed", None)
         seed_msg = f" (pytest-randomly seed: {seed})" if seed is not None else ""
         warnings.warn(
-            f"{excess} (owned_pools, va_reservations) memory resource(s) were left alive by "
-            f"tests that ran since {_mr_last_init_cuda_nodeid} ({held} still alive after a "
-            f"collect). A test or fixture that constructs a memory resource must request "
-            f"init_cuda.{seed_msg}",
+            f"Owned resources were left alive by tests that ran since {_mr_last_init_cuda_nodeid} "
+            f"({_format_counts(excess)}; still alive after a collect: {_format_counts(held)}). "
+            f"A test or fixture that constructs a memory resource must request init_cuda.{seed_msg}",
             UnreleasedMemoryResourceWarning,
             stacklevel=2,
         )
@@ -284,10 +288,10 @@ def init_cuda(request):
         if any(now < snap for now, snap in zip(live_after, live_at_setup)):
             released = tuple(s - n for s, n in zip(live_at_setup, live_after))
             warnings.warn(
-                f"{request.node.nodeid} released {released} (owned_pools, va_reservations) memory "
-                f"resource(s) that existed before it started (counts {live_after}, snapshot "
-                f"{live_at_setup}). A leak in this test could have been masked by that release. "
-                "Find what held the released resource.",
+                f"{request.node.nodeid} released owned resources that existed before it started "
+                f"({_format_counts(released)}; counts {_format_counts(live_after)}, snapshot "
+                f"{_format_counts(live_at_setup)}). A leak in this test could have been masked by "
+                "that release. Find what held the released resource.",
                 BaselineMemoryResourceReleasedWarning,
                 stacklevel=2,
             )
diff --git a/cuda_core/tests/helpers/memory.py b/cuda_core/tests/helpers/memory.py
index 69d4e895fe2..210706b59d6 100644
--- a/cuda_core/tests/helpers/memory.py
+++ b/cuda_core/tests/helpers/memory.py
@@ -14,11 +14,11 @@
 
 
 class UnreleasedMemoryResourceWarning(Warning):
-    """A test or fixture left a ``MemoryResource`` alive without ``init_cuda``."""
+    """A test or fixture left an owned memory pool or VA reservation alive without ``init_cuda``."""
 
 
 class BaselineMemoryResourceReleasedWarning(Warning):
-    """An ``init_cuda`` test released a resource that existed before it started."""
+    """An ``init_cuda`` test released an owned memory pool or VA reservation that existed before it started."""
 
 
 def skip_if_pinned_memory_unsupported(device):
diff --git a/cuda_core/tests/memory/test_init_cuda_collect_gate.py b/cuda_core/tests/memory/test_init_cuda_collect_gate.py
index 7a1fe1bcf6d..5f0f06d2988 100644
--- a/cuda_core/tests/memory/test_init_cuda_collect_gate.py
+++ b/cuda_core/tests/memory/test_init_cuda_collect_gate.py
@@ -179,7 +179,7 @@ def test_setup_warns_when_resources_left_alive_since_last_init_cuda(request, mon
     _patch_conftest_global(monkeypatch, func, "_live_resource_counts", _ScriptedCounts((1, 0)))
 
     gen = func(_FakeRequest())
-    with pytest.warns(UnreleasedMemoryResourceWarning, match="left alive"):
+    with pytest.warns(UnreleasedMemoryResourceWarning):
         next(gen)  # setup emits the leftover warning
     _finish_teardown(gen)  # live_after == live_at_setup -> no baseline warning
 
@@ -196,7 +196,7 @@ def test_teardown_warns_on_baseline_release(request, monkeypatch):
     gen = func(_FakeRequest())
     next(gen)  # setup: live_at_setup == (1, 0)
     with (
-        pytest.warns(BaselineMemoryResourceReleasedWarning, match="existed before it started"),
+        pytest.warns(BaselineMemoryResourceReleasedWarning),
         pytest.raises(StopIteration),
     ):
         next(gen)  # teardown: live_after == (0, 0) -> baseline-release warning
diff --git a/cuda_core/tests/memory/test_resource_live_count.py b/cuda_core/tests/memory/test_resource_live_count.py
index b299abf17b6..75cc6d956ca 100644
--- a/cuda_core/tests/memory/test_resource_live_count.py
+++ b/cuda_core/tests/memory/test_resource_live_count.py
@@ -104,9 +104,9 @@ def test_va_reservation_is_counted_until_freed(init_cuda):
         buf = vmm.allocate(4096)
     except NotImplementedError:
         pytest.skip("VMM handle type not implemented on this platform")
-    # An allocation reserves at least one address range.
-    assert _live_va_reservation_count() >= before + 1
-    buf.close()
+    with buf:
+        # An allocation reserves at least one address range.
+        assert _live_va_reservation_count() >= before + 1
     assert _live_va_reservation_count() == before