byteor_pipeline_exec/

single_ring.rs

//! SingleRing SHM executor and runtime.

use alloc::vec::Vec;

use byteor_pipeline_kernel::step::TransformOutcome;
use byteor_pipeline_spec::{validate_v1, PipelineSpecV1, StageOpV1};

use crate::ExecError;
use crate::{resolve_or_err, ResolvedStage, StageResolver};
use crate::{RoleKind, ThreadInitContext, ThreadInitHook};

type InspectFn = fn(&[u8]);
type ProcessFn = fn(&mut [u8], u8);

fn op_noop(_data: &[u8]) {}

fn op_add_u8(data: &mut [u8], delta: u8) {
    for b in data {
        *b = b.wrapping_add(delta);
    }
}

fn clamp_len(len: u32, max: usize) -> usize {
    (len as usize).min(max)
}

enum StageExecOp {
    BuiltIn(StageOpV1),
    Resolved { key: String, stage: ResolvedStage },
}

fn resolve_stage_op(
    op: StageOpV1,
    resolver: Option<&dyn StageResolver>,
) -> Result<StageExecOp, ExecError> {
    match op {
        StageOpV1::ResolverKey { stage } => {
            let Some(resolver) = resolver else {
                return Err(ExecError::new("missing stage resolver"));
            };
            let resolved = resolve_or_err(resolver, &stage).map_err(|e| {
                ExecError::with_context("resolve stage failed", format!("stage={stage}: {e}"))
            })?;
            Ok(StageExecOp::Resolved {
                key: stage,
                stage: resolved,
            })
        }
        other => Ok(StageExecOp::BuiltIn(other)),
    }
}

/// Wait/backoff policy presets for SingleRing hot loops.
///
/// These presets are intentionally minimal and deterministic:
/// - `MaxPerfSpin`: spin-first strategy for isolated CPUs.
/// - `StdBackoff`: cooperative backoff for non-HFT deployments.
/// - `Yield`: cooperative yield-only waiting for shared-host runs.
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub enum WaitPreset {
    /// Max-perf preset: spin-first waiting.
    MaxPerfSpin,
    /// Default backoff preset (std-only): yields/sleeps as needed.
    StdBackoff,
    /// Yield-only preset (std-only).
    Yield,
}

struct JoinBarrier {
    deps: Vec<byteor_pipeline_backings_shm::GatingBarrier>,
}

impl byteor_pipeline_kernel::SeqBarrier for JoinBarrier {
    fn available(&self) -> u64 {
        self.deps.iter().map(|b| b.available()).min().unwrap_or(0)
    }
}

enum StageBarrier {
    Cursor(byteor_pipeline_backings_shm::CursorBarrier),
    Gating(byteor_pipeline_backings_shm::GatingBarrier),
    Join(JoinBarrier),
}

impl byteor_pipeline_kernel::SeqBarrier for StageBarrier {
    fn available(&self) -> u64 {
        match self {
            StageBarrier::Cursor(b) => b.available(),
            StageBarrier::Gating(b) => b.available(),
            StageBarrier::Join(b) => b.available(),
        }
    }
}

enum WaitHolder {
    Spin(indexbus_core::SpinWait),
    Backoff(indexbus_core::StdBackoff),
    Yield(indexbus_core::YieldWait),
}

impl WaitHolder {
    #[inline]
    fn wait(&mut self) {
        match self {
            WaitHolder::Spin(w) => byteor_pipeline_kernel::WaitStrategy::wait(w),
            WaitHolder::Backoff(w) => byteor_pipeline_kernel::WaitStrategy::wait(w),
            WaitHolder::Yield(w) => byteor_pipeline_kernel::WaitStrategy::wait(w),
        }
    }
}

fn current_allowed_cpus() -> Vec<usize> {
    #[cfg(target_os = "linux")]
    {
        affinity::get_thread_affinity().unwrap_or_default()
    }

    #[cfg(not(target_os = "linux"))]
    {
        Vec::new()
    }
}

fn main_thread_cpu(cpus: &[usize]) -> Option<usize> {
    if cpus.len() > 1 {
        cpus.first().copied()
    } else {
        None
    }
}

fn worker_thread_cpu(cpus: &[usize], worker_idx: usize) -> Option<usize> {
    if cpus.len() <= 1 {
        return None;
    }

    Some(cpus[1 + (worker_idx % (cpus.len() - 1))])
}

fn runtime_stage_cpu(cpus: &[usize], stage_idx: usize) -> Option<usize> {
    if cpus.is_empty() {
        return None;
    }

    Some(cpus[stage_idx % cpus.len()])
}

struct ThreadAffinityGuard {
    previous: Vec<usize>,
    active: bool,
}

impl ThreadAffinityGuard {
    fn inactive() -> Self {
        Self {
            previous: Vec::new(),
            active: false,
        }
    }

    fn pin(cpu: Option<usize>) -> Result<Self, ExecError> {
        #[cfg(target_os = "linux")]
        {
            let Some(cpu) = cpu else {
                return Ok(Self::inactive());
            };

            let previous = affinity::get_thread_affinity().map_err(|e| {
                ExecError::with_context("read thread affinity failed", e.to_string())
            })?;

            if previous.len() == 1 && previous[0] == cpu {
                return Ok(Self {
                    previous,
                    active: false,
                });
            }

            affinity::set_thread_affinity([cpu]).map_err(|e| {
                ExecError::with_context("set thread affinity failed", format!("cpu={cpu}: {e}"))
            })?;

            Ok(Self {
                previous,
                active: true,
            })
        }

        #[cfg(not(target_os = "linux"))]
        {
            let _ = cpu;
            Ok(Self::inactive())
        }
    }
}

impl Drop for ThreadAffinityGuard {
    fn drop(&mut self) {
        #[cfg(target_os = "linux")]
        if self.active && !self.previous.is_empty() {
            let _ = affinity::set_thread_affinity(&self.previous);
        }
    }
}

fn run_stage_loop_stoppable<W: byteor_pipeline_kernel::WaitStrategy>(
    stage: &mut byteor_pipeline_backings_shm::ShmSequencedSlotsStage,
    barrier: &StageBarrier,
    wait: &mut W,
    stop: &std::sync::atomic::AtomicBool,
    op: &mut StageExecOp,
    iterations: usize,
    release: bool,
) -> Result<(), ExecError> {
    match op {
        StageExecOp::BuiltIn(op) => match op {
            StageOpV1::Identity => {
                let inspect_fn: InspectFn = op_noop;
                for _ in 0..iterations {
                    if stop.load(std::sync::atomic::Ordering::Relaxed) {
                        break;
                    }

                    let res =
                        stage.inspect_next_stoppable(barrier, wait, release, stop, |data, _len| {
                            inspect_fn(data);
                            Ok(())
                        });

                    match res {
                        Ok(_) => {}
                        Err(byteor_pipeline_backings_shm::SequencedSlotsError::Stopped) => break,
                        Err(e) => return Err(ExecError::with_source("stage failed", e)),
                    }
                }
            }
            StageOpV1::AddU8 { delta } => {
                let delta = *delta;
                let process_fn: ProcessFn = op_add_u8;
                for _ in 0..iterations {
                    if stop.load(std::sync::atomic::Ordering::Relaxed) {
                        break;
                    }

                    let res =
                        stage.process_next_stoppable(barrier, wait, release, stop, |data, len| {
                            let n = clamp_len(*len, data.len());
                            process_fn(&mut data[..n], delta);
                            Ok(())
                        });

                    match res {
                        Ok(_) => {}
                        Err(byteor_pipeline_backings_shm::SequencedSlotsError::Stopped) => break,
                        Err(e) => return Err(ExecError::with_source("stage failed", e)),
                    }
                }
            }
            StageOpV1::ResolverKey { stage } => {
                return Err(ExecError::with_context(
                    "unresolved stage",
                    format!("stage={stage}"),
                ));
            }
        },
        StageExecOp::Resolved {
            key,
            stage: resolved,
        } => {
            let mut scratch = vec![0u8; indexbus_core::INDEXBUS_SLOT_DATA_SIZE];

            for _ in 0..iterations {
                if stop.load(std::sync::atomic::Ordering::Relaxed) {
                    break;
                }

                let mut stage_err: Option<ExecError> = None;

                match resolved {
                    ResolvedStage::Inspect(f) => {
                        let res = stage.inspect_next_stoppable(
                            barrier,
                            wait,
                            release,
                            stop,
                            |data, len| {
                                let n = clamp_len(len, data.len());
                                if let Err(e) = f(&data[..n]) {
                                    stage_err = Some(ExecError::with_context(
                                        "resolved stage failed",
                                        format!("stage={key}: {e}"),
                                    ));
                                    return Err(
                                        byteor_pipeline_backings_shm::SequencedSlotsError::Stopped,
                                    );
                                }
                                Ok(())
                            },
                        );

                        match res {
                            Ok(_) => {}
                            Err(byteor_pipeline_backings_shm::SequencedSlotsError::Stopped) => {
                                if let Some(e) = stage_err {
                                    return Err(e);
                                }
                                break;
                            }
                            Err(e) => return Err(ExecError::with_source("stage failed", e)),
                        }
                    }
                    _ => {
                        let res = stage.process_next_stoppable(
                            barrier,
                            wait,
                            release,
                            stop,
                            |data, len| {
                                let n_in = clamp_len(*len, data.len());
                                let input = &data[..n_in];
                                let out = &mut scratch[..n_in];

                                let mut set_err = |e: ExecError| {
                                    stage_err = Some(e);
                                    Err(byteor_pipeline_backings_shm::SequencedSlotsError::Stopped)
                                };

                                match resolved {
                                    ResolvedStage::MapOk(f) => {
                                        let n = f(input, out);
                                        if n > input.len() {
                                            return set_err(ExecError::with_context(
                                                "resolved stage output too large",
                                                format!("stage={key} out={n} in={}", input.len()),
                                            ));
                                        }
                                        data[..n].copy_from_slice(&out[..n]);
                                        *len = n as u32;
                                        Ok(())
                                    }
                                    ResolvedStage::Map(f) => match f(input, out) {
                                        Ok(n) => {
                                            if n > input.len() {
                                                return set_err(ExecError::with_context(
                                                    "resolved stage output too large",
                                                    format!(
                                                        "stage={key} out={n} in={} (fallible)",
                                                        input.len()
                                                    ),
                                                ));
                                            }
                                            data[..n].copy_from_slice(&out[..n]);
                                            *len = n as u32;
                                            Ok(())
                                        }
                                        Err(e) => set_err(ExecError::with_context(
                                            "resolved stage failed",
                                            format!("stage={key}: {e}"),
                                        )),
                                    },
                                    ResolvedStage::Filter(f) => match f(input) {
                                        Ok(true) => Ok(()),
                                        Ok(false) => {
                                            *len = 0;
                                            Ok(())
                                        }
                                        Err(e) => set_err(ExecError::with_context(
                                            "resolved stage failed",
                                            format!("stage={key}: {e}"),
                                        )),
                                    },
                                    ResolvedStage::Inspect(f) => match f(input) {
                                        Ok(()) => Ok(()),
                                        Err(e) => set_err(ExecError::with_context(
                                            "resolved stage failed",
                                            format!("stage={key}: {e}"),
                                        )),
                                    },
                                    ResolvedStage::StatefulTransform(st) => match st.call(input, out)
                                    {
                                        Ok(TransformOutcome::Drop) => {
                                            *len = 0;
                                            Ok(())
                                        }
                                        Ok(TransformOutcome::ForwardInput) => Ok(()),
                                        Ok(TransformOutcome::ForwardOutput(n)) => {
                                            if n > input.len() {
                                                return set_err(ExecError::with_context(
                                                    "resolved stage output too large",
                                                    format!(
                                                        "stage={key} out={n} in={} (stateful transform)",
                                                        input.len()
                                                    ),
                                                ));
                                            }
                                            data[..n].copy_from_slice(&out[..n]);
                                            *len = n as u32;
                                            Ok(())
                                        }
                                        Ok(_) => set_err(ExecError::new(
                                            "unsupported transform outcome (non-exhaustive)",
                                        )),
                                        Err(e) => set_err(ExecError::with_context(
                                            "resolved stage failed",
                                            format!("stage={key}: {e}"),
                                        )),
                                    },
                                    ResolvedStage::Stateful(st) => match st.call(input, out) {
                                        Ok(n) => {
                                            if n > input.len() {
                                                return set_err(ExecError::with_context(
                                                    "resolved stage output too large",
                                                    format!(
                                                        "stage={key} out={n} in={} (stateful)",
                                                        input.len()
                                                    ),
                                                ));
                                            }
                                            data[..n].copy_from_slice(&out[..n]);
                                            *len = n as u32;
                                            Ok(())
                                        }
                                        Err(e) => set_err(ExecError::with_context(
                                            "resolved stage failed",
                                            format!("stage={key}: {e}"),
                                        )),
                                    },
                                }
                            },
                        );

                        match res {
                            Ok(_) => {}
                            Err(byteor_pipeline_backings_shm::SequencedSlotsError::Stopped) => {
                                if let Some(e) = stage_err {
                                    return Err(e);
                                }
                                break;
                            }
                            Err(e) => return Err(ExecError::with_source("stage failed", e)),
                        }
                    }
                }
            }
        }
    }

    Ok(())
}

fn run_stage_loop_iters<W: byteor_pipeline_kernel::WaitStrategy>(
    stage: &mut byteor_pipeline_backings_shm::ShmSequencedSlotsStage,
    barrier: &StageBarrier,
    wait: &mut W,
    op: &mut StageExecOp,
    iterations: usize,
    release: bool,
) -> Result<(), ExecError> {
    match op {
        StageExecOp::BuiltIn(op) => match op {
            StageOpV1::Identity => {
                let inspect_fn: InspectFn = op_noop;
                for _ in 0..iterations {
                    stage
                        .inspect_next(barrier, wait, release, |data, _len| {
                            inspect_fn(data);
                            Ok(())
                        })
                        .map_err(|e| ExecError::with_source("stage failed", e))?;
                }
            }
            StageOpV1::AddU8 { delta } => {
                let delta = *delta;
                let process_fn: ProcessFn = op_add_u8;
                for _ in 0..iterations {
                    stage
                        .process_next(barrier, wait, release, |data, len| {
                            let n = clamp_len(*len, data.len());
                            process_fn(&mut data[..n], delta);
                            Ok(())
                        })
                        .map_err(|e| ExecError::with_source("stage failed", e))?;
                }
            }
            StageOpV1::ResolverKey { stage } => {
                return Err(ExecError::with_context(
                    "unresolved stage",
                    format!("stage={stage}"),
                ));
            }
        },
        StageExecOp::Resolved {
            key,
            stage: resolved,
        } => {
            let mut scratch = vec![0u8; indexbus_core::INDEXBUS_SLOT_DATA_SIZE];
            for _ in 0..iterations {
                let mut stage_err: Option<ExecError> = None;

                match resolved {
                    ResolvedStage::Inspect(f) => {
                        let res = stage.inspect_next(barrier, wait, release, |data, len| {
                            let n = clamp_len(len, data.len());
                            if let Err(e) = f(&data[..n]) {
                                stage_err = Some(ExecError::with_context(
                                    "resolved stage failed",
                                    format!("stage={key}: {e}"),
                                ));
                                return Err(
                                    byteor_pipeline_backings_shm::SequencedSlotsError::Stopped,
                                );
                            }
                            Ok(())
                        });

                        match res {
                            Ok(_) => {}
                            Err(byteor_pipeline_backings_shm::SequencedSlotsError::Stopped) => {
                                return Err(stage_err
                                    .unwrap_or_else(|| ExecError::new("resolved stage stopped")));
                            }
                            Err(e) => return Err(ExecError::with_source("stage failed", e)),
                        }
                    }
                    _ => {
                        let res = stage.process_next(barrier, wait, release, |data, len| {
                            let n_in = clamp_len(*len, data.len());
                            let input = &data[..n_in];
                            let out = &mut scratch[..n_in];

                            let mut set_err = |e: ExecError| {
                                stage_err = Some(e);
                                Err(byteor_pipeline_backings_shm::SequencedSlotsError::Stopped)
                            };

                            match resolved {
                                ResolvedStage::MapOk(f) => {
                                    let n = f(input, out);
                                    if n > input.len() {
                                        return set_err(ExecError::with_context(
                                            "resolved stage output too large",
                                            format!("stage={key} out={n} in={}", input.len()),
                                        ));
                                    }
                                    data[..n].copy_from_slice(&out[..n]);
                                    *len = n as u32;
                                    Ok(())
                                }
                                ResolvedStage::Map(f) => match f(input, out) {
                                    Ok(n) => {
                                        if n > input.len() {
                                            return set_err(ExecError::with_context(
                                                "resolved stage output too large",
                                                format!(
                                                    "stage={key} out={n} in={} (fallible)",
                                                    input.len()
                                                ),
                                            ));
                                        }
                                        data[..n].copy_from_slice(&out[..n]);
                                        *len = n as u32;
                                        Ok(())
                                    }
                                    Err(e) => set_err(ExecError::with_context(
                                        "resolved stage failed",
                                        format!("stage={key}: {e}"),
                                    )),
                                },
                                ResolvedStage::Filter(f) => match f(input) {
                                    Ok(true) => Ok(()),
                                    Ok(false) => {
                                        *len = 0;
                                        Ok(())
                                    }
                                    Err(e) => set_err(ExecError::with_context(
                                        "resolved stage failed",
                                        format!("stage={key}: {e}"),
                                    )),
                                },
                                ResolvedStage::Inspect(f) => match f(input) {
                                    Ok(()) => Ok(()),
                                    Err(e) => set_err(ExecError::with_context(
                                        "resolved stage failed",
                                        format!("stage={key}: {e}"),
                                    )),
                                },
                                ResolvedStage::StatefulTransform(st) => match st.call(input, out)
                                {
                                    Ok(TransformOutcome::Drop) => {
                                        *len = 0;
                                        Ok(())
                                    }
                                    Ok(TransformOutcome::ForwardInput) => Ok(()),
                                    Ok(TransformOutcome::ForwardOutput(n)) => {
                                        if n > input.len() {
                                            return set_err(ExecError::with_context(
                                                "resolved stage output too large",
                                                format!(
                                                    "stage={key} out={n} in={} (stateful transform)",
                                                    input.len()
                                                ),
                                            ));
                                        }
                                        data[..n].copy_from_slice(&out[..n]);
                                        *len = n as u32;
                                        Ok(())
                                    }
                                    Ok(_) => set_err(ExecError::new(
                                        "unsupported transform outcome (non-exhaustive)",
                                    )),
                                    Err(e) => set_err(ExecError::with_context(
                                        "resolved stage failed",
                                        format!("stage={key}: {e}"),
                                    )),
                                },
                                ResolvedStage::Stateful(st) => match st.call(input, out) {
                                    Ok(n) => {
                                        if n > input.len() {
                                            return set_err(ExecError::with_context(
                                                "resolved stage output too large",
                                                format!(
                                                    "stage={key} out={n} in={} (stateful)",
                                                    input.len()
                                                ),
                                            ));
                                        }
                                        data[..n].copy_from_slice(&out[..n]);
                                        *len = n as u32;
                                        Ok(())
                                    }
                                    Err(e) => set_err(ExecError::with_context(
                                        "resolved stage failed",
                                        format!("stage={key}: {e}"),
                                    )),
                                },
                            }
                        });

                        match res {
                            Ok(_) => {}
                            Err(byteor_pipeline_backings_shm::SequencedSlotsError::Stopped) => {
                                return Err(stage_err
                                    .unwrap_or_else(|| ExecError::new("resolved stage stopped")));
                            }
                            Err(e) => return Err(ExecError::with_source("stage failed", e)),
                        }
                    }
                }
            }
        }
    }

    Ok(())
}

/// In-process SP runtime handle for SingleRing SHM execution.
pub struct SingleRingSpRuntime {
    stop: std::sync::Arc<std::sync::atomic::AtomicBool>,
    lane_path: std::path::PathBuf,
    active_gating: usize,
    handles: Vec<(usize, std::thread::JoinHandle<Result<(), ExecError>>)>,
}

impl SingleRingSpRuntime {
    /// Number of stage threads in this runtime.
    pub fn stage_count(&self) -> usize {
        self.handles.len()
    }

    /// Number of active gating cells used by this runtime.
    pub fn active_gating(&self) -> usize {
        self.active_gating
    }

    /// Snapshot cursor + gating sequences without stopping the runtime.
    pub fn snapshot(&self) -> Result<byteor_pipeline_backings_shm::SingleRingSnapshot, ExecError> {
        let lane = "ignored";
        let opts = byteor_pipeline_backings_shm::AttachOptions {
            blocking: false,
            prefault: false,
            path: Some(self.lane_path.clone()),
            queue: 0,
        };
        let shm =
            byteor_pipeline_backings_shm::ShmSequencedSlots::attach(&opts, lane).map_err(|e| {
                ExecError::with_context_source(
                    "attach shm failed",
                    format!("path={} lane={lane}", self.lane_path.display()),
                    e,
                )
            })?;
        Ok(shm.snapshot())
    }

    /// True if any stage thread has exited (successfully or with error).
    ///
    /// This is intended for control-plane supervision.
    pub fn any_stage_finished(&self) -> bool {
        self.handles.iter().any(|(_, h)| h.is_finished())
    }

    /// Request a cooperative stop.
    pub fn request_stop(&self) {
        self.stop.store(true, std::sync::atomic::Ordering::Relaxed);
    }

    /// Request stop and join stage threads, but return an error if they don't stop within `timeout`.
    ///
    /// Notes:
    /// - Rust threads cannot be force-killed safely; this is cooperative stop only.
    /// - On timeout, this returns an error that lists unfinished stages.
    pub fn stop_and_join_bounded(mut self, timeout: std::time::Duration) -> Result<(), ExecError> {
        self.request_stop();

        let start = std::time::Instant::now();
        let mut first_err: Option<ExecError> = None;

        // Join threads as they finish so we can report failures promptly.
        while !self.handles.is_empty() {
            let mut i = 0;
            while i < self.handles.len() {
                if self.handles[i].1.is_finished() {
                    let (stage_idx, h) = self.handles.swap_remove(i);
                    let res = match h.join() {
                        Ok(r) => r,
                        Err(_) => Err(ExecError::new("stage panic")),
                    };

                    if let Err(e) = res {
                        if first_err.is_none() {
                            first_err = Some(ExecError::with_context(
                                "stage exited",
                                format!("stage_idx={stage_idx}: {e}"),
                            ));
                        }
                    }

                    continue;
                }
                i += 1;
            }

            if self.handles.is_empty() {
                break;
            }

            if start.elapsed() >= timeout {
                let mut unfinished: Vec<usize> = self.handles.iter().map(|(idx, _)| *idx).collect();
                unfinished.sort_unstable();
                return Err(ExecError::with_context(
                    "stop/join timed out",
                    format!("unfinished_stages={unfinished:?}"),
                ));
            }

            std::thread::sleep(std::time::Duration::from_millis(10));
        }

        match first_err {
            Some(e) => Err(e),
            None => Ok(()),
        }
    }

    /// Request stop and join all stage threads.
    pub fn stop_and_join(mut self) -> Result<(), ExecError> {
        self.request_stop();

        let mut first_err: Option<ExecError> = None;
        for (stage_idx, h) in self.handles.drain(..) {
            let res = match h.join() {
                Ok(r) => r,
                Err(_) => Err(ExecError::new("stage panic")),
            };

            if let Err(e) = res {
                if first_err.is_none() {
                    first_err = Some(ExecError::with_context(
                        "stage exited",
                        format!("stage_idx={stage_idx}: {e}"),
                    ));
                }
            }
        }

        match first_err {
            Some(e) => Err(e),
            None => Ok(()),
        }
    }
}

/// Spawn a long-running SingleRing runtime that processes until stopped.
pub fn spawn_single_ring_shm_runtime(
    spec: &PipelineSpecV1,
    lane_path: std::path::PathBuf,
) -> Result<SingleRingSpRuntime, ExecError> {
    spawn_single_ring_shm_runtime_with_wait(spec, lane_path, WaitPreset::StdBackoff)
}

/// Spawn a long-running SingleRing runtime that processes until stopped, using a wait preset.
pub fn spawn_single_ring_shm_runtime_with_wait(
    spec: &PipelineSpecV1,
    lane_path: std::path::PathBuf,
    wait: WaitPreset,
) -> Result<SingleRingSpRuntime, ExecError> {
    spawn_single_ring_shm_runtime_with_wait_and_resolver(spec, lane_path, wait, None)
}

/// Spawn a long-running SingleRing runtime that processes until stopped, resolving any
/// `ResolverKey` stages using `resolver` (cold path).
pub fn spawn_single_ring_shm_runtime_with_wait_and_resolver(
    spec: &PipelineSpecV1,
    lane_path: std::path::PathBuf,
    wait: WaitPreset,
    resolver: Option<&dyn StageResolver>,
) -> Result<SingleRingSpRuntime, ExecError> {
    spawn_single_ring_shm_runtime_with_wait_and_resolver_and_prefault(
        spec, lane_path, wait, resolver, false,
    )
}

/// Spawn a long-running SingleRing runtime that processes until stopped, resolving any
/// `ResolverKey` stages using `resolver` (cold path), with explicit control over SHM prefault.
pub fn spawn_single_ring_shm_runtime_with_wait_and_resolver_and_prefault(
    spec: &PipelineSpecV1,
    lane_path: std::path::PathBuf,
    wait: WaitPreset,
    resolver: Option<&dyn StageResolver>,
    prefault: bool,
) -> Result<SingleRingSpRuntime, ExecError> {
    spawn_single_ring_shm_runtime_with_wait_and_resolver_and_thread_init_and_prefault(
        spec, lane_path, wait, resolver, None, prefault,
    )
}

/// Spawn a long-running SingleRing runtime that processes until stopped, resolving any
/// `ResolverKey` stages using `resolver` (cold path) and invoking `thread_init` at the start of
/// each stage thread (before entering hot loops).
pub fn spawn_single_ring_shm_runtime_with_wait_and_resolver_and_thread_init(
    spec: &PipelineSpecV1,
    lane_path: std::path::PathBuf,
    wait: WaitPreset,
    resolver: Option<&dyn StageResolver>,
    thread_init: Option<ThreadInitHook>,
) -> Result<SingleRingSpRuntime, ExecError> {
    spawn_single_ring_shm_runtime_with_wait_and_resolver_and_thread_init_and_prefault(
        spec,
        lane_path,
        wait,
        resolver,
        thread_init,
        false,
    )
}

/// Spawn a long-running SingleRing runtime that processes until stopped, resolving any
/// `ResolverKey` stages using `resolver` (cold path) and invoking `thread_init` at the start of
/// each stage thread (before entering hot loops), with explicit control over SHM prefault.
pub fn spawn_single_ring_shm_runtime_with_wait_and_resolver_and_thread_init_and_prefault(
    spec: &PipelineSpecV1,
    lane_path: std::path::PathBuf,
    wait: WaitPreset,
    resolver: Option<&dyn StageResolver>,
    thread_init: Option<ThreadInitHook>,
    prefault: bool,
) -> Result<SingleRingSpRuntime, ExecError> {
    validate_v1(spec).map_err(|e| ExecError::with_source("spec validation failed", e))?;
    let ring = match spec {
        PipelineSpecV1::SingleRing(ring) => ring,
        _ => return Err(ExecError::new("expected SingleRing spec")),
    };

    if ring.shards > 1 {
        return Err(ExecError::new(
            "sharded SingleRing requires the sharded executor helper",
        ));
    }

    let lane = "ignored";
    let stages = ring.stages.len();
    let active_gating = stages;
    if active_gating == 0 || active_gating > 4 {
        return Err(ExecError::new("single_ring gating exceeds Layout4"));
    }

    let opts = byteor_pipeline_backings_shm::AttachOptions {
        blocking: false,
        prefault,
        path: Some(lane_path.clone()),
        queue: 0,
    };
    let shm =
        byteor_pipeline_backings_shm::ShmSequencedSlots::attach(&opts, lane).map_err(|e| {
            ExecError::with_context_source(
                "attach shm failed",
                format!("path={} lane={lane}", lane_path.display()),
                e,
            )
        })?;

    let stop = std::sync::Arc::new(std::sync::atomic::AtomicBool::new(false));
    let iterations = usize::MAX;
    let auto_pin_cpus = if thread_init.is_none() {
        current_allowed_cpus()
    } else {
        Vec::new()
    };

    let mut handles = Vec::new();
    for (stage_idx, st) in ring.stages.iter().cloned().enumerate() {
        let opts_t = opts.clone();
        let stop_t = stop.clone();
        let cursor_t = byteor_pipeline_backings_shm::CursorBarrier::new(&shm);
        let wait_t = wait;
        let thread_init_t = thread_init.clone();
        let cpu_t = runtime_stage_cpu(&auto_pin_cpus, stage_idx);

        let barrier = if st.depends_on.is_empty() {
            StageBarrier::Cursor(cursor_t)
        } else if st.depends_on.len() == 1 {
            let dep = st.depends_on[0] as usize;
            StageBarrier::Gating(byteor_pipeline_backings_shm::GatingBarrier::new(&shm, dep))
        } else {
            let mut deps = Vec::new();
            for d in &st.depends_on {
                deps.push(byteor_pipeline_backings_shm::GatingBarrier::new(
                    &shm,
                    *d as usize,
                ));
            }
            StageBarrier::Join(JoinBarrier { deps })
        };

        let mut op = resolve_stage_op(st.op, resolver)?;

        let release = stage_idx + 1 == stages;

        let role_name = format!("byteor-single-ring:stage-{stage_idx}");
        let handle = std::thread::Builder::new()
            .name(role_name.clone())
            .spawn(move || {
                let result = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| {
                    let _affinity_guard = ThreadAffinityGuard::pin(cpu_t)?;

                    if let Some(h) = &thread_init_t {
                        let ctx = ThreadInitContext::new(role_name.clone(), RoleKind::Stage);
                        h(&ctx)?;
                    }

                    let mut stage = byteor_pipeline_backings_shm::ShmSequencedSlotsStage::attach(
                        &opts_t, lane, stage_idx,
                    )
                    .map_err(|_| {
                        ExecError::with_context(
                            "attach stage failed",
                            format!("stage_idx={stage_idx}"),
                        )
                    })?;

                    match wait_t {
                        WaitPreset::MaxPerfSpin => {
                            let mut wait = indexbus_core::SpinWait::default();
                            run_stage_loop_stoppable(
                                &mut stage,
                                &barrier,
                                &mut wait,
                                stop_t.as_ref(),
                                &mut op,
                                iterations,
                                release,
                            )?;
                        }
                        WaitPreset::StdBackoff => {
                            let mut wait = indexbus_core::StdBackoff::default();
                            run_stage_loop_stoppable(
                                &mut stage,
                                &barrier,
                                &mut wait,
                                stop_t.as_ref(),
                                &mut op,
                                iterations,
                                release,
                            )?;
                        }
                        WaitPreset::Yield => {
                            let mut wait = indexbus_core::YieldWait;
                            run_stage_loop_stoppable(
                                &mut stage,
                                &barrier,
                                &mut wait,
                                stop_t.as_ref(),
                                &mut op,
                                iterations,
                                release,
                            )?;
                        }
                    }

                    Ok::<(), ExecError>(())
                }));

                match result {
                    Ok(Ok(())) => Ok(()),
                    Ok(Err(e)) => {
                        stop_t.store(true, std::sync::atomic::Ordering::Relaxed);
                        Err(ExecError::with_context(
                            "stage failed",
                            format!("stage_idx={stage_idx}: {e}"),
                        ))
                    }
                    Err(_) => {
                        stop_t.store(true, std::sync::atomic::Ordering::Relaxed);
                        Err(ExecError::with_context(
                            "stage panic",
                            format!("stage_idx={stage_idx}"),
                        ))
                    }
                }
            })
            .map_err(|e| ExecError::with_source("failed to spawn stage thread", e))?;

        handles.push((stage_idx, handle));
    }

    Ok(SingleRingSpRuntime {
        stop,
        lane_path,
        active_gating,
        handles,
    })
}

/// Run a SingleRing spec over a SHM SequencedSlots backing.
///
/// This is a minimal v1 in-process executor intended for bring-up tests.
pub fn run_single_ring_shm(
    spec: &PipelineSpecV1,
    lane_path: std::path::PathBuf,
    input: &[u8],
) -> Result<Vec<u8>, ExecError> {
    run_single_ring_shm_with_wait(spec, lane_path, input, WaitPreset::MaxPerfSpin)
}

/// Run a SingleRing spec over a SHM SequencedSlots backing, using a wait preset.
pub fn run_single_ring_shm_with_wait(
    spec: &PipelineSpecV1,
    lane_path: std::path::PathBuf,
    input: &[u8],
    wait: WaitPreset,
) -> Result<Vec<u8>, ExecError> {
    run_single_ring_shm_with_wait_and_resolver(spec, lane_path, input, wait, None)
}

/// Run a SingleRing spec over a SHM SequencedSlots backing, resolving any `ResolverKey` stages
/// using `resolver` (cold path).
pub fn run_single_ring_shm_with_wait_and_resolver(
    spec: &PipelineSpecV1,
    lane_path: std::path::PathBuf,
    input: &[u8],
    wait: WaitPreset,
    resolver: Option<&dyn StageResolver>,
) -> Result<Vec<u8>, ExecError> {
    run_single_ring_shm_with_wait_and_resolver_and_prefault(
        spec, lane_path, input, wait, resolver, false,
    )
}

/// Run a SingleRing spec over a SHM SequencedSlots backing, resolving any `ResolverKey` stages
/// using `resolver` (cold path), with explicit control over SHM prefault.
pub fn run_single_ring_shm_with_wait_and_resolver_and_prefault(
    spec: &PipelineSpecV1,
    lane_path: std::path::PathBuf,
    input: &[u8],
    wait: WaitPreset,
    resolver: Option<&dyn StageResolver>,
    prefault: bool,
) -> Result<Vec<u8>, ExecError> {
    run_single_ring_shm_with_wait_and_resolver_and_thread_init_and_prefault(
        spec, lane_path, input, wait, resolver, None, prefault,
    )
}

/// Run a SingleRing spec over a SHM SequencedSlots backing, resolving any `ResolverKey` stages
/// using `resolver` (cold path) and invoking `thread_init` at the start of each stage thread.
///
/// This is a minimal v1 in-process executor intended for bring-up tests.
pub fn run_single_ring_shm_with_wait_and_resolver_and_thread_init(
    spec: &PipelineSpecV1,
    lane_path: std::path::PathBuf,
    input: &[u8],
    wait: WaitPreset,
    resolver: Option<&dyn StageResolver>,
    thread_init: Option<ThreadInitHook>,
) -> Result<Vec<u8>, ExecError> {
    run_single_ring_shm_with_wait_and_resolver_and_thread_init_and_prefault(
        spec,
        lane_path,
        input,
        wait,
        resolver,
        thread_init,
        false,
    )
}

/// Run a SingleRing spec over a SHM SequencedSlots backing, resolving any `ResolverKey` stages
/// using `resolver` (cold path) and invoking `thread_init` at the start of each stage thread,
/// with explicit control over SHM prefault.
pub fn run_single_ring_shm_with_wait_and_resolver_and_thread_init_and_prefault(
    spec: &PipelineSpecV1,
    lane_path: std::path::PathBuf,
    input: &[u8],
    wait: WaitPreset,
    resolver: Option<&dyn StageResolver>,
    thread_init: Option<ThreadInitHook>,
    prefault: bool,
) -> Result<Vec<u8>, ExecError> {
    validate_v1(spec).map_err(|e| ExecError::with_source("spec validation failed", e))?;
    let ring = match spec {
        PipelineSpecV1::SingleRing(ring) => ring,
        _ => return Err(ExecError::new("spec is not SingleRing")),
    };

    if ring.shards > 1 {
        return Err(ExecError::new(
            "sharded SingleRing requires the sharded executor helper",
        ));
    }

    let opts = byteor_pipeline_backings_shm::AttachOptions {
        blocking: false,
        prefault,
        path: Some(lane_path),
        queue: 0,
    };

    let lane = "ignored";
    let stages = ring.stages.len();
    let active_gating = stages;
    if active_gating == 0 || active_gating > 4 {
        return Err(ExecError::new("single_ring gating exceeds Layout4"));
    }

    let mut prod = byteor_pipeline_backings_shm::ShmSequencedSlotsProducer::attach(
        &opts,
        lane,
        1,
        active_gating,
    )
    .map_err(|e| {
        ExecError::with_context_source(
            "attach producer failed",
            format!("path={} lane={lane}", opts.path.as_ref().unwrap().display()),
            e,
        )
    })?;

    let shm =
        byteor_pipeline_backings_shm::ShmSequencedSlots::attach(&opts, lane).map_err(|e| {
            ExecError::with_context_source(
                "attach shm failed",
                format!("path={} lane={lane}", opts.path.as_ref().unwrap().display()),
                e,
            )
        })?;
    let cursor = byteor_pipeline_backings_shm::CursorBarrier::new(&shm);

    let mut handles = Vec::new();
    let iterations = input.len();
    let final_stage_idx = stages.saturating_sub(1);
    let auto_pin_cpus = if thread_init.is_none() {
        current_allowed_cpus()
    } else {
        Vec::new()
    };
    for (stage_idx, st) in ring
        .stages
        .iter()
        .cloned()
        .enumerate()
        .take(final_stage_idx)
    {
        let opts_t = opts.clone();
        let cursor_t = byteor_pipeline_backings_shm::CursorBarrier::new(&shm);
        let wait_t = wait;
        let thread_init_t = thread_init.clone();
        let cpu_t = worker_thread_cpu(&auto_pin_cpus, stage_idx);

        let barrier = if st.depends_on.is_empty() {
            StageBarrier::Cursor(cursor_t)
        } else if st.depends_on.len() == 1 {
            let dep = st.depends_on[0] as usize;
            StageBarrier::Gating(byteor_pipeline_backings_shm::GatingBarrier::new(&shm, dep))
        } else {
            let mut deps = Vec::new();
            for d in &st.depends_on {
                deps.push(byteor_pipeline_backings_shm::GatingBarrier::new(
                    &shm,
                    *d as usize,
                ));
            }
            StageBarrier::Join(JoinBarrier { deps })
        };

        let mut op = resolve_stage_op(st.op, resolver)?;

        let role_name = format!("byteor-single-ring:stage-{stage_idx}");
        let handle = std::thread::Builder::new()
            .name(role_name.clone())
            .spawn(move || {
                let _affinity_guard = ThreadAffinityGuard::pin(cpu_t)?;

                if let Some(h) = &thread_init_t {
                    let ctx = ThreadInitContext::new(role_name.clone(), RoleKind::Stage);
                    h(&ctx)?;
                }

                let mut stage = byteor_pipeline_backings_shm::ShmSequencedSlotsStage::attach(
                    &opts_t, lane, stage_idx,
                )
                .map_err(|e| {
                    ExecError::with_context_source(
                        "attach stage failed",
                        format!(
                            "path={} lane={lane} stage_idx={stage_idx}",
                            opts_t.path.as_ref().unwrap().display()
                        ),
                        e,
                    )
                })?;

                match wait_t {
                    WaitPreset::MaxPerfSpin => {
                        let mut wait = indexbus_core::SpinWait::default();
                        run_stage_loop_iters(
                            &mut stage, &barrier, &mut wait, &mut op, iterations, false,
                        )?;
                    }
                    WaitPreset::StdBackoff => {
                        let mut wait = indexbus_core::StdBackoff::default();
                        run_stage_loop_iters(
                            &mut stage, &barrier, &mut wait, &mut op, iterations, false,
                        )?;
                    }
                    WaitPreset::Yield => {
                        let mut wait = indexbus_core::YieldWait;
                        run_stage_loop_iters(
                            &mut stage, &barrier, &mut wait, &mut op, iterations, false,
                        )?;
                    }
                }

                Ok::<(), ExecError>(())
            })
            .map_err(|e| ExecError::with_source("failed to spawn stage thread", e))?;
        handles.push(handle);
    }

    let _main_affinity_guard = ThreadAffinityGuard::pin(main_thread_cpu(&auto_pin_cpus))?;

    let mut publish_wait = match wait {
        WaitPreset::MaxPerfSpin => WaitHolder::Spin(indexbus_core::SpinWait::default()),
        WaitPreset::StdBackoff => WaitHolder::Backoff(indexbus_core::StdBackoff::default()),
        WaitPreset::Yield => WaitHolder::Yield(indexbus_core::YieldWait),
    };

    // Final stage acts as the consumer: it releases slots and we collect output directly.
    let final_stage = ring
        .stages
        .get(final_stage_idx)
        .cloned()
        .ok_or_else(|| ExecError::new("single_ring requires at least one stage"))?;

    let consumer_barrier = if final_stage.depends_on.is_empty() {
        StageBarrier::Cursor(cursor)
    } else if final_stage.depends_on.len() == 1 {
        let dep = final_stage.depends_on[0] as usize;
        StageBarrier::Gating(byteor_pipeline_backings_shm::GatingBarrier::new(&shm, dep))
    } else {
        let mut deps = Vec::new();
        for d in &final_stage.depends_on {
            deps.push(byteor_pipeline_backings_shm::GatingBarrier::new(
                &shm,
                *d as usize,
            ));
        }
        StageBarrier::Join(JoinBarrier { deps })
    };

    let mut consumer_stage =
        byteor_pipeline_backings_shm::ShmSequencedSlotsStage::attach(&opts, lane, final_stage_idx)
            .map_err(|e| {
                ExecError::with_context_source(
                    "attach stage failed",
                    format!(
                        "path={} lane={lane} stage_idx={final_stage_idx}",
                        opts.path.as_ref().unwrap().display()
                    ),
                    e,
                )
            })?;
    let mut consumer_op = resolve_stage_op(final_stage.op, resolver)?;
    let mut scratch = vec![0u8; indexbus_core::INDEXBUS_SLOT_DATA_SIZE];

    let mut out = Vec::with_capacity(input.len());
    let mut wait = match wait {
        WaitPreset::MaxPerfSpin => WaitHolder::Spin(indexbus_core::SpinWait::default()),
        WaitPreset::StdBackoff => WaitHolder::Backoff(indexbus_core::StdBackoff::default()),
        WaitPreset::Yield => WaitHolder::Yield(indexbus_core::YieldWait),
    };

    for &b in input {
        loop {
            match prod.publish_bytes(&[b]) {
                Ok(_) => break,
                Err(byteor_pipeline_backings_shm::SequencedSlotsError::Full) => {
                    publish_wait.wait();
                }
                Err(e) => {
                    return Err(ExecError::with_context("publish failed", e.to_string()));
                }
            }
        }

        match &mut wait {
            WaitHolder::Spin(w) => run_consumer_step_collect_u8(
                &mut consumer_stage,
                &consumer_barrier,
                w,
                &mut consumer_op,
                &mut scratch,
                &mut out,
            )?,
            WaitHolder::Backoff(w) => run_consumer_step_collect_u8(
                &mut consumer_stage,
                &consumer_barrier,
                w,
                &mut consumer_op,
                &mut scratch,
                &mut out,
            )?,
            WaitHolder::Yield(w) => run_consumer_step_collect_u8(
                &mut consumer_stage,
                &consumer_barrier,
                w,
                &mut consumer_op,
                &mut scratch,
                &mut out,
            )?,
        }
    }

    for h in handles {
        h.join().map_err(|_| ExecError::new("stage panic"))??;
    }

    Ok(out)
}

/// Run a SingleRing spec over a SHM SequencedSlots backing treating `input` as a single record.
///
/// This is intended for message-shaped stages (e.g. IndexBus envelopes) where a stage expects a
/// whole message in one slot.
pub fn run_single_ring_shm_record_with_wait_and_resolver_and_thread_init(
    spec: &PipelineSpecV1,
    lane_path: std::path::PathBuf,
    input: &[u8],
    wait: WaitPreset,
    resolver: Option<&dyn StageResolver>,
    thread_init: Option<ThreadInitHook>,
) -> Result<Vec<u8>, ExecError> {
    validate_v1(spec).map_err(|e| ExecError::with_source("spec validation failed", e))?;
    let ring = match spec {
        PipelineSpecV1::SingleRing(ring) => ring,
        _ => return Err(ExecError::new("spec is not SingleRing")),
    };

    if ring.shards > 1 {
        return Err(ExecError::new(
            "sharded SingleRing requires the sharded executor helper",
        ));
    }

    if input.len() > indexbus_core::INDEXBUS_SLOT_DATA_SIZE {
        return Err(ExecError::with_context(
            "input too large for slot",
            format!(
                "in={} max={}",
                input.len(),
                indexbus_core::INDEXBUS_SLOT_DATA_SIZE
            ),
        ));
    }

    let opts = byteor_pipeline_backings_shm::AttachOptions {
        blocking: false,
        prefault: false,
        path: Some(lane_path),
        queue: 0,
    };

    let lane = "ignored";
    let stages = ring.stages.len();
    let active_gating = stages;
    if active_gating == 0 || active_gating > 4 {
        return Err(ExecError::new("single_ring gating exceeds Layout4"));
    }

    let mut prod = byteor_pipeline_backings_shm::ShmSequencedSlotsProducer::attach(
        &opts,
        lane,
        1,
        active_gating,
    )
    .map_err(|e| {
        ExecError::with_context_source(
            "attach producer failed",
            format!("path={} lane={lane}", opts.path.as_ref().unwrap().display()),
            e,
        )
    })?;

    let shm =
        byteor_pipeline_backings_shm::ShmSequencedSlots::attach(&opts, lane).map_err(|e| {
            ExecError::with_context_source(
                "attach shm failed",
                format!("path={} lane={lane}", opts.path.as_ref().unwrap().display()),
                e,
            )
        })?;
    let cursor = byteor_pipeline_backings_shm::CursorBarrier::new(&shm);

    let mut handles = Vec::new();
    let iterations = 1;
    let final_stage_idx = stages.saturating_sub(1);
    let auto_pin_cpus = if thread_init.is_none() {
        current_allowed_cpus()
    } else {
        Vec::new()
    };
    for (stage_idx, st) in ring
        .stages
        .iter()
        .cloned()
        .enumerate()
        .take(final_stage_idx)
    {
        let opts_t = opts.clone();
        let cursor_t = byteor_pipeline_backings_shm::CursorBarrier::new(&shm);
        let wait_t = wait;
        let thread_init_t = thread_init.clone();
        let cpu_t = worker_thread_cpu(&auto_pin_cpus, stage_idx);

        let barrier = if st.depends_on.is_empty() {
            StageBarrier::Cursor(cursor_t)
        } else if st.depends_on.len() == 1 {
            let dep = st.depends_on[0] as usize;
            StageBarrier::Gating(byteor_pipeline_backings_shm::GatingBarrier::new(&shm, dep))
        } else {
            let mut deps = Vec::new();
            for d in &st.depends_on {
                deps.push(byteor_pipeline_backings_shm::GatingBarrier::new(
                    &shm,
                    *d as usize,
                ));
            }
            StageBarrier::Join(JoinBarrier { deps })
        };

        let mut op = resolve_stage_op(st.op, resolver)?;

        let role_name = format!("byteor-single-ring:stage-{stage_idx}");
        let handle = std::thread::Builder::new()
            .name(role_name.clone())
            .spawn(move || {
                let _affinity_guard = ThreadAffinityGuard::pin(cpu_t)?;

                if let Some(h) = &thread_init_t {
                    let ctx = ThreadInitContext::new(role_name.clone(), RoleKind::Stage);
                    h(&ctx)?;
                }

                let mut stage = byteor_pipeline_backings_shm::ShmSequencedSlotsStage::attach(
                    &opts_t, lane, stage_idx,
                )
                .map_err(|e| {
                    ExecError::with_context_source(
                        "attach stage failed",
                        format!(
                            "path={} lane={lane} stage_idx={stage_idx}",
                            opts_t.path.as_ref().unwrap().display()
                        ),
                        e,
                    )
                })?;

                match wait_t {
                    WaitPreset::MaxPerfSpin => {
                        let mut wait = indexbus_core::SpinWait::default();
                        run_stage_loop_iters(
                            &mut stage, &barrier, &mut wait, &mut op, iterations, false,
                        )?;
                    }
                    WaitPreset::StdBackoff => {
                        let mut wait = indexbus_core::StdBackoff::default();
                        run_stage_loop_iters(
                            &mut stage, &barrier, &mut wait, &mut op, iterations, false,
                        )?;
                    }
                    WaitPreset::Yield => {
                        let mut wait = indexbus_core::YieldWait;
                        run_stage_loop_iters(
                            &mut stage, &barrier, &mut wait, &mut op, iterations, false,
                        )?;
                    }
                }

                Ok::<(), ExecError>(())
            })
            .map_err(|e| ExecError::with_source("failed to spawn stage thread", e))?;
        handles.push(handle);
    }

    let _main_affinity_guard = ThreadAffinityGuard::pin(main_thread_cpu(&auto_pin_cpus))?;

    let mut publish_wait = match wait {
        WaitPreset::MaxPerfSpin => WaitHolder::Spin(indexbus_core::SpinWait::default()),
        WaitPreset::StdBackoff => WaitHolder::Backoff(indexbus_core::StdBackoff::default()),
        WaitPreset::Yield => WaitHolder::Yield(indexbus_core::YieldWait),
    };

    let final_stage = ring
        .stages
        .get(final_stage_idx)
        .cloned()
        .ok_or_else(|| ExecError::new("single_ring requires at least one stage"))?;

    let consumer_barrier = if final_stage.depends_on.is_empty() {
        StageBarrier::Cursor(cursor)
    } else if final_stage.depends_on.len() == 1 {
        let dep = final_stage.depends_on[0] as usize;
        StageBarrier::Gating(byteor_pipeline_backings_shm::GatingBarrier::new(&shm, dep))
    } else {
        let mut deps = Vec::new();
        for d in &final_stage.depends_on {
            deps.push(byteor_pipeline_backings_shm::GatingBarrier::new(
                &shm,
                *d as usize,
            ));
        }
        StageBarrier::Join(JoinBarrier { deps })
    };

    let mut consumer_stage =
        byteor_pipeline_backings_shm::ShmSequencedSlotsStage::attach(&opts, lane, final_stage_idx)
            .map_err(|e| {
                ExecError::with_context_source(
                    "attach stage failed",
                    format!(
                        "path={} lane={lane} stage_idx={final_stage_idx}",
                        opts.path.as_ref().unwrap().display()
                    ),
                    e,
                )
            })?;
    let mut consumer_op = resolve_stage_op(final_stage.op, resolver)?;
    let mut scratch = vec![0u8; indexbus_core::INDEXBUS_SLOT_DATA_SIZE];

    let mut out = Vec::with_capacity(input.len());
    let mut wait = match wait {
        WaitPreset::MaxPerfSpin => WaitHolder::Spin(indexbus_core::SpinWait::default()),
        WaitPreset::StdBackoff => WaitHolder::Backoff(indexbus_core::StdBackoff::default()),
        WaitPreset::Yield => WaitHolder::Yield(indexbus_core::YieldWait),
    };

    loop {
        match prod.publish_bytes(input) {
            Ok(_) => break,
            Err(byteor_pipeline_backings_shm::SequencedSlotsError::Full) => publish_wait.wait(),
            Err(e) => return Err(ExecError::with_context("publish failed", e.to_string())),
        }
    }

    match &mut wait {
        WaitHolder::Spin(w) => run_consumer_step_collect_bytes(
            &mut consumer_stage,
            &consumer_barrier,
            w,
            &mut consumer_op,
            &mut scratch,
            &mut out,
        )?,
        WaitHolder::Backoff(w) => run_consumer_step_collect_bytes(
            &mut consumer_stage,
            &consumer_barrier,
            w,
            &mut consumer_op,
            &mut scratch,
            &mut out,
        )?,
        WaitHolder::Yield(w) => run_consumer_step_collect_bytes(
            &mut consumer_stage,
            &consumer_barrier,
            w,
            &mut consumer_op,
            &mut scratch,
            &mut out,
        )?,
    }

    for h in handles {
        h.join().map_err(|_| ExecError::new("stage panic"))??;
    }

    Ok(out)
}

fn run_consumer_step_collect_u8<W: byteor_pipeline_kernel::WaitStrategy>(
    stage: &mut byteor_pipeline_backings_shm::ShmSequencedSlotsStage,
    barrier: &StageBarrier,
    wait: &mut W,
    op: &mut StageExecOp,
    scratch: &mut [u8],
    out: &mut Vec<u8>,
) -> Result<(), ExecError> {
    match op {
        StageExecOp::BuiltIn(op) => match op {
            StageOpV1::Identity => {
                stage
                    .inspect_next(barrier, wait, true, |data, len| {
                        out.push(if len > 0 { data[0] } else { 0 });
                        Ok(())
                    })
                    .map_err(|e| ExecError::with_source("stage failed", e))?;
            }
            StageOpV1::AddU8 { delta } => {
                let delta = *delta;
                stage
                    .process_next(barrier, wait, true, |data, len| {
                        let n = clamp_len(*len, data.len());
                        op_add_u8(&mut data[..n], delta);
                        out.push(if *len > 0 { data[0] } else { 0 });
                        Ok(())
                    })
                    .map_err(|e| ExecError::with_source("stage failed", e))?;
            }
            StageOpV1::ResolverKey { stage } => {
                return Err(ExecError::with_context(
                    "unresolved stage",
                    format!("stage={stage}"),
                ));
            }
        },
        StageExecOp::Resolved {
            key,
            stage: resolved,
        } => {
            let mut stage_err: Option<ExecError> = None;

            match resolved {
                ResolvedStage::Inspect(f) => {
                    let res = stage.inspect_next(barrier, wait, true, |data, len| {
                        let n = clamp_len(len, data.len());
                        if let Err(e) = f(&data[..n]) {
                            stage_err = Some(ExecError::with_context(
                                "resolved stage failed",
                                format!("stage={key}: {e}"),
                            ));
                            return Err(byteor_pipeline_backings_shm::SequencedSlotsError::Stopped);
                        }
                        out.push(if len > 0 { data[0] } else { 0 });
                        Ok(())
                    });

                    match res {
                        Ok(_) => {}
                        Err(byteor_pipeline_backings_shm::SequencedSlotsError::Stopped) => {
                            return Err(stage_err.unwrap_or_else(|| {
                                ExecError::with_context(
                                    "resolved stage stopped",
                                    format!("stage={key}"),
                                )
                            }));
                        }
                        Err(e) => return Err(ExecError::with_source("stage failed", e)),
                    }
                }
                _ => {
                    let res = stage.process_next(barrier, wait, true, |data, len| {
                        let n_in = clamp_len(*len, data.len());
                        let input = &data[..n_in];
                        let out_buf = &mut scratch[..n_in];

                        let mut set_err = |e: ExecError| {
                            stage_err = Some(e);
                            Err(byteor_pipeline_backings_shm::SequencedSlotsError::Stopped)
                        };

                        match resolved {
                            ResolvedStage::MapOk(f) => {
                                let n = f(input, out_buf);
                                if n > input.len() {
                                    return set_err(ExecError::with_context(
                                        "resolved stage output too large",
                                        format!("stage={key} out={n} in={}", input.len()),
                                    ));
                                }
                                data[..n].copy_from_slice(&out_buf[..n]);
                                *len = n as u32;
                            }
                            ResolvedStage::Map(f) => match f(input, out_buf) {
                                Ok(n) => {
                                    if n > input.len() {
                                        return set_err(ExecError::with_context(
                                            "resolved stage output too large",
                                            format!(
                                                "stage={key} out={n} in={} (fallible)",
                                                input.len()
                                            ),
                                        ));
                                    }
                                    data[..n].copy_from_slice(&out_buf[..n]);
                                    *len = n as u32;
                                }
                                Err(e) => {
                                    return set_err(ExecError::with_context(
                                        "resolved stage failed",
                                        format!("stage={key}: {e}"),
                                    ));
                                }
                            },
                            ResolvedStage::Filter(f) => match f(input) {
                                Ok(true) => {}
                                Ok(false) => {
                                    *len = 0;
                                }
                                Err(e) => {
                                    return set_err(ExecError::with_context(
                                        "resolved stage failed",
                                        format!("stage={key}: {e}"),
                                    ));
                                }
                            },
                            ResolvedStage::Inspect(f) => match f(input) {
                                Ok(()) => {}
                                Err(e) => {
                                    return set_err(ExecError::with_context(
                                        "resolved stage failed",
                                        format!("stage={key}: {e}"),
                                    ));
                                }
                            },
                            ResolvedStage::StatefulTransform(st) => match st.call(input, out_buf) {
                                Ok(TransformOutcome::Drop) => {
                                    *len = 0;
                                }
                                Ok(TransformOutcome::ForwardInput) => {}
                                Ok(TransformOutcome::ForwardOutput(n)) => {
                                    if n > input.len() {
                                        return set_err(ExecError::with_context(
                                            "resolved stage output too large",
                                            format!(
                                                "stage={key} out={n} in={} (stateful transform)",
                                                input.len()
                                            ),
                                        ));
                                    }
                                    data[..n].copy_from_slice(&out_buf[..n]);
                                    *len = n as u32;
                                }
                                Ok(_) => {
                                    return set_err(ExecError::new(
                                        "unsupported transform outcome (non-exhaustive)",
                                    ));
                                }
                                Err(e) => {
                                    return set_err(ExecError::with_context(
                                        "resolved stage failed",
                                        format!("stage={key}: {e}"),
                                    ));
                                }
                            },
                            ResolvedStage::Stateful(st) => match st.call(input, out_buf) {
                                Ok(n) => {
                                    if n > input.len() {
                                        return set_err(ExecError::with_context(
                                            "resolved stage output too large",
                                            format!(
                                                "stage={key} out={n} in={} (stateful)",
                                                input.len()
                                            ),
                                        ));
                                    }
                                    data[..n].copy_from_slice(&out_buf[..n]);
                                    *len = n as u32;
                                }
                                Err(e) => {
                                    return set_err(ExecError::with_context(
                                        "resolved stage failed",
                                        format!("stage={key}: {e}"),
                                    ));
                                }
                            },
                        }

                        out.push(if *len > 0 { data[0] } else { 0 });
                        Ok(())
                    });

                    match res {
                        Ok(_) => {}
                        Err(byteor_pipeline_backings_shm::SequencedSlotsError::Stopped) => {
                            return Err(stage_err.unwrap_or_else(|| {
                                ExecError::with_context(
                                    "resolved stage stopped",
                                    format!("stage={key}"),
                                )
                            }));
                        }
                        Err(e) => return Err(ExecError::with_source("stage failed", e)),
                    }
                }
            }
        }
    }

    Ok(())
}

fn run_consumer_step_collect_bytes<W: byteor_pipeline_kernel::WaitStrategy>(
    stage: &mut byteor_pipeline_backings_shm::ShmSequencedSlotsStage,
    barrier: &StageBarrier,
    wait: &mut W,
    op: &mut StageExecOp,
    scratch: &mut [u8],
    out: &mut Vec<u8>,
) -> Result<(), ExecError> {
    out.clear();

    match op {
        StageExecOp::BuiltIn(op) => match op {
            StageOpV1::Identity => {
                stage
                    .inspect_next(barrier, wait, true, |data, len| {
                        let n = clamp_len(len, data.len());
                        out.extend_from_slice(&data[..n]);
                        Ok(())
                    })
                    .map_err(|e| ExecError::with_source("stage failed", e))?;
            }
            StageOpV1::AddU8 { delta } => {
                let delta = *delta;
                stage
                    .process_next(barrier, wait, true, |data, len| {
                        let n = clamp_len(*len, data.len());
                        op_add_u8(&mut data[..n], delta);
                        out.extend_from_slice(&data[..n]);
                        Ok(())
                    })
                    .map_err(|e| ExecError::with_source("stage failed", e))?;
            }
            StageOpV1::ResolverKey { stage } => {
                return Err(ExecError::with_context(
                    "unresolved stage",
                    format!("stage={stage}"),
                ));
            }
        },
        StageExecOp::Resolved {
            key,
            stage: resolved,
        } => {
            let mut stage_err: Option<ExecError> = None;

            match resolved {
                ResolvedStage::Inspect(f) => {
                    let res = stage.inspect_next(barrier, wait, true, |data, len| {
                        let n = clamp_len(len, data.len());
                        if let Err(e) = f(&data[..n]) {
                            stage_err = Some(ExecError::with_context(
                                "resolved stage failed",
                                format!("stage={key}: {e}"),
                            ));
                            return Err(byteor_pipeline_backings_shm::SequencedSlotsError::Stopped);
                        }
                        out.extend_from_slice(&data[..n]);
                        Ok(())
                    });

                    match res {
                        Ok(_) => {}
                        Err(byteor_pipeline_backings_shm::SequencedSlotsError::Stopped) => {
                            return Err(stage_err
                                .unwrap_or_else(|| ExecError::new("resolved stage stopped")));
                        }
                        Err(e) => return Err(ExecError::with_source("stage failed", e)),
                    }
                }
                _ => {
                    let res = stage.process_next(barrier, wait, true, |data, len| {
                        let n_in = clamp_len(*len, data.len());
                        let input = &data[..n_in];
                        let out_buf = &mut scratch[..n_in];

                        let mut set_err = |e: ExecError| {
                            stage_err = Some(e);
                            Err(byteor_pipeline_backings_shm::SequencedSlotsError::Stopped)
                        };

                        match resolved {
                            ResolvedStage::MapOk(f) => {
                                let n = f(input, out_buf);
                                if n > input.len() {
                                    return set_err(ExecError::with_context(
                                        "resolved stage output too large",
                                        format!("stage={key} out={n} in={}", input.len()),
                                    ));
                                }
                                data[..n].copy_from_slice(&out_buf[..n]);
                                *len = n as u32;
                            }
                            ResolvedStage::Map(f) => match f(input, out_buf) {
                                Ok(n) => {
                                    if n > input.len() {
                                        return set_err(ExecError::with_context(
                                            "resolved stage output too large",
                                            format!(
                                                "stage={key} out={n} in={} (fallible)",
                                                input.len()
                                            ),
                                        ));
                                    }
                                    data[..n].copy_from_slice(&out_buf[..n]);
                                    *len = n as u32;
                                }
                                Err(e) => {
                                    return set_err(ExecError::with_context(
                                        "resolved stage failed",
                                        format!("stage={key}: {e}"),
                                    ));
                                }
                            },
                            ResolvedStage::Filter(f) => match f(input) {
                                Ok(true) => {}
                                Ok(false) => {
                                    *len = 0;
                                }
                                Err(e) => {
                                    return set_err(ExecError::with_context(
                                        "resolved stage failed",
                                        format!("stage={key}: {e}"),
                                    ));
                                }
                            },
                            ResolvedStage::Inspect(f) => match f(input) {
                                Ok(()) => {}
                                Err(e) => {
                                    return set_err(ExecError::with_context(
                                        "resolved stage failed",
                                        format!("stage={key}: {e}"),
                                    ));
                                }
                            },
                            ResolvedStage::StatefulTransform(st) => match st.call(input, out_buf) {
                                Ok(TransformOutcome::Drop) => {
                                    *len = 0;
                                }
                                Ok(TransformOutcome::ForwardInput) => {}
                                Ok(TransformOutcome::ForwardOutput(n)) => {
                                    if n > input.len() {
                                        return set_err(ExecError::with_context(
                                            "resolved stage output too large",
                                            format!(
                                                "stage={key} out={n} in={} (stateful transform)",
                                                input.len()
                                            ),
                                        ));
                                    }
                                    data[..n].copy_from_slice(&out_buf[..n]);
                                    *len = n as u32;
                                }
                                Ok(_) => {
                                    return set_err(ExecError::new(
                                        "unsupported transform outcome (non-exhaustive)",
                                    ));
                                }
                                Err(e) => {
                                    return set_err(ExecError::with_context(
                                        "resolved stage failed",
                                        format!("stage={key}: {e}"),
                                    ));
                                }
                            },
                            ResolvedStage::Stateful(st) => match st.call(input, out_buf) {
                                Ok(n) => {
                                    if n > input.len() {
                                        return set_err(ExecError::with_context(
                                            "resolved stage output too large",
                                            format!(
                                                "stage={key} out={n} in={} (stateful)",
                                                input.len()
                                            ),
                                        ));
                                    }
                                    data[..n].copy_from_slice(&out_buf[..n]);
                                    *len = n as u32;
                                }
                                Err(e) => {
                                    return set_err(ExecError::with_context(
                                        "resolved stage failed",
                                        format!("stage={key}: {e}"),
                                    ));
                                }
                            },
                        }

                        let n_out = clamp_len(*len, data.len());
                        out.extend_from_slice(&data[..n_out]);
                        Ok(())
                    });

                    match res {
                        Ok(_) => {}
                        Err(byteor_pipeline_backings_shm::SequencedSlotsError::Stopped) => {
                            return Err(stage_err
                                .unwrap_or_else(|| ExecError::new("resolved stage stopped")));
                        }
                        Err(e) => return Err(ExecError::with_source("stage failed", e)),
                    }
                }
            }
        }
    }

    Ok(())
}

fn shard_lane_path(base: &std::path::Path, shard_idx: u32) -> std::path::PathBuf {
    let dir = base.parent().unwrap_or_else(|| std::path::Path::new("."));
    let stem = base
        .file_stem()
        .and_then(|s| s.to_str())
        .unwrap_or("byteor-single-ring");
    let ext = base.extension().and_then(|s| s.to_str());
    let name = match ext {
        Some(ext) => format!("{stem}-shard{shard_idx}.{ext}"),
        None => format!("{stem}-shard{shard_idx}"),
    };
    dir.join(name)
}

/// Run a PerKey-sharded SingleRing spec by partitioning input by shard key.
///
/// v1 sharding contract:
/// - the shard key is extracted deterministically from the message payload
/// - messages routed to the same shard preserve strict order
/// - no cross-shard ordering is implied
///
/// Returns one output vector per shard (in shard index order).
pub fn run_single_ring_shm_per_key_sharded_with_wait(
    spec: &PipelineSpecV1,
    lane_path: std::path::PathBuf,
    input: &[u8],
    wait: WaitPreset,
) -> Result<Vec<alloc::vec::Vec<u8>>, ExecError> {
    run_single_ring_shm_per_key_sharded_with_wait_and_resolver(spec, lane_path, input, wait, None)
}

/// Run a PerKey-sharded SingleRing spec by partitioning input by shard key,
/// resolving any `ResolverKey` stages using `resolver` (cold path).
///
/// Returns one output vector per shard (in shard index order).
pub fn run_single_ring_shm_per_key_sharded_with_wait_and_resolver(
    spec: &PipelineSpecV1,
    lane_path: std::path::PathBuf,
    input: &[u8],
    wait: WaitPreset,
    resolver: Option<&dyn StageResolver>,
) -> Result<Vec<alloc::vec::Vec<u8>>, ExecError> {
    run_single_ring_shm_per_key_sharded_with_wait_and_resolver_and_prefault(
        spec, lane_path, input, wait, resolver, false,
    )
}

/// Run a PerKey-sharded SingleRing spec by partitioning input by shard key,
/// resolving any `ResolverKey` stages using `resolver` (cold path), with explicit control over
/// SHM prefault.
pub fn run_single_ring_shm_per_key_sharded_with_wait_and_resolver_and_prefault(
    spec: &PipelineSpecV1,
    lane_path: std::path::PathBuf,
    input: &[u8],
    wait: WaitPreset,
    resolver: Option<&dyn StageResolver>,
    prefault: bool,
) -> Result<Vec<alloc::vec::Vec<u8>>, ExecError> {
    run_single_ring_shm_per_key_sharded_with_wait_and_resolver_and_thread_init_and_prefault(
        spec, lane_path, input, wait, resolver, None, prefault,
    )
}

/// Run a PerKey-sharded SingleRing spec by partitioning input by shard key,
/// resolving any `ResolverKey` stages using `resolver` (cold path), and invoking `thread_init`
/// for each stage thread (within each shard's executor).
///
/// Returns one output vector per shard (in shard index order).
pub fn run_single_ring_shm_per_key_sharded_with_wait_and_resolver_and_thread_init(
    spec: &PipelineSpecV1,
    lane_path: std::path::PathBuf,
    input: &[u8],
    wait: WaitPreset,
    resolver: Option<&dyn StageResolver>,
    thread_init: Option<ThreadInitHook>,
) -> Result<Vec<alloc::vec::Vec<u8>>, ExecError> {
    run_single_ring_shm_per_key_sharded_with_wait_and_resolver_and_thread_init_and_prefault(
        spec,
        lane_path,
        input,
        wait,
        resolver,
        thread_init,
        false,
    )
}

/// Run a PerKey-sharded SingleRing spec by partitioning input by shard key,
/// resolving any `ResolverKey` stages using `resolver` (cold path), invoking `thread_init`
/// for each stage thread, and explicitly controlling SHM prefault.
pub fn run_single_ring_shm_per_key_sharded_with_wait_and_resolver_and_thread_init_and_prefault(
    spec: &PipelineSpecV1,
    lane_path: std::path::PathBuf,
    input: &[u8],
    wait: WaitPreset,
    resolver: Option<&dyn StageResolver>,
    thread_init: Option<ThreadInitHook>,
    prefault: bool,
) -> Result<Vec<alloc::vec::Vec<u8>>, ExecError> {
    validate_v1(spec).map_err(|e| ExecError::with_source("spec validation failed", e))?;
    let ring = match spec {
        PipelineSpecV1::SingleRing(ring) => ring,
        _ => return Err(ExecError::new("spec is not SingleRing")),
    };

    let shards = ring.shards;
    if shards <= 1 {
        return Ok(alloc::vec::Vec::from([
            run_single_ring_shm_with_wait_and_resolver_and_thread_init_and_prefault(
                spec,
                lane_path,
                input,
                wait,
                resolver,
                thread_init,
                prefault,
            )?,
        ]));
    }

    // Partition input by shard key.
    let mut per_shard: alloc::vec::Vec<alloc::vec::Vec<u8>> =
        (0..shards).map(|_| alloc::vec::Vec::new()).collect();
    for &b in input {
        // v1 key extraction: first byte of payload (the only byte in this runner).
        let shard = (b as u32) % shards;
        per_shard[shard as usize].push(b);
    }

    // Run each shard independently over its own SHM file.
    //
    // Note: we use scoped threads so we can borrow `resolver` without requiring a `static`
    // lifetime or forcing all callers to allocate an `Arc<dyn StageResolver>`.
    let mut out = alloc::vec::Vec::with_capacity(shards as usize);
    std::thread::scope(|scope| {
        let mut handles = alloc::vec::Vec::with_capacity(shards as usize);
        for shard_idx in 0..shards {
            let mut spec_t = spec.clone();
            if let PipelineSpecV1::SingleRing(r) = &mut spec_t {
                r.shards = 1;
            }
            let path_t = shard_lane_path(&lane_path, shard_idx);
            let input_t = core::mem::take(&mut per_shard[shard_idx as usize]);
            let thread_init_t = thread_init.clone();
            handles.push(scope.spawn(move || {
                run_single_ring_shm_with_wait_and_resolver_and_thread_init_and_prefault(
                    &spec_t,
                    path_t,
                    &input_t,
                    wait,
                    resolver,
                    thread_init_t,
                    prefault,
                )
            }));
        }

        for h in handles {
            out.push(h.join().map_err(|_| ExecError::new("stage panic"))??);
        }
        Ok::<(), ExecError>(())
    })?;

    Ok(out)
}