byteor_pipeline_exec/

sp.rs

//! SP runtime scaffolding.

use byteor_pipeline_spec::{validate_v1, PipelineSpecV1};

use crate::ExecError;

/// Validate an SP spec.
pub fn validate_sp(spec: &PipelineSpecV1) -> Result<(), ExecError> {
    validate_v1(spec)
        .map_err(|e| ExecError::with_context("spec validation failed", e.message().to_string()))
}

#[cfg(feature = "shm")]
mod shm {
    use std::collections::BTreeMap;
    use std::path::PathBuf;
    use std::sync::atomic::{AtomicBool, AtomicU64, AtomicUsize, Ordering};
    use std::sync::Arc;
    use std::thread::JoinHandle;
    use std::time::{Duration, Instant};

    use byteor_pipeline_backings_shm::{
        attach_events_mpsc_rx, attach_events_mpsc_tx, attach_events_rx, attach_events_tx,
        attach_fanout_router, attach_fanout_rx, attach_fanout_tx, AttachOptions, BackingError,
        CursorBarrier, ShmEventsMpscRx, ShmEventsMpscTx, ShmEventsRx, ShmEventsTx, ShmFanoutRouter,
        ShmFanoutRx, ShmFanoutTx, ShmJournalPublisher, ShmJournalSubscriber, ShmSequencedSlots,
        ShmSequencedSlotsProducer, ShmSequencedSlotsSubscriber,
    };
    use byteor_pipeline_kernel::step::{publish_with_policy, TransformOutcome};
    use byteor_pipeline_kernel::{
        LaneRx, LaneRxBorrow, LaneTx, LaneTxError, LaneTxWith, LaneTxWithError, LaneTxWithResult,
    };
    use byteor_pipeline_spec::{
        EndpointKindV1, LaneGraphV1, LaneKindV1, MergePolicyV1, OnFullV1, RoleCfgV1,
    };
    use indexbus_core::{RouterMode, SpinWait, WaitStrategy};

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

    /// Start an SP runtime for a LaneGraph spec (SHM-backed, thread-per-role).
    pub fn run_sp(
        spec: &byteor_pipeline_spec::PipelineSpecV1,
        resolver: &dyn StageResolver,
        options: SpShmOptions,
    ) -> Result<LaneGraphSpRuntime, ExecError> {
        spawn_lane_graph_sp_shm_runtime(spec, resolver, options)
    }

    #[derive(Clone, Copy, Debug, PartialEq, Eq)]
    enum EventsAttachKind {
        Spsc,
        Mpsc,
    }

    fn backing_err(msg: &'static str, lane: &str, e: BackingError) -> ExecError {
        ExecError::with_context_source(msg, lane.to_string(), e)
    }

    fn sanitize_lane_name(lane: &str) -> String {
        lane.chars()
            .map(|c| {
                if c.is_ascii_alphanumeric() || c == '-' || c == '_' {
                    c
                } else {
                    '_'
                }
            })
            .collect()
    }

    /// SHM attach options for the LaneGraph SP runtime.
    ///
    /// If `dir` is set, each lane is backed by its own mmap file at
    /// `${dir}/${lane}.mmap` (with basic sanitization applied).
    #[derive(Clone, Debug, Default)]
    pub struct SpShmOptions {
        /// Whether to prefer blocking waits (non-HFT mode).
        pub blocking: bool,
        /// Optional directory where per-lane mmap files should be created.
        pub dir: Option<PathBuf>,
    }

    impl SpShmOptions {
        fn lane_path(&self, lane: &str) -> Option<PathBuf> {
            let dir = self.dir.as_ref()?;
            let lane = sanitize_lane_name(lane);
            Some(dir.join(format!("{lane}.mmap")))
        }

        /// Build backing attach options for a specific lane.
        pub fn lane_attach_options(&self, lane: &str) -> AttachOptions {
            AttachOptions {
                blocking: self.blocking,
                prefault: false,
                path: self.lane_path(lane),
                queue: 0,
            }
        }

        fn lane_attach_options_with_queue(&self, lane: &str, queue: usize) -> AttachOptions {
            AttachOptions {
                blocking: self.blocking,
                prefault: false,
                path: self.lane_path(lane),
                queue,
            }
        }

        fn ensure_dir(&self) -> Result<(), ExecError> {
            let Some(dir) = &self.dir else {
                return Ok(());
            };
            std::fs::create_dir_all(dir).map_err(|e| {
                ExecError::with_context_source(
                    "failed to create shm dir",
                    dir.display().to_string(),
                    e,
                )
            })
        }
    }

    enum AnyEventsRx {
        Spsc(ShmEventsRx),
        Mpsc(ShmEventsMpscRx),
    }

    impl LaneRx for AnyEventsRx {
        fn recv(&mut self, buf: &mut [u8]) -> Option<usize> {
            match self {
                AnyEventsRx::Spsc(rx) => rx.recv(buf),
                AnyEventsRx::Mpsc(rx) => rx.recv(buf),
            }
        }
    }

    impl LaneRxBorrow for AnyEventsRx {
        fn recv_with<F, R>(
            &mut self,
            f: F,
        ) -> core::result::Result<Option<R>, byteor_pipeline_kernel::LaneRxError>
        where
            F: FnOnce(&[u8]) -> R,
        {
            match self {
                AnyEventsRx::Spsc(rx) => rx.recv_with(f),
                AnyEventsRx::Mpsc(rx) => rx.recv_with(f),
            }
        }
    }

    enum AnyEventsTx {
        Spsc(ShmEventsTx),
        Mpsc(ShmEventsMpscTx),
    }

    struct RuntimeActivity {
        active_roles: AtomicUsize,
        progress_epoch: AtomicU64,
        roles: Vec<Arc<RuntimeRoleActivity>>,
    }

    #[derive(Debug)]
    struct RuntimeRoleActivity {
        name: String,
        role_kind: RoleKind,
        active_work: AtomicUsize,
        progress_epoch: AtomicU64,
    }

    impl RuntimeRoleActivity {
        fn new(name: String, role_kind: RoleKind) -> Self {
            Self {
                name,
                role_kind,
                active_work: AtomicUsize::new(0),
                progress_epoch: AtomicU64::new(0),
            }
        }

        fn snapshot(&self) -> LaneGraphRoleActivitySnapshot {
            LaneGraphRoleActivitySnapshot {
                name: self.name.clone(),
                role_kind: self.role_kind,
                active_work: self.active_work.load(Ordering::Relaxed),
                processed_count: self.progress_epoch.load(Ordering::Relaxed),
                progress_epoch: self.progress_epoch.load(Ordering::Relaxed),
            }
        }
    }

    /// Activity snapshot for a single LaneGraph role thread.
    #[derive(Clone, Debug, PartialEq, Eq)]
    pub struct LaneGraphRoleActivitySnapshot {
        /// Stable role name from the spec.
        pub name: String,
        /// Role kind.
        pub role_kind: RoleKind,
        /// Count of currently in-flight work units for this role.
        pub active_work: usize,
        /// Count of completed work units observed for this role.
        pub processed_count: u64,
        /// Monotonic progress counter incremented when the role completes work.
        pub progress_epoch: u64,
    }

    impl LaneGraphRoleActivitySnapshot {
        /// Returns true when the role is actively processing work.
        pub fn is_active(&self) -> bool {
            self.active_work > 0
        }

        /// Human-readable runtime status for operators.
        pub fn status(&self) -> &'static str {
            if self.is_active() {
                "active"
            } else {
                "idle"
            }
        }
    }

    impl RuntimeActivity {
        fn new(roles: Vec<Arc<RuntimeRoleActivity>>) -> Self {
            Self {
                active_roles: AtomicUsize::new(0),
                progress_epoch: AtomicU64::new(0),
                roles,
            }
        }

        fn start_work<'a>(&'a self, role: &'a RuntimeRoleActivity) -> RuntimeWorkGuard<'a> {
            self.active_roles.fetch_add(1, Ordering::Relaxed);
            role.active_work.fetch_add(1, Ordering::Relaxed);
            RuntimeWorkGuard {
                activity: self,
                role,
                progressed: false,
            }
        }
    }

    struct RuntimeWorkGuard<'a> {
        activity: &'a RuntimeActivity,
        role: &'a RuntimeRoleActivity,
        progressed: bool,
    }

    impl RuntimeWorkGuard<'_> {
        fn note_progress(&mut self) {
            if !self.progressed {
                self.progressed = true;
                self.activity.progress_epoch.fetch_add(1, Ordering::Relaxed);
                self.role.progress_epoch.fetch_add(1, Ordering::Relaxed);
            }
        }
    }

    impl Drop for RuntimeWorkGuard<'_> {
        fn drop(&mut self) {
            self.activity.active_roles.fetch_sub(1, Ordering::Relaxed);
            self.role.active_work.fetch_sub(1, Ordering::Relaxed);
        }
    }

    impl LaneTx for AnyEventsTx {
        fn publish(&mut self, msg: &[u8]) -> core::result::Result<(), LaneTxError> {
            match self {
                AnyEventsTx::Spsc(tx) => tx.publish(msg),
                AnyEventsTx::Mpsc(tx) => tx.publish(msg),
            }
        }
    }

    impl LaneTxWith for AnyEventsTx {
        fn publish_with<F>(&mut self, f: F) -> core::result::Result<(), LaneTxError>
        where
            F: FnOnce(&mut [u8]) -> usize,
        {
            match self {
                AnyEventsTx::Spsc(tx) => tx.publish_with(f),
                AnyEventsTx::Mpsc(tx) => tx.publish_with(f),
            }
        }
    }

    impl byteor_pipeline_kernel::LaneTxWithResult for AnyEventsTx {
        fn publish_with_result<F, E>(
            &mut self,
            f: F,
        ) -> core::result::Result<(), byteor_pipeline_kernel::LaneTxWithError<E>>
        where
            F: FnOnce(&mut [u8]) -> core::result::Result<usize, E>,
        {
            match self {
                AnyEventsTx::Spsc(tx) => tx.publish_with_result(f),
                AnyEventsTx::Mpsc(tx) => tx.publish_with_result(f),
            }
        }
    }

    fn events_attach_kind_by_lane(lg: &LaneGraphV1) -> BTreeMap<String, EventsAttachKind> {
        #[derive(Clone, Copy, Default)]
        struct LaneStats {
            producers: u32,
        }

        let mut stats = BTreeMap::<&str, LaneStats>::new();
        for lane in &lg.lanes {
            stats.insert(lane.name.as_str(), LaneStats::default());
        }

        for ep in &lg.endpoints {
            let Some(s) = stats.get_mut(ep.lane.as_str()) else {
                continue;
            };
            if matches!(ep.kind, EndpointKindV1::Ingress) {
                s.producers = s.producers.saturating_add(1);
            }
        }

        for role in &lg.roles {
            match role {
                RoleCfgV1::Stage(cfg) => {
                    if let Some(s) = stats.get_mut(cfg.tx.as_str()) {
                        s.producers = s.producers.saturating_add(1);
                    }
                }
                RoleCfgV1::Bridge(cfg) => {
                    if let Some(s) = stats.get_mut(cfg.tx.as_str()) {
                        s.producers = s.producers.saturating_add(1);
                    }
                }
                RoleCfgV1::Router(cfg) => {
                    for tx in &cfg.tx {
                        if let Some(s) = stats.get_mut(tx.as_str()) {
                            s.producers = s.producers.saturating_add(1);
                        }
                    }
                }
                RoleCfgV1::Merge(cfg) => {
                    if let Some(s) = stats.get_mut(cfg.tx.as_str()) {
                        s.producers = s.producers.saturating_add(1);
                    }
                }
            }
        }

        let mut out = BTreeMap::<String, EventsAttachKind>::new();
        for lane in &lg.lanes {
            if !matches!(lane.kind, LaneKindV1::Events) {
                continue;
            }
            let producers = stats
                .get(lane.name.as_str())
                .copied()
                .unwrap_or_default()
                .producers;

            // In v1, Events lanes may be multi-producer but must be single-consumer.
            // We choose SPSC for single-producer, MPSC otherwise.
            let kind = if producers > 1 {
                EventsAttachKind::Mpsc
            } else {
                EventsAttachKind::Spsc
            };
            out.insert(lane.name.clone(), kind);
        }
        out
    }

    fn attach_events_rx_any(
        opts: &SpShmOptions,
        lane: &str,
        kind: EventsAttachKind,
    ) -> Result<AnyEventsRx, ExecError> {
        let o = opts.lane_attach_options(lane);
        match kind {
            EventsAttachKind::Spsc => attach_events_rx(&o, lane)
                .map(AnyEventsRx::Spsc)
                .map_err(|e| backing_err("events rx attach failed", lane, e)),
            EventsAttachKind::Mpsc => attach_events_mpsc_rx(&o, lane)
                .map(AnyEventsRx::Mpsc)
                .map_err(|e| backing_err("events mpsc rx attach failed", lane, e)),
        }
    }

    fn attach_events_tx_any(
        opts: &SpShmOptions,
        lane: &str,
        kind: EventsAttachKind,
    ) -> Result<AnyEventsTx, ExecError> {
        let o = opts.lane_attach_options(lane);
        match kind {
            EventsAttachKind::Spsc => attach_events_tx(&o, lane)
                .map(AnyEventsTx::Spsc)
                .map_err(|e| backing_err("events tx attach failed", lane, e)),
            EventsAttachKind::Mpsc => attach_events_mpsc_tx(&o, lane)
                .map(AnyEventsTx::Mpsc)
                .map_err(|e| backing_err("events mpsc tx attach failed", lane, e)),
        }
    }

    fn attach_fanout_tx_one(opts: &SpShmOptions, lane: &str) -> Result<ShmFanoutTx, ExecError> {
        let o = opts.lane_attach_options(lane);
        attach_fanout_tx(&o, lane).map_err(|e| backing_err("fanout tx attach failed", lane, e))
    }

    fn attach_fanout_router_one(
        opts: &SpShmOptions,
        lane: &str,
    ) -> Result<ShmFanoutRouter, ExecError> {
        let o = opts.lane_attach_options(lane);
        attach_fanout_router(&o, lane)
            .map_err(|e| backing_err("fanout router attach failed", lane, e))
    }

    fn attach_fanout_rx_one(
        opts: &SpShmOptions,
        lane: &str,
        consumer_idx: usize,
    ) -> Result<ShmFanoutRx, ExecError> {
        let o = opts.lane_attach_options_with_queue(lane, consumer_idx);
        attach_fanout_rx(&o, lane).map_err(|e| backing_err("fanout rx attach failed", lane, e))
    }

    fn attach_journal_rx_one(
        opts: &SpShmOptions,
        lane: &str,
    ) -> Result<ShmJournalSubscriber, ExecError> {
        let o = opts.lane_attach_options(lane);
        byteor_pipeline_backings_shm::attach_journal_subscriber_tail(&o, lane)
            .map_err(|e| backing_err("journal rx attach failed", lane, e))
    }

    fn attach_journal_tx_one(
        opts: &SpShmOptions,
        lane: &str,
    ) -> Result<ShmJournalPublisher, ExecError> {
        let o = opts.lane_attach_options(lane);
        byteor_pipeline_backings_shm::attach_journal_publisher(
            &o,
            lane,
            indexbus_log::JournalPublisherConfig::default(),
        )
        .map_err(|e| backing_err("journal tx attach failed", lane, e))
    }

    struct SequencedSlotsRxOne {
        barrier: CursorBarrier,
        sub: ShmSequencedSlotsSubscriber,
    }

    fn attach_sequenced_slots_rx_one(
        opts: &SpShmOptions,
        lane: &str,
    ) -> Result<SequencedSlotsRxOne, ExecError> {
        let o = opts.lane_attach_options(lane);
        let shm = ShmSequencedSlots::attach(&o, lane)
            .map_err(|e| backing_err("sequenced-slots rx attach failed", lane, e))?;
        let barrier = CursorBarrier::new(&shm);
        let sub = ShmSequencedSlotsSubscriber::attach(&o, lane, 0)
            .map_err(|e| backing_err("sequenced-slots rx attach failed", lane, e))?;
        Ok(SequencedSlotsRxOne { barrier, sub })
    }

    fn attach_sequenced_slots_tx_one(
        opts: &SpShmOptions,
        lane: &str,
    ) -> Result<ShmSequencedSlotsProducer, ExecError> {
        let o = opts.lane_attach_options(lane);
        ShmSequencedSlotsProducer::attach(&o, lane, 1, 1)
            .map_err(|e| backing_err("sequenced-slots tx attach failed", lane, e))
    }

    fn publish_with_policy_with<Tx: LaneTxWith>(
        tx: &mut Tx,
        on_full: OnFullV1,
        wait: &mut impl WaitStrategy,
        mut fill: impl FnMut(&mut [u8]) -> usize,
    ) -> core::result::Result<(), LaneTxError> {
        loop {
            match tx.publish_with(|slot| fill(slot).min(slot.len())) {
                Ok(()) => {
                    wait.reset();
                    return Ok(());
                }
                Err(LaneTxError::Full) => match on_full {
                    OnFullV1::Drop => return Ok(()),
                    OnFullV1::Block => {
                        wait.wait();
                        continue;
                    }
                },
                Err(error) => return Err(error),
            }
        }
    }

    fn publish_with_policy_with_result<Tx: LaneTxWithResult, E>(
        tx: &mut Tx,
        on_full: OnFullV1,
        wait: &mut impl WaitStrategy,
        mut fill: impl FnMut(&mut [u8]) -> core::result::Result<usize, E>,
    ) -> core::result::Result<(), LaneTxWithError<E>> {
        loop {
            match tx.publish_with_result(|slot| fill(slot).map(|n| n.min(slot.len()))) {
                Ok(()) => {
                    wait.reset();
                    return Ok(());
                }
                Err(LaneTxWithError::Full) => match on_full {
                    OnFullV1::Drop => return Ok(()),
                    OnFullV1::Block => {
                        wait.wait();
                        continue;
                    }
                },
                Err(error @ LaneTxWithError::Encode(_)) => return Err(error),
                Err(error @ LaneTxWithError::Failed) => return Err(error),
            }
        }
    }

    fn run_stage_role_loop(
        stop: &AtomicBool,
        activity: &RuntimeActivity,
        role_activity: &RuntimeRoleActivity,
        rx: &mut AnyEventsRx,
        tx: &mut AnyEventsTx,
        mut stage: ResolvedStage,
        on_full: OnFullV1,
    ) -> Result<(), ExecError> {
        let mut idle_wait = SpinWait::default();
        let mut backpressure_wait = SpinWait::default();
        let mut scratch = [0u8; indexbus_core::INDEXBUS_SLOT_DATA_SIZE];

        while !stop.load(Ordering::Relaxed) {
            let did = match &mut stage {
                ResolvedStage::MapOk(map_ok) => {
                    let res = rx
                        .recv_with(|input| {
                            let mut work = activity.start_work(role_activity);
                            publish_with_policy_with(tx, on_full, &mut backpressure_wait, |slot| {
                                map_ok(input, slot)
                            })
                            .map_err(|_| ExecError::new("stage publish failed"))?;
                            work.note_progress();
                            Ok::<(), ExecError>(())
                        })
                        .map_err(|_| ExecError::new("stage recv failed"))?;
                    match res {
                        None => false,
                        Some(r) => {
                            r?;
                            true
                        }
                    }
                }
                ResolvedStage::Map(map) => {
                    let res = rx
                        .recv_with(|input| {
                            let mut work = activity.start_work(role_activity);
                            publish_with_policy_with_result(
                                tx,
                                on_full,
                                &mut backpressure_wait,
                                |slot| map(input, slot),
                            )
                            .map_err(|error| match error {
                                byteor_pipeline_kernel::LaneTxWithError::Encode(error) => {
                                    ExecError::with_context("stage error", error.to_string())
                                }
                                _ => ExecError::new("stage publish failed"),
                            })?;
                            work.note_progress();
                            Ok::<(), ExecError>(())
                        })
                        .map_err(|_| ExecError::new("stage recv failed"))?;
                    match res {
                        None => false,
                        Some(r) => {
                            r?;
                            true
                        }
                    }
                }
                ResolvedStage::Filter(filter) => {
                    let res = rx
                        .recv_with(|input| {
                            let mut work = activity.start_work(role_activity);
                            let keep = filter(input).map_err(|e| {
                                ExecError::with_context("stage error", e.to_string())
                            })?;
                            if keep {
                                publish_with_policy(tx, input, on_full, &mut backpressure_wait)
                                    .map_err(|_| ExecError::new("stage publish failed"))?;
                            }
                            work.note_progress();
                            Ok::<(), ExecError>(())
                        })
                        .map_err(|_| ExecError::new("stage recv failed"))?;
                    match res {
                        None => false,
                        Some(r) => {
                            r?;
                            true
                        }
                    }
                }
                ResolvedStage::Inspect(inspect) => {
                    let res = rx
                        .recv_with(|input| {
                            let mut work = activity.start_work(role_activity);
                            inspect(input).map_err(|e| {
                                ExecError::with_context("stage error", e.to_string())
                            })?;
                            publish_with_policy(tx, input, on_full, &mut backpressure_wait)
                                .map_err(|_| ExecError::new("stage publish failed"))?;
                            work.note_progress();
                            Ok::<(), ExecError>(())
                        })
                        .map_err(|_| ExecError::new("stage recv failed"))?;
                    match res {
                        None => false,
                        Some(r) => {
                            r?;
                            true
                        }
                    }
                }
                ResolvedStage::StatefulTransform(stateful) => {
                    let res = rx
                        .recv_with(|input| {
                            let mut work = activity.start_work(role_activity);
                            let outcome = stateful.call(input, &mut scratch).map_err(|e| {
                                ExecError::with_context("stage error", e.to_string())
                            })?;

                            match outcome {
                                TransformOutcome::Drop => {}
                                TransformOutcome::ForwardInput => {
                                    publish_with_policy(tx, input, on_full, &mut backpressure_wait)
                                        .map_err(|_| ExecError::new("stage publish failed"))?;
                                }
                                TransformOutcome::ForwardOutput(out_n) => {
                                    let out_n = out_n.min(scratch.len());
                                    publish_with_policy(
                                        tx,
                                        &scratch[..out_n],
                                        on_full,
                                        &mut backpressure_wait,
                                    )
                                    .map_err(|_| ExecError::new("stage publish failed"))?;
                                }
                                _ => {
                                    return Err(ExecError::new(
                                        "unsupported transform outcome (non-exhaustive)",
                                    ));
                                }
                            }

                            work.note_progress();
                            Ok::<(), ExecError>(())
                        })
                        .map_err(|_| ExecError::new("stage recv failed"))?;
                    match res {
                        None => false,
                        Some(r) => {
                            r?;
                            true
                        }
                    }
                }
                ResolvedStage::Stateful(stateful) => {
                    let res = rx
                        .recv_with(|input| {
                            let mut work = activity.start_work(role_activity);
                            publish_with_policy_with_result(
                                tx,
                                on_full,
                                &mut backpressure_wait,
                                |slot| stateful.call(input, slot),
                            )
                            .map_err(|error| match error {
                                byteor_pipeline_kernel::LaneTxWithError::Encode(error) => {
                                    ExecError::with_context("stage error", error.to_string())
                                }
                                _ => ExecError::new("stage publish failed"),
                            })?;
                            work.note_progress();
                            Ok::<(), ExecError>(())
                        })
                        .map_err(|_| ExecError::new("stage recv failed"))?;
                    match res {
                        None => false,
                        Some(r) => {
                            r?;
                            true
                        }
                    }
                }
            };

            if did {
                idle_wait.reset();
            } else {
                idle_wait.wait();
            }
        }

        Ok(())
    }

    fn run_stage_role_loop_copy<Rx, Tx>(
        stop: &AtomicBool,
        activity: &RuntimeActivity,
        role_activity: &RuntimeRoleActivity,
        rx: &mut Rx,
        tx: &mut Tx,
        mut stage: ResolvedStage,
        on_full: OnFullV1,
    ) -> Result<(), ExecError>
    where
        Rx: LaneRx,
        Tx: LaneTx,
    {
        let mut idle_wait = SpinWait::default();
        let mut backpressure_wait = SpinWait::default();
        let mut input = [0u8; indexbus_core::INDEXBUS_SLOT_DATA_SIZE];
        let mut scratch = [0u8; indexbus_core::INDEXBUS_SLOT_DATA_SIZE];

        while !stop.load(Ordering::Relaxed) {
            let Some(n) = rx.recv(&mut input) else {
                idle_wait.wait();
                continue;
            };

            let mut work = activity.start_work(role_activity);
            let input = &input[..n];
            match &mut stage {
                ResolvedStage::MapOk(map_ok) => {
                    let out_n = map_ok(input, &mut scratch).min(scratch.len());
                    publish_with_policy(tx, &scratch[..out_n], on_full, &mut backpressure_wait)
                        .map_err(|_| ExecError::new("stage publish failed"))?;
                }
                ResolvedStage::Map(map) => {
                    let out_n = map(input, &mut scratch)
                        .map_err(|e| ExecError::with_context("stage error", e.to_string()))?
                        .min(scratch.len());
                    publish_with_policy(tx, &scratch[..out_n], on_full, &mut backpressure_wait)
                        .map_err(|_| ExecError::new("stage publish failed"))?;
                }
                ResolvedStage::Filter(filter) => {
                    if filter(input)
                        .map_err(|e| ExecError::with_context("stage error", e.to_string()))?
                    {
                        publish_with_policy(tx, input, on_full, &mut backpressure_wait)
                            .map_err(|_| ExecError::new("stage publish failed"))?;
                    }
                }
                ResolvedStage::Inspect(inspect) => {
                    inspect(input)
                        .map_err(|e| ExecError::with_context("stage error", e.to_string()))?;
                    publish_with_policy(tx, input, on_full, &mut backpressure_wait)
                        .map_err(|_| ExecError::new("stage publish failed"))?;
                }
                ResolvedStage::StatefulTransform(stateful) => {
                    let outcome = stateful
                        .call(input, &mut scratch)
                        .map_err(|e| ExecError::with_context("stage error", e.to_string()))?;
                    match outcome {
                        TransformOutcome::Drop => {}
                        TransformOutcome::ForwardInput => {
                            publish_with_policy(tx, input, on_full, &mut backpressure_wait)
                                .map_err(|_| ExecError::new("stage publish failed"))?;
                        }
                        TransformOutcome::ForwardOutput(out_n) => {
                            let out_n = out_n.min(scratch.len());
                            publish_with_policy(
                                tx,
                                &scratch[..out_n],
                                on_full,
                                &mut backpressure_wait,
                            )
                            .map_err(|_| ExecError::new("stage publish failed"))?;
                        }
                        _ => {
                            return Err(ExecError::new(
                                "unsupported transform outcome (non-exhaustive)",
                            ));
                        }
                    }
                }
                ResolvedStage::Stateful(stateful) => {
                    let out_n = stateful
                        .call(input, &mut scratch)
                        .map_err(|e| ExecError::with_context("stage error", e.to_string()))?
                        .min(scratch.len());
                    publish_with_policy(tx, &scratch[..out_n], on_full, &mut backpressure_wait)
                        .map_err(|_| ExecError::new("stage publish failed"))?;
                }
            }

            work.note_progress();
            idle_wait.reset();
        }

        Ok(())
    }

    fn run_bridge_role_loop<Rx, Tx>(
        stop: &AtomicBool,
        activity: &RuntimeActivity,
        role_activity: &RuntimeRoleActivity,
        rx: &mut Rx,
        tx: &mut Tx,
        on_full: OnFullV1,
    ) -> Result<(), ExecError>
    where
        Rx: LaneRxBorrow,
        Tx: LaneTxWith,
    {
        let mut idle_wait = SpinWait::default();
        let mut backpressure_wait = SpinWait::default();

        while !stop.load(Ordering::Relaxed) {
            let res = rx
                .recv_with(|input| {
                    let mut work = activity.start_work(role_activity);
                    publish_with_policy_with(tx, on_full, &mut backpressure_wait, |slot| {
                        let n = input.len().min(slot.len());
                        slot[..n].copy_from_slice(&input[..n]);
                        n
                    })
                    .map_err(|_| ExecError::new("bridge publish failed"))?;
                    work.note_progress();
                    Ok::<(), ExecError>(())
                })
                .map_err(|_| ExecError::new("bridge recv failed"))?;
            let did = match res {
                None => false,
                Some(r) => {
                    r?;
                    true
                }
            };
            if did {
                idle_wait.reset();
            } else {
                idle_wait.wait();
            }
        }
        Ok(())
    }

    fn run_bridge_role_loop_copy<Rx, Tx>(
        stop: &AtomicBool,
        activity: &RuntimeActivity,
        role_activity: &RuntimeRoleActivity,
        rx: &mut Rx,
        tx: &mut Tx,
        on_full: OnFullV1,
    ) -> Result<(), ExecError>
    where
        Rx: LaneRx,
        Tx: LaneTx,
    {
        let mut idle_wait = SpinWait::default();
        let mut backpressure_wait = SpinWait::default();
        let mut buf = [0u8; indexbus_core::INDEXBUS_SLOT_DATA_SIZE];

        while !stop.load(Ordering::Relaxed) {
            let Some(n) = rx.recv(&mut buf) else {
                idle_wait.wait();
                continue;
            };

            let mut work = activity.start_work(role_activity);
            publish_with_policy(tx, &buf[..n], on_full, &mut backpressure_wait)
                .map_err(|_| ExecError::new("bridge publish failed"))?;
            work.note_progress();
            idle_wait.reset();
        }

        Ok(())
    }

    fn run_stage_role_loop_sequenced_slots(
        stop: &AtomicBool,
        activity: &RuntimeActivity,
        role_activity: &RuntimeRoleActivity,
        rx: &mut SequencedSlotsRxOne,
        tx: &mut ShmSequencedSlotsProducer,
        mut stage: ResolvedStage,
        on_full: OnFullV1,
    ) -> Result<(), ExecError> {
        let mut idle_wait = SpinWait::default();
        let mut backpressure_wait = SpinWait::default();
        let mut input = [0u8; indexbus_core::INDEXBUS_SLOT_DATA_SIZE];
        let mut scratch = [0u8; indexbus_core::INDEXBUS_SLOT_DATA_SIZE];

        while !stop.load(Ordering::Relaxed) {
            let n = match rx.sub.recv_next_into_from_stoppable(
                &rx.barrier,
                &mut idle_wait,
                &mut input,
                stop,
            ) {
                Ok((_seq, n)) => n,
                Err(byteor_pipeline_backings_shm::SequencedSlotsError::Stopped) => break,
                Err(e) => {
                    return Err(ExecError::with_source(
                        "sequenced-slots stage recv failed",
                        e,
                    ))
                }
            };

            let mut work = activity.start_work(role_activity);
            let input = &input[..n];
            match &mut stage {
                ResolvedStage::MapOk(map_ok) => {
                    let out_n = map_ok(input, &mut scratch).min(scratch.len());
                    publish_with_policy(tx, &scratch[..out_n], on_full, &mut backpressure_wait)
                        .map_err(|_| ExecError::new("stage publish failed"))?;
                }
                ResolvedStage::Map(map) => {
                    let out_n = map(input, &mut scratch)
                        .map_err(|e| ExecError::with_context("stage error", e.to_string()))?
                        .min(scratch.len());
                    publish_with_policy(tx, &scratch[..out_n], on_full, &mut backpressure_wait)
                        .map_err(|_| ExecError::new("stage publish failed"))?;
                }
                ResolvedStage::Filter(filter) => {
                    if filter(input)
                        .map_err(|e| ExecError::with_context("stage error", e.to_string()))?
                    {
                        publish_with_policy(tx, input, on_full, &mut backpressure_wait)
                            .map_err(|_| ExecError::new("stage publish failed"))?;
                    }
                }
                ResolvedStage::Inspect(inspect) => {
                    inspect(input)
                        .map_err(|e| ExecError::with_context("stage error", e.to_string()))?;
                    publish_with_policy(tx, input, on_full, &mut backpressure_wait)
                        .map_err(|_| ExecError::new("stage publish failed"))?;
                }
                ResolvedStage::StatefulTransform(stateful) => {
                    let outcome = stateful
                        .call(input, &mut scratch)
                        .map_err(|e| ExecError::with_context("stage error", e.to_string()))?;
                    match outcome {
                        TransformOutcome::Drop => {}
                        TransformOutcome::ForwardInput => {
                            publish_with_policy(tx, input, on_full, &mut backpressure_wait)
                                .map_err(|_| ExecError::new("stage publish failed"))?;
                        }
                        TransformOutcome::ForwardOutput(out_n) => {
                            let out_n = out_n.min(scratch.len());
                            publish_with_policy(
                                tx,
                                &scratch[..out_n],
                                on_full,
                                &mut backpressure_wait,
                            )
                            .map_err(|_| ExecError::new("stage publish failed"))?;
                        }
                        _ => {
                            return Err(ExecError::new(
                                "unsupported transform outcome (non-exhaustive)",
                            ));
                        }
                    }
                }
                ResolvedStage::Stateful(stateful) => {
                    let out_n = stateful
                        .call(input, &mut scratch)
                        .map_err(|e| ExecError::with_context("stage error", e.to_string()))?
                        .min(scratch.len());
                    publish_with_policy(tx, &scratch[..out_n], on_full, &mut backpressure_wait)
                        .map_err(|_| ExecError::new("stage publish failed"))?;
                }
            }

            work.note_progress();
            idle_wait.reset();
        }

        Ok(())
    }

    fn run_bridge_role_loop_sequenced_slots(
        stop: &AtomicBool,
        activity: &RuntimeActivity,
        role_activity: &RuntimeRoleActivity,
        rx: &mut SequencedSlotsRxOne,
        tx: &mut ShmSequencedSlotsProducer,
        on_full: OnFullV1,
    ) -> Result<(), ExecError> {
        let mut idle_wait = SpinWait::default();
        let mut backpressure_wait = SpinWait::default();
        let mut buf = [0u8; indexbus_core::INDEXBUS_SLOT_DATA_SIZE];

        while !stop.load(Ordering::Relaxed) {
            let n = match rx.sub.recv_next_into_from_stoppable(
                &rx.barrier,
                &mut idle_wait,
                &mut buf,
                stop,
            ) {
                Ok((_seq, n)) => n,
                Err(byteor_pipeline_backings_shm::SequencedSlotsError::Stopped) => break,
                Err(e) => {
                    return Err(ExecError::with_source(
                        "sequenced-slots bridge recv failed",
                        e,
                    ));
                }
            };

            let mut work = activity.start_work(role_activity);
            publish_with_policy(tx, &buf[..n], on_full, &mut backpressure_wait)
                .map_err(|_| ExecError::new("bridge publish failed"))?;
            work.note_progress();
            idle_wait.reset();
        }

        Ok(())
    }

    fn run_merge_role_loop<Rx, Tx>(
        stop: &AtomicBool,
        activity: &RuntimeActivity,
        role_activity: &RuntimeRoleActivity,
        rxs: &mut [Rx],
        tx: &mut Tx,
        policy: MergePolicyV1,
        on_full: OnFullV1,
    ) -> Result<(), ExecError>
    where
        Rx: LaneRx,
        Tx: LaneTx,
    {
        if rxs.is_empty() {
            return Err(ExecError::new("merge requires at least one rx lane"));
        }
        if !matches!(policy, MergePolicyV1::RoundRobin) {
            return Err(ExecError::new("unsupported merge policy"));
        }

        let mut idle_wait = SpinWait::default();
        let mut backpressure_wait = SpinWait::default();
        let mut rr: usize = 0;
        let mut buf = [0u8; indexbus_core::INDEXBUS_SLOT_DATA_SIZE];

        while !stop.load(Ordering::Relaxed) {
            let n = rxs.len();
            let start = rr % n;
            let mut did_any = false;

            for offset in 0..n {
                let idx = (start + offset) % n;
                let Some(msg_n) = rxs[idx].recv(&mut buf) else {
                    continue;
                };
                let mut work = activity.start_work(role_activity);
                publish_with_policy(tx, &buf[..msg_n], on_full, &mut backpressure_wait)
                    .map_err(|_| ExecError::new("merge publish failed"))?;
                work.note_progress();
                rr = (idx + 1) % n;
                did_any = true;
                break;
            }

            if !did_any {
                rr = (start + 1) % n;
                idle_wait.wait();
            } else {
                idle_wait.reset();
            }
        }

        Ok(())
    }

    fn run_router_role_loop(
        stop: &AtomicBool,
        activity: &RuntimeActivity,
        role_activity: &RuntimeRoleActivity,
        router: &ShmFanoutRouter,
        fanout_rxs: &mut [ShmFanoutRx],
        txs: &mut [AnyEventsTx],
        on_full: OnFullV1,
    ) -> Result<(), ExecError> {
        if fanout_rxs.is_empty() || txs.is_empty() {
            return Err(ExecError::new("router requires non-empty lanes"));
        }
        if fanout_rxs.len() != txs.len() {
            return Err(ExecError::new("router lane count mismatch"));
        }

        let mut idle_wait = SpinWait::default();
        let mut backpressure_wait = SpinWait::default();
        let mut buf = [0u8; indexbus_core::INDEXBUS_SLOT_DATA_SIZE];

        while !stop.load(Ordering::Relaxed) {
            let routed = router.route_once(RouterMode::Broadcast);
            if !routed {
                idle_wait.wait();
                continue;
            }

            let mut work = activity.start_work(role_activity);
            idle_wait.reset();

            for (i, (rx, tx)) in fanout_rxs.iter_mut().zip(txs.iter_mut()).enumerate() {
                let Some(n) = rx.recv(&mut buf) else {
                    return Err(ExecError::with_context(
                        "fanout consumer recv failed",
                        format!("consumer {i}"),
                    ));
                };
                publish_with_policy(tx, &buf[..n], on_full, &mut backpressure_wait)
                    .map_err(|_| ExecError::new("router publish failed"))?;
            }

            work.note_progress();
        }

        Ok(())
    }

    /// Handle for a spawned LaneGraph SP runtime.
    pub struct LaneGraphSpRuntime {
        stop: Arc<AtomicBool>,
        activity: Arc<RuntimeActivity>,
        joins: Vec<JoinHandle<Result<(), ExecError>>>,
    }

    impl LaneGraphSpRuntime {
        /// Snapshot per-role runtime activity for operator-facing monitoring.
        pub fn role_activity_snapshot(&self) -> Vec<LaneGraphRoleActivitySnapshot> {
            self.activity
                .roles
                .iter()
                .map(|role| role.snapshot())
                .collect()
        }

        /// Wait until all roles are idle and no new progress is observed for a short settle window.
        pub fn wait_until_idle(&self, timeout: Duration) -> Result<(), ExecError> {
            const POLL_INTERVAL: Duration = Duration::from_millis(10);
            const IDLE_SETTLE: Duration = Duration::from_millis(20);

            let start = Instant::now();
            let mut stable_since: Option<Instant> = None;
            let mut last_epoch = self.activity.progress_epoch.load(Ordering::Relaxed);

            loop {
                if self.joins.iter().any(|j| j.is_finished()) {
                    return Err(ExecError::new("role exited before runtime became idle"));
                }

                let active_roles = self.activity.active_roles.load(Ordering::Relaxed);
                let epoch = self.activity.progress_epoch.load(Ordering::Relaxed);

                if active_roles == 0 {
                    if epoch == last_epoch {
                        let stable = stable_since.get_or_insert_with(Instant::now);
                        if stable.elapsed() >= IDLE_SETTLE {
                            return Ok(());
                        }
                    } else {
                        last_epoch = epoch;
                        stable_since = Some(Instant::now());
                    }
                } else {
                    last_epoch = epoch;
                    stable_since = None;
                }

                if start.elapsed() >= timeout {
                    return Err(ExecError::with_context(
                        "lane graph idle wait timed out",
                        format!(
                            "timeout_ms={} active_roles={} progress_epoch={epoch}",
                            timeout.as_millis(),
                            active_roles,
                        ),
                    ));
                }

                std::thread::sleep(POLL_INTERVAL);
            }
        }

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

        /// Join all role threads.
        pub fn join(&mut self) -> Result<(), ExecError> {
            for j in self.joins.drain(..) {
                let r = j
                    .join()
                    .map_err(|_| ExecError::new("role thread panicked"))?;
                r?;
            }
            Ok(())
        }

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

    /// Spawn a SHM-backed LaneGraph SP runtime (thread-per-role).
    ///
    /// Current scope: Events + FanoutBroadcast + Journal + SequencedSlots.
    pub fn spawn_lane_graph_sp_shm_runtime(
        spec: &byteor_pipeline_spec::PipelineSpecV1,
        resolver: &dyn StageResolver,
        options: SpShmOptions,
    ) -> Result<LaneGraphSpRuntime, ExecError> {
        spawn_lane_graph_sp_shm_runtime_with_thread_init(spec, resolver, options, None)
    }

    /// Spawn a SHM-backed LaneGraph SP runtime (thread-per-role), invoking `thread_init` at the
    /// start of each role thread (before entering hot loops).
    pub fn spawn_lane_graph_sp_shm_runtime_with_thread_init(
        spec: &byteor_pipeline_spec::PipelineSpecV1,
        resolver: &dyn StageResolver,
        options: SpShmOptions,
        thread_init: Option<ThreadInitHook>,
    ) -> Result<LaneGraphSpRuntime, ExecError> {
        options.ensure_dir()?;

        crate::validate_sp(spec)?;

        let byteor_pipeline_spec::PipelineSpecV1::LaneGraph(lg) = spec else {
            return Err(ExecError::new("spec is not a lane graph"));
        };

        let lane_kinds: BTreeMap<String, LaneKindV1> =
            lg.lanes.iter().map(|l| (l.name.clone(), l.kind)).collect();
        let lane_kinds = Arc::new(lane_kinds);

        let events_kinds = Arc::new(events_attach_kind_by_lane(lg));

        let on_full = lg.on_full;

        let stop = Arc::new(AtomicBool::new(false));
        let role_activities = lg
            .roles
            .iter()
            .map(|role| {
                Arc::new(RuntimeRoleActivity::new(
                    role.name().to_string(),
                    role_kind(role),
                ))
            })
            .collect::<Vec<_>>();
        let activity = Arc::new(RuntimeActivity::new(role_activities));
        let mut joins: Vec<JoinHandle<Result<(), ExecError>>> = Vec::with_capacity(lg.roles.len());

        for (role, role_activity) in lg.roles.iter().cloned().zip(activity.roles.iter().cloned()) {
            let stop_t = stop.clone();
            let activity_t = activity.clone();
            let role_activity_t = role_activity.clone();
            let options_t = options.clone();
            let lane_kinds_t = lane_kinds.clone();
            let events_kinds_t = events_kinds.clone();
            let thread_init_t = thread_init.clone();

            let role_name = role.name().to_string();

            let join = match role {
                RoleCfgV1::Stage(cfg) => {
                    let stage = resolve_or_err(resolver, cfg.stage.as_str())?;
                    std::thread::Builder::new()
                        .name(format!("byteor-sp:{}", role_name))
                        .spawn(move || {
                            if let Some(h) = &thread_init_t {
                                let ctx =
                                    ThreadInitContext::new(role_name.clone(), RoleKind::Stage);
                                h(&ctx)?;
                            }

                            match (
                                lane_kinds_t.get(cfg.rx.as_str()).copied(),
                                lane_kinds_t.get(cfg.tx.as_str()).copied(),
                            ) {
                                (Some(LaneKindV1::Events), Some(LaneKindV1::Events)) => {
                                    let rx_kind = events_kinds_t
                                        .get(cfg.rx.as_str())
                                        .copied()
                                        .unwrap_or(EventsAttachKind::Mpsc);
                                    let tx_kind = events_kinds_t
                                        .get(cfg.tx.as_str())
                                        .copied()
                                        .unwrap_or(EventsAttachKind::Mpsc);

                                    let mut rx =
                                        attach_events_rx_any(&options_t, cfg.rx.as_str(), rx_kind)?;
                                    let mut tx =
                                        attach_events_tx_any(&options_t, cfg.tx.as_str(), tx_kind)?;

                                    run_stage_role_loop(
                                        stop_t.as_ref(),
                                        activity_t.as_ref(),
                                        role_activity_t.as_ref(),
                                        &mut rx,
                                        &mut tx,
                                        stage,
                                        on_full,
                                    )
                                }
                                (Some(LaneKindV1::Journal), Some(LaneKindV1::Journal)) => {
                                    let mut rx =
                                        attach_journal_rx_one(&options_t, cfg.rx.as_str())?;
                                    let mut tx =
                                        attach_journal_tx_one(&options_t, cfg.tx.as_str())?;

                                    run_stage_role_loop_copy(
                                        stop_t.as_ref(),
                                        activity_t.as_ref(),
                                        role_activity_t.as_ref(),
                                        &mut rx,
                                        &mut tx,
                                        stage,
                                        on_full,
                                    )
                                }
                                (
                                    Some(LaneKindV1::SequencedSlots { .. }),
                                    Some(LaneKindV1::SequencedSlots { .. }),
                                ) => {
                                    let mut rx =
                                        attach_sequenced_slots_rx_one(&options_t, cfg.rx.as_str())?;
                                    let mut tx =
                                        attach_sequenced_slots_tx_one(&options_t, cfg.tx.as_str())?;

                                    run_stage_role_loop_sequenced_slots(
                                        stop_t.as_ref(),
                                        activity_t.as_ref(),
                                        role_activity_t.as_ref(),
                                        &mut rx,
                                        &mut tx,
                                        stage,
                                        on_full,
                                    )
                                }
                                _ => Err(ExecError::new("unsupported stage rx lane kind")),
                            }
                        })
                        .map_err(|e| ExecError::with_source("failed to spawn role thread", e))?
                }
                RoleCfgV1::Bridge(cfg) => std::thread::Builder::new()
                    .name(format!("byteor-sp:{}", role_name))
                    .spawn(move || {
                        if let Some(h) = &thread_init_t {
                            let ctx = ThreadInitContext::new(role_name.clone(), RoleKind::Bridge);
                            h(&ctx)?;
                        }

                        let Some(rx_kind) = lane_kinds_t.get(cfg.rx.as_str()).copied() else {
                            return Err(ExecError::new("unknown bridge rx lane"));
                        };
                        let Some(tx_kind) = lane_kinds_t.get(cfg.tx.as_str()).copied() else {
                            return Err(ExecError::new("unknown bridge tx lane"));
                        };

                        match (rx_kind, tx_kind) {
                            (LaneKindV1::Events, LaneKindV1::Events) => {
                                let rx_attach = events_kinds_t
                                    .get(cfg.rx.as_str())
                                    .copied()
                                    .unwrap_or(EventsAttachKind::Mpsc);
                                let tx_attach = events_kinds_t
                                    .get(cfg.tx.as_str())
                                    .copied()
                                    .unwrap_or(EventsAttachKind::Mpsc);
                                let mut rx =
                                    attach_events_rx_any(&options_t, cfg.rx.as_str(), rx_attach)?;
                                let mut tx =
                                    attach_events_tx_any(&options_t, cfg.tx.as_str(), tx_attach)?;
                                run_bridge_role_loop(
                                    stop_t.as_ref(),
                                    activity_t.as_ref(),
                                    role_activity_t.as_ref(),
                                    &mut rx,
                                    &mut tx,
                                    on_full,
                                )
                            }
                            (LaneKindV1::Events, LaneKindV1::FanoutBroadcast { .. }) => {
                                let rx_attach = events_kinds_t
                                    .get(cfg.rx.as_str())
                                    .copied()
                                    .unwrap_or(EventsAttachKind::Mpsc);
                                let mut rx =
                                    attach_events_rx_any(&options_t, cfg.rx.as_str(), rx_attach)?;
                                let mut tx = attach_fanout_tx_one(&options_t, cfg.tx.as_str())?;
                                run_bridge_role_loop(
                                    stop_t.as_ref(),
                                    activity_t.as_ref(),
                                    role_activity_t.as_ref(),
                                    &mut rx,
                                    &mut tx,
                                    on_full,
                                )
                            }
                            (LaneKindV1::Journal, LaneKindV1::Journal) => {
                                let mut rx = attach_journal_rx_one(&options_t, cfg.rx.as_str())?;
                                let mut tx = attach_journal_tx_one(&options_t, cfg.tx.as_str())?;
                                run_bridge_role_loop_copy(
                                    stop_t.as_ref(),
                                    activity_t.as_ref(),
                                    role_activity_t.as_ref(),
                                    &mut rx,
                                    &mut tx,
                                    on_full,
                                )
                            }
                            (
                                LaneKindV1::SequencedSlots { .. },
                                LaneKindV1::SequencedSlots { .. },
                            ) => {
                                let mut rx =
                                    attach_sequenced_slots_rx_one(&options_t, cfg.rx.as_str())?;
                                let mut tx =
                                    attach_sequenced_slots_tx_one(&options_t, cfg.tx.as_str())?;
                                run_bridge_role_loop_sequenced_slots(
                                    stop_t.as_ref(),
                                    activity_t.as_ref(),
                                    role_activity_t.as_ref(),
                                    &mut rx,
                                    &mut tx,
                                    on_full,
                                )
                            }
                            _ => Err(ExecError::new("unsupported bridge lane kinds")),
                        }
                    })
                    .map_err(|e| ExecError::with_source("failed to spawn role thread", e))?,
                RoleCfgV1::Merge(cfg) => std::thread::Builder::new()
                    .name(format!("byteor-sp:{}", role_name))
                    .spawn(move || {
                        if let Some(h) = &thread_init_t {
                            let ctx = ThreadInitContext::new(role_name.clone(), RoleKind::Merge);
                            h(&ctx)?;
                        }

                        match lane_kinds_t.get(cfg.tx.as_str()).copied() {
                            Some(LaneKindV1::Events) => {
                                let tx_attach = events_kinds_t
                                    .get(cfg.tx.as_str())
                                    .copied()
                                    .unwrap_or(EventsAttachKind::Mpsc);
                                let mut tx =
                                    attach_events_tx_any(&options_t, cfg.tx.as_str(), tx_attach)?;

                                let mut rxs: Vec<AnyEventsRx> = Vec::with_capacity(cfg.rx.len());
                                for rx_lane in &cfg.rx {
                                    let Some(LaneKindV1::Events) =
                                        lane_kinds_t.get(rx_lane.as_str()).copied()
                                    else {
                                        return Err(ExecError::new(
                                            "unsupported merge rx lane kind",
                                        ));
                                    };
                                    let rx_attach = events_kinds_t
                                        .get(rx_lane.as_str())
                                        .copied()
                                        .unwrap_or(EventsAttachKind::Mpsc);
                                    rxs.push(attach_events_rx_any(
                                        &options_t,
                                        rx_lane.as_str(),
                                        rx_attach,
                                    )?);
                                }

                                run_merge_role_loop(
                                    stop_t.as_ref(),
                                    activity_t.as_ref(),
                                    role_activity_t.as_ref(),
                                    &mut rxs,
                                    &mut tx,
                                    cfg.policy,
                                    on_full,
                                )
                            }
                            Some(LaneKindV1::Journal) => {
                                let mut tx = attach_journal_tx_one(&options_t, cfg.tx.as_str())?;

                                let mut rxs: Vec<ShmJournalSubscriber> =
                                    Vec::with_capacity(cfg.rx.len());
                                for rx_lane in &cfg.rx {
                                    let Some(LaneKindV1::Journal) =
                                        lane_kinds_t.get(rx_lane.as_str()).copied()
                                    else {
                                        return Err(ExecError::new(
                                            "unsupported merge rx lane kind",
                                        ));
                                    };
                                    rxs.push(attach_journal_rx_one(&options_t, rx_lane.as_str())?);
                                }

                                run_merge_role_loop(
                                    stop_t.as_ref(),
                                    activity_t.as_ref(),
                                    role_activity_t.as_ref(),
                                    &mut rxs,
                                    &mut tx,
                                    cfg.policy,
                                    on_full,
                                )
                            }
                            _ => Err(ExecError::new("unsupported merge tx lane kind")),
                        }
                    })
                    .map_err(|e| ExecError::with_source("failed to spawn role thread", e))?,
                RoleCfgV1::Router(cfg) => std::thread::Builder::new()
                    .name(format!("byteor-sp:{}", role_name))
                    .spawn(move || {
                        if let Some(h) = &thread_init_t {
                            let ctx = ThreadInitContext::new(role_name.clone(), RoleKind::Router);
                            h(&ctx)?;
                        }

                        let Some(LaneKindV1::FanoutBroadcast { consumers }) =
                            lane_kinds_t.get(cfg.rx.as_str()).copied()
                        else {
                            return Err(ExecError::new("unsupported router rx lane kind"));
                        };
                        let consumers = consumers as usize;
                        if consumers > 4 {
                            return Err(ExecError::new(
                                "fanout consumers > 4 is not supported by shm backing",
                            ));
                        }
                        if consumers != cfg.tx.len() {
                            return Err(ExecError::new("router tx lane count mismatch"));
                        }

                        let router = attach_fanout_router_one(&options_t, cfg.rx.as_str())?;

                        let mut fanout_rxs: Vec<ShmFanoutRx> = Vec::with_capacity(consumers);
                        for idx in 0..consumers {
                            fanout_rxs.push(attach_fanout_rx_one(
                                &options_t,
                                cfg.rx.as_str(),
                                idx,
                            )?);
                        }

                        let mut txs: Vec<AnyEventsTx> = Vec::with_capacity(consumers);
                        for tx_lane in &cfg.tx {
                            let Some(LaneKindV1::Events) =
                                lane_kinds_t.get(tx_lane.as_str()).copied()
                            else {
                                return Err(ExecError::new("router tx lanes must be events"));
                            };
                            let tx_attach = events_kinds_t
                                .get(tx_lane.as_str())
                                .copied()
                                .unwrap_or(EventsAttachKind::Mpsc);
                            txs.push(attach_events_tx_any(
                                &options_t,
                                tx_lane.as_str(),
                                tx_attach,
                            )?);
                        }

                        run_router_role_loop(
                            stop_t.as_ref(),
                            activity_t.as_ref(),
                            role_activity_t.as_ref(),
                            &router,
                            &mut fanout_rxs,
                            &mut txs,
                            on_full,
                        )
                    })
                    .map_err(|e| ExecError::with_source("failed to spawn role thread", e))?,
            };

            joins.push(join);
        }

        Ok(LaneGraphSpRuntime {
            stop,
            activity,
            joins,
        })
    }

    fn role_kind(role: &RoleCfgV1) -> RoleKind {
        match role {
            RoleCfgV1::Stage(_) => RoleKind::Stage,
            RoleCfgV1::Bridge(_) => RoleKind::Bridge,
            RoleCfgV1::Router(_) => RoleKind::Router,
            RoleCfgV1::Merge(_) => RoleKind::Merge,
        }
    }

    /// Run exactly one LaneGraph role in the current process using SHM-backed lanes.
    ///
    /// This is the “MP role runner” primitive: external supervisors can run each role in its own
    /// process, while this function runs the selected role loop until `stop` is set.
    ///
    /// Current scope: Events + FanoutBroadcast + Journal + SequencedSlots.
    pub fn run_lane_graph_mp_role_shm(
        spec: &byteor_pipeline_spec::PipelineSpecV1,
        resolver: &dyn StageResolver,
        options: SpShmOptions,
        role_name: &str,
        stop: &AtomicBool,
    ) -> Result<(), ExecError> {
        options.ensure_dir()?;

        crate::validate_sp(spec)?;

        let byteor_pipeline_spec::PipelineSpecV1::LaneGraph(lg) = spec else {
            return Err(ExecError::new("spec is not a lane graph"));
        };

        let role = lg
            .roles
            .iter()
            .find(|r| r.name() == role_name)
            .ok_or_else(|| ExecError::with_context("unknown role", role_name.to_string()))?;

        let lane_kinds: BTreeMap<String, LaneKindV1> =
            lg.lanes.iter().map(|l| (l.name.clone(), l.kind)).collect();
        let events_kinds = events_attach_kind_by_lane(lg);
        let on_full = lg.on_full;
        let role_activity = RuntimeRoleActivity::new(role.name().to_string(), role_kind(role));
        let activity = RuntimeActivity::new(Vec::new());

        match role {
            RoleCfgV1::Stage(cfg) => {
                let stage = resolve_or_err(resolver, cfg.stage.as_str())?;
                match (
                    lane_kinds.get(cfg.rx.as_str()).copied(),
                    lane_kinds.get(cfg.tx.as_str()).copied(),
                ) {
                    (Some(LaneKindV1::Events), Some(LaneKindV1::Events)) => {
                        let rx_kind = events_kinds
                            .get(cfg.rx.as_str())
                            .copied()
                            .unwrap_or(EventsAttachKind::Mpsc);
                        let tx_kind = events_kinds
                            .get(cfg.tx.as_str())
                            .copied()
                            .unwrap_or(EventsAttachKind::Mpsc);

                        let mut rx = attach_events_rx_any(&options, cfg.rx.as_str(), rx_kind)?;
                        let mut tx = attach_events_tx_any(&options, cfg.tx.as_str(), tx_kind)?;
                        run_stage_role_loop(
                            stop,
                            &activity,
                            &role_activity,
                            &mut rx,
                            &mut tx,
                            stage,
                            on_full,
                        )
                    }
                    (Some(LaneKindV1::Journal), Some(LaneKindV1::Journal)) => {
                        let mut rx = attach_journal_rx_one(&options, cfg.rx.as_str())?;
                        let mut tx = attach_journal_tx_one(&options, cfg.tx.as_str())?;
                        run_stage_role_loop_copy(
                            stop,
                            &activity,
                            &role_activity,
                            &mut rx,
                            &mut tx,
                            stage,
                            on_full,
                        )
                    }
                    (
                        Some(LaneKindV1::SequencedSlots { .. }),
                        Some(LaneKindV1::SequencedSlots { .. }),
                    ) => {
                        let mut rx = attach_sequenced_slots_rx_one(&options, cfg.rx.as_str())?;
                        let mut tx = attach_sequenced_slots_tx_one(&options, cfg.tx.as_str())?;
                        run_stage_role_loop_sequenced_slots(
                            stop,
                            &activity,
                            &role_activity,
                            &mut rx,
                            &mut tx,
                            stage,
                            on_full,
                        )
                    }
                    _ => Err(ExecError::new("unsupported stage rx lane kind")),
                }
            }
            RoleCfgV1::Bridge(cfg) => {
                let Some(rx_kind) = lane_kinds.get(cfg.rx.as_str()).copied() else {
                    return Err(ExecError::new("unknown bridge rx lane"));
                };
                let Some(tx_kind) = lane_kinds.get(cfg.tx.as_str()).copied() else {
                    return Err(ExecError::new("unknown bridge tx lane"));
                };

                match (rx_kind, tx_kind) {
                    (LaneKindV1::Events, LaneKindV1::Events) => {
                        let rx_attach = events_kinds
                            .get(cfg.rx.as_str())
                            .copied()
                            .unwrap_or(EventsAttachKind::Mpsc);
                        let tx_attach = events_kinds
                            .get(cfg.tx.as_str())
                            .copied()
                            .unwrap_or(EventsAttachKind::Mpsc);
                        let mut rx = attach_events_rx_any(&options, cfg.rx.as_str(), rx_attach)?;
                        let mut tx = attach_events_tx_any(&options, cfg.tx.as_str(), tx_attach)?;
                        run_bridge_role_loop(
                            stop,
                            &activity,
                            &role_activity,
                            &mut rx,
                            &mut tx,
                            on_full,
                        )
                    }
                    (LaneKindV1::Events, LaneKindV1::FanoutBroadcast { .. }) => {
                        let rx_attach = events_kinds
                            .get(cfg.rx.as_str())
                            .copied()
                            .unwrap_or(EventsAttachKind::Mpsc);
                        let mut rx = attach_events_rx_any(&options, cfg.rx.as_str(), rx_attach)?;
                        let mut tx = attach_fanout_tx_one(&options, cfg.tx.as_str())?;
                        run_bridge_role_loop(
                            stop,
                            &activity,
                            &role_activity,
                            &mut rx,
                            &mut tx,
                            on_full,
                        )
                    }
                    (LaneKindV1::Journal, LaneKindV1::Journal) => {
                        let mut rx = attach_journal_rx_one(&options, cfg.rx.as_str())?;
                        let mut tx = attach_journal_tx_one(&options, cfg.tx.as_str())?;
                        run_bridge_role_loop_copy(
                            stop,
                            &activity,
                            &role_activity,
                            &mut rx,
                            &mut tx,
                            on_full,
                        )
                    }
                    (LaneKindV1::SequencedSlots { .. }, LaneKindV1::SequencedSlots { .. }) => {
                        let mut rx = attach_sequenced_slots_rx_one(&options, cfg.rx.as_str())?;
                        let mut tx = attach_sequenced_slots_tx_one(&options, cfg.tx.as_str())?;
                        run_bridge_role_loop_sequenced_slots(
                            stop,
                            &activity,
                            &role_activity,
                            &mut rx,
                            &mut tx,
                            on_full,
                        )
                    }
                    _ => Err(ExecError::new("unsupported bridge lane kinds")),
                }
            }
            RoleCfgV1::Merge(cfg) => match lane_kinds.get(cfg.tx.as_str()).copied() {
                Some(LaneKindV1::Events) => {
                    let tx_attach = events_kinds
                        .get(cfg.tx.as_str())
                        .copied()
                        .unwrap_or(EventsAttachKind::Mpsc);
                    let mut tx = attach_events_tx_any(&options, cfg.tx.as_str(), tx_attach)?;

                    let mut rxs: Vec<AnyEventsRx> = Vec::with_capacity(cfg.rx.len());
                    for rx_lane in &cfg.rx {
                        let Some(LaneKindV1::Events) = lane_kinds.get(rx_lane.as_str()).copied()
                        else {
                            return Err(ExecError::new("unsupported merge rx lane kind"));
                        };
                        let rx_attach = events_kinds
                            .get(rx_lane.as_str())
                            .copied()
                            .unwrap_or(EventsAttachKind::Mpsc);
                        rxs.push(attach_events_rx_any(&options, rx_lane.as_str(), rx_attach)?);
                    }

                    run_merge_role_loop(
                        stop,
                        &activity,
                        &role_activity,
                        &mut rxs,
                        &mut tx,
                        cfg.policy,
                        on_full,
                    )
                }
                Some(LaneKindV1::Journal) => {
                    let mut tx = attach_journal_tx_one(&options, cfg.tx.as_str())?;
                    let mut rxs: Vec<ShmJournalSubscriber> = Vec::with_capacity(cfg.rx.len());
                    for rx_lane in &cfg.rx {
                        let Some(LaneKindV1::Journal) = lane_kinds.get(rx_lane.as_str()).copied()
                        else {
                            return Err(ExecError::new("unsupported merge rx lane kind"));
                        };
                        rxs.push(attach_journal_rx_one(&options, rx_lane.as_str())?);
                    }

                    run_merge_role_loop(
                        stop,
                        &activity,
                        &role_activity,
                        &mut rxs,
                        &mut tx,
                        cfg.policy,
                        on_full,
                    )
                }
                _ => Err(ExecError::new("unsupported merge tx lane kind")),
            },
            RoleCfgV1::Router(cfg) => {
                let Some(LaneKindV1::FanoutBroadcast { consumers }) =
                    lane_kinds.get(cfg.rx.as_str()).copied()
                else {
                    return Err(ExecError::new("unsupported router rx lane kind"));
                };
                let consumers = consumers as usize;
                if consumers > 4 {
                    return Err(ExecError::new(
                        "fanout consumers > 4 is not supported by shm backing",
                    ));
                }
                if consumers != cfg.tx.len() {
                    return Err(ExecError::new("router tx lane count mismatch"));
                }

                let router = attach_fanout_router_one(&options, cfg.rx.as_str())?;

                let mut fanout_rxs: Vec<ShmFanoutRx> = Vec::with_capacity(consumers);
                for idx in 0..consumers {
                    fanout_rxs.push(attach_fanout_rx_one(&options, cfg.rx.as_str(), idx)?);
                }

                let mut txs: Vec<AnyEventsTx> = Vec::with_capacity(consumers);
                for tx_lane in &cfg.tx {
                    let Some(LaneKindV1::Events) = lane_kinds.get(tx_lane.as_str()).copied() else {
                        return Err(ExecError::new("router tx lanes must be events"));
                    };
                    let tx_attach = events_kinds
                        .get(tx_lane.as_str())
                        .copied()
                        .unwrap_or(EventsAttachKind::Mpsc);
                    txs.push(attach_events_tx_any(&options, tx_lane.as_str(), tx_attach)?);
                }

                run_router_role_loop(
                    stop,
                    &activity,
                    &role_activity,
                    &router,
                    &mut fanout_rxs,
                    &mut txs,
                    on_full,
                )
            }
        }
    }
}

#[cfg(feature = "shm")]
pub use shm::{
    run_lane_graph_mp_role_shm, run_sp, spawn_lane_graph_sp_shm_runtime,
    spawn_lane_graph_sp_shm_runtime_with_thread_init, LaneGraphRoleActivitySnapshot,
    LaneGraphSpRuntime, SpShmOptions,
};