perf(0.69.4): batch: qwen35 persistent embed buffer (#4316) + hot-path allocs + stale q8 (#4258)

crates/aprender-serve/src/cuda/executor/core.rs

@@ -186,6 +186,7 @@ impl CudaExecutor {
             num_sms: context.multiprocessor_count().unwrap_or(8) as u32,
             // PMAT-027: Q8 activation cache starts invalid
             q8_activation_valid: false,
+            q8_activation_src: (0, 0),
             fp8_activation_cache_key: None,
             fp8_weight_scales: HashMap::new(),
             fp8_act_scale_buf: None,

crates/aprender-serve/src/cuda/executor/executor_api.rs

@@ -493,6 +493,27 @@ impl CudaExecutor {
         self.stream.synchronize()
     }
 
+    /// Overwrite a resident device buffer from the host, queued on the
+    /// execution stream: no allocation and no host wait. The copy lands after
+    /// every kernel already queued and before every kernel queued later, so a
+    /// buffer reused token after token is never overwritten while a previous
+    /// token's kernels still read it (#4316).
+    ///
+    /// # Safety
+    /// `src` must stay valid until the execution stream is synchronized.
+    ///
+    /// # Errors
+    /// `src` and `dst` differ in length, or the transfer fails.
+    pub unsafe fn upload_into_async(
+        &self,
+        dst: &mut GpuBuffer<f32>,
+        src: &[f32],
+    ) -> Result<(), GpuError> {
+        // SAFETY: the caller guarantees `src` outlives the next sync of
+        // `self.stream`, which is the stream the copy is queued on.
+        unsafe { dst.copy_from_host_async(src, &self.stream) }
+    }
+
     /// Get memory pool statistics (IMP-900d)
     #[must_use]
     pub fn pool_stats(&self) -> PoolStats {

crates/aprender-serve/src/cuda/executor/gdn_ops.rs

@@ -50,6 +50,9 @@ impl CudaExecutor {
     ) -> Result<(), GpuError> {
         for (i, &p) in ptrs.iter().enumerate() {
             validate_device_ptr(p, &format!("{kernel_name} arg {i}"))?;
+            // #4258: any pointer argument may be an output (several GDN kernels
+            // write in place), so any of them can stale the Q8 activation.
+            self.q8_activation_written(p);
         }
         let mut args: Vec<u64> = ptrs.to_vec();
         let mut raw: Vec<*mut std::ffi::c_void> = args
@@ -103,6 +106,9 @@ impl CudaExecutor {
     ) -> Result<(), GpuError> {
         for (i, &p) in ptrs.iter().enumerate() {
             validate_device_ptr(p, &format!("{kernel_name} arg {i}"))?;
+            // #4258: any pointer argument may be an output (several GDN kernels
+            // write in place), so any of them can stale the Q8 activation.
+            self.q8_activation_written(p);
         }
         let mut args: Vec<u64> = ptrs.to_vec();
         args.extend_from_slice(scalars);

crates/aprender-serve/src/cuda/executor/layer_norm_gpu.rs

@@ -150,6 +150,7 @@ impl CudaExecutor {
         hidden_size: u32,
         epsilon: f32,
     ) -> Result<(), GpuError> {
+        self.q8_activation_written(output.as_ptr()); // #4258
         // GH-559 DIAGNOSTIC: CPU RMSNorm bypass for Blackwell
         static CPU_RMSNORM: std::sync::OnceLock<bool> = std::sync::OnceLock::new();
         let use_cpu_rmsnorm = *CPU_RMSNORM.get_or_init(|| {
@@ -304,6 +305,7 @@ impl CudaExecutor {
         num_heads: u32,
         epsilon: f32,
     ) -> Result<(), GpuError> {
+        self.q8_activation_written(output.as_ptr()); // #4258
         let kernel_type = KernelType::PerHeadRmsNorm {
             head_dim,
             num_heads,

crates/aprender-serve/src/cuda/executor/mod.rs

@@ -299,6 +299,8 @@ mod poison_trace_test;
 // FALSIFY-QDOT-008 (#3111): the Q5_K GEMV against gguf-py's values of a llama.cpp block
 #[cfg(test)]
 mod tests_q5k_ggml;
+#[cfg(test)]
+mod tests_q8_activation_staleness;
 
 // COV-003 through COV-006 (layer preload, kv_cache, attention, quantized)
 #[cfg(test)]
@@ -635,6 +637,11 @@ pub struct CudaExecutor {
     // Set to true after q8_quantize_into; callers invalidate (set false)
     // when the input buffer content changes (e.g. after RMSNorm write).
     q8_activation_valid: bool,
+    // #4258: which activation the Q8 cache holds — (source buffer ptr, element
+    // count). `ensure_q8_activation` re-quantizes when a GEMV's input is not this
+    // buffer, and every in-place writer clears `q8_activation_valid` when it writes
+    // this buffer, so a DP4A GEMV never reads another activation's Q8_1 bytes.
+    q8_activation_src: (u64, u32),
     // PMAT-084: FP8 activation cache — skip redundant absmax+convert when
     // multiple FP8 GEMMs share the same input (QKV phase, FFN gate+up).
     // Saves 84 kernel pairs per prefill (3 per layer × 28 layers).

crates/aprender-serve/src/cuda/executor/q4k_mwv_gemv.rs

@@ -126,10 +126,7 @@ impl CudaExecutor {
         let q8_buf = unsafe { GpuBuffer::<u8>::from_raw_parts(q8_ptr, q8_len) };
 
         // Step 1: Quantize activations to Q8_1 (skip if already valid — PMAT-027)
-        if !self.q8_activation_valid {
-            self.q8_quantize_into(input, &q8_buf, k)?;
-            self.q8_activation_valid = true;
-        }
+        self.ensure_q8_activation(input, &q8_buf, k)?;
 
         // Step 2: Launch DP4A GEMV kernel
         let num_warps = self.gpu_profile.mwv_warps;
@@ -231,10 +228,7 @@ impl CudaExecutor {
         let q8_buf = unsafe { GpuBuffer::<u8>::from_raw_parts(q8_ptr, q8_len) };
 
         // Step 1: Quantize activations to Q8_1 (skip if already valid — PMAT-027)
-        if !self.q8_activation_valid {
-            self.q8_quantize_into(input, &q8_buf, k)?;
-            self.q8_activation_valid = true;
-        }
+        self.ensure_q8_activation(input, &q8_buf, k)?;
 
         // Step 2: Launch half-warp DP4A GEMV kernel
         let num_warps = self.gpu_profile.mwv_warps;
@@ -475,10 +469,7 @@ impl CudaExecutor {
         // SAFETY: constructs a non-owning `GpuBuffer` view over an already-allocated device region (`ptr`, element count `len`) that stays live for the kernel call; the view is `leak()`ed afterwards so its Drop never frees the borrowed device allocation (no double-free).
         let q8_buf = unsafe { GpuBuffer::<u8>::from_raw_parts(q8_ptr, q8_len) };
 
-        if !self.q8_activation_valid {
-            self.q8_quantize_into(input, &q8_buf, k)?;
-            self.q8_activation_valid = true;
-        }
+        self.ensure_q8_activation(input, &q8_buf, k)?;
 
         let num_warps = self.gpu_profile.mwv_warps;
         let kernel_type = KernelType::FusedGateUpSwigluHwDp4aQ4KGemv { k, n };
@@ -617,10 +608,7 @@ impl CudaExecutor {
         // SAFETY: constructs a non-owning `GpuBuffer` view over an already-allocated device region (`ptr`, element count `len`) that stays live for the kernel call; the view is `leak()`ed afterwards so its Drop never frees the borrowed device allocation (no double-free).
         let q8_buf = unsafe { GpuBuffer::<u8>::from_raw_parts(q8_ptr, q8_len) };
 
-        if !self.q8_activation_valid {
-            self.q8_quantize_into(input, &q8_buf, k)?;
-            self.q8_activation_valid = true;
-        }
+        self.ensure_q8_activation(input, &q8_buf, k)?;
 
         let num_warps = self.gpu_profile.mwv_warps;
         let kernel_type = KernelType::FusedKVHwDp4aQ4KGemv { k, n };
@@ -726,11 +714,7 @@ impl CudaExecutor {
         // Skip Q8 quantize when the same input buffer was already quantized
         // (e.g., K/V projections share input with Q, up shares with gate).
         // Saves 3 Q8 launches per layer × 28 layers = 84 launches per step.
-        if !self.q8_activation_valid {
-            let total_elements = m * k;
-            self.q8_quantize_into(input, &q8_buf, total_elements)?;
-            self.q8_activation_valid = true;
-        }
+        self.ensure_q8_activation(input, &q8_buf, m * k)?;
 
         // Step 2: Launch batched HW DP4A kernel
         let num_warps = self.gpu_profile.mwv_warps;

crates/aprender-serve/src/cuda/executor/residual.rs

@@ -309,6 +309,7 @@ impl CudaExecutor {
         output: &GpuBuffer<f32>,
         n: u32,
     ) -> Result<(), GpuError> {
+        self.q8_activation_written(output.as_ptr()); // #4258
         let kernel_type = KernelType::ResidualAdd { n };
         let kernel_name = self.kernels.kernel_name(&kernel_type);
         // GH-129: PTX is n-independent (n is a runtime param), so use constant cache key.

crates/aprender-serve/src/cuda/executor/silu.rs

@@ -235,6 +235,7 @@ impl CudaExecutor {
         output: &GpuBuffer<f32>,
         n: u32,
     ) -> Result<(), GpuError> {
+        self.q8_activation_written(output.as_ptr()); // #4258
         let kernel_type = KernelType::FusedSwiglu { n };
         let kernel_name = self.kernels.kernel_name(&kernel_type);
         let cache_key = format!("fused_swiglu_{}", n);

crates/aprender-serve/src/cuda/executor/tests_q8_activation_staleness.rs

@@ -0,0 +1,188 @@
+//! #4258: the DP4A GEMVs cache their Q8_1 activation behind `q8_activation_valid`. Before
+//! #4258 nothing on the Qwen3.5 path cleared that flag, so every DP4A GEMV after the first
+//! reused the FIRST quantized activation — across projections, layers and tokens.
+//!
+//! Each test runs a DP4A Q4_K GEMV whose input is NOT the activation last quantized, and
+//! judges it against the float (Mwv) GEMV of the same input. A stale activation reproduces
+//! the previous GEMV's output exactly; a fresh one lands within Q8_1 rounding of the float.
+
+use super::CudaExecutor;
+use crate::cuda::gpu_profile::Q4kVariant;
+use serial_test::serial;
+use trueno_gpu::driver::GpuBuffer;
+
+const N: u32 = 32;
+const K: u32 = 256;
+/// Q8_1 activation rounding on these inputs is well under 1% relative L2.
+const FRESH_TOL: f32 = 0.05;
+/// The two inputs are built so their float outputs sit far apart; below this the test
+/// could not tell a stale activation from a fresh one.
+const DISTINCT_FLOOR: f32 = 0.2;
+
+/// `N` rows of one Q4_K super-block each: d = 1.0, dmin = 0, every 6-bit scale 1,
+/// pseudo-random nibbles.
+fn q4k_weights() -> Vec<u8> {
+    let mut seed = 0x2545_f491_u32;
+    let mut out = Vec::with_capacity(N as usize * 144);
+    for _ in 0..N {
+        out.extend_from_slice(&[0x00, 0x3C, 0x00, 0x00]);
+        out.extend_from_slice(&[0x01; 12]);
+        for _ in 0..128 {
+            seed = seed.wrapping_mul(1_103_515_245).wrapping_add(12_345);
+            out.push((seed >> 16) as u8);
+        }
+    }
+    out
+}
+
+fn input_a() -> Vec<f32> {
+    (0..K).map(|i| (i as f32 * 0.37).sin()).collect()
+}
+
+fn input_b() -> Vec<f32> {
+    (0..K)
+        .map(|i| (i as f32 * 0.11).cos() * 2.0 - 0.5)
+        .collect()
+}
+
+fn rel_l2(got: &[f32], want: &[f32]) -> f32 {
+    let num: f32 = got.iter().zip(want).map(|(g, w)| (g - w).powi(2)).sum();
+    let den: f32 = want.iter().map(|w| w * w).sum();
+    (num / den).sqrt()
+}
+
+struct Rig {
+    ex: CudaExecutor,
+    w: u64,
+    out: GpuBuffer<f32>,
+}
+
+impl Rig {
+    fn new(mut ex: CudaExecutor) -> Self {
+        ex.init_workspace(K as usize, K as usize)
+            .expect("init_workspace");
+        ex.load_quantized_weights("q8_staleness_4258", &q4k_weights())
+            .expect("load Q4_K");
+        let w = ex
+            .get_quantized_weight_ptr("q8_staleness_4258")
+            .expect("weight ptr");
+        let out = GpuBuffer::new(ex.context(), N as usize).expect("out");
+        Self { ex, w, out }
+    }
+
+    fn buf(&self, host: &[f32]) -> GpuBuffer<f32> {
+        GpuBuffer::from_host(self.ex.context(), host).expect("buffer")
+    }
+
+    fn gemv(&mut self, variant: Q4kVariant, input: &GpuBuffer<f32>) -> Vec<f32> {
+        self.ex.gpu_profile.q4k = variant;
+        self.ex
+            .q4k_gemv_into(self.w, input, &self.out, N, K)
+            .expect("gemv");
+        self.ex.synchronize().expect("sync");
+        let mut host = vec![0.0f32; N as usize];
+        self.out.copy_to_host(&mut host).expect("copy out");
+        host
+    }
+
+    /// The float GEMV of whatever `input` holds right now.
+    fn float_ref(&mut self, input: &GpuBuffer<f32>) -> Vec<f32> {
+        self.ex.synchronize().expect("sync");
+        let mut host = vec![0.0f32; K as usize];
+        input.copy_to_host(&mut host).expect("copy in");
+        let fresh = self.buf(&host);
+        self.gemv(Q4kVariant::Mwv, &fresh)
+    }
+}
+
+const DP4A: [Q4kVariant; 2] = [Q4kVariant::HwDp4a, Q4kVariant::MwvDp4a];
+
+/// Two DP4A GEMVs on two DIFFERENT buffers: the second must quantize its own input.
+#[test]
+#[serial]
+fn a_dp4a_gemv_on_a_different_buffer_quantizes_that_buffer() {
+    let ex = crate::cuda_executor_or_skip!(0);
+    let mut rig = Rig::new(ex);
+    let (a, b) = (rig.buf(&input_a()), rig.buf(&input_b()));
+    let (ref_a, ref_b) = (rig.float_ref(&a), rig.float_ref(&b));
+    assert!(rel_l2(&ref_a, &ref_b) > DISTINCT_FLOOR, "inputs too alike");
+
+    for variant in DP4A {
+        let got_b = rig.gemv(variant, &b);
+        let got_a = rig.gemv(variant, &a);
+        let (err_b, err_a) = (rel_l2(&got_b, &ref_b), rel_l2(&got_a, &ref_a));
+        eprintln!("[4258] {variant:?}: first input {err_b:.3e}, second input {err_a:.3e}");
+        assert!(
+            err_b < FRESH_TOL,
+            "{variant:?}: the first DP4A GEMV is off ({err_b})"
+        );
+        assert!(
+            err_a < FRESH_TOL,
+            "{variant:?}: a DP4A GEMV on a second buffer is {err_a} from the float GEMV of \
+             that buffer — it reused the first buffer's Q8_1 activation (#4258)"
+        );
+    }
+}
+
+/// One buffer, rewritten in place by each Qwen3.5 decode primitive that feeds a GEMV: the
+/// DP4A GEMV after the write must quantize the NEW contents.
+#[test]
+#[serial]
+fn a_dp4a_gemv_after_a_qwen35_writer_rewrites_its_input_requantizes() {
+    let ex = crate::cuda_executor_or_skip!(0);
+    let mut rig = Rig::new(ex);
+    let (a_host, b_host) = (input_a(), input_b());
+    let (a, b) = (rig.buf(&a_host), rig.buf(&b_host));
+    let ones = rig.buf(&vec![1.0f32; K as usize]);
+
+    type Writer =
+        fn(&mut CudaExecutor, &GpuBuffer<f32>, &GpuBuffer<f32>, &GpuBuffer<f32>, &GpuBuffer<f32>);
+    let writers: [(&str, Writer); 5] = [
+        ("rmsnorm_into", |ex, _a, b, one, x| {
+            ex.rmsnorm_into(b, one, x, K, 1e-6).expect("rmsnorm_into");
+        }),
+        ("per_head_rmsnorm_into", |ex, _a, b, one, x| {
+            // `one` is K ones; per-head gamma reads only its first 64.
+            ex.per_head_rmsnorm_into(b, one, x, 64, K / 64, 1e-6)
+                .expect("per_head_rmsnorm_into");
+        }),
+        ("residual_add_into", |ex, a, b, _one, x| {
+            ex.residual_add_into(a, b, x, K).expect("residual_add_into");
+        }),
+        ("fused_swiglu_into", |ex, a, b, _one, x| {
+            ex.fused_swiglu_into(b, a, x, K).expect("fused_swiglu_into");
+        }),
+        ("gdn_sigmoid_gate_into", |ex, _a, b, _one, x| {
+            ex.gdn_sigmoid_gate_into(x, b, K)
+                .expect("gdn_sigmoid_gate_into");
+        }),
+    ];
+
+    // Every `x` stays alive for the whole test: a freed buffer's address is handed
+    // straight back to the next allocation, and a host upload into it is not a write
+    // the executor sees (see `ensure_q8_activation`).
+    let xs: Vec<GpuBuffer<f32>> = (0..DP4A.len() * writers.len())
+        .map(|_| rig.buf(&a_host))
+        .collect();
+    let mut xs = xs.iter();
+    for variant in DP4A {
+        for (name, write) in writers {
+            let x = xs.next().expect("one x per case");
+            let before = rig.gemv(variant, x);
+            write(&mut rig.ex, &a, &b, &ones, x);
+            let want = rig.float_ref(x);
+            assert!(
+                rel_l2(&before, &want) > DISTINCT_FLOOR,
+                "{name}: write changed too little"
+            );
+            let got = rig.gemv(variant, x);
+            let err = rel_l2(&got, &want);
+            eprintln!("[4258] {variant:?} after {name}: {err:.3e}");
+            assert!(
+                err < FRESH_TOL,
+                "{variant:?}: the DP4A GEMV after `{name}` rewrote its input is {err} from the \
+                 float GEMV of the new contents — it reused the stale Q8_1 activation (#4258)"
+            );
+        }
+    }
+}

crates/aprender-serve/src/cuda/executor/tiled_gemv.rs

@@ -333,6 +333,38 @@ impl CudaExecutor {
         Ok(buf_output)
     }
 
+    /// #4258: quantize the first `n` elements of `input` into `q8_buf` unless the
+    /// Q8 activation cache already holds exactly that. The cache is keyed on the
+    /// source buffer, so a GEMV over a different buffer quantizes its own input
+    /// instead of reusing the last one (PMAT-027 keyed only on a bool).
+    ///
+    /// The executor only sees writes made by its own kernels, which invalidate
+    /// through [`Self::q8_activation_written`]. A host upload into a buffer, or a
+    /// freed buffer whose address is reallocated, is invisible to it: the caller
+    /// must clear `q8_activation_valid` before a DP4A GEMV reads such a buffer.
+    pub(crate) fn ensure_q8_activation(
+        &mut self,
+        input: &GpuBuffer<f32>,
+        q8_buf: &GpuBuffer<u8>,
+        n: u32,
+    ) -> Result<(), GpuError> {
+        let src = (input.as_ptr(), n);
+        if !self.q8_activation_valid || self.q8_activation_src != src {
+            self.q8_quantize_into(input, q8_buf, n)?;
+            self.q8_activation_valid = true;
+            self.q8_activation_src = src;
+        }
+        Ok(())
+    }
+
+    /// #4258: a kernel is about to write device buffer `dst`; if that buffer is
+    /// the one the Q8 activation cache was quantized from, the cache is stale.
+    pub(crate) fn q8_activation_written(&mut self, dst: u64) {
+        if dst == self.q8_activation_src.0 {
+            self.q8_activation_valid = false;
+        }
+    }
+
     /// PAR-PERF-DP4A: Q8 quantize into PRE-ALLOCATED buffer (zero allocation)
     ///
     /// Five-Whys root cause (2026-02-09):