foxhunt/crates/ml-alpha/examples/alpha_rl_train.rs

// The per-step diag `json!{...}` block now spans ≥ 130 fields after the
// adaptive-clamp additions, exceeding serde_json's default macro recursion
// budget of 128. Bumping at the crate root keeps the expansion within limits.
#![recursion_limit = "256"]

//! Integrated RL trainer CLI (DQN + PPO on shared Mamba2 -> CfC encoder).
//!
//! Drives `IntegratedTrainer::step_with_lobsim` against a real `LobSimCuda`
//! backed by MBP-10 windows loaded via `MultiHorizonLoader::next_sequence_pair`
//! (R2) for true `(s_t, s_{t+1})` adjacency. The trainer combines:
//!   * R3: GPU-resident EMA + advantage/return kernels.
//!   * R4: GPU Thompson + Double-DQN argmax + log π_old kernels.
//!   * R5: 7 RL controllers + target-net soft update.
//!   * R6: GPU-pure env step (extract_realized_pnl_delta, apply_reward_scale,
//!     actions_to_market_targets).
//!   * R7a/R7b: host-lift of every per-step orchestration loop.
//!   * R7c-data: true `h_{t+1}` / `V(s_{t+1})` / Bellman target on `h_{t+1}`.
//!   * R7d: PER push/sample + off-policy DQN with stop-grad on encoder.
//!
//! Emits `alpha_rl_train_summary.json` on exit. Gate G8 (NaN abort): if any
//! per-head loss is non-finite at any step, the binary exits with code 2 so
//! the cluster smoke harness can terminate the workflow on first divergence.
//!
//! Usage:
//!   alpha_rl_train \
//!     --mbp10-data-dir   /data/futures-baseline-mbp10/ES.FUT \
//!     --predecoded-dir   /feature-cache/predecoded \
//!     --out              /feature-cache/alpha-rl-runs/<sha> \
//!     --n-steps          1000 \
//!     --seed             16962 \
//!     --instrument-mode  front-month \
//!     --gpu-idx          0
//!
//! Plan-doc reference: docs/superpowers/plans/2026-05-23-integrated-rl-trainer-rebuild.md
//! Section R8 (CLI binary + Argo template + dispatcher).

use anyhow::{Context, Result};
use clap::Parser;
use data::providers::databento::dbn_parser::InstrumentFilter;
// Use the canonical Action enum size, NOT a literal. Caught 2026-05-24
// during SP20 P4 dogfood: a `0..9` range survived the N_ACTIONS=9→11
// bump and silently dropped HalfFlat samples (a9, a10 showed 0% in
// diag despite P_MIN=0.02 floor guaranteeing 2% each). Per the
// emerging meta-pattern: any constant that mirrors a kernel-side
// structural dimension MUST reference the Rust const, not duplicate
// the literal value.
use ml_alpha::rl::common::N_ACTIONS;
use ml_alpha::data::gpu_dataset::{GpuDataLoader, GpuDataset};
use ml_alpha::data::loader::{
    discover_mbp10_files_sorted, MultiHorizonLoader, MultiHorizonLoaderConfig,
    DEFAULT_OUTCOME_LABEL_COST_ES,
};
use ml_alpha::heads::HORIZONS;
// ISV slot imports — only the slots still read DIRECTLY in this file.
// Phase A of the 2026-05-31 checkpoints+eval-diag plan moved all ~140
// diag-block ISV reads behind `IntegratedTrainer::build_diag_value`, so
// the previous wall of imports collapsed to the four slots still in
// the per-N-step stderr ticker.
use ml_alpha::rl::isv_slots::{
    RL_CONF_GATE_FIRED_COUNT_INDEX, RL_FRD_GATE_FIRED_COUNT_INDEX, RL_GAMMA_INDEX,
    RL_HEAT_CAP_FIRED_COUNT_INDEX, RL_PER_ALPHA_INDEX, RL_PPO_CLIP_INDEX, RL_PYRAMID_ADD_COUNT_INDEX,
    RL_REWARD_SCALE_INDEX,
};
use ml_alpha::trainer::diag_staging::DiagStaging;
use ml_alpha::trainer::integrated::{
    DiagInputs, IntegratedStepStats, IntegratedTrainer, IntegratedTrainerConfig,
};
use ml_alpha::trainer::perception::PerceptionTrainerConfig;
use ml_backtesting::sim::LobSimCuda;
use ml_core::device::MlDevice;
use serde::Serialize;
use std::io::{BufWriter, Write};
use std::path::PathBuf;
use std::process;

#[derive(Parser)]
#[command(name = "alpha_rl_train")]
struct Cli {
    /// Directory containing MBP-10 .dbn.zst files. Per
    /// `feedback_mbp10_mandatory`, this argument is required — no
    /// synthetic-data fallback in the production training path.
    #[arg(long)]
    mbp10_data_dir: PathBuf,

    /// Directory holding predecoded sidecar caches (created on miss).
    #[arg(long)]
    predecoded_dir: PathBuf,

    /// Output directory for summary JSON + per-step log artefacts.
    #[arg(long)]
    out: PathBuf,

    /// Total `step_with_lobsim` calls to run. The validation smoke
    /// (audit prerequisite) ships with 1000; production scale-up runs
    /// 50k+. Each call processes one `(s_t, s_{t+1})` sequence pair
    /// across `n_backtests` parallel LobSim instances.
    #[arg(long, default_value_t = 1000)]
    n_steps: usize,

    /// Encoder sequence length (snapshots per step). Must match
    /// `PerceptionTrainerConfig::seq_len`. Default 32 matches the
    /// validated single-scale `1:32` window from alpha_train.
    #[arg(long, default_value_t = 32)]
    seq_len: usize,

    /// Batch dimension — number of parallel LobSim backtests run in
    /// lockstep per step. Larger batches amortise the encoder forward
    /// cost across more trajectories but push GPU memory pressure
    /// linearly. Default 16 — fixes the b_size=1 signal starvation
    /// per `pearl_b_size_1_signal_starvation_blocks_q_learning`
    /// (Q stayed at uniform baseline ln(21)=3.04 across all b_size=1
    /// smokes regardless of controller tuning; 16 parallel
    /// transitions per Adam step give 16× the gradient signal). Was
    /// 1 (validation-smoke default); production sweep can push to
    /// 32-64 on L40S 48GB.
    #[arg(long, default_value_t = 16)]
    n_backtests: usize,

    /// PER buffer capacity. At b_size=16, capacity/16 = unique steps
    /// retained. 32768/16 = 2048 steps of replay depth.
    #[arg(long, default_value_t = 32768)]
    per_capacity: usize,

    /// Random seed pinning RNG + Xavier init + ISV controller bootstrap
    /// state. Per `pearl_scoped_init_seed_for_reproducibility`.
    #[arg(long, default_value_t = 16962)]
    seed: u64,

    /// MBP-10 instrument filter — `all`, `front-month`, or `id=<N>`.
    /// `front-month` is the canonical setting for ES (auto-detects the
    /// dominant contract per file, handles quarterly roll).
    #[arg(long, default_value = "front-month", value_parser = parse_instrument_mode)]
    instrument_mode: InstrumentFilter,

    /// CUDA device index. Single-GPU dev (RTX 3050 Ti) uses 0; cluster
    /// L40S/H100 also 0 (one GPU per pod per the alpha-rl template).
    #[arg(long, default_value_t = 0)]
    gpu_idx: usize,

    /// Round-trip trade cost in price units for the D-style outcome
    /// labels (consumed by the perception aux head, not by the RL
    /// path directly — but the loader needs it). Default 0.5 = 2 ticks
    /// × $0.25/tick for ES.
    #[arg(long, default_value_t = DEFAULT_OUTCOME_LABEL_COST_ES)]
    outcome_label_cost: f32,

    /// How often (in steps) to flush a progress line to stderr. The
    /// summary JSON is written ONCE at the end; this controls in-flight
    /// visibility only.
    #[arg(long, default_value_t = 100)]
    log_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
    /// blocks; fold k trains on blocks [0..k+1) and evaluates on block
    /// [k+1] (purged walk-forward). At `fold_idx=0` the first block
    /// is the train window, second is eval. At `fold_idx=K-2` the
    /// trainer sees almost all the data and evaluates on the final
    /// block. The dispatcher (`scripts/argo-alpha-rl.sh --n-folds K`)
    /// fans out K-1 workflows, one per fold; the aggregator computes
    /// mean ± SD of the per-fold profit_factor.
    ///
    /// Default 0 + n-folds=1 = single-window mode (no eval split).
    #[arg(long, default_value_t = 0)]
    fold_idx: usize,

    /// Total walk-forward fold count. Set with `--fold-idx` for the
    /// multi-fold G8 gate. n_folds=1 disables the train/eval split
    /// (single-window mode — only meaningful for smoke validation).
    /// n_folds≥3 enables a real evaluation: train on first
    /// `fold_idx + 1` of `n_folds` data blocks, evaluate on the next.
    #[arg(long, default_value_t = 1)]
    n_folds: usize,

    /// Eval-phase step count. After the train phase completes its
    /// `--n-steps`, the eval phase runs `--n-eval-steps` additional
    /// steps on the held-out fold using the same trainer
    /// (acknowledged minor in-eval learning contamination at b_size=1
    /// — pure eval mode is a follow-up). LobSim trade records are
    /// drained at the END so the summary reflects eval-phase trades
    /// only (train-phase trades are flushed when the eval phase
    /// starts).
    ///
    /// 0 disables the eval phase entirely (single-window training
    /// only — the summary's profit_factor will reflect the full run
    /// including train).
    #[arg(long, default_value_t = 0)]
    n_eval_steps: usize,

    /// Per-step diagnostic JSONL path. One record per step capturing:
    ///   * step number, wall time
    ///   * loss components (l_bce/q/pi/v/aux/total) + λs (5)
    ///   * all 7 RL controller outputs (γ τ ε coef n_roll per_α scale)
    ///   * all 5 per-head learning rates (lr_bce/q/pi/v/aux)
    ///   * all 7 EMA inputs the controllers consume
    ///   * replay buffer length
    ///   * per-step reward + done aggregates (DtoH b_size floats per stat)
    ///   * per-step action histogram (counts per action class)
    ///
    /// Default: `<out>/diag.jsonl`. Pass `--diag-jsonl /dev/null` (or
    /// any unwritable path that errors at open) to disable. The dump
    /// happens AFTER each `step_with_lobsim` so it sees the freshly-
    /// adapted ISV values. Writes are line-buffered + flushed every
    /// `log_every` steps so a `tail -f` shows in-flight progress.
    ///
    /// Critically: this provides the per-step diagnostic visibility
    /// the cluster smoke needs to detect anomalies early. Without it,
    /// the smoke is "blind" — only the final summary is observable,
    /// and a controller drifting off-anchor mid-run is invisible until
    /// it produces a non-finite loss (G8 NaN abort).
    #[arg(long)]
    diag_jsonl: Option<PathBuf>,

    /// Per-step diagnostic JSONL path for the EVAL phase (Phase A of
    /// the 2026-05-31 checkpoints+eval-diag plan). Same 643-leaf
    /// schema (642 scalars + 1 boolean) as `--diag-jsonl` — produced by the SAME builder
    /// (`IntegratedTrainer::build_diag_value`), guaranteeing parity
    /// per `feedback_single_source_of_truth_no_duplicates`.
    ///
    /// Motivation: cluster run alpha-rl-8ll7j completed with
    /// +$61,513 final pnl but max_dd -$444,512 during the eval
    /// phase — and the eval phase emitted ZERO per-step diag, so
    /// the drawdown trajectory was invisible. This file unblocks
    /// that diagnostic surface.
    ///
    /// Default: `<out>/eval_diag.jsonl`. Only written when the eval
    /// phase runs (`--n-eval-steps > 0` AND walk-forward enabled).
    #[arg(long)]
    eval_diag_jsonl: Option<PathBuf>,
}

fn parse_instrument_mode(s: &str) -> Result<InstrumentFilter, String> {
    let trimmed = s.trim();
    if trimmed.eq_ignore_ascii_case("all") {
        return Ok(InstrumentFilter::All);
    }
    if trimmed.eq_ignore_ascii_case("front-month")
        || trimmed.eq_ignore_ascii_case("front_month")
    {
        return Ok(InstrumentFilter::FrontMonth);
    }
    if let Some(num) = trimmed.strip_prefix("id=") {
        let id: u32 = num.parse().map_err(|e| {
            format!("instrument-mode: expected `id=<u32>` but failed to parse '{num}': {e}")
        })?;
        return Ok(InstrumentFilter::Id(id));
    }
    Err(format!(
        "instrument-mode: expected `all`, `front-month`, or `id=<N>`; got '{s}'"
    ))
}

#[derive(Serialize, Default)]
struct AlphaRlTrainSummary {
    n_steps_planned: usize,
    n_steps_completed: usize,
    seq_len: usize,
    n_backtests: usize,
    per_capacity: usize,
    seed: u64,

    /// True iff `n_steps_completed == n_steps_planned` AND no NaN
    /// abort fired. False if early-terminated for any reason (NaN,
    /// loader EOF, panic — though panics also exit nonzero).
    completed_clean: bool,

    /// Final-step loss components (raw, not λ-weighted). All NaN if
    /// trainer never returned a finite step.
    final_l_bce: f32,
    final_l_q: f32,
    final_l_pi: f32,
    final_l_v: f32,
    final_l_aux: f32,
    final_l_total: f32,

    /// Replay buffer state at exit.
    final_replay_len: usize,

    /// Step at which the first NaN was observed, if any. -1 means
    /// no NaN observed.
    nan_abort_step: i64,
}

fn main() -> Result<()> {
    tracing_subscriber::fmt()
        .with_env_filter(tracing_subscriber::EnvFilter::from_default_env())
        .init();

    let cli = Cli::parse();

    std::fs::create_dir_all(&cli.out)
        .with_context(|| format!("mkdir -p {}", cli.out.display()))?;
    std::fs::create_dir_all(&cli.predecoded_dir)
        .with_context(|| format!("mkdir -p {}", cli.predecoded_dir.display()))?;

    let dev = MlDevice::cuda(cli.gpu_idx)
        .with_context(|| format!("MlDevice::cuda({})", cli.gpu_idx))?;
    eprintln!("CUDA device {} initialised", cli.gpu_idx);

    // Loader: paired (s_t, s_{t+1}) sequences. inference_only=false so
    // next_sequence_pair is available (the labels it preloads are
    // unused by step_with_lobsim, but the API requires the label
    // precomputation pass per `feedback_no_partial_refactor`).
    let all_files = discover_mbp10_files_sorted(&cli.mbp10_data_dir)
        .with_context(|| format!("discover_mbp10_files_sorted({})", cli.mbp10_data_dir.display()))?;
    anyhow::ensure!(
        !all_files.is_empty(),
        "no MBP-10 .dbn.zst files under {}",
        cli.mbp10_data_dir.display()
    );
    eprintln!("loader: {} MBP-10 files total", all_files.len());

    // Walk-forward fold split per `pearl_single_window_oos_is_not_oos`.
    // Slice the file list into `n_folds` equal blocks. fold k uses
    // blocks [0..=k] for training and block [k+1] for evaluation. With
    // n_folds=1 (default) train_files = all_files and eval_files is
    // empty (single-window mode, no eval phase).
    let (train_files, eval_files) = if cli.n_folds <= 1 || cli.n_eval_steps == 0 {
        eprintln!("walk-forward: disabled (n_folds={}, n_eval_steps={})", cli.n_folds, cli.n_eval_steps);
        (all_files, Vec::new())
    } else {
        anyhow::ensure!(
            cli.fold_idx + 1 < cli.n_folds,
            "fold_idx {} requires fold_idx + 1 < n_folds ({}); the last block must be the eval window",
            cli.fold_idx,
            cli.n_folds
        );
        let n_files = all_files.len();
        let block_size = n_files / cli.n_folds;
        anyhow::ensure!(
            block_size >= 1,
            "not enough MBP-10 files ({}) to split into n_folds={} (need ≥ n_folds files)",
            n_files,
            cli.n_folds
        );
        // Train: blocks 0..=fold_idx, Eval: block fold_idx+1.
        let train_end = (cli.fold_idx + 1) * block_size;
        let eval_end = (cli.fold_idx + 2) * block_size;
        let train: Vec<_> = all_files[..train_end].to_vec();
        let eval: Vec<_> = all_files[train_end..eval_end.min(n_files)].to_vec();
        eprintln!(
            "walk-forward fold {}/{}: train={} files ({}..{}), eval={} files ({}..{})",
            cli.fold_idx, cli.n_folds, train.len(), 0, train_end,
            eval.len(), train_end, eval_end.min(n_files)
        );
        (train, eval)
    };

    let multi_resolution: ml_alpha::data::aggregation::MultiResolutionConfig =
        format!("1:{}", cli.seq_len)
            .parse()
            .with_context(|| format!("multi-resolution `1:{}`", cli.seq_len))?;
    anyhow::ensure!(
        multi_resolution.total_positions() == cli.seq_len,
        "internal: single-scale 1:{} should produce total_positions={} but got {}",
        cli.seq_len,
        cli.seq_len,
        multi_resolution.total_positions()
    );

    let loader_cfg = MultiHorizonLoaderConfig {
        files: train_files,
        predecoded_dir: cli.predecoded_dir.clone(),
        multi_resolution: multi_resolution.clone(),
        horizons: HORIZONS,
        // Soft cap above n_steps × n_backtests × 2 so next_sequence_pair
        // never returns Ok(None) mid-run. Each step samples n_backtests
        // pairs (one per batch slot) to build the B×K snapshot tensor
        // that forward_encoder consumes — per-batch market diversity is
        // what gives the gradient-variance reduction promised by
        // `pearl_b_size_1_signal_starvation_blocks_q_learning`.
        n_max_sequences: cli.n_steps
            .saturating_mul(cli.n_backtests.max(1))
            .saturating_mul(2)
            .max(64),
        seed: cli.seed,
        inference_only: false,
        outcome_label_cost: cli.outcome_label_cost,
        instrument_filter: cli.instrument_mode,
        frd_horizon_ticks: ml_alpha::rl::common::FRD_HORIZON_TICKS,
        frd_bucket_range_sigma: ml_alpha::rl::common::FRD_BUCKET_RANGE_SIGMA,
    };
    let loader = MultiHorizonLoader::new(&loader_cfg).context("MultiHorizonLoader::new")?;

    // ── GPU-resident dataset upload. ────────────────────────────────
    // Upload ALL pre-converted snapshots + FRD labels to GPU memory as
    // a single contiguous device buffer. Per-step data loading becomes
    // pure GPU kernel work (sample + gather) — zero CPU data work.
    let gpu_dataset: GpuDataset = loader
        .upload_to_gpu(&dev, cli.n_backtests.max(1), cli.seed)
        .context("loader.upload_to_gpu")?;
    let mut gpu_data_loader = GpuDataLoader::new(&dev, cli.n_backtests.max(1))
        .context("GpuDataLoader::new")?;
    eprintln!(
        "GPU dataset uploaded: {} snapshots, {} files, {:.1} GB",
        gpu_dataset.total_snapshots,
        gpu_dataset.n_files,
        (gpu_dataset.total_snapshots * 216) as f64 / 1e9,
    );

    // Trainer + LobSim.
    let perception_cfg = PerceptionTrainerConfig {
        seq_len: cli.seq_len,
        n_batch: cli.n_backtests,
        ..PerceptionTrainerConfig::default()
    };
    let trainer_cfg = IntegratedTrainerConfig {
        perception: perception_cfg,
        dqn_seed: cli.seed.wrapping_add(0xDA),
        ppo_seed: cli.seed.wrapping_add(0xBE),
        per_capacity: cli.per_capacity,
        per_seed: cli.seed.wrapping_add(0x9E37_79B9),
    };
    let mut trainer =
        IntegratedTrainer::new(&dev, trainer_cfg).context("IntegratedTrainer::new")?;
    let mut sim = LobSimCuda::new(cli.n_backtests, &dev).context("LobSimCuda::new")?;
    eprintln!(
        "trainer + LobSimCuda initialised: seq_len={} n_backtests={} per_capacity={}",
        cli.seq_len, cli.n_backtests, cli.per_capacity
    );

    let mut summary = AlphaRlTrainSummary {
        n_steps_planned: cli.n_steps,
        seq_len: cli.seq_len,
        n_backtests: cli.n_backtests,
        per_capacity: cli.per_capacity,
        seed: cli.seed,
        nan_abort_step: -1,
        ..Default::default()
    };
    let mut last_stats: Option<IntegratedStepStats> = None;

    // ── Per-step JSONL diag writer. ──────────────────────────────────
    // Default path: `<out>/diag.jsonl`. The smoke run's failure-mode
    // catalogue (NaN abort, controller drift, replay stagnation, reward
    // explosion) is detected via post-hoc inspection of this file —
    // without it, the run is blind beyond the eprintln every-N-step
    // ticks.
    let diag_path = cli
        .diag_jsonl
        .clone()
        .unwrap_or_else(|| cli.out.join("diag.jsonl"));
    let diag_file = std::fs::File::create(&diag_path)
        .with_context(|| format!("create diag jsonl {}", diag_path.display()))?;
    let mut diag = BufWriter::new(diag_file);
    eprintln!("per-step diag JSONL: {}", diag_path.display());

    // Async non-blocking diagnostic staging: separate CUDA stream +
    // double-buffered mapped-pinned memory. ALL per-step device reads
    // (ISV, rewards, dones, actions, raw_rewards, trade_duration,
    // outcome_ema, position_lots, pyramid_count, unit state x4,
    // close_unit_index, frd_logits) use async DtoD copies — zero
    // stream syncs on the training stream. The host reads the PREVIOUS
    // step's buffer while current step copies run on the diag stream.
    let mut diag_staging = DiagStaging::new(&dev, cli.n_backtests)
        .context("DiagStaging::new")?;

    // Per-step scratch (replaced each step via mapped-pinned helpers).

    // ── Training loop. ───────────────────────────────────────────────
    // Windowed action distribution — EMA-smoothed across recent steps
    // so action_entropy is meaningful at b_size=1 (per-step act_hist
    // is one-hot, H=0; windowed dist captures actual policy variety).
    // α = 1/1000 → half-life ≈ 690 steps. Tracks the recent ~1k steps'
    // action distribution; entropy of normalised windowed_act_hist is
    // a real exploration signal.
    // SP20 P4: action histogram width MUST track the Action enum
    // size. Always reference the const (NOT a literal) — see
    // import-site comment for the dogfood-caught literal-drift bug.
    let mut windowed_act_hist: [f32; N_ACTIONS] = [0.0; N_ACTIONS];
    const WINDOWED_ACT_ALPHA: f32 = 1.0 / 1000.0;

    let mut pnl_cum_usd: f64 = 0.0;
    // B-7 observability: pnl_cum_usd uses POST-clamp rewards / scale (un-scales
    // but doesn't un-clamp), so the diag's training pnl signal is truncated to
    // [-clamp_loss, +clamp_win] per close. realized_pnl_cum_usd sums raw_rewards
    // (pre-scale, pre-clamp) — direct USD trade pnl. Divergence between the two
    // exposes the reward-hacking gap the clamp introduces.
    let mut realized_pnl_cum_usd: f64 = 0.0;
    let mut win_count: u64 = 0;
    let mut total_trades: u64 = 0;
    let mut hold_time_sum: f64 = 0.0;
    let mut trail_fired_total: u64 = 0;
    let mut trail_tighten_total: u64 = 0;
    let mut trail_loosen_total: u64 = 0;
    let mut pyramid_added_total: u64 = 0;
    let mut partial_flat_total: u64 = 0;
    let mut partial_flat_long_total: u64 = 0;
    let mut partial_flat_short_total: u64 = 0;
    let mut conf_gate_total: u64 = 0;
    let mut frd_gate_total: u64 = 0;
    let mut heat_cap_total: u64 = 0;

    let t_start = std::time::Instant::now();
    for step in 0..cli.n_steps {
        // GPU-resident data loading: sample_and_gather + gather_next +
        // gather_current + gather_frd_labels all run as GPU kernels
        // inside step_with_lobsim_gpu. Zero CPU data work per step.
        let stats = trainer
            .step_with_lobsim_gpu(&mut gpu_data_loader, &gpu_dataset, &mut sim)
            .with_context(|| format!("step_with_lobsim_gpu at step {step}"))?;

        // Gate G8: NaN abort. Per `feedback_stop_on_anomaly` +
        // `feedback_kill_runs_on_anomaly_quickly` the cluster smoke
        // tail-watcher kills the workflow on non-zero exit; here we
        // signal that with exit code 2 (distinct from clean exit 0
        // and panic-derived exit codes).
        let non_finite = !stats.l_total.is_finite()
            || !stats.l_bce.is_finite()
            || !stats.l_q.is_finite()
            || !stats.l_pi.is_finite()
            || !stats.l_v.is_finite()
            || !stats.l_aux.is_finite();
        if non_finite {
            eprintln!(
                "G8 NaN ABORT at step {step}: l_bce={} l_q={} l_pi={} l_v={} l_aux={} l_total={}",
                stats.l_bce, stats.l_q, stats.l_pi, stats.l_v, stats.l_aux, stats.l_total
            );
            summary.n_steps_completed = step;
            summary.nan_abort_step = step as i64;
            summary.completed_clean = false;
            write_summary(&cli.out, &summary)?;
            process::exit(2);
        }

        // Event-sync the diag snapshot and swap buffers. Previous step's
        // async DtoD copies complete here. Placed AFTER step_with_lobsim
        // so the diag copies have had the entire training step to finish —
        // the event sync is instant (~0 us). On step 0 this is a no-op
        // (event sync on a freshly-created event returns immediately).
        diag_staging.sync_and_swap().context("diag sync_and_swap")?;

        // Launch async DtoD copies of ALL diag buffers into the current
        // staging buffer. Non-blocking on the training stream — the copies
        // run on diag_staging's separate stream.
        diag_staging
            .snapshot_async(
                trainer.isv_dev_ptr,
                trainer.rewards_d.raw_ptr(),
                trainer.dones_d.raw_ptr(),
                trainer.actions_d.raw_ptr(),
                trainer.raw_rewards_d.raw_ptr(),
                trainer.trade_duration_emit_d.raw_ptr(),
                trainer.outcome_ema_d.raw_ptr(),
                trainer.prev_position_lots_d.raw_ptr(),
                trainer.pyramid_units_count_d.raw_ptr(),
                trainer.unit_entry_price_d.raw_ptr(),
                trainer.unit_entry_step_d.raw_ptr(),
                trainer.unit_lots_d.raw_ptr(),
                trainer.unit_trail_distance_d.raw_ptr(),
                trainer.close_unit_index_d.raw_ptr(),
                trainer.frd_logits_d.raw_ptr(),
            )
            .context("diag snapshot_async")?;

        // ── Per-step diag dump (async staging reads). ────────────────
        // ALL diag reads come from the DiagStaging double-buffer — zero
        // GPU stalls on the training stream. On step 0 these report
        // zeros (no prior snapshot); from step 1+ they report step N-1's
        // data (one-step delay, acceptable for diagnostics).
        let rewards_host = diag_staging.read_rewards();
        let dones_host = diag_staging.read_dones();
        let actions_raw = diag_staging.read_actions_raw();
        let actions_host: Vec<i32> = actions_raw
            .iter()
            .map(|f| f.to_bits() as i32)
            .collect();
        let raw_rewards_host = diag_staging.read_raw_rewards();
        let trade_duration_host = diag_staging.read_trade_duration();
        let outcome_ema_host = diag_staging.read_outcome_ema();
        let position_lots_host = diag_staging.read_position_lots();
        let pyramid_count_host = diag_staging.read_pyramid_count();
        let unit_entry_price_host = diag_staging.read_unit_entry_price();
        let unit_entry_step_host = diag_staging.read_unit_entry_step();
        let unit_lots_host = diag_staging.read_unit_lots();
        let unit_trail_host = diag_staging.read_unit_trail();
        let close_unit_index_host = diag_staging.read_close_unit_index();

        // Per-step action histogram (counts) — still computed here because
        // the windowed EMA and several per-step counters (tighten/loosen/
        // half-flat) read directly from it. `build_diag_value` recomputes
        // its own act_hist internally — the two are bit-equivalent.
        let mut act_hist = [0u32; N_ACTIONS];
        for &a in &actions_host {
            if (a as usize) < N_ACTIONS {
                act_hist[a as usize] += 1;
            }
        }
        // EMA-smoothed windowed action distribution (cross-step state).
        // Mutated here so the next step inherits the updated EMA;
        // build_diag_value reads the post-update value via `windowed_act_hist`
        // in DiagInputs.
        let total_actions: u32 = act_hist.iter().sum();
        if total_actions > 0 {
            for i in 0..N_ACTIONS {
                let p_step = (act_hist[i] as f32) / (total_actions as f32);
                windowed_act_hist[i] = (1.0 - WINDOWED_ACT_ALPHA)
                    * windowed_act_hist[i]
                    + WINDOWED_ACT_ALPHA * p_step;
            }
        }
        let reward_sum: f32 = rewards_host.iter().sum();
        let done_count: u32 = dones_host.iter().map(|&d| if d > 0.5 { 1 } else { 0 }).sum();

        // Per-trade cumulative stats (surfer validation) — mutate the
        // train-loop running counters (pnl_cum_usd / total_trades /
        // win_count / hold_time_sum). build_diag_value receives the
        // post-update values via DiagInputs.
        let current_scale = trainer.read_isv_host(RL_REWARD_SCALE_INDEX);
        for b in 0..cli.n_backtests {
            if dones_host[b] > 0.5 {
                let pnl_usd = if current_scale > 1e-9 {
                    rewards_host[b] as f64 / current_scale as f64
                } else {
                    rewards_host[b] as f64
                };
                pnl_cum_usd += pnl_usd;
                // B-7 observability: raw_rewards is the shaped pnl in USD
                // BEFORE scale and clamp (see rl_fused_reward_pipeline.cu).
                // Direct trade-pnl signal — compare against pnl_cum_usd above
                // to surface clamp-truncated tail losses.
                realized_pnl_cum_usd += raw_rewards_host[b] as f64;
                total_trades += 1;
                if pnl_usd > 0.0 { win_count += 1; }
                hold_time_sum += trade_duration_host[b] as f64;
            }
        }
        let win_rate = if total_trades > 0 { win_count as f64 / total_trades as f64 } else { 0.0 };

        // Snapshot the ISV slots we read on the host side. Determinism
        // Phase 1 (2026-06-02) turned `build_diag_value` into `&mut self`
        // (needed to launch the checksum kernels), so the `isv` slice
        // can no longer co-exist with the trainer mutable borrow.
        // Copying the eight specific values we use is cheaper than
        // re-grabbing the slice twice and keeps the diff local.
        let isv_trail_fired_step    = trainer.read_isv_host(RL_CONF_GATE_FIRED_COUNT_INDEX) as u64;
        let isv_pyramid_add_step    = trainer.read_isv_host(RL_PYRAMID_ADD_COUNT_INDEX)    as u64;
        let isv_conf_gate_step      = trainer.read_isv_host(RL_CONF_GATE_FIRED_COUNT_INDEX) as u64;
        let isv_frd_gate_step       = trainer.read_isv_host(RL_FRD_GATE_FIRED_COUNT_INDEX)  as u64;
        let isv_heat_cap_step       = trainer.read_isv_host(RL_HEAT_CAP_FIRED_COUNT_INDEX)  as u64;
        let isv_gamma               = trainer.read_isv_host(RL_GAMMA_INDEX);
        let isv_ppo_clip            = trainer.read_isv_host(RL_PPO_CLIP_INDEX);
        let isv_per_alpha           = trainer.read_isv_host(RL_PER_ALPHA_INDEX);
        let isv_reward_scale        = trainer.read_isv_host(RL_REWARD_SCALE_INDEX);

        // Per-step counters derived from action histogram + ISV diag slots.
        let trail_fired_step = isv_trail_fired_step; // reusing — trail stop uses actions override
        let tighten_step = act_hist[ml_alpha::rl::common::Action::TrailTighten as usize] as u64;
        let loosen_step = act_hist[ml_alpha::rl::common::Action::TrailLoosen as usize] as u64;
        let pyramid_add_step = isv_pyramid_add_step;
        let half_flat_long_step = act_hist[ml_alpha::rl::common::Action::HalfFlatLong as usize] as u64;
        let half_flat_short_step = act_hist[ml_alpha::rl::common::Action::HalfFlatShort as usize] as u64;
        let partial_flat_step = half_flat_long_step + half_flat_short_step;
        let conf_gate_step = isv_conf_gate_step;
        let frd_gate_step = isv_frd_gate_step;
        let heat_cap_step = isv_heat_cap_step;

        trail_fired_total += trail_fired_step;
        trail_tighten_total += tighten_step;
        trail_loosen_total += loosen_step;
        pyramid_added_total += pyramid_add_step;
        partial_flat_total += partial_flat_step;
        partial_flat_long_total += half_flat_long_step;
        partial_flat_short_total += half_flat_short_step;
        conf_gate_total += conf_gate_step;
        frd_gate_total += frd_gate_step;
        heat_cap_total += heat_cap_step;

        // Build the per-step diag record via the single-source-of-truth
        // builder on the trainer (Phase A of the 2026-05-31
        // checkpoints+eval-diag plan). Same builder is invoked from the
        // eval phase below, guaranteeing schema parity per
        // `feedback_single_source_of_truth_no_duplicates`.
        let frd_logits_host = diag_staging.read_frd_logits();
        let diag_inputs = DiagInputs {
            b_size: cli.n_backtests,
            stats: &stats,
            rewards: rewards_host,
            dones: dones_host,
            actions: &actions_host,
            raw_rewards: raw_rewards_host,
            trade_duration: trade_duration_host,
            outcome_ema: outcome_ema_host,
            position_lots: &position_lots_host,
            pyramid_count: &pyramid_count_host,
            unit_entry_price: unit_entry_price_host,
            unit_entry_step: &unit_entry_step_host,
            unit_lots: &unit_lots_host,
            unit_trail: unit_trail_host,
            close_unit_index: &close_unit_index_host,
            frd_logits: frd_logits_host,
            pnl_cum_usd,
            realized_pnl_cum_usd,
            total_trades,
            win_count,
            hold_time_sum,
            trail_fired_total,
            trail_tighten_total,
            trail_loosen_total,
            pyramid_added_total,
            partial_flat_total,
            partial_flat_long_total,
            partial_flat_short_total,
            conf_gate_total,
            frd_gate_total,
            heat_cap_total,
            windowed_act_hist: &windowed_act_hist,
        };
        let record = trainer.build_diag_value(
            step as u64,
            t_start.elapsed().as_secs_f32(),
            &diag_inputs,
        ).context("build_diag_value (train)")?;
        writeln!(diag, "{}", record).context("diag: writeln jsonl")?;

        if step % cli.log_every == 0 || step == cli.n_steps - 1 {
            diag.flush().context("diag: flush")?;
            let elapsed = t_start.elapsed().as_secs_f32();
            let sps = if elapsed > 0.0 { (step + 1) as f32 / elapsed } else { 0.0 };
            eprintln!(
                "step {:>6}/{}: l_q={:.4} l_pi={:.4} l_v={:.4} l_total={:.4} \
                 γ={:.4} ε={:.4} per_α={:.4} scale={:.4} \
                 replay={} dones={} rew_sum={:.3} sps={:.0} pnl=${:.0} wr={:.2} elapsed={:.1}s",
                step,
                cli.n_steps,
                stats.l_q,
                stats.l_pi,
                stats.l_v,
                stats.l_total,
                isv_gamma,
                isv_ppo_clip,
                isv_per_alpha,
                isv_reward_scale,
                trainer.gpu_replay.capacity.min(step + 1),
                done_count,
                reward_sum,
                sps,
                pnl_cum_usd,
                win_rate,
                elapsed,
            );
        }
        last_stats = Some(stats);
        summary.n_steps_completed = step + 1;
    }
    diag.flush().context("diag: final flush")?;

    // ── Eval phase (walk-forward G8). ────────────────────────────────
    // After the train phase, run n_eval_steps additional step_with_lobsim
    // calls on the held-out eval files. Trade records accumulated by
    // LobSim during eval are isolated from train-phase records via the
    // `head_before_eval` checkpoint — eval records are sliced as
    // `all_records[head_before_eval..]` post-eval.
    //
    // Note: this is NOT pure-eval mode — step_with_lobsim still runs
    // its learning machinery (Adam steps, PER updates). At b_size=1
    // the per-step learning effect is small relative to the policy
    // already accumulated in train; the eval profit_factor approximates
    // the train-end policy's OOS performance. True pure-eval (forward
    // only, no backward) is a follow-up architectural change.
    if cli.n_eval_steps > 0 && !eval_files.is_empty() {
        // Adaptive risk management (spec 2026-05-30): reset Layer 1
        // session-level state on train→eval transition. Clears session
        // pnl, DD-triggered flag, consec-loss counter, cooldown counter.
        // Kelly + inventory EMAs intentionally preserved.
        trainer
            .reset_session_state()
            .context("reset_session_state at train→eval boundary")?;

        let head_before_per_b = sim
            .read_per_backtest_trade_counts()
            .context("snapshot per-backtest trade-count pre-eval")?;
        let head_before_min = head_before_per_b.iter().min().copied().unwrap_or(0);
        let head_before_max = head_before_per_b.iter().max().copied().unwrap_or(0);
        let head_before_sum: u64 =
            head_before_per_b.iter().map(|&h| h as u64).sum();
        eprintln!(
            "── eval phase: {} steps on {} held-out files; pre-eval head per b_size={} \
             accounts: min={} max={} sum={} ──",
            cli.n_eval_steps,
            eval_files.len(),
            head_before_per_b.len(),
            head_before_min,
            head_before_max,
            head_before_sum
        );

        let eval_loader_cfg = MultiHorizonLoaderConfig {
            files: eval_files.clone(),
            predecoded_dir: cli.predecoded_dir.clone(),
            multi_resolution,
            horizons: HORIZONS,
            n_max_sequences: cli.n_eval_steps
                .saturating_mul(cli.n_backtests.max(1))
                .saturating_mul(2)
                .max(64),
            // Different seed for eval so sequence sampling is independent
            // of the train seed (deterministic given the eval seed).
            seed: cli.seed.wrapping_add(0xE7AE),
            inference_only: false,
            outcome_label_cost: cli.outcome_label_cost,
            instrument_filter: cli.instrument_mode,
            frd_horizon_ticks: ml_alpha::rl::common::FRD_HORIZON_TICKS,
            frd_bucket_range_sigma: ml_alpha::rl::common::FRD_BUCKET_RANGE_SIGMA,
        };
        let eval_loader = MultiHorizonLoader::new(&eval_loader_cfg)
            .context("MultiHorizonLoader::new (eval)")?;

        // Upload eval dataset to GPU (separate from train dataset).
        let eval_gpu_dataset = eval_loader
            .upload_to_gpu(&dev, cli.n_backtests.max(1), cli.seed.wrapping_add(0xE7AE))
            .context("eval loader.upload_to_gpu")?;
        let mut eval_gpu_loader = GpuDataLoader::new(&dev, cli.n_backtests.max(1))
            .context("GpuDataLoader::new (eval)")?;

        // ── Per-step eval-phase diag writer (Phase A of the 2026-05-31
        // checkpoints+eval-diag plan). Same 643-leaf schema (642 scalars
        // + 1 boolean) as train's diag.jsonl — emitted by the SAME builder.
        // The eval-phase
        // trajectory was previously invisible (alpha-rl-8ll7j: +$61k
        // pnl, -$444k drawdown, zero per-step records).
        let eval_diag_path = cli
            .eval_diag_jsonl
            .clone()
            .unwrap_or_else(|| cli.out.join("eval_diag.jsonl"));
        let eval_diag_file = std::fs::File::create(&eval_diag_path)
            .with_context(|| format!("create eval diag jsonl {}", eval_diag_path.display()))?;
        let mut eval_diag = BufWriter::new(eval_diag_file);
        eprintln!("per-step eval diag JSONL: {}", eval_diag_path.display());

        // ── Eval-phase running counters (independent of train counters)
        // ───────────────────────────────────────────────────────────────
        // reset_session_state above clears device-side Layer 1 state but
        // does NOT zero the host-side counters maintained by THIS loop;
        // they reflect the eval window only (matches eval_summary.json's
        // post-checkpoint trade slicing).
        let mut eval_windowed_act_hist: [f32; N_ACTIONS] = [0.0; N_ACTIONS];
        let mut eval_pnl_cum_usd: f64 = 0.0;
        // B-7 observability: mirror of realized_pnl_cum_usd for the eval
        // window. See declaration in train loop for rationale.
        let mut eval_realized_pnl_cum_usd: f64 = 0.0;
        let mut eval_win_count: u64 = 0;
        let mut eval_total_trades: u64 = 0;
        let mut eval_hold_time_sum: f64 = 0.0;
        let mut eval_trail_fired_total: u64 = 0;
        let mut eval_trail_tighten_total: u64 = 0;
        let mut eval_trail_loosen_total: u64 = 0;
        let mut eval_pyramid_added_total: u64 = 0;
        let mut eval_partial_flat_total: u64 = 0;
        let mut eval_partial_flat_long_total: u64 = 0;
        let mut eval_partial_flat_short_total: u64 = 0;
        let mut eval_conf_gate_total: u64 = 0;
        let mut eval_frd_gate_total: u64 = 0;
        let mut eval_heat_cap_total: u64 = 0;

        for eval_step in 0..cli.n_eval_steps {
            let stats = trainer
                .step_with_lobsim_gpu(&mut eval_gpu_loader, &eval_gpu_dataset, &mut sim)
                .with_context(|| format!("step_with_lobsim_gpu eval step {eval_step}"))?;
            if !stats.l_total.is_finite() {
                eprintln!("G8 NaN ABORT during eval at step {eval_step}");
                process::exit(2);
            }

            // Mirror train-phase diag pipeline: sync prior staging copies,
            // then enqueue async DtoD copies of THIS step's device buffers.
            // Per `feedback_no_partial_refactor` the eval path uses the
            // SAME staging machinery as train (separate diag stream, no
            // sync on the training stream, one-step latency).
            diag_staging.sync_and_swap().context("eval diag sync_and_swap")?;
            diag_staging
                .snapshot_async(
                    trainer.isv_dev_ptr,
                    trainer.rewards_d.raw_ptr(),
                    trainer.dones_d.raw_ptr(),
                    trainer.actions_d.raw_ptr(),
                    trainer.raw_rewards_d.raw_ptr(),
                    trainer.trade_duration_emit_d.raw_ptr(),
                    trainer.outcome_ema_d.raw_ptr(),
                    trainer.prev_position_lots_d.raw_ptr(),
                    trainer.pyramid_units_count_d.raw_ptr(),
                    trainer.unit_entry_price_d.raw_ptr(),
                    trainer.unit_entry_step_d.raw_ptr(),
                    trainer.unit_lots_d.raw_ptr(),
                    trainer.unit_trail_distance_d.raw_ptr(),
                    trainer.close_unit_index_d.raw_ptr(),
                    trainer.frd_logits_d.raw_ptr(),
                )
                .context("eval diag snapshot_async")?;

            // Host-side reads from the (just-swapped) staging buffer.
            let rewards_host = diag_staging.read_rewards();
            let dones_host = diag_staging.read_dones();
            let actions_raw = diag_staging.read_actions_raw();
            let actions_host: Vec<i32> = actions_raw
                .iter()
                .map(|f| f.to_bits() as i32)
                .collect();
            let raw_rewards_host = diag_staging.read_raw_rewards();
            let trade_duration_host = diag_staging.read_trade_duration();
            let outcome_ema_host = diag_staging.read_outcome_ema();
            let position_lots_host = diag_staging.read_position_lots();
            let pyramid_count_host = diag_staging.read_pyramid_count();
            let unit_entry_price_host = diag_staging.read_unit_entry_price();
            let unit_entry_step_host = diag_staging.read_unit_entry_step();
            let unit_lots_host = diag_staging.read_unit_lots();
            let unit_trail_host = diag_staging.read_unit_trail();
            let close_unit_index_host = diag_staging.read_close_unit_index();
            let frd_logits_host = diag_staging.read_frd_logits();

            // Per-step action histogram (mirrors train pipeline — needed
            // for windowed-EMA + per-counter updates below).
            let mut act_hist = [0u32; N_ACTIONS];
            for &a in &actions_host {
                if (a as usize) < N_ACTIONS {
                    act_hist[a as usize] += 1;
                }
            }
            let total_actions: u32 = act_hist.iter().sum();
            if total_actions > 0 {
                for i in 0..N_ACTIONS {
                    let p_step = (act_hist[i] as f32) / (total_actions as f32);
                    eval_windowed_act_hist[i] = (1.0 - WINDOWED_ACT_ALPHA)
                        * eval_windowed_act_hist[i]
                        + WINDOWED_ACT_ALPHA * p_step;
                }
            }

            // Per-trade cumulative stats — same scale-recovery as train.
            let current_scale = trainer.read_isv_host(RL_REWARD_SCALE_INDEX);
            for b in 0..cli.n_backtests {
                if dones_host[b] > 0.5 {
                    let pnl_usd = if current_scale > 1e-9 {
                        rewards_host[b] as f64 / current_scale as f64
                    } else {
                        rewards_host[b] as f64
                    };
                    eval_pnl_cum_usd += pnl_usd;
                    // B-7 observability: raw pnl (pre-scale, pre-clamp) — see
                    // train-loop counterpart.
                    eval_realized_pnl_cum_usd += raw_rewards_host[b] as f64;
                    eval_total_trades += 1;
                    if pnl_usd > 0.0 { eval_win_count += 1; }
                    eval_hold_time_sum += trade_duration_host[b] as f64;
                }
            }

            let isv = trainer.isv_host_slice();
            // Per-step ISV-derived counters — same indices as train loop.
            let trail_fired_step = isv[RL_CONF_GATE_FIRED_COUNT_INDEX] as u64;
            let tighten_step = act_hist[ml_alpha::rl::common::Action::TrailTighten as usize] as u64;
            let loosen_step = act_hist[ml_alpha::rl::common::Action::TrailLoosen as usize] as u64;
            let pyramid_add_step = isv[RL_PYRAMID_ADD_COUNT_INDEX] as u64;
            let half_flat_long_step = act_hist[ml_alpha::rl::common::Action::HalfFlatLong as usize] as u64;
            let half_flat_short_step = act_hist[ml_alpha::rl::common::Action::HalfFlatShort as usize] as u64;
            let partial_flat_step = half_flat_long_step + half_flat_short_step;
            let conf_gate_step = isv[RL_CONF_GATE_FIRED_COUNT_INDEX] as u64;
            let frd_gate_step = isv[RL_FRD_GATE_FIRED_COUNT_INDEX] as u64;
            let heat_cap_step = isv[RL_HEAT_CAP_FIRED_COUNT_INDEX] as u64;

            eval_trail_fired_total += trail_fired_step;
            eval_trail_tighten_total += tighten_step;
            eval_trail_loosen_total += loosen_step;
            eval_pyramid_added_total += pyramid_add_step;
            eval_partial_flat_total += partial_flat_step;
            eval_partial_flat_long_total += half_flat_long_step;
            eval_partial_flat_short_total += half_flat_short_step;
            eval_conf_gate_total += conf_gate_step;
            eval_frd_gate_total += frd_gate_step;
            eval_heat_cap_total += heat_cap_step;

            // Build inputs and emit the per-step record. Same builder as
            // train → schema parity is guaranteed by construction.
            let diag_inputs = DiagInputs {
                b_size: cli.n_backtests,
                stats: &stats,
                rewards: rewards_host,
                dones: dones_host,
                actions: &actions_host,
                raw_rewards: raw_rewards_host,
                trade_duration: trade_duration_host,
                outcome_ema: outcome_ema_host,
                position_lots: &position_lots_host,
                pyramid_count: &pyramid_count_host,
                unit_entry_price: unit_entry_price_host,
                unit_entry_step: &unit_entry_step_host,
                unit_lots: &unit_lots_host,
                unit_trail: unit_trail_host,
                close_unit_index: &close_unit_index_host,
                frd_logits: frd_logits_host,
                pnl_cum_usd: eval_pnl_cum_usd,
                realized_pnl_cum_usd: eval_realized_pnl_cum_usd,
                total_trades: eval_total_trades,
                win_count: eval_win_count,
                hold_time_sum: eval_hold_time_sum,
                trail_fired_total: eval_trail_fired_total,
                trail_tighten_total: eval_trail_tighten_total,
                trail_loosen_total: eval_trail_loosen_total,
                pyramid_added_total: eval_pyramid_added_total,
                partial_flat_total: eval_partial_flat_total,
                partial_flat_long_total: eval_partial_flat_long_total,
                partial_flat_short_total: eval_partial_flat_short_total,
                conf_gate_total: eval_conf_gate_total,
                frd_gate_total: eval_frd_gate_total,
                heat_cap_total: eval_heat_cap_total,
                windowed_act_hist: &eval_windowed_act_hist,
            };
            // Eval-phase records use a separate step namespace: continue
            // the train-phase step index (`cli.n_steps + eval_step`) so
            // post-hoc tooling can concatenate train + eval JSONL and
            // get a monotone step axis.
            let record = trainer.build_diag_value(
                (cli.n_steps + eval_step) as u64,
                t_start.elapsed().as_secs_f32(),
                &diag_inputs,
            ).context("build_diag_value (eval)")?;
            writeln!(eval_diag, "{}", record).context("eval diag: writeln jsonl")?;

            if eval_step % cli.log_every == 0 || eval_step == cli.n_eval_steps - 1 {
                eval_diag.flush().context("eval diag: flush")?;
                eprintln!(
                    "eval {:>5}/{}: l_total={:.4} elapsed={:.1}s",
                    eval_step, cli.n_eval_steps, stats.l_total,
                    t_start.elapsed().as_secs_f32()
                );
            }
        }
        eval_diag.flush().context("eval diag: final flush")?;

        // Aggregate eval-phase trades across ALL b_size accounts, with
        // correct per-account head_before slicing. Replaces the previous
        // single-account read of backtest 0 + aggregate-vs-per-account
        // scale mismatch (see spec
        // docs/superpowers/specs/2026-05-31-eval-summary-trade-aggregation-design.md
        // for the root-cause analysis).
        let all_per_b = sim
            .read_trade_records_all()
            .context("read all backtests' trade records post-eval")?;
        let head_after_per_b = sim
            .read_per_backtest_trade_counts()
            .context("snapshot per-backtest trade-count post-eval")?;

        let cap_u32 = ml_backtesting::lob::TRADE_LOG_CAP as u32;
        let mut eval_records: Vec<ml_backtesting::order::TradeRecord> = Vec::new();
        let mut n_eval_trades_seen: u64 = 0;
        let mut n_eval_trades_dropped: u64 = 0;
        let mut n_pre_eval_wrapped: u64 = 0;

        for (b, records) in all_per_b.iter().enumerate() {
            let head_before = head_before_per_b[b];
            let head_after = head_after_per_b[b];
            let eval_count_total = head_after.saturating_sub(head_before);
            n_eval_trades_seen += eval_count_total as u64;

            // The ring's contents cover cumulative-stream indices
            // [max(0, head_after - cap), head_after).
            let ring_start_stream_idx = head_after.saturating_sub(cap_u32);

            if head_before < ring_start_stream_idx {
                // Some pre-eval trades had wrapped out before eval started —
                // diagnostic only, doesn't affect eval slicing.
                n_pre_eval_wrapped += (ring_start_stream_idx - head_before) as u64;
            }

            if eval_count_total > cap_u32 {
                // Eval-phase trades wrapped (lost). Should be 0 with
                // TRADE_LOG_CAP=4096 at typical cluster scale.
                n_eval_trades_dropped += (eval_count_total - cap_u32) as u64;
            }

            // Slice the ring contents to eval-only.
            // ring index of first eval trade = head_before − ring_start_stream_idx
            // (clamped at 0 if all pre-eval already wrapped out).
            let eval_start_in_ring =
                head_before.saturating_sub(ring_start_stream_idx) as usize;
            if eval_start_in_ring < records.len() {
                eval_records.extend_from_slice(&records[eval_start_in_ring..]);
            }
        }

        if n_eval_trades_dropped > 0 {
            eprintln!(
                "warning: {} eval trades wrapped out across {} accounts \
                 (TRADE_LOG_CAP={}); summary based on {} captured eval trades",
                n_eval_trades_dropped,
                all_per_b.len(),
                cap_u32,
                eval_records.len()
            );
        }
        if n_pre_eval_wrapped > 0 {
            eprintln!(
                "info: {} pre-eval trades had already wrapped before eval phase \
                 (no effect on eval summary)",
                n_pre_eval_wrapped
            );
        }
        eprintln!(
            "eval phase trade accounting: n_eval_trades_seen={} n_captured={} \
             n_dropped={} b_size={}",
            n_eval_trades_seen,
            eval_records.len(),
            n_eval_trades_dropped,
            all_per_b.len()
        );

        // Synthesise a pnl_curve from per-trade cumulative PnL for
        // compute_summary's max_drawdown calc.
        let mut pnl_curve = Vec::with_capacity(eval_records.len());
        let mut cum = 0.0f32;
        for r in &eval_records {
            cum += (r.realised_pnl_usd_fp as f32) / 100.0;
            pnl_curve.push(cum);
        }
        let eval_summary =
            ml_backtesting::artifacts::compute_summary(&eval_records, &pnl_curve);

        eprintln!(
            "eval summary: n_trades={} pnl_usd={:.2} pf={:.3} sharpe_ann={:.3} \
             max_dd_usd={:.2} win_rate={:.3} | seen={} dropped={} b={}",
            eval_summary.n_trades,
            eval_summary.total_pnl_usd,
            eval_summary.profit_factor,
            eval_summary.sharpe_ann,
            eval_summary.max_drawdown_usd,
            eval_summary.win_rate,
            n_eval_trades_seen,
            n_eval_trades_dropped,
            all_per_b.len(),
        );

        // Write eval_summary.json with the existing compute_summary fields
        // PLUS four new aggregation-aware fields:
        //   n_eval_trades_seen          — true total cumulative dones across b_size
        //   n_eval_trades_dropped       — eval trades lost to ring wrap (0 at typical scale)
        //   n_pre_eval_trades_wrapped   — pre-eval trades wrapped before eval (diagnostic)
        //   b_size                      — context for downstream interpretation
        let aggregated_summary = serde_json::json!({
            "n_trades":                   eval_summary.n_trades,
            "total_pnl_usd":              eval_summary.total_pnl_usd,
            "profit_factor":              eval_summary.profit_factor,
            "sharpe_ann":                 eval_summary.sharpe_ann,
            "max_drawdown_usd":           eval_summary.max_drawdown_usd,
            "win_rate":                   eval_summary.win_rate,
            "n_eval_trades_seen":         n_eval_trades_seen,
            "n_eval_trades_dropped":      n_eval_trades_dropped,
            "n_pre_eval_trades_wrapped":  n_pre_eval_wrapped,
            "b_size":                     cli.n_backtests,
        });

        let eval_summary_path = cli.out.join("eval_summary.json");
        let f = std::fs::File::create(&eval_summary_path)
            .with_context(|| format!("create {}", eval_summary_path.display()))?;
        serde_json::to_writer_pretty(f, &aggregated_summary)
            .with_context(|| format!("write {}", eval_summary_path.display()))?;
        eprintln!("eval summary written: {}", eval_summary_path.display());
    }

    // ── Compose + emit summary. ──────────────────────────────────────
    if let Some(s) = last_stats {
        summary.final_l_bce = s.l_bce;
        summary.final_l_q = s.l_q;
        summary.final_l_pi = s.l_pi;
        summary.final_l_v = s.l_v;
        summary.final_l_aux = s.l_aux;
        summary.final_l_total = s.l_total;
    }
    summary.final_replay_len = trainer.gpu_replay.capacity.min(cli.n_steps);
    summary.completed_clean = summary.n_steps_completed == summary.n_steps_planned;
    write_summary(&cli.out, &summary)?;

    eprintln!(
        "alpha_rl_train complete: {} / {} steps in {:.1}s (replay len = {})",
        summary.n_steps_completed,
        summary.n_steps_planned,
        t_start.elapsed().as_secs_f32(),
        summary.final_replay_len
    );
    Ok(())
}

fn write_summary(out_dir: &std::path::Path, summary: &AlphaRlTrainSummary) -> Result<()> {
    let path = out_dir.join("alpha_rl_train_summary.json");
    let f = std::fs::File::create(&path)
        .with_context(|| format!("create {}", path.display()))?;
    serde_json::to_writer_pretty(f, summary)
        .with_context(|| format!("write {}", path.display()))?;
    eprintln!("summary written: {}", path.display());
    Ok(())
}