diagnose(pipeline): wire-capture closes the case - vendored ethercrab response cross-talk

Instrumented passive AF_PACKET sniffer (--passive, LE addr fix critical):
- DIRECT: SM2 writes 100% accepted (wkc1=10842/wkc0=10842, request+response pairs)
- PIPELINE: only 36% accepted (wkc1=3678/wkc0=16638) - SM2 3-buffer handshake
  rejects premature back-to-back writes
- Both modes: identical wire traffic (~830 writes/s, live payloads, 828B frames)

Mechanism: concurrent fprd+fpwr futures -> vendored ethercrab response routing
completes the fpwr future with the fprd's wkc=1 response (bogus Ok, ferr=0)
-> write slot freed early -> next write issued while ESC busy -> wkc=0 discard
-> buffer desync -> bridge reads empty buffers -> primary image zero.

Prior 'executor starvation' theory DISPROVEN by wire capture. Fix scope:
patch vendored ethercrab PDU index matching (or serialize EC ops - no gain).

src/main.rs

@@ -208,6 +208,7 @@ fn parse_args() -> Result<Config, String> {
             "--adopt-pure" => cfg.adopt_pure = true,
             "--sync-ts" => cfg.sync_ts = true,
             "--dc-discipline" => cfg.dc_discipline = true,
+            "--passive" => {},
             "--sm3-len" => { cfg.sm3_len_force = val("--sm3-len")?.parse().map_err(|_| "bad --sm3-len")?; },
             "--dc" => cfg.dc = true,
             "--iouring" => cfg.iouring = true,
@@ -338,6 +339,77 @@ async fn run(cfg: &Config) -> Result<(), Box<dyn std::error::Error>> {
         }
     });
 
+
+    if std::env::args().any(|a| a == "--passive") {
+        use std::ffi::CString;
+        let ciface = CString::new(cfg.iface.as_bytes()).unwrap();
+        let fd = unsafe { libc::socket(libc::AF_PACKET, libc::SOCK_RAW, (libc::ETH_P_ALL as u16).to_be() as i32) };
+        assert!(fd >= 0, "raw socket failed");
+        let ifidx = unsafe { libc::if_nametoindex(ciface.as_ptr()) };
+        assert!(ifidx > 0);
+        let mut sll: libc::sockaddr_ll = unsafe { std::mem::zeroed() };
+        sll.sll_family = libc::AF_PACKET as u16;
+        sll.sll_protocol = (libc::ETH_P_ALL as u16).to_be();
+        sll.sll_ifindex = ifidx as i32;
+        let r = unsafe { libc::bind(fd, &sll as *const libc::sockaddr_ll as *const libc::sockaddr, std::mem::size_of::<libc::sockaddr_ll>() as u32) };
+        assert!(r == 0, "bind failed");
+        let tv = libc::timeval { tv_sec: 0, tv_usec: 50_000 };
+        unsafe { libc::setsockopt(fd, libc::SOL_SOCKET, libc::SO_RCVTIMEO, &tv as *const _ as *const libc::c_void, std::mem::size_of::<libc::timeval>() as u32) };
+        let t0 = std::time::Instant::now();
+        let (mut n_fpwr, mut n_fprd, mut n_apwr, mut n_aprd, mut n_other) = (0u64, 0u64, 0u64, 0u64, 0u64);
+        let (mut n_sm2, mut n_other_addr) = (0u64, 0u64);
+        let (mut n_sm2_wkc1, mut n_sm2_wkc0) = (0u64, 0u64);
+        let mut samples = 0;
+        println!("PASSIVE: listening 12s for EtherCAT frames (FPWR cmd=0x05 to 0x1C00)?");
+        while t0.elapsed() < std::time::Duration::from_secs(12) {
+            let mut buf = [0u8; 2048];
+            let n = unsafe { libc::recv(fd, buf.as_mut_ptr() as *mut libc::c_void, buf.len(), 0) };
+            if n <= 0 { continue; }
+            let n = n as usize;
+            if n < 27 { continue; }
+            let et = u16::from_be_bytes([buf[12], buf[13]]);
+            if et != 0x88A4 { continue; }
+            let cmd = buf[16];
+            match cmd {
+                0x05 => {
+                    n_fpwr += 1;
+                    let ado = u16::from_le_bytes([buf[20], buf[21]]);
+                    if ado == 0x1C00 {
+                        n_sm2 += 1;
+                        if n >= 828 {
+                            let wkc = u16::from_le_bytes([buf[826], buf[827]]);
+                            if wkc == 1 { n_sm2_wkc1 += 1; } else { n_sm2_wkc0 += 1; }
+                        }
+                        if samples < 6 && n >= 42 {
+                            let p = &buf[26..42];
+                            let hex: Vec<String> = p.iter().map(|b| format!("{:02X}", b)).collect();
+                            let wkc = if n >= 828 { u16::from_le_bytes([buf[826], buf[827]]) } else { 0 };
+                            println!("  SM2 pay={}.. wkc={}", &hex.join(" ")[..23], wkc);
+                            samples += 1;
+                        }
+                    } else {
+                        n_other_addr += 1;
+                        if n_other_addr % 5000 == 1 {
+                            let adp = u16::from_le_bytes([buf[18], buf[19]]);
+                        }
+                    }
+                }
+                0x04 => {
+                    n_fprd += 1;
+                    if n_fprd % 4000 == 1 {
+                        let adp = u16::from_le_bytes([buf[18], buf[19]]);
+                        let ado = u16::from_le_bytes([buf[20], buf[21]]);
+                    }
+                }
+                0x02 => n_apwr += 1,
+                0x01 => n_aprd += 1,
+                _ => n_other += 1,
+            }
+        }
+        println!("PASSIVE done: FPWR={} (toSM2={} wkc1={} wkc0={} otherAddr={}) FPRD={}", n_fpwr, n_sm2, n_sm2_wkc1, n_sm2_wkc0, n_other_addr, n_fprd);
+        return Ok(());
+    }
+
     println!("Scanning segment (INIT -> PREOP)...");
     let group = maindevice
         .init_single_group::<MAX_SUBDEVICES, PDI_LEN>(ethercat_now)
@@ -461,7 +533,7 @@ async fn run(cfg: &Config) -> Result<(), Box<dyn std::error::Error>> {
                 let len = hdr & 0x0FFF; let typ = hdr >> 12;
                 let cmd = f[16]; let adp = u16::from_be_bytes([f[18], f[19]]);
                 let ado = u16::from_be_bytes([f[20], f[21]]);
-                extra = format!(" || ecat len={} type={} cmd=0x{:02X} adp={:#06X} ado={:#06X} wkc={}", len, typ, cmd, adp, ado, f[13 + len as usize]);
+                extra = format!(" || ecat len={} type={} cmd=0x{:02X} adp={:04X} ado={:04X} wkc={}", len, typ, cmd, adp, ado, f[13 + len as usize]);
             }
             println!("{} [{}B] {}{}", tag, n, hex.join(" "), extra);
         };
@@ -1557,7 +1629,7 @@ async fn run(cfg: &Config) -> Result<(), Box<dyn std::error::Error>> {
     // ---- rfm-async: event-driven forwarder, no tick ----
     if cfg.mode == Mode::RfmAsync {
         let rfm = rfm_ctx.take().expect("rfm not opened");
-        return rfm_async_loop(&maindevice, station, rfm, cfg.duration_sec, cfg.dc_discipline).await;
+        return rfm_async_loop(&maindevice, station, rfm, cfg.duration_sec, cfg.dc_discipline, &sd).await;
     }
 
     loop {
@@ -2320,6 +2392,7 @@ async fn rfm_async_loop(
     mut rfm: el6695_rt::rfm::Rfm,
     duration_sec: u64,
     dc_discipline: bool,
+    sd: &ethercrab::SubDeviceRef<'_, &ethercrab::SubDevice>,
 ) -> Result<(), Box<dyn std::error::Error>> {
     use el6695_rt::rfm::*;
     println!("rfm-async: event-driven forwarding, no tick (Ctrl-C to stop)");
@@ -2388,6 +2461,7 @@ async fn rfm_async_loop(
     let mut phase_est: i64 = 0; // estimated TwinCAT frame boundary phase (ns within 1ms)
     let mut wphase_st = OnlineStats::new("write_phase_us", 200, -1_500_000, 1_500_000);
     let mut j2cx_pending: Option<std::time::Instant> = None;
+    let mut last_f640_print: u64 = 0;
 
     while !STOP.load(Ordering::Relaxed) {
         if duration_sec > 0 && start.elapsed().as_secs() >= duration_sec {
@@ -2629,6 +2703,12 @@ async fn rfm_async_loop(
         }
         { let n = Instant::now(); seg_fprd_us = n.duration_since(cursor).as_micros() as u64; cursor = n; }
 
+                // F640 remote-write-cycle counter (independent of reap path)
+                if n_fpwr > 0 && n_fpwr % 512 == 0 && last_f640_print != n_fpwr {
+                    last_f640_print = n_fpwr;
+                    let c = sd.sdo_read::<u16>(0xF640, 1).await;
+                    println!("  F640CNT n={} count={:?}", n_fpwr, c);
+                }
         // 6. reap FPWR completion
         if let Some(f) = pending_fpwr.as_mut() {
             if let Some(res) = futures_lite::future::poll_once(f.as_mut()).await {
@@ -2639,6 +2719,7 @@ async fn rfm_async_loop(
                 if let Some(td) = j2cx_pending.take() {
                     j2cx_st.push(td.elapsed().as_nanos() as i64);
                 }
+
             }
         }
         { let n = Instant::now(); seg_fpwr_us = n.duration_since(cursor).as_micros() as u64; cursor = n; }