byteor_pipeline_lower/

lane_graph.rs

#[cfg(feature = "alloc")]
use alloc::format;
#[cfg(feature = "alloc")]
use alloc::string::{String, ToString};
#[cfg(feature = "alloc")]
use alloc::vec;
#[cfg(feature = "alloc")]
use alloc::vec::Vec;

#[cfg(feature = "alloc")]
use byteor_pipeline_plan::{
    ExecutionTargetV1, LaneGraphAuthoringV2, LaneIntent, PlanBoundary, PlanBoundaryKind,
    PlanConnectionDestination, PlanConnectionSource, PlanStage,
    PipelinePlanSourceV2, PipelinePlanV2,
};
use byteor_pipeline_spec::{
    BridgeRoleCfgV1, EndpointCfgV1, EndpointKindV1, LaneCfgV1, LaneGraphV1, LaneKindV1,
    MergePolicyV1, MergeRoleCfgV1, PipelineSpecV1, RoleCfgV1, RouterRoleCfgV1,
    StageRoleCfgV1,
};

use crate::common::{
    collect_stage_names, edge_required_kind, has_egress_boundary, has_ingress_boundary,
    inlet_key, intent_to_kind, outlet_key, reclassify_lower_error_v2, reject_v2, sanitize_id,
    validate_for_lowering_v2,
};
use crate::{
    LowerDiagnosticCategoryV2, LowerDiagnosticV2, LowerError, LowerOptionsV1, LowerReportV2,
    Result,
};

/// Lower a full-parity authoring document into `PipelineSpecV1::LaneGraph`.
#[cfg(feature = "alloc")]
pub fn lower_lane_graph_v2(
    plan: &PipelinePlanV2,
    opts: &LowerOptionsV1,
) -> Result<LowerReportV2> {
    if plan.target != ExecutionTargetV1::LaneGraph {
        return Err(reject_v2(
            ExecutionTargetV1::LaneGraph,
            LowerDiagnosticCategoryV2::TargetMismatch,
            "explicit plan target does not match requested lowering target",
        ));
    }

    let PipelinePlanSourceV2::LaneGraph(source) = &plan.source else {
        return Err(reject_v2(
            ExecutionTargetV1::LaneGraph,
            LowerDiagnosticCategoryV2::SourcePayloadMismatch,
            "lane_graph lowering requires a LaneGraph source payload",
        ));
    };

    let plan_diagnostics = validate_for_lowering_v2(plan, ExecutionTargetV1::LaneGraph)?;
    let mut lower_diagnostics = Vec::new();
    let mut effective_opts = opts.clone();
    effective_opts.on_full = source.on_full;
    let spec = lower_lane_graph_inner(
        &plan.name,
        &plan.stages,
        &plan.boundaries,
        source,
        &effective_opts,
        &mut lower_diagnostics,
    )
        .map_err(|err| reclassify_lower_error_v2(ExecutionTargetV1::LaneGraph, err))?;
    byteor_pipeline_spec::validate_v1(&spec)
        .map_err(|_| {
            reject_v2(
                ExecutionTargetV1::LaneGraph,
                LowerDiagnosticCategoryV2::SpecValidation,
                "lane_graph lowering produced an invalid v1 spec",
            )
        })?;

    Ok(LowerReportV2 {
        spec,
        plan_diagnostics,
        lower_diagnostics,
    })
}

#[cfg(feature = "alloc")]
fn lower_lane_graph_inner(
    plan_name: &str,
    stages: &[PlanStage],
    boundaries: &[PlanBoundary],
    source: &LaneGraphAuthoringV2,
    opts: &LowerOptionsV1,
    diagnostics: &mut Vec<LowerDiagnosticV2>,
) -> Result<PipelineSpecV1> {
    if plan_name.trim().is_empty() {
        return Err(LowerError::Invalid("plan.name must be non-empty"));
    }
    if stages.is_empty() {
        return Err(LowerError::Invalid("plan.stages must be non-empty"));
    }

    let stage_names = collect_stage_names(stages)?;

    {
        let mut seen: Vec<String> = Vec::new();
        for boundary in boundaries {
            if boundary.name.trim().is_empty() {
                return Err(LowerError::Invalid("boundary name must be non-empty"));
            }
            let key = format!("{:?}:{}", boundary.kind, boundary.name);
            if seen.contains(&key) {
                return Err(LowerError::Invalid("boundaries must be unique per kind"));
            }
            seen.push(key);
        }
    }

    let boundary_kind = intent_to_kind(source.boundary_lane)?;

    let mut incoming_by_inlet: Vec<(String, Vec<usize>)> = Vec::new();
    for (index, edge) in source.connections.iter().enumerate() {
        match &edge.from {
            PlanConnectionSource::StageOut { stage } => {
                if !stage_names.contains(&stage.as_str()) {
                    return Err(LowerError::Invalid("connection source stage does not exist"));
                }
            }
            PlanConnectionSource::BoundaryIngress { boundary } => {
                if !has_ingress_boundary(boundaries, boundary) {
                    return Err(LowerError::Invalid(
                        "connection source ingress boundary does not exist",
                    ));
                }
            }
            _ => {
                return Err(LowerError::Unsupported(
                    "connection source variant is not supported in v1 lowering",
                ));
            }
        }

        match &edge.to {
            PlanConnectionDestination::StageIn { stage } => {
                if !stage_names.contains(&stage.as_str()) {
                    return Err(LowerError::Invalid(
                        "connection destination stage does not exist",
                    ));
                }
            }
            PlanConnectionDestination::BoundaryEgress { boundary } => {
                if !has_egress_boundary(boundaries, boundary) {
                    return Err(LowerError::Invalid(
                        "connection destination egress boundary does not exist",
                    ));
                }
            }
            _ => {
                return Err(LowerError::Unsupported(
                    "connection destination variant is not supported in v1 lowering",
                ));
            }
        }

        if matches!(edge.intent, LaneIntent::FanoutBroadcast { consumers: 0 }) {
            return Err(LowerError::Invalid("fanout consumers must be > 0"));
        }

        let key = inlet_key(&edge.to);
        match incoming_by_inlet.iter_mut().find(|(current, _)| current == &key) {
            Some((_, entries)) => entries.push(index),
            None => incoming_by_inlet.push((key, vec![index])),
        }
    }

    let mut stage_in_kind: Vec<(String, LaneKindV1)> = Vec::new();
    for stage_name in &stage_names {
        let key = format!("stage:{stage_name}");
        let edge_indices = incoming_by_inlet
            .iter()
            .find(|(current, _)| current == &key)
            .map(|(_, entries)| entries.as_slice())
            .unwrap_or(&[]);
        if edge_indices.is_empty() {
            return Err(LowerError::Invalid(
                "stage nodes must have at least one incoming edge",
            ));
        }
        let mut kind: Option<LaneKindV1> = None;
        for &edge_index in edge_indices {
            let resolved = edge_required_kind(&source.connections[edge_index], boundary_kind)?;
            kind = match kind {
                None => Some(resolved),
                Some(prev) if prev == resolved => Some(prev),
                Some(_) => {
                    return Err(LowerError::Invalid(
                        "all incoming edges into a stage must resolve to the same lane kind",
                    ));
                }
            };
        }
        stage_in_kind.push(((*stage_name).to_string(), kind.unwrap()));
    }

    let mut egress_in_kind: Vec<(String, LaneKindV1)> = Vec::new();
    for boundary in boundaries
        .iter()
        .filter(|boundary| boundary.kind == PlanBoundaryKind::Egress)
    {
        let key = format!("egress:{}", boundary.name);
        let edge_indices = incoming_by_inlet
            .iter()
            .find(|(current, _)| current == &key)
            .map(|(_, entries)| entries.as_slice())
            .unwrap_or(&[]);
        if edge_indices.is_empty() {
            return Err(LowerError::Invalid(
                "egress boundaries must have at least one incoming edge",
            ));
        }
        let mut kind: Option<LaneKindV1> = None;
        for &edge_index in edge_indices {
            let resolved = edge_required_kind(&source.connections[edge_index], boundary_kind)?;
            kind = match kind {
                None => Some(resolved),
                Some(prev) if prev == resolved => Some(prev),
                Some(_) => {
                    return Err(LowerError::Invalid(
                        "all incoming edges into an egress boundary must resolve to the same lane kind",
                    ));
                }
            };
        }
        egress_in_kind.push((boundary.name.clone(), kind.unwrap()));
    }

    let mut lanes = Vec::new();
    let mut roles = Vec::new();

    let mut boundary_lane_ingress: Vec<(String, String)> = Vec::new();
    for boundary in boundaries
        .iter()
        .filter(|boundary| boundary.kind == PlanBoundaryKind::Ingress)
    {
        let lane_name = format!("lane_boundary_{}_out", sanitize_id(&boundary.name));
        lanes.push(LaneCfgV1 {
            name: lane_name.clone(),
            kind: boundary_kind,
        });
        boundary_lane_ingress.push((boundary.name.clone(), lane_name));
    }

    let mut stage_lane_out: Vec<(String, String)> = Vec::new();
    for (stage, kind) in &stage_in_kind {
        let lane_name = format!("lane_node_{}_out", sanitize_id(stage));
        lanes.push(LaneCfgV1 {
            name: lane_name.clone(),
            kind: *kind,
        });
        stage_lane_out.push((stage.clone(), lane_name));
    }

    let mut outgoing_by_outlet: Vec<(String, Vec<usize>)> = Vec::new();
    for (index, edge) in source.connections.iter().enumerate() {
        let key = outlet_key(&edge.from);
        match outgoing_by_outlet.iter_mut().find(|(current, _)| current == &key) {
            Some((_, entries)) => entries.push(index),
            None => outgoing_by_outlet.push((key, vec![index])),
        }
    }

    let mut outlet_base_lane: Vec<(String, (String, LaneKindV1))> = Vec::new();
    for (key, _) in &outgoing_by_outlet {
        if let Some(stage) = key.strip_prefix("stage:") {
            let lane = stage_lane_out
                .iter()
                .find(|(name, _)| name == stage)
                .map(|(_, lane)| lane.clone())
                .ok_or(LowerError::Invalid("stage output lane missing"))?;
            let kind = stage_in_kind
                .iter()
                .find(|(name, _)| name == stage)
                .map(|(_, kind)| *kind)
                .ok_or(LowerError::Invalid("stage input kind missing"))?;
            outlet_base_lane.push((key.clone(), (lane, kind)));
        } else if let Some(boundary) = key.strip_prefix("ingress:") {
            let lane = boundary_lane_ingress
                .iter()
                .find(|(name, _)| name == boundary)
                .map(|(_, lane)| lane.clone())
                .ok_or(LowerError::Invalid("boundary output lane missing"))?;
            outlet_base_lane.push((key.clone(), (lane, boundary_kind)));
        }
    }

    let mut delivered_lane_by_edge =
        vec![(String::new(), LaneKindV1::Events); source.connections.len()];

    for (out_key, edge_indices) in &outgoing_by_outlet {
        if edge_indices.len() <= 1 {
            continue;
        }

        let mut consumers: Option<u32> = None;
        for &edge_index in edge_indices {
            match source.connections[edge_index].intent {
                LaneIntent::FanoutBroadcast { consumers: current } => {
                    consumers = match consumers {
                        None => Some(current),
                        Some(prev) if prev == current => Some(prev),
                        Some(_) => {
                            return Err(LowerError::Invalid(
                                "fanout edges from a single outlet must use the same consumers value",
                            ));
                        }
                    };
                }
                _ => {
                    return Err(LowerError::Invalid(
                        "out-degree > 1 requires fanout intent on all outgoing edges",
                    ));
                }
            }
        }

        let consumers = consumers.unwrap_or(0);
        if consumers as usize != edge_indices.len() {
            return Err(LowerError::Invalid(
                "fanout consumers must match the number of outgoing edges",
            ));
        }

        let (src_lane, src_kind) = outlet_base_lane
            .iter()
            .find(|(key, _)| key == out_key)
            .map(|(_, value)| value.clone())
            .ok_or(LowerError::Invalid("fanout outlet base lane missing"))?;
        if src_kind != LaneKindV1::Events {
            return Err(LowerError::Invalid(
                "fanout outlets must produce Events lanes in v1",
            ));
        }

        let fanout_lane = format!("lane_fanout_{}", sanitize_id(out_key));
        lanes.push(LaneCfgV1 {
            name: fanout_lane.clone(),
            kind: LaneKindV1::FanoutBroadcast { consumers },
        });

        let bridge_role = format!("role_bridge_fanout_{}", sanitize_id(out_key));
            diagnostics.push(LowerDiagnosticV2::InsertedBridge {
            role: bridge_role.clone(),
            rx: src_lane.clone(),
            tx: fanout_lane.clone(),
            from: src_kind,
            to: LaneKindV1::FanoutBroadcast { consumers },
        });
        roles.push(RoleCfgV1::Bridge(BridgeRoleCfgV1 {
            name: bridge_role,
            rx: src_lane,
            tx: fanout_lane.clone(),
        }));

        let router_role = format!("role_router_{}", sanitize_id(out_key));
        let mut tx_lanes = Vec::with_capacity(consumers as usize);
        for consumer_index in 0..consumers {
            let lane_name = format!("lane_fanout_{}_c{consumer_index}", sanitize_id(out_key));
            lanes.push(LaneCfgV1 {
                name: lane_name.clone(),
                kind: LaneKindV1::Events,
            });
            tx_lanes.push(lane_name);
        }

        diagnostics.push(LowerDiagnosticV2::InsertedRouter {
            role: router_role.clone(),
            rx: fanout_lane.clone(),
            tx: tx_lanes.clone(),
            consumers,
        });
        roles.push(RoleCfgV1::Router(RouterRoleCfgV1 {
            name: router_role,
            rx: fanout_lane,
            tx: tx_lanes.clone(),
        }));

        for (index, &edge_index) in edge_indices.iter().enumerate() {
            delivered_lane_by_edge[edge_index] = (tx_lanes[index].clone(), LaneKindV1::Events);
        }
    }

    for (out_key, edge_indices) in &outgoing_by_outlet {
        if edge_indices.len() > 1 || edge_indices.is_empty() {
            continue;
        }

        let edge_index = edge_indices[0];
        if matches!(source.connections[edge_index].intent, LaneIntent::FanoutBroadcast { .. }) {
            return Err(LowerError::Invalid(
                "fanout intent requires out-degree greater than one",
            ));
        }

        let dest_kind = edge_required_kind(&source.connections[edge_index], boundary_kind)?;
        let (src_lane, src_kind) = outlet_base_lane
            .iter()
            .find(|(key, _)| key == out_key)
            .map(|(_, value)| value.clone())
            .ok_or(LowerError::Invalid("outlet base lane missing"))?;
        if src_kind == dest_kind {
            delivered_lane_by_edge[edge_index] = (src_lane, dest_kind);
        } else {
            let edge_lane = format!("lane_edge_{edge_index}");
            lanes.push(LaneCfgV1 {
                name: edge_lane.clone(),
                kind: dest_kind,
            });
            let bridge_role = format!("role_bridge_{edge_index}");
            diagnostics.push(LowerDiagnosticV2::InsertedBridge {
                role: bridge_role.clone(),
                rx: src_lane.clone(),
                tx: edge_lane.clone(),
                from: src_kind,
                to: dest_kind,
            });
            roles.push(RoleCfgV1::Bridge(BridgeRoleCfgV1 {
                name: bridge_role,
                rx: src_lane,
                tx: edge_lane.clone(),
            }));
            delivered_lane_by_edge[edge_index] = (edge_lane, dest_kind);
        }
    }

    for (lane, _) in &delivered_lane_by_edge {
        if lane.is_empty() {
            return Err(LowerError::Invalid(
                "internal error: edge did not get an assigned delivered lane",
            ));
        }
    }

    let mut delivered_by_inlet: Vec<(String, Vec<(String, LaneKindV1)>)> = Vec::new();
    for (index, edge) in source.connections.iter().enumerate() {
        let key = inlet_key(&edge.to);
        let delivered = delivered_lane_by_edge[index].clone();
        match delivered_by_inlet.iter_mut().find(|(current, _)| current == &key) {
            Some((_, entries)) => entries.push(delivered),
            None => delivered_by_inlet.push((key, vec![delivered])),
        }
    }

    let mut stage_lane_in: Vec<(String, String)> = Vec::new();
    for (stage, kind) in &stage_in_kind {
        let key = format!("stage:{stage}");
        let incoming = delivered_by_inlet
            .iter()
            .find(|(current, _)| current == &key)
            .map(|(_, entries)| entries.as_slice())
            .unwrap_or(&[]);
        let rx_lane = if incoming.len() == 1 {
            incoming[0].0.clone()
        } else {
            let tx_lane = format!("lane_merge_node_{}", sanitize_id(stage));
            lanes.push(LaneCfgV1 {
                name: tx_lane.clone(),
                kind: *kind,
            });
            let rx_lanes: Vec<String> = incoming.iter().map(|(lane, _)| lane.clone()).collect();
            let role = format!("role_merge_node_{}", sanitize_id(stage));
            diagnostics.push(LowerDiagnosticV2::InsertedMerge {
                role: role.clone(),
                rx: rx_lanes.clone(),
                tx: tx_lane.clone(),
                policy: MergePolicyV1::RoundRobin,
            });
            roles.push(RoleCfgV1::Merge(MergeRoleCfgV1 {
                name: role,
                rx: rx_lanes,
                tx: tx_lane.clone(),
                policy: MergePolicyV1::RoundRobin,
            }));
            tx_lane
        };
        stage_lane_in.push((stage.clone(), rx_lane));
    }

    let mut boundary_lane_egress: Vec<(String, String)> = Vec::new();
    for (boundary, kind) in &egress_in_kind {
        let key = format!("egress:{boundary}");
        let incoming = delivered_by_inlet
            .iter()
            .find(|(current, _)| current == &key)
            .map(|(_, entries)| entries.as_slice())
            .unwrap_or(&[]);
        let lane = if incoming.len() == 1 {
            incoming[0].0.clone()
        } else {
            let tx_lane = format!("lane_merge_boundary_{}", sanitize_id(boundary));
            lanes.push(LaneCfgV1 {
                name: tx_lane.clone(),
                kind: *kind,
            });
            let rx_lanes: Vec<String> = incoming.iter().map(|(lane, _)| lane.clone()).collect();
            let role = format!("role_merge_boundary_{}", sanitize_id(boundary));
            diagnostics.push(LowerDiagnosticV2::InsertedMerge {
                role: role.clone(),
                rx: rx_lanes.clone(),
                tx: tx_lane.clone(),
                policy: MergePolicyV1::RoundRobin,
            });
            roles.push(RoleCfgV1::Merge(MergeRoleCfgV1 {
                name: role,
                rx: rx_lanes,
                tx: tx_lane.clone(),
                policy: MergePolicyV1::RoundRobin,
            }));
            tx_lane
        };
        boundary_lane_egress.push((boundary.clone(), lane));
    }

    for stage in stages {
        let rx = stage_lane_in
            .iter()
            .find(|(name, _)| name == &stage.name)
            .map(|(_, lane)| lane.clone())
            .ok_or(LowerError::Invalid("stage rx lane missing"))?;
        let tx = stage_lane_out
            .iter()
            .find(|(name, _)| name == &stage.name)
            .map(|(_, lane)| lane.clone())
            .ok_or(LowerError::Invalid("stage tx lane missing"))?;
        roles.push(RoleCfgV1::Stage(StageRoleCfgV1 {
            name: format!("role_stage_{}", sanitize_id(&stage.name)),
            stage: stage.stage.clone(),
            rx,
            tx,
        }));
    }

    let mut endpoints = Vec::new();
    for boundary in boundaries {
        match boundary.kind {
            PlanBoundaryKind::Ingress => {
                let lane = boundary_lane_ingress
                    .iter()
                    .find(|(name, _)| name == &boundary.name)
                    .map(|(_, lane)| lane.clone())
                    .ok_or(LowerError::Invalid("ingress boundary lane missing"))?;
                endpoints.push(EndpointCfgV1 {
                    name: boundary.name.clone(),
                    kind: EndpointKindV1::Ingress,
                    lane,
                });
            }
            PlanBoundaryKind::Egress => {
                let lane = boundary_lane_egress
                    .iter()
                    .find(|(name, _)| name == &boundary.name)
                    .map(|(_, lane)| lane.clone())
                    .ok_or(LowerError::Invalid("egress boundary lane missing"))?;
                endpoints.push(EndpointCfgV1 {
                    name: boundary.name.clone(),
                    kind: EndpointKindV1::Egress,
                    lane,
                });
            }
        }
    }

    Ok(PipelineSpecV1::LaneGraph(LaneGraphV1 {
        endpoints,
        lanes,
        roles,
        on_full: opts.on_full,
    }))
}