ethercat-linux/src/main.rs

//! J1900 <-> TwinCAT 1 kHz communication verification over an EL6695 bridge.
//!
//! SECONDARY-SIDE build: J1900 talks to the EL6695 secondary port. The
//! secondary side SM layout (probed from EEPROM) uses SM0/TX (0x1000, 1024 B,
//! ctrl=0x26) for J1900->TwinCAT data and SM1/RX (0x1600, 1024 B, ctrl=0x22)
//! for TwinCAT->J1900 data. Unlike the primary side there is NO external
//! timestamp object (0x1A02) — the data is raw 1024 B payload with no prefix.
//!
//! Two modes:
//! - `ticker` (default): echo TwinCAT's payload back on SM0 with the local
//!   cycle number stamped at bytes 8..16.
//! - `probe`: read-only; dump the first bytes of SM1 every N cycles and do
//!   NOT write to SM0 (avoids disturbing TwinCAT's view).
//!
//! Graceful SIGINT/SIGTERM shutdown with full statistics.

use std::sync::atomic::{AtomicBool, Ordering};
use std::time::{Duration, Instant};
use std::os::unix::io::AsRawFd;

use el6695_rt::{bridge, regs, rt, stats};
use ethercrab::{
    std::{ethercat_now, tx_rx_task, tx_rx_task_io_uring},
    MainDevice, MainDeviceConfig, PduStorage, Timeouts,
};
use stats::OnlineStats;

const MAX_SUBDEVICES: usize = 16;
const MAX_PDU_DATA: usize = 1100;
const MAX_FRAMES: usize = 16;
const PDI_LEN: usize = 3072;

// Secondary side SM layout (ESI-defined process data SMs)
const SM_TX_ADDR: u16 = 0x1C00;  // SM2: outputs (J1900 writes data to TwinCAT)
const SM_TX_LEN: usize = 800;
const SM_RX_ADDR: u16 = 0x8E00;  // SM3: inputs (J1900 reads data from TwinCAT)
// Full PDO: 2B diagnostic word + 8B payload (single-entry twin_layout blob
// expands 0x6000:01 to 8 bytes).
const SM_RX_LEN: usize = 800;   // SM3 buffer (no diag word, pure 8B payload)
const SM_RX_PLD_LEN: usize = 800; // actual payload after stripping diagnostic word
const RX_PREFIX: usize = 0;      // no diag word prefix (0x1C13=[0x1A08] only)

const PAYLOAD_LEN: usize = 800;

/// Abort after this many consecutive cycles with a TX or RX error.
const MAX_ERR_STREAK: u64 = 1000;

static STOP: AtomicBool = AtomicBool::new(false);
static PDU_STORAGE: std::sync::OnceLock<PduStorage<MAX_FRAMES, MAX_PDU_DATA>> =
    std::sync::OnceLock::new();

extern "C" fn on_signal(_sig: libc::c_int) {
    STOP.store(true, Ordering::Relaxed);
}

#[derive(Clone, Copy, PartialEq)]
enum Mode {
    Ticker,
    Probe,
    Verify,  // CoE audit: read 0xF008, layout hash, SM, identity
    /// RFM ping: periodic reflective-memory write, wait for remote echo,
    /// measure round-trip latency (same-clock RTT).
    RfmPing,
    /// RFM echo: poll REQ mailbox, copy to REP mailbox (remote ping peer).
    RfmEcho,
    /// RFM bridge: EL6695 RX -> RFM -> remote echo -> RFM -> EL6695 TX.
    /// End-to-end CX<->x86 latency via reflective memory hop.
    RfmBridge,
    /// One-shot hardware ring-continuity check (prints ring=OK/BROKEN, exits).
    RfmRing,
}

#[derive(Clone, Copy, Debug, PartialEq)]
enum Role {
    /// Full PDO config owner: writes 0x1608/0x1A08/0x1C12/0x1C13.
    /// Used when J1900 is the first/only master to configure the bridge.
    ConfigOwner,
    /// Read-only consumer: never writes PDO mapping/assignment.
    /// Adopts whatever layout the primary-side owner (TwinCAT) left.
    Consumer,
}

struct Config {
    iface: String,
    mode: Mode,
    role: Role,
    duration_sec: u64,
    cycle_us: u64,
    core: usize,
    prio: u8,
    quiet: bool,
    /// Toggle an EL2202 digital output every cycle (oscilloscope jitter
    /// waveform). Off by default: the outputs may drive real actuators.
    el2202: bool,
    /// Toggle BOTH EL2202 channels (0x03/0x00): output 1 -> EL1252 readback,
    /// output 2 -> oscilloscope, same edges in two views.
    el2202_dual: bool,
    /// Read EL1252 DC latch timestamps every cycle (edge interval/phase
    /// statistics; wire the signal under test to EL1252 channel 1).
    el1252: bool,
    /// Drive an EL2262 XFC output with DC-sync-locked edges (reference
    /// square wave on the DC 1 ms grid).
    el2262: bool,
    /// Read the EL1252 every N cycles (frame-budget throttle when combined
    /// with other per-cycle frames; edges are latched so nothing is lost
    /// beyond time resolution).
    el1252_every: u32,
    /// Spin-lock mode: instead of a free-running ticker, poll FPRD until the
    /// payload changes (new TwinCAT frame), then process immediately.
    /// Phase-locks J1900 processing to the primary master's frame arrival.
    spin: bool,
    /// Enable DC SYNC0 on the EL6695 secondary (cycle = --cycle-us) and print
    /// per-second DC phase/drift telemetry in the report line.
    dc: bool,
    /// Use the io_uring-based blocking TX/RX loop instead of AF_PACKET poll.
    iouring: bool,
    /// RFM device path override (e.g. /dev/rfm2g1) for multi-card hosts.
    rfm_dev: Option<String>,
}

fn parse_args() -> Result<Config, String> {
    let mut cfg = Config {
        iface: "eth1".into(),
        mode: Mode::Ticker,
        role: Role::ConfigOwner,
        duration_sec: 60,
        cycle_us: 1000,
        core: 2,
        prio: 80,
        quiet: false,
        el2202: false,
        el2202_dual: false,
        el1252: false,
        el2262: false,
        el1252_every: 1,
        spin: false,
        dc: false,
        iouring: false,
        rfm_dev: None,
    };
    let mut args = std::env::args().skip(1).peekable();
    if let Some(a) = args.peek() {
        if !a.starts_with("--") {
            cfg.iface = args.next().unwrap();
        }
    }
    while let Some(a) = args.next() {
        let mut val = |name: &str| -> Result<String, String> {
            args.next().ok_or_else(|| format!("missing value for {}", name))
        };
        match a.as_str() {
            "--mode" => {
                cfg.mode = match val("--mode")?.as_str() {
                    "ticker" => Mode::Ticker,
                    "probe" => Mode::Probe,
                    "verify" => Mode::Verify,
                    "rfm-ping" => Mode::RfmPing,
                    "rfm-echo" => Mode::RfmEcho,
                    "rfm-bridge" => Mode::RfmBridge,
                    "rfm-ring" => Mode::RfmRing,
                    other => return Err(format!("unknown mode '{}'", other)),
                }
            }
            "--duration" => {
                cfg.duration_sec = val("--duration")?
                    .parse()
                    .map_err(|_| "bad --duration")?
            }
            "--cycle-us" => cfg.cycle_us = val("--cycle-us")?.parse().map_err(|_| "bad --cycle-us")?,
            "--core" => cfg.core = val("--core")?.parse().map_err(|_| "bad --core")?,
            "--prio" => cfg.prio = val("--prio")?.parse().map_err(|_| "bad --prio")?,
            "--quiet" => cfg.quiet = true,
            "--spin" => cfg.spin = true,
            "--dc" => cfg.dc = true,
            "--iouring" => cfg.iouring = true,
            "--rfm-dev" => {
                let n: u32 = val("--rfm-dev")?.parse().map_err(|_| "bad --rfm-dev (use card index, e.g. 0 or 1)")?;
                cfg.rfm_dev = Some(format!("/dev/rfm2g{}", n));
            }
            "--el2202" => cfg.el2202 = true,
            "--el2202-dual" => {
                cfg.el2202 = true;
                cfg.el2202_dual = true;
            }
            "--el1252" => cfg.el1252 = true,
            "--el2262" => cfg.el2262 = true,
            "--consumer" => cfg.role = Role::Consumer,
            "--el1252-every" => {
                cfg.el1252_every = val("--el1252-every")?.parse().map_err(|_| "bad --el1252-every")?
            }
            other => return Err(format!("unknown argument '{}'", other)),
        }
    }
    if cfg.cycle_us == 0 {
        return Err("--cycle-us must be > 0".into());
    }
    Ok(cfg)
}

fn main() {
    let cfg = match parse_args() {
        Ok(c) => c,
        Err(e) => {
            eprintln!("error: {}", e);
            eprintln!(
                "usage: el6695_rt [iface] [--mode ticker|probe] [--duration SEC] \
                 [--cycle-us N] [--core N] [--prio N] [--quiet] [--el2202] [--consumer]"
            );
            std::process::exit(2);
        }
    };

    unsafe {
        let mut sa: libc::sigaction = core::mem::zeroed();
        sa.sa_sigaction = on_signal as usize;
        sa.sa_flags = 0;
        libc::sigemptyset(&mut sa.sa_mask);
        libc::sigaction(libc::SIGINT, &sa, core::ptr::null_mut());
        libc::sigaction(libc::SIGTERM, &sa, core::ptr::null_mut());
    }

    let code = match futures_lite::future::block_on(run(&cfg)) {
        Ok(()) => 0,
        Err(e) => {
            eprintln!("Failed: {}", e);
            1
        }
    };
    std::process::exit(code);
}

async fn run(cfg: &Config) -> Result<(), Box<dyn std::error::Error>> {
    let cycle_ns = cfg.cycle_us * 1000;

    println!("el6695-rt: J1900 <-> TwinCAT 1 kHz verification (EL6695 bridge, secondary side)");
    println!(
        "  iface={} mode={} duration={}s cycle={}us core={} prio={}",
        cfg.iface,
        match cfg.mode {
            Mode::Ticker => "ticker",
            Mode::Probe => "probe",
            Mode::Verify => "verify",
            Mode::RfmPing => "rfm-ping",
            Mode::RfmEcho => "rfm-echo",
            Mode::RfmBridge => "rfm-bridge",
            Mode::RfmRing => "rfm-ring",
        },
        cfg.duration_sec,
        cfg.cycle_us,
        cfg.core,
        cfg.prio
    );

    let pdu_storage = PDU_STORAGE.get_or_init(PduStorage::new);

    // ---- RFM-only modes: no EtherCAT bring-up needed ----
    if matches!(cfg.mode, Mode::RfmPing | Mode::RfmEcho | Mode::RfmRing) {
        if cfg.mode == Mode::RfmRing {
            let rfm = el6695_rt::rfm::Rfm::open_path(cfg.rfm_dev.as_deref())?;
            let fd = rfm.device_fd.as_raw_fd();
            let ok = el6695_rt::rfm::ring_cont(fd)?;
            println!("ring={}", if ok { "OK" } else { "BROKEN" });
            std::process::exit(if ok { 0 } else { 1 });
        }
        return rfm_mode_main(cfg).await;
    }

    let (tx, rx, pdu_loop) = pdu_storage.try_split().map_err(|_| "PDU storage already in use")?;

    let maindevice = MainDevice::new(
        pdu_loop,
        Timeouts {
            state_transition: Duration::from_millis(10000),
            mailbox_response: Duration::from_millis(2000),
            eeprom: Duration::from_millis(500),
            ..Default::default()
        },
        MainDeviceConfig::default(),
    );

    let iface = cfg.iface.clone();
    let use_iouring = cfg.iouring;
    let _txrx = std::thread::spawn(move || {
        if let Err(e) = rt::promote_current_thread(1, 79, false) {
            eprintln!("tx_rx RT promotion failed (continuing): {}", e);
        }
        if use_iouring {
            match tx_rx_task_io_uring(&iface, tx, rx) {
                Ok(_) => {}
                Err(e) => eprintln!("io_uring tx_rx task ended with error: {:?}", e),
            }
        } else {
            let task = tx_rx_task(&iface, tx, rx).expect("create tx_rx task");
            if let Err(e) = futures_lite::future::block_on(task) {
                eprintln!("tx_rx task ended with error: {:?}", e);
            }
        }
    });

    println!("Scanning segment (INIT -> PREOP)...");
    let group = maindevice
        .init_single_group::<MAX_SUBDEVICES, PDI_LEN>(ethercat_now)
        .await?;

    let mut el6695_index = None;
    for (idx, subdevice) in group.iter(&maindevice).enumerate() {
        if subdevice.name().contains("EL6695") {
            el6695_index = Some(idx);
            println!("[{}] Found EL6695", idx);
        }
    }
    let el6695_index = el6695_index.ok_or("EL6695 not found on bus")?;
    let sd = group.subdevice(&maindevice, el6695_index)?;
    let station = sd.configured_address();
    println!("EL6695 station address: 0x{:04X}", station);

    // ---- Probe SM registers ----
    async fn read_sm(md: &MainDevice<'_>, st: u16, n: u8) -> (u16, u16, u8, u8) {
        let start = ethercrab::Command::fprd(st, regs::sm_base(n as u16) + regs::SM_OFF_START).receive::<u16>(md).await.unwrap_or(0xFFFF);
        let len = ethercrab::Command::fprd(st, regs::sm_base(n as u16) + regs::SM_OFF_LEN).receive::<u16>(md).await.unwrap_or(0xFFFF);
        let ctrl = ethercrab::Command::fprd(st, regs::sm_base(n as u16) + regs::SM_OFF_CTRL).receive::<u8>(md).await.unwrap_or(0xFF);
        let ena = ethercrab::Command::fprd(st, regs::sm_base(n as u16) + regs::SM_OFF_ENABLE).receive::<u8>(md).await.unwrap_or(0xFF);
        (start, len, ctrl, ena)
    }
    for n in [0u8, 1, 2, 3] {
        let (s, l, c, e) = read_sm(&maindevice, station, n).await;
        println!("  SM{}: start=0x{:04X} len={} ctrl=0x{:02X} enable=0x{:02X}", n, s, l, c, e);
    }
    // Probe FMMU registers
    for n in 0u16..4 {
        let reg = 0x0600 + n * 16;
        match ethercrab::Command::fprd(station, reg)
            .receive_slice(&maindevice, 16)
            .await
        {
            Ok(pdu) => {
                let raw: &[u8] = &pdu;
                let logical_start = u32::from_le_bytes(raw[0..4].try_into().unwrap());
                let length = u16::from_le_bytes(raw[4..6].try_into().unwrap());
                let phy_start = u16::from_le_bytes(raw[8..10].try_into().unwrap());
                println!("  FMMU{}: logical_start=0x{:08X} len={} physical=0x{:04X} enable=0x{:02X}", n, logical_start, length, phy_start, raw[12]);
            }
            Err(e) => println!("  FMMU{} read error: {:?}", n, e),
        }
    }

    // ---- Configure SM2/SM3 for process data (SM0/SM1 are EEPROM-fixed mailbox) ----
    // 0x1C12/0x1C13 exist (empty), 0x1C10/0x1C11 do not.
    // SM2/SM3 registers are writable because they are not defined in EEPROM.
    println!("\nTesting SM0/SM1 data access...");
    // Read SM1 (buffered input, 0x1600)
    match ethercrab::Command::fprd(station, 0x1600)
        .receive_slice(&maindevice, 32)
        .await
    {
        Ok(pdu) => {
            let raw: &[u8] = &pdu;
            println!("  SM1 FPRD(0x1600,32) OK: {:02X?}", &raw[..32.min(raw.len())]);
        }
        Err(e) => println!("  SM1 FPRD error: {:?}", e),
    }
    // Write SM0 (mailbox, 0x1000)
    match ethercrab::Command::fpwr(station, 0x1000)
        .send(&maindevice, &[0xBBu8; 16])
        .await
    {
        Ok(_) => println!("  SM0 FPWR(0x1000,16) OK"),
        Err(e) => println!("  SM0 FPWR error: {:?}", e),
    }
    // Read back SM0
    match ethercrab::Command::fprd(station, 0x1000)
        .receive_slice(&maindevice, 16)
        .await
    {
        Ok(pdu) => {
            let raw: &[u8] = &pdu;
            println!("  SM0 FPRD(0x1000,16) back: {:02X?}", &raw[..16.min(raw.len())]);
        }
        Err(e) => println!("  SM0 FPRD back error: {:?}", e),
    }
    
    // Remove SM2/SM3 config (was causing SAFEOP 0x001D)
    // Safety: clear SM2/SM3 in case previous runs left stale values
    let _ = ethercrab::Command::fpwr(station, regs::sm_base(2) + regs::SM_OFF_ENABLE)
        .send(&maindevice, 0x00u8).await;
    let _ = ethercrab::Command::fpwr(station, regs::sm_base(3) + regs::SM_OFF_ENABLE)
        .send(&maindevice, 0x00u8).await;
    
    // ---- Verify mode: CoE audit (then exit, no SAFEOP) ----
    if cfg.mode == Mode::Verify {
        println!("\n=== CoE Audit ===");
        // 0xF008: NoCoeStorage / vendor reset
        let f008_0 = sd.sdo_read::<u32>(0xF008, 0).await;
        let f008_1 = sd.sdo_read::<u8>(0xF008, 1).await;
        println!("  0xF008:00 = {:?}  sub1={:?}", f008_0, f008_1);
        let no_coe_storage = f008_0.unwrap_or(0) == 0x12345678;
        println!("  NoCoeStorage: {}", if no_coe_storage { "⚠ ENABLED (0x12345678) — config will NOT persist to EEPROM" } else { "✓ disabled" });
        
        // 0x1018: Identity
        for sub in 1..=4u8 {
            let v = sd.sdo_read::<u32>(0x1018, sub).await;
            let label = match sub { 1 => "Vendor", 2 => "Product", 3 => "Revision", _ => "Serial" };
            println!("  0x1018:{:02} ({}) = 0x{:08X}", sub, label, v.unwrap_or(0xFFFFFFFF));
        }
        
        // Full PDO layout
        let snap = bridge::snapshot_layout(&sd, &maindevice, station).await;
        let h = bridge::compute_layout_hash(&snap);
        println!("  0x1C12 assign: {:?}", snap.tx_assign);
        println!("  0x1C13 assign: {:?}", snap.rx_assign);
        println!("  0x1608 entries: {} (total {} bits)", snap.tx_entries.len(), snap.tx_total_bits);
        if !snap.tx_entries.is_empty() {
            for (i, (idx, sub, bl)) in snap.tx_entries.iter().enumerate().take(3) {
                println!("    [{:02}] 0x{:04X}:{:02} @ {} bits", i+1, idx, sub, bl);
            }
            if snap.tx_entries.len() > 3 {
                println!("    ... ({} more)", snap.tx_entries.len() - 3);
            }
        }
        println!("  0x1A08 entries: {} (total {} bits)", snap.rx_entries.len(), snap.rx_total_bits);
        if !snap.rx_entries.is_empty() {
            for (i, (idx, sub, bl)) in snap.rx_entries.iter().enumerate().take(3) {
                println!("    [{:02}] 0x{:04X}:{:02} @ {} bits", i+1, idx, sub, bl);
            }
            if snap.rx_entries.len() > 3 {
                println!("    ... ({} more)", snap.rx_entries.len() - 3);
            }
        }
        println!("  SM2 len={} SM3 len={}", snap.sm2_len, snap.sm3_len);
        println!("  Layout CRC32 = 0x{:08X}", h);
        println!("  Expected hash = 0x{:08X}", bridge::EXPECTED_LAYOUT_HASH.load(std::sync::atomic::Ordering::Relaxed));
        println!("  0x6000[0] = {}", sd.sdo_read::<u8>(0x6000, 0).await.unwrap_or(0xFF));
        println!("  0x7000[0] = {}", sd.sdo_read::<u8>(0x7000, 0).await.unwrap_or(0xFF));
        
        // AL state / error
        let (al, _) = bridge::drive_state(&maindevice, station, 0x0002).await;
        println!("  AL=0x{:04X} err={}", al, al & regs::AL_ERROR != 0);
        println!("=== CoE Audit complete ===");
        return Ok(());
    }
    
    // ---- ensure_link: adopt-if-valid (Consumer) or full reconfigure (ConfigOwner) ----
    println!("\nensure_link: role={:?} ...", cfg.role);
    
    // Read current PDO layout from the bridge
    let layout_ok = bridge::layout_matches(&sd, &maindevice, station).await.unwrap_or(false);
    let (al, alc) = bridge::drive_state(&maindevice, station, 0x0002).await;
    let has_err = al & regs::AL_ERROR != 0;
    let al_state = al & 0x0F;
    println!("  AL=0x{:04X} err={} layout_ok={}", al, has_err, layout_ok);
    
    match cfg.role {
        Role::Consumer => {
            // ---- Consumer: idempotent writes, no INIT→PREOP cycle ----
            // SM/PDO config must ALWAYS be written after scan (scanning wipes
            // SM registers and PDO assign/map).  The writes are idempotent:
            // same layout as the expected expanded 1024-byte config.
            // Clear any PREOP+ERR (L1 recovery)
            if has_err {
                println!("  Consumer: clearing PREOP+ERR ...");
                let _ = ethercrab::Command::fpwr(station, regs::AL_CONTROL)
                    .send(&maindevice, 0x0002u16).await;
                std::thread::sleep(Duration::from_millis(300));
            }
            // Skip INIT→PREOP — preserves TwinCAT's persisted CoE objects.
            println!("  Consumer: idempotent SM/FMMU/PDO config");
            // SM2/SM3 registers
            let _ = ethercrab::Command::fpwr(station, regs::sm_base(2) + regs::SM_OFF_ENABLE)
                .send(&maindevice, 0x00u8).await;
            let _ = ethercrab::Command::fpwr(station, regs::sm_base(2) + regs::SM_OFF_START)
                .send(&maindevice, 0x1C00u16).await;
            let _ = ethercrab::Command::fpwr(station, regs::sm_base(2) + regs::SM_OFF_LEN)
                .send(&maindevice, (PAYLOAD_LEN as u16)).await;
            let _ = ethercrab::Command::fpwr(station, regs::sm_base(2) + regs::SM_OFF_CTRL)
                .send(&maindevice, 0x64u8).await;
            let _ = ethercrab::Command::fpwr(station, regs::sm_base(2) + regs::SM_OFF_ENABLE)
                .send(&maindevice, 0x01u8).await;
            let _ = ethercrab::Command::fpwr(station, regs::sm_base(3) + regs::SM_OFF_ENABLE)
                .send(&maindevice, 0x00u8).await;
            let _ = ethercrab::Command::fpwr(station, regs::sm_base(3) + regs::SM_OFF_START)
                .send(&maindevice, 0x8E00u16).await;
            let _ = ethercrab::Command::fpwr(station, regs::sm_base(3) + regs::SM_OFF_LEN)
                .send(&maindevice, (PAYLOAD_LEN as u16)).await;
            let _ = ethercrab::Command::fpwr(station, regs::sm_base(3) + regs::SM_OFF_CTRL)
                .send(&maindevice, 0x20u8).await;
            let _ = ethercrab::Command::fpwr(station, regs::sm_base(3) + regs::SM_OFF_ENABLE)
                .send(&maindevice, 0x01u8).await;
            bridge::write_fmmus_secondary(&maindevice, station, PAYLOAD_LEN as u16, PAYLOAD_LEN as u16).await;
            // PDO maps and assigns (whitelisted, idempotent)
            println!("  Consumer: write PDO mappings...");
            let blob = bridge::twin_layout_blob_sized(0x7000, PAYLOAD_LEN);
            let _ = bridge::sdo_write_ca_checked(&sd, 0x1608, 0, &blob).await;
            let blob = bridge::twin_layout_blob_sized(0x6000, PAYLOAD_LEN);
            let _ = bridge::sdo_write_ca_checked(&sd, 0x1A08, 0, &blob).await;
            println!("  Consumer: write PDO assignments...");
            let blob = bridge::assign_blob(&[0x1608]);
            let _ = bridge::sdo_write_ca_checked(&sd, 0x1C12, 0, &blob).await;
            let blob = bridge::assign_blob(&[0x1A08]);
            let _ = bridge::sdo_write_ca_checked(&sd, 0x1C13, 0, &blob).await;
            let _o7 = sd.sdo_read::<u8>(0x7000, 0).await.unwrap_or(0xFF);
            let _o6 = sd.sdo_read::<u8>(0x6000, 0).await.unwrap_or(0xFF);
            let snap = bridge::snapshot_layout(&sd, &maindevice, station).await;
            let h = bridge::compute_layout_hash(&snap);
            println!("  Consumer: dyn 0x6000[0]={} 0x7000[0]={} hash=0x{:08X}", _o6, _o7, h);
        }

        Role::ConfigOwner => {
            // ---- ConfigOwner: full INIT→PREOP + PDO rebuild ----
            // Clear any PREOP+ERR by re-requesting PREOP
            if has_err {
                println!("  ConfigOwner: clearing PREOP+ERR ...");
                let _ = ethercrab::Command::fpwr(station, regs::AL_CONTROL)
                    .send(&maindevice, 0x0002u16).await;
                std::thread::sleep(Duration::from_millis(300));
            }
            // INIT→PREOP cycle to reset CoE state
            println!("  INIT→PREOP cycle...");
            let (al_i, _) = bridge::drive_state(&maindevice, station, 0x0001).await;
            println!("    INIT: {}", bridge::verdict(al_i, 0));
            if al_i & 0x0F != 0x01 {
                return Err(format!("INIT failed: {}", bridge::verdict(al_i, 0)).into());
            }
            let (al_p, alc_p) = bridge::drive_state(&maindevice, station, 0x0002).await;
            println!("    PREOP: {}", bridge::verdict(al_p, alc_p));
            if al_p & 0x0F != 0x02 {
                return Err(format!("PREOP failed: {}", bridge::verdict(al_p, alc_p)).into());
            }
            // Write PDO mappings (0x1608 TxPDO, 0x1A08 RxPDO)
            // Whitelist-checked: only {0x1608,0x1A08,0x1C12,0x1C13,0xF008,0xF800} allowed.
            println!("  Write PDO mappings...");
            let blob = bridge::twin_layout_blob_sized(0x7000, PAYLOAD_LEN);
            let _ = bridge::sdo_write_ca_checked(&sd, 0x1608, 0, &blob).await;
            let blob = bridge::twin_layout_blob_sized(0x6000, PAYLOAD_LEN);
            let _ = bridge::sdo_write_ca_checked(&sd, 0x1A08, 0, &blob).await;
            // Write PDO assignments (triggers 0x6000/0x7000 object creation)
            println!("  Write PDO assignments...");
            let blob = bridge::assign_blob(&[0x1608]);
            let _ = bridge::sdo_write_ca_checked(&sd, 0x1C12, 0, &blob).await;
            let blob = bridge::assign_blob(&[0x1A08]);
            let _ = bridge::sdo_write_ca_checked(&sd, 0x1C13, 0, &blob).await;
            // Verify the dynamic objects were instantiated
            let _o7 = sd.sdo_read::<u8>(0x7000, 0).await.unwrap_or(0xFF);
            let _o6 = sd.sdo_read::<u8>(0x6000, 0).await.unwrap_or(0xFF);
            // Compute layout hash after successful config
            let snap = bridge::snapshot_layout(&sd, &maindevice, station).await;
            let h = bridge::compute_layout_hash(&snap);
            println!("  dyn: 0x6000[0]={} 0x7000[0]={}  layout_hash=0x{:08X}", _o6, _o7, h);
            // Configure SM2/SM3 + FMMU
            let _ = ethercrab::Command::fpwr(station, regs::sm_base(2) + regs::SM_OFF_ENABLE)
                .send(&maindevice, 0x00u8).await;
            let _ = ethercrab::Command::fpwr(station, regs::sm_base(2) + regs::SM_OFF_START)
                .send(&maindevice, 0x1C00u16).await;
            let _ = ethercrab::Command::fpwr(station, regs::sm_base(2) + regs::SM_OFF_LEN)
                .send(&maindevice, (PAYLOAD_LEN as u16)).await;
            let _ = ethercrab::Command::fpwr(station, regs::sm_base(2) + regs::SM_OFF_CTRL)
                .send(&maindevice, 0x64u8).await;
            let _ = ethercrab::Command::fpwr(station, regs::sm_base(2) + regs::SM_OFF_ENABLE)
                .send(&maindevice, 0x01u8).await;
            let _ = ethercrab::Command::fpwr(station, regs::sm_base(3) + regs::SM_OFF_ENABLE)
                .send(&maindevice, 0x00u8).await;
            let _ = ethercrab::Command::fpwr(station, regs::sm_base(3) + regs::SM_OFF_START)
                .send(&maindevice, 0x8E00u16).await;
            let _ = ethercrab::Command::fpwr(station, regs::sm_base(3) + regs::SM_OFF_LEN)
                .send(&maindevice, (PAYLOAD_LEN as u16)).await;
            let _ = ethercrab::Command::fpwr(station, regs::sm_base(3) + regs::SM_OFF_CTRL)
                .send(&maindevice, 0x20u8).await;
            let _ = ethercrab::Command::fpwr(station, regs::sm_base(3) + regs::SM_OFF_ENABLE)
                .send(&maindevice, 0x01u8).await;
            bridge::write_fmmus_secondary(&maindevice, station, PAYLOAD_LEN as u16, PAYLOAD_LEN as u16).await;
        }
    }
    
    // SAFEOP
    let (al, alc) = bridge::drive_state(&maindevice, station, 0x0004).await;
    println!("  SAFEOP: {}", bridge::verdict(al, alc));
    if al & 0x0F != 0x04 || al & 0x10 != 0 {
        return Err(format!("Cannot reach SAFEOP: {}", bridge::verdict(al, alc)).into());
    }
    println!("  >>> SAFEOP achieved! <<<");
    let (al2, alc2) = bridge::drive_state(&maindevice, station, 0x0008).await;
    println!("  OP: {}", bridge::verdict(al2, alc2));
    if al2 & 0x0F != 0x08 || al2 & 0x10 != 0 {
        return Err(format!("OP rejected: {}", bridge::verdict(al2, alc2)).into());
    }
    println!("  >>> OP achieved! <<<");

    // ---- Optional DC SYNC0 on the EL6695 secondary ----
    // ethercrab init already distributed the system time (offset/propagation
    // delay/static drift). Here we only start the SYNC0 unit on --cycle-us
    // grid and verify the clock actually advances.
    let mut dc_ref: Option<(u64, u64)> = None; // (dc_time, mono_time) at setup
    if cfg.dc {
        let t1 = ethercrab::Command::fprd(station, regs::DC_SYSTEM_TIME)
            .receive::<u64>(&maindevice)
            .await
            .unwrap_or(0);
        std::thread::sleep(Duration::from_millis(100));
        let t2 = ethercrab::Command::fprd(station, regs::DC_SYSTEM_TIME)
            .receive::<u64>(&maindevice)
            .await
            .unwrap_or(0);
        let adv = t2.wrapping_sub(t1);
        println!("DC system time: {} -> {} (adv {} ns in 100ms)", t1, t2, adv);
        if adv < 50_000_000 || adv > 500_000_000 {
            println!("DC: clock not advancing sanely; SYNC0 setup skipped");
        } else {
            // Diagnostics: which DC register groups actually exist?
            for (reg, name) in [
                (0x0900u16, "rx_time_port0(u32)"),
                (0x0920u16, "sys_time_offset(u64)"),
                (0x0928u16, "prop_delay(u32)"),
                (0x0930u16, "drift... (u16)"),
                (0x0980u16, "sync_activation(u8)"),
                (0x0981u16, "sync_active_byte(u8)"),
                (0x09A0u16, "sync0_cycle(u32)"),
                (0x09AEu16, "sync0_status(u8)"),
            ] {
                match ethercrab::Command::fprd(station, reg)
                    .receive_slice(&maindevice, 8)
                    .await
                {
                    Ok(pdu) => {
                        let raw: &[u8] = &pdu;
                        println!("  DC probe 0x{:04X} {}: {:02X?}", reg, name, &raw[..8.min(raw.len())]);
                    }
                    Err(e) => println!("  DC probe 0x{:04X} {}: ERR {:?}", reg, name, e),
                }
            }
            // CoE view: 0x1C32 (SM2 outputs) / 0x1C33 (SM3 inputs) sync params
            for idx in [0x1C32u16, 0x1C33] {
                let n = sd.sdo_read::<u8>(idx, 0).await.unwrap_or(0xFF);
                println!("  CoE 0x{:04X}: {} subentries", idx, n);
                for (sub, name) in [(0x01u8, "sync_mode"), (0x04, "modes_supported"), (0x0B, "sm_event_missed"), (0x0C, "cycle_exceeded")] {
                    match sd.sdo_read::<u16>(idx, sub).await {
                        Ok(x) => println!("  CoE 0x{:04X}:{:02X} {} = 0x{:04X}", idx, sub, name, x),
                        Err(e) => println!("  CoE 0x{:04X}:{:02X} {} ERR {:?}", idx, sub, name, e),
                    }
                }
                match sd.sdo_read::<u32>(idx, 0x02).await {
                    Ok(x) => println!("  CoE 0x{:04X}:02 cycle_time_ns = {}", idx, x),
                    Err(e) => println!("  CoE 0x{:04X}:02 cycle_time_ns ERR {:?}", idx, e),
                }
                match sd.sdo_read::<u8>(idx, 0x20).await {
                    Ok(x) => println!("  CoE 0x{:04X}:20 sync_error = {}", idx, x),
                    Err(e) => println!("  CoE 0x{:04X}:20 sync_error ERR {:?}", idx, e),
                }
            }
            let start = t2.wrapping_add(100_000_000);
            let w1 = ethercrab::Command::fpwr(station, regs::DC_SYNC_START_TIME)
                .send(&maindevice, start)
                .await;
            println!("  DC write start_time(0x0990): {:?}", w1.as_ref().map(|_| "OK").map_err(|e| format!("{:?}", e)));
            let w2 = ethercrab::Command::fpwr(station, regs::DC_SYNC0_CYCLE)
                .send(&maindevice, cycle_ns as u32)
                .await;
            println!("  DC write sync0_cycle(0x09A0): {:?}", w2.as_ref().map(|_| "OK").map_err(|e| format!("{:?}", e)));
            let w3 = regs::rmw_u8(
                &maindevice,
                station,
                regs::DC_SYNC_ACTIVATION,
                regs::DC_SYNC0_ACTIVATE,
                0,
            )
            .await;
            println!("  DC rmw activation(0x0980): {:?}", w3.as_ref().map(|v| format!("0x{:02X}", v)).map_err(|e| format!("{:?}", e)));
            // rmw needs a read (fails here); try a blind full-byte write instead
            let w4 = ethercrab::Command::fpwr(station, regs::DC_SYNC_ACTIVATION)
                .send(&maindevice, regs::DC_SYNC0_ACTIVATE)
                .await;
            println!("  DC blind write activation(0x0980)=0x01: {:?}", w4.as_ref().map(|_| "OK").map_err(|e| format!("{:?}", e)));
            std::thread::sleep(Duration::from_millis(150));
            // Try selecting DC-Synchron SYNC0 via firmware object (mode 2).
            // May be rejected in OP; worth observing the abort code either way.
            let mut coe_dc_ok = true;
            for idx in [0x1C32u16, 0x1C33] {
                match sd.sdo_write::<u16>(idx, 0x01, 2).await {
                    Ok(_) => println!("  CoE write 0x{:04X}:01 sync_mode=2 OK", idx),
                    Err(e) => {
                        coe_dc_ok = false;
                        println!("  CoE write 0x{:04X}:01 sync_mode=2 ERR {:?}", idx, e);
                    }
                }
                match sd.sdo_read::<u16>(idx, 0x01).await {
                    Ok(x) => println!("  CoE-after 0x{:04X}:01 sync_mode = {}", idx, x),
                    Err(e) => {
                        coe_dc_ok = false;
                        println!("  CoE-after 0x{:04X}:01 sync_mode ERR {:?}", idx, e);
                    }
                }
            }
            if coe_dc_ok {
                let t_dc = ethercrab::Command::fprd(station, regs::DC_SYSTEM_TIME)
                    .receive::<u64>(&maindevice)
                    .await
                    .unwrap_or(0);
                dc_ref = Some((t_dc, rt::now_ns()));
                println!("DC-Synchron: firmware sync_mode=2 accepted, DC telemetry on");
            }
            let act = ethercrab::Command::fprd(station, regs::DC_SYNC_ACTIVATION)
                .receive::<u8>(&maindevice)
                .await;
            let cyc0 = ethercrab::Command::fprd(station, regs::DC_SYNC0_CYCLE)
                .receive::<u32>(&maindevice)
                .await;
            let t3 = ethercrab::Command::fprd(station, regs::DC_SYSTEM_TIME)
                .receive::<u64>(&maindevice)
                .await
                .unwrap_or(0);
            let (al, alc) = bridge::drive_state(&maindevice, station, 0x0008).await;
            println!("  AL after DC writes: {}", bridge::verdict(al, alc));
            println!(
                "DC SYNC0: act={:?} cycle={:?} start=+100ms (t3={} start_delta={}ns)",
                act.as_ref().map(|v| format!("0x{:02X}", v)).map_err(|e| format!("{:?}", e)),
                cyc0.as_ref().map_err(|e| format!("{:?}", e)),
                t3,
                t3.wrapping_sub(start) as i64
            );
            if matches!(act, Ok(a) if a & regs::DC_SYNC0_ACTIVATE != 0)
                && matches!(cyc0, Ok(c) if c == cycle_ns as u32)
            {
                dc_ref = Some((t3, rt::now_ns()));
                println!("DC SYNC0: active on EL6695 secondary");
            } else {
                println!("DC SYNC0: readback mismatch; running without DC telemetry");
            }
        }
    }

    // ---- Optional EL2202 oscilloscope waveform (toggle one DO per cycle) ----
    const EL2202_OUT_ADDR: u16 = 0x0F00;
    let mut el2202: Option<(u16, u16)> = None;
    if cfg.el2202 {
        let mut found = false;
        for (idx, subdevice) in group.iter(&maindevice).enumerate() {
            if subdevice.name().contains("EL2202") {
                found = true;
                let sd2 = group.subdevice(&maindevice, idx)?;
                let st2 = sd2.configured_address();
                println!("[{}] Found EL2202, station 0x{:04X}", idx, st2);
                let _ = ethercrab::Command::fpwr(st2, regs::sm_base(0) + regs::SM_OFF_START)
                    .send(&maindevice, EL2202_OUT_ADDR)
                    .await;
                let _ = ethercrab::Command::fpwr(st2, regs::sm_base(0) + regs::SM_OFF_LEN)
                    .send(&maindevice, 1u16)
                    .await;
                let _ = ethercrab::Command::fpwr(st2, regs::sm_base(0) + regs::SM_OFF_CTRL)
                    .send(&maindevice, regs::SM_CTRL_EL2202_OUTPUTS)
                    .await;
                let _ = regs::rmw_u8(
                    &maindevice,
                    st2,
                    regs::sm_base(0) + regs::SM_OFF_ENABLE,
                    regs::SM_ENABLE | regs::SM_ENABLE_FAMILY,
                    0,
                )
                .await;
                let (al, alc) = bridge::drive_state(&maindevice, st2, regs::AL_SAFEOP).await;
                println!("EL2202 SAFEOP: {}", bridge::verdict(al, alc));
                let (al, alc) = bridge::drive_state(&maindevice, st2, regs::AL_OP).await;
                println!("EL2202 OP:     {}", bridge::verdict(al, alc));
                if al & regs::AL_STATE_MASK != regs::AL_OP || al & regs::AL_ERROR != 0 {
                    return Err(format!("EL2202 OP rejected: {}", bridge::verdict(al, alc)).into());
                }
                let _ = ethercrab::Command::fpwr(st2, EL2202_OUT_ADDR)
                    .send(&maindevice, regs::EL2202_CH1_DATA)
                    .await;
                let rb = ethercrab::Command::fprd(st2, EL2202_OUT_ADDR)
                    .receive::<u8>(&maindevice)
                    .await
                    .unwrap_or(0xFF);
                if rb & regs::EL2202_CH1_DATA == regs::EL2202_CH1_DATA {
                    println!(
                        "EL2202 scope output: verified, toggle 1 byte @0x{:04X} every cycle",
                        EL2202_OUT_ADDR
                    );
                    let _ = ethercrab::Command::fpwr(st2, EL2202_OUT_ADDR)
                        .send(&maindevice, 0x00u8)
                        .await;
                    el2202 = Some((st2, EL2202_OUT_ADDR));
                } else {
                    println!("EL2202 output readback failed (0x{:02X}); scope output disabled", rb);
                }
            }
        }
        if !found {
            println!("--el2202 given but no EL2202 on bus; scope output disabled");
        }
        if found && el2202.is_none() {
            println!("EL2202 present but output setup failed; scope output disabled");
        }
    }

    // ---- Optional EL2262 DC-locked reference output ----
    const EL2262_OUT_ADDR: u16 = 0x1000;
    let mut el2262: Option<u16> = None;
    if cfg.el2262 {
        let mut found = false;
        for (idx, subdevice) in group.iter(&maindevice).enumerate() {
            if subdevice.name().contains("EL2262") {
                found = true;
                let sd3 = group.subdevice(&maindevice, idx)?;
                let st3 = sd3.configured_address();
                println!("[{}] Found EL2262, station 0x{:04X}", idx, st3);
                let t = ethercrab::Command::fprd(st3, regs::DC_SYSTEM_TIME)
                    .receive::<u64>(&maindevice)
                    .await
                    .unwrap_or(0);
                let start = t.wrapping_add(100_000_000);
                let _ = ethercrab::Command::fpwr(st3, regs::DC_SYNC_START_TIME)
                    .send(&maindevice, start)
                    .await;
                let _ = ethercrab::Command::fpwr(st3, regs::DC_SYNC0_CYCLE)
                    .send(&maindevice, cycle_ns as u32)
                    .await;
                let _ = ethercrab::Command::fpwr(st3, regs::DC_SYNC1_CYCLE)
                    .send(&maindevice, cycle_ns as u32)
                    .await;
                let _ = regs::rmw_u8(
                    &maindevice,
                    st3,
                    regs::DC_SYNC_ACTIVATION,
                    regs::DC_SYNC0_ACTIVATE | regs::DC_SYNC1_ACTIVATE,
                    0,
                )
                .await;
                println!("EL2262 DC sync: sync0/1 cycle={}ns start=+100ms (t={})", cycle_ns, t);
                let _ = ethercrab::Command::fpwr(st3, regs::sm_base(0) + regs::SM_OFF_START)
                    .send(&maindevice, EL2262_OUT_ADDR)
                    .await;
                let _ = ethercrab::Command::fpwr(st3, regs::sm_base(0) + regs::SM_OFF_LEN)
                    .send(&maindevice, 1u16)
                    .await;
                let _ = ethercrab::Command::fpwr(st3, regs::sm_base(0) + regs::SM_OFF_CTRL)
                    .send(&maindevice, regs::SM_CTRL_OUTPUTS_ESI)
                    .await;
                let _ = regs::rmw_u8(
                    &maindevice,
                    st3,
                    regs::sm_base(0) + regs::SM_OFF_ENABLE,
                    regs::SM_ENABLE | regs::SM_ENABLE_FAMILY,
                    0,
                )
                .await;
                let (al, alc) = bridge::drive_state(&maindevice, st3, regs::AL_SAFEOP).await;
                println!("EL2262 SAFEOP: {}", bridge::verdict(al, alc));
                let (al, alc) = bridge::drive_state(&maindevice, st3, regs::AL_OP).await;
                println!("EL2262 OP:     {}", bridge::verdict(al, alc));
                if al & regs::AL_STATE_MASK != regs::AL_OP || al & regs::AL_ERROR != 0 {
                    return Err(format!("EL2262 OP rejected: {}", bridge::verdict(al, alc)).into());
                }
                el2262 = Some(st3);
            }
        }
        if !found {
            println!("--el2262 given but no EL2262 on bus; reference output disabled");
        }
    }

    // ---- Optional EL1252 DC latch timestamp readback ----
    let mut el1252: Option<u16> = None;
    if cfg.el1252 {
        let mut found = false;
        for (idx, subdevice) in group.iter(&maindevice).enumerate() {
            if subdevice.name().contains("EL1252") {
                found = true;
                let sd4 = group.subdevice(&maindevice, idx)?;
                let st4 = sd4.configured_address();
                println!("[{}] Found EL1252, station 0x{:04X} (latch readback)", idx, st4);
                let _ = ethercrab::Command::fpwr(st4, regs::sm_base(0) + regs::SM_OFF_START)
                    .send(&maindevice, 0x1000u16)
                    .await;
                let _ = ethercrab::Command::fpwr(st4, regs::sm_base(0) + regs::SM_OFF_LEN)
                    .send(&maindevice, 1u16)
                    .await;
                let _ = ethercrab::Command::fpwr(st4, regs::sm_base(0) + regs::SM_OFF_CTRL)
                    .send(&maindevice, regs::SM_CTRL_EL1252_INPUTS)
                    .await;
                let _ = regs::rmw_u8(
                    &maindevice,
                    st4,
                    regs::sm_base(0) + regs::SM_OFF_ENABLE,
                    regs::SM_ENABLE,
                    0,
                )
                .await;
                let (al, alc) = bridge::drive_state(&maindevice, st4, regs::AL_SAFEOP).await;
                println!("EL1252 SAFEOP: {}", bridge::verdict(al, alc));
                let (al, alc) = bridge::drive_state(&maindevice, st4, regs::AL_OP).await;
                println!("EL1252 OP:     {}", bridge::verdict(al, alc));
                if al & regs::AL_STATE_MASK != regs::AL_OP || al & regs::AL_ERROR != 0 {
                    return Err(format!("EL1252 OP rejected: {}", bridge::verdict(al, alc)).into());
                }
                el1252 = Some(st4);
            }
        }
        if !found {
            println!("--el1252 given but no EL1252 on bus; latch readback disabled");
        }
    }

    // ---- Pre-allocate everything, then promote to RT ----
    let mut tx_buf = vec![0u8; SM_TX_LEN];
    let mut rx_buf = vec![0u8; SM_RX_LEN];
    let mut jitter_st = OnlineStats::new("period_jitter", 100, -5_000_000, 5_000_000);
    let mut exec_st = OnlineStats::new("exec_time", 1_000, 0, 20_000_000);
    let mut rx_rtt_st = OnlineStats::new("fprd_rtt", 1_000, 0, 20_000_000);
    let mut tx_rtt_st = OnlineStats::new("fpwr_rtt", 1_000, 0, 20_000_000);

    println!("Promoting to RT: core={}, fifo={}, mlock=true", cfg.core, cfg.prio);
    if let Err(e) = rt::promote_current_thread(cfg.core, cfg.prio, true) {
        eprintln!("RT promotion failed (needs root, continuing anyway): {}", e);
    }
    rt::prefault_stack(256 * 1024);

    // ---- Real-time data loop ----
    println!(
        "\n>>> running: {} loop, echo payload, stats (Ctrl-C for graceful stop)",
        match cfg.mode {
            Mode::Ticker => "ticker",
            Mode::Probe => "probe (read-only, no echo)",
            Mode::Verify => "verify (CoE audit, no data loop)",
            Mode::RfmBridge => "rfm-bridge (EL6695 <-> RFM <-> remote)",
            Mode::RfmPing | Mode::RfmEcho | Mode::RfmRing => unreachable!(),
        }
    );
    let start = Instant::now();
    let mut ticker = rt::Ticker::new(cycle_ns);

    let mut cyc: u64 = 0;
    let mut tx_err: u64 = 0;
    let mut rx_err: u64 = 0;
    let mut err_streak: u64 = 0;
    let mut no_tc_seq: u64 = 0;
    let mut el2202_err: u64 = 0;
    let mut el2262_err: u64 = 0;
    let mut el1252_err: u64 = 0;
    let mut latch_pos0: u64 = 0;
    let mut latch_neg0: u64 = 0;
    let mut latch_pos1: u64 = 0;
    let mut latch_neg1: u64 = 0;
    let mut pos_edges: u64 = 0;
    let mut neg_edges: u64 = 0;
    let mut pos1_edges: u64 = 0;
    let mut el1252_st: [u8; 2] = [0, 0];
    let mut el1252_in: u8 = 0;
    let mut latch_interval_st = OnlineStats::new("el1252_ch1_edge_interval", 1_000, 0, 20_000_000);
    let mut latch_phase_st = OnlineStats::new("el1252_ch1_edge_phase", 100, 0, 1_000_000);
    let mut latch1_interval_st = OnlineStats::new("el1252_ch2_edge_interval", 1_000, 0, 20_000_000);
    let mut latch1_phase_st = OnlineStats::new("el1252_ch2_edge_phase", 100, 0, 1_000_000);
    let mut last_report = Instant::now();
    // J1900 data verification counters
    let mut prev_tc_seq: u64 = 0;
    let mut rx_stale: u64 = 0;      // tc_seq unchanged (stale read)
    let mut rx_nonmono: u64 = 0;     // tc_seq decreased (non-monotonic)
    let mut comp_err: u64 = 0;        // internal computation errors
    let mut payload_checked: u64 = 0; // J1900 payload verification cycles
    let mut payload_mismatch: u64 = 0; // J1900 payload mismatch bytes accumulated
    let mut prev_rx8 = [0u8; 8];       // spin-lock: previous payload (change detect)
    let mut last_fresh: Option<Instant> = None; // spin-lock: last fresh-data instant

    // Async pipelined FPWR: two alternating leaked tx slots so the in-flight
    // write future can hold a 'static buffer while the response is reaped
    // at the start of the next cycle.
    let tx_slot_a = Box::leak(Box::new([0u8; PAYLOAD_LEN]));
    let tx_slot_b = Box::leak(Box::new([0u8; PAYLOAD_LEN]));
    let mut tx_slot_toggle = false;
    let mut pending_tx: Option<
        std::pin::Pin<Box<dyn std::future::Future<Output = Result<(), ethercrab::error::Error>> + '_>>,
    > = None;
    let mut tx_late: u64 = 0;

    let mut aborted: Option<String> = None;

    // RFM bridge context (only opened in rfm-bridge mode).
    let mut rfm_ctx: Option<el6695_rt::rfm::Rfm> = if cfg.mode == Mode::RfmBridge {
        let r = el6695_rt::rfm::Rfm::open_path(cfg.rfm_dev.as_deref())?;
        println!(
            "rfm-bridge: BAR2 reflective mem = {} MB, node_id={}",
            r.mem.len() / (1024 * 1024),
            r.node_id
        );
        Some(r)
    } else {
        None
    };
    let mut bridge_req_seq: u64 = 0;
    let mut bridge_ok: u64 = 0;
    let mut bridge_timeouts: u64 = 0;
    let mut bridge_rtt_st = OnlineStats::new("rfm_hop_rtt", 100, 0, 20_000_000);

    loop {
        if STOP.load(Ordering::Relaxed) {
            println!("\n(stop requested, finishing cycle {})", cyc);
            break;
        }
        if cfg.duration_sec > 0 && start.elapsed().as_secs() >= cfg.duration_sec {
            break;
        }

        let mut rx_ok = true;
        let spin_mode = cfg.spin && cfg.mode == Mode::Ticker;
        let mut jitter_ns: i64 = 0;

        if spin_mode {
            // ---- Spin-lock: poll FPRD until the payload changes ----
            // Phase-locks processing to TwinCAT's frame arrival instead of a
            // free-running ticker. Timeout at 1.5 periods keeps the loop
            // alive (counted as stale) if the primary master stops.
            let spin_t0 = Instant::now();
            let timeout_ns = cycle_ns + cycle_ns / 2;
            let mut fresh = false;
            loop {
                match ethercrab::Command::fprd(station, SM_RX_ADDR)
                    .receive_slice(&maindevice, SM_RX_LEN as u16)
                    .await
                {
                    Ok(pdu) => {
                        let raw: &[u8] = &pdu;
                        let n = raw.len().min(SM_RX_LEN);
                        if n > RX_PREFIX {
                            let copy_len = (n - RX_PREFIX).min(SM_RX_PLD_LEN);
                            rx_buf[..copy_len].copy_from_slice(&raw[RX_PREFIX..RX_PREFIX + copy_len]);
                        }
                        if rx_buf[..8] != prev_rx8 {
                            prev_rx8.copy_from_slice(&rx_buf[..8]);
                            fresh = true;
                            break;
                        }
                    }
                    Err(_) => {
                        rx_err += 1;
                    }
                }
                if spin_t0.elapsed().as_nanos() as u64 > timeout_ns {
                    break;
                }
            }
            if !fresh {
                rx_ok = false;
                rx_stale += 1;
            }
            // jitter = deviation of fresh-data arrival interval from nominal
            let now = Instant::now();
            if let Some(prev) = last_fresh {
                let dt = now.duration_since(prev).as_nanos() as i64 - cycle_ns as i64;
                jitter_st.push(dt);
            }
            last_fresh = Some(now);
        } else {
            jitter_ns = ticker.wait_next_period();
            jitter_st.push(jitter_ns);
        }

        let t0 = Instant::now();
        cyc += 1;

        // Reap async FPWR kicked last cycle: the spin-detect above gave the
        // response ~one full cycle to arrive, so this should never block.
        if let Some(mut f) = pending_tx.take() {
            match futures_lite::future::poll_once(f.as_mut()).await {
                Some(Ok(_)) => {}
                Some(Err(_)) => {
                    tx_err += 1;
                }
                None => {
                    tx_late += 1;
                    let _ = f.await;
                }
            }
        }

        if !spin_mode {
            // 1. Read SM3 (TwinCAT data via bridge)
            let t_rx0 = Instant::now();
            match ethercrab::Command::fprd(station, SM_RX_ADDR)
                .receive_slice(&maindevice, SM_RX_LEN as u16)
                .await
            {
                Ok(pdu) => {
                    let raw: &[u8] = &pdu;
                    // Strip 2-byte diagnostic word prefix (0x1A01) from plain mode
                    let n = raw.len().min(SM_RX_LEN);
                    if n > RX_PREFIX {
                        let copy_len = (n - RX_PREFIX).min(SM_RX_PLD_LEN);
                        rx_buf[..copy_len].copy_from_slice(&raw[RX_PREFIX..RX_PREFIX + copy_len]);
                    }
                }
                Err(_) => {
                    rx_ok = false;
                    rx_err += 1;
                }
            }
            rx_rtt_st.push(t_rx0.elapsed().as_nanos() as i64);
        }
        if cyc <= 3 && rx_ok {
            let live: u32 = rx_buf[..SM_RX_PLD_LEN.min(256)].iter().map(|&b| if b != 0 { 1 } else { 0 }).sum();
            let show = SM_RX_LEN.min(64);
            println!("  rx_buf[0..{}]: {} non-zero (of {}) at cycle {}", show, live, SM_RX_PLD_LEN.min(256), cyc);
            print!("  rx_buf[..{}] hex: ", show);
            for b in rx_buf[..show].iter() { print!("{:02X}", b); }
            println!();
        }

        // 2. In ticker mode: prove J1900 processing by transforming TwinCAT's
        //    tc_seq instead of blind echo.  The function f(x)=x*7+3 is a
        //    non-trivial mathematical transform that no hardware bridge could
        //    produce — it proves J1900 read, computed AND wrote the data.
        //    The original tc_seq is preserved at offset 272 for TwinCAT to
        //    verify:  aIn1[0..7] == f(aOut1.tc_seq).
        //    In probe mode: skip TX (read-only observation).
        //    In rfm-bridge mode: forward the RX payload to the RFM mailbox and
        //    use the remote (J1900) processed REP as the TX payload, so the
        //    PLC-measured RTT covers the full CX->EL6695->D2000->RFM->x86 path.
        if cfg.mode == Mode::RfmBridge {
            if rx_ok {
                let rfm_ref = rfm_ctx.as_mut().expect("rfm not opened");
                let mem_ptr = rfm_ref.mem.as_ptr() as *mut u8;
                let n = SM_RX_PLD_LEN.min(el6695_rt::rfm::MB_MAX_PAYLOAD);
                let ts = el6695_rt::rfm::now_ns();
                bridge_req_seq = bridge_req_seq.wrapping_add(1);
                unsafe {
                    let mb = el6695_rt::rfm::Mailbox::at(mem_ptr, el6695_rt::rfm::MB_REQ_OFF);
                    el6695_rt::rfm::dev_write(mb.payload.as_mut_ptr(), &rx_buf[..n]);
                    mb.len = n as u32;
                    mb.ts_ns = ts;
                    el6695_rt::rfm::barrier();
                    mb.seq = bridge_req_seq;
                    el6695_rt::rfm::barrier();
                    mb.magic = el6695_rt::rfm::MB_MAGIC;
                }
                // Spin-poll REP for the remote-processed payload. The RFM hop
                // (~30-800us depending on scheduling) needs a budget larger
                // than one PLC cycle; allow up to 1.5ms so the full-chain RTT
                // is actually measured instead of timing out to local echo.
                let deadline = Instant::now() + Duration::from_nanos(1_500_000);
                let mut got = false;
                loop {
                    unsafe {
                        let rep = el6695_rt::rfm::Mailbox::at(mem_ptr, el6695_rt::rfm::MB_REP_OFF);
                        el6695_rt::rfm::barrier();
                        if rep.magic == el6695_rt::rfm::MB_MAGIC
                            && rep.seq == bridge_req_seq
                            && rep.len as usize >= SM_RX_PLD_LEN.min(el6695_rt::rfm::MB_MAX_PAYLOAD)
                        {
                            let m = SM_RX_PLD_LEN.min(el6695_rt::rfm::MB_MAX_PAYLOAD);
                            el6695_rt::rfm::dev_read(&mut tx_buf[..m], rep.payload.as_ptr());
                            // Apply f(x)=x*7+3 to the echoed frame's tc_seq
                            // here on the D2000 side (single transform point,
                            // robust): the remote just reflects the frame, the
                            // PLC's echo check needs [0..8] = f(rx tc_seq).
                            if m >= 280 {
                                let rx_seq = u64::from_le_bytes(rx_buf[..8].try_into().unwrap_or([0; 8]));
                                let processed = rx_seq.wrapping_mul(7).wrapping_add(3);
                                tx_buf[..8].copy_from_slice(&processed.to_le_bytes());
                            }
                            bridge_rtt_st.push(el6695_rt::rfm::now_ns().saturating_sub(rep.ts_ns) as i64);
                            got = true;
                            break;
                        }
                    }
                    if Instant::now() > deadline {
                        break;
                    }
                    core::hint::spin_loop();
                }
                if got {
                    bridge_ok += 1;
                } else {
                    bridge_timeouts += 1;
                    // Fall back to local echo so downstream sees valid data
                    let echo_len = SM_RX_PLD_LEN.min(tx_buf.len());
                    tx_buf[..echo_len].copy_from_slice(&rx_buf[..echo_len]);
                }
                payload_checked += 1;
            }
        }
        if cfg.mode == Mode::Ticker {
            if rx_ok {
                if PAYLOAD_LEN == 8 {
                    // 8B proof mode: per-byte transform tx[i] = rx[i] XOR 0x5A.
                    // Each byte is read, transformed and written back
                    // independently, so TwinCAT can verify all 8 bytes.
                    for i in 0..PAYLOAD_LEN {
                        tx_buf[i] = rx_buf[i] ^ 0x5A;
                    }
                    payload_checked += 1;
                } else {
                // Echo received data within available bounds
                let echo_len = SM_RX_PLD_LEN.min(tx_buf.len());
                tx_buf[..echo_len].copy_from_slice(&rx_buf[..echo_len]);
                // Read TwinCAT's sequence number
                let tc_seq = u64::from_le_bytes(
                    rx_buf[..8].try_into().unwrap_or([0; 8]),
                );
                // J1900 payload verification: pattern at [16..255] = (tc_seq + i) % 256
                // (only present in 1KB layout)
                if SM_RX_PLD_LEN >= 256 {
                    let mut mism: u32 = 0;
                    for i in 16..256 {
                        let expected = ((tc_seq + i as u64) % 256) as u8;
                        if rx_buf[i] != expected {
                            mism += 1;
                        }
                    }
                    payload_checked += 1;
                    payload_mismatch += mism as u64;
                    if PAYLOAD_LEN >= 280 {
                        // Write J1900 payload mismatch count for TwinCAT (offset 268)
                        tx_buf[268..272].copy_from_slice(&mism.to_le_bytes());
                    }
                }
                // Transform: f(x) = x * 7 + 3 (irreversible by simple subtraction)
                let processed = tc_seq.wrapping_mul(7).wrapping_add(3);
                tx_buf[..8].copy_from_slice(&processed.to_le_bytes());
                if PAYLOAD_LEN >= 280 {
                    // Preserve original for TwinCAT verification (offset 272)
                    tx_buf[272..280].copy_from_slice(&tc_seq.to_le_bytes());
                }
                }
            }
        }
        // TX (FPWR) must run for BOTH ticker and rfm-bridge modes (it was
        // previously gated inside Mode::Ticker, so rfm-bridge never wrote
        // SM2 and the PLC read zeros). Probe mode stays read-only.
        if cfg.mode != Mode::Probe {
            if PAYLOAD_LEN >= 16 {
                tx_buf[8..16].copy_from_slice(&cyc.to_le_bytes());
            }
            if PAYLOAD_LEN >= 280 {
                tx_buf[264..268].copy_from_slice(&(jitter_ns as i32).to_le_bytes());
            }

            if spin_mode {
                // Async pipelined FPWR: kick the frame onto the wire now
                // (poll_once drives it to sendable), reap response next
                // cycle. Frees ~240us of blocking write time per cycle.
                let p = if tx_slot_toggle {
                    tx_slot_b.as_mut_ptr()
                } else {
                    tx_slot_a.as_mut_ptr()
                };
                tx_slot_toggle = !tx_slot_toggle;
                let slot: &'static mut [u8] =
                    unsafe { std::slice::from_raw_parts_mut(p, PAYLOAD_LEN) };
                slot.copy_from_slice(&tx_buf[..PAYLOAD_LEN]);
                let mut fut = Box::pin(
                    ethercrab::Command::fpwr(station, SM_TX_ADDR).send(&maindevice, &slot[..]),
                );
                // Kick: drive the write future for up to ~20us so the frame
                // is marked sendable and the TX thread puts it on the wire
                // NOW. poll_once alone does not reach sendable state.
                let t_kick = Instant::now();
                let _ = futures_lite::future::or(
                    async { fut.as_mut().await.is_ok() },
                    async {
                        while t_kick.elapsed().as_micros() < 20 {
                            futures_lite::future::yield_now().await;
                        }
                        false
                    },
                )
                .await;
                pending_tx = Some(fut);
            } else {
                let t_tx0 = Instant::now();
                let tx_ok = ethercrab::Command::fpwr(station, SM_TX_ADDR)
                    .send(&maindevice, &tx_buf[..PAYLOAD_LEN])
                    .await
                    .is_ok();
                tx_rtt_st.push(t_tx0.elapsed().as_nanos() as i64);
                if !tx_ok {
                    tx_err += 1;
                }
            }
        }

        // Optional EL2202 scope waveform
        if let Some((st2, addr)) = el2202 {
            let v: u8 = if cfg.el2202_dual {
                (cyc & 1) as u8 * (regs::EL2202_CH1_DATA | regs::EL2202_CH2_DATA)
            } else {
                (cyc & 1) as u8 * regs::EL2202_CH1_DATA
            };
            if ethercrab::Command::fpwr(st2, addr)
                .send(&maindevice, v)
                .await
                .is_err()
            {
                el2202_err += 1;
            }
        }

        // Optional EL2262 DC-locked reference output
        if let Some(st3) = el2262 {
            let v: u8 = (cyc & 1) as u8;
            if ethercrab::Command::fpwr(st3, EL2262_OUT_ADDR)
                .send(&maindevice, v)
                .await
                .is_err()
            {
                el2262_err += 1;
            }
        }

        // Optional EL1252 latch readback
        if let Some(st4) = el1252 {
            if cyc % (cfg.el1252_every.max(1) as u64) == 0 {
                match ethercrab::Command::fprd(st4, 0x09AEu16)
                .receive_slice(&maindevice, 34)
                .await
            {
                Ok(pdu) => {
                    let e: &[u8] = &pdu;
                    if e.len() >= 34 {
                        el1252_st = [e[0], e[1]];
                        let pos0 = u64::from_le_bytes(e[2..10].try_into().unwrap());
                        let neg0 = u64::from_le_bytes(e[10..18].try_into().unwrap());
                        let pos1 = u64::from_le_bytes(e[18..26].try_into().unwrap());
                        let neg1 = u64::from_le_bytes(e[26..34].try_into().unwrap());
                        if pos0 != 0 && pos0 != latch_pos0 {
                            if latch_pos0 != 0 {
                                latch_interval_st.push((pos0 - latch_pos0) as i64);
                            }
                            latch_phase_st.push((pos0 % cycle_ns) as i64);
                            latch_pos0 = pos0;
                            pos_edges += 1;
                        }
                        if neg0 != 0 && neg0 != latch_neg0 {
                            latch_neg0 = neg0;
                            neg_edges += 1;
                        }
                        if pos1 != 0 && pos1 != latch_pos1 {
                            if latch_pos1 != 0 {
                                latch1_interval_st.push((pos1 - latch_pos1) as i64);
                            }
                            latch1_phase_st.push((pos1 % cycle_ns) as i64);
                            latch_pos1 = pos1;
                            pos1_edges += 1;
                        }
                        if neg1 != 0 && neg1 != latch_neg1 {
                            latch_neg1 = neg1;
                        }
                    }
                }
                Err(_) => el1252_err += 1,
            }
            }
        }

        if rx_ok && (cfg.mode == Mode::Probe || !tx_err > 0) {
            err_streak = 0;
        } else {
            err_streak += 1;
            if err_streak > MAX_ERR_STREAK {
                aborted = Some(format!(
                    "aborting: {} consecutive cycles with TX/RX errors (link dead?)",
                    err_streak
                ));
            }
        }

        let tc_seq = u64::from_le_bytes(
            rx_buf[..8].try_into().unwrap_or([0; 8]),
        );
        if tc_seq == 0 {
            no_tc_seq += 1;
        }
        // J1900 data verification
        if rx_ok && cfg.mode == Mode::Ticker {
            if tc_seq == prev_tc_seq {
                rx_stale += 1;
            }
            if prev_tc_seq != 0 && tc_seq != 0 {
                let delta = (tc_seq as i64).wrapping_sub(prev_tc_seq as i64);
                if delta < 0 {
                    rx_nonmono += 1;
                }
            }
            prev_tc_seq = tc_seq;
            // Self-consistency: verify proc == tc_seq * 7 + 3
            // (always true for 1KB ticker mode; 8B mode uses XOR echo instead)
            if PAYLOAD_LEN != 8 {
                let computed = tc_seq.wrapping_mul(7).wrapping_add(3);
                if cfg.mode == Mode::Ticker && tx_buf[..8] != computed.to_le_bytes() {
                    comp_err += 1; // internal computation error (should never fire)
                }
            }
        }

        exec_st.push(t0.elapsed().as_nanos() as i64);

        if !cfg.quiet && last_report.elapsed() >= Duration::from_millis(1000) {
            if let Some(st4) = el1252 {
                if let Ok(b) = ethercrab::Command::fprd(st4, 0x1000u16)
                    .receive::<u8>(&maindevice)
                    .await
                {
                    el1252_in = b;
                }
            }
            let proc_val = tc_seq.wrapping_mul(7).wrapping_add(3);
            let stale_pct = if cyc > 0 { (rx_stale * 100 / cyc) } else { 0 };
            let mut dc_info = String::new();
            if let Some((dc0, mono0)) = dc_ref {
                if let Ok(dc_now) = ethercrab::Command::fprd(station, regs::DC_SYSTEM_TIME)
                    .receive::<u64>(&maindevice)
                    .await
                {
                    let mono_now = rt::now_ns();
                    let drift = (dc_now.wrapping_sub(dc0)) as i64 - (mono_now - mono0) as i64;
                    dc_info = format!(
                        " dcph={} dcdr={}ns",
                        dc_now % cycle_ns,
                        drift
                    );
                }
            }
            println!(
                "S tms={} cyc={} raw={} proc={} txe={} rxe={} late={} stale={}% mono={} cerr={} mism={} e0={} e1={}{}",
                start.elapsed().as_millis(),
                cyc,
                tc_seq,
                proc_val,
                tx_err,
                rx_err,
                tx_late,
                stale_pct,
                rx_nonmono,
                comp_err,
                payload_mismatch,
                pos_edges,
                pos1_edges,
                dc_info,
            );
            if cfg.mode == Mode::RfmBridge {
                println!(
                    "  bridge: ok={} timeouts={} rfm_hop_avg={}us",
                    bridge_ok,
                    bridge_timeouts,
                    if bridge_ok > 0 { bridge_rtt_st.mean() / 1000.0 } else { 0.0 },
                );
            }
            last_report = Instant::now();
        }

        if aborted.is_some() {
            break;
        }
    }

    // Reap any in-flight async FPWR before reporting
    if let Some(f) = pending_tx.take() {
        let _ = f.await;
    }

    // ---- DC post-run: re-read firmware sync counters accumulated during the run ----
    if cfg.dc {
        println!("\n=== DC post-run sync counters (0x1C32 SM2 / 0x1C33 SM3) ===");
        for idx in [0x1C32u16, 0x1C33] {
            let mode = sd.sdo_read::<u16>(idx, 0x01).await.unwrap_or(0xFFFF);
            let missed = sd.sdo_read::<u16>(idx, 0x0B).await.unwrap_or(0xFFFF);
            let exceeded = sd.sdo_read::<u16>(idx, 0x0C).await.unwrap_or(0xFFFF);
            let serr = sd.sdo_read::<u8>(idx, 0x20).await.unwrap_or(0xFF);
            println!(
                "  0x{:04X}: sync_mode={} sm_event_missed={} cycle_exceeded={} sync_error={}",
                idx, mode, missed, exceeded, serr
            );
        }
    }

    // ---- Final report ----
    println!("\n=== timing stats ===");
    jitter_st.report(1000, "us");
    exec_st.report(1000, "us");
    rx_rtt_st.report(1000, "us");
    tx_rtt_st.report(1000, "us");
    if cfg.mode == Mode::RfmBridge {
        bridge_rtt_st.report(1000, "us");
        println!(
            "rfm-bridge: ok={} timeouts={} (rep_miss_rate={:.2}%)",
            bridge_ok,
            bridge_timeouts,
            if bridge_ok + bridge_timeouts > 0 {
                bridge_timeouts as f64 * 100.0 / (bridge_ok + bridge_timeouts) as f64
            } else {
                0.0
            }
        );
    }
    if el1252.is_some() {
        println!("=== EL1252 latch: rising-edge intervals and phase in cycle ===");
        latch_interval_st.report(1000, "us");
        latch_phase_st.report(1000, "us");
        latch1_interval_st.report(1000, "us");
        latch1_phase_st.report(1000, "us");
        println!(
            "el1252: ch1 pos_edges={}, ch1 neg_edges={}, ch2 pos_edges={}, read_err={}",
            pos_edges, neg_edges, pos1_edges, el1252_err
        );
    }
    println!("=== J1900 data verification ===");
    println!("  tc_seq: stale={} ({}%) nonmono={} comp_err={}", rx_stale,
        if cyc > 0 { rx_stale * 100 / cyc } else { 0 },
        rx_nonmono, comp_err);
    println!("  payload: checked={} mismatch={}", payload_checked, payload_mismatch);
    println!(
        "\n=== done: {} cycles in {:.1}s, tx_err={}, rx_err={}, no_tc_seq={}, el2202_err={}, el2262_err={} ===",
        cyc,
        start.elapsed().as_secs_f64(),
        tx_err,
        rx_err,
        no_tc_seq,
        el2202_err,
        el2262_err
    );

    if let Some(msg) = aborted {
        return Err(msg.into());
    }
    Ok(())
}

// =====================================================================
// RFM (PCI-5565 reflective memory) modes
// =====================================================================

/// RFM ping/echo modes: pure reflective-memory latency, no EtherCAT.
async fn rfm_mode_main(cfg: &Config) -> Result<(), Box<dyn std::error::Error>> {
    let rfm = el6695_rt::rfm::Rfm::open_path(cfg.rfm_dev.as_deref())?;
    println!(
        "rfm: {} BAR reflective mem = {} MB, node_id={}",
        cfg.rfm_dev.as_deref().unwrap_or("/dev/rfm2g0(default)"),
        rfm.mem.len() / (1024 * 1024),
        rfm.node_id
    );
    let mem_ptr = rfm.mem.as_ptr() as *mut u8;

    // RT-rotate this thread onto its own core (both ping and echo spin;
    // sharing a core with the peer would add ~0.5ms scheduling latency).
    if let Err(e) = rt::promote_current_thread(cfg.core, cfg.prio, true) {
        eprintln!("RT promotion failed (continuing): {}", e);
    }

    // Clear both mailboxes so a stale REQ/REP from a previous run cannot
    // poison the first cycle's RTT sample.
    unsafe {
        let zero = [0u8; 32];
        el6695_rt::rfm::dev_write(mem_ptr.add(el6695_rt::rfm::MB_REQ_OFF), &zero);
        el6695_rt::rfm::dev_write(mem_ptr.add(el6695_rt::rfm::MB_REP_OFF), &zero);
        el6695_rt::rfm::barrier();
    }

    let mut rtt = OnlineStats::new("rfm_rtt", 100, 0, 5_000_000);
    let mut period = OnlineStats::new("period", 100, 0, 10_000_000);
    let cycle_ns = cfg.cycle_us * 1000;
    let mut seq: u64 = 0;
    let start = Instant::now();
    let mut last_rep_seq: u64 = 0;
    let mut timeouts: u64 = 0;
    let mut dseq_sum: u64 = 0;
    let mut last_report = Instant::now();
    let mut last_cyc: u64 = 0;

    match cfg.mode {
        Mode::RfmPing => {
            let mut next = Instant::now();
            while start.elapsed().as_secs() < cfg.duration_sec || cfg.duration_sec == 0 {
                if STOP.load(Ordering::Relaxed) {
                    break;
                }
                // ---- write REQ ----
                let ts = el6695_rt::rfm::now_ns();
                unsafe {
                    let mb = el6695_rt::rfm::Mailbox::at(mem_ptr, el6695_rt::rfm::MB_REQ_OFF);
                    let req_seq = seq.wrapping_add(1);
                    let pattern = [(seq & 0xFF) as u8; 64];
                    el6695_rt::rfm::dev_write(mb.payload.as_mut_ptr(), &pattern);
                    mb.len = 64;
                    mb.ts_ns = ts;
                    el6695_rt::rfm::barrier();
                    mb.seq = req_seq;
                    el6695_rt::rfm::barrier();
                    mb.magic = el6695_rt::rfm::MB_MAGIC;
                }
                // ---- spin-wait REP ----
                let deadline = Instant::now() + Duration::from_nanos(cycle_ns / 2);
                let mut got = false;
                loop {
                    unsafe {
                        let rep = el6695_rt::rfm::Mailbox::at(mem_ptr, el6695_rt::rfm::MB_REP_OFF);
                        el6695_rt::rfm::barrier();
                        if rep.magic == el6695_rt::rfm::MB_MAGIC && rep.seq != last_rep_seq {
                            dseq_sum += seq.wrapping_add(1).saturating_sub(rep.seq);
                            last_rep_seq = rep.seq;
                            let rtt_ns = el6695_rt::rfm::now_ns().saturating_sub(rep.ts_ns);
                            rtt.push(rtt_ns as i64);
                            got = true;
                            break;
                        }
                    }
                    if Instant::now() > deadline {
                        break;
                    }
                    core::hint::spin_loop();
                }
                if !got {
                    timeouts += 1;
                }
                seq = seq.wrapping_add(1);
                period.push(next.elapsed().as_nanos() as i64);
                next += Duration::from_nanos(cycle_ns);
                let now = Instant::now();
                if next > now {
                    spin_sleep(next - now);
                } else {
                    next = now;
                }
                if last_report.elapsed().as_secs() >= 1 {
                    println!(
                        "S tms={} cyc={} rtt_n={} timeouts={}",
                        start.elapsed().as_millis(),
                        seq,
                        rtt_n(&rtt),
                        timeouts
                    );
                    last_cyc = seq;
                    last_report = Instant::now();
                }
            }
        }
        Mode::RfmEcho => {
            println!("rfm: echo mode (reflect REQ -> REP), ctrl-C to stop");
            let mut last_req_seq: u64 = 0;
            while !STOP.load(Ordering::Relaxed) {
                unsafe {
                    let req = el6695_rt::rfm::Mailbox::at(mem_ptr, el6695_rt::rfm::MB_REQ_OFF);
                    el6695_rt::rfm::barrier();
                    if req.magic == el6695_rt::rfm::MB_MAGIC && req.seq != last_req_seq {
                        let n = (req.len as usize).min(el6695_rt::rfm::MB_MAX_PAYLOAD);
                        let cur_seq = req.seq;
                        let cur_ts = req.ts_ns;
                        let cur_len = req.len;
                        let mut buf = [0u8; el6695_rt::rfm::MB_MAX_PAYLOAD];
                        el6695_rt::rfm::dev_read(&mut buf[..n], req.payload.as_ptr());
                        let rep = el6695_rt::rfm::Mailbox::at(mem_ptr, el6695_rt::rfm::MB_REP_OFF);
                        el6695_rt::rfm::dev_write(rep.payload.as_mut_ptr(), &buf[..n]);
                        // Echo is a dumb reflector: f(x) is applied by the
                        // D2000 bridge (single transform point, avoids double
                        // transform).
                        rep.len = cur_len;
                        rep.ts_ns = cur_ts;
                        el6695_rt::rfm::barrier();
                        rep.seq = cur_seq;
                        el6695_rt::rfm::barrier();
                        rep.magic = el6695_rt::rfm::MB_MAGIC;
                        last_req_seq = cur_seq;
                        seq = seq.wrapping_add(1);
                    }
                }
                core::hint::spin_loop();
                if last_report.elapsed().as_secs() >= 1 {
                    println!("S tms={} echoed={}", start.elapsed().as_millis(), seq);
                    last_report = Instant::now();
                }
            }
        }
        _ => unreachable!(),
    }

    println!("\n=== rfm done: {} ops in {:.1}s, timeouts={} ===", seq, start.elapsed().as_secs_f64(), timeouts);
    print_stats(&rtt);
    print_stats(&period);
    let _ = last_cyc;
    Ok(())
}

fn rtt_n(s: &OnlineStats) -> u64 {
    s.len()
}

#[inline]
fn spin_sleep(d: Duration) {
    let target = Instant::now() + d;
    while Instant::now() < target {
        if d > Duration::from_micros(200) {
            std::thread::yield_now();
        } else {
            core::hint::spin_loop();
        }
    }
}

fn print_stats(s: &OnlineStats) {
    s.report(1000, "us");
}