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

"""GraphNode subclasses — EmptyNode through SwitchNode."""

from __future__ import annotations

from libc.stddef cimport size_t
from libc.stdint cimport uintptr_t
from libc.string cimport memset as c_memset

from cuda.bindings cimport cydriver

from cuda.core._event cimport Event
from cuda.core._kernel_arg_handler cimport ParamHolder
from cuda.core._launch_config cimport LaunchConfig
from cuda.core._memory._buffer cimport Buffer
from cuda.core._module cimport Kernel
from cuda.core.graph._graph_definition cimport GraphCondition, GraphDefinition
from cuda.core.graph._graph_node cimport (
    GraphNode,
    _get_memcpy_memory_type,
    _init_memcpy_params,
    _resolve_memcpy_operand,
)
from cuda.core._resource_handles cimport (
    EventHandle,
    GraphExecHandle,
    GraphHandle,
    GraphNodeHandle,
    KernelHandle,
    OpaqueHandle,
    PreparedAttachment,
    PreparedChildGraphUpdate,
    PreparedExecAttachment,
    as_cu,
    as_intptr,
    create_child_graph_handle,
    create_event_handle_ref,
    create_kernel_handle_ref,
    graph_commit_attachment,
    graph_commit_child_graph_update,
    graph_commit_exec_attachment,
    graph_get_attachment,
    graph_node_get_graph,
    graph_prepare_attachment,
    graph_prepare_child_graph_update,
    graph_prepare_exec_attachment,
    make_opaque_py,
)
from cuda.core._utils.cuda_utils cimport HANDLE_RETURN, _parse_fill_value
from cuda.core._utils.version cimport cy_binding_version, cy_driver_version

from cuda.core.graph._host_callback cimport (
    _is_py_host_trampoline,
    _resolve_host_callback,
)

from cuda.core._utils.cuda_utils import driver, handle_return
from cuda.core.typing import GraphConditionalType

__all__ = [
    'AllocNode',
    'ChildGraphNode',
    'ConditionalNode',
    'EmptyNode',
    'EventRecordNode',
    'EventWaitNode',
    'ExecutableChildGraphNode',
    'ExecutableEventRecordNode',
    'ExecutableEventWaitNode',
    'ExecutableGraphNode',
    'ExecutableHostCallbackNode',
    'ExecutableKernelNode',
    'ExecutableMemcpyNode',
    'ExecutableMemsetNode',
    'FreeNode',
    'HostCallbackNode',
    'IfElseNode',
    'IfNode',
    'KernelNode',
    'MemcpyNode',
    'MemsetNode',
    'SwitchNode',
    'WhileNode',
]


cdef bint _has_cuGraphNodeGetParams = False
cdef bint _version_checked = False


cdef void _require_graph_node_update_support() except *:
    cdef tuple version = cy_driver_version()
    if version < (12, 2, 0):
        raise RuntimeError(
            "Graph node mutation requires CUDA driver 12.2 or newer; "
            f"using driver version {'.'.join(map(str, version))}"
        )
    version = cy_binding_version()
    if version < (12, 2, 0):
        raise RuntimeError(
            "Graph node mutation requires cuda.bindings 12.2 or newer; "
            f"using cuda.bindings version {'.'.join(map(str, version))}"
        )


cdef void _set_definition_node_params(
        const GraphNodeHandle& h_node,
        cydriver.CUgraphNodeParams* params,
        OpaqueHandle owner0,
        OpaqueHandle owner1=OpaqueHandle(),
        cydriver.CUcontext update_ctx=NULL) except *:
    _require_graph_node_update_support()

    cdef GraphHandle h_graph = graph_node_get_graph(h_node)
    cdef cydriver.CUgraphNode node = as_cu(h_node)
    cdef cydriver.CUcontext previous_ctx = NULL
    cdef bint restore_ctx = False
    cdef PreparedAttachment prepared

    HANDLE_RETURN(graph_prepare_attachment(
        h_graph, owner0, owner1, &prepared))
    if update_ctx != NULL:
        with nogil:
            HANDLE_RETURN(cydriver.cuCtxGetCurrent(&previous_ctx))
            if previous_ctx != update_ctx:
                HANDLE_RETURN(cydriver.cuCtxSetCurrent(update_ctx))
                restore_ctx = True
    try:
        with nogil:
            HANDLE_RETURN(cydriver.cuGraphNodeSetParams(node, params))
    finally:
        if restore_ctx:
            with nogil:
                HANDLE_RETURN(cydriver.cuCtxSetCurrent(previous_ctx))
    HANDLE_RETURN(graph_commit_attachment(prepared, node))


cdef void _set_executable_node_params(
        const GraphExecHandle& h_exec,
        const GraphNodeHandle& h_node,
        cydriver.CUgraphNodeParams* params,
        OpaqueHandle owner0=OpaqueHandle(),
        OpaqueHandle owner1=OpaqueHandle()) except *:
    _require_graph_node_update_support()

    cdef cydriver.CUgraphExec graph_exec = as_cu(h_exec)
    cdef cydriver.CUgraphNode node = as_cu(h_node)
    if graph_exec == NULL:
        raise ValueError("executable graph has been closed")
    if node == NULL:
        raise ValueError("source graph node is no longer valid")

    cdef PreparedExecAttachment prepared
    HANDLE_RETURN(graph_prepare_exec_attachment(
        h_exec, owner0, owner1, &prepared))

    cdef cydriver.CUresult status
    with nogil:
        status = cydriver.cuGraphExecNodeSetParams(
            graph_exec, node, params)
    if status == cydriver.CUDA_SUCCESS:
        graph_commit_exec_attachment(prepared)
    HANDLE_RETURN(status)


cdef bint _get_executable_node_enabled(
        const GraphExecHandle& h_exec,
        const GraphNodeHandle& h_node) except *:
    _require_graph_node_update_support()

    cdef cydriver.CUgraphExec graph_exec = as_cu(h_exec)
    cdef cydriver.CUgraphNode node = as_cu(h_node)
    cdef unsigned int enabled
    if graph_exec == NULL:
        raise ValueError("executable graph has been closed")
    if node == NULL:
        raise ValueError("source graph node is no longer valid")
    with nogil:
        HANDLE_RETURN(cydriver.cuGraphNodeGetEnabled(
            graph_exec, node, &enabled))
    return enabled != 0


cdef void _set_executable_node_enabled(
        const GraphExecHandle& h_exec,
        const GraphNodeHandle& h_node,
        bint enabled) except *:
    _require_graph_node_update_support()

    cdef cydriver.CUgraphExec graph_exec = as_cu(h_exec)
    cdef cydriver.CUgraphNode node = as_cu(h_node)
    if graph_exec == NULL:
        raise ValueError("executable graph has been closed")
    if node == NULL:
        raise ValueError("source graph node is no longer valid")
    with nogil:
        HANDLE_RETURN(cydriver.cuGraphNodeSetEnabled(
            graph_exec, node, <unsigned int>enabled))


cdef bint _check_node_get_params():
    global _has_cuGraphNodeGetParams, _version_checked
    if not _version_checked:
        from cuda.core._utils.version import binding_version, driver_version
        _has_cuGraphNodeGetParams = (
            driver_version() >= (13, 2, 0) and binding_version() >= (13, 2, 0)
        )
        _version_checked = True
    return _has_cuGraphNodeGetParams


cdef void _reject_unsupported_kernel_node(
        cydriver.CUgraphNode node) except *:
    cdef cydriver.CUkernelNodeAttrValue cluster
    cdef cydriver.CUkernelNodeAttrValue cooperative

    c_memset(&cluster, 0, sizeof(cluster))
    c_memset(&cooperative, 0, sizeof(cooperative))
    with nogil:
        HANDLE_RETURN(cydriver.cuGraphKernelNodeGetAttribute(
            node, <cydriver.CUkernelNodeAttrID>(
                cydriver.CU_KERNEL_NODE_ATTRIBUTE_CLUSTER_DIMENSION),
            &cluster))
        HANDLE_RETURN(cydriver.cuGraphKernelNodeGetAttribute(
            node, <cydriver.CUkernelNodeAttrID>(
                cydriver.CU_KERNEL_NODE_ATTRIBUTE_COOPERATIVE),
            &cooperative))
    if (cluster.clusterDim.x != 0 or cluster.clusterDim.y != 0 or
            cluster.clusterDim.z != 0 or cooperative.cooperative != 0):
        raise NotImplementedError(
            "updating clustered or cooperative kernel nodes is not supported")


cdef bint _is_supported_memcpy_descriptor(
        cydriver.CUDA_MEMCPY3D* params) noexcept nogil:
    return (
        (params.srcMemoryType == cydriver.CU_MEMORYTYPE_HOST or
         params.srcMemoryType == cydriver.CU_MEMORYTYPE_DEVICE)
        and (params.dstMemoryType == cydriver.CU_MEMORYTYPE_HOST or
             params.dstMemoryType == cydriver.CU_MEMORYTYPE_DEVICE)
        and params.srcXInBytes == 0
        and params.srcY == 0
        and params.srcZ == 0
        and params.srcLOD == 0
        and params.srcPitch == 0
        and params.srcHeight == 0
        and params.dstXInBytes == 0
        and params.dstY == 0
        and params.dstZ == 0
        and params.dstLOD == 0
        and params.dstPitch == 0
        and params.dstHeight == 0
        and params.Height == 1
        and params.Depth == 1
        and params.reserved0 == NULL
        and params.reserved1 == NULL
    )


cdef class EmptyNode(GraphNode):
    """An empty (synchronization) node."""

    @staticmethod
    cdef EmptyNode _create_impl(GraphNodeHandle h_node):
        cdef EmptyNode n = EmptyNode.__new__(EmptyNode)
        n._h_node = h_node
        return n

    def __repr__(self) -> str:
        return f"<EmptyNode handle=0x{as_intptr(self._h_node):x}>"


cdef class KernelNode(GraphNode):
    """A kernel launch node.

    Properties
    ----------
    grid : tuple of int
        Grid dimensions (gridDimX, gridDimY, gridDimZ).
    block : tuple of int
        Block dimensions (blockDimX, blockDimY, blockDimZ).
    shmem_size : int
        Dynamic shared memory size in bytes.
    kernel : Kernel
        The kernel object for this launch node.
    config : LaunchConfig
        A LaunchConfig reconstructed from this node's parameters.
    """

    @staticmethod
    cdef KernelNode _create_with_params(GraphNodeHandle h_node,
                                        tuple grid, tuple block, unsigned int shmem_size,
                                        KernelHandle h_kernel):
        """Create from known params (called by launch() builder)."""
        cdef KernelNode n = KernelNode.__new__(KernelNode)
        n._h_node = h_node
        n._grid = grid
        n._block = block
        n._shmem_size = shmem_size
        n._h_kernel = h_kernel
        return n

    @staticmethod
    cdef KernelNode _create_from_driver(GraphNodeHandle h_node):
        """Create by fetching params from the driver (called by _create factory)."""
        cdef cydriver.CUgraphNode node = as_cu(h_node)
        cdef cydriver.CUDA_KERNEL_NODE_PARAMS params
        with nogil:
            HANDLE_RETURN(cydriver.cuGraphKernelNodeGetParams(node, &params))
        cdef KernelHandle h_kernel = create_kernel_handle_ref(params.kern)
        return KernelNode._create_with_params(
            h_node,
            (params.gridDimX, params.gridDimY, params.gridDimZ),
            (params.blockDimX, params.blockDimY, params.blockDimZ),
            params.sharedMemBytes,
            h_kernel)

    def __repr__(self) -> str:
        return (f"<KernelNode handle=0x{as_intptr(self._h_node):x}"
                f" kernel=0x{as_intptr(self._h_kernel):x}>")

    def update(
        self,
        *,
        config: LaunchConfig | None = None,
        kernel: Kernel | None = None,
        args=None,
    ) -> None:
        """Replace selected kernel launch parameters.

        Omitted parameters preserve their current values. Changing ``kernel``
        requires ``args``, including ``args=()`` for a no-argument kernel.
        Clustered and cooperative kernel nodes are not supported.

        .. warning::

            Use caution when a retained kernel argument directly or indirectly
            owns a graph. Any reference cycle involving the argument and a
            graph that retains it cannot be broken by Python's cyclic garbage
            collector. Use a weak reference to break such cycles.
        """
        cdef LaunchConfig c_config
        cdef Kernel c_kernel
        cdef ParamHolder arg_holder
        cdef object kernel_args
        cdef KernelHandle h_kernel = self._h_kernel
        cdef OpaqueHandle kernel_owner
        cdef OpaqueHandle args_owner
        cdef GraphHandle h_graph = graph_node_get_graph(self._h_node)
        cdef cydriver.CUgraphNode node = as_cu(self._h_node)
        cdef cydriver.CUgraphNodeParams params

        if config is not None:
            c_config = config
            if (c_config.cluster is not None or
                    c_config.is_cooperative):
                raise NotImplementedError(
                    "updating clustered or cooperative kernel nodes is not "
                    "supported")
        _require_graph_node_update_support()
        _reject_unsupported_kernel_node(node)
        if kernel is not None:
            if args is None:
                raise ValueError("changing kernel requires args")
            c_kernel = kernel
            h_kernel = c_kernel._h_kernel
        if args is not None:
            arg_holder = ParamHolder(args)

        c_memset(&params, 0, sizeof(params))
        params.type = cydriver.CU_GRAPH_NODE_TYPE_KERNEL
        with nogil:
            HANDLE_RETURN(cydriver.cuGraphKernelNodeGetParams(
                node, <cydriver.CUDA_KERNEL_NODE_PARAMS*>&params.kernel))
        HANDLE_RETURN(graph_get_attachment(
            h_graph, node, &kernel_owner, &args_owner))

        if config is not None:
            params.kernel.gridDimX = c_config.grid[0]
            params.kernel.gridDimY = c_config.grid[1]
            params.kernel.gridDimZ = c_config.grid[2]
            params.kernel.blockDimX = c_config.block[0]
            params.kernel.blockDimY = c_config.block[1]
            params.kernel.blockDimZ = c_config.block[2]
            params.kernel.sharedMemBytes = c_config.shmem_size
        if kernel is not None:
            params.kernel.kern = as_cu(h_kernel)
            params.kernel.func = <cydriver.CUfunction>NULL
            params.kernel.ctx = <cydriver.CUcontext>NULL
            kernel_owner = h_kernel
        if args is not None:
            params.kernel.kernelParams = <void**><uintptr_t>arg_holder.ptr
            params.kernel.extra = NULL
            kernel_args = arg_holder.kernel_args
            if kernel_args is None:
                args_owner = OpaqueHandle()
            else:
                args_owner = make_opaque_py(kernel_args)

        _set_definition_node_params(
            self._h_node, &params, kernel_owner, args_owner)
        self._grid = (
            params.kernel.gridDimX,
            params.kernel.gridDimY,
            params.kernel.gridDimZ,
        )
        self._block = (
            params.kernel.blockDimX,
            params.kernel.blockDimY,
            params.kernel.blockDimZ,
        )
        self._shmem_size = params.kernel.sharedMemBytes
        self._h_kernel = h_kernel

    @property
    def grid(self) -> tuple[int, int, int]:
        """Grid dimensions as a 3-tuple (gridDimX, gridDimY, gridDimZ)."""
        return self._grid

    @property
    def block(self) -> tuple[int, int, int]:
        """Block dimensions as a 3-tuple (blockDimX, blockDimY, blockDimZ)."""
        return self._block

    @property
    def shmem_size(self) -> int:
        """Dynamic shared memory size in bytes."""
        return self._shmem_size

    @property
    def kernel(self) -> Kernel:
        """The Kernel object for this launch node."""
        return Kernel._from_handle(self._h_kernel)

    @property
    def config(self) -> LaunchConfig:
        """A LaunchConfig reconstructed from this node's grid, block, and shmem_size.

        Note: cluster dimensions and is_cooperative are not preserved
        by the CUDA driver's kernel node params, so they are not included.
        """
        return LaunchConfig(grid=self._grid, block=self._block,
                            shmem_size=self._shmem_size)


cdef class AllocNode(GraphNode):
    """A memory allocation node.

    Properties
    ----------
    dptr : int
        The device pointer for the allocation.
    bytesize : int
        The number of bytes allocated.
    device_id : int
        The device on which the allocation was made.
    memory_type : GraphMemoryType | str
        The type of memory allocated.
    peer_access : tuple of int
        Device IDs that have read-write access to this allocation.
    """

    @staticmethod
    cdef AllocNode _create_with_params(GraphNodeHandle h_node,
                                       cydriver.CUdeviceptr dptr, size_t bytesize,
                                       int device_id, str memory_type, tuple peer_access):
        """Create from known params (called by allocate() builder)."""
        cdef AllocNode n = AllocNode.__new__(AllocNode)
        n._h_node = h_node
        n._dptr = dptr
        n._bytesize = bytesize
        n._device_id = device_id
        n._memory_type = memory_type
        n._peer_access = peer_access
        return n

    @staticmethod
    cdef AllocNode _create_from_driver(GraphNodeHandle h_node):
        """Create by fetching params from the driver (called by _create factory)."""
        cdef cydriver.CUgraphNode node = as_cu(h_node)
        cdef cydriver.CUDA_MEM_ALLOC_NODE_PARAMS params
        with nogil:
            HANDLE_RETURN(cydriver.cuGraphMemAllocNodeGetParams(node, &params))

        cdef str memory_type
        if params.poolProps.allocType == cydriver.CUmemAllocationType.CU_MEM_ALLOCATION_TYPE_PINNED:
            if params.poolProps.location.type == cydriver.CUmemLocationType.CU_MEM_LOCATION_TYPE_HOST:
                memory_type = "host"
            else:
                memory_type = "device"
        else:
            IF CUDA_CORE_BUILD_MAJOR >= 13:
                if params.poolProps.allocType == cydriver.CUmemAllocationType.CU_MEM_ALLOCATION_TYPE_MANAGED:
                    memory_type = "managed"
                else:
                    memory_type = "device"
            ELSE:
                memory_type = "device"

        cdef list peer_ids = []
        cdef size_t i
        for i in range(params.accessDescCount):
            peer_ids.append(<int>params.accessDescs[i].location.id)

        return AllocNode._create_with_params(
            h_node, params.dptr, params.bytesize,
            <int>params.poolProps.location.id, memory_type, tuple(peer_ids))

    def __repr__(self) -> str:
        return (f"<AllocNode handle=0x{as_intptr(self._h_node):x}"
                f" dptr=0x{self._dptr:x} size={self._bytesize}>")

    @property
    def dptr(self) -> int:
        """The device pointer for the allocation."""
        return self._dptr

    @property
    def bytesize(self) -> int:
        """The number of bytes allocated."""
        return self._bytesize

    @property
    def device_id(self) -> int:
        """The device on which the allocation was made."""
        return self._device_id

    @property
    def memory_type(self) -> str:
        """The type of memory: ``"device"``, ``"host"``, or ``"managed"``."""
        return self._memory_type

    @property
    def peer_access(self) -> tuple[int, ...]:
        """Device IDs with read-write access to this allocation."""
        return self._peer_access


cdef class FreeNode(GraphNode):
    """A memory deallocation node.

    Properties
    ----------
    dptr : int
        The device pointer being freed.
    """

    @staticmethod
    cdef FreeNode _create_with_params(GraphNodeHandle h_node,
                                      cydriver.CUdeviceptr dptr):
        """Create from known params (called by deallocate() builder)."""
        cdef FreeNode n = FreeNode.__new__(FreeNode)
        n._h_node = h_node
        n._dptr = dptr
        return n

    @staticmethod
    cdef FreeNode _create_from_driver(GraphNodeHandle h_node):
        """Create by fetching params from the driver (called by _create factory)."""
        cdef cydriver.CUgraphNode node = as_cu(h_node)
        cdef cydriver.CUdeviceptr dptr
        with nogil:
            HANDLE_RETURN(cydriver.cuGraphMemFreeNodeGetParams(node, &dptr))
        return FreeNode._create_with_params(h_node, dptr)

    def __repr__(self) -> str:
        return f"<FreeNode handle=0x{as_intptr(self._h_node):x} dptr=0x{self._dptr:x}>"

    @property
    def dptr(self) -> int:
        """The device pointer being freed."""
        return self._dptr


cdef class MemsetNode(GraphNode):
    """A memset node.

    Properties
    ----------
    dptr : int
        The destination device pointer.
    value : int
        The fill value.
    element_size : int
        Element size in bytes (1, 2, or 4).
    width : int
        Width of the row in elements.
    height : int
        Number of rows.
    pitch : int
        Pitch in bytes (unused if height is 1).
    """

    @staticmethod
    cdef MemsetNode _create_with_params(GraphNodeHandle h_node,
                                        cydriver.CUdeviceptr dptr, unsigned int value,
                                        unsigned int element_size, size_t width,
                                        size_t height, size_t pitch):
        """Create from known params (called by memset() builder)."""
        cdef MemsetNode n = MemsetNode.__new__(MemsetNode)
        n._h_node = h_node
        n._dptr = dptr
        n._value = value
        n._element_size = element_size
        n._width = width
        n._height = height
        n._pitch = pitch
        return n

    @staticmethod
    cdef MemsetNode _create_from_driver(GraphNodeHandle h_node):
        """Create by fetching params from the driver (called by _create factory)."""
        cdef cydriver.CUgraphNode node = as_cu(h_node)
        cdef cydriver.CUDA_MEMSET_NODE_PARAMS params
        with nogil:
            HANDLE_RETURN(cydriver.cuGraphMemsetNodeGetParams(node, &params))
        return MemsetNode._create_with_params(
            h_node, params.dst, params.value,
            params.elementSize, params.width, params.height, params.pitch)

    def __repr__(self) -> str:
        return (f"<MemsetNode handle=0x{as_intptr(self._h_node):x}"
                f" dptr=0x{self._dptr:x} value={self._value}>")

    def update(
        self,
        *,
        dst: Buffer | int | None = None,
        value=None,
        width: int | None = None,
        height: int | None = None,
        pitch: int | None = None,
        dst_owner=None,
    ) -> None:
        """Replace selected memset parameters.

        Omitted parameters preserve their current values. ``dst_owner`` may
        only accompany a raw-address ``dst``.

        With CUDA 12.2 through 13.1, the node's intended CUDA context must be
        current when this method is called. CUDA driver and ``cuda.bindings``
        versions 13.2 and newer preserve the recorded context automatically.

        .. warning::

            Use caution when a retained operand owner directly or indirectly
            owns a graph. Any reference cycle involving the owner and a graph
            that retains it cannot be broken by Python's cyclic garbage
            collector. Use a weak reference to break such cycles.
        """
        cdef OpaqueHandle dst_attachment_owner
        cdef GraphHandle h_graph
        cdef cydriver.CUgraphNode node = as_cu(self._h_node)
        cdef cydriver.CUcontext ctx = NULL
        cdef cydriver.CUDA_MEMSET_NODE_PARAMS current
        cdef cydriver.CUgraphNodeParams params
        cdef object queried

        if dst is None and dst_owner is not None:
            raise ValueError("dst_owner requires dst")
        if (dst is None and value is None and width is None and
                height is None and pitch is None):
            return

        c_memset(&params, 0, sizeof(params))
        params.type = cydriver.CU_GRAPH_NODE_TYPE_MEMSET
        with nogil:
            HANDLE_RETURN(cydriver.cuGraphMemsetNodeGetParams(
                node, &current))
        if _check_node_get_params():
            queried = handle_return(driver.cuGraphNodeGetParams(
                <uintptr_t>node))
            ctx = <cydriver.CUcontext><uintptr_t>int(queried.memset.ctx)
        else:
            with nogil:
                HANDLE_RETURN(cydriver.cuCtxGetCurrent(&ctx))

        cdef cydriver.CUdeviceptr c_dst = current.dst
        cdef unsigned int c_value = current.value
        cdef unsigned int c_element_size = current.elementSize
        cdef size_t c_width = current.width
        cdef size_t c_height = current.height
        cdef size_t c_pitch = current.pitch

        if dst is None:
            h_graph = graph_node_get_graph(self._h_node)
            HANDLE_RETURN(graph_get_attachment(
                h_graph, node,
                &dst_attachment_owner, NULL))
        else:
            dst_attachment_owner = _resolve_memcpy_operand(
                dst, dst_owner, "dst", &c_dst)

        if value is not None:
            c_value, c_element_size = _parse_fill_value(value)
        if width is not None:
            c_width = width
        if height is not None:
            c_height = height
        if pitch is not None:
            c_pitch = pitch

        params.memset.dst = c_dst
        params.memset.value = c_value
        params.memset.elementSize = c_element_size
        params.memset.width = c_width
        params.memset.height = c_height
        params.memset.pitch = c_pitch
        params.memset.ctx = ctx

        _set_definition_node_params(
            self._h_node, &params, dst_attachment_owner,
            OpaqueHandle(), params.memset.ctx)
        self._dptr = c_dst
        self._value = c_value
        self._element_size = c_element_size
        self._width = c_width
        self._height = c_height
        self._pitch = c_pitch

    @property
    def dptr(self) -> int:
        """The destination device pointer."""
        return self._dptr

    @property
    def value(self) -> int:
        """The fill value."""
        return self._value

    @property
    def element_size(self) -> int:
        """Element size in bytes (1, 2, or 4)."""
        return self._element_size

    @property
    def width(self) -> int:
        """Width of the row in elements."""
        return self._width

    @property
    def height(self) -> int:
        """Number of rows."""
        return self._height

    @property
    def pitch(self) -> int:
        """Pitch in bytes (unused if height is 1)."""
        return self._pitch


cdef class MemcpyNode(GraphNode):
    """A memcpy node.

    Properties
    ----------
    dst : int
        The destination pointer.
    src : int
        The source pointer.
    size : int
        The number of bytes copied.
    """

    @staticmethod
    cdef MemcpyNode _create_with_params(GraphNodeHandle h_node,
                                        cydriver.CUdeviceptr dst, cydriver.CUdeviceptr src,
                                        size_t size, cydriver.CUmemorytype dst_type,
                                        cydriver.CUmemorytype src_type):
        """Create from known params (called by memcpy() builder)."""
        cdef MemcpyNode n = MemcpyNode.__new__(MemcpyNode)
        n._h_node = h_node
        n._dst = dst
        n._src = src
        n._size = size
        n._dst_type = dst_type
        n._src_type = src_type
        return n

    @staticmethod
    cdef MemcpyNode _create_from_driver(GraphNodeHandle h_node):
        """Create by fetching params from the driver (called by _create factory)."""
        cdef cydriver.CUgraphNode node = as_cu(h_node)
        cdef cydriver.CUDA_MEMCPY3D params
        with nogil:
            HANDLE_RETURN(cydriver.cuGraphMemcpyNodeGetParams(node, &params))

        cdef cydriver.CUdeviceptr dst
        cdef cydriver.CUdeviceptr src
        if params.dstMemoryType == cydriver.CU_MEMORYTYPE_HOST:
            dst = <cydriver.CUdeviceptr><uintptr_t>params.dstHost
        else:
            dst = params.dstDevice
        if params.srcMemoryType == cydriver.CU_MEMORYTYPE_HOST:
            src = <cydriver.CUdeviceptr><uintptr_t>params.srcHost
        else:
            src = params.srcDevice

        return MemcpyNode._create_with_params(
            h_node, dst, src, params.WidthInBytes,
            params.dstMemoryType, params.srcMemoryType)

    def __repr__(self) -> str:
        cdef str dt = "H" if self._dst_type == cydriver.CU_MEMORYTYPE_HOST else "D"
        cdef str st = "H" if self._src_type == cydriver.CU_MEMORYTYPE_HOST else "D"
        return (f"<MemcpyNode handle=0x{as_intptr(self._h_node):x}"
                f" dst=0x{self._dst:x}({dt}) src=0x{self._src:x}({st}) size={self._size}>")

    def update(
        self,
        *,
        dst: Buffer | int | None = None,
        src: Buffer | int | None = None,
        size: int | None = None,
        dst_owner=None,
        src_owner=None,
    ) -> None:
        """Replace selected memcpy parameters.

        Omitted parameters preserve their current values. ``dst_owner`` and
        ``src_owner`` may only accompany their corresponding raw addresses.
        Multidimensional, pitched, offset, and array-backed memcpy nodes are
        not supported.

        With CUDA 12.2 through 13.1, the node's intended CUDA context must be
        current when this method is called. CUDA driver and ``cuda.bindings``
        versions 13.2 and newer preserve the recorded context automatically.

        .. warning::

            Use caution when a retained operand owner directly or indirectly
            owns a graph. Any reference cycle involving the owner and a graph
            that retains it cannot be broken by Python's cyclic garbage
            collector. Use a weak reference to break such cycles.
        """
        cdef cydriver.CUdeviceptr c_dst = self._dst
        cdef cydriver.CUdeviceptr c_src = self._src
        cdef OpaqueHandle dst_attachment_owner
        cdef OpaqueHandle src_attachment_owner
        cdef GraphHandle h_graph = graph_node_get_graph(self._h_node)
        cdef cydriver.CUgraphNode node = as_cu(self._h_node)
        cdef cydriver.CUcontext ctx = NULL
        cdef cydriver.CUgraphNodeParams params
        cdef cydriver.CUmemorytype c_dst_type
        cdef cydriver.CUmemorytype c_src_type
        cdef object queried

        if dst is None and dst_owner is not None:
            raise ValueError("dst_owner requires dst")
        if src is None and src_owner is not None:
            raise ValueError("src_owner requires src")
        if dst is None and src is None and size is None:
            return

        c_memset(&params, 0, sizeof(params))
        params.type = cydriver.CU_GRAPH_NODE_TYPE_MEMCPY
        with nogil:
            HANDLE_RETURN(cydriver.cuGraphMemcpyNodeGetParams(
                node, &params.memcpy.copyParams))
        if _check_node_get_params():
            queried = handle_return(driver.cuGraphNodeGetParams(
                <uintptr_t>node))
            ctx = <cydriver.CUcontext><uintptr_t>int(
                queried.memcpy.copyCtx)
        else:
            with nogil:
                HANDLE_RETURN(cydriver.cuCtxGetCurrent(&ctx))
        params.memcpy.copyCtx = ctx

        if not _is_supported_memcpy_descriptor(&params.memcpy.copyParams):
            raise NotImplementedError(
                "updating multidimensional, pitched, offset, or array-backed "
                "memcpy nodes is not supported")

        c_dst_type = params.memcpy.copyParams.dstMemoryType
        c_src_type = params.memcpy.copyParams.srcMemoryType
        if c_dst_type == cydriver.CU_MEMORYTYPE_HOST:
            c_dst = <cydriver.CUdeviceptr><uintptr_t>(
                params.memcpy.copyParams.dstHost)
        elif c_dst_type == cydriver.CU_MEMORYTYPE_DEVICE:
            c_dst = params.memcpy.copyParams.dstDevice
        else:
            raise NotImplementedError(
                f"unsupported destination memory type: {int(c_dst_type)}")
        if c_src_type == cydriver.CU_MEMORYTYPE_HOST:
            c_src = <cydriver.CUdeviceptr><uintptr_t>(
                params.memcpy.copyParams.srcHost)
        elif c_src_type == cydriver.CU_MEMORYTYPE_DEVICE:
            c_src = params.memcpy.copyParams.srcDevice
        else:
            raise NotImplementedError(
                f"unsupported source memory type: {int(c_src_type)}")

        HANDLE_RETURN(graph_get_attachment(
            h_graph, node,
            &dst_attachment_owner, &src_attachment_owner))
        if dst is not None:
            dst_attachment_owner = _resolve_memcpy_operand(
                dst, dst_owner, "dst", &c_dst)
            c_dst_type = _get_memcpy_memory_type(c_dst)
            params.memcpy.copyParams.dstMemoryType = c_dst_type
            params.memcpy.copyParams.dstHost = NULL
            params.memcpy.copyParams.dstDevice = 0
            params.memcpy.copyParams.dstArray = NULL
            params.memcpy.copyParams.reserved1 = NULL
            if c_dst_type == cydriver.CU_MEMORYTYPE_HOST:
                params.memcpy.copyParams.dstHost = <void*><uintptr_t>c_dst
            else:
                params.memcpy.copyParams.dstDevice = c_dst
        if src is not None:
            src_attachment_owner = _resolve_memcpy_operand(
                src, src_owner, "src", &c_src)
            c_src_type = _get_memcpy_memory_type(c_src)
            params.memcpy.copyParams.srcMemoryType = c_src_type
            params.memcpy.copyParams.srcHost = NULL
            params.memcpy.copyParams.srcDevice = 0
            params.memcpy.copyParams.srcArray = NULL
            params.memcpy.copyParams.reserved0 = NULL
            if c_src_type == cydriver.CU_MEMORYTYPE_HOST:
                params.memcpy.copyParams.srcHost = <void*><uintptr_t>c_src
            else:
                params.memcpy.copyParams.srcDevice = c_src
        if size is not None:
            params.memcpy.copyParams.WidthInBytes = size

        _set_definition_node_params(
            self._h_node, &params,
            dst_attachment_owner, src_attachment_owner,
            params.memcpy.copyCtx)
        self._dst = c_dst
        self._src = c_src
        self._size = params.memcpy.copyParams.WidthInBytes
        self._dst_type = c_dst_type
        self._src_type = c_src_type

    @property
    def dst(self) -> int:
        """The destination pointer."""
        return self._dst

    @property
    def src(self) -> int:
        """The source pointer."""
        return self._src

    @property
    def size(self) -> int:
        """The number of bytes copied."""
        return self._size


cdef class ChildGraphNode(GraphNode):
    """A child graph node.

    Properties
    ----------
    child_graph : GraphDefinition
        The embedded graph definition (non-owning wrapper).
    """

    @staticmethod
    cdef ChildGraphNode _create_with_params(GraphNodeHandle h_node,
                                            GraphHandle h_child_graph):
        """Create from known params (called by embed() builder)."""
        cdef ChildGraphNode n = ChildGraphNode.__new__(ChildGraphNode)
        n._h_node = h_node
        n._h_child_graph = h_child_graph
        return n

    @staticmethod
    cdef ChildGraphNode _create_from_driver(GraphNodeHandle h_node):
        """Create by fetching params from the driver (called by _create factory)."""
        cdef cydriver.CUgraphNode node = as_cu(h_node)
        cdef cydriver.CUgraph child_graph = NULL
        with nogil:
            HANDLE_RETURN(cydriver.cuGraphChildGraphNodeGetGraph(node, &child_graph))
        cdef GraphHandle h_graph = graph_node_get_graph(h_node)
        cdef GraphHandle h_child = create_child_graph_handle(
            child_graph, h_graph, node)
        return ChildGraphNode._create_with_params(h_node, h_child)

    def __repr__(self) -> str:
        return (f"<ChildGraphNode handle=0x{as_intptr(self._h_node):x}"
                f" child=0x{as_intptr(self._h_child_graph):x}>")

    def update(self, child: GraphDefinition) -> None:
        """Replace the embedded graph with a clone of ``child``.

        ``child`` must belong to an independent graph hierarchy.
        """
        cdef GraphHandle h_parent = graph_node_get_graph(self._h_node)
        cdef GraphHandle h_replacement
        cdef cydriver.CUgraphNode node = as_cu(self._h_node)
        cdef cydriver.CUgraphNodeParams params
        cdef cydriver.CUresult commit_status
        cdef PreparedChildGraphUpdate prepared

        _require_graph_node_update_support()
        c_memset(&params, 0, sizeof(params))
        params.type = cydriver.CU_GRAPH_NODE_TYPE_GRAPH
        params.graph.graph = as_cu(child._h_graph)

        HANDLE_RETURN(graph_prepare_child_graph_update(
            h_parent, self._h_child_graph, node,
            child._h_graph, &prepared))
        with nogil:
            HANDLE_RETURN(cydriver.cuGraphNodeSetParams(
                node, &params))
        try:
            commit_status = graph_commit_child_graph_update(
                prepared, &h_replacement)
        finally:
            if h_replacement:
                self._h_child_graph = h_replacement
        HANDLE_RETURN(commit_status)

    @property
    def child_graph(self) -> GraphDefinition:
        """The embedded graph definition (non-owning wrapper)."""
        return GraphDefinition._from_handle(self._h_child_graph)


cdef class EventRecordNode(GraphNode):
    """An event record node.

    Properties
    ----------
    event : Event
        The event being recorded.
    """

    @staticmethod
    cdef EventRecordNode _create_with_params(GraphNodeHandle h_node,
                                             EventHandle h_event):
        """Create from known params (called by record() builder)."""
        cdef EventRecordNode n = EventRecordNode.__new__(EventRecordNode)
        n._h_node = h_node
        n._h_event = h_event
        return n

    @staticmethod
    cdef EventRecordNode _create_from_driver(GraphNodeHandle h_node):
        """Create by fetching params from the driver (called by _create factory)."""
        cdef cydriver.CUgraphNode node = as_cu(h_node)
        cdef cydriver.CUevent event
        with nogil:
            HANDLE_RETURN(cydriver.cuGraphEventRecordNodeGetEvent(node, &event))
        cdef EventHandle h_event = create_event_handle_ref(event)
        return EventRecordNode._create_with_params(h_node, h_event)

    def __repr__(self) -> str:
        return (f"<EventRecordNode handle=0x{as_intptr(self._h_node):x}"
                f" event=0x{as_intptr(self._h_event):x}>")

    def update(self, event: Event) -> None:
        """Replace the event recorded by this node."""
        cdef OpaqueHandle event_owner = event._h_event
        cdef cydriver.CUgraphNodeParams params

        c_memset(&params, 0, sizeof(params))
        params.type = cydriver.CU_GRAPH_NODE_TYPE_EVENT_RECORD
        params.eventRecord.event = as_cu(event._h_event)

        _set_definition_node_params(
            self._h_node, &params, event_owner)
        self._h_event = event._h_event

    @property
    def event(self) -> Event:
        """The event being recorded."""
        return Event._from_handle(self._h_event)


cdef class EventWaitNode(GraphNode):
    """An event wait node.

    Properties
    ----------
    event : Event
        The event being waited on.
    """

    @staticmethod
    cdef EventWaitNode _create_with_params(GraphNodeHandle h_node,
                                           EventHandle h_event):
        """Create from known params (called by wait() builder)."""
        cdef EventWaitNode n = EventWaitNode.__new__(EventWaitNode)
        n._h_node = h_node
        n._h_event = h_event
        return n

    @staticmethod
    cdef EventWaitNode _create_from_driver(GraphNodeHandle h_node):
        """Create by fetching params from the driver (called by _create factory)."""
        cdef cydriver.CUgraphNode node = as_cu(h_node)
        cdef cydriver.CUevent event
        with nogil:
            HANDLE_RETURN(cydriver.cuGraphEventWaitNodeGetEvent(node, &event))
        cdef EventHandle h_event = create_event_handle_ref(event)
        return EventWaitNode._create_with_params(h_node, h_event)

    def __repr__(self) -> str:
        return (f"<EventWaitNode handle=0x{as_intptr(self._h_node):x}"
                f" event=0x{as_intptr(self._h_event):x}>")

    def update(self, event: Event) -> None:
        """Replace the event waited on by this node."""
        cdef OpaqueHandle event_owner = event._h_event
        cdef cydriver.CUgraphNodeParams params

        c_memset(&params, 0, sizeof(params))
        params.type = cydriver.CU_GRAPH_NODE_TYPE_WAIT_EVENT
        params.eventWait.event = as_cu(event._h_event)

        _set_definition_node_params(
            self._h_node, &params, event_owner)
        self._h_event = event._h_event

    @property
    def event(self) -> Event:
        """The event being waited on."""
        return Event._from_handle(self._h_event)


cdef class HostCallbackNode(GraphNode):
    """A host callback node.

    Properties
    ----------
    callback : callable or None
        The Python callable (None for ctypes function pointer callbacks).
    """

    @staticmethod
    cdef HostCallbackNode _create_with_params(GraphNodeHandle h_node,
                                              object callable_obj, cydriver.CUhostFn fn,
                                              void* user_data):
        """Create from known params (called by callback() builder)."""
        cdef HostCallbackNode n = HostCallbackNode.__new__(HostCallbackNode)
        n._h_node = h_node
        n._callable = callable_obj
        n._fn = fn
        n._user_data = user_data
        return n

    @staticmethod
    cdef HostCallbackNode _create_from_driver(GraphNodeHandle h_node):
        """Create by fetching params from the driver (called by _create factory)."""
        cdef cydriver.CUgraphNode node = as_cu(h_node)
        cdef cydriver.CUDA_HOST_NODE_PARAMS params
        with nogil:
            HANDLE_RETURN(cydriver.cuGraphHostNodeGetParams(node, &params))

        cdef object callable_obj = None
        if _is_py_host_trampoline(params.fn):
            callable_obj = <object>params.userData

        return HostCallbackNode._create_with_params(
            h_node, callable_obj, params.fn, params.userData)

    def __repr__(self) -> str:
        if self._callable is not None:
            name = getattr(self._callable, '__name__', '?')
            return (f"<HostCallbackNode handle=0x{as_intptr(self._h_node):x}"
                    f" callback={name}>")
        return (f"<HostCallbackNode handle=0x{as_intptr(self._h_node):x}"
                f" cfunc=0x{<uintptr_t>self._fn:x}>")

    def update(self, fn, *, user_data=None) -> None:
        """Replace the callback and user-data binding for this node.

        ``fn`` accepts the same forms as :meth:`~graph.GraphNode.callback`: a
        Python callable, or a ctypes function pointer whose declared prototype
        matches ``CUhostFn`` (``void (*)(void*)``). A mismatched ctypes
        prototype raises ``TypeError``.

        .. warning::

            Callbacks must not call CUDA API functions. Doing so may
            deadlock or corrupt driver state.

            Use caution when a Python callback retains an object that owns a
            graph. Any reference cycle involving the callback and a graph that
            retains it cannot be broken by Python's cyclic garbage collector.
            Use a weak reference to break such cycles.
        """
        cdef cydriver.CUhostFn c_fn
        cdef void* c_user_data
        cdef OpaqueHandle fn_owner, data_owner
        cdef cydriver.CUgraphNodeParams params

        _resolve_host_callback(
            fn, user_data, &c_fn, &c_user_data, &fn_owner, &data_owner)
        c_memset(&params, 0, sizeof(params))
        params.type = cydriver.CU_GRAPH_NODE_TYPE_HOST
        params.host.fn = c_fn
        params.host.userData = c_user_data

        _set_definition_node_params(
            self._h_node, &params, fn_owner, data_owner)
        self._callable = fn if _is_py_host_trampoline(c_fn) else None
        self._fn = c_fn
        self._user_data = c_user_data

    @property
    def callback(self):
        """The Python callable, or None for ctypes function pointer callbacks."""
        return self._callable


cdef class ConditionalNode(GraphNode):
    """Base class for conditional nodes.

    When created via builder methods (if_then, if_else, while_loop, switch),
    a specific subclass (IfNode, IfElseNode, WhileNode, SwitchNode) is
    returned. When reconstructed from the driver on CUDA 13.2+, the
    correct subclass is determined via cuGraphNodeGetParams. On older
    drivers, this base class is used as a fallback.

    Properties
    ----------
    condition : GraphCondition or None
        The condition variable controlling execution (None pre-13.2).
    cond_type : str or None
        The conditional type ("if", "while", or "switch"; None pre-13.2).
    branches : tuple of GraphDefinition
        The body graphs for each branch (empty pre-13.2).
    """

    @staticmethod
    cdef ConditionalNode _create_from_driver(GraphNodeHandle h_node):
        cdef ConditionalNode n
        if not _check_node_get_params():
            n = ConditionalNode.__new__(ConditionalNode)
            n._h_node = h_node
            n._condition = None
            n._cond_type = cydriver.CU_GRAPH_COND_TYPE_IF
            n._branches = ()
            return n

        cdef cydriver.CUgraphNode node = as_cu(h_node)
        params = handle_return(driver.cuGraphNodeGetParams(
            <uintptr_t>node))
        cond_params = params.conditional
        cdef int cond_type_int = int(cond_params.type)
        cdef unsigned int size = int(cond_params.size)

        cdef GraphCondition condition = GraphCondition.__new__(GraphCondition)
        condition._c_handle = <cydriver.CUgraphConditionalHandle>(
            <unsigned long long>int(cond_params.handle))

        cdef GraphHandle h_graph = graph_node_get_graph(h_node)
        cdef list branch_list = []
        cdef unsigned int i
        cdef GraphHandle h_branch
        if cond_params.phGraph_out is not None:
            for i in range(size):
                h_branch = create_child_graph_handle(
                    <cydriver.CUgraph><uintptr_t>int(cond_params.phGraph_out[i]),
                    h_graph, node)
                branch_list.append(GraphDefinition._from_handle(h_branch))
        cdef tuple branches = tuple(branch_list)

        cdef type cls
        if cond_type_int == <int>cydriver.CU_GRAPH_COND_TYPE_IF:
            if size == 1:
                cls = IfNode
            else:
                cls = IfElseNode
        elif cond_type_int == <int>cydriver.CU_GRAPH_COND_TYPE_WHILE:
            cls = WhileNode
        else:
            cls = SwitchNode

        n = cls.__new__(cls)
        n._h_node = h_node
        n._condition = condition
        n._cond_type = <cydriver.CUgraphConditionalNodeType>cond_type_int
        n._branches = branches
        return n

    def __repr__(self) -> str:
        return f"<ConditionalNode handle=0x{as_intptr(self._h_node):x}>"

    @property
    def condition(self) -> GraphCondition | None:
        """The condition variable controlling execution."""
        return self._condition

    @property
    def cond_type(self) -> GraphConditionalType | None:
        """The conditional type: GraphConditionalType.IF, .WHILE, or .SWITCH

        Returns None when reconstructed from the driver pre-CUDA 13.2,
        as the conditional type cannot be determined.
        """
        if self._condition is None:
            return None
        if self._cond_type == cydriver.CU_GRAPH_COND_TYPE_IF:
            return GraphConditionalType("if")
        elif self._cond_type == cydriver.CU_GRAPH_COND_TYPE_WHILE:
            return GraphConditionalType("while")
        else:
            return GraphConditionalType("switch")

    @property
    def branches(self) -> tuple[GraphDefinition, ...]:
        """The body graphs for each branch as a tuple of GraphDefinition.

        Returns an empty tuple when reconstructed from the driver
        pre-CUDA 13.2.
        """
        return self._branches


cdef class IfNode(ConditionalNode):
    """An if-conditional node."""

    def __repr__(self) -> str:
        return (f"<IfNode handle=0x{as_intptr(self._h_node):x}"
                f" condition=0x{<unsigned long long>self._condition._c_handle:x}>")

    @property
    def then(self) -> GraphDefinition:
        """The 'then' branch graph."""
        return self._branches[0]


cdef class IfElseNode(ConditionalNode):
    """An if-else conditional node."""

    def __repr__(self) -> str:
        return (f"<IfElseNode handle=0x{as_intptr(self._h_node):x}"
                f" condition=0x{<unsigned long long>self._condition._c_handle:x}>")

    @property
    def then(self) -> GraphDefinition:
        """The ``then`` branch graph (executed when condition is non-zero)."""
        return self._branches[0]

    @property
    def else_(self) -> GraphDefinition:
        """The ``else`` branch graph (executed when condition is zero)."""
        return self._branches[1]


cdef class WhileNode(ConditionalNode):
    """A while-loop conditional node."""

    def __repr__(self) -> str:
        return (f"<WhileNode handle=0x{as_intptr(self._h_node):x}"
                f" condition=0x{<unsigned long long>self._condition._c_handle:x}>")

    @property
    def body(self) -> GraphDefinition:
        """The loop body graph."""
        return self._branches[0]


cdef class SwitchNode(ConditionalNode):
    """A switch conditional node."""

    def __repr__(self) -> str:
        return (f"<SwitchNode handle=0x{as_intptr(self._h_node):x}"
                f" condition=0x{<unsigned long long>self._condition._c_handle:x}>")


cdef class ExecutableGraphNode:
    """A lightweight view pairing an executable graph with a source node.

    Create executable-node views with ``graph[node]``. CUDA validates that the
    node identifies a node in the executable graph when an operation is
    performed.
    """

    def __init__(self):
        raise RuntimeError(
            "directly constructing an executable graph node is not supported")

    def __repr__(self) -> str:
        return (
            f"<{type(self).__name__} graph=0x{as_intptr(self._h_graph_exec):x}"
            f" node=0x{as_intptr(self._h_node):x}>"
        )


cdef class ExecutableKernelNode(ExecutableGraphNode):
    """An executable kernel-node view."""

    def update(
        self,
        *,
        config: LaunchConfig,
        kernel: Kernel,
        args,
    ) -> None:
        """Replace all kernel launch parameters for future launches.

        ``args`` must contain the complete argument sequence; use ``args=()``
        for a no-argument kernel. Clustered and cooperative launch
        configurations are not supported.
        """
        cdef LaunchConfig c_config = config
        cdef Kernel c_kernel = kernel
        cdef ParamHolder arg_holder
        cdef object kernel_args
        cdef OpaqueHandle kernel_owner = c_kernel._h_kernel
        cdef OpaqueHandle args_owner
        cdef cydriver.CUgraphNodeParams params

        if c_config.cluster is not None or c_config.is_cooperative:
            raise NotImplementedError(
                "updating clustered or cooperative kernel nodes is not "
                "supported")
        arg_holder = ParamHolder(args)

        c_memset(&params, 0, sizeof(params))
        params.type = cydriver.CU_GRAPH_NODE_TYPE_KERNEL
        params.kernel.kern = as_cu(c_kernel._h_kernel)
        params.kernel.func = <cydriver.CUfunction>NULL
        params.kernel.gridDimX = c_config.grid[0]
        params.kernel.gridDimY = c_config.grid[1]
        params.kernel.gridDimZ = c_config.grid[2]
        params.kernel.blockDimX = c_config.block[0]
        params.kernel.blockDimY = c_config.block[1]
        params.kernel.blockDimZ = c_config.block[2]
        params.kernel.sharedMemBytes = c_config.shmem_size
        params.kernel.kernelParams = <void**><uintptr_t>arg_holder.ptr
        params.kernel.extra = NULL
        params.kernel.ctx = <cydriver.CUcontext>NULL

        kernel_args = arg_holder.kernel_args
        if kernel_args is not None:
            args_owner = make_opaque_py(kernel_args)
        _set_executable_node_params(
            self._h_graph_exec, self._h_node, &params,
            kernel_owner, args_owner)

    @property
    def is_enabled(self) -> bool:
        """Whether this node is enabled in the executable graph."""
        return _get_executable_node_enabled(
            self._h_graph_exec, self._h_node)

    def enable(self) -> None:
        """Enable this node in the executable graph."""
        _set_executable_node_enabled(
            self._h_graph_exec, self._h_node, True)

    def disable(self) -> None:
        """Disable this node in the executable graph."""
        _set_executable_node_enabled(
            self._h_graph_exec, self._h_node, False)


cdef class ExecutableMemsetNode(ExecutableGraphNode):
    """An executable memset-node view."""

    def update(
        self,
        *,
        dst: Buffer | int,
        value,
        size_t width,
        size_t height=1,
        size_t pitch=0,
    ) -> None:
        """Replace all memset parameters for future launches."""
        cdef cydriver.CUdeviceptr c_dst
        cdef OpaqueHandle dst_owner = _resolve_memcpy_operand(
            dst, None, "dst", &c_dst)
        cdef unsigned int c_value
        cdef unsigned int element_size
        c_value, element_size = _parse_fill_value(value)

        cdef cydriver.CUcontext ctx = NULL
        cdef cydriver.CUgraphNodeParams params
        with nogil:
            HANDLE_RETURN(cydriver.cuCtxGetCurrent(&ctx))

        c_memset(&params, 0, sizeof(params))
        params.type = cydriver.CU_GRAPH_NODE_TYPE_MEMSET
        params.memset.dst = c_dst
        params.memset.value = c_value
        params.memset.elementSize = element_size
        params.memset.width = width
        params.memset.height = height
        params.memset.pitch = pitch
        params.memset.ctx = ctx
        _set_executable_node_params(
            self._h_graph_exec, self._h_node, &params, dst_owner)

    @property
    def is_enabled(self) -> bool:
        """Whether this node is enabled in the executable graph."""
        return _get_executable_node_enabled(
            self._h_graph_exec, self._h_node)

    def enable(self) -> None:
        """Enable this node in the executable graph."""
        _set_executable_node_enabled(
            self._h_graph_exec, self._h_node, True)

    def disable(self) -> None:
        """Disable this node in the executable graph."""
        _set_executable_node_enabled(
            self._h_graph_exec, self._h_node, False)


cdef class ExecutableMemcpyNode(ExecutableGraphNode):
    """An executable memcpy-node view."""

    def update(
        self,
        *,
        dst: Buffer | int,
        src: Buffer | int,
        size_t size,
    ) -> None:
        """Replace all one-dimensional memcpy parameters for future launches."""
        cdef cydriver.CUdeviceptr c_dst
        cdef cydriver.CUdeviceptr c_src
        cdef OpaqueHandle dst_owner = _resolve_memcpy_operand(
            dst, None, "dst", &c_dst)
        cdef OpaqueHandle src_owner = _resolve_memcpy_operand(
            src, None, "src", &c_src)
        cdef cydriver.CUmemorytype dst_type
        cdef cydriver.CUmemorytype src_type
        cdef cydriver.CUcontext ctx = NULL
        cdef cydriver.CUgraphNodeParams params

        c_memset(&params, 0, sizeof(params))
        params.type = cydriver.CU_GRAPH_NODE_TYPE_MEMCPY
        _init_memcpy_params(
            c_dst, c_src, size, &params.memcpy.copyParams,
            &dst_type, &src_type)
        with nogil:
            HANDLE_RETURN(cydriver.cuCtxGetCurrent(&ctx))
        params.memcpy.copyCtx = ctx
        _set_executable_node_params(
            self._h_graph_exec, self._h_node, &params,
            dst_owner, src_owner)

    @property
    def is_enabled(self) -> bool:
        """Whether this node is enabled in the executable graph."""
        return _get_executable_node_enabled(
            self._h_graph_exec, self._h_node)

    def enable(self) -> None:
        """Enable this node in the executable graph."""
        _set_executable_node_enabled(
            self._h_graph_exec, self._h_node, True)

    def disable(self) -> None:
        """Disable this node in the executable graph."""
        _set_executable_node_enabled(
            self._h_graph_exec, self._h_node, False)


cdef class ExecutableChildGraphNode(ExecutableGraphNode):
    """An executable child-graph-node view."""

    def update(self, child: GraphDefinition) -> None:
        """Replace the embedded graph parameters for future launches."""
        cdef cydriver.CUgraphNodeParams params
        c_memset(&params, 0, sizeof(params))
        params.type = cydriver.CU_GRAPH_NODE_TYPE_GRAPH
        params.graph.graph = as_cu(child._h_graph)
        _set_executable_node_params(
            self._h_graph_exec, self._h_node, &params)


cdef class ExecutableEventRecordNode(ExecutableGraphNode):
    """An executable event-record-node view."""

    def update(self, event: Event) -> None:
        """Replace the event recorded by future launches."""
        cdef OpaqueHandle event_owner = event._h_event
        cdef cydriver.CUgraphNodeParams params
        c_memset(&params, 0, sizeof(params))
        params.type = cydriver.CU_GRAPH_NODE_TYPE_EVENT_RECORD
        params.eventRecord.event = as_cu(event._h_event)
        _set_executable_node_params(
            self._h_graph_exec, self._h_node, &params, event_owner)


cdef class ExecutableEventWaitNode(ExecutableGraphNode):
    """An executable event-wait-node view."""

    def update(self, event: Event) -> None:
        """Replace the event waited on by future launches."""
        cdef OpaqueHandle event_owner = event._h_event
        cdef cydriver.CUgraphNodeParams params
        c_memset(&params, 0, sizeof(params))
        params.type = cydriver.CU_GRAPH_NODE_TYPE_WAIT_EVENT
        params.eventWait.event = as_cu(event._h_event)
        _set_executable_node_params(
            self._h_graph_exec, self._h_node, &params, event_owner)


cdef class ExecutableHostCallbackNode(ExecutableGraphNode):
    """An executable host-callback-node view."""

    def update(self, fn, *, user_data=None) -> None:
        """Replace the callback and user-data binding for future launches.

        ``fn`` may be a Python callable, or a ctypes function pointer whose
        declared prototype matches ``CUhostFn`` (``void (*)(void*)``); a
        mismatched prototype raises ``TypeError``.

        .. warning::

            Callbacks must not call CUDA API functions. Doing so may deadlock
            or corrupt driver state.
        """
        cdef cydriver.CUhostFn c_fn
        cdef void* c_user_data
        cdef OpaqueHandle fn_owner
        cdef OpaqueHandle data_owner
        cdef cydriver.CUgraphNodeParams params

        _resolve_host_callback(
            fn, user_data, &c_fn, &c_user_data, &fn_owner, &data_owner)
        c_memset(&params, 0, sizeof(params))
        params.type = cydriver.CU_GRAPH_NODE_TYPE_HOST
        params.host.fn = c_fn
        params.host.userData = c_user_data
        _set_executable_node_params(
            self._h_graph_exec, self._h_node, &params,
            fn_owner, data_owner)


cdef ExecutableGraphNode create_executable_node_view(
        const GraphExecHandle& h_exec,
        GraphNode node):
    cdef type view_type
    if isinstance(node, KernelNode):
        view_type = ExecutableKernelNode
    elif isinstance(node, MemsetNode):
        view_type = ExecutableMemsetNode
    elif isinstance(node, MemcpyNode):
        view_type = ExecutableMemcpyNode
    elif isinstance(node, ChildGraphNode):
        view_type = ExecutableChildGraphNode
    elif isinstance(node, EventRecordNode):
        view_type = ExecutableEventRecordNode
    elif isinstance(node, EventWaitNode):
        view_type = ExecutableEventWaitNode
    elif isinstance(node, HostCallbackNode):
        view_type = ExecutableHostCallbackNode
    else:
        raise TypeError(
            f"{type(node).__name__} does not support executable updates")

    if as_cu(h_exec) == NULL:
        raise ValueError("executable graph has been closed")
    if as_cu(node._h_node) == NULL:
        raise ValueError("source graph node is no longer valid")

    cdef ExecutableGraphNode view = view_type.__new__(view_type)
    view._h_graph_exec = h_exec
    view._h_node = node._h_node
    return view
