datum-core 0.10.5

//! Graph construction: the builder, the `GraphDsl` entry points, and the
//! immutable `GraphBlueprint` they produce.
//!
//! Building a graph is side-effect-free (blueprint vs. materialization): `add`
//! registers a stage's ports, `connect`/`wire` records an edge after
//! `validate_connection` checks port kind, element `TypeId`, and single-use, and
//! `finish` verifies the returned [`Shape`] matches the open ports. Execution
//! starts only when a `run_*` method (defined in the `executor` module) is
//! called on the resulting [`GraphBlueprint`].

use super::*;
use crate::{Attribute, Attributes};

type PartialGraphBuilder<S> = dyn Fn(&mut GraphBuilder) -> StreamResult<S> + Send + Sync;

#[derive(Clone, Debug)]
struct PortRecord {
    kind: PortKind,
    type_id: TypeId,
    type_name: &'static str,
    name: Arc<str>,
}

#[derive(Clone, Debug)]
pub(super) struct Edge {
    pub(super) outlet: PortId,
    pub(super) inlet: PortId,
}

#[derive(Clone)]
pub(super) struct StageRecord {
    pub(super) spec: StageSpec,
    pub(super) logic_factory: Option<Arc<dyn Fn() -> GraphStageLogic + Send + Sync>>,
}

impl std::fmt::Debug for StageRecord {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("StageRecord")
            .field("spec", &self.spec)
            .field("has_logic", &self.logic_factory.is_some())
            .finish()
    }
}

/// Mutable scratch state for assembling a dynamic graph inside a `GraphDsl`
/// closure.
///
/// Tracks registered stages, their ports, the edges wired between them, and any
/// deferred wiring errors (raised by `wire`, which collects rather than
/// short-circuits). `finish` consumes it into a validated [`GraphBlueprint`].
/// Use this builder for cycles, runtime-data-dependent topology, erased
/// interop, `connect_any`, and method-based `wire` construction. For static
/// typed graphs where double-wiring should be rejected by Rust moves, use
/// [`TypedGraphBuilder`](super::TypedGraphBuilder).
#[derive(Debug, Default)]
pub struct GraphBuilder {
    allocator: PortAllocator,
    ports: HashMap<PortId, PortRecord>,
    stages: Vec<StageRecord>,
    edges: Vec<Edge>,
    errors: Vec<StreamError>,
}

impl GraphBuilder {
    #[must_use]
    pub fn add<G: GraphStage>(&mut self, stage: G) -> G::Shape {
        self.add_with_attributes(stage, Attributes::default())
    }

    #[must_use]
    pub fn add_with_attributes<G: GraphStage>(
        &mut self,
        stage: G,
        attributes: Attributes,
    ) -> G::Shape {
        let shape = stage.allocate_shape(&mut self.allocator);
        let inlets = shape.inlets();
        let outlets = shape.outlets();
        self.ports.reserve(inlets.len() + outlets.len());

        for inlet in &inlets {
            self.register_inlet(inlet);
        }
        for outlet in &outlets {
            self.register_outlet(outlet);
        }
        let spec = stage
            .stage_spec_with_ports(&shape, inlets, outlets)
            .add_attributes(attributes);
        let logic_factory = if matches!(spec.kind, StageKind::Opaque) {
            let shape_clone = shape.clone();
            Some(Arc::new(move || stage.create_logic(&shape_clone))
                as Arc<dyn Fn() -> GraphStageLogic + Send + Sync>)
        } else {
            None
        };
        self.stages.push(StageRecord {
            spec,
            logic_factory,
        });
        shape
    }

    #[must_use]
    pub fn add_named<G: GraphStage>(&mut self, stage: G, name: impl Into<String>) -> G::Shape {
        self.add_with_attributes(stage, Attributes::named(name))
    }

    pub fn connect<T: 'static>(&mut self, outlet: Outlet<T>, inlet: Inlet<T>) -> StreamResult<()> {
        self.connect_any(outlet.erase(), inlet.erase())
    }

    /// Connect type-erased ports for dynamic graph construction and interop.
    ///
    /// This remains an explicit compatibility surface for cases where the
    /// element type is only known at runtime. Prefer [`connect`](Self::connect)
    /// or [`TypedGraphBuilder`](super::TypedGraphBuilder) for static Rust
    /// graphs.
    pub fn connect_any(&mut self, outlet: AnyOutlet, inlet: AnyInlet) -> StreamResult<()> {
        match self.connect_any_unrecorded(outlet, inlet) {
            Ok(()) => Ok(()),
            Err(error) => self.record_error(error),
        }
    }

    pub fn import<S: Shape>(&mut self, graph: &PartialGraph<S>) -> StreamResult<S> {
        graph.build(self)
    }

    /// Apply a dynamic/method-based wiring spec, deferring errors to `finish`.
    ///
    /// `wire` is useful for graph DSL ergonomics and interop-style topology
    /// construction. Static Rust graphs can use [`TypedGraphBuilder`] when
    /// compile-time double-wire rejection is desired.
    pub fn wire<W: WireSpec>(&mut self, spec: W) -> &mut Self {
        if let Err(error) = spec.apply(self) {
            let _ = self.record_error(error);
        }
        self
    }

    /// Apply a dynamic/method-based wiring spec, returning wiring errors
    /// immediately.
    ///
    /// This is the fallible counterpart to [`wire`](Self::wire) and remains a
    /// runtime-validated dynamic/interop surface.
    pub fn try_wire<W: WireSpec>(&mut self, spec: W) -> StreamResult<&mut Self> {
        match spec.apply(self) {
            Ok(()) => Ok(self),
            Err(error) => {
                self.errors.push(error.clone());
                Err(error)
            }
        }
    }

    pub(super) fn connect_any_unrecorded(
        &mut self,
        outlet: AnyOutlet,
        inlet: AnyInlet,
    ) -> StreamResult<()> {
        self.validate_connection(&outlet, &inlet)?;
        self.edges.push(Edge {
            outlet: outlet.id(),
            inlet: inlet.id(),
        });
        Ok(())
    }

    pub(super) fn is_outlet_connected(&self, outlet: &AnyOutlet) -> bool {
        self.edges.iter().any(|edge| edge.outlet == outlet.id())
    }

    pub(super) fn is_inlet_connected(&self, inlet: &AnyInlet) -> bool {
        self.edges.iter().any(|edge| edge.inlet == inlet.id())
    }

    fn record_error(&mut self, error: StreamError) -> StreamResult<()> {
        self.errors.push(error.clone());
        Err(error)
    }

    fn register_inlet(&mut self, inlet: &AnyInlet) {
        self.ports.insert(
            inlet.id(),
            PortRecord {
                kind: PortKind::Inlet,
                type_id: inlet.type_id(),
                type_name: inlet.type_name(),
                name: Arc::clone(&inlet.name),
            },
        );
    }

    fn register_outlet(&mut self, outlet: &AnyOutlet) {
        self.ports.insert(
            outlet.id(),
            PortRecord {
                kind: PortKind::Outlet,
                type_id: outlet.type_id(),
                type_name: outlet.type_name(),
                name: Arc::clone(&outlet.name),
            },
        );
    }

    fn validate_connection(&self, outlet: &AnyOutlet, inlet: &AnyInlet) -> StreamResult<()> {
        let outlet_record = self.ports.get(&outlet.id()).ok_or_else(|| {
            StreamError::GraphValidation(format!("unknown outlet {}", outlet.name()))
        })?;
        let inlet_record = self.ports.get(&inlet.id()).ok_or_else(|| {
            StreamError::GraphValidation(format!("unknown inlet {}", inlet.name()))
        })?;

        if outlet_record.kind != PortKind::Outlet {
            return Err(StreamError::GraphValidation(format!(
                "{} is not an outlet",
                outlet_record.name
            )));
        }
        if inlet_record.kind != PortKind::Inlet {
            return Err(StreamError::GraphValidation(format!(
                "{} is not an inlet",
                inlet_record.name
            )));
        }
        if outlet_record.type_id != inlet_record.type_id {
            return Err(StreamError::GraphValidation(format!(
                "cannot connect outlet {} ({}) to inlet {} ({})",
                outlet_record.name,
                outlet_record.type_name,
                inlet_record.name,
                inlet_record.type_name
            )));
        }
        if self.edges.iter().any(|edge| edge.outlet == outlet.id()) {
            return Err(StreamError::GraphValidation(format!(
                "outlet {} is already connected",
                outlet_record.name
            )));
        }
        if self.edges.iter().any(|edge| edge.inlet == inlet.id()) {
            return Err(StreamError::GraphValidation(format!(
                "inlet {} is already connected",
                inlet_record.name
            )));
        }

        Ok(())
    }

    pub(super) fn finish<S: Shape>(self, shape: S) -> StreamResult<GraphBlueprint<S>> {
        let mut errors = self.errors;
        let connected_inlets: HashSet<PortId> = self.edges.iter().map(|edge| edge.inlet).collect();
        let connected_outlets: HashSet<PortId> =
            self.edges.iter().map(|edge| edge.outlet).collect();

        let open_inlets: HashSet<PortId> = self
            .ports
            .iter()
            .filter_map(|(id, port)| {
                (port.kind == PortKind::Inlet && !connected_inlets.contains(id)).then_some(*id)
            })
            .collect();
        let open_outlets: HashSet<PortId> = self
            .ports
            .iter()
            .filter_map(|(id, port)| {
                (port.kind == PortKind::Outlet && !connected_outlets.contains(id)).then_some(*id)
            })
            .collect();

        let result_inlets: HashSet<PortId> = shape.inlets().iter().map(AnyInlet::id).collect();
        let result_outlets: HashSet<PortId> = shape.outlets().iter().map(AnyOutlet::id).collect();

        for inlet in shape.inlets() {
            match self.ports.get(&inlet.id()) {
                Some(port)
                    if port.kind == PortKind::Inlet
                        && port.type_id == inlet.type_id()
                        && port.name.as_ref() == inlet.name() => {}
                Some(port) if port.kind == PortKind::Inlet => {
                    errors.push(StreamError::GraphValidation(format!(
                        "result shape inlet {} does not match registered inlet {} ({})",
                        inlet.name(),
                        port.name,
                        port.type_name
                    )));
                }
                Some(port) => errors.push(StreamError::GraphValidation(format!(
                    "result shape references non-inlet port {}",
                    port.name
                ))),
                None => errors.push(StreamError::GraphValidation(format!(
                    "result shape references unknown inlet {}",
                    inlet.name()
                ))),
            }
        }
        for outlet in shape.outlets() {
            match self.ports.get(&outlet.id()) {
                Some(port)
                    if port.kind == PortKind::Outlet
                        && port.type_id == outlet.type_id()
                        && port.name.as_ref() == outlet.name() => {}
                Some(port) if port.kind == PortKind::Outlet => {
                    errors.push(StreamError::GraphValidation(format!(
                        "result shape outlet {} does not match registered outlet {} ({})",
                        outlet.name(),
                        port.name,
                        port.type_name
                    )));
                }
                Some(port) => errors.push(StreamError::GraphValidation(format!(
                    "result shape references non-outlet port {}",
                    port.name
                ))),
                None => errors.push(StreamError::GraphValidation(format!(
                    "result shape references unknown outlet {}",
                    outlet.name()
                ))),
            }
        }

        if open_inlets != result_inlets {
            errors.push(StreamError::GraphValidation(format!(
                "result shape inlets do not match open inlets: open={:?}, result={:?}",
                describe_ports(&self.ports, &open_inlets),
                describe_ports(&self.ports, &result_inlets)
            )));
        }
        if open_outlets != result_outlets {
            errors.push(StreamError::GraphValidation(format!(
                "result shape outlets do not match open outlets: open={:?}, result={:?}",
                describe_ports(&self.ports, &open_outlets),
                describe_ports(&self.ports, &result_outlets)
            )));
        }

        if !errors.is_empty() {
            return Err(StreamError::GraphValidation(
                errors
                    .into_iter()
                    .map(|error| error.to_string())
                    .collect::<Vec<_>>()
                    .join("; "),
            ));
        }

        let segments = compute_segments(&self.stages);
        Ok(GraphBlueprint {
            shape,
            stages: self.stages,
            edges: self.edges,
            segments,
            attributes: Attributes::default(),
        })
    }
}

fn describe_ports(ports: &HashMap<PortId, PortRecord>, ids: &HashSet<PortId>) -> Vec<String> {
    let mut names = ids
        .iter()
        .map(|id| {
            ports
                .get(id)
                .map(|port| port.name.as_ref().to_owned())
                .unwrap_or_else(|| format!("unknown#{}", id.as_usize()))
        })
        .collect::<Vec<_>>();
    names.sort();
    names
}

fn compute_segments(stages: &[StageRecord]) -> Vec<FusedSegment> {
    let mut segments = Vec::with_capacity(1);
    let mut current = Vec::with_capacity(stages.len());

    for (index, stage) in stages.iter().enumerate() {
        if stage.spec.async_boundary && !current.is_empty() {
            segments.push(FusedSegment {
                stage_indices: std::mem::take(&mut current),
            });
        }
        current.push(index);
        if stage.spec.async_boundary {
            segments.push(FusedSegment {
                stage_indices: std::mem::take(&mut current),
            });
        }
    }

    if !current.is_empty() {
        segments.push(FusedSegment {
            stage_indices: current,
        });
    }

    segments
}

/// Entry points for building graphs. `create` takes a closure returning the
/// shape directly, `try_create` a closure returning `StreamResult<Shape>` (so
/// `?` works on `connect`/`try_wire`), and `partial` builds a reusable
/// [`PartialGraph`] fragment.
pub struct GraphDsl;

impl GraphDsl {
    pub fn create<S, F>(build: F) -> StreamResult<GraphBlueprint<S>>
    where
        S: Shape,
        F: FnOnce(&mut GraphBuilder) -> S,
    {
        let mut builder = GraphBuilder::default();
        let shape = build(&mut builder);
        builder.finish(shape)
    }

    pub fn try_create<S, F>(build: F) -> StreamResult<GraphBlueprint<S>>
    where
        S: Shape,
        F: FnOnce(&mut GraphBuilder) -> StreamResult<S>,
    {
        let mut builder = GraphBuilder::default();
        let shape = build(&mut builder)?;
        builder.finish(shape)
    }

    pub fn partial<S, F>(build: F) -> PartialGraph<S>
    where
        S: Shape,
        F: Fn(&mut GraphBuilder) -> StreamResult<S> + Send + Sync + 'static,
    {
        PartialGraph {
            build: Arc::new(build),
            attributes: Attributes::default(),
        }
    }
}

/// Anything exposing a graph [`Shape`] — implemented by [`GraphBlueprint`].
pub trait Graph {
    type Shape: Shape;

    fn shape(&self) -> Self::Shape;
}

/// A maximal run of stages with no async boundary between them, executed
/// together by the fused executor. Boundaries (`AsyncBoundary` stages) split a
/// graph into consecutive segments.
#[derive(Clone, Debug)]
pub struct FusedSegment {
    stage_indices: Vec<usize>,
}

impl FusedSegment {
    #[must_use]
    pub fn stage_indices(&self) -> &[usize] {
        &self.stage_indices
    }
}

/// An immutable, validated graph ready to run. Produced by `GraphDsl::create`/
/// `try_create`. Carries the external [`Shape`], the stages, the wired edges,
/// the precomputed fused segments, and graph-level [`Attributes`]. The `run_*`
/// methods are defined in the `executor` module; running one never mutates the
/// blueprint, so it can be reused and run concurrently.
pub struct GraphBlueprint<S: Shape> {
    pub(super) shape: S,
    pub(super) stages: Vec<StageRecord>,
    pub(super) edges: Vec<Edge>,
    pub(super) segments: Vec<FusedSegment>,
    pub(super) attributes: Attributes,
}

impl<S: Shape + Clone> Clone for GraphBlueprint<S> {
    fn clone(&self) -> Self {
        Self {
            shape: self.shape.clone(),
            stages: self.stages.clone(),
            edges: self.edges.clone(),
            segments: self.segments.clone(),
            attributes: self.attributes.clone(),
        }
    }
}

impl<S: Shape + fmt::Debug> fmt::Debug for GraphBlueprint<S> {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        f.debug_struct("GraphBlueprint")
            .field("shape", &self.shape)
            .field("stages", &self.stages)
            .field("edges", &self.edges)
            .field("segments", &self.segments)
            .field("attributes", &self.attributes)
            .finish()
    }
}

impl<S: Shape> GraphBlueprint<S> {
    #[must_use]
    pub fn shape(&self) -> S {
        self.shape.clone()
    }

    #[must_use]
    pub fn stage_count(&self) -> usize {
        self.stages.len()
    }

    #[must_use]
    pub fn edge_count(&self) -> usize {
        self.edges.len()
    }

    #[must_use]
    pub fn segments(&self) -> &[FusedSegment] {
        &self.segments
    }

    #[must_use]
    pub fn attributes(&self) -> &Attributes {
        &self.attributes
    }

    /// Effective post-merge attributes for every stage in execution order.
    ///
    /// Graph-level attributes are inherited by each stage and stage-level attributes override them
    /// per key, matching Akka's closest-wins mental model.
    #[must_use]
    pub fn effective_stage_attributes(&self) -> Vec<FusedNodeAttributes> {
        self.stage_attribute_reports()
    }

    #[must_use]
    pub fn with_attributes(mut self, attributes: Attributes) -> Self {
        self.attributes = attributes;
        self
    }

    #[must_use]
    pub fn add_attributes(mut self, attributes: Attributes) -> Self {
        self.attributes = self.attributes.and(attributes);
        self
    }

    #[must_use]
    pub fn named(self, name: impl Into<String>) -> Self {
        self.add_attributes(Attributes::named(name))
    }

    pub(super) fn effective_attributes_for_stage(&self, stage: &StageRecord) -> Attributes {
        self.attributes.clone().and(stage.spec.attributes().clone())
    }

    pub(super) fn stage_attribute_reports(&self) -> Vec<FusedNodeAttributes> {
        self.stages
            .iter()
            .enumerate()
            .map(|(index, stage)| {
                let attributes = self.effective_attributes_for_stage(stage);
                FusedNodeAttributes {
                    stage_index: index,
                    stage_name: stage.spec.name().to_owned(),
                    effective_name: attributes
                        .name()
                        .map(str::to_owned)
                        .unwrap_or_else(|| stage.spec.name().to_owned()),
                    attributes,
                }
            })
            .collect()
    }

    pub(super) fn plan_report(
        &self,
        selected_tier: FusedExecutorTier,
        tier_changes: Vec<FusedTierChange>,
    ) -> FusedPlanReport {
        FusedPlanReport {
            selected_tier,
            stage_attributes: self.stage_attribute_reports(),
            tier_changes,
        }
    }
}

impl<S: Shape> Graph for GraphBlueprint<S> {
    type Shape = S;

    fn shape(&self) -> Self::Shape {
        self.shape()
    }
}

/// A reusable graph fragment: a builder closure plus its [`Shape`], importable
/// into multiple parent graphs via [`GraphBuilder::import`]. Still a blueprint —
/// the closure runs (allocating fresh ports) each time it is imported.
/// Aliased as [`ImportedGraph`].
#[derive(Clone)]
pub struct PartialGraph<S: Shape> {
    build: Arc<PartialGraphBuilder<S>>,
    attributes: Attributes,
}

impl<S: Shape> PartialGraph<S> {
    pub fn build(&self, builder: &mut GraphBuilder) -> StreamResult<S> {
        (self.build)(builder)
    }

    #[must_use]
    pub fn attributes(&self) -> &Attributes {
        &self.attributes
    }

    #[must_use]
    pub fn with_attributes(mut self, attributes: Attributes) -> Self {
        self.attributes = attributes;
        self
    }

    #[must_use]
    pub fn add_attributes(mut self, attributes: Attributes) -> Self {
        self.attributes = self.attributes.and(attributes);
        self
    }

    #[must_use]
    pub fn named(self, name: impl Into<String>) -> Self {
        self.add_attributes(Attributes::named(name))
    }
}

impl<S: Shape> std::fmt::Debug for PartialGraph<S> {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("PartialGraph")
            .field("attributes", &self.attributes)
            .finish_non_exhaustive()
    }
}

pub type ImportedGraph<S> = PartialGraph<S>;

/// Execution bound for the fused executor. `event_limit` caps the number of
/// push/pull events a single run may take, so an unproductive cycle surfaces
/// [`StreamError::EventLimitExceeded`] instead of hanging. Defaults to 100M.
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub struct FusedExecutionConfig {
    pub event_limit: usize,
}

impl Default for FusedExecutionConfig {
    fn default() -> Self {
        Self {
            event_limit: 100_000_000,
        }
    }
}

/// Execution settings for the current graph async-boundary benchmark path.
///
/// This path validates a typed-linear graph and uses Ractor-backed async
/// regions with bounded handoff queues to measure real boundary crossing cost.
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub struct AsyncBoundaryExecutionConfig {
    pub fused: FusedExecutionConfig,
    pub buffer_size: usize,
}

impl Default for AsyncBoundaryExecutionConfig {
    fn default() -> Self {
        Self {
            fused: FusedExecutionConfig::default(),
            buffer_size: 16,
        }
    }
}

/// Result of a collecting run: the `output` vector plus instrumentation
/// (`events` processed, `async_boundary_crossings`) and the selected graph execution plan.
/// Returned by the `*_with_input_report` methods.
#[derive(Clone, Debug, PartialEq, Eq)]
pub struct FusedExecutionReport<T> {
    pub output: Vec<T>,
    pub events: usize,
    pub async_boundary_crossings: usize,
    pub plan: FusedPlanReport,
    pub node_metrics: Vec<FusedNodeMetrics>,
}

/// Result of a terminal (count/fold) run: the reduced `result` plus the same
/// instrumentation and plan data as [`FusedExecutionReport`]. Returned by the
/// `*_count_*_report` / `*_fold_*_report` methods.
#[derive(Clone, Debug, PartialEq, Eq)]
pub struct FusedTerminalReport<T> {
    pub result: T,
    pub events: usize,
    pub async_boundary_crossings: usize,
    pub plan: FusedPlanReport,
    pub node_metrics: Vec<FusedNodeMetrics>,
}

impl<T> FusedExecutionReport<T> {
    pub(super) fn new(
        output: Vec<T>,
        events: usize,
        async_boundary_crossings: usize,
        plan: FusedPlanReport,
    ) -> Self {
        Self {
            output,
            events,
            async_boundary_crossings,
            plan,
            node_metrics: Vec::new(),
        }
    }
}

impl<T> FusedTerminalReport<T> {
    pub(super) fn new(
        result: T,
        events: usize,
        async_boundary_crossings: usize,
        plan: FusedPlanReport,
    ) -> Self {
        Self {
            result,
            events,
            async_boundary_crossings,
            plan,
            node_metrics: Vec::new(),
        }
    }
}

impl<T> FusedExecutionReport<T> {
    pub(super) fn with_node_metrics(mut self, node_metrics: Vec<FusedNodeMetrics>) -> Self {
        self.node_metrics = node_metrics;
        self
    }

    /// Return an immutable point-in-time copy of this run's node metrics.
    #[must_use]
    pub fn metrics_snapshot(&self) -> Vec<FusedNodeMetrics> {
        self.node_metrics.clone()
    }
}

impl<T> FusedTerminalReport<T> {
    pub(super) fn with_node_metrics(mut self, node_metrics: Vec<FusedNodeMetrics>) -> Self {
        self.node_metrics = node_metrics;
        self
    }

    /// Return an immutable point-in-time copy of this run's node metrics.
    #[must_use]
    pub fn metrics_snapshot(&self) -> Vec<FusedNodeMetrics> {
        self.node_metrics.clone()
    }
}

/// Executor tier selected for a fused graph run.
#[derive(Clone, Copy, Debug, PartialEq, Eq, Hash)]
pub enum FusedExecutorTier {
    TypedLinear,
    TypedAcyclicJunction,
    TypedMergeSequence,
    TypedMergeLatest,
    TypedCyclicFeedback,
    AsyncBoundary,
    TypedFanIn,
    TypedConcat,
    TypedInterleave,
    TypedMergePreferred,
    Erased,
}

/// Plan metadata attached to fused execution reports.
#[derive(Clone, Debug, PartialEq, Eq)]
pub struct FusedPlanReport {
    pub selected_tier: FusedExecutorTier,
    pub stage_attributes: Vec<FusedNodeAttributes>,
    pub tier_changes: Vec<FusedTierChange>,
}

impl FusedPlanReport {
    pub(super) fn empty(selected_tier: FusedExecutorTier) -> Self {
        Self {
            selected_tier,
            stage_attributes: Vec::new(),
            tier_changes: Vec::new(),
        }
    }
}

/// Effective attributes for one graph stage.
#[derive(Clone, Debug, PartialEq, Eq)]
pub struct FusedNodeAttributes {
    pub stage_index: usize,
    pub stage_name: String,
    pub effective_name: String,
    pub attributes: Attributes,
}

/// Why a run selected a different tier than Auto would normally choose, or why Auto fell through
/// from typed planning to erased execution.
#[derive(Clone, Copy, Debug, PartialEq, Eq, Hash)]
pub enum FusedTierChangeReason {
    FusionErasedOnlyRequested,
    FusionTypedOnlyConstrained,
    AutoTypedUnsupportedFallback,
}

/// A tier selection event visible in an execution report.
#[derive(Clone, Debug, PartialEq, Eq)]
pub struct FusedTierChange {
    pub requested_attribute: Option<Attribute>,
    pub reason: FusedTierChangeReason,
    pub selected_tier: FusedExecutorTier,
    pub affected_nodes: Vec<String>,
}