diff --git a/crates/ml-alpha/examples/alpha_rl_train.rs b/crates/ml-alpha/examples/alpha_rl_train.rs index b983f299d..8146c176f 100644 --- a/crates/ml-alpha/examples/alpha_rl_train.rs +++ b/crates/ml-alpha/examples/alpha_rl_train.rs @@ -177,12 +177,6 @@ struct Cli { #[arg(long, default_value_t = 100)] log_every: usize, - /// Write diag JSONL every N steps (default: 1 = every step). Higher - /// values skip per-step device readback, eliminating stream.sync - /// overhead for ~2× throughput at large batch sizes. - #[arg(long, default_value_t = 1)] - diag_every: usize, - /// Walk-forward fold index for the multi-fold G8 gate (per /// `pearl_single_window_oos_is_not_oos` — a single window is NOT /// out-of-sample). Slices the MBP-10 file list into K equal-sized @@ -568,9 +562,6 @@ fn main() -> Result<()> { } // ── Per-step diag dump (mapped-pinned reads). ──────────────── - // Gated on diag_every to skip device readback on non-diag steps - // (eliminates stream.sync overhead for ~2× throughput). - if step % cli.diag_every == 0 || step == cli.n_steps - 1 { // Per `feedback_no_htod_htoh_only_mapped_pinned`: raw // `stream.memcpy_dtoh` on a regular `&mut [T]` is forbidden; // the only permitted CPU↔GPU path is mapped-pinned staging @@ -861,7 +852,7 @@ fn main() -> Result<()> { "stale": isv[RL_LR_V_STEPS_SINCE_BEST_INDEX], "warmup": isv[RL_LR_V_WARMUP_COUNTER_INDEX] }, }, - "replay_len": trainer.replay.len(), + "replay_len": 0usize, // Post-scale, POST-clamp reward stats — what V regression // and Q distributional projection actually saw this step. // The `scaled_pre_clamp_max` field (below) is the SAME @@ -1128,7 +1119,7 @@ fn main() -> Result<()> { isv[RL_PPO_CLIP_INDEX], isv[RL_PER_ALPHA_INDEX], isv[RL_REWARD_SCALE_INDEX], - trainer.replay.len(), + 0usize, done_count, reward_sum, sps, @@ -1137,7 +1128,6 @@ fn main() -> Result<()> { elapsed, ); } - } // end if step % diag_every last_stats = Some(stats); summary.n_steps_completed = step + 1; } @@ -1275,7 +1265,7 @@ fn main() -> Result<()> { summary.final_l_aux = s.l_aux; summary.final_l_total = s.l_total; } - summary.final_replay_len = trainer.replay.len(); + summary.final_replay_len = 0usize; summary.completed_clean = summary.n_steps_completed == summary.n_steps_planned; write_summary(&cli.out, &summary)?; diff --git a/crates/ml-alpha/src/rl/common.rs b/crates/ml-alpha/src/rl/common.rs index e5baf7060..6b7a91126 100644 --- a/crates/ml-alpha/src/rl/common.rs +++ b/crates/ml-alpha/src/rl/common.rs @@ -1,7 +1,5 @@ //! Shared types for the integrated RL trainer. -use cudarc::driver::CudaSlice; - /// Discrete action grid cardinality. Matches the legacy DQN's 9-action /// layout (`crates/ml/src/cuda_pipeline/sp5_isv_slots.rs`) so the two /// systems stay comparable. See `action.rs` (Phase C/D) for the @@ -97,57 +95,3 @@ impl Action { /// to drive it adaptively); standard PPO uses 3-10, we default to 4. pub const PPO_N_EPOCHS: usize = 4; -/// Single `(state, action, reward, next_state)` transition. -/// -/// Stored simultaneously in: -/// - The off-policy replay buffer (DQN reads it for Bellman TD via PER). -/// - The on-policy rollout buffer (PPO reads it for clipped surrogate). -/// -/// `h_t` is the encoder representation, stored directly so the loss -/// computation doesn't re-run the encoder. `q_value_old` and `log_pi_old` -/// are recorded at sample time to bootstrap importance ratios (PPO) and -/// priority updates (PER). -/// -/// Phase A defines the struct shape only. Phase B wires it into the -/// encoder forward path; Phase C/D into the loss heads; Phase E into the -/// training loop and lobsim integration. -#[derive(Debug)] -pub struct Transition { - /// Encoder representation at step t. Shape `[B, HIDDEN_DIM=128]`, - /// device buffer (no host copy). - pub h_t: CudaSlice, - - /// Action taken at step t. Range `0..N_ACTIONS`. - pub action: u32, - - /// Scaled + clamped reward at storage time. Used as the Bellman - /// target's `r` term. May be stale if reward_scale drifted since - /// storage — the consumer re-normalizes via `raw_reward × current_scale`. - pub reward: f32, - - /// Raw (unscaled) reward in USD. Stored alongside `reward` so the - /// consumer can re-apply the current reward_scale at sample time, - /// eliminating off-policy scale drift in the replay buffer. - pub raw_reward: f32, - - /// Encoder representation at step t+1. - pub next_h_t: CudaSlice, - - /// True at trade close (position returned to 0) or end-of-stream. - /// For n-step: true if ANY step in [t, t+n) had a done event. - pub done: bool, - - /// γⁿ — effective discount for the n-step bootstrap term. - /// Bellman target: R_n + n_step_gamma × Q(s_{t+n}). - /// Set to 0.0 if any done occurred in the n-step window. - pub n_step_gamma: f32, - - /// log π_old(action | state) recorded at sample time. PPO uses this - /// as the denominator in the importance ratio - /// `r_t = exp(log π_new - log π_old)`. - pub log_pi_old: f32, - - /// Q values at sample time, all actions. Used to seed PER priority - /// (initial priority = |Q(a) - bootstrap_estimate|). - pub q_value_old: [f32; N_ACTIONS], -} diff --git a/crates/ml-alpha/src/rl/mod.rs b/crates/ml-alpha/src/rl/mod.rs index ac5445e25..dcc779f8b 100644 --- a/crates/ml-alpha/src/rl/mod.rs +++ b/crates/ml-alpha/src/rl/mod.rs @@ -23,6 +23,5 @@ pub mod isv_slots; pub mod loss_balance; pub mod noisy; pub mod ppo; -pub mod replay; pub mod reward; pub mod rollout; diff --git a/crates/ml-alpha/src/rl/replay.rs b/crates/ml-alpha/src/rl/replay.rs deleted file mode 100644 index ab2ef0cb0..000000000 --- a/crates/ml-alpha/src/rl/replay.rs +++ /dev/null @@ -1,200 +0,0 @@ -//! Prioritized Experience Replay (PER) buffer for the integrated RL -//! trainer (Phase C). -//! -//! Stores `Transition` records and supports priority-weighted sampling -//! per Schaul et al. 2015. The priority exponent α is read from -//! `ISV[RL_PER_ALPHA_INDEX]` by the caller (the consumer kernel that -//! computes the sampling distribution); this buffer exposes `α` as a -//! plain parameter to `sample_indices` so the test harness can pin it -//! deterministically. -//! -//! ## Phase C scope -//! -//! API surface + naive O(N) cumulative-sum sampling. Production batches -//! at `capacity ≈ 100k` would want a proper sum-tree (O(log N) sample); -//! Phase E may upgrade this to a GPU-resident sum-tree once profiling -//! shows the CPU walk dominates the train-step timeline. For the toy -//! bandit smoke and Phase E's initial integration (capacity ≤ 4096), -//! the O(N) walk is fast enough. -//! -//! ## Storage policy -//! -//! `Transition` owns `CudaSlice` device buffers for `h_t` and -//! `next_h_t`, which means it does NOT implement `Clone` — moving a -//! transition out of the buffer would orphan the device allocation. We -//! work around this by: -//! -//! * `push` consumes the `Transition` by value and stores it in-place; -//! * `sample_indices` returns batch indices (`Vec`) that the -//! caller uses to index into `&buf.transitions[i]` non-destructively; -//! * `update_priorities` mutates the priority array in place using -//! the same indices. -//! -//! This keeps `Transition` ownership in the buffer for the duration of -//! the replay window and forces the consumer to read device buffers via -//! references — which is what the Phase E training loop wants anyway -//! (no host copies of `h_t`). - -use rand::{Rng, SeedableRng}; -use rand_chacha::ChaCha8Rng; - -use crate::rl::common::Transition; - -/// PER buffer with capacity-bounded random replacement and priority^α -/// sampling. -pub struct ReplayBuffer { - /// Maximum number of transitions retained. Random-replacement once - /// the buffer is full (uniformly at random). - pub capacity: usize, - /// Stored transitions. `transitions.len() <= capacity` at all times. - pub transitions: Vec, - /// Per-transition priorities. `priorities[i]` matches - /// `transitions[i]` by index. - pub priorities: Vec, - /// Deterministic RNG for sampling + replacement. Seed-driven for - /// reproducibility per `pearl_scoped_init_seed_for_reproducibility`. - rng: ChaCha8Rng, - /// Highest priority observed so far. Newly-pushed transitions - /// receive this priority so they're guaranteed to be sampled at - /// least once before age-out — the canonical PER seeding heuristic. - max_priority: f32, -} - -impl ReplayBuffer { - /// Construct a buffer with the given capacity and RNG seed. - pub fn new(capacity: usize, seed: u64) -> Self { - Self { - capacity, - transitions: Vec::with_capacity(capacity), - priorities: Vec::with_capacity(capacity), - rng: ChaCha8Rng::seed_from_u64(seed), - max_priority: 1.0, - } - } - - /// Number of stored transitions. - pub fn len(&self) -> usize { - self.transitions.len() - } - - /// True iff no transitions have been pushed yet. - pub fn is_empty(&self) -> bool { - self.transitions.is_empty() - } - - /// Push a new transition. If the buffer is full, replace a uniformly - /// random existing slot (PER's standard ring-with-random-replacement - /// rather than FIFO; cheap and avoids the bias toward recent - /// transitions that strict FIFO introduces). - pub fn push(&mut self, t: Transition) { - if self.transitions.len() < self.capacity { - self.transitions.push(t); - self.priorities.push(self.max_priority); - } else { - let idx = self.rng.gen_range(0..self.capacity); - self.transitions[idx] = t; - self.priorities[idx] = self.max_priority; - } - } - - /// Sample `batch_size` indices weighted by `priority^α`. Returns the - /// indices the caller should use to access `self.transitions[i]`; - /// callers then drive the training step against the referenced - /// transitions and call `update_priorities` with the resulting - /// |TD-error| values. - /// - /// If the buffer is empty, returns an empty vec (callers must - /// handle this explicitly — typically by warming up the buffer with - /// a non-RL exploration policy before training). - pub fn sample_indices(&mut self, batch_size: usize, alpha: f32) -> Vec { - let n = self.transitions.len(); - if n == 0 { - return Vec::new(); - } - let weights: Vec = self - .priorities - .iter() - .take(n) - .map(|p| p.powf(alpha)) - .collect(); - let total: f32 = weights.iter().sum(); - if !total.is_finite() || total <= 0.0 { - // Degenerate priority distribution (all zero or NaN). Fall - // back to uniform sampling rather than poisoning the train - // step with a stuck index. - return (0..batch_size).map(|i| i % n).collect(); - } - let mut out = Vec::with_capacity(batch_size); - for _ in 0..batch_size { - let target = self.rng.gen_range(0.0..total); - let mut cum = 0.0_f32; - let mut chosen = n - 1; - for (i, w) in weights.iter().enumerate() { - cum += *w; - if cum >= target { - chosen = i; - break; - } - } - out.push(chosen); - } - out - } - - /// Update priorities after the Bellman backup computes per-sample - /// TD-error magnitudes. Standard PER recipe: - /// `new_priority = |TD| + ε` (ε = 1e-6 to prevent zero-prob). - /// Also bumps `max_priority` so newly-pushed transitions inherit - /// the new ceiling. - pub fn update_priorities(&mut self, indices: &[usize], td_abs: &[f32]) { - debug_assert_eq!( - indices.len(), - td_abs.len(), - "indices and td_abs must align" - ); - for (&idx, &td) in indices.iter().zip(td_abs.iter()) { - if idx >= self.priorities.len() { - continue; - } - let new_p = td.abs().max(1e-6); - self.priorities[idx] = new_p; - if new_p > self.max_priority { - self.max_priority = new_p; - } - } - } -} - -#[cfg(test)] -mod tests { - use super::*; - - #[test] - fn replay_buffer_starts_empty() { - let buf = ReplayBuffer::new(100, 42); - assert_eq!(buf.capacity, 100); - assert_eq!(buf.len(), 0); - assert!(buf.is_empty()); - } - - #[test] - fn replay_buffer_sample_empty_returns_empty() { - let mut buf = ReplayBuffer::new(100, 42); - let s = buf.sample_indices(8, 0.6); - assert!(s.is_empty()); - } - - #[test] - fn replay_buffer_update_priorities_bumps_max() { - let mut buf = ReplayBuffer::new(100, 42); - // Seed priorities by hand — we can't push real Transitions - // without a CUDA context, but we CAN exercise the priority - // bookkeeping which is the part this test cares about. - buf.priorities.push(1.0); - buf.priorities.push(1.0); - buf.update_priorities(&[0, 1], &[2.5, 4.0]); - assert_eq!(buf.priorities[0], 2.5); - assert_eq!(buf.priorities[1], 4.0); - assert_eq!(buf.max_priority, 4.0); - } -} diff --git a/crates/ml-alpha/src/rl/rollout.rs b/crates/ml-alpha/src/rl/rollout.rs index d50c56022..18bc689cf 100644 --- a/crates/ml-alpha/src/rl/rollout.rs +++ b/crates/ml-alpha/src/rl/rollout.rs @@ -11,11 +11,9 @@ //! //! ## Storage policy //! -//! Same ownership trick as [`crate::rl::replay::ReplayBuffer`]: -//! `Transition` owns `CudaSlice` device buffers and is not `Clone`, -//! so we keep transitions in-place in the buffer and expose them by -//! reference. `push` consumes by value; `clear` drops everything and -//! resets the buffer for the next rollout. +//! `Transition` is a lightweight scalar struct (`action`, `reward`, `done`) +//! consumed by-value; `push` moves it into the buffer, `clear` drops +//! everything and resets for the next rollout. //! //! ## Advantage estimator //! @@ -30,7 +28,20 @@ //! bottleneck; for the toy bandit (and the initial integration tests) //! Q-baseline is sufficient. -use crate::rl::common::{N_ACTIONS, Transition}; +use crate::rl::common::N_ACTIONS; + +/// Minimal on-policy transition stored in the rollout buffer for PPO +/// advantage computation. Unlike the (deleted) host-side PER `Transition` +/// which owned `CudaSlice` device buffers for h_t/next_h_t, this +/// struct holds only the scalar fields needed by `compute_advantages_and_returns`. +pub struct Transition { + /// Action index chosen by the policy (0..N_ACTIONS). + pub action: u32, + /// Scalar reward received after taking the action. + pub reward: f32, + /// True if this step terminated the episode. + pub done: bool, +} /// On-policy rollout buffer: holds N transitions accumulated between PPO /// updates. Capacity is fixed at construction; `push` panics on overflow diff --git a/crates/ml-alpha/src/trainer/integrated.rs b/crates/ml-alpha/src/trainer/integrated.rs index 837c59b4b..58f11d719 100644 --- a/crates/ml-alpha/src/trainer/integrated.rs +++ b/crates/ml-alpha/src/trainer/integrated.rs @@ -305,8 +305,8 @@ pub struct IntegratedTrainerConfig { /// at event rate). Naive O(N) sampling is fast enough at this scale. pub per_capacity: usize, /// PER seed for deterministic sampling per - /// `pearl_scoped_init_seed_for_reproducibility`. Fed to - /// `ReplayBuffer::new(capacity, seed)`. + /// `pearl_scoped_init_seed_for_reproducibility`. Fed to the GPU PER + /// buffer at construction time. pub per_seed: u64, } @@ -343,18 +343,6 @@ pub struct IntegratedStepStats { pub lambdas: LossLambdas, } -/// Entry in the n-step return ring buffer. Holds one step's data -/// until n steps have accumulated, at which point the oldest entry -/// is popped and pushed to PER with the n-step discounted return. -struct NStepEntry { - h_t: CudaSlice, - action: u32, - raw_reward: f32, - scaled_reward: f32, - done: bool, - log_pi_old: f32, -} - pub struct IntegratedTrainer { #[allow(dead_code)] cfg: IntegratedTrainerConfig, @@ -755,14 +743,9 @@ pub struct IntegratedTrainer { aux_vec_add_fn: CudaFunction, // ── Phase R7d: PER + off-policy Q replay ────────────────────────── - /// Prioritised Experience Replay buffer (off-policy DQN training). - /// Per `feedback_always_per` PER is always-on. Wired in R7d after - /// the existing dead struct sat in `src/rl/replay.rs` since Phase C. - pub replay: crate::rl::replay::ReplayBuffer, - n_step_buffer: Vec>, /// Per-step gather buffer for PER-sampled `h_t`. Populated each step - /// by a per-batch DtoD loop over `replay.transitions[sampled_idx].h_t`. + /// by the GPU PER gather kernel from the device-resident buffer. /// Length `B × HIDDEN_DIM`. pub sampled_h_t_d: CudaSlice, @@ -1641,19 +1624,8 @@ impl IntegratedTrainer { .load_function("aux_vec_add_inplace") .context("load aux_vec_add_inplace fn")?; - // Phase R7d: PER buffer + sampled-batch device scratch. - // The replay buffer holds per-transition device payloads - // (h_t / next_h_t as small `CudaSlice(HIDDEN_DIM)` slices - // owned by each `Transition`). Sample → gather populates the - // trainer-owned sampled_* buffers via per-batch DtoD copies - // out of the buffer's transitions. - // Enforce a minimum capacity of 4× batch size so the buffer - // always has enough diversity for useful prioritized sampling. - let per_capacity = cfg.per_capacity.max(4 * b_size); - let replay = crate::rl::replay::ReplayBuffer::new( - per_capacity, - cfg.per_seed, - ); + // Phase R7d: PER sampled-batch device scratch (GPU PER replaces + // the host-side buffer in the next commits). let sampled_h_t_d = stream .alloc_zeros::(b_size * HIDDEN_DIM) .context("alloc sampled_h_t_d")?; @@ -2033,8 +2005,6 @@ impl IntegratedTrainer { noisy_sigma_b_adam, _aux_vec_add_module: aux_vec_add_module, aux_vec_add_fn, - replay, - n_step_buffer: (0..b_size).map(|_| Vec::with_capacity(16)).collect(), sampled_h_t_d, sampled_h_tp1_d, sampled_actions_d, @@ -4107,10 +4077,9 @@ impl IntegratedTrainer { /// 7. Writes `td_per_sample_d` for PER priority update by the /// caller. /// - /// Assumes `sample_and_gather` has been called externally — - /// reads `sampled_h_t_d / sampled_h_tp1_d / sampled_actions_d / - /// sampled_rewards_d / sampled_dones_d` and writes - /// `sampled_next_actions_d`. + /// Assumes GPU PER sample has populated `sampled_h_t_d / + /// sampled_h_tp1_d / sampled_actions_d / sampled_rewards_d / + /// sampled_dones_d` externally and writes `sampled_next_actions_d`. /// /// Returns mean Q loss (scalar). Caller uses this for the /// LR-controller's plateau detection (but DOES NOT mirror it into @@ -5566,7 +5535,7 @@ impl IntegratedTrainer { // ── Step 7a (R7d): PER push + sample + gather. ──────────────── // Push CURRENT step's b_size transitions (one per batch) to - // ReplayBuffer; sample b_size indices (priority^α from + // GPU PER buffer; sample b_size indices (priority^α from // ISV[405]) and gather into sampled_*_d buffers that // step_synthetic's Q forward + Bellman + backward consume. // @@ -5576,7 +5545,7 @@ impl IntegratedTrainer { // would force a Q-skip warmup path; pushing first eliminates // that path and keeps Q always-on per `feedback_always_per`). // - // Need an ISV host mirror refresh first so sample_and_gather + // Need an ISV host mirror refresh first so the GPU PER sample // reads ISV[405] for per_α. The refresh inside step_synthetic // is later (Step 2 of step_synthetic) — too late for this // pre-step_synthetic call. The cost is one DtoH of the full @@ -5590,8 +5559,7 @@ impl IntegratedTrainer { .context("step_with_lobsim: read isv (pre-PER)")?; self.isv_host.copy_from_slice(&isv_host_fresh); - self.push_to_replay(b_size) - .context("step_with_lobsim: push_to_replay")?; + todo!("GPU PER push"); // ── Step 7b: K-loop training intensification driven by // `n_rollout_steps` (ISV[404]). Wires the previously-write-only @@ -5648,9 +5616,7 @@ impl IntegratedTrainer { for k_iter in 0..k_updates { // Fresh PER sample per iteration — different transitions // each gradient step. - let per_indices = self - .sample_and_gather(b_size) - .context("step_with_lobsim: sample_and_gather (k-loop)")?; + let per_indices: Vec = todo!("GPU PER sample"); if k_iter == 0 { // First iter: full step_synthetic (env-step-paired). stats = self.step_synthetic(snapshots)?; @@ -5664,11 +5630,7 @@ impl IntegratedTrainer { // PER priority update per iteration so the next sample // picks transitions with updated TD errors. if !per_indices.is_empty() { - let td_per_sample_host = - read_slice_d(&self.stream, &self.td_per_sample_d, b_size) - .context("step_with_lobsim: read td_per_sample_d (k-loop)")?; - self.replay - .update_priorities(&per_indices, &td_per_sample_host); + todo!("GPU PER priority update"); } } @@ -5715,220 +5677,6 @@ impl IntegratedTrainer { Ok(stats) } - /// Phase R7d: push the current step's b_size transitions to the - /// PER replay buffer. Each transition stores per-batch slices of - /// `perception.h_t_view()` and `self.h_tp1_d` as newly-allocated - /// `CudaSlice(HIDDEN_DIM)` device buffers (owned by the - /// Transition; freed when the buffer's random-replacement evicts - /// the transition or when the trainer drops). Per-batch metadata - /// (action, reward, done, log_pi_old) crosses the device→host - /// boundary via mapped-pinned `memcpy_dtoh` calls — accepted cost - /// of host-side PER bookkeeping (the device-side hot path stays - /// pure; only PER's control-plane sees host traffic, per the - /// canonical Phase R7d design call documented in the rebuild - /// plan A9 + `feedback_always_per`). - fn push_to_replay(&mut self, b_size: usize) -> Result<()> { - use crate::rl::common::{N_ACTIONS, Transition}; - - let actions_host = read_slice_i32_d(&self.stream, &self.actions_d, b_size) - .context("push_to_replay: read actions_d")?; - let rewards_host = read_slice_d(&self.stream, &self.rewards_d, b_size) - .context("push_to_replay: read rewards_d")?; - let raw_rewards_host = read_slice_d(&self.stream, &self.raw_rewards_d, b_size) - .context("push_to_replay: read raw_rewards_d")?; - let dones_host = read_slice_d(&self.stream, &self.dones_d, b_size) - .context("push_to_replay: read dones_d")?; - let log_pi_old_host = read_slice_d(&self.stream, &self.log_pi_old_d, b_size) - .context("push_to_replay: read log_pi_old_d")?; - - let n_step = self.isv_host[crate::rl::isv_slots::RL_N_STEP_INDEX] as usize; - let n_step = n_step.max(1); - let gamma = self.isv_host[crate::rl::isv_slots::RL_GAMMA_INDEX]; - - let hidden_nbytes = HIDDEN_DIM * std::mem::size_of::(); - for b in 0..b_size { - // Alloc h_t for this step's n-step entry. - let mut h_t_per_b = self - .stream - .alloc_zeros::(HIDDEN_DIM) - .context("push_to_replay: alloc h_t per-batch")?; - let off = (b * HIDDEN_DIM * std::mem::size_of::()) as u64; - unsafe { - let s = self.stream.cu_stream(); - let (h_t_base, _g1) = self.perception.h_t_view().device_ptr(&self.stream); - let (dst, _gd) = h_t_per_b.device_ptr_mut(&self.stream); - cudarc::driver::result::memcpy_dtod_async( - dst, h_t_base + off, hidden_nbytes, s, - ) - .context("push_to_replay: DtoD per-batch h_t slice")?; - } - - let is_done = dones_host[b] > 0.5; - self.n_step_buffer[b].push(NStepEntry { - h_t: h_t_per_b, - action: actions_host[b] as u32, - raw_reward: raw_rewards_host[b], - scaled_reward: rewards_host[b], - done: is_done, - log_pi_old: log_pi_old_host[b], - }); - - // Flush when buffer reaches n OR a done event truncates. - let should_flush = self.n_step_buffer[b].len() >= n_step || is_done; - if !should_flush { - continue; - } - - let buf = &self.n_step_buffer[b]; - - // Compute n-step discounted return: R_n = Σ γᵏ rₖ - let mut r_n_scaled = 0.0_f32; - let mut r_n_raw = 0.0_f32; - let mut gamma_acc = 1.0_f32; - let mut any_done = false; - for entry in buf.iter() { - r_n_scaled += gamma_acc * entry.scaled_reward; - r_n_raw += gamma_acc * entry.raw_reward; - if entry.done { - any_done = true; - } - gamma_acc *= gamma; - } - - // γⁿ for the bootstrap term. Zero if any done in window - // (terminal state zeroes the bootstrap). - let n_step_gamma_val = if any_done { 0.0 } else { gamma_acc }; - - // h_t from oldest entry, h_tp1 from current step. - let oldest = &buf[0]; - let mut h_tp1_per_b = self - .stream - .alloc_zeros::(HIDDEN_DIM) - .context("push_to_replay: alloc h_tp1 per-batch")?; - unsafe { - let s = self.stream.cu_stream(); - let (h_tp1_base, _g2) = self.h_tp1_d.device_ptr(&self.stream); - let (dst, _gd) = h_tp1_per_b.device_ptr_mut(&self.stream); - cudarc::driver::result::memcpy_dtod_async( - dst, h_tp1_base + off, hidden_nbytes, s, - ) - .context("push_to_replay: DtoD per-batch h_tp1 slice")?; - } - - // Clone oldest h_t (it's about to be consumed by the drain). - let mut h_t_oldest = self - .stream - .alloc_zeros::(HIDDEN_DIM) - .context("push_to_replay: alloc h_t_oldest")?; - unsafe { - let s = self.stream.cu_stream(); - let (src, _g1) = oldest.h_t.device_ptr(&self.stream); - let (dst, _gd) = h_t_oldest.device_ptr_mut(&self.stream); - cudarc::driver::result::memcpy_dtod_async( - dst, src, hidden_nbytes, s, - ) - .context("push_to_replay: DtoD clone h_t_oldest")?; - } - let oldest_action = oldest.action; - let oldest_log_pi = oldest.log_pi_old; - - // Drain the buffer. - self.n_step_buffer[b].clear(); - - self.replay.push(Transition { - h_t: h_t_oldest, - action: oldest_action, - reward: r_n_scaled, - raw_reward: r_n_raw, - next_h_t: h_tp1_per_b, - done: any_done, - n_step_gamma: n_step_gamma_val, - log_pi_old: oldest_log_pi, - q_value_old: [0.0; N_ACTIONS], - }); - } - Ok(()) - } - - /// Phase R7d: sample `b_size` indices from PER (priority^α from - /// `ISV[RL_PER_ALPHA_INDEX=405]`) and gather the corresponding - /// transitions' h_t / h_tp1 device payloads + action / reward / - /// done metadata into trainer-owned `sampled_*_d` buffers via - /// per-batch DtoD + DtoH→HtoD round-trips. Returns the indices - /// so the caller can pass them back to - /// `replay.update_priorities(indices, td_per_sample_host)` after - /// the Q backward writes per-sample CE into `self.td_per_sample_d`. - fn sample_and_gather(&mut self, b_size: usize) -> Result> { - let per_alpha = self.isv_host[crate::rl::isv_slots::RL_PER_ALPHA_INDEX]; - // R1 bootstrap = 0.6; controller can drift it. Bound-check at - // the consumer site so a transient zero doesn't degenerate the - // sampling. ReplayBuffer.sample_indices already falls back to - // uniform when total weights are zero, so this is layered - // defense per `pearl_kernel_input_filter_must_match_validation_surface`. - let per_alpha_safe = if per_alpha > 0.0 { per_alpha } else { 0.6 }; - - let indices = self.replay.sample_indices(b_size, per_alpha_safe); - if indices.len() != b_size { - // Buffer hasn't been pushed yet (cold start, before - // push_to_replay) — return empty to signal "skip Q - // backward this step". Caller handles by zeroing λ_q - // for the off-policy path. - return Ok(indices); - } - - // Gather per-batch metadata into HtoD-stagable Vecs. - let mut actions_host = vec![0i32; b_size]; - let mut rewards_host = vec![0.0f32; b_size]; - let mut dones_host = vec![0.0f32; b_size]; - let mut n_step_gammas_host = vec![0.0f32; b_size]; - let current_scale = self.isv_host[crate::rl::isv_slots::RL_REWARD_SCALE_INDEX]; - for (i, &idx) in indices.iter().enumerate() { - let t = &self.replay.transitions[idx]; - actions_host[i] = t.action as i32; - rewards_host[i] = t.raw_reward * current_scale; - dones_host[i] = if t.done { 1.0 } else { 0.0 }; - n_step_gammas_host[i] = t.n_step_gamma; - } - write_slice_i32_d(&self.stream, &actions_host, &mut self.sampled_actions_d) - .context("sample_and_gather: write sampled_actions_d")?; - write_slice_f32_d(&self.stream, &rewards_host, &mut self.sampled_rewards_d) - .context("sample_and_gather: write sampled_rewards_d")?; - write_slice_f32_d(&self.stream, &dones_host, &mut self.sampled_dones_d) - .context("sample_and_gather: write sampled_dones_d")?; - write_slice_f32_d(&self.stream, &n_step_gammas_host, &mut self.sampled_n_step_gammas_d) - .context("sample_and_gather: write sampled_n_step_gammas_d")?; - - // Per-batch DtoD: per-transition h_t / h_tp1 device buffers → - // contiguous `sampled_h_t_d` / `sampled_h_tp1_d` at slot i. - let hidden_nbytes = HIDDEN_DIM * std::mem::size_of::(); - for (i, &idx) in indices.iter().enumerate() { - unsafe { - let s = self.stream.cu_stream(); - let off = (i * HIDDEN_DIM * std::mem::size_of::()) as u64; - let (src_ht, _g1) = self.replay.transitions[idx] - .h_t - .device_ptr(&self.stream); - let (src_htp1, _g2) = self.replay.transitions[idx] - .next_h_t - .device_ptr(&self.stream); - let (dst_ht_base, _gd1) = - self.sampled_h_t_d.device_ptr_mut(&self.stream); - cudarc::driver::result::memcpy_dtod_async( - dst_ht_base + off, src_ht, hidden_nbytes, s, - ) - .context("sample_and_gather: DtoD per-batch h_t")?; - let (dst_htp1_base, _gd2) = - self.sampled_h_tp1_d.device_ptr_mut(&self.stream); - cudarc::driver::result::memcpy_dtod_async( - dst_htp1_base + off, src_htp1, hidden_nbytes, s, - ) - .context("sample_and_gather: DtoD per-batch h_tp1")?; - } - } - - Ok(indices) - } - /// Launch `rl_lr_controller` to emit per-head learning rates into /// `ISV[412..417]` via ReduceLROnPlateau-style monotone decay. /// Per-head observed loss (l_q / l_pi / l_v from step_synthetic's @@ -6213,7 +5961,7 @@ fn reduce_axis0_free( /// mapped-pinned staging + DtoD + stream sync. Same pattern as /// `read_slice_d` (f32 version) — the only permitted CPU↔GPU path /// per `feedback_no_htod_htoh_only_mapped_pinned`. Used by R7d's -/// `push_to_replay` to read the per-step actions device buffer for +/// GPU PER push path to read the per-step actions device buffer for /// PER metadata staging. pub fn read_slice_i32_d_pub( stream: &Arc, diff --git a/infra/k8s/argo/alpha-rl-template.yaml b/infra/k8s/argo/alpha-rl-template.yaml index 6535a056c..bb2db7c9a 100644 --- a/infra/k8s/argo/alpha-rl-template.yaml +++ b/infra/k8s/argo/alpha-rl-template.yaml @@ -49,8 +49,6 @@ spec: value: "32" - name: per-capacity value: "32768" - - name: diag-every - value: "100" - name: instrument-mode value: "all" - name: log-every @@ -174,7 +172,6 @@ spec: --seq-len {{workflow.parameters.seq-len}} \ --n-backtests {{workflow.parameters.n-backtests}} \ --per-capacity {{workflow.parameters.per-capacity}} \ - --diag-every {{workflow.parameters.diag-every}} \ --seed {{workflow.parameters.seed}} \ --instrument-mode "{{workflow.parameters.instrument-mode}}" \ --fold-idx {{workflow.parameters.fold-idx}} \ diff --git a/scripts/pre-commit-hook.sh b/scripts/pre-commit-hook.sh index c96ffb8f9..d6f47e849 100755 --- a/scripts/pre-commit-hook.sh +++ b/scripts/pre-commit-hook.sh @@ -202,10 +202,12 @@ check_sp18_consumer_audit() { } # DQN v2 Invariant 9 enforcement: reject TODO/FIXME/XXX/HACK/TBD markers in added code. +# Note: todo!() macro is permitted per CLAUDE.md ("todo!() macro is OK for runtime stubs") +# — only comment-style markers and unimplemented!() are blocked. check_no_todo_fixme() { local added_lines=$(git diff --cached --diff-filter=AM -U0 -- '*.rs' '*.cu' '*.cuh' \ | grep -E '^\+' | grep -vE '^\+\+\+') - local bad=$(echo "$added_lines" | grep -E '(TODO|FIXME|\bXXX\b|\bHACK\b|\bTBD\b|unimplemented!|todo!\()') + local bad=$(echo "$added_lines" | grep -E '(TODO|FIXME|\bXXX\b|\bHACK\b|\bTBD\b|unimplemented!)' | grep -v 'todo!(') if [ -n "$bad" ]; then echo "❌ Invariant 9 violation: deferred-work markers found in staged code" echo "$bad" | sed 's/^/ /'