gitadmin/foxhunt
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

refactor(sp4): A13 code-quality pass — DRY helper, named constant, doc fix, param cleanup

Browse Source 
Addresses 5 IMPORTANT items from A13 code-quality review:

1. apply_pearls_to_slot helper extracted into sp4_wiener_ema.rs.
   Collapses ~30 lines of read_volatile / pearls_ad_update / write_volatile
   per-slot block into a single unsafe fn. 12 launchers now consume the
   helper (5 SP4 producers A5-A9, 6 A13 retrofits A13.0-A13.5, plus the
   inline label-scale block in aux_heads_forward Step 2b — including 2
   apply_pearls closures inside multi-slot launchers and the cross-boundary
   GpuExperienceCollector consumer). Pearl C rate_deficit site untouched
   (different mapped-pinned buffer + Rust ema array, doesn't share the
   helper's pointer-based contract).

2. REWARD_COMPONENT_COUNT named constant added next to
   REWARD_POPART_EMA_INDEX in gpu_dqn_trainer.rs. Replaces 3 hardcoded `6`
   literals across collector launcher (block_dim, Pearls A+D loop) and
   training_loop fold-reset range arithmetic. Mirrors MOE_NUM_EXPERTS /
   SL_NUM_FEATURE_GROUPS invariant-guard pattern with debug_assert_eq! in
   the collector launcher. Kernel literal `6` retained (allows nvcc full
   unroll); kernel comment now documents the host-side invariant.

3. SP4_PRODUCER_COUNT, SP4_WIENER_FLOATS_PER_SLOT, SP4_WIENER_TOTAL_FLOATS
   promoted from fn-local consts inside `pub fn new` to module-level
   `pub const`s in gpu_dqn_trainer.rs, re-exported via cuda_pipeline::mod.
   All 13 redeclarations in tests/sp4_producer_unit_tests.rs replaced
   with single `use ml::cuda_pipeline::SP4_PRODUCER_COUNT;` import. Future
   buffer growth requires single-file edit.

4. _ema_alpha_unused: f32 caller-compat shim removed from 6 retrofitted
   launchers (launch_h_s2_rms_ema, launch_aux_heads_loss_ema,
   launch_vsn_mask_ema, launch_moe_expert_util_ema, launch_iqn_quantile_ema,
   launch_reward_component_ema_inplace) and the FusedTrainingCtx proxy.
   All callers in training_loop.rs + fused_training.rs updated to drop
   the unused argument per feedback_no_legacy_aliases (no soft-deprecated
   wrappers; rename all call sites directly). Doc-comments updated from
   "α dropped per SP4 — argument preserved so callers compile unchanged"
   to "α derived adaptively from per-slot Pearls A+D Wiener state — see
   sp4_wiener_ema::pearls_ad_update".

5. Stale doc reference fixed at gpu_aux_heads.rs:391 — comment referenced
   non-existent launch_label_scale_ema_with_pearls function. Now correctly
   points at the inline Pearls A+D block in GpuDqnTrainer::aux_heads_forward
   Step 2b (which now consumes apply_pearls_to_slot).

Pure refactor — no spec or behaviour change. Same kernel launches, same
Pearls A+D semantics, same ISV slot writes.

cargo check -p ml --offline clean (12 pre-existing warnings, no new);
cargo test -p ml --lib --offline sp4_wiener_ema 7/7 passing
(6 originals + apply_pearls_to_slot_pearl_a_bootstrap_path);
cargo test -p ml --lib --offline state_reset_registry 3/3 passing.

Refs: A13 review (commits aada419de..c5add566d).

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
This commit is contained in:
jgrusewski
2026-05-01 08:42:40 +02:00
parent c5add566db
commit 4c231fa812
10 changed files with 362 additions and 353 deletions
Download Patch File Download Diff File Expand all files Collapse all files

9
crates/ml/src/cuda_pipeline/gpu_aux_heads.rs
View File

@@ -386,10 +386,11 @@ impl AuxHeadsForwardOps {
///
/// Consumes the dedicated label buffer that `aux_heads_forward` populated
/// via `strided_gather` of `next_states[:, 0]`. The kernel writes ONLY
/// the step observation; the host launcher
/// (`launch_label_scale_ema_with_pearls` on `GpuDqnTrainer`) syncs
/// the stream and applies Pearls A+D before any consumer reads
/// ISV[AUX_LABEL_SCALE_EMA_INDEX].
/// the step observation; the inline Pearls A+D update lives in
/// `GpuDqnTrainer::aux_heads_forward` Step 2b (immediately after this
/// kernel returns and before `next_bar_loss_reduce` runs), syncing the
/// stream and applying `apply_pearls_to_slot` so consumers see the
/// up-to-date `ISV[AUX_LABEL_SCALE_EMA_INDEX]` mid-step.
pub(crate) fn launch_label_scale_ema(
&self,
stream: &Arc<CudaStream>,

503
crates/ml/src/cuda_pipeline/gpu_dqn_trainer.rs
View File

@@ -707,8 +707,49 @@ pub const TLOB_REGIME_FOCUS_EMA_INDEX: usize = 60;
// ─── C.2 Reward-component attribution (Plan 3 Task 1, spec §4.C.2) ────────────
// Per-component mean |reward| EMA. Producer: reward_component_ema GPU kernel
// (per-step, single-block, 6 threads). Consumer: HEALTH_DIAG reward_split line.
// Component indices match reward_components_per_sample layout in experience_kernels.cu.
// (per-step, single-block, REWARD_COMPONENT_COUNT threads). Consumer:
// HEALTH_DIAG reward_split line. Component indices match
// reward_components_per_sample layout in experience_kernels.cu.
/// Number of reward components in the C.2 attribution layout.
///
/// Used by:
/// - `reward_component_ema_kernel.cu` block-dim and per-thread index gate
/// - `gpu_experience_collector::launch_reward_component_ema_inplace` block-dim
/// and Pearls A+D loop range
/// - `state_reset_registry::reset_named_state` ISV slot range
/// `[REWARD_POPART_EMA_INDEX, REWARD_POPART_EMA_INDEX + REWARD_COMPONENT_COUNT)`
///
/// Single source of truth — bumping this value requires the kernel block-dim,
/// the launcher loop, the SP4 scratch slot reservation `63..63+N`, and the
/// reset dispatch arm to all migrate together (per `feedback_no_partial_refactor.md`).
pub const REWARD_COMPONENT_COUNT: usize = 6;
/// Total number of SP4 producer scratch slots — sized to hold one f32 step
/// observation per producer kernel writing into `producer_step_scratch_buf`.
///
/// Layout (single source of truth — extending this requires the launchers,
/// the Wiener-state buffer, and the SP4 producer unit tests to co-migrate
/// per `feedback_no_partial_refactor.md`):
/// - 40 SP4 Layer A producers (Tasks A5-A9): TARGET_Q_BOUND, ATOM_POS×4,
/// WEIGHT/ADAM_M/ADAM_V/WD_RATE×8, L1_LAMBDA, GRAD_CLIP, H_S2 — slots 0..40
/// - 29 Task-A13 retrofit producers — slots 40..69
///
/// Co-located with `SP4_WIENER_FLOATS_PER_SLOT` and `SP4_WIENER_TOTAL_FLOATS`
/// so the relationship `wiener_total = count × floats_per_slot` is visible at
/// declaration time. Re-exported via `cuda_pipeline::SP4_PRODUCER_COUNT`
/// for consumption by `crates/ml/tests/sp4_producer_unit_tests.rs`.
pub const SP4_PRODUCER_COUNT: usize = 69;
/// Per-slot Wiener-state float count (`sample_var`, `diff_var`, `x_lag`).
/// Matches `WienerState`'s flat layout in `sp4_wiener_ema.rs`.
pub const SP4_WIENER_FLOATS_PER_SLOT: usize = 3;
/// Total floats in the SP4 Wiener-state buffer (one triple per producer slot).
/// Equals `SP4_PRODUCER_COUNT × SP4_WIENER_FLOATS_PER_SLOT = 207`.
pub const SP4_WIENER_TOTAL_FLOATS: usize =
SP4_PRODUCER_COUNT * SP4_WIENER_FLOATS_PER_SLOT;
/// ISV slot [63] — mean |popart reward| EMA (final on-policy reward, pre-PopArt).
pub const REWARD_POPART_EMA_INDEX: usize = 63;
/// ISV slot [64] — mean |counterfactual reward| EMA.
@@ -8776,22 +8817,20 @@ impl GpuDqnTrainer {
/// kernel). Reduces RMS = sqrt(sum_sq / (B*SH2)) over the trainer's
/// `save_h_s2 [B, SH2]` buffer and writes the step observation to
/// `producer_step_scratch_buf[SCRATCH_IDX=40]`. Synchronises the stream,
/// then applies Pearls A+D host-side via `pearls_ad_update` using
/// then applies Pearls A+D host-side via `apply_pearls_to_slot` using
/// zero-copy mapped-pinned reads of ISV[H_S2_RMS_EMA_INDEX=96] +
/// `wiener_state_buf[(40)*3..(40)*3+3]` — slot 40 in the SP4 producer
/// scratch / Wiener-state layout (first retrofit slot after the 40 SP4
/// producers).
///
/// **α dropped per SP4 — Pearls A+D adapt α from per-slot signal-vs-noise
/// variance.** The `_ema_alpha_unused` argument is preserved so callers
/// (training_loop.rs) compile unchanged; α is now derived adaptively per
/// step inside `pearls_ad_update`.
/// α is derived adaptively from per-slot Pearls A+D Wiener state — see
/// `sp4_wiener_ema::pearls_ad_update`.
///
/// Producer-only — 2c.3c.6 wires the consumer in `mag_concat_qdir`'s
/// adaptive-scale path. Cold-path-cadence: launched once per training step
/// alongside `launch_reward_component_ema`. No atomicAdd, no DtoH.
pub fn launch_h_s2_rms_ema(&self, _ema_alpha_unused: f32) -> Result<(), MLError> {
use crate::cuda_pipeline::sp4_wiener_ema::{pearls_ad_update, WienerState};
pub fn launch_h_s2_rms_ema(&self) -> Result<(), MLError> {
use crate::cuda_pipeline::sp4_wiener_ema::apply_pearls_to_slot;
// `isv_signals_dev_ptr` / `isv_signals_pinned` are set by the constructor
// via `cuMemHostGetDevicePointer_v2` on a `cuMemAllocHost_v2`-backed
@@ -8832,43 +8871,26 @@ impl GpuDqnTrainer {
.map_err(|e| MLError::ModelError(format!("h_s2_rms_ema sync: {e}")))?;
// ── Pearls A+D host-side update (zero-copy mapped-pinned reads) ──
let step_obs = unsafe {
std::ptr::read_volatile(self.producer_step_scratch_buf.host_ptr.add(SCRATCH_IDX))
};
// Degenerate-signal short-circuit: kernel writes 0.0 if every
// sample is exactly 0 (e.g., before the first online forward
// populates `save_h_s2`). Updating the Wiener state with 0 would
// advance `x_lag` and suppress the next genuine first-observation.
let step_obs = unsafe {
std::ptr::read_volatile(self.producer_step_scratch_buf.host_ptr.add(SCRATCH_IDX))
};
if step_obs == 0.0 { return Ok(()); }
let isv_idx = H_S2_RMS_EMA_INDEX;
// Wiener offset: SCRATCH_IDX * 3 (one Wiener triple per scratch slot).
// SCRATCH_IDX=40 → wiener_offset=120, well within 207 floats.
let wiener_offset = SCRATCH_IDX * 3;
// Safety: `isv_signals_pinned` is mapped-pinned `*mut f32` of length
// ISV_TOTAL_DIM (>= 171); H_S2_RMS_EMA_INDEX=96 < ISV_TOTAL_DIM.
// `wiener_state_buf.host_ptr` is mapped-pinned `*mut f32` of length
// `SP4_PRODUCER_COUNT * 3 = 207`; `wiener_offset + 2 < 207`.
let prev_x_mean = unsafe { std::ptr::read_volatile(self.isv_signals_pinned.add(isv_idx)) };
let mut state = unsafe {
let p = self.wiener_state_buf.host_ptr;
WienerState {
sample_var: std::ptr::read_volatile(p.add(wiener_offset)),
diff_var: std::ptr::read_volatile(p.add(wiener_offset + 1)),
x_lag: std::ptr::read_volatile(p.add(wiener_offset + 2)),
}
};
let new_x_mean = pearls_ad_update(prev_x_mean, &mut state, step_obs);
unsafe {
std::ptr::write_volatile(self.isv_signals_pinned.add(isv_idx), new_x_mean);
let wp = self.wiener_state_buf.host_ptr;
std::ptr::write_volatile(wp.add(wiener_offset), state.sample_var);
std::ptr::write_volatile(wp.add(wiener_offset + 1), state.diff_var);
std::ptr::write_volatile(wp.add(wiener_offset + 2), state.x_lag);
apply_pearls_to_slot(
self.producer_step_scratch_buf.host_ptr,
SCRATCH_IDX,
self.isv_signals_pinned,
H_S2_RMS_EMA_INDEX,
self.wiener_state_buf.host_ptr,
SCRATCH_IDX * 3,
);
}
Ok(())
}
@@ -8896,7 +8918,7 @@ impl GpuDqnTrainer {
/// commit; the ISV slot now holds an EMA but is not yet read.
pub fn launch_sp4_target_q_p99(&self) -> Result<(), MLError> {
use crate::cuda_pipeline::sp4_isv_slots::{TARGET_Q_BOUND_INDEX, SP4_SLOT_BASE};
use crate::cuda_pipeline::sp4_wiener_ema::{pearls_ad_update, WienerState};
use crate::cuda_pipeline::sp4_wiener_ema::apply_pearls_to_slot;
// Producer-step-scratch slot assignment. Stable layout — Tasks A6-A11
// extend with subsequent indices in producer-launch order:
@@ -8959,43 +8981,31 @@ impl GpuDqnTrainer {
// The kernel issued `__threadfence_system()` after writing
// `producer_step_scratch_buf[SCRATCH_IDX]`; the post-sync read here
// is the canonical mapped-pinned host-visible read pattern.
let step_obs = unsafe {
std::ptr::read_volatile(self.producer_step_scratch_buf.host_ptr.add(SCRATCH_IDX))
};
// Degenerate-signal short-circuit: kernel writes 0.0 when count==0
// or every sample is exactly 0 (e.g., before the first target-net
// forward populates `denoise_target_q_buf`). Pearl A's sentinel
// branch would still fire on a real 0 step_obs at fold reset, but
// updating the Wiener state with 0 advances `x_lag` and would
// suppress the next genuine first-observation. Simpler: skip.
let step_obs = unsafe {
std::ptr::read_volatile(self.producer_step_scratch_buf.host_ptr.add(SCRATCH_IDX))
};
if step_obs == 0.0 { return Ok(()); }
let isv_idx = TARGET_Q_BOUND_INDEX;
let wiener_offset = (isv_idx - SP4_SLOT_BASE) * 3;
// Safety: `isv_signals_pinned` is a mapped-pinned `*mut f32` of
// length `ISV_TOTAL_DIM`; `isv_idx < ISV_TOTAL_DIM` (171 > 131).
// `wiener_state_buf.host_ptr` is a mapped-pinned `*mut f32` of
// length `SP4_PRODUCER_COUNT * 3 = 207`; `wiener_offset + 2 < 207`.
let prev_x_mean = unsafe { std::ptr::read_volatile(self.isv_signals_pinned.add(isv_idx)) };
let mut state = unsafe {
let p = self.wiener_state_buf.host_ptr;
WienerState {
sample_var: std::ptr::read_volatile(p.add(wiener_offset)),
diff_var: std::ptr::read_volatile(p.add(wiener_offset + 1)),
x_lag: std::ptr::read_volatile(p.add(wiener_offset + 2)),
}
};
let new_x_mean = pearls_ad_update(prev_x_mean, &mut state, step_obs);
unsafe {
std::ptr::write_volatile(self.isv_signals_pinned.add(isv_idx), new_x_mean);
let wp = self.wiener_state_buf.host_ptr;
std::ptr::write_volatile(wp.add(wiener_offset), state.sample_var);
std::ptr::write_volatile(wp.add(wiener_offset + 1), state.diff_var);
std::ptr::write_volatile(wp.add(wiener_offset + 2), state.x_lag);
apply_pearls_to_slot(
self.producer_step_scratch_buf.host_ptr,
SCRATCH_IDX,
self.isv_signals_pinned,
isv_idx,
self.wiener_state_buf.host_ptr,
(isv_idx - SP4_SLOT_BASE) * 3,
);
}
Ok(())
}
@@ -9029,7 +9039,7 @@ impl GpuDqnTrainer {
use crate::cuda_pipeline::sp4_isv_slots::{
atom_pos_bound, ATOM_POS_BOUND_BASE, SP4_BRANCH_COUNT, SP4_SLOT_BASE,
};
use crate::cuda_pipeline::sp4_wiener_ema::{pearls_ad_update, WienerState};
use crate::cuda_pipeline::sp4_wiener_ema::apply_pearls_to_slot;
// Producer-step-scratch slot assignment (matches `launch_sp4_target_q_p99`'s
// documented stable layout): slots 1..1+SP4_BRANCH_COUNT == 1..5 are
@@ -9097,13 +9107,13 @@ impl GpuDqnTrainer {
// The kernel issued `__threadfence_system()` after each per-
// branch write; the post-sync read here is the canonical
// mapped-pinned host-visible read pattern.
let step_obs = unsafe {
std::ptr::read_volatile(self.producer_step_scratch_buf.host_ptr.add(scratch_idx))
};
// Same degenerate-zero short-circuit rationale as A5: skip the
// ISV/Wiener mutation when step_obs == 0 (e.g., before the first
// `recompute_atom_positions` populates the branch slice). Pearl A's
// sentinel branch fires on the next genuine first observation.
let step_obs = unsafe {
std::ptr::read_volatile(self.producer_step_scratch_buf.host_ptr.add(scratch_idx))
};
if step_obs == 0.0 { continue; }
let isv_idx = atom_pos_bound(branch);
@@ -9113,33 +9123,20 @@ impl GpuDqnTrainer {
// ATOM_POS_BOUND_BASE. The const-fn `atom_pos_bound` already
// checks `branch < SP4_BRANCH_COUNT` via debug_assert.
debug_assert_eq!(isv_idx, ATOM_POS_BOUND_BASE + branch);
let wiener_offset = (isv_idx - SP4_SLOT_BASE) * 3;
// Safety: `isv_signals_pinned` is a mapped-pinned `*mut f32` of
// length `ISV_TOTAL_DIM`; `isv_idx ∈ [132, 136) < 171`.
// `wiener_state_buf.host_ptr` is a mapped-pinned `*mut f32` of
// length `SP4_PRODUCER_COUNT * 3 = 207`; `wiener_offset + 2 < 207`.
let prev_x_mean = unsafe {
std::ptr::read_volatile(self.isv_signals_pinned.add(isv_idx))
};
let mut state = unsafe {
let p = self.wiener_state_buf.host_ptr;
WienerState {
sample_var: std::ptr::read_volatile(p.add(wiener_offset)),
diff_var: std::ptr::read_volatile(p.add(wiener_offset + 1)),
x_lag: std::ptr::read_volatile(p.add(wiener_offset + 2)),
}
};
let new_x_mean = pearls_ad_update(prev_x_mean, &mut state, step_obs);
unsafe {
std::ptr::write_volatile(self.isv_signals_pinned.add(isv_idx), new_x_mean);
let wp = self.wiener_state_buf.host_ptr;
std::ptr::write_volatile(wp.add(wiener_offset), state.sample_var);
std::ptr::write_volatile(wp.add(wiener_offset + 1), state.diff_var);
std::ptr::write_volatile(wp.add(wiener_offset + 2), state.x_lag);
apply_pearls_to_slot(
self.producer_step_scratch_buf.host_ptr,
scratch_idx,
self.isv_signals_pinned,
isv_idx,
self.wiener_state_buf.host_ptr,
(isv_idx - SP4_SLOT_BASE) * 3,
);
}
}
Ok(())
@@ -9197,7 +9194,7 @@ impl GpuDqnTrainer {
adam_m_bound, adam_v_bound, weight_bound, wd_rate, ParamGroup,
L1_LAMBDA_TRUNK_INDEX, SP4_PARAM_GROUP_COUNT, SP4_SLOT_BASE,
};
use crate::cuda_pipeline::sp4_wiener_ema::{pearls_ad_update, WienerState};
use crate::cuda_pipeline::sp4_wiener_ema::apply_pearls_to_slot;
// Producer-step-scratch slot assignment (matches the stable layout
// documented in `launch_sp4_target_q_p99` and extended for Task A7):
@@ -9361,45 +9358,34 @@ impl GpuDqnTrainer {
// ── Phase 2: Pearls A+D host-side update per output slot ──
// Helper closure: read step_obs from scratch slot, apply
// pearls_ad_update with the slot's Wiener state, write back.
// Captured-by-reference into closure to keep code DRY across the
// 4 (or 5 for group 0) outputs per group.
// `apply_pearls_to_slot` with the slot's Wiener state. Captured-by-
// reference into closure to keep code DRY across the 4 (or 5 for
// group 0) outputs per group.
//
// Safety: `isv_signals_pinned` is a mapped-pinned `*mut f32` of
// length `ISV_TOTAL_DIM`; SP4 ISV slots are bounded
// [SP4_SLOT_BASE, SP4_SLOT_END) = [131, 171).
// `wiener_state_buf.host_ptr` is mapped-pinned `*mut f32` of
// length `SP4_PRODUCER_COUNT * 3 = 207`; wiener_offset + 2 < 207.
let apply_pearls = |isv_idx: usize, scratch_idx: usize| {
let step_obs = unsafe {
std::ptr::read_volatile(self.producer_step_scratch_buf.host_ptr.add(scratch_idx))
};
// Same degenerate-zero short-circuit as A5/A6: skip the
// ISV/Wiener mutation when step_obs == 0 (e.g., before the
// first forward populates the buffer, or in the L1-lambda case
// when the trunk weight matrix collapses to a single feature
// and entropy_deficit hits the natural 0 of the formula).
let step_obs = unsafe {
std::ptr::read_volatile(self.producer_step_scratch_buf.host_ptr.add(scratch_idx))
};
if step_obs == 0.0 { return; }
let wiener_offset = (isv_idx - SP4_SLOT_BASE) * 3;
// Safety: `isv_signals_pinned` is a mapped-pinned `*mut f32` of
// length `ISV_TOTAL_DIM`; SP4 ISV slots are bounded
// [SP4_SLOT_BASE, SP4_SLOT_END) = [131, 171).
// `wiener_state_buf.host_ptr` is mapped-pinned `*mut f32` of
// length `SP4_PRODUCER_COUNT * 3 = 207`; wiener_offset + 2 < 207.
let prev_x_mean = unsafe {
std::ptr::read_volatile(self.isv_signals_pinned.add(isv_idx))
};
let mut state = unsafe {
let p = self.wiener_state_buf.host_ptr;
WienerState {
sample_var: std::ptr::read_volatile(p.add(wiener_offset)),
diff_var: std::ptr::read_volatile(p.add(wiener_offset + 1)),
x_lag: std::ptr::read_volatile(p.add(wiener_offset + 2)),
}
};
let new_x_mean = pearls_ad_update(prev_x_mean, &mut state, step_obs);
unsafe {
std::ptr::write_volatile(self.isv_signals_pinned.add(isv_idx), new_x_mean);
let wp = self.wiener_state_buf.host_ptr;
std::ptr::write_volatile(wp.add(wiener_offset), state.sample_var);
std::ptr::write_volatile(wp.add(wiener_offset + 1), state.diff_var);
std::ptr::write_volatile(wp.add(wiener_offset + 2), state.x_lag);
apply_pearls_to_slot(
self.producer_step_scratch_buf.host_ptr,
scratch_idx,
self.isv_signals_pinned,
isv_idx,
self.wiener_state_buf.host_ptr,
(isv_idx - SP4_SLOT_BASE) * 3,
);
}
};
@@ -9541,7 +9527,7 @@ impl GpuDqnTrainer {
/// consumer migration follows once all producers (A5-A9) land.
pub fn launch_sp4_grad_norm_p99(&self) -> Result<(), MLError> {
use crate::cuda_pipeline::sp4_isv_slots::{GRAD_CLIP_BOUND_INDEX, SP4_SLOT_BASE};
use crate::cuda_pipeline::sp4_wiener_ema::{pearls_ad_update, WienerState};
use crate::cuda_pipeline::sp4_wiener_ema::apply_pearls_to_slot;
// Producer-step-scratch slot 38 is reserved for GRAD_CLIP_BOUND per
// the stable layout documented on `launch_sp4_target_q_p99`:
@@ -9588,42 +9574,30 @@ impl GpuDqnTrainer {
.map_err(|e| MLError::ModelError(format!("grad_norm_p99_update sync: {e}")))?;
// ── Pearls A+D host-side update (zero-copy mapped-pinned reads) ──
let step_obs = unsafe {
std::ptr::read_volatile(self.producer_step_scratch_buf.host_ptr.add(SCRATCH_IDX))
};
// Degenerate-signal short-circuit: kernel writes 0.0 if grad_norm
// hasn't been populated yet (e.g., before first backward). Avoid
// advancing Wiener state with a zero observation, which would
// suppress the next genuine first-observation. Same semantics as
// Tasks A5/A6/A7.
let step_obs = unsafe {
std::ptr::read_volatile(self.producer_step_scratch_buf.host_ptr.add(SCRATCH_IDX))
};
if step_obs == 0.0 { return Ok(()); }
let isv_idx = GRAD_CLIP_BOUND_INDEX;
let wiener_offset = (isv_idx - SP4_SLOT_BASE) * 3;
// Safety: `isv_signals_pinned` is a mapped-pinned `*mut f32` of
// length `ISV_TOTAL_DIM`; `isv_idx < ISV_TOTAL_DIM` (171 > 168).
// `wiener_state_buf.host_ptr` is a mapped-pinned `*mut f32` of
// length `SP4_PRODUCER_COUNT * 3 = 207`; `wiener_offset + 2 < 207`.
let prev_x_mean = unsafe { std::ptr::read_volatile(self.isv_signals_pinned.add(isv_idx)) };
let mut state = unsafe {
let p = self.wiener_state_buf.host_ptr;
WienerState {
sample_var: std::ptr::read_volatile(p.add(wiener_offset)),
diff_var: std::ptr::read_volatile(p.add(wiener_offset + 1)),
x_lag: std::ptr::read_volatile(p.add(wiener_offset + 2)),
}
};
let new_x_mean = pearls_ad_update(prev_x_mean, &mut state, step_obs);
unsafe {
std::ptr::write_volatile(self.isv_signals_pinned.add(isv_idx), new_x_mean);
let wp = self.wiener_state_buf.host_ptr;
std::ptr::write_volatile(wp.add(wiener_offset), state.sample_var);
std::ptr::write_volatile(wp.add(wiener_offset + 1), state.diff_var);
std::ptr::write_volatile(wp.add(wiener_offset + 2), state.x_lag);
apply_pearls_to_slot(
self.producer_step_scratch_buf.host_ptr,
SCRATCH_IDX,
self.isv_signals_pinned,
isv_idx,
self.wiener_state_buf.host_ptr,
(isv_idx - SP4_SLOT_BASE) * 3,
);
}
Ok(())
}
@@ -9657,7 +9631,7 @@ impl GpuDqnTrainer {
/// before this commit.
pub fn launch_sp4_h_s2_p99(&self) -> Result<(), MLError> {
use crate::cuda_pipeline::sp4_isv_slots::{H_S2_BOUND_INDEX, SP4_SLOT_BASE};
use crate::cuda_pipeline::sp4_wiener_ema::{pearls_ad_update, WienerState};
use crate::cuda_pipeline::sp4_wiener_ema::apply_pearls_to_slot;
// Producer-step-scratch slot 39 is reserved for H_S2_BOUND per the
// stable layout documented on `launch_sp4_target_q_p99`.
@@ -9699,41 +9673,29 @@ impl GpuDqnTrainer {
.map_err(|e| MLError::ModelError(format!("h_s2_p99_update sync: {e}")))?;
// ── Pearls A+D host-side update (zero-copy mapped-pinned reads) ──
let step_obs = unsafe {
std::ptr::read_volatile(self.producer_step_scratch_buf.host_ptr.add(SCRATCH_IDX))
};
// Degenerate-signal short-circuit: kernel writes 0.0 if every
// sample is exactly 0 (e.g., before the first online forward
// populates `save_h_s2`). Updating the Wiener state with 0 would
// advance `x_lag` and suppress the next genuine first-observation.
let step_obs = unsafe {
std::ptr::read_volatile(self.producer_step_scratch_buf.host_ptr.add(SCRATCH_IDX))
};
if step_obs == 0.0 { return Ok(()); }
let isv_idx = H_S2_BOUND_INDEX;
let wiener_offset = (isv_idx - SP4_SLOT_BASE) * 3;
// Safety: `isv_signals_pinned` is a mapped-pinned `*mut f32` of
// length `ISV_TOTAL_DIM`; `isv_idx < ISV_TOTAL_DIM` (171 > 169).
// `wiener_state_buf.host_ptr` is a mapped-pinned `*mut f32` of
// length `SP4_PRODUCER_COUNT * 3 = 207`; `wiener_offset + 2 < 207`.
let prev_x_mean = unsafe { std::ptr::read_volatile(self.isv_signals_pinned.add(isv_idx)) };
let mut state = unsafe {
let p = self.wiener_state_buf.host_ptr;
WienerState {
sample_var: std::ptr::read_volatile(p.add(wiener_offset)),
diff_var: std::ptr::read_volatile(p.add(wiener_offset + 1)),
x_lag: std::ptr::read_volatile(p.add(wiener_offset + 2)),
}
};
let new_x_mean = pearls_ad_update(prev_x_mean, &mut state, step_obs);
unsafe {
std::ptr::write_volatile(self.isv_signals_pinned.add(isv_idx), new_x_mean);
let wp = self.wiener_state_buf.host_ptr;
std::ptr::write_volatile(wp.add(wiener_offset), state.sample_var);
std::ptr::write_volatile(wp.add(wiener_offset + 1), state.diff_var);
std::ptr::write_volatile(wp.add(wiener_offset + 2), state.x_lag);
apply_pearls_to_slot(
self.producer_step_scratch_buf.host_ptr,
SCRATCH_IDX,
self.isv_signals_pinned,
isv_idx,
self.wiener_state_buf.host_ptr,
(isv_idx - SP4_SLOT_BASE) * 3,
);
}
Ok(())
}
@@ -9804,15 +9766,14 @@ impl GpuDqnTrainer {
/// floats (matches `h_s2_rms_ema_kernel.cu`'s footprint exactly — the
/// kernel reuses one 256-float scratchpad across `num_groups` separate
/// reductions, see `vsn_mask_ema_kernel.cu` for the layout rationale).
pub fn launch_vsn_mask_ema(&self, _ema_alpha_unused: f32) -> Result<(), MLError> {
pub fn launch_vsn_mask_ema(&self) -> Result<(), MLError> {
// SP4 Layer A Task A13.3 retrofit (2026-05-01): kernel writes per-group
// step observation to `producer_step_scratch_buf[53..53+SL_NUM_FEATURE_GROUPS)`;
// host-side Pearls A+D maps to ISV[105..111).
//
// **α dropped per SP4 — Pearls A+D adapt α from per-slot signal-vs-noise
// variance.** The `_ema_alpha_unused` argument is preserved so callers
// (training_loop.rs) compile unchanged.
use crate::cuda_pipeline::sp4_wiener_ema::{pearls_ad_update, WienerState};
// α is derived adaptively from per-slot Pearls A+D Wiener state — see
// `sp4_wiener_ema::pearls_ad_update`.
use crate::cuda_pipeline::sp4_wiener_ema::apply_pearls_to_slot;
debug_assert!(self.isv_signals_dev_ptr != 0,
"launch_vsn_mask_ema: isv_signals_dev_ptr must be allocated by constructor");
@@ -9852,34 +9813,24 @@ impl GpuDqnTrainer {
for g in 0..num_groups {
let scratch_idx = SCRATCH_FIRST + g;
let isv_idx = VSN_MASK_GROUP_0_EMA_INDEX + g;
let step_obs = unsafe {
std::ptr::read_volatile(self.producer_step_scratch_buf.host_ptr.add(scratch_idx))
};
// VSN mask is a softmax output (rows sum to 1, all entries
// ∈ [0,1]). A degenerate 0 at this site means the kernel hasn't
// run (cold start before VSN forward populates `vsn_mask_buf`).
// Wiener state untouched in that case.
let step_obs = unsafe {
std::ptr::read_volatile(self.producer_step_scratch_buf.host_ptr.add(scratch_idx))
};
if step_obs == 0.0 { continue; }
let wiener_offset = scratch_idx * 3;
let prev_x_mean = unsafe {
std::ptr::read_volatile(self.isv_signals_pinned.add(isv_idx))
};
let mut state = unsafe {
let p = self.wiener_state_buf.host_ptr;
WienerState {
sample_var: std::ptr::read_volatile(p.add(wiener_offset)),
diff_var: std::ptr::read_volatile(p.add(wiener_offset + 1)),
x_lag: std::ptr::read_volatile(p.add(wiener_offset + 2)),
}
};
let new_x_mean = pearls_ad_update(prev_x_mean, &mut state, step_obs);
unsafe {
std::ptr::write_volatile(self.isv_signals_pinned.add(isv_idx), new_x_mean);
let wp = self.wiener_state_buf.host_ptr;
std::ptr::write_volatile(wp.add(wiener_offset), state.sample_var);
std::ptr::write_volatile(wp.add(wiener_offset + 1), state.diff_var);
std::ptr::write_volatile(wp.add(wiener_offset + 2), state.x_lag);
apply_pearls_to_slot(
self.producer_step_scratch_buf.host_ptr,
scratch_idx,
self.isv_signals_pinned,
isv_idx,
self.wiener_state_buf.host_ptr,
scratch_idx * 3,
);
}
}
Ok(())
@@ -9914,11 +9865,10 @@ impl GpuDqnTrainer {
/// (slots 41=next-bar, 42=regime). Wiener state at
/// `wiener_state_buf[123..129)` (= scratch slots 41..43 × 3).
///
/// **α dropped per SP4 — Pearls A+D adapt α from per-slot signal-vs-noise
/// variance.** The `_ema_alpha_unused` argument is preserved so callers
/// (training_loop.rs) compile unchanged.
pub fn launch_aux_heads_loss_ema(&self, _ema_alpha_unused: f32) -> Result<(), MLError> {
use crate::cuda_pipeline::sp4_wiener_ema::{pearls_ad_update, WienerState};
/// α is derived adaptively from per-slot Pearls A+D Wiener state — see
/// `sp4_wiener_ema::pearls_ad_update`.
pub fn launch_aux_heads_loss_ema(&self) -> Result<(), MLError> {
use crate::cuda_pipeline::sp4_wiener_ema::apply_pearls_to_slot;
debug_assert!(self.isv_signals_dev_ptr != 0,
"launch_aux_heads_loss_ema: isv_signals_dev_ptr must be allocated by constructor");
@@ -9943,34 +9893,24 @@ impl GpuDqnTrainer {
(SCRATCH_IDX_NEXT_BAR, AUX_NEXT_BAR_MSE_EMA_INDEX),
(SCRATCH_IDX_REGIME, AUX_REGIME_CE_EMA_INDEX),
] {
let step_obs = unsafe {
std::ptr::read_volatile(self.producer_step_scratch_buf.host_ptr.add(scratch_idx))
};
// Degenerate-zero short-circuit: skip if reduction yielded 0
// (e.g., before the first aux head forward populates the scalar
// buffers). Updating Wiener state with 0 would advance x_lag and
// suppress the next genuine first-observation.
let step_obs = unsafe {
std::ptr::read_volatile(self.producer_step_scratch_buf.host_ptr.add(scratch_idx))
};
if step_obs == 0.0 { continue; }
let wiener_offset = scratch_idx * 3;
let prev_x_mean = unsafe {
std::ptr::read_volatile(self.isv_signals_pinned.add(isv_idx))
};
let mut state = unsafe {
let p = self.wiener_state_buf.host_ptr;
WienerState {
sample_var: std::ptr::read_volatile(p.add(wiener_offset)),
diff_var: std::ptr::read_volatile(p.add(wiener_offset + 1)),
x_lag: std::ptr::read_volatile(p.add(wiener_offset + 2)),
}
};
let new_x_mean = pearls_ad_update(prev_x_mean, &mut state, step_obs);
unsafe {
std::ptr::write_volatile(self.isv_signals_pinned.add(isv_idx), new_x_mean);
let wp = self.wiener_state_buf.host_ptr;
std::ptr::write_volatile(wp.add(wiener_offset), state.sample_var);
std::ptr::write_volatile(wp.add(wiener_offset + 1), state.diff_var);
std::ptr::write_volatile(wp.add(wiener_offset + 2), state.x_lag);
apply_pearls_to_slot(
self.producer_step_scratch_buf.host_ptr,
scratch_idx,
self.isv_signals_pinned,
isv_idx,
self.wiener_state_buf.host_ptr,
scratch_idx * 3,
);
}
}
Ok(())
@@ -10190,11 +10130,10 @@ impl GpuDqnTrainer {
/// - ISV[MOE_GATE_ENTROPY_EMA_INDEX=126]
/// ← Wiener offset 52*3=156
///
/// **α dropped per SP4 — Pearls A+D adapt α from per-slot signal-vs-noise
/// variance.** The `_ema_alpha_unused` argument is preserved so callers
/// (training_loop.rs) compile unchanged.
pub fn launch_moe_expert_util_ema(&self, _ema_alpha_unused: f32) -> Result<(), MLError> {
use crate::cuda_pipeline::sp4_wiener_ema::{pearls_ad_update, WienerState};
/// α is derived adaptively from per-slot Pearls A+D Wiener state — see
/// `sp4_wiener_ema::pearls_ad_update`.
pub fn launch_moe_expert_util_ema(&self) -> Result<(), MLError> {
use crate::cuda_pipeline::sp4_wiener_ema::apply_pearls_to_slot;
debug_assert!(self.isv_signals_dev_ptr != 0,
"launch_moe_expert_util_ema: isv_signals_dev_ptr must be non-zero");
@@ -10219,34 +10158,23 @@ impl GpuDqnTrainer {
// ── Pearls A+D host-side for all 9 slots (8 experts + entropy) ──
let apply_pearls = |scratch_idx: usize, isv_idx: usize| {
let step_obs = unsafe {
std::ptr::read_volatile(self.producer_step_scratch_buf.host_ptr.add(scratch_idx))
};
// Skip degenerate-zero observations — gate softmax always sums
// to 1, so col_means are non-zero in steady state. A 0 here
// means the kernel hasn't run (cold start before forward
// populates moe_gate_softmax_buf) — Wiener state untouched.
let step_obs = unsafe {
std::ptr::read_volatile(self.producer_step_scratch_buf.host_ptr.add(scratch_idx))
};
if step_obs == 0.0 { return; }
let wiener_offset = scratch_idx * 3;
let prev_x_mean = unsafe {
std::ptr::read_volatile(self.isv_signals_pinned.add(isv_idx))
};
let mut state = unsafe {
let p = self.wiener_state_buf.host_ptr;
WienerState {
sample_var: std::ptr::read_volatile(p.add(wiener_offset)),
diff_var: std::ptr::read_volatile(p.add(wiener_offset + 1)),
x_lag: std::ptr::read_volatile(p.add(wiener_offset + 2)),
}
};
let new_x_mean = pearls_ad_update(prev_x_mean, &mut state, step_obs);
unsafe {
std::ptr::write_volatile(self.isv_signals_pinned.add(isv_idx), new_x_mean);
let wp = self.wiener_state_buf.host_ptr;
std::ptr::write_volatile(wp.add(wiener_offset), state.sample_var);
std::ptr::write_volatile(wp.add(wiener_offset + 1), state.diff_var);
std::ptr::write_volatile(wp.add(wiener_offset + 2), state.x_lag);
apply_pearls_to_slot(
self.producer_step_scratch_buf.host_ptr,
scratch_idx,
self.isv_signals_pinned,
isv_idx,
self.wiener_state_buf.host_ptr,
scratch_idx * 3,
);
}
};
@@ -10553,36 +10481,25 @@ impl GpuDqnTrainer {
self.stream.synchronize()
.map_err(|e| MLError::ModelError(format!("aux_label_scale_ema sync: {e}")))?;
{
use crate::cuda_pipeline::sp4_wiener_ema::{pearls_ad_update, WienerState};
use crate::cuda_pipeline::sp4_wiener_ema::apply_pearls_to_slot;
// Skip if the reduction yielded 0 — degenerate input
// (Wiener state untouched, sentinel preserved for first
// genuine non-zero observation).
let step_obs = unsafe {
std::ptr::read_volatile(
self.producer_step_scratch_buf.host_ptr.add(SCRATCH_IDX_LABEL_SCALE)
)
};
// Skip if the reduction yielded 0 — degenerate input
// (Wiener state untouched, sentinel preserved for first
// genuine non-zero observation).
if step_obs != 0.0 {
let isv_idx = AUX_LABEL_SCALE_EMA_INDEX;
let wiener_offset = SCRATCH_IDX_LABEL_SCALE * 3;
let prev_x_mean = unsafe {
std::ptr::read_volatile(self.isv_signals_pinned.add(isv_idx))
};
let mut state = unsafe {
let p = self.wiener_state_buf.host_ptr;
WienerState {
sample_var: std::ptr::read_volatile(p.add(wiener_offset)),
diff_var: std::ptr::read_volatile(p.add(wiener_offset + 1)),
x_lag: std::ptr::read_volatile(p.add(wiener_offset + 2)),
}
};
let new_x_mean = pearls_ad_update(prev_x_mean, &mut state, step_obs);
unsafe {
std::ptr::write_volatile(self.isv_signals_pinned.add(isv_idx), new_x_mean);
let wp = self.wiener_state_buf.host_ptr;
std::ptr::write_volatile(wp.add(wiener_offset), state.sample_var);
std::ptr::write_volatile(wp.add(wiener_offset + 1), state.diff_var);
std::ptr::write_volatile(wp.add(wiener_offset + 2), state.x_lag);
apply_pearls_to_slot(
self.producer_step_scratch_buf.host_ptr,
SCRATCH_IDX_LABEL_SCALE,
self.isv_signals_pinned,
AUX_LABEL_SCALE_EMA_INDEX,
self.wiener_state_buf.host_ptr,
SCRATCH_IDX_LABEL_SCALE * 3,
);
}
}
}
@@ -10830,7 +10747,6 @@ impl GpuDqnTrainer {
save_q_online_ptr: u64,
tba: usize,
num_quantiles: usize,
_ema_alpha_unused: f32,
) -> Result<(), MLError> {
// SP4 Layer A Task A13.4 retrofit (2026-05-01): kernel writes 4
// step observations (mean |Q| at p05/p25/p75/p95) to
@@ -10838,10 +10754,9 @@ impl GpuDqnTrainer {
// host-side, mapping each scratch slot to its ISV slot (99/100/
// 101/102) and updating Wiener-EMA state.
//
// **α dropped per SP4 — Pearls A+D adapt α from per-slot signal-
// vs-noise variance.** The `_ema_alpha_unused` argument is preserved
// so callers (training_loop.rs / fused_training.rs) compile unchanged.
use crate::cuda_pipeline::sp4_wiener_ema::{pearls_ad_update, WienerState};
// α is derived adaptively from per-slot Pearls A+D Wiener state — see
// `sp4_wiener_ema::pearls_ad_update`.
use crate::cuda_pipeline::sp4_wiener_ema::apply_pearls_to_slot;
if save_q_online_ptr == 0 {
return Ok(());
@@ -10891,34 +10806,24 @@ impl GpuDqnTrainer {
];
for &(off, isv_idx) in SLOT_PAIRS.iter() {
let scratch_idx = SCRATCH_FIRST + off;
let step_obs = unsafe {
std::ptr::read_volatile(self.producer_step_scratch_buf.host_ptr.add(scratch_idx))
};
// Skip degenerate-zero observations — IQN's per-quantile
// mean|Q| is non-zero in steady state. A 0 here means the
// kernel hasn't run (cold start before IQN forward populates
// `save_q_online`).
let step_obs = unsafe {
std::ptr::read_volatile(self.producer_step_scratch_buf.host_ptr.add(scratch_idx))
};
if step_obs == 0.0 { continue; }
let wiener_offset = scratch_idx * 3;
let prev_x_mean = unsafe {
std::ptr::read_volatile(self.isv_signals_pinned.add(isv_idx))
};
let mut state = unsafe {
let p = self.wiener_state_buf.host_ptr;
WienerState {
sample_var: std::ptr::read_volatile(p.add(wiener_offset)),
diff_var: std::ptr::read_volatile(p.add(wiener_offset + 1)),
x_lag: std::ptr::read_volatile(p.add(wiener_offset + 2)),
}
};
let new_x_mean = pearls_ad_update(prev_x_mean, &mut state, step_obs);
unsafe {
std::ptr::write_volatile(self.isv_signals_pinned.add(isv_idx), new_x_mean);
let wp = self.wiener_state_buf.host_ptr;
std::ptr::write_volatile(wp.add(wiener_offset), state.sample_var);
std::ptr::write_volatile(wp.add(wiener_offset + 1), state.diff_var);
std::ptr::write_volatile(wp.add(wiener_offset + 2), state.x_lag);
apply_pearls_to_slot(
self.producer_step_scratch_buf.host_ptr,
scratch_idx,
self.isv_signals_pinned,
isv_idx,
self.wiener_state_buf.host_ptr,
scratch_idx * 3,
);
}
}
Ok(())
@@ -13121,12 +13026,12 @@ impl GpuDqnTrainer {
// Per `feedback_no_htod_htoh_only_mapped_pinned`: mapped pinned
// (`cuMemHostAlloc(DEVICEMAP|PORTABLE)`) is the only allowed CPU↔GPU
// path for these state buffers.
const SP4_PRODUCER_COUNT: usize = 69; // 40 SP4 + 29 Task-A13 retrofit producers
const SP4_WIENER_FLOATS_PER_SLOT: usize = 3; // sample_var, diff_var, x_lag
const SP4_WIENER_TOTAL_FLOATS: usize = SP4_PRODUCER_COUNT * SP4_WIENER_FLOATS_PER_SLOT; // 207
//
// SP4_PRODUCER_COUNT / SP4_WIENER_FLOATS_PER_SLOT / SP4_WIENER_TOTAL_FLOATS
// are module-level `pub(crate) const`s declared at the top of this
// file so launch sites + tests share the single source of truth.
// SP4 Task A14: SP4_PARAM_GROUP_COUNT, MAX_BLOCKS_PER_ADAM, and
// SP4_ENGAGE_BUF_LEN are now public constants in `sp4_isv_slots`
// so launch sites + helpers can reference them without re-declaring.
// SP4_ENGAGE_BUF_LEN are public constants in `sp4_isv_slots`.
let wiener_state_buf = unsafe { MappedF32Buffer::new(SP4_WIENER_TOTAL_FLOATS) }
.map_err(|e| MLError::ModelError(format!("SP4 wiener_state_buf alloc: {e}")))?;

58
crates/ml/src/cuda_pipeline/gpu_experience_collector.rs
View File

@@ -2283,9 +2283,8 @@ impl GpuExperienceCollector {
/// the threaded mapped-pinned host pointers (no HtoD/DtoH per
/// `feedback_no_htod_htoh_only_mapped_pinned.md`).
///
/// **α dropped per SP4 — Pearls A+D adapt α from per-slot signal-
/// vs-noise variance.** The `_ema_alpha_unused` argument is preserved
/// so callers (training_loop.rs) compile unchanged.
/// α is derived adaptively from per-slot Pearls A+D Wiener state — see
/// `sp4_wiener_ema::pearls_ad_update`.
///
/// After the Pearls A+D update completes, resets
/// `reward_components_per_sample` to zeros via memset (same as other
@@ -2296,18 +2295,27 @@ impl GpuExperienceCollector {
pub fn launch_reward_component_ema_inplace(
&mut self,
n_samples: usize,
_ema_alpha_unused: f32,
) -> Result<(), crate::MLError> {
use crate::cuda_pipeline::gpu_dqn_trainer::REWARD_POPART_EMA_INDEX;
use crate::cuda_pipeline::sp4_wiener_ema::{pearls_ad_update, WienerState};
use crate::cuda_pipeline::gpu_dqn_trainer::{
REWARD_COMPONENT_COUNT, REWARD_POPART_EMA_INDEX,
};
use crate::cuda_pipeline::sp4_wiener_ema::apply_pearls_to_slot;
if n_samples == 0 || self.isv_signals_dev_ptr == 0 {
return Ok(());
}
// SP4 Task A13.5: scratch slot 63 — first of 6 contiguous reward-
// component slots. Wiener offsets: (63+c)*3 = 189..207 for c in 0..6.
// SP4 Task A13.5: scratch slot 63 — first of REWARD_COMPONENT_COUNT
// contiguous reward-component slots. Wiener offsets:
// (63+c)*3 for c in 0..REWARD_COMPONENT_COUNT.
const SCRATCH_FIRST: usize = 63;
// Layout invariant: bumping REWARD_COMPONENT_COUNT requires the
// kernel block_dim and the reward_components_per_sample row stride
// to migrate together (see kernel comment in
// `reward_component_ema_kernel.cu`).
debug_assert_eq!(REWARD_COMPONENT_COUNT, 6,
"reward-component layout invariant — kernel block_dim and \
reward_components_per_sample stride must update together");
// The trainer's Pearls A+D buffers MUST be wired before the first
// launch. `FusedTrainingCtx::wire_reward_component_pearls_buffers`
@@ -2332,7 +2340,11 @@ impl GpuExperienceCollector {
.arg(&n_i32)
.arg(&scratch_dev)
.arg(&scratch_first_i32)
.launch(LaunchConfig { grid_dim: (1, 1, 1), block_dim: (6, 1, 1), shared_mem_bytes: 0 })
.launch(LaunchConfig {
grid_dim: (1, 1, 1),
block_dim: (REWARD_COMPONENT_COUNT as u32, 1, 1),
shared_mem_bytes: 0,
})
.map_err(|e| crate::MLError::ModelError(format!("reward_component_ema: {e}")))?;
}
// Cold-path producer — sync before reading the mapped-pinned scratch
@@ -2340,36 +2352,30 @@ impl GpuExperienceCollector {
self.stream.synchronize()
.map_err(|e| crate::MLError::ModelError(format!("reward_component_ema sync: {e}")))?;
// ── Pearls A+D host-side update for all 6 component slots ──
// ── Pearls A+D host-side update for all REWARD_COMPONENT_COUNT slots ──
let scratch_host = self.reward_component_pearls_scratch_host_ptr;
let wiener_host = self.reward_component_pearls_wiener_host_ptr;
let isv_host = self.reward_component_pearls_isv_pinned_ptr;
for c in 0..6 {
for c in 0..REWARD_COMPONENT_COUNT {
let scratch_idx = SCRATCH_FIRST + c;
let isv_idx = REWARD_POPART_EMA_INDEX + c;
let step_obs = unsafe { std::ptr::read_volatile(scratch_host.add(scratch_idx)) };
// Reward components 2/5 (trail/bonus) are always 0.0 placeholders
// in this design — they should not advance Wiener state. The
// generic short-circuit also covers cold-start before the first
// collect_experiences_gpu populates `reward_components_per_sample`.
let step_obs = unsafe { std::ptr::read_volatile(scratch_host.add(scratch_idx)) };
if step_obs == 0.0 { continue; }
let wiener_offset = scratch_idx * 3;
let prev_x_mean = unsafe { std::ptr::read_volatile(isv_host.add(isv_idx)) };
let mut state = unsafe {
WienerState {
sample_var: std::ptr::read_volatile(wiener_host.add(wiener_offset)),
diff_var: std::ptr::read_volatile(wiener_host.add(wiener_offset + 1)),
x_lag: std::ptr::read_volatile(wiener_host.add(wiener_offset + 2)),
}
};
let new_x_mean = pearls_ad_update(prev_x_mean, &mut state, step_obs);
unsafe {
std::ptr::write_volatile(isv_host.add(isv_idx), new_x_mean);
std::ptr::write_volatile(wiener_host.add(wiener_offset), state.sample_var);
std::ptr::write_volatile(wiener_host.add(wiener_offset + 1), state.diff_var);
std::ptr::write_volatile(wiener_host.add(wiener_offset + 2), state.x_lag);
apply_pearls_to_slot(
scratch_host,
scratch_idx,
isv_host,
isv_idx,
wiener_host,
scratch_idx * 3,
);
}
}

8
crates/ml/src/cuda_pipeline/mod.rs
View File

@@ -75,9 +75,15 @@ pub use sp4_isv_slots::{
};
pub mod sp4_wiener_ema;
pub use sp4_wiener_ema::{
pearls_ad_update, WienerState,
apply_pearls_to_slot, pearls_ad_update, WienerState,
ALPHA_META, EPS_DIV, EPS_CLAMP_FLOOR,
};
// SP4 producer-step-scratch buffer constants — single source of truth for
// launch sites + tests. `pub(crate)` in `gpu_dqn_trainer.rs` so the
// re-export through this module gates external visibility (ml::cuda_pipeline::*).
pub use gpu_dqn_trainer::{
SP4_PRODUCER_COUNT, SP4_WIENER_FLOATS_PER_SLOT, SP4_WIENER_TOTAL_FLOATS,
};
// gpu_replay_buffer moved to ml-dqn crate
/// Maximum bytes allowed for a single GPU upload (2 GB safety limit).

6
crates/ml/src/cuda_pipeline/reward_component_ema_kernel.cu
View File

@@ -38,6 +38,12 @@ extern "C" __global__ void reward_component_ema(
float* __restrict__ scratch_buf, /* producer_step_scratch_buf */
int scratch_first_index /* slot 63 — first of 6 contiguous slots */
) {
/* Per-thread one component (REWARD_COMPONENT_COUNT=6 in production layout —
* see `REWARD_COMPONENT_COUNT` in `gpu_dqn_trainer.rs`). The host launcher
* sets `block_dim = (REWARD_COMPONENT_COUNT, 1, 1)`; the literal 6 below
* is intentional kernel-side — keeping it as a compile-time literal allows
* `nvcc` to fully unroll the row-stride arithmetic. The host-side
* `debug_assert_eq!(REWARD_COMPONENT_COUNT, 6, ...)` guards the contract. */
int c = (int)threadIdx.x;
if (c >= 6 || blockIdx.x != 0) return;

94
crates/ml/src/cuda_pipeline/sp4_wiener_ema.rs
View File

@@ -94,6 +94,61 @@ pub fn pearls_ad_update(
new_x_mean
}
/// Apply Pearls A+D for one ISV slot using mapped-pinned host pointers.
///
/// Reads `step_observation` from `scratch_host[scratch_idx]`, reads previous
/// `prev_x_mean` from `isv_pinned[isv_idx]`, reads/updates the per-slot
/// `WienerState` at `wiener_host[wiener_offset .. wiener_offset+3]`. Writes
/// the new ISV value back to `isv_pinned[isv_idx]` and the updated Wiener
/// state back to `wiener_host`. All accesses use `read_volatile` /
/// `write_volatile` on raw mapped-pinned pointers — caller is responsible
/// for ensuring the pointers are non-null and the GPU has finished writing
/// `scratch_host[scratch_idx]` (typically via `stream.synchronize()`).
///
/// The companion degenerate-zero short-circuit is kept at the call site so
/// each consumer can document the per-signal semantic of the zero
/// observation (e.g., "kernel hasn't run yet" vs "always-zero placeholder
/// component"); the helper itself unconditionally applies Pearls A+D.
///
/// # Safety
///
/// All three pointers MUST be non-null mapped-pinned pointers from
/// `cuMemHostAlloc(... DEVICEMAP)`. Indices MUST be within their respective
/// buffer bounds — `scratch_idx` in `[0, SP4_PRODUCER_COUNT)`, `isv_idx`
/// in `[0, ISV_TOTAL_DIM)`, `wiener_offset + 2` in
/// `[0, SP4_PRODUCER_COUNT * 3)`. Caller MUST hold a synchronisation
/// barrier between the kernel that wrote `scratch_host[scratch_idx]` and
/// this call.
#[inline]
pub unsafe fn apply_pearls_to_slot(
scratch_host: *const f32,
scratch_idx: usize,
isv_pinned: *mut f32,
isv_idx: usize,
wiener_host: *mut f32,
wiener_offset: usize,
) {
debug_assert!(!scratch_host.is_null(),
"apply_pearls_to_slot: scratch_host must be non-null");
debug_assert!(!isv_pinned.is_null(),
"apply_pearls_to_slot: isv_pinned must be non-null");
debug_assert!(!wiener_host.is_null(),
"apply_pearls_to_slot: wiener_host must be non-null");
let step_obs = std::ptr::read_volatile(scratch_host.add(scratch_idx));
let prev_x_mean = std::ptr::read_volatile(isv_pinned.add(isv_idx));
let mut state = WienerState {
sample_var: std::ptr::read_volatile(wiener_host.add(wiener_offset)),
diff_var: std::ptr::read_volatile(wiener_host.add(wiener_offset + 1)),
x_lag: std::ptr::read_volatile(wiener_host.add(wiener_offset + 2)),
};
let new_x_mean = pearls_ad_update(prev_x_mean, &mut state, step_obs);
std::ptr::write_volatile(isv_pinned.add(isv_idx), new_x_mean);
std::ptr::write_volatile(wiener_host.add(wiener_offset), state.sample_var);
std::ptr::write_volatile(wiener_host.add(wiener_offset + 1), state.diff_var);
std::ptr::write_volatile(wiener_host.add(wiener_offset + 2), state.x_lag);
}
#[cfg(test)]
mod tests {
use super::*;
@@ -167,6 +222,45 @@ mod tests {
assert_eq!(EPS_CLAMP_FLOOR, 1.0);
}
#[test]
fn apply_pearls_to_slot_pearl_a_bootstrap_path() {
// Smoke test: apply_pearls_to_slot using heap-allocated arrays
// (stand-ins for mapped-pinned buffers — same `*mut f32`/`*const f32`
// pointer ABI). Verifies the helper reads/writes through the raw
// pointers and that Pearl A's sentinel branch fires when both
// prev_x_mean and x_lag are zero.
let mut scratch = [0.0_f32; 16];
let mut isv = [0.0_f32; 16];
let mut wiener = [0.0_f32; 48]; // 16 slots × 3 floats
scratch[5] = 3.14;
// prev isv[7] = 0.0 (sentinel)
// wiener[9..12] = 0.0 (sentinel)
unsafe {
apply_pearls_to_slot(
scratch.as_ptr(),
5,
isv.as_mut_ptr(),
7,
wiener.as_mut_ptr(),
9,
);
}
assert_eq!(isv[7], 3.14,
"Pearl A: first observation written directly to isv slot");
assert_eq!(wiener[9 + 2], 3.14,
"Pearl A: x_lag seeded with first observation");
assert_eq!(wiener[9], 0.0, "sample_var unchanged at Pearl A bootstrap");
assert_eq!(wiener[9 + 1], 0.0, "diff_var unchanged at Pearl A bootstrap");
// Other slots untouched.
assert_eq!(isv[0], 0.0);
assert_eq!(isv[15], 0.0);
assert_eq!(scratch[0], 0.0);
}
#[test]
fn pearl_a_only_fires_when_both_x_mean_and_x_lag_are_zero() {
// If prev_x_mean is 0 but x_lag is non-zero (e.g., a previous Pearl A

6
crates/ml/src/trainers/dqn/fused_training.rs
View File

@@ -3073,8 +3073,9 @@ impl FusedTrainingCtx {
/// Plan 4 Task 3 (E.3): launch the IQN multi-quantile diagnostic EMA
/// kernel. Reads `gpu_iqn.save_q_online` and EMAs the four off-median
/// fixed quantiles into ISV[99..103). No-op when IQN is inactive (no
/// quantile surface to read).
pub(crate) fn launch_iqn_quantile_ema(&self, ema_alpha: f32) -> Result<(), crate::MLError> {
/// quantile surface to read). α is derived adaptively from per-slot
/// Pearls A+D Wiener state — see `sp4_wiener_ema::pearls_ad_update`.
pub(crate) fn launch_iqn_quantile_ema(&self) -> Result<(), crate::MLError> {
let iqn = match self.gpu_iqn.as_ref() {
Some(iqn) => iqn,
None => return Ok(()),
@@ -3083,7 +3084,6 @@ impl FusedTrainingCtx {
iqn.save_q_online_ptr(),
iqn.tba(),
iqn.num_quantiles(),
ema_alpha,
)
}

15
crates/ml/src/trainers/dqn/trainer/training_loop.rs
View File

@@ -3401,7 +3401,7 @@ impl DQNTrainer {
}).unwrap_or(0);
let ema_alpha = 0.05_f32;
if let Some(ref mut c) = self.gpu_experience_collector {
if let Err(e) = c.launch_reward_component_ema_inplace(n_main, ema_alpha) {
if let Err(e) = c.launch_reward_component_ema_inplace(n_main) {
tracing::warn!("C.2 reward_component_ema launch failed: {e}");
}
if let Err(e) = c.launch_trade_attempt_rate_ema_inplace(n_main, ema_alpha) {
@@ -3445,7 +3445,7 @@ impl DQNTrainer {
// `mag_concat_qdir`'s adaptive-scale path. Same launch
// cadence as the Plan 3 producers above.
if let Some(ref fused) = self.fused_ctx {
if let Err(e) = fused.trainer().launch_h_s2_rms_ema(ema_alpha) {
if let Err(e) = fused.trainer().launch_h_s2_rms_ema() {
tracing::warn!("Plan 4 2c.3c.5 h_s2_rms_ema launch failed: {e}");
}
}
@@ -3473,7 +3473,7 @@ impl DQNTrainer {
// skipped — already in the greedy-Q diagnostic. Same launch
// cadence as `h_s2_rms_ema` above. No-op when IQN is inactive.
if let Some(ref fused) = self.fused_ctx {
if let Err(e) = fused.launch_iqn_quantile_ema(ema_alpha) {
if let Err(e) = fused.launch_iqn_quantile_ema() {
tracing::warn!("Plan 4 Task 3 iqn_quantile_ema launch failed: {e}");
}
}
@@ -3488,7 +3488,7 @@ impl DQNTrainer {
// focus monitoring; no consumer kernel reads slots [105..111).
// Same launch cadence as `h_s2_rms_ema` / `iqn_quantile_ema`.
if let Some(ref fused) = self.fused_ctx {
if let Err(e) = fused.trainer().launch_vsn_mask_ema(ema_alpha) {
if let Err(e) = fused.trainer().launch_vsn_mask_ema() {
tracing::warn!("Plan 4 1B-iii vsn_mask_ema launch failed: {e}");
}
}
@@ -3504,7 +3504,7 @@ impl DQNTrainer {
// a5f23b28f): ISV producer launches MUST run BEFORE the
// HEALTH_DIAG line emit further down in this loop body.
if let Some(ref fused) = self.fused_ctx {
if let Err(e) = fused.trainer().launch_aux_heads_loss_ema(ema_alpha) {
if let Err(e) = fused.trainer().launch_aux_heads_loss_ema() {
tracing::warn!("Plan 4 Task 6 Commit B aux_heads_loss_ema launch failed: {e}");
}
}
@@ -3513,7 +3513,7 @@ impl DQNTrainer {
// Reads `moe_gate_softmax_buf [B, K]` (valid after the captured
// forward graph ran) and EMA-updates ISV[118..127).
if let Some(ref fused) = self.fused_ctx {
if let Err(e) = fused.trainer().launch_moe_expert_util_ema(ema_alpha) {
if let Err(e) = fused.trainer().launch_moe_expert_util_ema() {
tracing::warn!("Phase 3 T3.5 launch_moe_expert_util_ema failed: {e}");
}
}
@@ -5472,7 +5472,8 @@ impl DQNTrainer {
"isv_reward_component_emas" => {
if let Some(ref fused) = self.fused_ctx {
for idx in crate::cuda_pipeline::gpu_dqn_trainer::REWARD_POPART_EMA_INDEX
..crate::cuda_pipeline::gpu_dqn_trainer::REWARD_POPART_EMA_INDEX + 6
..crate::cuda_pipeline::gpu_dqn_trainer::REWARD_POPART_EMA_INDEX
+ crate::cuda_pipeline::gpu_dqn_trainer::REWARD_COMPONENT_COUNT
{
fused.trainer().write_isv_signal_at(idx, 0.0);
}

14
crates/ml/tests/sp4_producer_unit_tests.rs
View File

@@ -23,6 +23,7 @@ use std::sync::Arc;
use cudarc::driver::{CudaContext, CudaFunction, CudaStream, LaunchConfig, PushKernelArg};
use ml::cuda_pipeline::mapped_pinned::{MappedF32Buffer, MappedI32Buffer, MappedU64Buffer};
use ml::cuda_pipeline::SP4_PRODUCER_COUNT;
/// Test-only cubin for the SP4 histogram-p99 wrapper kernel. Built by
/// `crates/ml/build.rs`; the cubin path is the standard `OUT_DIR` slot
@@ -324,7 +325,6 @@ fn sp4_target_q_p99_first_observation_writes_step_p99_via_pearls_ad() {
// production-allocated TARGET_Q_BOUND slot; we use the same index so
// the kernel-direct test keeps the layout invariant the production
// launcher depends on.
const SP4_PRODUCER_COUNT: usize = 69;
const SCRATCH_IDX_TARGET_Q: usize = 0;
// ── Kernel ──
@@ -438,7 +438,6 @@ fn sp4_atom_pos_p99_per_branch_writes_distinct_isv_slots() {
// Same scratch shape as `launch_sp4_atom_pos_p99_all_branches` writes
// into in production. SCRATCH_BASE=1 mirrors the launcher's documented
// slot assignment (slot 0 = TARGET_Q_BOUND, slots 1..5 = ATOM_POS_BOUND[0..4]).
const SP4_PRODUCER_COUNT: usize = 69;
const SCRATCH_BASE: usize = 1;
const SHARED_BYTES: u32 = (256 / 32) * 256 * 4;
const N_PER_BRANCH: usize = 4096;
@@ -652,7 +651,6 @@ fn launch_sp4_param_group_oracle_for_group(
k_in: i32,
h_dim: i32,
) -> Vec<f32> {
const SP4_PRODUCER_COUNT: usize = 69;
const SCRATCH_BASE_W: usize = 5;
const SCRATCH_BASE_M: usize = 13;
const SCRATCH_BASE_V: usize = 21;
@@ -802,7 +800,6 @@ fn sp4_param_group_oracle_per_group_writes_distinct_isv_slots() {
SP4_PARAM_GROUP_COUNT, SP4_SLOT_BASE, WD_RATE_BASE, WEIGHT_BOUND_BASE,
};
const SP4_PRODUCER_COUNT: usize = 69;
const SCRATCH_BASE_W: usize = 5;
const SCRATCH_BASE_M: usize = 13;
const SCRATCH_BASE_V: usize = 21;
@@ -1145,7 +1142,6 @@ fn sp4_grad_norm_p99_pearl_a_first_observation_replaces_scalar() {
// Production scratch shape (47 = SP4 producers count). Slot 38 is the
// production-allocated GRAD_CLIP_BOUND slot (per the documented stable
// layout in `launch_sp4_target_q_p99` / `launch_sp4_grad_norm_p99`).
const SP4_PRODUCER_COUNT: usize = 69;
const SCRATCH_IDX_GRAD_CLIP: usize = 38;
// Single-element grad_norm scalar — known synthetic value chosen to be
@@ -1312,7 +1308,6 @@ fn sp4_h_s2_p99_writes_step_p99_to_scratch_via_pearl_a() {
// Production scratch shape (47 = SP4 producers count). Slot 39 is the
// production-allocated H_S2_BOUND slot per the documented stable layout
// in `launch_sp4_target_q_p99` / `launch_sp4_h_s2_p99`.
const SP4_PRODUCER_COUNT: usize = 69;
const SCRATCH_IDX_H_S2: usize = 39;
const SHARED_BYTES: u32 = (256 / 32) * 256 * 4;
@@ -1435,7 +1430,6 @@ fn sp4_h_s2_rms_ema_writes_step_rms_via_pearl_a_then_converges_pearl_d() {
const SH2: usize = 64;
const N: usize = B * SH2;
const SCRATCH_IDX: usize = 40;
const SP4_PRODUCER_COUNT: usize = 69;
const BLOCK_DIM: u32 = 256;
const SHARED_BYTES: u32 = BLOCK_DIM * std::mem::size_of::<f32>() as u32;
@@ -1554,7 +1548,6 @@ fn sp4_aux_heads_loss_ema_writes_step_obs_via_pearl_a_then_converges_pearl_d() {
const SCRATCH_IDX_NEXT_BAR: usize = 41;
const SCRATCH_IDX_REGIME: usize = 42;
const SP4_PRODUCER_COUNT: usize = 69;
let nb_loss: f32 = 0.5;
let rg_loss: f32 = 1.2;
@@ -1632,7 +1625,6 @@ fn sp4_aux_label_scale_ema_writes_step_obs_via_pearl_a_then_converges_pearl_d()
use ml::cuda_pipeline::sp4_wiener_ema::{pearls_ad_update, WienerState};
const SCRATCH_IDX: usize = 43;
const SP4_PRODUCER_COUNT: usize = 69;
const B: usize = 256;
const BLOCK_DIM: u32 = 256;
const SHARED_BYTES: u32 = BLOCK_DIM * std::mem::size_of::<f32>() as u32;
@@ -1727,7 +1719,6 @@ fn sp4_moe_expert_util_ema_writes_step_obs_via_pearl_a_then_converges_pearl_d()
const K: usize = 8;
const SCRATCH_IDX_UTIL_BASE: usize = 44;
const SCRATCH_IDX_ENTROPY: usize = 52;
const SP4_PRODUCER_COUNT: usize = 69;
// Uniform gate: each row sums to 1, all entries equal 1/K.
let uniform: f32 = 1.0 / K as f32;
@@ -1839,7 +1830,6 @@ fn sp4_vsn_mask_ema_writes_step_obs_via_pearl_a_then_converges_pearl_d() {
const B: usize = 128;
const NUM_GROUPS: usize = 6;
const SCRATCH_FIRST: usize = 53;
const SP4_PRODUCER_COUNT: usize = 69;
const BLOCK_DIM: u32 = 256;
const SHARED_BYTES: u32 = BLOCK_DIM * std::mem::size_of::<f32>() as u32;
@@ -1947,7 +1937,6 @@ fn sp4_iqn_quantile_ema_writes_step_obs_via_pearl_a_then_converges_pearl_d() {
const Q: usize = 5;
const TBA: usize = 12;
const SCRATCH_FIRST: usize = 59;
const SP4_PRODUCER_COUNT: usize = 69;
const BLOCK_DIM: u32 = 256;
const SHARED_BYTES: u32 = BLOCK_DIM * std::mem::size_of::<f32>() as u32;
@@ -2070,7 +2059,6 @@ fn sp4_reward_component_ema_writes_step_obs_via_pearl_a_then_converges_pearl_d()
const N: usize = 128;
const NUM_COMPONENTS: usize = 6;
const SCRATCH_FIRST: usize = 63;
const SP4_PRODUCER_COUNT: usize = 69;
// reward_components[i*6 + c] = (c+1) with sign flip on odd i, so
// mean(|r_c|) = c+1 ∈ {1, 2, 3, 4, 5, 6}.

2
docs/dqn-wire-up-audit.md
View File

@@ -2337,4 +2337,6 @@ SP4 Layer A Task A13.4 — iqn_quantile_ema retrofit (Pearls A+D) (2026-05-01):
SP4 Layer A Task A13 follow-up — fold-reset sentinel contract restored (2026-05-01): A13.0..A13.5 zeroed the Wiener triples at fold boundary via the bulk `sp4_wiener_state` reset but left the companion ISV-slot dispatch arms in `state_reset_registry::reset_named_state` writing pre-SP4 cold-start values (1.0, 1/6, 1/8, ln(8)) — defeating Pearl A's sentinel branch (`prev_x_mean == 0 AND state.x_lag == 0`). Updated the 5 retrofit dispatch arms (`isv_h_s2_rms_ema`, `isv_vsn_mask_g{0..5}_ema`, `isv_aux_label_scale_ema`, `isv_moe_expert_util_ema`, `isv_moe_gate_entropy_ema`) to write 0.0 in lockstep with the Wiener-state half. Registry descriptions updated in the same commit per `feedback_no_partial_refactor.md` (doc + dispatch are two halves of the same contract). Stale `141`/`2048`/`47` Wiener-buffer comments updated to `207`/`2816`/`69` in `state_reset_registry.rs`, `training_loop.rs`, and `gpu_dqn_trainer.rs`. `cargo check -p ml --offline` clean; `sp4_wiener_ema` 6/6 + `state_reset_registry` 3/3 tests passing.
SP4 Layer A Task A13 code-quality refactor — apply_pearls_to_slot helper + REWARD_COMPONENT_COUNT named constant + SP4_PRODUCER_COUNT module-level promotion + _ema_alpha_unused removal + label-scale doc fix (2026-05-01): five IMPORTANT items from the A13 code-quality review addressed in a single coordinated commit per `feedback_no_partial_refactor.md`. (1) New `apply_pearls_to_slot(scratch_host, scratch_idx, isv_pinned, isv_idx, wiener_host, wiener_offset)` helper in `sp4_wiener_ema.rs` collapses the ~30-line read_volatile/`pearls_ad_update`/write_volatile per-slot block into a single `unsafe fn`; consumed by 11 launchers (SP4 producers `launch_sp4_target_q_p99`, `launch_sp4_atom_pos_p99_all_branches`, `launch_sp4_param_group_oracles_all_groups` (closure), `launch_sp4_grad_norm_p99`, `launch_sp4_h_s2_p99`; A13 retrofit `launch_h_s2_rms_ema`, `launch_aux_heads_loss_ema`, `launch_vsn_mask_ema`, `launch_moe_expert_util_ema` (closure), `launch_iqn_quantile_ema`; cross-boundary `GpuExperienceCollector::launch_reward_component_ema_inplace`) plus the inline label-scale block in `aux_heads_forward` Step 2b. Pearl C `pearl_c_rate_deficit_state_buf` site at `gpu_dqn_trainer.rs:21436` left untouched — different mapped-pinned buffer + Rust-side ema array means it doesn't share the helper's signature contract; refactoring would force a different abstraction. New unit test `apply_pearls_to_slot_pearl_a_bootstrap_path` exercises the helper with heap arrays standing in for mapped-pinned pointers (`*const f32`/`*mut f32` ABI is identical). (2) New `pub const REWARD_COMPONENT_COUNT: usize = 6` constant declared next to `REWARD_POPART_EMA_INDEX` replaces 4 hardcoded `6` literals: `block_dim: (REWARD_COMPONENT_COUNT as u32, 1, 1)` in `launch_reward_component_ema_inplace`, `for c in 0..REWARD_COMPONENT_COUNT` Pearls A+D loop, `REWARD_POPART_EMA_INDEX + REWARD_COMPONENT_COUNT` in `state_reset_registry::reset_named_state` ISV-slot range, plus a `debug_assert_eq!(REWARD_COMPONENT_COUNT, 6, "reward-component layout invariant — kernel block_dim and reward_components_per_sample stride must update together")` invariant guard mirroring `MOE_NUM_EXPERTS` / `SL_NUM_FEATURE_GROUPS` style. Kernel `reward_component_ema_kernel.cu` retains the literal `6` (allows nvcc to fully unroll the row-stride arithmetic) but its comment now documents the host-side `REWARD_COMPONENT_COUNT` invariant. (3) `SP4_PRODUCER_COUNT` / `SP4_WIENER_FLOATS_PER_SLOT` / `SP4_WIENER_TOTAL_FLOATS` promoted from fn-local consts inside `pub fn new` to module-level `pub const`s in `gpu_dqn_trainer.rs`, re-exported via `cuda_pipeline::mod.rs` so `crates/ml/tests/sp4_producer_unit_tests.rs` can `use ml::cuda_pipeline::SP4_PRODUCER_COUNT;` instead of redeclaring the value at 13 test sites. All 13 redeclarations deleted. (4) `_ema_alpha_unused: f32` parameter removed from 6 retrofit launchers (`launch_h_s2_rms_ema`, `launch_aux_heads_loss_ema`, `launch_vsn_mask_ema`, `launch_moe_expert_util_ema`, `launch_iqn_quantile_ema`, `launch_reward_component_ema_inplace`) and the `FusedTrainingCtx::launch_iqn_quantile_ema` proxy; all callers in `training_loop.rs` updated per `feedback_no_legacy_aliases.md` (no soft-deprecated wrappers — rename all call sites directly). Doc-comments on each launcher updated from "α dropped per SP4 — Pearls A+D adapt α from per-slot signal-vs-noise variance. The `_ema_alpha_unused` argument is preserved so callers compile unchanged" to "α is derived adaptively from per-slot Pearls A+D Wiener state — see `sp4_wiener_ema::pearls_ad_update`". (5) Stale doc reference at `gpu_aux_heads.rs::launch_label_scale_ema:391` referencing non-existent `launch_label_scale_ema_with_pearls` corrected to point at the actual inline Pearls A+D block in `GpuDqnTrainer::aux_heads_forward` Step 2b (which now consumes `apply_pearls_to_slot`). Pure refactor — no spec or behaviour change; same kernel launches, same Pearls A+D semantics, same ISV slot writes. `cargo check -p ml --offline` clean (12 pre-existing warnings, no new); `cargo test -p ml --lib --offline sp4_wiener_ema` 7/7 passing (6 originals + 1 new helper test); `cargo test -p ml --lib --offline state_reset_registry` 3/3 passing.
SP4 Layer A Task A13.5 — reward_component_ema retrofit (Pearls A+D) + cross-boundary wiring + orphan deletion (2026-05-01): retrofit the existing per-step reward-component EMA producer in `crates/ml/src/cuda_pipeline/reward_component_ema_kernel.cu` AND wire the host-side Pearls A+D update across the trainer/collector boundary (mirroring the A14/A15 Pearl C wiring path) AND delete the trainer's orphan `launch_reward_component_ema` per `feedback_wire_everything_up.md`. Kernel signature changed: drops `(isv_out, ema_alpha, isv_reward_base_slot)` for `(scratch_buf, scratch_first_index=63)`. Single-block 6-thread (one per component); each thread reduces mean |r_c| over `n_samples` samples and writes the step observation to `scratch_buf[scratch_first_index + c]` for c in 0..6 with `__threadfence_system()`. **6 ISV slots wired with Pearls A+D**: ISV[REWARD_POPART_EMA_INDEX..+6) = ISV[63..69), with Wiener offsets `(63+c)*3 = 189..207` for c in 0..6 — these are the LAST slots in the post-A13 207-float `wiener_state_buf`. **Cross-boundary wiring (mirrors A14/A15 Pearl C precedent)**: 4 new fields on `GpuExperienceCollector` — `reward_component_pearls_wiener_host_ptr: *mut f32`, `reward_component_pearls_scratch_dev_ptr: u64`, `reward_component_pearls_scratch_host_ptr: *mut f32`, `reward_component_pearls_isv_pinned_ptr: *mut f32` — all `NULL`/`0` until wired. New setter `set_reward_component_pearls_buffers(wiener_host, scratch_dev, scratch_host, isv_pinned)` on the collector. New accessors on `GpuDqnTrainer`: `wiener_state_buf_host_ptr()`, `producer_step_scratch_buf_dev_ptr()`, `producer_step_scratch_buf_host_ptr()`, `isv_signals_pinned_ptr()`. New wire helper `FusedTrainingCtx::wire_reward_component_pearls_buffers(&mut self, &mut GpuExperienceCollector)` pulls all four pointers from the trainer and pushes them into the collector. Wired once in `training_loop.rs::init_gpu_experience_collector` immediately after `set_curiosity_pearl_c_buffers` (mirrors the A14/A15 wire call site). The retrofitted `launch_reward_component_ema_inplace` on the collector now: launches the kernel with `(rewards_ptr, n_samples, scratch_dev, scratch_first_index=63)`, syncs the stream, applies Pearls A+D in a `for c in 0..6` loop (read prev_x_mean from `isv_pinned`, read Wiener triple from `wiener_host`, call `pearls_ad_update`, write back ISV + Wiener) — all via mapped-pinned host pointers (no HtoD/DtoH per `feedback_no_htod_htoh_only_mapped_pinned.md`), then memsets `reward_components_per_sample` to zero (preserving original behaviour). Degenerate-zero short-circuit per slot — covers the always-zero placeholder components (c=2 trail, c=5 bonus) plus cold-start before first `collect_experiences_gpu`. **Orphan deletion** per `feedback_wire_everything_up.md`: removed `GpuDqnTrainer::launch_reward_component_ema` (lines 8775-8796 — zero call sites pre-deletion), removed the trainer-side `reward_component_ema_kernel: CudaFunction` field (zero consumers post-launcher-deletion), removed the cubin loader at line 11591-11596, removed the field from the constructor's struct-init at line 14677. The `REWARD_COMPONENT_EMA_CUBIN` static remains because the collector still loads from it. **Unit test** `sp4_reward_component_ema_writes_step_obs_via_pearl_a_then_converges_pearl_d` (`#[ignore]`-gated for GPU): drives kernel with N=128 reward_components where `r[i*6+c] = sign(i) × (c+1)` (analytical mean|r_c| = c+1 for c in 0..6). Asserts each scratch slot ∈ ±1e-5 of (c+1), non-target slots remain 0; verifies Pearl A bootstrap on component 0 + Pearl D convergence (1000 stationary observations within 1% of 1.0). The cross-boundary wiring path is exercised in production by the integration smoke harness; this kernel-direct test isolates the kernel signature retrofit + Pearls A+D semantics. Per `feedback_no_atomicadd.md`, `feedback_no_htod_htoh_only_mapped_pinned.md`, `feedback_no_partial_refactor.md`, `feedback_wire_everything_up.md`. `cargo check -p ml --lib --tests --offline` clean (11 pre-existing warnings, no new warnings); `cargo test -p ml --lib sp4_wiener_ema --offline` 6/6 passing.