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 c2a03808968..7cf76b05c6e 100644
--- a/cuda_core/cuda/core/_memory/_buffer.pyi
+++ b/cuda_core/cuda/core/_memory/_buffer.pyi
@@ -356,3 +356,6 @@ class _MemAttrs(TypedDict):
     is_device_accessible: bool
     is_host_accessible: bool
     is_managed: bool
+
+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 b1ee356c245..9a1ab411bdc 100644
--- a/cuda_core/cuda/core/_memory/_buffer.pyx
+++ b/cuda_core/cuda/core/_memory/_buffer.pyx
@@ -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,
@@ -33,6 +35,18 @@ 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
 
+
+# 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_va_reservation_count() -> int:
+    return live_va_reservation_count()
+
 import warnings
 from collections.abc import Sequence
 from typing import TYPE_CHECKING
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/pytest.ini b/cuda_core/pytest.ini
index 7c6ec1c3273..6c510783748 100644
--- a/cuda_core/pytest.ini
+++ b/cuda_core/pytest.ini
@@ -6,6 +6,19 @@
 addopts = --showlocals --durations=20
 pythonpath = tests
 norecursedirs = cython
+filterwarnings =
+    # 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 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/AGENTS.md b/cuda_core/tests/AGENTS.md
index 0bb6a1a3928..4cd50de44a2 100644
--- a/cuda_core/tests/AGENTS.md
+++ b/cuda_core/tests/AGENTS.md
@@ -60,13 +60,24 @@ 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
+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
+collection, so the collect covers only objects created during the run.
+
+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
 
@@ -239,7 +250,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/conftest.py b/cuda_core/tests/conftest.py
index f7d0b5857e8..82dea0a8919 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_owned_mempool_count, _live_va_reservation_count
 from cuda.core._utils.cuda_utils import handle_return
 
 
@@ -125,6 +131,56 @@ def _item_uses_init_cuda(item):
     return "init_cuda" in getattr(item, "fixturenames", ())
 
 
+# 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.
+
+
+def _live_resource_counts():
+    """(owned memory pools, VA reservations) live process-wide, as a tuple.
+
+    Only *owned* driver resources are counted (see the C++ accessors), so this
+    excludes borrowed pool references and non-owning MemoryResource subclasses.
+    """
+    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.
+_mr_live_at_item_start = (0, 0)
+_mr_last_init_cuda_nodeid = None
+
+
+@pytest.hookimpl(wrapper=True)
+def pytest_runtest_protocol(item, nextitem):
+    # 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:
+        return (yield)
+    finally:
+        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 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:
     """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 +228,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 +239,62 @@ 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 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()
+        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"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,
+        )
+
+    # 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:
+        # 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 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.
+        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
         # 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 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 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,
+            )
 
 
 def _device_unset_current() -> bool:
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 da263ab81bc..210706b59d6 100644
--- a/cuda_core/tests/helpers/memory.py
+++ b/cuda_core/tests/helpers/memory.py
@@ -7,6 +7,18 @@
 
 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):
+    """A test or fixture left an owned memory pool or VA reservation alive without ``init_cuda``."""
+
+
+class BaselineMemoryResourceReleasedWarning(Warning):
+    """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
new file mode 100644
index 00000000000..5f0f06d2988
--- /dev/null
+++ b/cuda_core/tests/memory/test_init_cuda_collect_gate.py
@@ -0,0 +1,202 @@
+# 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 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 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
+import inspect
+
+import pytest
+from helpers.memory import (
+    BaselineMemoryResourceReleasedWarning,
+    UnreleasedMemoryResourceWarning,
+)
+
+
+class _ScriptedCounts:
+    """Callable stand-in for ``conftest._live_resource_counts``.
+
+    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 (the resource-count hook and snapshots)"
+)
+
+
+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 _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.
+
+    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
+
+
+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_counts_unchanged(request, monkeypatch):
+    func = _init_cuda_func(request)
+    # Disable the leftover check so only the teardown gate can call gc.collect,
+    # and script equal setup/teardown counts.
+    _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())
+    next(gen)  # setup: live_at_setup == (2, 1)
+    before = len(collects)
+    _finish_teardown(gen)  # teardown: live_after == (2, 1) -> no collect
+
+    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_a_count_grows(request, monkeypatch):
+    func = _init_cuda_func(request)
+    _patch_conftest_global(monkeypatch, func, "_mr_expected_live", None)
+    # Owned-pool count grows from 2 to 3 across the test.
+    _patch_conftest_global(monkeypatch, func, "_live_resource_counts", _ScriptedCounts((2, 1), (3, 1)))
+    collects = _spy_on_gc_collect(monkeypatch)
+
+    gen = func(_FakeRequest())
+    next(gen)  # setup: live_at_setup == (2, 1)
+    before = len(collects)
+    _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 -------------------------------------------------------
+
+
+@_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)
+    # Force the leftover-check precondition: the item started with one more owned
+    # pool than the baseline recorded after the previous init_cuda test.
+    _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):
+        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)
+    _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.
+    _patch_conftest_global(monkeypatch, func, "_live_resource_counts", _ScriptedCounts((1, 0), (0, 0)))
+
+    gen = func(_FakeRequest())
+    next(gen)  # setup: live_at_setup == (1, 0)
+    with (
+        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
new file mode 100644
index 00000000000..75cc6d956ca
--- /dev/null
+++ b/cuda_core/tests/memory/test_resource_live_count.py
@@ -0,0 +1,144 @@
+# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+# SPDX-License-Identifier: Apache-2.0
+
+"""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.
+
+Tests compare against a count read at their start, because resources retained
+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 threading
+
+import pytest
+from helpers.constants import POOL_SIZE
+from helpers.memory import VMM_HANDLE_TYPE
+
+from cuda.core import (
+    DeviceMemoryResource,
+    DeviceMemoryResourceOptions,
+    VirtualMemoryResource,
+    VirtualMemoryResourceOptions,
+)
+from cuda.core._memory._buffer import _live_owned_mempool_count, _live_va_reservation_count
+
+_THREAD_UNSAFE_MARKER = pytest.mark.thread_unsafe(reason="asserts on the process-global owned-resource counts")
+
+
+def _skip_without_mempools(device):
+    if not device.properties.memory_pools_supported:
+        pytest.skip("Device does not support memory pools")
+
+
+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="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_owned_mempool_count() == before
+
+
+@_THREAD_UNSAFE_MARKER
+@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
+    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)
+    # Disable automatic collection so a background gc cannot free the cycle
+    # 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:
+        if enabled:
+            gc.enable()
+
+
+@_THREAD_UNSAFE_MARKER
+@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")
+    with buf:
+        # An allocation reserves at least one address range.
+        assert _live_va_reservation_count() >= before + 1
+    assert _live_va_reservation_count() == before
+
+
+@_THREAD_UNSAFE_MARKER
+@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 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
+    # 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():
+        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
diff --git a/cuda_core/tests/memory_ipc/test_errors.py b/cuda_core/tests/memory_ipc/test_errors.py
index a09c4f44ac9..9d1a3fef591 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()
+            # 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)
+            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..48fbc9d37ce 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)
+    # 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_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:
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)
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..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,
@@ -158,9 +159,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 +177,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 +195,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 +235,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 +265,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 +286,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)
@@ -341,7 +335,14 @@ def fill_env(request, init_cuda):
         mr = DummyUnifiedMemoryResource(device)
     else:
         mr = DummyPinnedMemoryResource(device)
-    return device, mr
+    try:
+        yield device, mr
+    finally:
+        # 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
 
 
 _FILL_SIZE = 64  # Keep small; divisible by 1/2/4.
@@ -428,9 +429,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 +449,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 +473,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 +635,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 +774,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 +797,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
@@ -882,10 +879,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))
 
@@ -898,7 +894,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
@@ -907,8 +903,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)
 
@@ -1041,11 +1036,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 +1054,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 +1089,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 +1113,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 +1129,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 +1138,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 +1152,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 +1173,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
@@ -1282,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
 
 
@@ -1366,7 +1354,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 +1397,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
@@ -1458,7 +1446,7 @@ def test_vmm_allocator_policy_configuration():
         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",
@@ -1502,7 +1490,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 +2112,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 +2511,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 +2553,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 +2918,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 +3089,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)