# SPDX-FileCopyrightText: Copyright (c) 2024-2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
# SPDX-License-Identifier: Apache-2.0

import concurrent.futures
import ctypes
import gc
import multiprocessing as mp
import sys
import textwrap
import threading
import warnings

from cuda.bindings import driver

try:
    import numpy as np
except ImportError:
    np = None
import platform
import re

import pytest
from cuda_python_test_helpers.subprocess_runner import run_python_snippet
from helpers import supports_ipc_mempool
from helpers.buffers import (
    DummyDeviceMemoryResource,
    DummyHostMemoryResource,
    DummyUnifiedMemoryResource,
    NumpyHostMemoryResource,
    StubMemoryResource,
    make_instrumented_memory_resource,
    thread_unsafe_on_windows,
)
from helpers.child_processes import child_timeout_sec, kill_subprocesses
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,
    skip_if_pinned_memory_unsupported,
)
from helpers.nanosleep_kernel import NanosleepKernel

from cuda.core import (
    Buffer,
    CUDAOutOfMemoryError,
    CUDAWarning,
    Device,
    DeviceMemoryResource,
    DeviceMemoryResourceOptions,
    GraphMemoryResource,
    LegacyPinnedMemoryResource,
    ManagedBuffer,
    ManagedMemoryResource,
    ManagedMemoryResourceOptions,
    MemoryResource,
    PinnedMemoryResource,
    PinnedMemoryResourceOptions,
    StreamOptions,
    VirtualMemoryBuffer,
    VirtualMemoryResource,
    VirtualMemoryResourceOptions,
)
from cuda.core._dlpack import DLDeviceType
from cuda.core._memory._ipc import IPCBufferDescriptor
from cuda.core._stream import default_stream
from cuda.core._utils.cuda_utils import CUDAError, handle_return
from cuda.core._utils.version import driver_version
from cuda.core.typing import (
    ManagedMemoryLocationType,
    VirtualMemoryAccessType,
    VirtualMemoryAllocationType,
    VirtualMemoryGranularityType,
    VirtualMemoryHandleType,
    VirtualMemoryLocationType,
)
from cuda.core.utils import StridedMemoryView
from cuda_python_test_helpers import IS_WINDOWS

CHILD_TIMEOUT_SEC = child_timeout_sec()


def _allocate_pinned_buffer_or_xfail(mr, size, *, device):
    try:
        return mr.allocate(size, stream=device.default_stream)
    except CUDAError as exc:
        if "CUDA_ERROR_OUT_OF_MEMORY" in str(exc):
            pytest.xfail("TODO(#9999): Resolve CUDA_ERROR_OUT_OF_MEMORY")
        raise
    except RuntimeError as exc:
        if "Failed to allocate memory from pool" in str(exc):
            pytest.xfail("TODO(#9999): Resolve Failed to allocate memory from pool")
        raise


class DummyPinnedMemoryResource(MemoryResource):
    # cuMemAllocHost / cuMemFreeHost are synchronous; stream is accepted
    # for interface conformance but ignored.
    def __init__(self, device):
        self.device = device

    def allocate(self, size, *, stream=None) -> Buffer:
        ptr = handle_return(driver.cuMemAllocHost(size))
        return Buffer.from_handle(ptr=ptr, size=size, mr=self)

    def deallocate(self, ptr, size, *, stream=None):
        handle_return(driver.cuMemFreeHost(ptr))

    @property
    def is_device_accessible(self) -> bool:
        return True

    @property
    def is_host_accessible(self) -> bool:
        return True

    @property
    def device_id(self) -> int:
        raise RuntimeError("the pinned memory resource is not bound to any GPU")


class NullMemoryResource(DummyHostMemoryResource):
    @property
    def is_host_accessible(self) -> bool:
        return False


def test_package_contents():
    expected = [
        "Buffer",
        "DeviceMemoryResource",
        "DeviceMemoryResourceOptions",
        "GraphMemoryResource",
        "LegacyPinnedMemoryResource",
        "ManagedBuffer",
        "ManagedMemoryResource",
        "ManagedMemoryResourceOptions",
        "MemoryResource",
        "PinnedMemoryResource",
        "PinnedMemoryResourceOptions",
        "VirtualMemoryBuffer",
        "VirtualMemoryResource",
        "VirtualMemoryResourceOptions",
    ]
    d = {}
    exec("from cuda.core._memory import *", d)  # noqa: S102
    d = {k: v for k, v in d.items() if not k.startswith("__")}
    assert sorted(expected) == sorted(d.keys())


def buffer_initialization(dummy_mr: MemoryResource):
    buffer = dummy_mr.allocate(size=1024)
    assert buffer.handle != 0
    assert buffer.size == 1024
    assert buffer.memory_resource == dummy_mr
    assert buffer.is_device_accessible == dummy_mr.is_device_accessible
    assert buffer.is_host_accessible == dummy_mr.is_host_accessible
    assert not buffer.is_mapped
    buffer.close()


def test_buffer_initialization(init_cuda):
    device = init_cuda
    buffer_initialization(DummyDeviceMemoryResource(device))
    buffer_initialization(DummyHostMemoryResource())
    buffer_initialization(DummyUnifiedMemoryResource(device))
    buffer_initialization(DummyPinnedMemoryResource(device))
    with pytest.raises(TypeError):
        buffer_initialization(MemoryResource())


@pytest.mark.agent_authored(model="gpt-5.6-sol")
def test_buffer_direct_init_forbidden():
    """Buffers must come from a MemoryResource, never from ``Buffer()``."""
    with pytest.raises(RuntimeError, match=r"^Buffer objects cannot be instantiated directly\."):
        Buffer()


@pytest.mark.agent_authored(model="gpt-5.6-sol")
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 = init_cuda
    mr = DummyDeviceMemoryResource(device)
    buf = mr.allocate(size=64, stream=device.default_stream)
    with buf as entered:
        assert entered is buf
        assert buf.handle != 0
    assert buf.handle == 0
    assert buf.memory_resource is None

    buf = mr.allocate(size=64, stream=device.default_stream)
    with pytest.raises(RuntimeError, match="^boom$"), buf:
        raise RuntimeError("boom")
    assert buf.handle == 0


@pytest.mark.agent_authored(model="gpt-5.6-sol")
def test_memory_resource_abstract_stubs(init_cuda):
    """Every abstract MemoryResource member reports itself as unimplemented."""
    device = init_cuda
    mr = MemoryResource()
    stream = device.default_stream
    with pytest.raises(TypeError, match=r"^MemoryResource\.allocate must be implemented"):
        mr.allocate(1, stream=stream)
    with pytest.raises(TypeError, match=r"^MemoryResource\.deallocate must be implemented"):
        mr.deallocate(0, 1, stream=stream)
    with pytest.raises(TypeError, match=r"^MemoryResource\.is_device_accessible must be implemented"):
        _ = mr.is_device_accessible
    with pytest.raises(TypeError, match=r"^MemoryResource\.is_host_accessible must be implemented"):
        _ = mr.is_host_accessible
    with pytest.raises(TypeError, match=r"^MemoryResource\.device_id must be implemented"):
        _ = mr.device_id


def buffer_copy_to(dummy_mr: MemoryResource, device: Device, check=False):
    src_buffer = dummy_mr.allocate(size=1024)
    dst_buffer = dummy_mr.allocate(size=1024)
    stream = device.create_stream()

    if check:
        src_ptr = ctypes.cast(src_buffer.handle, ctypes.POINTER(ctypes.c_byte))
        for i in range(1024):
            src_ptr[i] = ctypes.c_byte(i)

    src_buffer.copy_to(dst_buffer, stream=stream)
    device.sync()

    if check:
        dst_ptr = ctypes.cast(dst_buffer.handle, ctypes.POINTER(ctypes.c_byte))

        for i in range(10):
            assert dst_ptr[i] == src_ptr[i]

    dst_buffer.close()
    src_buffer.close()


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)


def buffer_copy_from(dummy_mr: MemoryResource, device, check=False):
    src_buffer = dummy_mr.allocate(size=1024)
    dst_buffer = dummy_mr.allocate(size=1024)
    stream = device.create_stream()

    if check:
        src_ptr = ctypes.cast(src_buffer.handle, ctypes.POINTER(ctypes.c_byte))
        for i in range(1024):
            src_ptr[i] = ctypes.c_byte(i)

    dst_buffer.copy_from(src_buffer, stream=stream)
    device.sync()

    if check:
        dst_ptr = ctypes.cast(dst_buffer.handle, ctypes.POINTER(ctypes.c_byte))

        for i in range(10):
            assert dst_ptr[i] == src_ptr[i]

    dst_buffer.close()
    src_buffer.close()


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(init_cuda):
    device = init_cuda
    mr = DummyDeviceMemoryResource(device)
    stream = device.create_stream()
    src_buffer = mr.allocate(size=1024)
    dst_buffer = mr.allocate(size=2048)

    with pytest.raises(ValueError, match="buffer sizes mismatch"):
        src_buffer.copy_to(dst_buffer, stream=stream)

    dst_buffer.close()
    src_buffer.close()


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)
    dst_buffer = mr.allocate(size=2048)

    with pytest.raises(ValueError, match="buffer sizes mismatch"):
        dst_buffer.copy_from(src_buffer, stream=stream)

    dst_buffer.close()
    src_buffer.close()


@pytest.mark.agent_authored(model="gpt-5.6-sol")
def test_copy_to_auto_dst_requires_memory_resource():
    """``copy_to()`` cannot mint a destination without a memory resource."""
    device = Device()
    device.set_current()
    owner = (ctypes.c_byte * 32)()
    buf = Buffer.from_handle(ctypes.addressof(owner), 32, owner=owner)
    try:
        with pytest.raises(ValueError, match="does not have a memory_resource"):
            buf.copy_to(stream=device.default_stream)
    finally:
        buf.close()


def _bytes_repeat(pattern: bytes, size: int) -> bytes:
    assert len(pattern) > 0
    assert size % len(pattern) == 0
    return pattern * (size // len(pattern))


def _pattern_bytes(value) -> bytes:
    if isinstance(value, int):
        return bytes([value])
    return bytes(memoryview(value).cast("B"))


@pytest.fixture(params=["device", "unified", "pinned"])
def fill_env(request, init_cuda):
    device = init_cuda
    if request.param == "device":
        mr = DummyDeviceMemoryResource(device)
    elif request.param == "unified":
        mr = DummyUnifiedMemoryResource(device)
    else:
        mr = DummyPinnedMemoryResource(device)
    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.

_FILL_CASES = [
    # int -> 1-byte pattern
    pytest.param(0x42, _FILL_SIZE, None, id="int-0x42"),
    pytest.param(-1, _FILL_SIZE, OverflowError, id="int-neg"),
    pytest.param(256, _FILL_SIZE, OverflowError, id="int-256"),
    pytest.param(1000, _FILL_SIZE, OverflowError, id="int-1000"),
    # bad type
    pytest.param("invalid", _FILL_SIZE, TypeError, id="bad-type-str"),
    # bytes-like patterns
    pytest.param(b"\x7f", _FILL_SIZE, None, id="bytes-1"),
    pytest.param(b"\x34\x12", _FILL_SIZE, None, id="bytes-2"),
    pytest.param(b"\xef\xbe\xad\xde", _FILL_SIZE, None, id="bytes-4"),
    pytest.param(b"\x34\x12", _FILL_SIZE + 1, ValueError, id="bytes-2-bad-size"),
    pytest.param(b"\xef\xbe\xad\xde", _FILL_SIZE + 2, ValueError, id="bytes-4-bad-size"),
    pytest.param(b"", _FILL_SIZE, ValueError, id="bytes-0"),
    pytest.param(b"\x01\x02\x03", _FILL_SIZE, ValueError, id="bytes-3"),
]

if np is not None:
    _FILL_CASES.extend(
        [
            # 8-bit patterns
            pytest.param(np.uint8(0), _FILL_SIZE, None, id="np-uint8-0"),
            pytest.param(np.uint8(255), _FILL_SIZE, None, id="np-uint8-255"),
            pytest.param(np.int8(-1), _FILL_SIZE, None, id="np-int8--1"),
            pytest.param(np.int8(127), _FILL_SIZE, None, id="np-int8-127"),
            pytest.param(np.int8(-128), _FILL_SIZE, None, id="np-int8--128"),
            # 16-bit patterns
            pytest.param(np.uint16(0x1234), _FILL_SIZE, None, id="np-uint16-0x1234"),
            pytest.param(np.uint16(0xFFFF), _FILL_SIZE, None, id="np-uint16-0xFFFF"),
            pytest.param(np.int16(-1), _FILL_SIZE, None, id="np-int16--1"),
            pytest.param(np.int16(32767), _FILL_SIZE, None, id="np-int16-max"),
            pytest.param(np.int16(-32768), _FILL_SIZE, None, id="np-int16-min"),
            pytest.param(np.uint16(0x1234), _FILL_SIZE + 1, ValueError, id="np-uint16-bad-size"),
            # 32-bit patterns
            pytest.param(np.uint32(0xDEADBEEF), _FILL_SIZE, None, id="np-uint32-0xDEADBEEF"),
            pytest.param(np.uint32(0xFFFFFFFF), _FILL_SIZE, None, id="np-uint32-0xFFFFFFFF"),
            pytest.param(np.int32(-1), _FILL_SIZE, None, id="np-int32--1"),
            pytest.param(np.int32(2147483647), _FILL_SIZE, None, id="np-int32-max"),
            pytest.param(np.int32(-2147483648), _FILL_SIZE, None, id="np-int32-min"),
            pytest.param(np.uint32(0xDEADBEEF), _FILL_SIZE + 2, ValueError, id="np-uint32-bad-size"),
            # float32 (bit-pattern fill)
            pytest.param(np.float32(1.0), _FILL_SIZE, None, id="np-float32-1.0"),
            # 64-bit patterns should error (8-byte pattern)
            pytest.param(np.uint64(0), _FILL_SIZE, ValueError, id="np-uint64-err"),
            pytest.param(np.int64(0), _FILL_SIZE, ValueError, id="np-int64-err"),
            pytest.param(np.float64(0), _FILL_SIZE, ValueError, id="np-float64-err"),
        ]
    )


@thread_unsafe_on_windows
@pytest.mark.parametrize("value,size,exc", _FILL_CASES)
def test_buffer_fill(fill_env, value, size, exc):
    device, mr = fill_env
    stream = device.create_stream()
    buffer = mr.allocate(size=size)
    try:
        if exc is not None:
            with pytest.raises(exc):
                buffer.fill(value, stream=stream)
            return

        buffer.fill(value, stream=stream)
        device.sync()

        # Verify contents only for host-accessible buffers.
        if buffer.is_host_accessible:
            pat = _pattern_bytes(value)
            got = ctypes.string_at(int(buffer.handle), size)
            assert got == _bytes_repeat(pat, size)
    finally:
        buffer.close()


def buffer_close(dummy_mr: MemoryResource):
    buffer = dummy_mr.allocate(size=1024)
    buffer.close()
    assert buffer.handle == 0
    assert buffer.memory_resource is None


def test_buffer_close(init_cuda):
    device = init_cuda
    buffer_close(DummyDeviceMemoryResource(device))
    buffer_close(DummyHostMemoryResource())
    buffer_close(DummyUnifiedMemoryResource(device))
    buffer_close(DummyPinnedMemoryResource(device))


def test_buffer_external_host():
    a = (ctypes.c_byte * 20)()
    ptr = ctypes.addressof(a)
    buffer = Buffer.from_handle(ptr, 20, owner=a)
    assert buffer.owner is a
    assert not buffer.is_device_accessible
    assert buffer.is_host_accessible
    assert buffer.device_id == -1
    buffer.close()


@pytest.mark.parametrize("change_device", [True, False])
def test_buffer_external_device(init_cuda, change_device):
    n = len(Device.get_all_devices())
    if n < 1:
        pytest.skip("No devices found")
    dev_id = n - 1
    d = Device(dev_id)
    d.set_current()
    buffer_ = d.allocate(size=32, stream=d.default_stream)

    if change_device:
        # let's switch to a different device if possibe
        # to make sure we get the original device id
        d = Device(0)
        d.set_current()

    buffer = Buffer.from_handle(int(buffer_.handle), 32)
    assert buffer.is_device_accessible
    assert not buffer.is_host_accessible
    assert buffer.device_id == dev_id
    buffer.close()
    buffer_.close()


@pytest.mark.parametrize("change_device", [True, False])
def test_buffer_external_pinned_alloc(init_cuda, change_device):
    n = len(Device.get_all_devices())
    if n < 1:
        pytest.skip("No devices found")
    dev_id = n - 1
    d = Device(dev_id)
    d.set_current()
    mr = DummyPinnedMemoryResource(d)
    buffer_ = mr.allocate(size=32)

    if change_device:
        # let's switch to a different device if possibe
        # to make sure we get the original device id
        d = Device(0)
        d.set_current()

    buffer = Buffer.from_handle(int(buffer_.handle), 32)
    assert buffer.is_device_accessible
    assert buffer.is_host_accessible
    assert buffer.device_id == dev_id
    buffer.close()
    buffer_.close()


@pytest.mark.parametrize("change_device", [True, False])
def test_buffer_external_pinned_registered(change_device):
    n = len(Device.get_all_devices())
    if n < 1:
        pytest.skip("No devices found")
    dev_id = n - 1
    d = Device(dev_id)
    d.set_current()
    a = (ctypes.c_byte * 20)()
    ptr = ctypes.addressof(a)

    buffer = Buffer.from_handle(ptr, 20, owner=ptr)
    assert not buffer.is_device_accessible
    assert buffer.is_host_accessible
    assert buffer.device_id == -1

    handle_return(driver.cuMemHostRegister(ptr, 20, 0))
    try:
        if change_device:
            # let's switch to a different device if possibe
            # to make sure we get the original device id
            d = Device(0)
            d.set_current()

        buffer = Buffer.from_handle(ptr, 20, owner=ptr)
        assert buffer.is_device_accessible
        assert buffer.is_host_accessible
        assert buffer.device_id == dev_id
        buffer.close()
    finally:
        handle_return(driver.cuMemHostUnregister(ptr))


@pytest.mark.parametrize("change_device", [True, False])
def test_buffer_external_managed(change_device):
    n = len(Device.get_all_devices())
    if n < 1:
        pytest.skip("No devices found")
    dev_id = n - 1
    d = Device(dev_id)
    d.set_current()
    ptr = None
    try:
        ptr = handle_return(driver.cuMemAllocManaged(32, driver.CUmemAttach_flags.CU_MEM_ATTACH_GLOBAL.value))
        if change_device:
            # let's switch to a different device if possibe
            # to make sure we get the original device id
            d = Device(0)
            d.set_current()
        buffer = Buffer.from_handle(ptr, 32)
        assert buffer.is_device_accessible
        assert buffer.is_host_accessible
        assert buffer.device_id == dev_id
    finally:
        if ptr is not None:
            handle_return(driver.cuMemFree(ptr))


def test_mr_deallocate_called_on_close():
    """Buffer.from_handle(mr=mr) calls mr.deallocate() on close (issue #1619)."""
    device = Device()
    device.set_current()
    TrackingMR, telemetry = make_instrumented_memory_resource(DummyDeviceMemoryResource, track_active=True)
    mr = TrackingMR(device)
    buf = mr.allocate(1024)
    assert len(telemetry["active"]) == 1
    buf.close()
    assert len(telemetry["active"]) == 0


def test_mr_deallocate_called_on_gc():
    """Buffer.from_handle(mr=mr) calls mr.deallocate() on GC (issue #1619)."""
    import gc

    device = Device()
    device.set_current()
    TrackingMR, telemetry = make_instrumented_memory_resource(DummyDeviceMemoryResource, track_active=True)
    mr = TrackingMR(device)
    buf = mr.allocate(1024)
    assert len(telemetry["active"]) == 1
    del buf
    gc.collect()
    assert len(telemetry["active"]) == 0


def test_mr_deallocate_receives_stream():
    """Buffer.close(stream) forwards the stream to mr.deallocate() (issue #1619)."""
    device = Device()
    device.set_current()
    stream = device.create_stream()
    CapturingMR, telemetry = make_instrumented_memory_resource(DummyDeviceMemoryResource, record_streams=True)
    mr = CapturingMR(device)
    buf = mr.allocate(1024)
    buf.close(stream)
    assert telemetry["deallocations"][-1]["stream"].handle == stream.handle


@pytest.mark.agent_authored(model="gpt-5.6")
@pytest.mark.parametrize(
    ("configuration", "destruction"),
    [
        ("initialization", "close"),
        ("initialization", "gc"),
        ("setter", "close"),
        ("setter", "gc"),
        ("close", "close"),
    ],
)
def test_buffer_deallocation_stream_configuration_paths(configuration, destruction):
    """Creation, mutation, and close overrides use the requested stream."""
    import gc

    device = Device()
    device.set_current()
    initial_stream = device.create_stream()
    target_stream = device.create_stream()
    CapturingMR, telemetry = make_instrumented_memory_resource(record_streams=True)
    mr = CapturingMR(device)

    stream = target_stream if configuration == "initialization" else initial_stream
    buf = Buffer.from_handle(1, 1024, mr=mr, stream=stream)
    if configuration == "setter":
        handle = buf.handle
        buf.set_deallocation_stream(target_stream)
        assert buf.handle == handle
        assert buf.size == 1024

    if destruction == "close":
        buf.close(stream=target_stream if configuration == "close" else None)
    else:
        del buf
        gc.collect()

    assert len(telemetry["deallocations"]) == 1
    assert telemetry["deallocations"][0]["stream"].handle == target_stream.handle


@pytest.mark.agent_authored(model="gpt-5.6")
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)

    with pytest.raises(TypeError, match="stream is required"):
        buf.set_deallocation_stream(None)

    buf.close()
    with pytest.raises(RuntimeError, match="Buffer has been closed"):
        buf.set_deallocation_stream(stream)


@pytest.mark.agent_authored(model="gpt-5.6")
def test_closed_deallocation_stream_does_not_mutate_buffer():
    device = Device()
    device.set_current()
    initial_stream = device.create_stream()
    closed_stream = device.create_stream()
    CapturingMR, telemetry = make_instrumented_memory_resource(record_streams=True)
    mr = CapturingMR(device)
    buf = Buffer.from_handle(1, 1024, mr=mr, stream=initial_stream)
    closed_stream.close()

    with pytest.raises(RuntimeError, match="Stream has been closed"):
        buf.set_deallocation_stream(closed_stream)
    assert not buf.is_closed

    with pytest.raises(RuntimeError, match="Stream has been closed"):
        buf.close(stream=closed_stream)
    assert not buf.is_closed

    buf.close()
    assert len(telemetry["deallocations"]) == 1
    assert telemetry["deallocations"][0]["stream"].handle == initial_stream.handle


@pytest.mark.agent_authored(model="gpt-5.6")
def test_closed_buffer_rejected_before_active_operations():
    device = Device()
    device.set_current()
    stream = device.create_stream()
    closed = Buffer.from_handle(1, 16, owner=object())
    live = Buffer.from_handle(2, 16, owner=object())
    closed.close()

    for operation in (
        lambda: closed.copy_to(live, stream=stream),
        lambda: closed.copy_from(live, stream=stream),
        lambda: closed.fill(0, stream=stream),
        closed.__dlpack__,
        closed.__dlpack_device__,
        lambda: closed.device_id,
        lambda: closed.is_device_accessible,
        lambda: closed.is_host_accessible,
        lambda: closed.is_managed,
        lambda: closed.ipc_descriptor,
    ):
        with pytest.raises(RuntimeError, match="Buffer has been closed"):
            operation()

    with pytest.raises(RuntimeError, match="Buffer has been closed"):
        live.copy_to(closed, stream=stream)
    with pytest.raises(RuntimeError, match="Buffer has been closed"):
        live.copy_from(closed, stream=stream)
    live.close()


@pytest.mark.agent_authored(model="gpt-5.6")
def test_closed_memory_pool_rejected_before_active_operations(mempool_device):
    mr = DeviceMemoryResource(mempool_device)
    peer_access = mr.peer_accessible_by
    mr.close()

    assert mr.is_closed
    assert bool(mr) is True  # Preserve backward-compatible truthiness after close.
    for operation in (
        lambda: mr.allocate(16, stream=mempool_device.default_stream),
        lambda: mr.attributes,
        lambda: mr.peer_accessible_by,
        lambda: len(peer_access),
        lambda: mr.allocation_handle,
    ):
        with pytest.raises(RuntimeError, match="DeviceMemoryResource has been closed"):
            operation()


@pytest.mark.parametrize("buffer_type", [Buffer, ManagedBuffer])
def test_from_handle_mr_records_default_stream(buffer_type):
    """When a Buffer/ManagedBuffer is minted via :meth:`from_handle` with ``mr``
    but without an explicit ``stream=``, the deallocation stream is recorded at
    creation as ``default_stream()`` (not chosen later in the destructor).
    See `#2497`.
    """
    import gc

    device = Device()
    device.set_current()
    CapturingMR, telemetry = make_instrumented_memory_resource(record_streams=True)
    mr = CapturingMR(device)
    # ptr=1 is fine because StubMemoryResource.deallocate does not free.
    buf = buffer_type.from_handle(1, 1024, mr=mr)
    del buf
    gc.collect()

    assert telemetry["deallocations"], "deallocate was not invoked (callback raised and leaked)"
    assert telemetry["deallocations"][-1]["stream"].handle == default_stream().handle


@pytest.mark.agent_authored(model="cursor-grok-4.5")
@pytest.mark.parametrize("buffer_type", [Buffer, ManagedBuffer])
def test_from_handle_mr_records_explicit_stream(buffer_type):
    """Buffer/ManagedBuffer.from_handle(..., mr=mr, stream=s) stores s for teardown."""
    import gc

    device = Device()
    device.set_current()
    stream = device.create_stream()
    CapturingMR, telemetry = make_instrumented_memory_resource(record_streams=True)
    mr = CapturingMR(device)
    buf = buffer_type.from_handle(1, 1024, mr=mr, stream=stream)
    del buf
    gc.collect()

    assert telemetry["deallocations"][-1]["stream"].handle == stream.handle


@pytest.mark.agent_authored(model="cursor-grok-4.5")
@pytest.mark.parametrize("buffer_type", [Buffer, ManagedBuffer])
def test_from_handle_stream_requires_mr(buffer_type):
    device = Device()
    device.set_current()
    stream = device.create_stream()
    with pytest.raises(ValueError, match="stream requires a memory resource"):
        buffer_type.from_handle(1, 1024, stream=stream)


@pytest.mark.agent_authored(model="claude-sonnet-4-6")
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 = init_cuda
    stream = device.create_stream()
    mr = StubMemoryResource(device)
    # Use a real stream at creation so _init succeeds without a current context later.
    buf = Buffer.from_handle(1, 1024, mr=mr, stream=stream)

    with no_current_context():
        assert current_context_handle() == 0
        with pytest.raises(RuntimeError, match="no CUDA context is current"):
            buf.close(stream=default_stream())

    buf.close()  # clean up using the recorded stream (which carries a 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(init_cuda, buffer_type):
    """Owning from_handle with the default stream needs a current context."""
    device = init_cuda
    mr = StubMemoryResource(device)
    with no_current_context():
        assert current_context_handle() == 0
        with pytest.raises(RuntimeError, match="no CUDA context is current"):
            buffer_type.from_handle(1, 1024, mr=mr)


@pytest.mark.agent_authored(model="gpt-5.6")
@pytest.mark.parametrize("buffer_type", [Buffer, ManagedBuffer])
def test_from_handle_mr_explicit_stream_without_current_context(buffer_type):
    """A context-bound stream makes owning from_handle context-independent."""
    device = Device()
    device.set_current()
    stream = device.create_stream()
    CapturingMR, telemetry = make_instrumented_memory_resource(record_streams=True)
    mr = CapturingMR(device)
    with no_current_context():
        assert current_context_handle() == 0
        buf = buffer_type.from_handle(1, 1024, mr=mr, stream=stream)
        buf.close()
        assert current_context_handle() == 0

    assert telemetry["deallocations"][-1]["stream"].handle == stream.handle


_HOST_ONLY_MRS = [
    DummyHostMemoryResource,
    pytest.param(NumpyHostMemoryResource, marks=pytest.mark.skipif(np is None, reason="numpy is not installed")),
]


@pytest.mark.thread_unsafe(reason="records process-global warnings and mutates the context stack")
@pytest.mark.agent_authored(model="claude-fable-5-1")
@pytest.mark.parametrize("mr_cls", _HOST_ONLY_MRS)
def test_from_handle_host_only_mr_without_current_context(mr_cls):
    """Host-only memory needs no current context to create or free a Buffer."""
    device = Device()
    device.set_current()
    mr = mr_cls()

    previous = handle_return(driver.cuCtxPopCurrent())
    assert int(previous) != 0
    try:
        assert int(handle_return(driver.cuCtxGetCurrent())) == 0
        with assert_no_cuda_warning():
            buf = mr.allocate(64)
            assert buf.is_host_accessible
            buf.close()
        assert int(handle_return(driver.cuCtxGetCurrent())) == 0
    finally:
        handle_return(driver.cuCtxSetCurrent(previous))


def _host_only_child_main(mr_cls):
    """Allocate and free host-only memory in a process that never initialized CUDA."""
    # Warnings do not cross processes: check for a CUDAWarning here, where a
    # failed assertion becomes a non-zero exit code for the parent to see.
    with assert_no_cuda_warning():
        buf = mr_cls().allocate(64)
        assert buf.is_host_accessible
        buf.close()
    err, _ = driver.cuCtxGetCurrent()
    assert err == driver.CUresult.CUDA_ERROR_NOT_INITIALIZED, err


@pytest.mark.agent_authored(model="claude-fable-5-1")
@pytest.mark.parametrize("mr_cls", _HOST_ONLY_MRS)
def test_from_handle_host_only_mr_without_cuda_init(mr_cls):
    """Host-only buffers work in a spawned process that never initializes CUDA."""
    process = mp.Process(target=_host_only_child_main, args=(mr_cls,))
    process.start()
    process.join(timeout=CHILD_TIMEOUT_SEC)
    survivors = kill_subprocesses(process)
    assert not survivors, "child did not exit within timeout"
    assert process.exitcode == 0, f"child exited with {process.exitcode}"


@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(init_cuda):
    """Destructor-path MR failures are contained and reported as CUDAWarning."""
    device = init_cuda
    FailingMR, _ = make_instrumented_memory_resource(deallocate_error=RuntimeError("expected deallocation failure"))
    buf = Buffer.from_handle(1, 1024, mr=FailingMR(device))

    with pytest.warns(
        CUDAWarning,
        match=r"mr\.deallocate\(0x[0-9a-f]+\) failed during Buffer destruction.*expected deallocation failure",
    ):
        buf.close()


@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(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
    with a clean error state and restore the pending exception afterwards.
    Before the fix, reporting a failed ``deallocate`` cleared the pending
    exception, and the caller returned an error with no exception set, which
    Python reports as ``SystemError`` (found in review of #2917).
    """
    device = init_cuda
    FailingMR, _ = make_instrumented_memory_resource(deallocate_error=RuntimeError("expected deallocation failure"))
    mr = FailingMR(device)

    def reject(buf):
        raise TypeError("rejected")

    with pytest.warns(CUDAWarning, match="expected deallocation failure"), pytest.raises(TypeError, match="rejected"):
        # The temporary Buffer is released while TypeError propagates.
        reject(Buffer.from_handle(1, 1024, mr=mr))


@pytest.mark.thread_unsafe(reason="records process-global warnings and mutates the context stack")
@pytest.mark.agent_authored(model="cursor-grok-4.5")
@pytest.mark.parametrize("replace_stream", [False, True])
def test_mr_deallocation_without_current_context(init_cuda, replace_stream):
    """MR-backed Buffer teardown activates the recorded context when none is current."""
    TrackingMR, telemetry = make_instrumented_memory_resource(DummyDeviceMemoryResource, track_active=True)
    mr = TrackingMR(init_cuda)
    buf = mr.allocate(1024)
    stream = init_cuda.create_stream() if replace_stream else None
    assert len(telemetry["active"]) == 1

    with no_current_context():
        assert current_context_handle() == 0

        with assert_no_cuda_warning():
            buf.close(stream)

        assert len(telemetry["active"]) == 0
        assert current_context_handle() == 0


@pytest.mark.thread_unsafe(reason="records process-global warnings and mutates the context stack")
@pytest.mark.agent_authored(model="cursor-grok-4.5")
@pytest.mark.parametrize("replace_stream", [False, True])
def test_mr_deallocation_with_foreign_context(device_x2, replace_stream):
    """MR-backed Buffer teardown switches away from an unrelated current context."""
    alloc_dev, foreign_dev = device_x2
    alloc_dev.set_current()
    TrackingMR, telemetry = make_instrumented_memory_resource(DummyDeviceMemoryResource, track_active=True)
    mr = TrackingMR(alloc_dev)
    buf = mr.allocate(1024)
    stream = alloc_dev.create_stream() if replace_stream else None
    assert len(telemetry["active"]) == 1
    alloc_ctx = current_context_handle()

    foreign_dev.set_current()
    foreign_ctx = current_context_handle()
    assert foreign_ctx != 0
    assert foreign_ctx != alloc_ctx

    try:
        with assert_no_cuda_warning():
            buf.close(stream)

        assert len(telemetry["active"]) == 0
        assert current_context_handle() == foreign_ctx
    finally:
        alloc_dev.set_current()


@pytest.mark.agent_authored(model="claude-opus-5")
def test_mr_deallocate_raises_on_driver_error(mempool_device):
    """An explicit mr.deallocate() call propagates driver errors to the caller.

    Buffer teardown must not raise, so the containment lives in the destruction
    callback rather than in deallocate() itself. See `#2497`.
    """
    dev = mempool_device
    stream = dev.create_stream()
    mr = DeviceMemoryResource(dev, DeviceMemoryResourceOptions(max_size=POOL_SIZE))

    with pytest.raises(CUDAError):
        mr.deallocate(0xDEADBEEF, 256, stream=stream)


@pytest.mark.thread_unsafe(reason="records process-global warnings and mutates the context stack")
@pytest.mark.agent_authored(model="cursor-grok-4.5")
def test_pool_buffer_deallocates_without_current_context(mempool_device):
    """Pool Buffer.close frees on the recorded stream with no current context."""
    dev = mempool_device
    stream = dev.create_stream()
    mr = DeviceMemoryResource(dev, DeviceMemoryResourceOptions(max_size=POOL_SIZE))
    size = 256
    buf = mr.allocate(size, stream=stream)
    stream.sync()
    used_after_alloc = mr.attributes.used_mem_current

    with no_current_context():
        assert current_context_handle() == 0

        with assert_no_cuda_warning():
            buf.close()
        stream.sync()

        assert mr.attributes.used_mem_current < used_after_alloc
        assert current_context_handle() == 0


@pytest.mark.thread_unsafe(reason="records process-global warnings and mutates the context stack")
@pytest.mark.agent_authored(model="cursor-grok-4.5")
def test_pool_buffer_deallocates_with_foreign_context(mempool_device_x2):
    """Pool Buffer.close frees under the recorded context while another is current."""
    alloc_dev, foreign_dev = mempool_device_x2
    alloc_dev.set_current()
    stream = alloc_dev.create_stream()
    mr = DeviceMemoryResource(alloc_dev, DeviceMemoryResourceOptions(max_size=POOL_SIZE))
    size = 256
    buf = mr.allocate(size, stream=stream)
    stream.sync()
    used_after_alloc = mr.attributes.used_mem_current
    alloc_ctx = current_context_handle()

    foreign_dev.set_current()
    foreign_ctx = current_context_handle()
    assert foreign_ctx != 0
    assert foreign_ctx != alloc_ctx

    try:
        with assert_no_cuda_warning():
            buf.close()
        assert current_context_handle() == foreign_ctx

        # Observe the free on the allocation device, then restore the foreign context.
        alloc_dev.set_current()
        stream.sync()
        assert mr.attributes.used_mem_current < used_after_alloc
        foreign_dev.set_current()
    finally:
        alloc_dev.set_current()


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(init_cuda), owner=a)


def test_owner_close():
    a = (ctypes.c_byte * 20)()
    ptr = ctypes.addressof(a)
    before = sys.getrefcount(a)
    buffer = Buffer.from_handle(ptr, 20, owner=a)
    assert sys.getrefcount(a) != before
    buffer.close()
    after = sys.getrefcount(a)
    assert after == before


def test_buffer_dunder_dlpack(init_cuda):
    device = init_cuda
    dummy_mr = DummyDeviceMemoryResource(device)
    buffer = dummy_mr.allocate(size=1024)
    capsule = buffer.__dlpack__()
    assert "dltensor" in repr(capsule)
    capsule = buffer.__dlpack__(max_version=(1, 0))
    assert "dltensor" in repr(capsule)
    with pytest.raises(BufferError, match=r"^Sorry, not supported: dl_device other than None$"):
        buffer.__dlpack__(dl_device=())
    with pytest.raises(BufferError, match=r"^Sorry, not supported: copy=True$"):
        buffer.__dlpack__(copy=True)
    with pytest.raises(BufferError, match=r"^Expected max_version tuple\[int, int\], got \(\)$"):
        buffer.__dlpack__(max_version=())
    with pytest.raises(BufferError, match=r"^Expected max_version tuple\[int, int\], got \(9, 8, 7\)$"):
        buffer.__dlpack__(max_version=(9, 8, 7))


@pytest.mark.parametrize(
    ("DummyMR", "expected"),
    [
        (DummyDeviceMemoryResource, (DLDeviceType.kDLCUDA, 0)),
        (DummyHostMemoryResource, (DLDeviceType.kDLCPU, 0)),
        (DummyUnifiedMemoryResource, (DLDeviceType.kDLCUDAManaged, 0)),
        (DummyPinnedMemoryResource, (DLDeviceType.kDLCUDAHost, 0)),
    ],
)
def test_buffer_dunder_dlpack_device_success(DummyMR, expected):
    device = Device()
    device.set_current()
    dummy_mr = DummyMR() if DummyMR is DummyHostMemoryResource else DummyMR(device)
    buffer = dummy_mr.allocate(size=1024)
    assert buffer.__dlpack_device__() == expected


def test_buffer_dunder_dlpack_device_failure(init_cuda):
    # avoids an error capturing the default stream with no context
    # (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(init_cuda):
    # avoids an error capturing the default stream with no context
    # (init_cuda has already set a current context)
    dummy_mr = NullMemoryResource()
    buffer = dummy_mr.allocate(size=1024)
    before = sys.getrefcount(buffer)
    with pytest.raises(BufferError, match="buffer is neither device-accessible nor host-accessible"):
        buffer.__dlpack__()
    after = sys.getrefcount(buffer)
    # we use the buffer refcount as sentinel for proper clean-up here,
    # hoping that malloc and frees did the right thing
    # as they are handled by the same deleter
    assert after == before


def test_managed_buffer_dlpack_roundtrip_device_type(init_cuda):
    """Verify that a managed Buffer round-trips through DLPack with kDLCUDAManaged."""
    device = init_cuda
    skip_if_managed_memory_unsupported(device)
    mr = DummyUnifiedMemoryResource(device)
    buf = mr.allocate(size=1024)

    # Buffer-level classification should report managed.
    assert buf.__dlpack_device__() == (DLDeviceType.kDLCUDAManaged, 0)

    # The end-to-end path: Buffer -> DLPack capsule -> StridedMemoryView
    # must preserve kDLCUDAManaged rather than downgrading to kDLCUDAHost.
    view = StridedMemoryView.from_any_interface(buf, stream_ptr=-1)
    assert view.__dlpack_device__() == (int(DLDeviceType.kDLCUDAManaged), 0)


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
    on a CU_MEM_ALLOCATION_TYPE_MANAGED pool, which does not set
    CU_POINTER_ATTRIBUTE_IS_MANAGED.  The Buffer must still report itself as
    managed via its memory resource so that DLPack consumers (e.g. CCCL's
    make_tma_descriptor) accept the buffer.
    """
    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)

    assert mr.is_managed
    assert buf.is_managed
    assert buf.__dlpack_device__() == (DLDeviceType.kDLCUDAManaged, 0)

    view = StridedMemoryView.from_any_interface(buf, stream_ptr=-1)
    assert view.__dlpack_device__() == (int(DLDeviceType.kDLCUDAManaged), 0)


@pytest.mark.agent_authored(model="claude-opus-5-5")
def test_managed_memory_resource_allocate_rejects_graph_builder(init_cuda):
    device = init_cuda
    skip_if_managed_memory_unsupported(device)
    mr = None
    gb = None
    try:
        mr = create_managed_memory_resource_or_skip(ManagedMemoryResourceOptions(preferred_location=device.device_id))
        gb = device.create_graph_builder()
        with pytest.raises(TypeError, match="Only Stream is supported .*got GraphBuilder"):
            mr.allocate(64, stream=gb)
    finally:
        if gb is not None:
            gb.close()
        if mr is not None:
            mr.close()


@pytest.mark.parametrize("mr_kind", ["device", "pinned"])
def test_non_managed_resources_report_not_managed(init_cuda, mr_kind):
    """Non-managed memory resources must report is_managed=False."""
    device = init_cuda
    if not device.properties.memory_pools_supported:
        pytest.skip("Device does not support mempool operations")
    if mr_kind == "device":
        mr = DeviceMemoryResource(device)
    else:
        skip_if_pinned_memory_unsupported(device)
        mr = create_pinned_memory_resource_or_xfail(xfail_device=device)
    assert mr.is_managed is False
    buf = (
        _allocate_pinned_buffer_or_xfail(mr, 1024, device=device)
        if mr_kind == "pinned"
        else mr.allocate(1024, stream=device.default_stream)
    )
    assert buf.is_managed is False
    buf.close()


@pytest.mark.parametrize("use_device_object", [True, False])
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 = init_cuda

    if not device.properties.memory_pools_supported:
        pytest.skip("Device does not support mempool operations")

    # 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
    mr = DeviceMemoryResource(device_arg)

    # Verify basic properties
    assert mr.device_id == device.device_id
    assert mr.is_device_accessible
    assert not mr.is_host_accessible
    assert not mr.is_ipc_enabled

    # Test allocation/deallocation works
    buffer = mr.allocate(1024, stream=device.default_stream)
    assert buffer.size == 1024
    assert buffer.device_id == device.device_id
    buffer.close()


def test_pinned_memory_resource_initialization(init_cuda):
    device = Device()
    skip_if_pinned_memory_unsupported(device)

    device.set_current()

    mr = create_pinned_memory_resource_or_xfail(xfail_device=device)
    assert mr.is_device_accessible
    assert mr.is_host_accessible

    # Test allocation/deallocation works
    buffer = _allocate_pinned_buffer_or_xfail(mr, 1024, device=device)
    assert buffer.size == 1024
    assert buffer.device_id == -1  # Not bound to any GPU
    assert buffer.is_host_accessible
    assert buffer.memory_resource == mr
    assert buffer.is_device_accessible
    buffer.close()


@pytest.mark.agent_authored(model="cursor-grok-4.5")
def test_pinned_memory_resource_rejects_unsupported_host_pool(init_cuda):
    """allocate() must fail on devices without host memory pool support (see #2486)."""
    if driver_version() < (13, 0, 0):
        pytest.skip("Generic HOST memory pools require CUDA 13.0 or later")

    device = init_cuda
    if device.properties.host_memory_pools_supported:
        pytest.skip("Device supports host memory pools")

    try:
        mr = PinnedMemoryResource(PinnedMemoryResourceOptions(max_size=POOL_SIZE))
    except CUDAError as exc:
        if "CUDA_ERROR_NOT_SUPPORTED" in str(exc):
            pytest.skip("PinnedMemoryResource is not supported on this platform/device")
        raise
    try:
        with pytest.raises(RuntimeError, match="does not support.*LegacyPinnedMemoryResource"):
            mr.allocate(1024, stream=device.default_stream)
    finally:
        mr.close()


def test_managed_memory_resource_initialization(init_cuda):
    device = Device()
    skip_if_managed_memory_unsupported(device)

    device.set_current()

    mr = create_managed_memory_resource_or_skip()
    assert mr.is_device_accessible
    assert mr.is_host_accessible

    # Test allocation/deallocation works
    buffer = mr.allocate(1024, stream=device.default_stream)
    assert buffer.size == 1024
    assert buffer.is_host_accessible  # But accessible from host
    assert buffer.memory_resource == mr
    assert buffer.is_device_accessible
    buffer.close()


def get_handle_type():
    if IS_WINDOWS:
        return (("win32", None), ("win32_kmt", None))
    else:
        return (("posix_fd", None),)


MIB = 1024 * 1024


def _vmm_device_or_skip():
    device = Device()
    device.set_current()
    if not device.properties.virtual_memory_management_supported:
        pytest.skip("Virtual memory management is not supported on this device")
    return device


def _vmm_resource(device, **options):
    options.setdefault("handle_type", VMM_HANDLE_TYPE)
    return VirtualMemoryResource(device, config=VirtualMemoryResourceOptions(**options))


def _vmm_fill(buf, value, *, offset=0, size=None):
    """Write ``value`` to ``size`` bytes of ``buf`` at ``offset`` and wait for the write."""
    size = buf.size - offset if size is None else size
    handle_return(driver.cuMemsetD8(int(buf.handle) + offset, value, size))
    handle_return(driver.cuCtxSynchronize())


def _vmm_read(buf, offset, size):
    """Copy ``size`` bytes of ``buf`` at ``offset`` to the host."""
    host = (ctypes.c_ubyte * size)()
    handle_return(driver.cuMemcpyDtoH(ctypes.addressof(host), int(buf.handle) + offset, size))
    return bytes(host)


def _vmm_assert_released(reservations):
    """Assert that no mapping and no reservation remain for each ``(ptr, size)`` reservation.

    Both queries ask the driver about one address, so no other process can
    disturb them, unlike a device-wide free-memory count. A mapping lookup
    fails once the range is unmapped, and freeing an address range fails once
    it is no longer reserved. With every mapping gone and every buffer's own
    reference released, the driver frees the physical memory.
    """
    for ptr, size in reservations:
        status, handle = driver.cuMemRetainAllocationHandle(ptr)
        if status == driver.CUresult.CUDA_SUCCESS:
            handle_return(driver.cuMemRelease(handle))
            pytest.fail(f"address {ptr:#x} is still mapped")
        (status,) = driver.cuMemAddressFree(ptr, size)
        assert status != driver.CUresult.CUDA_SUCCESS, f"range {ptr:#x}+{size:#x} was still reserved"


def _vmm_reserve_at(addr, size):
    """Reserve ``size`` bytes at ``addr``. Return the reservation, or None if the driver placed it elsewhere."""
    ptr = handle_return(driver.cuMemAddressReserve(size, 0, addr, 0))
    if int(ptr) != addr:
        handle_return(driver.cuMemAddressFree(ptr, size))
        return None
    return ptr


def _vmm_allocate_in_free_hole(device, size, extra):
    """Allocate ``size`` bytes at the start of a free address range ``extra`` bytes longer.

    Reserving and freeing a range finds free address space; a resource with
    that address as its hint then places the buffer there, so the range right
    after the buffer is known to be free. Sizes round up to the granularity.
    Skips when the driver places the buffer elsewhere. Returns the resource
    and the buffer.
    """
    probe = _vmm_resource(device).allocate(1)
    gran = probe.size
    probe.close()
    size = -(-size // gran) * gran
    total = size + -(-extra // gran) * gran
    hole = handle_return(driver.cuMemAddressReserve(total, 0, 0, 0))
    handle_return(driver.cuMemAddressFree(hole, total))
    mr = _vmm_resource(device, addr_hint=int(hole))
    buf = mr.allocate(size)
    if int(buf.handle) != int(hole):
        buf.close()
        pytest.skip("the driver did not place the allocation at the requested address")
    return mr, buf


@pytest.mark.parametrize("use_device_object", [True, False])
@pytest.mark.parametrize("handle_type", get_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
    using CUDA VMM APIs with default configuration.
    """
    device = _vmm_device_or_skip()

    handle_type, security_attribute = handle_type  # unpack
    options = VirtualMemoryResourceOptions(handle_type=handle_type)
    # Create VMM allocator with default config
    device_arg = device if use_device_object else device.device_id
    vmm_mr = VirtualMemoryResource(device_arg, config=options)

    # Test basic allocation
    try:
        buffer = vmm_mr.allocate(4096)
    except NotImplementedError:
        assert handle_type == "win32"
        return
    assert isinstance(buffer, VirtualMemoryBuffer)
    assert buffer.size >= 4096  # May be aligned up
    assert buffer.device_id == device.device_id
    assert buffer.memory_resource == vmm_mr
    # POSIX file descriptors can be shared; Win32 KMT handles have no transport.
    assert vmm_mr.is_ipc_enabled is (handle_type == "posix_fd")

    # Test deallocation
    buffer.close()
    assert buffer.size == 0

    # Test multiple allocations
    buffers = []
    for i in range(5):
        buf = vmm_mr.allocate(1024 * (i + 1))
        buffers.append(buf)
        assert buf.size >= 1024 * (i + 1)

    # Clean up
    for buf in buffers:
        buf.close()


@pytest.mark.agent_authored(model="claude-fable-5-1")
def test_vmm_allocator_policy_configuration(init_cuda):
    """Test VMM allocator with different policy configurations.

    The resource applies its configuration to every allocation. A
    configuration passed to ``modify_allocation`` applies to the chunk that
    call adds, and the resource keeps its own (#2909).
    """
    device = _vmm_device_or_skip()

    # Skip if GPU Direct RDMA is not supported
    if not device.properties.gpu_direct_rdma_supported:
        pytest.skip("This test requires a device that supports GPU Direct RDMA")

    # Test with custom VMM config
    custom_config = VirtualMemoryResourceOptions(
        allocation_type=VirtualMemoryAllocationType.PINNED,
        location_type=VirtualMemoryLocationType.DEVICE,
        granularity=VirtualMemoryGranularityType.MINIMUM,
        gpu_direct_rdma=True,
        handle_type=VirtualMemoryHandleType.POSIX_FD if not IS_WINDOWS else VirtualMemoryHandleType.WIN32_KMT,
        peers=(),
        self_access=VirtualMemoryAccessType.READ_WRITE,
        peer_access=VirtualMemoryAccessType.READ_WRITE,
    )

    vmm_mr = VirtualMemoryResource(device, config=custom_config)

    # Verify configuration is applied
    assert vmm_mr.config == custom_config
    assert vmm_mr.config.gpu_direct_rdma is True
    assert vmm_mr.config.granularity == "minimum"

    # Test allocation with custom config
    try:
        buffer = vmm_mr.allocate(8192)
    except CUDAError as exc:
        msg = str(exc)
        if "CUDA_ERROR_INVALID_DEVICE" in msg:
            pytest.xfail("TODO(#1300): Failing on Jetson AGX Orin P3730")
        raise
    assert buffer.size >= 8192
    assert buffer.device_id == device.device_id

    # A read-only configuration for the chunk the grow adds.
    new_config = VirtualMemoryResourceOptions(
        allocation_type="pinned",
        location_type="device",
        granularity="recommended",
        gpu_direct_rdma=False,
        handle_type=VMM_HANDLE_TYPE,
        peers=(),
        self_access="r",  # Read-only access
        peer_access="r",
    )

    try:
        grown = vmm_mr.modify_allocation(buffer, buffer.size + 8192, config=new_config)
    except CUDAError as exc:
        msg = str(exc)
        if "CUDA_ERROR_UNKNOWN" in msg:
            pytest.xfail("TODO(#1300): Known to fail already with CTK 13.0 (Windows)")
        raise
    assert grown.size >= buffer.size + 8192
    # The resource keeps its own configuration.
    assert vmm_mr.config == custom_config
    # The chunk the input already mapped keeps read-write access; the chunk
    # the grow added has the read-only access the per-call configuration asked for.
    location = driver.CUmemLocation()
    location.type = driver.CUmemLocationType.CU_MEM_LOCATION_TYPE_DEVICE
    location.id = device.device_id
    access = driver.CUmemAccess_flags
    assert (
        handle_return(driver.cuMemGetAccess(location, int(grown.handle))) == access.CU_MEM_ACCESS_FLAGS_PROT_READWRITE
    )
    assert (
        handle_return(driver.cuMemGetAccess(location, int(grown.handle) + buffer.size))
        == access.CU_MEM_ACCESS_FLAGS_PROT_READ
    )
    _vmm_fill(grown, 0x42, size=buffer.size)
    assert _vmm_read(grown, 0, 16) == bytes([0x42]) * 16
    assert len(_vmm_read(grown, buffer.size, 16)) == 16
    # A later allocation uses the resource's configuration, which is writable.
    later = vmm_mr.allocate(4096)
    _vmm_fill(later, 0x24)

    # Clean up
    later.close()
    buffer.close()
    grown.close()


@pytest.mark.agent_authored(model="claude-fable-5-1")
@pytest.mark.parametrize("handle_type", get_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
    stays open. A request the buffer already covers returns the buffer itself.
    """
    device = _vmm_device_or_skip()

    handle_type, security_attribute = handle_type  # unpack
    options = VirtualMemoryResourceOptions(handle_type=handle_type)
    vmm_mr = VirtualMemoryResource(device, config=options)

    # Create initial allocation
    try:
        buffer = vmm_mr.allocate(2 * MIB)
    except NotImplementedError:
        assert handle_type == "win32"
        return
    original_size = buffer.size

    # Grow the allocation
    grown_buffer = vmm_mr.modify_allocation(buffer, 4 * MIB)

    # Verify growth
    assert grown_buffer is not buffer
    assert grown_buffer.size >= 4 * MIB
    assert grown_buffer.size > original_size
    # The input stays open, at its old size and address.
    assert buffer.size == original_size
    assert int(buffer.handle) != 0

    # Requests the buffer already covers return a full alias, never the buffer
    # itself, so closing the result never closes the input.
    same = vmm_mr.modify_allocation(grown_buffer, 4 * MIB)
    smaller = vmm_mr.modify_allocation(grown_buffer, 2 * MIB)
    assert same is not grown_buffer and smaller is not grown_buffer
    assert same.size == smaller.size == grown_buffer.size
    assert int(same.handle) == int(smaller.handle) == int(grown_buffer.handle)
    same.close()
    smaller.close()
    _vmm_fill(grown_buffer, 0x11)  # the input is still mapped

    # Clean up
    buffer.close()
    grown_buffer.close()


@pytest.mark.agent_authored(model="claude-fable-5-1")
def test_vmm_grow_in_place_keeps_pointer(init_cuda):
    """With free address space after the buffer, a grow extends it in place and keeps the pointer."""
    device = _vmm_device_or_skip()
    mr, buf = _vmm_allocate_in_free_hole(device, 2 * MIB, 4 * MIB)
    size = buf.size
    grown = mr.modify_allocation(buf, 2 * size)
    if int(grown.handle) != int(buf.handle):
        grown.close()
        buf.close()
        pytest.skip("the driver did not grant the adjacent address range")
    assert grown.size == 2 * size
    _vmm_fill(grown, 0x10)
    assert _vmm_read(buf, 0, size) == bytes([0x10]) * size
    buf.close()
    grown.close()


@pytest.mark.agent_authored(model="claude-fable-5-1")
def test_vmm_grow_preserves_contents_and_aliases_input(init_cuda):
    """A grown buffer sees the input's contents, and a write through either buffer is visible through the other."""
    device = _vmm_device_or_skip()
    mr = _vmm_resource(device)
    buf = mr.allocate(2 * MIB)
    size = buf.size
    _vmm_fill(buf, 0xAB)

    grown = mr.modify_allocation(buf, size + 2 * MIB)
    assert isinstance(grown, VirtualMemoryBuffer)
    assert grown.size == 2 * size
    assert _vmm_read(grown, 0, size) == bytes([0xAB]) * size

    _vmm_fill(grown, 0xCD, size=size)
    assert _vmm_read(buf, 0, size) == bytes([0xCD]) * size
    _vmm_fill(grown, 0xEF, offset=size)
    assert _vmm_read(grown, size, size) == bytes([0xEF]) * size

    # Closing the input first leaves the result usable.
    buf.close()
    assert _vmm_read(grown, 0, 16) == bytes([0xCD]) * 16
    grown.close()


@pytest.mark.agent_authored(model="claude-fable-5-1")
@pytest.mark.thread_unsafe(reason="records process-global warnings")
def test_vmm_grow_moves_when_adjacent_range_is_taken(init_cuda):
    """When the address range after the buffer is taken, the grow maps the existing memory at a new address.

    Neither the grow nor the closes emit a warning (#2877).
    """
    device = _vmm_device_or_skip()
    mr, buf = _vmm_allocate_in_free_hole(device, 2 * MIB, 4 * MIB)
    size = buf.size
    _vmm_fill(buf, 0x5A)

    decoy = _vmm_reserve_at(int(buf.handle) + size, size)
    if decoy is None:
        buf.close()
        pytest.skip("the driver did not grant a reservation right after the buffer")
    with assert_no_cuda_warning():
        try:
            grown = mr.modify_allocation(buf, 2 * size)
        finally:
            handle_return(driver.cuMemAddressFree(decoy, size))

        # The range after the buffer was taken, so the grow had to move.
        assert int(grown.handle) != int(buf.handle)
        assert grown.size == 2 * size
        assert _vmm_read(grown, 0, size) == bytes([0x5A]) * size
        _vmm_fill(buf, 0x3C)
        assert _vmm_read(grown, 0, size) == bytes([0x3C]) * size

        # Closing the result first leaves the input usable.
        grown.close()
        assert _vmm_read(buf, 0, 16) == bytes([0x3C]) * 16
        buf.close()


@pytest.mark.agent_authored(model="claude-fable-5-1")
@pytest.mark.thread_unsafe(reason="records process-global warnings")
def test_vmm_unaligned_and_repeated_grow_close_cleanly(init_cuda):
    """An unaligned request rounds up to the granularity (#2907), a grown buffer
    grows again (#2908), and every alias closes without a warning."""
    device = _vmm_device_or_skip()
    mr = _vmm_resource(device)
    with assert_no_cuda_warning():
        buf = mr.allocate(1)
        gran = buf.size
        assert gran >= 1

        second = mr.modify_allocation(buf, gran + 1)
        assert second.size == 2 * gran
        third = mr.modify_allocation(second, 3 * gran)
        assert third.size == 3 * gran

        # A second grow of the input builds its own result. The input's range
        # is immutable, so this grow cannot reuse the chunk ``second`` added.
        alias = mr.modify_allocation(buf, 2 * gran)
        assert alias is not buf
        assert alias.size == 2 * gran

        # Every buffer maps the first chunk.
        _vmm_fill(third, 0x77)
        assert _vmm_read(alias, 0, gran) == bytes([0x77]) * gran

        third.close()
        buf.close()
        alias.close()
        second.close()


@pytest.mark.agent_authored(model="claude-fable-5-1")
@pytest.mark.thread_unsafe(reason="spawns worker threads that grow shared buffers")
def test_vmm_concurrent_grows_of_aliased_buffers_are_memory_safe(init_cuda):
    """Growing and closing aliases of one buffer from several threads never crashes.

    Reduced from the reproduction on PR #2917. A range is immutable, so a grow
    reads the input's range and builds a new one, and ``modify_allocation``
    never returns its input, so closing a result never closes a buffer another
    thread is using. Which grow extends in place and which one moves depends on
    the address space the driver has free, as for one thread; the process must
    survive and every buffer must close cleanly.
    """
    device = _vmm_device_or_skip()
    mr = _vmm_resource(device)
    nthreads = 4
    errors = []

    def work(buf, i, gate):
        try:
            gate.wait()
            for j in range(6):
                mr.modify_allocation(buf, (i * 8 + j + 2) * 2 * MIB).close()
        except BaseException as exc:  # pragma: no cover - surfaced below
            errors.append(exc)

    with assert_no_cuda_warning():
        for _ in range(8):
            root = mr.allocate(2 * MIB)
            alias = mr.modify_allocation(root, 8 * MIB)
            gate = threading.Barrier(nthreads)
            threads = [threading.Thread(target=work, args=(root if i % 2 else alias, i, gate)) for i in range(nthreads)]
            for t in threads:
                t.start()
            for t in threads:
                t.join(timeout=CHILD_TIMEOUT_SEC)
            assert not any(t.is_alive() for t in threads), "a grow thread did not finish"
            alias.close()
            root.close()
    assert errors == []


@pytest.mark.agent_authored(model="claude-fable-5-1")
def test_vmm_modify_allocation_rejects_foreign_or_closed_buffers(init_cuda):
    """modify_allocation accepts only open buffers this resource returned."""
    device = _vmm_device_or_skip()
    mr = _vmm_resource(device)
    other_mr = _vmm_resource(device)
    foreign = DummyDeviceMemoryResource(device).allocate(64)
    sibling = other_mr.allocate(4096)
    closed = mr.allocate(4096)
    closed.close()
    try:
        with pytest.raises(TypeError):
            mr.modify_allocation(foreign, 4096)
        with pytest.raises(TypeError):
            mr.modify_allocation(sibling, 8192)
        with pytest.raises(RuntimeError, match="closed"):
            mr.modify_allocation(closed, 8192)
    finally:
        foreign.close()
        sibling.close()


@pytest.mark.agent_authored(model="claude-fable-5-1")
def test_vmm_modify_allocation_validates_per_call_config(init_cuda):
    """A per-call config passes the constructor's option checks and must keep the resource's location."""
    device = _vmm_device_or_skip()
    mr = _vmm_resource(device)
    buf = mr.allocate(4096)
    ptr, size = int(buf.handle), buf.size
    try:
        with pytest.raises(ValueError, match="location_type"):
            mr.modify_allocation(
                buf, 2 * size, config=VirtualMemoryResourceOptions(location_type="host", handle_type=None)
            )
        if not device.properties.gpu_direct_rdma_supported:
            with pytest.raises(RuntimeError, match="GPU Direct RDMA"):
                mr.modify_allocation(
                    buf,
                    2 * size,
                    config=VirtualMemoryResourceOptions(handle_type=VMM_HANDLE_TYPE, gpu_direct_rdma=True),
                )
        # The rejected requests leave the buffer untouched.
        assert int(buf.handle) == ptr
        assert buf.size == size
    finally:
        buf.close()


@pytest.mark.agent_authored(model="claude-fable-5-1")
def test_vmm_host_modify_allocation_rejects_exportable_handle_type(init_cuda):
    """On a host-located resource, a per-call config with an exportable handle type is rejected before any driver call."""
    device = Device()
    if not device.properties.host_virtual_memory_management_supported:
        pytest.skip("Host virtual memory management is not supported on this device")
    mr = VirtualMemoryResource(device, config=VirtualMemoryResourceOptions(location_type="host", handle_type=None))
    buf = mr.allocate(4096)
    try:
        with pytest.raises(ValueError, match="handle_type=None"):
            mr.modify_allocation(buf, 2 * buf.size, config=VirtualMemoryResourceOptions(location_type="host"))
    finally:
        buf.close()


@pytest.mark.agent_authored(model="claude-fable-5-1")
@pytest.mark.thread_unsafe(reason="records process-global warnings")
def test_vmm_allocate_zero_size(init_cuda):
    """allocate(0) returns an empty buffer without a driver call; a grow of it inherits its stream."""
    device = _vmm_device_or_skip()
    mr = _vmm_resource(device)
    s = device.create_stream()
    try:
        buf = mr.allocate(0, stream=s)
        assert isinstance(buf, VirtualMemoryBuffer)
        assert buf.size == 0
        assert int(buf.handle) == 0
        grown = mr.modify_allocation(buf, 4096)
        assert grown.size >= 4096
        # The grown buffer's deallocation is ordered on the stream passed to
        # allocate(0): its close waits for the work queued there. The sleep
        # kernel does not touch the buffer, so a missing wait fails the event
        # check instead of faulting. The close is wrapped so a skipped sync
        # surfaces as a CUDAWarning instead of a bare event-check failure.
        NanosleepKernel(device, sleep_duration_ms=200).launch(s)
        done = s.record()
        with assert_no_cuda_warning():
            grown.close()
        assert done.is_done
        buf.close()
    finally:
        s.sync()
        s.close()


@pytest.mark.agent_authored(model="claude-fable-5-1")
def test_vmm_allocate_without_access_descriptors(init_cuda):
    """self_access=None with no peers creates and maps memory; no access call is made."""
    device = _vmm_device_or_skip()
    mr = _vmm_resource(device, self_access=None)
    buf = mr.allocate(4096)
    assert buf.size >= 4096
    grown = mr.modify_allocation(buf, 2 * buf.size)
    grown.close()
    buf.close()


@pytest.mark.agent_authored(model="claude-fable-5-1")
def test_vmm_close_on_capturing_stream_raises(init_cuda):
    """An explicit close on a capturing stream raises and leaves the buffer open."""
    device = _vmm_device_or_skip()
    mr = _vmm_resource(device)
    buf = mr.allocate(4096)
    gb = device.create_graph_builder().begin_building()
    try:
        with pytest.raises(RuntimeError, match="capturing stream"):
            buf.close(gb)
    finally:
        gb.end_building()
        gb.close()
    assert int(buf.handle) != 0
    buf.close()


@pytest.mark.agent_authored(model="claude-fable-5-1")
@pytest.mark.thread_unsafe(reason="records process-global warnings")
@pytest.mark.parametrize("close_input_first", [True, False])
def test_vmm_close_synchronizes_recorded_streams(init_cuda, close_input_first):
    """The closes that unmap wait for the work queued on every recorded stream (#2886).

    A sleep kernel holds each stream. Whether the grow shared the range (one
    close synchronizes both streams) or moved it (each close synchronizes its
    own stream), both streams must be idle once both buffers are closed. The
    kernels do not touch the buffers, so a missing wait fails the event checks
    instead of faulting.
    """
    device = _vmm_device_or_skip()
    mr = _vmm_resource(device)
    sleeper = NanosleepKernel(device, sleep_duration_ms=200)
    s1 = device.create_stream()
    s2 = device.create_stream()
    try:
        buf = mr.allocate(8 * MIB, stream=s1)
        grown = mr.modify_allocation(buf, 2 * buf.size)
        grown.set_deallocation_stream(s2)
        first, second = (buf, grown) if close_input_first else (grown, buf)
        with assert_no_cuda_warning():
            sleeper.launch(s1)
            done1 = s1.record()
            sleeper.launch(s2)
            done2 = s2.record()
            first.close()
            second.close()
        assert done1.is_done
        assert done2.is_done
    finally:
        s1.sync()
        s2.sync()
        s1.close()
        s2.close()


@pytest.mark.agent_authored(model="claude-fable-5-1")
@pytest.mark.thread_unsafe(reason="records process-global warnings")
def test_vmm_deallocate_frees_wrapped_pointer(init_cuda):
    """deallocate() unmaps and frees one reservation wrapped with Buffer.from_handle."""
    device = _vmm_device_or_skip()
    mr = _vmm_resource(device)
    prop = driver.CUmemAllocationProp()
    prop.type = driver.CUmemAllocationType.CU_MEM_ALLOCATION_TYPE_PINNED
    prop.location.type = driver.CUmemLocationType.CU_MEM_LOCATION_TYPE_DEVICE
    prop.location.id = device.device_id
    # The helper returns a plain int; the CUDA 13.0 bindings' struct setter
    # accepts only the enum.
    prop.requestedHandleTypes = driver.CUmemAllocationHandleType(
        VirtualMemoryResourceOptions._handle_type_to_driver(VMM_HANDLE_TYPE)
    )
    size = handle_return(
        driver.cuMemGetAllocationGranularity(
            prop, driver.CUmemAllocationGranularity_flags.CU_MEM_ALLOC_GRANULARITY_RECOMMENDED
        )
    )
    desc = driver.CUmemAccessDesc()
    desc.location.type = prop.location.type
    desc.location.id = prop.location.id
    desc.flags = driver.CUmemAccess_flags.CU_MEM_ACCESS_FLAGS_PROT_READWRITE

    handle = handle_return(driver.cuMemCreate(size, prop, 0))
    ptr = handle_return(driver.cuMemAddressReserve(size, 0, 0, 0))
    handle_return(driver.cuMemMap(ptr, size, 0, handle, 0))
    handle_return(driver.cuMemSetAccess(ptr, size, [desc], 1))
    handle_return(driver.cuMemRelease(handle))  # the mapping keeps the memory alive

    buf = Buffer.from_handle(ptr=int(ptr), size=size, mr=mr)
    _vmm_fill(buf, 0x11)
    with assert_no_cuda_warning():
        buf.close()
    _vmm_assert_released([(int(ptr), size)])


@pytest.mark.agent_authored(model="gpt-6-sol")
def test_vmm_modify_allocation_rejects_non_buffer(init_cuda):
    """``modify_allocation`` rejects inputs that are not Buffer objects."""
    mr = _vmm_resource(_vmm_device_or_skip())
    with pytest.raises(TypeError, match=r"Argument 'buf' has incorrect type .*Buffer, got object"):
        mr.modify_allocation(object(), 4096)


@pytest.mark.agent_authored(model="gpt-6-sol")
def test_vmm_modify_allocation_rejects_non_vmm_buffer(init_cuda):
    """modify_allocation rejects non-VMM Buffers without closing them."""
    mr = _vmm_resource(_vmm_device_or_skip())
    stream = init_cuda.create_stream()
    try:
        buf = init_cuda.memory_resource.allocate(64, stream=stream)
        try:
            assert isinstance(buf, Buffer)
            assert not isinstance(buf, VirtualMemoryBuffer)
            with pytest.raises(TypeError, match="buf must be a VirtualMemoryBuffer"):
                mr.modify_allocation(buf, 4096)
            assert not buf.is_closed
        finally:
            buf.close(stream=stream)
            stream.sync()
    finally:
        stream.close()


@pytest.mark.agent_authored(model="claude-fable-5-1")
def test_vmm_buffers_alive_at_shutdown_are_freed_quietly(init_cuda):
    """Buffers still referenced at interpreter exit are freed without a warning or an error."""
    device = _vmm_device_or_skip()
    code = textwrap.dedent(
        f"""
        import warnings

        from cuda.core import CUDAWarning, Device, VirtualMemoryResource, VirtualMemoryResourceOptions

        warnings.simplefilter("error", CUDAWarning)
        dev = Device({device.device_id})
        dev.set_current()
        mr = VirtualMemoryResource(dev, config=VirtualMemoryResourceOptions(handle_type={VMM_HANDLE_TYPE!r}))
        keep = [mr.allocate(4096)]
        keep.append(mr.modify_allocation(keep[0], 2 * keep[0].size))
        """
    )
    # The runner starts the child in an empty directory, so the installed
    # package is imported rather than the source tree the test runs from.
    result = run_python_snippet(code, timeout=CHILD_TIMEOUT_SEC)
    assert result.stderr == ""


@pytest.mark.parametrize(
    "options",
    [
        pytest.param({"addr_align": 3}, id="invalid-alignment"),
        pytest.param({"peers": (-1,)}, id="invalid-peer"),
    ],
)
@pytest.mark.thread_unsafe(reason="records process-global warnings")
@pytest.mark.agent_authored(model="gpt-6-sol")
def test_failed_vmm_allocation_does_not_block_another_resource(init_cuda, options):
    """A failed allocation leaves its config unchanged and another VMM resource usable."""
    device = _vmm_device_or_skip()
    invalid_resource = _vmm_resource(device, **options)
    error_match = "CUDA_ERROR_INVALID_VALUE" if "addr_align" in options else r"CUDA_ERROR_INVALID_(VALUE|DEVICE)"
    with assert_no_cuda_warning(), pytest.raises(CUDAError, match=error_match):
        invalid_resource.allocate(4096)
    for name, value in options.items():
        assert getattr(invalid_resource.config, name) == value

    # Use a separate resource with valid options; the invalid configuration is unchanged.
    resource = _vmm_resource(device)
    buffer = resource.allocate(4096)
    ptr, size = int(buffer.handle), buffer.size
    try:
        _vmm_fill(buffer, 0x5C)
        assert _vmm_read(buffer, 0, 16) == bytes([0x5C]) * 16
    finally:
        with assert_no_cuda_warning():
            buffer.close()
    _vmm_assert_released([(ptr, size)])


@pytest.mark.agent_authored(model="claude-fable-5-1")
@pytest.mark.thread_unsafe(reason="records process-global warnings")
def test_vmm_failed_grow_leaves_input_intact(init_cuda):
    """A grow the driver rejects raises, unwinds what it created, and leaves the input untouched (#2345)."""
    device = _vmm_device_or_skip()
    mr = _vmm_resource(device)
    buf = mr.allocate(4096)
    _vmm_fill(buf, 0x5C)
    ptr, size = int(buf.handle), buf.size
    # No address space of this size exists: the adjacent probe fails, and so
    # does the reservation for the move. Nothing may be freed by hand.
    with assert_no_cuda_warning(), pytest.raises(CUDAError):
        mr.modify_allocation(buf, 1 << 62)
    assert int(buf.handle) == ptr
    assert buf.size == size
    assert _vmm_read(buf, 0, 16) == bytes([0x5C]) * 16
    with assert_no_cuda_warning():
        buf.close()
    _vmm_assert_released([(ptr, size)])


@pytest.mark.agent_authored(model="claude-fable-5-1")
def test_vmm_size_rounding_overflow_raises(init_cuda):
    """A size whose rounding to the granularity does not fit in size_t raises instead of wrapping.

    Without the check the rounded size wraps to zero: ``allocate`` would ask the
    driver for zero bytes, and ``modify_allocation`` would treat the request as
    already covered and return the input buffer.
    """
    device = _vmm_device_or_skip()
    mr = _vmm_resource(device)
    too_large = 2**64 - 1
    with pytest.raises(OverflowError):
        mr.allocate(too_large)
    buf = mr.allocate(4096)
    try:
        with pytest.raises(OverflowError):
            mr.modify_allocation(buf, too_large)
        assert buf.size >= 4096
    finally:
        buf.close()


@pytest.mark.agent_authored(model="claude-fable-5-1")
@pytest.mark.thread_unsafe(reason="records process-global warnings")
@pytest.mark.parametrize("mode", ["global", "thread_local"])
def test_vmm_close_during_unrelated_capture_keeps_capture_valid(init_cuda, mode):
    """Closing a buffer synchronizes its stream without invalidating a capture on another stream.

    The driver treats cuStreamSynchronize as unsafe while a global-mode capture
    is active anywhere, or a thread-local capture is active on this thread; the
    range deleter runs the sync in relaxed mode so the unrelated capture
    survives and end_building succeeds.
    """
    device = _vmm_device_or_skip()
    mr = _vmm_resource(device)
    s = device.create_stream()
    buf = mr.allocate(4096, stream=s)
    gb = device.create_graph_builder().begin_building(mode=mode)
    try:
        with assert_no_cuda_warning():
            buf.close()
    finally:
        # Raises if the close invalidated the capture.
        gb.end_building()
        gb.close()
        s.close()


@pytest.mark.agent_authored(model="claude-fable-5-1")
@pytest.mark.thread_unsafe(reason="records process-global warnings")
def test_vmm_release_from_gc_on_capturing_stream_reports_and_unmaps(init_cuda):
    """A release from garbage collection while the recorded stream is capturing warns once and unmaps.

    Only an explicit close can raise; garbage collection skips the sync,
    reports it, unmaps the range, and leaves the capture intact.
    """
    device = _vmm_device_or_skip()
    mr = _vmm_resource(device)
    buf = mr.allocate(4096)
    ptr, size = int(buf.handle), buf.size
    gb = device.create_graph_builder().begin_building()
    try:
        buf.set_deallocation_stream(gb)
        with warnings.catch_warnings(record=True) as records:
            warnings.simplefilter("always", CUDAWarning)
            del buf
            gc.collect()
        reports = [str(r.message) for r in records if issubclass(r.category, CUDAWarning)]
        assert len(reports) == 1 and "capturing" in reports[0], reports
    finally:
        gb.end_building()
        gb.close()
    _vmm_assert_released([(ptr, size)])


@pytest.mark.agent_authored(model="claude-fable-5-1")
@pytest.mark.thread_unsafe(reason="records process-global warnings")
def test_vmm_release_on_default_stream_during_blocking_capture_reports_and_unmaps(init_cuda):
    """A release ordered on the default stream while a blocking stream in its context is capturing warns once and unmaps.

    ``allocate()`` without a stream records the legacy default stream. While a
    blocking stream in the same context is capturing, the driver reports any
    legacy-stream operation as an implicit capture dependency, and a
    synchronization would invalidate the capture. The release skips the
    synchronization, reports it, unmaps the range, and leaves the capture
    intact. Streams from ``Device.create_stream()`` are non-blocking by default
    and never interact with the legacy stream, so this needs a blocking one.
    """
    device = _vmm_device_or_skip()
    mr = _vmm_resource(device)
    buf = mr.allocate(4096)
    ptr, size = int(buf.handle), buf.size
    blocking = device.create_stream(options=StreamOptions(nonblocking=False))
    gb = blocking.create_graph_builder().begin_building()
    try:
        with warnings.catch_warnings(record=True) as records:
            warnings.simplefilter("always", CUDAWarning)
            buf.close()
        reports = [str(r.message) for r in records if issubclass(r.category, CUDAWarning)]
        assert len(reports) == 1 and "legacy stream" in reports[0], reports
    finally:
        # Raises if the release invalidated the capture.
        gb.end_building()
        gb.close()
        blocking.close()
    _vmm_assert_released([(ptr, size)])


@pytest.mark.agent_authored(model="claude-fable-5-1")
def test_vmm_close_without_stream_argument_on_capturing_stream_raises(init_cuda):
    """close() with no argument checks the recorded deallocation stream and refuses to close during its capture."""
    device = _vmm_device_or_skip()
    mr = _vmm_resource(device)
    buf = mr.allocate(4096)
    gb = device.create_graph_builder().begin_building()
    try:
        buf.set_deallocation_stream(gb)
        with pytest.raises(RuntimeError, match="capturing stream"):
            buf.close()
    finally:
        gb.end_building()
        gb.close()
    assert int(buf.handle) != 0
    buf.close()


@pytest.mark.thread_unsafe(reason="records process-global warnings")
@pytest.mark.agent_authored(model="claude-fable-5-1")
@pytest.mark.parametrize("mode", ["allocate", "grow", "grow_moved"])
def test_vmm_allocate_close_does_not_leak(init_cuda, mode):
    """Closing every buffer of a range unmaps it and frees each reservation, with no teardown failure.

    Each mode closes the buffers and then asks the driver about the exact
    addresses the range used, instead of comparing a device-wide free-memory
    count that other processes can move.
    """
    device = Device()
    if not device.properties.virtual_memory_management_supported:
        pytest.skip("Virtual memory management is not supported on this device")

    mr = _vmm_resource(device)
    if mode == "grow_moved":
        # Place a 2 MiB buffer at the start of a free hole and take the range
        # right after it, so the grow has to move. The driver honors this
        # placement for 2 MiB where it declined 8 MiB on Linux.
        mr, buf = _vmm_allocate_in_free_hole(device, 2 * MIB, 4 * MIB)
        decoy = _vmm_reserve_at(int(buf.handle) + buf.size, buf.size)
        if decoy is None:
            buf.close()
            pytest.skip("the driver did not grant a reservation right after the buffer")
        buffers = [buf]
        try:
            buffers.append(mr.modify_allocation(buf, 2 * buf.size))
        finally:
            handle_return(driver.cuMemAddressFree(decoy, buf.size))
        assert int(buffers[1].handle) != int(buf.handle)
    else:
        buf = mr.allocate(8 * MIB)
        buffers = [buf]
        if mode == "grow":
            buffers.append(mr.modify_allocation(buf, 2 * buf.size))

    # The reservations the range holds: one per allocate, and one more per grow.
    base, size = int(buf.handle), buf.size
    reservations = [(base, size)]
    if len(buffers) > 1:
        grown = buffers[1]
        if int(grown.handle) == base:
            # Grew in place: a second reservation right after the first.
            reservations.append((base + size, grown.size - size))
        else:
            reservations.append((int(grown.handle), grown.size))

    # Deleters report a failed unmap, address free or release as a CUDAWarning.
    with assert_no_cuda_warning():
        for b in buffers:
            b.close()
    _vmm_assert_released(reservations)


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
    when gpu_direct_rdma=True is requested but the device doesn't support virtual memory management.
    """
    device = Device()
    device.set_current()

    # Skip if virtual memory management is not supported (we need it for VMM)
    if not device.properties.virtual_memory_management_supported:
        pytest.skip("Virtual memory management is not supported on this device")

    # Skip if GPU Direct RDMA is supported (we want to test the unsupported case)
    if device.properties.gpu_direct_rdma_supported:
        pytest.skip("This test requires a device that doesn't support GPU Direct RDMA")

    # Test that requesting RDMA on an unsupported device throws an exception
    options = VirtualMemoryResourceOptions(gpu_direct_rdma=True)
    with pytest.raises(RuntimeError, match="GPU Direct RDMA is not supported on this device"):
        VirtualMemoryResource(device, config=options)


@pytest.mark.agent_authored(model="claude-fable-5-1")
@pytest.mark.thread_unsafe(reason="records process-global warnings and mutates the context stack")
def test_vmm_host_location_allocate_without_current_context(init_cuda):
    """A host-located resource allocates and closes with no CUDA context current."""
    device = Device()
    if not device.properties.host_virtual_memory_management_supported:
        pytest.skip("Host virtual memory management is not supported on this device")
    mr = VirtualMemoryResource(device, config=VirtualMemoryResourceOptions(location_type="host", handle_type=None))
    with no_current_context(), assert_no_cuda_warning():
        buf = mr.allocate(4096)
        assert buf.size >= 4096
        assert mr.is_host_accessible
        buf.close()


@pytest.mark.agent_authored(model="claude-fable-5-1")
@pytest.mark.thread_unsafe(reason="records process-global warnings")
def test_vmm_host_location_grow_and_close(init_cuda):
    """A host-located resource grows a buffer twice and every alias closes without a warning.

    The first grow extends the range in place or moves it. The second grow is
    forced to move when the driver grants a decoy reservation right after the
    buffer, so the existing host allocations are remapped at a new address.
    Host-located buffers record no default stream, so the closes unmap without
    a synchronization.
    """
    device = Device()
    if not device.properties.host_virtual_memory_management_supported:
        pytest.skip("Host virtual memory management is not supported on this device")
    mr = VirtualMemoryResource(device, config=VirtualMemoryResourceOptions(location_type="host", handle_type=None))
    with assert_no_cuda_warning():
        buf = mr.allocate(4096)
        size = buf.size
        grown = mr.modify_allocation(buf, 2 * size)
        assert isinstance(grown, VirtualMemoryBuffer)
        assert grown.size == 2 * size
        assert grown.is_host_accessible and not grown.is_device_accessible
        assert grown.device_id == -1
        # The input stays open at its old size.
        assert buf.size == size
        assert int(buf.handle) != 0

        decoy = _vmm_reserve_at(int(grown.handle) + grown.size, size)
        try:
            moved = mr.modify_allocation(grown, 3 * size)
        finally:
            if decoy is not None:
                handle_return(driver.cuMemAddressFree(decoy, size))
        assert moved.size == 3 * size
        if decoy is not None:
            # The range after the buffer was taken, so the grow had to move.
            assert int(moved.handle) != int(grown.handle)

        buf.close()
        grown.close()
        moved.close()
        assert moved.size == 0


@pytest.mark.agent_authored(model="claude-opus-5-5")
def test_device_memory_resource_out_of_memory_is_memory_error(init_cuda):
    device = Device()
    if not device.properties.memory_pools_supported:
        pytest.skip("Device does not support mempool operations")
    device.set_current()

    mr = DeviceMemoryResource(device, options=DeviceMemoryResourceOptions(max_size=POOL_SIZE))
    try:
        # Larger than any device, and than the pool's max_size, which the
        # driver rounds up to its allocation granularity.
        with pytest.raises(MemoryError) as e:
            mr.allocate(1 << 46, stream=device.default_stream)
        assert isinstance(e.value, CUDAOutOfMemoryError)
        assert isinstance(e.value, CUDAError)
        assert "CUDA_ERROR_OUT_OF_MEMORY" in str(e.value)

        # The failed request leaves the pool usable.
        buffer = mr.allocate(POOL_SIZE // 2, stream=device.default_stream)
        buffer.close()
    finally:
        mr.close()


def test_device_memory_resource_with_options(init_cuda):
    device = Device()
    if not device.properties.memory_pools_supported:
        pytest.skip("Device does not support mempool operations")

    device.set_current()

    # Test basic pool creation
    options = DeviceMemoryResourceOptions(max_size=POOL_SIZE)
    mr = DeviceMemoryResource(device, options=options)
    assert mr.device_id == device.device_id
    assert mr.is_device_accessible
    assert not mr.is_host_accessible
    assert not mr.is_ipc_enabled

    # Test allocation and deallocation
    buffer1 = mr.allocate(1024, stream=device.default_stream)
    assert buffer1.handle != 0
    assert buffer1.size == 1024
    assert buffer1.memory_resource == mr
    buffer1.close()

    # Test multiple allocations
    buffer1 = mr.allocate(1024, stream=device.default_stream)
    buffer2 = mr.allocate(2048, stream=device.default_stream)
    assert buffer1.handle != buffer2.handle
    assert buffer1.size == 1024
    assert buffer2.size == 2048
    buffer1.close()
    buffer2.close()

    # Test stream-based allocation
    stream = device.create_stream()
    buffer = mr.allocate(1024, stream=stream)
    assert buffer.handle != 0
    buffer.close(stream)

    # Test memory copying between buffers from same pool
    stream = device.create_stream()
    src_buffer = mr.allocate(64, stream=stream)
    dst_buffer = mr.allocate(64, stream=stream)
    src_buffer.copy_to(dst_buffer, stream=stream)
    device.sync()
    dst_buffer.close()
    src_buffer.close()
    # TODO(seberg): 2026-06: mr close may be unsafe with incomplete `buf.close()`
    device.sync()


def test_pinned_memory_resource_with_options(init_cuda):
    device = Device()
    skip_if_pinned_memory_unsupported(device)

    device.set_current()

    # Test basic pool creation
    options = PinnedMemoryResourceOptions(max_size=POOL_SIZE)
    mr = PinnedMemoryResource(options)
    assert mr.device_id == -1  # Not bound to any GPU
    assert mr.is_device_accessible
    assert mr.is_host_accessible
    assert not mr.is_ipc_enabled

    # Test allocation and deallocation
    buffer1 = mr.allocate(1024, stream=device.default_stream)
    assert buffer1.handle != 0
    assert buffer1.size == 1024
    assert buffer1.memory_resource == mr
    buffer1.close()

    # Test multiple allocations
    buffer1 = mr.allocate(1024, stream=device.default_stream)
    buffer2 = mr.allocate(2048, stream=device.default_stream)
    assert buffer1.handle != buffer2.handle
    assert buffer1.size == 1024
    assert buffer2.size == 2048
    buffer1.close()
    buffer2.close()

    # Test stream-based allocation
    stream = device.create_stream()
    buffer = mr.allocate(1024, stream=stream)
    assert buffer.handle != 0
    buffer.close(stream)

    # Test memory copying between buffers from same pool
    stream = device.create_stream()
    src_buffer = mr.allocate(64, stream=stream)
    dst_buffer = mr.allocate(64, stream=stream)
    src_buffer.copy_to(dst_buffer, stream=stream)
    device.sync()
    dst_buffer.close()
    src_buffer.close()
    # TODO(seberg): 2026-06: mr close may be unsafe with incomplete `buf.close()`
    device.sync()


def test_managed_memory_resource_with_options(init_cuda):
    device = Device()
    skip_if_managed_memory_unsupported(device)

    device.set_current()

    # Test basic pool creation
    options = ManagedMemoryResourceOptions()
    mr = create_managed_memory_resource_or_skip(options)
    assert mr.is_device_accessible
    assert mr.is_host_accessible
    assert not mr.is_ipc_enabled

    # Test allocation and deallocation
    buffer1 = mr.allocate(1024, stream=device.default_stream)
    assert buffer1.handle != 0
    assert buffer1.size == 1024
    assert buffer1.memory_resource == mr
    buffer1.close()

    # Test multiple allocations
    buffer1 = mr.allocate(1024, stream=device.default_stream)
    buffer2 = mr.allocate(2048, stream=device.default_stream)
    assert buffer1.handle != buffer2.handle
    assert buffer1.size == 1024
    assert buffer2.size == 2048
    buffer1.close()
    buffer2.close()

    # Test stream-based allocation
    stream = device.create_stream()
    buffer = mr.allocate(1024, stream=stream)
    assert buffer.handle != 0
    buffer.close(stream)

    # Test memory copying between buffers from same pool
    stream = device.create_stream()
    src_buffer = mr.allocate(64, stream=stream)
    dst_buffer = mr.allocate(64, stream=stream)
    src_buffer.copy_to(dst_buffer, stream=stream)
    device.sync()
    dst_buffer.close()
    src_buffer.close()
    # TODO(seberg): 2026-06: mr close may be unsafe with incomplete `buf.close()`
    device.sync()


def test_managed_memory_resource_preferred_location_default(init_cuda):
    """preferred_location property returns None when no preference is set."""
    device = Device()
    skip_if_managed_memory_unsupported(device)
    device.set_current()

    mr = create_managed_memory_resource_or_skip()
    assert mr.preferred_location is None


def test_managed_memory_resource_preferred_location_device(init_cuda):
    """preferred_location returns ("device", ordinal) for device preference."""
    device = Device()
    skip_if_managed_memory_unsupported(device)
    device.set_current()

    # Legacy style
    opts = ManagedMemoryResourceOptions(preferred_location=device.device_id)
    mr = create_managed_memory_resource_or_skip(opts)
    assert mr.preferred_location == ("device", device.device_id)

    # Explicit style
    opts = ManagedMemoryResourceOptions(
        preferred_location=device.device_id,
        preferred_location_type=ManagedMemoryLocationType.DEVICE,
    )
    mr = create_managed_memory_resource_or_skip(opts)
    assert mr.preferred_location == ("device", device.device_id)


def test_managed_memory_resource_preferred_location_host(init_cuda):
    """preferred_location returns ("host", None) for host preference."""
    device = Device()
    skip_if_managed_memory_unsupported(device)
    device.set_current()

    # Legacy style
    opts = ManagedMemoryResourceOptions(preferred_location=-1)
    mr = create_managed_memory_resource_or_skip(opts)
    assert mr.preferred_location == ("host", None)

    # Explicit style
    opts = ManagedMemoryResourceOptions(preferred_location_type="host")
    mr = create_managed_memory_resource_or_skip(opts)
    assert mr.preferred_location == ("host", None)


def test_managed_memory_resource_preferred_location_host_numa(init_cuda):
    """preferred_location returns ("host_numa", id) for NUMA preference."""
    device = Device()
    skip_if_managed_memory_unsupported(device)
    device.set_current()

    numa_id = device.properties.host_numa_id
    if numa_id < 0:
        pytest.skip("System does not support NUMA")

    # Auto-resolved from current device
    opts = ManagedMemoryResourceOptions(preferred_location_type="host_numa")
    mr = create_managed_memory_resource_or_skip(opts)
    assert mr.preferred_location == ("host_numa", numa_id)

    # Explicit NUMA node ID
    opts = ManagedMemoryResourceOptions(
        preferred_location=numa_id,
        preferred_location_type="host_numa",
    )
    mr = create_managed_memory_resource_or_skip(opts)
    assert mr.preferred_location == ("host_numa", numa_id)


def test_managed_memory_resource_preferred_location_validation(init_cuda):
    """Invalid preferred_location combinations raise errors."""
    device = Device()
    skip_if_managed_memory_unsupported(device)
    device.set_current()

    # Invalid preferred_location_type
    with pytest.raises(ValueError, match="'invalid' is not a valid ManagedMemoryLocationType. Must be "):
        ManagedMemoryResource(
            ManagedMemoryResourceOptions(
                preferred_location_type="invalid",
            )
        )

    # "device" requires a non-negative int
    with pytest.raises(ValueError, match="must be a device ordinal"):
        ManagedMemoryResource(
            ManagedMemoryResourceOptions(
                preferred_location_type="device",
            )
        )
    with pytest.raises(ValueError, match="must be a device ordinal"):
        ManagedMemoryResource(
            ManagedMemoryResourceOptions(
                preferred_location=-1,
                preferred_location_type="device",
            )
        )

    # "host" requires preferred_location=None
    with pytest.raises(ValueError, match="must be None"):
        ManagedMemoryResource(
            ManagedMemoryResourceOptions(
                preferred_location=0,
                preferred_location_type="host",
            )
        )

    # "host_numa" rejects negative IDs
    with pytest.raises(ValueError, match="must be a NUMA node ID"):
        ManagedMemoryResource(
            ManagedMemoryResourceOptions(
                preferred_location=-1,
                preferred_location_type="host_numa",
            )
        )

    # Legacy mode rejects invalid negative values
    with pytest.raises(ValueError, match="preferred_location must be"):
        ManagedMemoryResource(
            ManagedMemoryResourceOptions(
                preferred_location=-2,
            )
        )


@pytest.mark.thread_unsafe(reason="Uses mock.")
def test_managed_memory_resource_host_numa_auto_resolve_failure(init_cuda):
    """host_numa with None raises RuntimeError when NUMA ID cannot be determined."""
    from unittest.mock import MagicMock, patch

    device = Device()
    skip_if_managed_memory_unsupported(device)
    device.set_current()

    mock_dev = MagicMock()
    mock_dev.properties.host_numa_id = -1

    with (
        patch("cuda.core._device.Device", return_value=mock_dev),
        pytest.raises(RuntimeError, match="Cannot determine host NUMA ID"),
    ):
        ManagedMemoryResource(
            ManagedMemoryResourceOptions(
                preferred_location_type="host_numa",
            )
        )


def test_mempool_ipc_errors(mempool_device):
    """Test error cases when IPC operations are disabled."""
    device = mempool_device
    options = DeviceMemoryResourceOptions(max_size=POOL_SIZE, ipc_enabled=False)
    mr = DeviceMemoryResource(device, options=options)
    buffer = mr.allocate(64, stream=device.default_stream)
    ipc_error_msg = "Memory resource is not IPC-enabled"

    with pytest.raises(RuntimeError, match=ipc_error_msg):
        _ = mr.allocation_handle

    with pytest.raises(RuntimeError, match=ipc_error_msg):
        _ = buffer.ipc_descriptor

    with pytest.raises(RuntimeError, match=ipc_error_msg):
        handle = IPCBufferDescriptor._init(b"", 0)
        Buffer.from_ipc_descriptor(mr, handle, stream=device.default_stream)

    buffer.close()
    # TODO(seberg): 2026-06: mr close may be unsafe with incomplete `buf.close()`
    device.sync()


def test_pinned_mempool_ipc_basic(init_cuda):
    """Test basic IPC functionality for PinnedMemoryResource."""
    device = init_cuda

    skip_if_pinned_memory_unsupported(device)

    if platform.system() == "Windows":
        pytest.skip("IPC not implemented for Windows")

    if not supports_ipc_mempool(device):
        pytest.skip("Driver rejects IPC-enabled mempool creation on this platform")

    # Test IPC-enabled PinnedMemoryResource creation
    options = PinnedMemoryResourceOptions(max_size=POOL_SIZE, ipc_enabled=True)
    mr = PinnedMemoryResource(options)
    assert mr.is_ipc_enabled
    assert mr.is_device_accessible
    assert mr.is_host_accessible
    assert mr.device_id == -1  # pinned memory is not device-specific
    assert mr.numa_id >= 0  # IPC requires a concrete NUMA node

    # Test allocation handle export
    alloc_handle = mr.allocation_handle
    assert alloc_handle is not None

    # Test buffer allocation
    buffer = mr.allocate(1024, stream=device.default_stream)
    assert buffer.size == 1024
    assert buffer.is_device_accessible
    assert buffer.is_host_accessible

    # Test IPC descriptor
    ipc_desc = buffer.ipc_descriptor
    assert ipc_desc is not None
    assert ipc_desc.size == 1024

    buffer.close()
    # TODO(seberg): 2026-06: mr close may be unsafe with incomplete `buf.close()`
    device.sync()
    mr.close()


def test_pinned_mempool_ipc_errors(init_cuda):
    """Test error cases when IPC operations are disabled for PinnedMemoryResource."""
    device = init_cuda

    skip_if_pinned_memory_unsupported(device)

    # Test with IPC disabled (default)
    options = PinnedMemoryResourceOptions(max_size=POOL_SIZE, ipc_enabled=False)
    mr = PinnedMemoryResource(options)
    assert not mr.is_ipc_enabled
    assert mr.device_id == -1
    assert mr.numa_id == -1  # Non-IPC uses OS-managed placement

    buffer = mr.allocate(64, stream=device.default_stream)
    ipc_error_msg = "Memory resource is not IPC-enabled"

    with pytest.raises(RuntimeError, match=ipc_error_msg):
        _ = mr.allocation_handle

    with pytest.raises(RuntimeError, match=ipc_error_msg):
        _ = buffer.ipc_descriptor

    with pytest.raises(RuntimeError, match=ipc_error_msg):
        handle = IPCBufferDescriptor._init(b"", 0)
        Buffer.from_ipc_descriptor(mr, handle, stream=device.default_stream)

    buffer.close()
    # TODO(seberg): 2026-06: mr close may be unsafe with incomplete `buf.close()`
    device.sync()
    mr.close()


def test_pinned_mr_numa_id_default_no_ipc(init_cuda):
    """numa_id defaults to -1 (OS-managed) when IPC is disabled."""
    device = Device()
    skip_if_pinned_memory_unsupported(device)

    mr = create_pinned_memory_resource_or_xfail(PinnedMemoryResourceOptions(max_size=POOL_SIZE), xfail_device=device)
    assert mr.numa_id == -1
    mr.close()

    mr = create_pinned_memory_resource_or_xfail(
        PinnedMemoryResourceOptions(ipc_enabled=False, max_size=POOL_SIZE), xfail_device=device
    )
    assert mr.numa_id == -1
    mr.close()


def test_pinned_mr_numa_id_default_with_ipc(init_cuda):
    """numa_id is derived from the current device when IPC is enabled."""
    device = Device()
    skip_if_pinned_memory_unsupported(device)

    if platform.system() == "Windows":
        pytest.skip("IPC not implemented for Windows")
    if not supports_ipc_mempool(device):
        pytest.skip("Driver rejects IPC-enabled mempool creation on this platform")

    expected_numa_id = device.properties.host_numa_id
    if expected_numa_id < 0:
        pytest.skip("System does not support NUMA")

    mr = create_pinned_memory_resource_or_xfail(
        PinnedMemoryResourceOptions(ipc_enabled=True, max_size=POOL_SIZE), xfail_device=device
    )
    assert mr.numa_id == expected_numa_id
    mr.close()


def test_pinned_mr_numa_id_explicit(init_cuda):
    """Explicit numa_id is used regardless of ipc_enabled."""
    device = Device()
    skip_if_pinned_memory_unsupported(device)

    host_numa_id = device.properties.host_numa_id
    if host_numa_id < 0:
        pytest.skip("System does not support NUMA")

    mr = create_pinned_memory_resource_or_xfail(
        PinnedMemoryResourceOptions(numa_id=host_numa_id, max_size=POOL_SIZE), xfail_device=device
    )
    assert mr.numa_id == host_numa_id
    mr.close()

    if platform.system() == "Windows":
        pytest.skip("IPC not implemented for Windows")
    if not supports_ipc_mempool(device):
        pytest.skip("Driver rejects IPC-enabled mempool creation on this platform")

    mr = create_pinned_memory_resource_or_xfail(
        PinnedMemoryResourceOptions(ipc_enabled=True, numa_id=host_numa_id, max_size=POOL_SIZE),
        xfail_device=device,
    )
    assert mr.numa_id == host_numa_id
    mr.close()


def test_pinned_mr_numa_id_negative_error(init_cuda):
    """Negative numa_id raises ValueError."""
    device = Device()
    skip_if_pinned_memory_unsupported(device)

    with pytest.raises(ValueError, match="numa_id must be >= 0"):
        # uncapped-pool-ok: numa_id is validated before the pool is created
        PinnedMemoryResource(PinnedMemoryResourceOptions(numa_id=-1))

    with pytest.raises(ValueError, match="numa_id must be >= 0"):
        # uncapped-pool-ok: numa_id is validated before the pool is created
        PinnedMemoryResource(PinnedMemoryResourceOptions(numa_id=-42))


@pytest.mark.parametrize("ipc_enabled", [True, False])
@pytest.mark.parametrize(
    "property_name,expected_type",
    [
        ("reuse_follow_event_dependencies", bool),
        ("reuse_allow_opportunistic", bool),
        ("reuse_allow_internal_dependencies", bool),
        ("release_threshold", int),
        ("reserved_mem_current", int),
        ("reserved_mem_high", int),
        ("used_mem_current", int),
        ("used_mem_high", int),
    ],
)
def test_mempool_attributes(ipc_enabled, memory_resource_factory, property_name, expected_type):
    """Test all properties of memory pool attributes for all memory resource types."""
    MR, MRops = memory_resource_factory
    device = Device()

    if MR is DeviceMemoryResource and not device.properties.memory_pools_supported:
        pytest.skip("Device does not support mempool operations")
    elif MR is PinnedMemoryResource:
        skip_if_pinned_memory_unsupported(device)
    elif MR is ManagedMemoryResource:
        skip_if_managed_memory_unsupported(device)

    # ManagedMemoryResource does not support IPC
    if MR is ManagedMemoryResource and ipc_enabled:
        pytest.skip(f"{MR.__name__} does not support IPC")

    device.set_current()

    if platform.system() == "Windows":
        return  # IPC not implemented for Windows

    if ipc_enabled and not supports_ipc_mempool(device):
        pytest.skip("Driver rejects IPC-enabled mempool creation on this platform")

    if MR is DeviceMemoryResource:
        options = MRops(max_size=POOL_SIZE, ipc_enabled=ipc_enabled)
        mr = MR(device, options=options)
        assert mr.is_ipc_enabled == ipc_enabled
    elif MR is PinnedMemoryResource:
        options = MRops(max_size=POOL_SIZE, ipc_enabled=ipc_enabled)
        mr = MR(options)
        assert mr.is_ipc_enabled == ipc_enabled
    elif MR is ManagedMemoryResource:
        options = MRops()
        mr = create_managed_memory_resource_or_skip(options)
        assert not mr.is_ipc_enabled

    # Get the property value
    value = getattr(mr.attributes, property_name)

    # Test type
    assert isinstance(value, expected_type), f"{property_name} should return {expected_type}, got {type(value)}"

    # Test value constraints
    if expected_type is int:
        assert value >= 0, f"{property_name} should be non-negative"

    # Test memory usage properties with actual allocations
    if property_name in ["reserved_mem_current", "used_mem_current"]:
        # Allocate some memory and check if values increase
        initial_value = value
        buffer = None
        try:
            buffer = mr.allocate(1024, stream=device.default_stream)
            new_value = getattr(mr.attributes, property_name)
            assert new_value >= initial_value, f"{property_name} should increase or stay same after allocation"
        finally:
            if buffer is not None:
                buffer.close()

    # Test high watermark properties
    if property_name in ["reserved_mem_high", "used_mem_high"]:
        # High watermark should never be less than current
        current_prop = property_name.replace("_high", "_current")
        current_value = getattr(mr.attributes, current_prop)
        assert value >= current_value, f"{property_name} should be >= {current_prop}"


def _create_owned_pool_or_skip(memory_resource_factory, device):
    MR, MRops = memory_resource_factory
    if MR is DeviceMemoryResource:
        if not device.properties.memory_pools_supported:
            pytest.skip("Device does not support mempool operations")
        return MR(device, options=MRops(max_size=POOL_SIZE))
    if MR is PinnedMemoryResource:
        skip_if_pinned_memory_unsupported(device)
        return MR(options=MRops(max_size=POOL_SIZE))
    skip_if_managed_memory_unsupported(device)
    return create_managed_memory_resource_or_skip(options=MRops())


@pytest.mark.agent_authored(model="claude-opus-5-5")
def test_mempool_attributes_set(memory_resource_factory):
    device = Device()
    device.set_current()
    mr = _create_owned_pool_or_skip(memory_resource_factory, device)
    attributes = mr.attributes
    try:
        for name in (
            "reuse_follow_event_dependencies",
            "reuse_allow_opportunistic",
            "reuse_allow_internal_dependencies",
        ):
            for flag in (False, True):
                setattr(attributes, name, flag)
                assert getattr(attributes, name) is flag

        attributes.release_threshold = 1 << 20
        assert attributes.release_threshold == 1 << 20
        attributes.release_threshold = 0
        assert attributes.release_threshold == 0
        for value in (-1, 2**64):
            with pytest.raises(OverflowError):
                attributes.release_threshold = value

        for name in ("reserved_mem_high", "used_mem_high"):
            with pytest.raises(AttributeError, match=f"'{name}' may only be set to zero"):
                setattr(attributes, name, 1)
        for name in ("reserved_mem_current", "used_mem_current"):
            with pytest.raises(AttributeError):
                setattr(attributes, name, 0)
    finally:
        mr.close()


class _IntSubclass(int):
    pass


@pytest.mark.agent_authored(model="claude-opus-5-5")
@pytest.mark.parametrize(
    "make_value",
    [
        pytest.param(lambda: True, id="bool"),
        pytest.param(lambda: _IntSubclass(4096), id="int-subclass"),
        pytest.param(lambda: np.uint64(4096), id="np-uint64"),
    ],
)
def test_mempool_release_threshold_accepts_index_types(memory_resource_factory, make_value):
    device = Device()
    device.set_current()
    mr = _create_owned_pool_or_skip(memory_resource_factory, device)
    try:
        value = make_value()
        mr.attributes.release_threshold = value
        assert mr.attributes.release_threshold == int(value)
    finally:
        mr.close()


@pytest.mark.agent_authored(model="claude-opus-5-5")
@pytest.mark.parametrize("kind", ["reserved", "used"])
def test_mempool_watermark_reset(memory_resource_factory, kind):
    device = Device()
    device.set_current()
    mr = _create_owned_pool_or_skip(memory_resource_factory, device)
    high, current = f"{kind}_mem_high", f"{kind}_mem_current"
    try:
        buffer = mr.allocate(POOL_SIZE // 2, stream=device.default_stream)
        buffer.close()
        device.sync()
        # Return the freed memory so that the reserved watermark is above the
        # current value too.
        mr.trim()
        assert getattr(mr.attributes, high) >= POOL_SIZE // 2
        assert getattr(mr.attributes, current) < POOL_SIZE // 2

        setattr(mr.attributes, high, 0)
        assert getattr(mr.attributes, high) == getattr(mr.attributes, current)
    finally:
        mr.close()


@pytest.mark.agent_authored(model="claude-opus-5-5")
def test_mempool_trim(memory_resource_factory):
    device = Device()
    device.set_current()
    mr = _create_owned_pool_or_skip(memory_resource_factory, device)
    try:
        # Keep freed memory reserved so that only trim() returns it.
        mr.attributes.release_threshold = 2**64 - 1
        buffer = mr.allocate(POOL_SIZE // 2, stream=device.default_stream)
        buffer.close()
        device.sync()
        reserved = mr.attributes.reserved_mem_current
        assert reserved > 0

        mr.trim(min_bytes_to_keep=reserved)
        assert mr.attributes.reserved_mem_current == reserved

        mr.trim()
        assert mr.attributes.reserved_mem_current == 0
    finally:
        mr.close()

    with pytest.raises(RuntimeError, match="has been closed"):
        mr.trim()


@pytest.mark.agent_authored(model="claude-opus-5-5")
@pytest.mark.thread_unsafe(reason="a global-mode capture affects every thread")
@pytest.mark.parametrize(
    ("mode", "caller"),
    # A thread-local capture does not restrict other threads, so pairing it
    # with other_thread would pass without relaxed mode.
    [("global", "same_thread"), ("global", "other_thread"), ("thread_local", "same_thread")],
)
def test_mempool_calls_during_capture_keep_capture_valid(memory_resource_factory, mode, caller):
    """Pool attribute reads, writes and trim() run in relaxed capture mode.

    The driver refuses these calls, and invalidates the capture, while a
    global-mode capture is active on any thread or a thread-local one on the
    calling thread.
    """
    device = Device()
    device.set_current()
    mr = _create_owned_pool_or_skip(memory_resource_factory, device)

    def use_pool():
        device.set_current()
        attributes = mr.attributes
        attributes.release_threshold = attributes.release_threshold
        attributes.reuse_allow_opportunistic = attributes.reuse_allow_opportunistic
        attributes.used_mem_high = 0
        mr.trim()

    gb = device.create_graph_builder().begin_building(mode=mode)
    try:
        if caller == "same_thread":
            use_pool()
        else:
            with concurrent.futures.ThreadPoolExecutor(max_workers=1) as pool:
                pool.submit(use_pool).result()
    finally:
        # Raises if a pool call invalidated the capture.
        gb.end_building()
        gb.close()
        mr.close()


def test_mempool_attributes_repr(memory_resource_factory):
    """Test the repr of memory pool attributes for all memory resource types."""
    MR, MRops = memory_resource_factory
    device = Device()

    if MR is DeviceMemoryResource and not device.properties.memory_pools_supported:
        pytest.skip("Device does not support mempool operations")
    elif MR is PinnedMemoryResource:
        skip_if_pinned_memory_unsupported(device)
    elif MR is ManagedMemoryResource:
        skip_if_managed_memory_unsupported(device)

    device.set_current()

    if MR is DeviceMemoryResource:
        mr = MR(device, options=DeviceMemoryResourceOptions(max_size=2048))
    elif MR is PinnedMemoryResource:
        mr = MR(options=PinnedMemoryResourceOptions(max_size=2048))
    elif MR is ManagedMemoryResource:
        mr = create_managed_memory_resource_or_skip(options=ManagedMemoryResourceOptions())

    buffer1 = mr.allocate(64, stream=device.default_stream)
    buffer2 = mr.allocate(64, stream=device.default_stream)
    buffer1.close()
    assert re.match(
        r".*Attributes\(release_threshold=\d+, reserved_mem_current=\d+, reserved_mem_high=\d+, "
        r"reuse_allow_internal_dependencies=(True|False), reuse_allow_opportunistic=(True|False), "
        r"reuse_follow_event_dependencies=(True|False), used_mem_current=\d+, used_mem_high=\d+\)",
        str(mr.attributes),
    )
    buffer2.close()


def test_mempool_attributes_ownership(memory_resource_factory):
    """Ensure the attributes bundle keeps the pool alive via the handle."""
    MR, MRops = memory_resource_factory
    device = Device()

    if MR is DeviceMemoryResource and not device.properties.memory_pools_supported:
        pytest.skip("Device does not support mempool operations")
    elif MR is PinnedMemoryResource:
        skip_if_pinned_memory_unsupported(device)
    elif MR is ManagedMemoryResource:
        skip_if_managed_memory_unsupported(device)

    # Skip if IPC mempool is not supported on this platform/device (only relevant for DeviceMemoryResource)
    if MR is DeviceMemoryResource and not supports_ipc_mempool(device):
        pytest.skip("Driver rejects IPC-enabled mempool creation on this platform")

    device.set_current()

    if MR is DeviceMemoryResource:
        mr = MR(device, DeviceMemoryResourceOptions(max_size=POOL_SIZE))
    elif MR is PinnedMemoryResource:
        mr = MR(PinnedMemoryResourceOptions(max_size=POOL_SIZE))
    elif MR is ManagedMemoryResource:
        mr = create_managed_memory_resource_or_skip(ManagedMemoryResourceOptions())

    attributes = mr.attributes
    mr.close()
    del mr

    # The attributes bundle keeps the pool alive via MemoryPoolHandle,
    # so accessing attributes still works even after the MR is deleted.
    _ = attributes.used_mem_high  # Should not raise


# Ensure that memory views dellocate their reference to dlpack tensors
@pytest.mark.skipif(np is None, reason="numpy is not installed")
def test_strided_memory_view_leak():
    arr = np.zeros(1048576, dtype=np.uint8)
    before = sys.getrefcount(arr)
    for idx in range(10):
        StridedMemoryView.from_any_interface(arr, stream_ptr=-1)
    after = sys.getrefcount(arr)
    assert before == after


def test_strided_memory_view_refcnt():
    # Use Fortran ordering so strides is used
    a = np.zeros((64, 4), dtype=np.uint8, order="F")
    av = StridedMemoryView.from_any_interface(a, stream_ptr=-1)
    # segfaults if refcnt is wrong
    assert av.shape[0] == 64
    assert sys.getrefcount(av.shape) >= 2

    assert av.strides[0] == 1
    assert av.strides[1] == 64
    assert sys.getrefcount(av.strides) >= 2


def test_graph_memory_resource_object(init_cuda):
    device = Device()
    gmr1 = GraphMemoryResource(device)
    gmr2 = GraphMemoryResource(device)
    gmr3 = GraphMemoryResource(device.device_id)

    # These objects are interned.
    assert gmr1 is gmr2 is gmr3
    assert gmr1 == gmr2 == gmr3


@pytest.mark.skipif(np is None, reason="numpy is not installed")
def test_strided_memory_view_dlpack_errors():
    arr = np.zeros(64, dtype=np.uint8)
    smv = StridedMemoryView.from_any_interface(arr, stream_ptr=-1)
    with pytest.raises(BufferError, match="dl_device other than None"):
        smv.__dlpack__(dl_device=())
    with pytest.raises(BufferError, match="copy=True"):
        smv.__dlpack__(copy=True)
    with pytest.raises(BufferError, match="Expected max_version"):
        smv.__dlpack__(max_version=(9, 8, 7))


def test_memory_resource_alloc_zero_bytes(init_cuda, memory_resource_factory):
    MR, MROps = memory_resource_factory

    device = Device()
    device.set_current()

    if MR is DeviceMemoryResource and not device.properties.memory_pools_supported:
        pytest.skip("Device does not support mempool operations")
    elif MR is PinnedMemoryResource:
        skip_if_pinned_memory_unsupported(device)
        mr = create_pinned_memory_resource_or_xfail(xfail_device=device)
    elif MR is ManagedMemoryResource:
        skip_if_managed_memory_unsupported(device)
        mr = create_managed_memory_resource_or_skip(MROps(preferred_location=device.device_id))
    else:
        assert MR is DeviceMemoryResource
        mr = MR(device)

    buffer = mr.allocate(0, stream=device.default_stream)
    device.sync()
    assert buffer.handle >= 0
    assert buffer.size == 0
    assert buffer.device_id == mr.device_id


def test_strided_memory_view_not_in_top_level_namespace():
    # Regression test for issue #2027: StridedMemoryView and
    # args_viewable_as_strided_memory must only be exposed via
    # cuda.core.utils, not the top-level cuda.core namespace.
    import cuda.core
    import cuda.core.utils

    assert not hasattr(cuda.core, "StridedMemoryView")
    assert not hasattr(cuda.core, "args_viewable_as_strided_memory")

    assert hasattr(cuda.core.utils, "StridedMemoryView")
    assert hasattr(cuda.core.utils, "args_viewable_as_strided_memory")


def test_top_level_namespace_excludes_known_leaks():
    # Hardening test: lock the public top-level namespace against
    # accidental re-introduction of known-leaked symbols.
    import cuda.core

    public = {n for n in dir(cuda.core) if not n.startswith("_")}
    leaked = {"StridedMemoryView", "args_viewable_as_strided_memory"}
    assert not (public & leaked)


def test_legacy_pinned_allocate_zero_size(init_cuda):
    """LegacyPinnedMemoryResource.allocate(0) skips the driver call and uses ptr=0."""
    mr = LegacyPinnedMemoryResource()
    buf = mr.allocate(0)
    assert buf.size == 0
    # No driver call was made; handle is the sentinel 0.
    assert int(buf.handle) == 0


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


def test_synchronous_memory_resource_basic(init_cuda):
    """_SynchronousMemoryResource exercises properties and allocate paths (zero, non-zero, with-stream)."""
    from cuda.core._memory._synchronous_memory_resource import _SynchronousMemoryResource

    dev = Device()
    mr = _SynchronousMemoryResource(dev.device_id)
    assert mr.is_device_accessible is True
    assert mr.is_host_accessible is False
    assert mr.device_id == dev.device_id

    # Zero-size allocation takes the ptr=0 fast path.
    zero_buf = mr.allocate(0)
    assert zero_buf.size == 0
    assert int(zero_buf.handle) == 0
    zero_buf.close(stream=None)

    # Non-zero allocation goes through cuMemAlloc; close with stream=None for
    # the simple path. The explicit-stream close path is covered separately.
    buf = mr.allocate(64)
    assert buf.size == 64
    assert int(buf.handle) != 0
    buf.close(stream=None)

    # allocate(size, stream=stream) exercises Stream_accept validation on the
    # allocate side (cuMemAlloc is synchronous so the stream is accepted but unused).
    stream = dev.create_stream()
    buf2 = mr.allocate(32, stream=stream)
    assert buf2.size == 32
    assert int(buf2.handle) != 0
    buf2.close(stream=None)
    stream.close()


def test_synchronous_memory_resource_deallocate_accepts_stream(init_cuda):
    """_SynchronousMemoryResource.deallocate accepts an explicit stream."""
    from cuda.core._memory._synchronous_memory_resource import _SynchronousMemoryResource

    dev = Device()
    mr = _SynchronousMemoryResource(dev.device_id)
    buf = mr.allocate(64)
    stream = dev.create_stream()
    buf.close(stream=stream)
    stream.close()


@pytest.mark.agent_authored(model="gpt-5.6")
def test_synchronous_memory_resource_uses_its_context(device_x2):
    """Synchronous allocation targets its stored context and restores the current one."""
    from cuda.core._memory._synchronous_memory_resource import _SynchronousMemoryResource

    alloc_dev, current_dev = device_x2
    alloc_dev.set_current()
    stream = alloc_dev.create_stream()
    mr = _SynchronousMemoryResource(alloc_dev.device_id, alloc_dev.context)

    current_dev.set_current()
    current_context = current_context_handle()

    buf = mr.allocate(64, stream=stream)
    try:
        pointer_context = handle_return(
            driver.cuPointerGetAttribute(
                driver.CUpointer_attribute.CU_POINTER_ATTRIBUTE_CONTEXT,
                int(buf.handle),
            )
        )
        pointer_device = handle_return(
            driver.cuPointerGetAttribute(
                driver.CUpointer_attribute.CU_POINTER_ATTRIBUTE_DEVICE_ORDINAL,
                int(buf.handle),
            )
        )
        assert int(pointer_context) == int(alloc_dev.context.handle)
        assert pointer_device == alloc_dev.device_id
        assert current_context_handle() == current_context
    finally:
        buf.close(stream=stream)
        stream.close()

    assert current_context_handle() == current_context


@pytest.mark.agent_authored(model="gpt-5.6")
def test_synchronous_memory_resource_restores_context_after_failure(device_x2):
    """A failed synchronous allocation restores the context that was current."""
    from cuda.core._memory._synchronous_memory_resource import _SynchronousMemoryResource

    alloc_dev, current_dev = device_x2
    alloc_dev.set_current()
    mr = _SynchronousMemoryResource(alloc_dev.device_id, alloc_dev.context)
    current_dev.set_current()
    current_context = current_context_handle()

    with pytest.raises(CUDAError):
        mr.allocate(sys.maxsize)

    assert current_context_handle() == current_context


@pytest.mark.thread_unsafe(reason="records process-global warnings and mutates the context stack")
@pytest.mark.agent_authored(model="claude-sonnet-5")
def test_synchronous_memory_resource_default_stream_deallocates_in_own_context(device_x2):
    """Buffer teardown with no explicit stream frees in the resource's own
    context, not whatever context happens to be current at close() time."""
    from cuda.core._memory._synchronous_memory_resource import _SynchronousMemoryResource

    alloc_dev, current_dev = device_x2
    alloc_dev.set_current()
    mr = _SynchronousMemoryResource(alloc_dev.device_id, alloc_dev.context)

    current_dev.set_current()
    current_context = current_context_handle()

    buf = mr.allocate(64)  # no explicit stream: records a context-bound default token
    assert current_context_handle() == current_context

    with assert_no_cuda_warning():
        buf.close()  # no explicit stream: reuses the recorded token
    assert current_context_handle() == current_context


@pytest.mark.thread_unsafe(reason="records process-global warnings and mutates the context stack")
@pytest.mark.agent_authored(model="claude-sonnet-5")
def test_synchronous_memory_resource_allocate_without_current_context(device_x2):
    """allocate()/close() with no explicit stream succeed with no context
    current, instead of raising or leaking the allocation (#2311)."""
    from cuda.core._memory._synchronous_memory_resource import _SynchronousMemoryResource

    alloc_dev, current_dev = device_x2
    alloc_dev.set_current()
    mr = _SynchronousMemoryResource(alloc_dev.device_id, alloc_dev.context)
    current_dev.set_current()

    with no_current_context(), assert_no_cuda_warning():
        buf = mr.allocate(64)
        assert current_context_handle() == 0
        buf.close()
        assert current_context_handle() == 0


@pytest.mark.parametrize(
    ("method", "spec", "match"),
    [
        ("_access_to_flags", "bogus", "Unknown access spec"),
        ("_allocation_type_to_driver", "bogus", "Unsupported allocation_type"),
        ("_location_type_to_driver", "bogus", "Unsupported location_type"),
        ("_handle_type_to_driver", "bogus", "Unsupported handle_type"),
        ("_granularity_to_driver", "bogus", "Unsupported granularity"),
    ],
)
def test_vmm_options_spec_validators_raise(method, spec, match):
    """Every VMM spec validator static method rejects unknown strings with ValueError."""
    fn = getattr(VirtualMemoryResourceOptions, method)
    with pytest.raises(ValueError, match=match):
        fn(spec)


@pytest.mark.parametrize("system", ["Linux", "Windows"])
@pytest.mark.thread_unsafe(reason="patches platform.system")
def test_vmm_options_default_handle_type(monkeypatch, system):
    monkeypatch.setattr(platform, "system", lambda: system)
    options = VirtualMemoryResourceOptions()
    expected = None if system == "Windows" else VirtualMemoryHandleType.POSIX_FD
    assert options.handle_type == expected
    assert VirtualMemoryResourceOptions(handle_type="win32_kmt").handle_type == "win32_kmt"
    assert VirtualMemoryResourceOptions(handle_type=None).handle_type is None
    with pytest.raises(ValueError):
        VirtualMemoryResourceOptions(allocation_type=VirtualMemoryAllocationType.MANAGED)


def test_vmm_options_handle_type_win32_raises():
    """_handle_type_to_driver raises NotImplementedError for 'win32'."""
    with pytest.raises(NotImplementedError, match="win32 is currently not supported"):
        VirtualMemoryResourceOptions._handle_type_to_driver("win32")


@pytest.mark.agent_authored(model="claude-opus-5-5")
def test_vmm_options_store_peers_as_tuple():
    """The options store a one-shot peers iterable as a tuple, so every reader sees all the peers."""
    options = VirtualMemoryResourceOptions(peers=(peer for peer in (1, 2)))
    assert options.peers == (1, 2)


@pytest.mark.agent_authored(model="claude-fable-5-1")
@pytest.mark.parametrize("location_type", ["host", "host_numa"])
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" and "host_numa" alike when
    deciding the resource is not bound to a device, so is_host_accessible must
    agree; otherwise a NUMA-located resource claims to be neither host- nor
    device-accessible. The driver rejects an exportable handle type for HOST
    memory only, so the default handle type is rejected at construction for
    "host" and accepted for "host_numa".
    """
    device = Device()
    device.set_current()
    numa = {"numa_id": 0} if location_type == "host_numa" else {}
    if location_type == "host":
        with pytest.raises(ValueError, match="handle_type=None"):
            VirtualMemoryResource(device, config=VirtualMemoryResourceOptions(location_type=location_type))
    else:
        VirtualMemoryResource(device, config=VirtualMemoryResourceOptions(location_type=location_type, **numa))
    mr = VirtualMemoryResource(
        device, config=VirtualMemoryResourceOptions(location_type=location_type, handle_type=None, **numa)
    )
    assert mr.device is None
    assert mr.is_host_accessible is True
    assert mr.device_id == -1


@pytest.mark.thread_unsafe(reason="records process-global warnings")
@pytest.mark.agent_authored(model="claude-opus-5-5")
def test_vmm_host_numa_allocates_and_copies(init_cuda):
    """A "host_numa" resource allocates on its NUMA node, and the host can use the memory.

    The driver rejects CU_MEM_LOCATION_TYPE_HOST_NUMA with location id -1, so
    the allocation works only if numa_id reaches the allocation properties
    (#2694). The resource grants access to its own location only, so the data
    goes through the host. A per-call config cannot move the buffer to another
    node.
    """
    device = Device()
    device.set_current()
    props = device.properties
    if not props.host_numa_virtual_memory_management_supported:
        pytest.skip("Host NUMA virtual memory management is not supported on this device")
    if props.host_numa_id < 0:
        pytest.skip("The system reports no host NUMA node")
    mr = VirtualMemoryResource(
        device,
        config=VirtualMemoryResourceOptions(location_type="host_numa", numa_id=props.host_numa_id, handle_type=None),
    )
    with assert_no_cuda_warning():
        buf = mr.allocate(4096)
        try:
            assert buf.is_host_accessible
            ctypes.memset(int(buf.handle), 0xAB, buf.size)
            assert ctypes.string_at(int(buf.handle), buf.size) == b"\xab" * buf.size
            other_node = VirtualMemoryResourceOptions(
                location_type="host_numa", numa_id=props.host_numa_id + 1, handle_type=None
            )
            with pytest.raises(ValueError, match="config.numa_id"):
                mr.modify_allocation(buf, 2 * buf.size, other_node)
        finally:
            buf.close()


@pytest.mark.agent_authored(model="claude-opus-5-5")
def test_vmm_host_numa_current_rejected(init_cuda):
    """The driver does not accept "host_numa_current" for physical allocations, so construction rejects it."""
    device = Device()
    device.set_current()
    with pytest.raises(ValueError, match="does not accept it for physical allocations"):
        VirtualMemoryResource(
            device, config=VirtualMemoryResourceOptions(location_type="host_numa_current", handle_type=None)
        )


@pytest.mark.agent_authored(model="claude-opus-5-5")
@pytest.mark.parametrize(
    ("options", "match"),
    [
        pytest.param({"location_type": "host_numa"}, "requires numa_id", id="host_numa-missing"),
        pytest.param({"location_type": "host_numa", "numa_id": -1}, "numa_id must be >= 0", id="host_numa-negative"),
        pytest.param({"location_type": "host", "numa_id": 0}, "only valid with", id="host-misplaced"),
        pytest.param({"location_type": "device", "numa_id": 0}, "only valid with", id="device-misplaced"),
    ],
)
def test_vmm_numa_id_validation(init_cuda, options, match):
    """numa_id is required for "host_numa", must be non-negative, and is rejected for other location types."""
    device = _vmm_device_or_skip()
    with pytest.raises(ValueError, match=match):
        VirtualMemoryResource(device, config=VirtualMemoryResourceOptions(handle_type=None, **options))


def test_device_memory_resource_peer_accessible_by_non_owned(mempool_device):
    """peer_accessible_by on a non-owned (default) DMR queries the driver live."""
    dev = mempool_device
    # The default DeviceMemoryResource(device) wraps the current device's
    # default pool, i.e. _mempool_owned is False, so accessing
    # peer_accessible_by exercises the live _DMR_query_peer_access path.
    mr = DeviceMemoryResource(dev)
    peers = mr.peer_accessible_by
    assert all(isinstance(p, Device) for p in peers)
    # __contains__ accepts int dev_ids; the owning device is never a peer.
    assert dev.device_id not in peers


def test_dmr_mempool_get_access_self(mempool_device):
    """DMR_mempool_get_access returns 'rw' when querying the owning device itself."""
    from cuda.core._memory._device_memory_resource import DMR_mempool_get_access

    mr = DeviceMemoryResource(mempool_device)
    # The owning device always has read-write access to its own pool.
    assert DMR_mempool_get_access(mr, mempool_device.device_id) == "rw"


def test_dmr_mempool_get_access_peer(mempool_device_x2):
    """DMR_mempool_get_access reflects peer access state for a different device."""
    from cuda.core._memory._device_memory_resource import DMR_mempool_get_access

    dev, peer = mempool_device_x2
    # Owned pool avoids peer-access contamination from other tests; max_size
    # caps VA to dodge Windows MCDM OOM
    mr = DeviceMemoryResource(dev, DeviceMemoryResourceOptions(max_size=POOL_SIZE))

    # Fresh owned pool: peer has no access.
    assert DMR_mempool_get_access(mr, peer.device_id) == ""
    # After granting access, peer has read-write.
    mr.peer_accessible_by = [peer]
    assert DMR_mempool_get_access(mr, peer.device_id) == "rw"
    # After revoking, peer is back to no access.
    mr.peer_accessible_by = []
    assert DMR_mempool_get_access(mr, peer.device_id) == ""


@pytest.mark.thread_unsafe(reason="depends on the driver handing back the pool handle that was just destroyed")
@pytest.mark.agent_authored(model="claude-fable-5-1")
def test_closed_pool_peer_access_not_inherited_by_recycled_handle(mempool_device_x2):
    """Peer access granted to a closed pool does not survive into a pool that reuses its handle."""
    from cuda.core._memory._device_memory_resource import DMR_mempool_get_access

    dev, peer = mempool_device_x2
    options = DeviceMemoryResourceOptions(max_size=POOL_SIZE)
    mr = DeviceMemoryResource(dev, options)
    mr.peer_accessible_by = [peer]
    assert DMR_mempool_get_access(mr, peer.device_id) == "rw"
    old_handle = int(mr.handle)
    mr.close()

    # The driver usually hands the freed handle straight back. Keep the misses
    # alive so that a retry cannot land on the same address twice.
    pools = []
    try:
        for _ in range(8):
            recycled = DeviceMemoryResource(dev, options)
            pools.append(recycled)
            if int(recycled.handle) == old_handle:
                break
        else:
            pytest.skip("the driver did not recycle the destroyed pool handle")
        # Unless the peer access is revoked before the pool is destroyed, the
        # recycled handle still reports it (nvbug 5698116).
        assert DMR_mempool_get_access(recycled, peer.device_id) == ""
    finally:
        for pool in pools:
            pool.close()


def test_dmr_peer_accessible_by_setter_empty(mempool_device):
    """Assigning an empty peer-access set to a fresh owned pool is a no-op."""
    # max_size caps VA to dodge Windows MCDM OOM
    mr = DeviceMemoryResource(mempool_device, options=DeviceMemoryResourceOptions(max_size=POOL_SIZE))
    assert set(mr.peer_accessible_by) == set()
    mr.peer_accessible_by = []
    assert set(mr.peer_accessible_by) == set()


@pytest.mark.agent_authored(model="claude-opus-4.8")
def test_mempool_attributes_cannot_instantiate_directly():
    """_MemPoolAttributes cannot be instantiated directly."""
    from cuda.core._memory._memory_pool import _MemPoolAttributes

    with pytest.raises(RuntimeError, match="cannot be instantiated directly"):
        _MemPoolAttributes()


@pytest.mark.agent_authored(model="claude-opus-4.8")
def test_dmr_handle_and_ownership(mempool_device):
    """An options-created pool is handle-owning with a live handle; wrapping the device's current pool is non-owning."""
    owned = DeviceMemoryResource(mempool_device, DeviceMemoryResourceOptions(max_size=POOL_SIZE))
    assert owned.is_handle_owned is True
    handle = owned.handle
    assert handle is not None
    assert int(handle) != 0

    non_owned = DeviceMemoryResource(mempool_device)
    assert non_owned.is_handle_owned is False


@pytest.mark.agent_authored(model="claude-opus-4.8")
def test_dmr_deallocate_frees_pool_pointer(mempool_device):
    """Closing a Buffer.from_handle(..., mr=mr) view frees the pointer via the Python
    _MemPool.deallocate path; the pool's in-use bytes drop back."""
    dev = mempool_device
    stream = dev.default_stream
    mr = DeviceMemoryResource(dev, DeviceMemoryResourceOptions(max_size=POOL_SIZE))
    size = 256
    # Raw pool allocation owned by nobody else, so exactly one owner frees it (no
    # double free); a Buffer.from_handle view then routes teardown through the
    # Python deallocate path that mr.allocate()'s C++-direct free would skip.
    ptr = handle_return(driver.cuMemAllocFromPoolAsync(size, mr.handle, stream.handle))
    stream.sync()
    used_after_alloc = mr.attributes.used_mem_current
    assert used_after_alloc >= size
    buf = Buffer.from_handle(int(ptr), size, mr=mr)
    buf.close(stream)
    stream.sync()
    assert int(buf.handle) == 0
    # In-use bytes fell back, so the pointer was actually returned (buf.handle == 0
    # alone wouldn't prove it: the deleter callback swallows a failed free).
    assert mr.attributes.used_mem_current < used_after_alloc


@pytest.mark.agent_authored(model="claude-opus-4.8")
def test_dmr_close_is_idempotent(mempool_device):
    """Closing an owned DeviceMemoryResource twice is safe (the second close is a no-op)."""
    mr = DeviceMemoryResource(mempool_device, DeviceMemoryResourceOptions(max_size=POOL_SIZE))
    assert mr.is_handle_owned is True
    assert int(mr.handle) != 0
    mr.close()
    # First close releases the pool handle itself, not just ownership.
    assert int(mr.handle) == 0
    assert mr.is_handle_owned is False
    mr.close()  # no-op on the now-null handle
    assert int(mr.handle) == 0
    assert mr.is_handle_owned is False


@pytest.mark.agent_authored(model="claude-opus-4.8")
def test_dmr_ipc_enabled_unsupported_raises(mempool_device):
    """Requesting an IPC-enabled pool where memory IPC is unsupported raises RuntimeError."""
    if not IS_WINDOWS:
        pytest.skip("memory IPC is supported on this platform; unsupported-raise path is Windows-only")
    with pytest.raises(RuntimeError, match="IPC is not available"):
        # uncapped-pool-ok: IPC support is checked before the pool is created
        DeviceMemoryResource(mempool_device, DeviceMemoryResourceOptions(ipc_enabled=True))