feat(cuda): add monitoring reduction kernel

Replaces per-launch rewards/actions download with single epoch-end
reduction. monitoring_reduce kernel computes mean, std, min, max,
Sharpe estimate, and per-action counts via parallel reduction.
Single 48-byte download instead of N*8 bytes per kernel launch.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

crates/ml/src/cuda_pipeline/gpu_monitoring.rs

@@ -0,0 +1,127 @@
+#![allow(unsafe_code)]
+
+//! GPU monitoring reduction — aggregates per-experience rewards/actions
+//! into a compact summary without downloading full arrays.
+
+use std::sync::Arc;
+use candle_core::cuda_backend::cudarc;
+use cudarc::driver::{CudaFunction, CudaSlice, CudaStream, LaunchConfig, PushKernelArg};
+use cudarc::nvrtc::Ptx;
+use crate::MLError;
+
+/// Compact monitoring summary from GPU reduction (48 bytes).
+#[derive(Debug, Clone, Default)]
+pub struct MonitoringSummary {
+    pub mean_reward: f32,
+    pub reward_std: f32,
+    pub min_reward: f32,
+    pub max_reward: f32,
+    pub sharpe_estimate: f32,
+    pub action_counts: [usize; 5],
+    pub total_experiences: usize,
+}
+
+/// GPU monitoring reducer.
+#[allow(missing_debug_implementations)]
+pub struct GpuMonitoringReducer {
+    stream: Arc<CudaStream>,
+    kernel_func: CudaFunction,
+    summary_buf: CudaSlice<f32>,  // [12]
+}
+
+fn compile_monitoring_ptx() -> Result<Ptx, String> {
+    let kernel_src = include_str!("monitoring_kernel.cu");
+    cudarc::nvrtc::compile_ptx(kernel_src)
+        .map_err(|e| format!("monitoring kernel CUDA compilation failed: {e}"))
+}
+
+impl GpuMonitoringReducer {
+    pub fn new(stream: &Arc<CudaStream>) -> Result<Self, MLError> {
+        let context = stream.context();
+        let kernel_src = include_str!("monitoring_kernel.cu");
+        let ptx: Ptx = cudarc::nvrtc::compile_ptx(kernel_src)
+            .map_err(|e| MLError::ModelError(format!("monitoring kernel compile: {e}")))?;
+        let module = context.load_module(ptx)
+            .map_err(|e| MLError::ModelError(format!("monitoring module load: {e}")))?;
+        let kernel_func = module.load_function("monitoring_reduce")
+            .map_err(|e| MLError::ModelError(format!("monitoring_reduce load: {e}")))?;
+        let summary_buf = stream.alloc_zeros::<f32>(12)
+            .map_err(|e| MLError::ModelError(format!("monitoring summary alloc: {e}")))?;
+
+        Ok(Self { stream: Arc::clone(stream), kernel_func, summary_buf })
+    }
+
+    /// Launch reduction over rewards/actions buffers already on GPU.
+    /// Does NOT synchronize — caller must sync before reading result.
+    pub fn reduce(
+        &mut self,
+        rewards: &CudaSlice<f32>,
+        actions: &CudaSlice<i32>,
+        n: usize,
+    ) -> Result<(), MLError> {
+        let config = LaunchConfig {
+            grid_dim: (1, 1, 1),
+            block_dim: (256, 1, 1),
+            shared_mem_bytes: 0,
+        };
+        unsafe {
+            self.stream
+                .launch_builder(&self.kernel_func)
+                .arg(rewards)
+                .arg(actions)
+                .arg(&self.summary_buf)
+                .arg(&(n as i32))
+                .launch(config)
+                .map_err(|e| MLError::ModelError(format!("monitoring_reduce launch: {e}")))?;
+        }
+        Ok(())
+    }
+
+    /// Download summary from GPU (single 48-byte transfer).
+    pub fn download_summary(&self) -> Result<MonitoringSummary, MLError> {
+        let mut raw = vec![0.0_f32; 12];
+        self.stream.memcpy_dtoh(&self.summary_buf, &mut raw)
+            .map_err(|e| MLError::ModelError(format!("monitoring download: {e}")))?;
+        Ok(MonitoringSummary {
+            mean_reward: raw[0],
+            reward_std: raw[1],
+            min_reward: raw[2],
+            max_reward: raw[3],
+            sharpe_estimate: raw[4],
+            action_counts: [
+                raw[5] as usize, raw[6] as usize, raw[7] as usize,
+                raw[8] as usize, raw[9] as usize,
+            ],
+            total_experiences: raw[10] as usize,
+        })
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn test_monitoring_summary_default() {
+        let s = MonitoringSummary::default();
+        assert_eq!(s.total_experiences, 0);
+        assert_eq!(s.action_counts, [0; 5]);
+    }
+
+    /// Verify the PTX compiles without errors (skips gracefully when NVRTC is absent).
+    #[test]
+    fn test_monitoring_ptx_compilation() {
+        let result = compile_monitoring_ptx();
+        if let Err(ref e) = result {
+            if e.contains("NVRTC")
+                || e.contains("nvrtc")
+                || e.contains("not found")
+                || e.contains("No such file")
+            {
+                // NVRTC not installed — acceptable on CPU-only machines.
+                return;
+            }
+            panic!("monitoring kernel PTX compilation failed: {e}");
+        }
+    }
+}

crates/ml/src/cuda_pipeline/mod.rs

@@ -26,6 +26,8 @@ pub mod gpu_action_selector;
 pub mod gpu_statistics;
 #[cfg(feature = "cuda")]
 pub mod gpu_training_guard;
+#[cfg(feature = "cuda")]
+pub mod gpu_monitoring;
 // gpu_replay_buffer moved to ml-dqn crate
 
 /// Maximum bytes allowed for a single GPU upload (2 GB safety limit).

crates/ml/src/cuda_pipeline/monitoring_kernel.cu

@@ -0,0 +1,86 @@
+// Custom atomic min/max for float (must be defined BEFORE the kernel).
+__device__ float atomicMin_float(float* addr, float val) {
+    int* addr_as_int = (int*)addr;
+    int old = *addr_as_int, assumed;
+    do {
+        assumed = old;
+        old = atomicCAS(addr_as_int, assumed,
+            __float_as_int(fminf(val, __int_as_float(assumed))));
+    } while (assumed != old);
+    return __int_as_float(old);
+}
+
+__device__ float atomicMax_float(float* addr, float val) {
+    int* addr_as_int = (int*)addr;
+    int old = *addr_as_int, assumed;
+    do {
+        assumed = old;
+        old = atomicCAS(addr_as_int, assumed,
+            __float_as_int(fmaxf(val, __int_as_float(assumed))));
+    } while (assumed != old);
+    return __int_as_float(old);
+}
+
+// Reduce per-experience rewards and actions into a compact summary.
+// One block, parallel reduction across N elements.
+extern "C" __global__ void monitoring_reduce(
+    const float* __restrict__ rewards,   // [N]
+    const int*   __restrict__ actions,   // [N]
+    float*       summary,                // [12]: mean, std, min, max, sharpe, counts[5], total, _pad
+    int          N
+) {
+    __shared__ float s_sum;
+    __shared__ float s_sq_sum;
+    __shared__ float s_min;
+    __shared__ float s_max;
+    __shared__ int   s_counts[5];
+
+    int tid = threadIdx.x;
+    int stride = blockDim.x;
+
+    // Init shared memory
+    if (tid == 0) {
+        s_sum = 0.0f; s_sq_sum = 0.0f;
+        s_min = 1e30f; s_max = -1e30f;
+        for (int i = 0; i < 5; i++) s_counts[i] = 0;
+    }
+    __syncthreads();
+
+    // Thread-local accumulators
+    float local_sum = 0.0f, local_sq = 0.0f;
+    float local_min = 1e30f, local_max = -1e30f;
+    int local_counts[5] = {0, 0, 0, 0, 0};
+
+    for (int i = tid; i < N; i += stride) {
+        float r = rewards[i];
+        local_sum += r;
+        local_sq += r * r;
+        local_min = fminf(local_min, r);
+        local_max = fmaxf(local_max, r);
+        int a = actions[i];
+        if (a >= 0 && a < 5) local_counts[a]++;
+    }
+
+    // Warp reduction then atomic to shared
+    atomicAdd(&s_sum, local_sum);
+    atomicAdd(&s_sq_sum, local_sq);
+    atomicMin_float(&s_min, local_min);
+    atomicMax_float(&s_max, local_max);
+    for (int i = 0; i < 5; i++) atomicAdd(&s_counts[i], local_counts[i]);
+    __syncthreads();
+
+    // Thread 0 writes summary
+    if (tid == 0) {
+        float mean = s_sum / (float)N;
+        float var = s_sq_sum / (float)N - mean * mean;
+        float std = sqrtf(fmaxf(var, 0.0f));
+        summary[0] = mean;
+        summary[1] = std;
+        summary[2] = s_min;
+        summary[3] = s_max;
+        summary[4] = (std > 1e-8f) ? mean / std : 0.0f;  // Sharpe estimate
+        for (int i = 0; i < 5; i++) summary[5 + i] = (float)s_counts[i];
+        summary[10] = (float)N;
+        summary[11] = 0.0f;  // padding
+    }
+}