obzenflow_runtime 0.2.1

// SPDX-License-Identifier: MIT OR Apache-2.0
// SPDX-FileCopyrightText: 2025-2026 ObzenFlow Contributors
// https://obzenflow.dev

use super::fsm::{FlowStopMode, PipelineEvent, PipelineState};
use crate::errors::FlowError;
use crate::journal::RunSubstrateState;
use crate::stages::common::stage_handle::STOP_REASON_TIMEOUT;
use crate::stages::LivenessSnapshots;
use crate::supervised_base::{HandleError, StandardHandle, SupervisorHandle};
use obzenflow_core::event::{SystemEvent, WriterId};
use obzenflow_core::journal::Journal;
use obzenflow_core::StageId;
use obzenflow_topology::Topology;
use serde::{Deserialize, Serialize};
use std::collections::HashMap;
use std::io;
use std::sync::Arc;
use std::time::Duration;

type ContractAttachments = Arc<HashMap<(StageId, StageId), Vec<String>>>;

pub(crate) struct FlowHandleExtras {
    pub topology: Option<Arc<Topology>>,
    pub flow_name: String,
    pub contract_attachments: Option<ContractAttachments>,
    pub system_journal: Option<Arc<dyn Journal<SystemEvent>>>,
    pub pipeline_writer_id: WriterId,
    pub liveness_snapshots: Option<LivenessSnapshots>,
    /// The selected run substrate (FLOWIP-120u): durable with its locator, or ephemeral.
    pub run_substrate: RunSubstrateState,
    /// FLOWIP-010: the build-resolved effective config, carried out of the
    /// build so the host can serve the per-flow/per-stage read surface.
    pub flow_effective_config: Option<Arc<crate::runtime_config::FlowEffectiveConfig>>,
}

/// Structural middleware configuration for a stage (FLOWIP-059).
///
/// Contains both the ordered list of middleware names and their static configuration
/// snapshots for the topology observability API.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct MiddlewareStackConfig {
    /// Ordered list of middleware names in the stack
    pub stack: Vec<String>,
    /// Circuit breaker static config (if present)
    #[serde(skip_serializing_if = "Option::is_none")]
    pub circuit_breaker: Option<serde_json::Value>,
    /// Rate limiter static config (if present)
    #[serde(skip_serializing_if = "Option::is_none")]
    pub rate_limiter: Option<serde_json::Value>,
}

impl MiddlewareStackConfig {
    /// Create a new middleware stack config with just names (no detailed config)
    pub fn names_only(stack: Vec<String>) -> Self {
        Self {
            stack,
            circuit_breaker: None,
            rate_limiter: None,
        }
    }
}

/// Immediate outcome for externally requested pipeline start admission.
#[derive(Debug, Clone, PartialEq)]
pub enum FlowStartControlOutcome {
    /// `Run` was accepted and sent while the pipeline was ready.
    Started { state: PipelineState },
    /// The pipeline was already running, so no duplicate `Run` was sent.
    AlreadyRunning { state: PipelineState },
    /// The pipeline cannot accept `Run` in the observed state.
    Rejected {
        state: PipelineState,
        reason: &'static str,
    },
}

/// Flow handle for external control - the public API returned by the DSL
///
/// This is a wrapper that combines:
/// - A standard handle for FSM control (event sending, state watching, lifecycle)
/// - Pipeline-specific functionality (metrics export)
///
/// This is the only supervisor handle that gets exposed to DSL users,
/// so it needs to provide all functionality they might need.
pub struct FlowHandle {
    /// The standard handle for FSM control
    handle: StandardHandle<PipelineEvent, PipelineState>,

    /// Pipeline-specific: Metrics access (read-only)
    metrics_exporter: Option<Arc<dyn obzenflow_core::metrics::MetricsExporter>>,

    /// Flow topology for visualization (read-only)
    topology: Option<Arc<Topology>>,

    /// User-specified flow name from flow! macro
    flow_name: String,

    /// Structural contract names per edge (for topology observability).
    ///
    /// As of FLOWIP-114b, the canonical `topology` carries stage typing,
    /// join metadata, subgraph membership, and middleware annotations
    /// directly. Contracts remain a side map because they are derived in
    /// `PipelineBuilder::build` from the topology shape and are not yet
    /// baked into the canonical `Topology`.
    contract_attachments: Option<ContractAttachments>,

    /// System journal for lifecycle events (for SSE / observability)
    system_journal: Option<Arc<dyn Journal<SystemEvent>>>,

    /// Writer identity for this pipeline's lifecycle facts in the system journal.
    pipeline_writer_id: WriterId,

    /// Flow-scoped stage liveness snapshots (FLOWIP-063e).
    liveness_snapshots: Option<LivenessSnapshots>,

    /// The selected run substrate (FLOWIP-120u).
    run_substrate: RunSubstrateState,

    /// FLOWIP-010: the build-resolved effective config with provenance.
    flow_effective_config: Option<Arc<crate::runtime_config::FlowEffectiveConfig>>,
}

impl FlowHandle {
    /// Create a new flow handle from a standard handle and extras
    pub(crate) fn new(
        handle: StandardHandle<PipelineEvent, PipelineState>,
        metrics_exporter: Option<Arc<dyn obzenflow_core::metrics::MetricsExporter>>,
        extras: FlowHandleExtras,
    ) -> Self {
        let FlowHandleExtras {
            topology,
            flow_name,
            contract_attachments,
            system_journal,
            pipeline_writer_id,
            liveness_snapshots,
            run_substrate,
            flow_effective_config,
        } = extras;

        Self {
            handle,
            metrics_exporter,
            topology,
            flow_name,
            contract_attachments,
            system_journal,
            pipeline_writer_id,
            liveness_snapshots,
            run_substrate,
            flow_effective_config,
        }
    }

    /// The run substrate selected at composition: durable with its current-run
    /// locator, or ephemeral with none (FLOWIP-120u).
    pub fn run_substrate(&self) -> &RunSubstrateState {
        &self.run_substrate
    }

    /// The build-resolved effective config (FLOWIP-010), when the flow was
    /// built through the DSL path that materializes it.
    pub fn flow_effective_config(
        &self,
    ) -> Option<&Arc<crate::runtime_config::FlowEffectiveConfig>> {
        self.flow_effective_config.as_ref()
    }

    /// Try to start the pipeline using only the currently observed state.
    ///
    /// This is intended for non-blocking control surfaces such as HTTP Play.
    /// It does not wait for readiness. Callers that want blocking startup
    /// semantics should use `start()`, `run()`, or `run_with_metrics()`.
    pub async fn start_if_ready_now(&self) -> Result<FlowStartControlOutcome, FlowError> {
        const NOT_READY_REASON: &str = "pipeline is not ready for run";

        let state = self.current_state();
        match state {
            PipelineState::ReadyForRun => {
                self.send_event(PipelineEvent::Run).await?;
                Ok(FlowStartControlOutcome::Started { state })
            }
            PipelineState::Running => Ok(FlowStartControlOutcome::AlreadyRunning { state }),
            _ => Ok(FlowStartControlOutcome::Rejected {
                state,
                reason: NOT_READY_REASON,
            }),
        }
    }

    /// Wait until the pipeline is ready to accept `Run`.
    ///
    /// Returns successfully when the pipeline reaches `ReadyForRun`, or when
    /// `Running` is already observed. Returns an error if the pipeline reaches a
    /// terminal, aborting, or post-source state before it is ready.
    pub async fn wait_for_ready(&self) -> Result<(), FlowError> {
        let mut state_rx = self.state_receiver();

        loop {
            let state = state_rx.borrow().clone();
            match state {
                PipelineState::ReadyForRun | PipelineState::Running => return Ok(()),
                PipelineState::Failed { reason, .. } => {
                    return Err(FlowError::ExecutionFailed(Box::new(io::Error::other(
                        reason,
                    ))));
                }
                PipelineState::AbortRequested { reason, .. } => {
                    return Err(FlowError::ExecutionFailed(Box::new(io::Error::other(
                        format!("{reason:?}"),
                    ))));
                }
                PipelineState::SourceCompleted => {
                    return Err(FlowError::ExecutionFailed(Box::new(io::Error::other(
                        "Pipeline source completed before it became ready for run",
                    ))));
                }
                PipelineState::Draining => {
                    return Err(FlowError::ExecutionFailed(Box::new(io::Error::other(
                        "Pipeline entered draining before it became ready for run",
                    ))));
                }
                PipelineState::Drained => {
                    return Err(FlowError::ExecutionFailed(Box::new(io::Error::other(
                        "Pipeline drained before it became ready for run",
                    ))));
                }
                PipelineState::Created
                | PipelineState::Materializing
                | PipelineState::Materialized => {}
            }

            state_rx.changed().await.map_err(|_| {
                FlowError::ExecutionFailed(Box::new(io::Error::new(
                    io::ErrorKind::BrokenPipe,
                    "Pipeline state channel closed before readiness",
                )))
            })?;
        }
    }

    /// Start the pipeline without waiting for completion.
    ///
    /// This waits until the pipeline reaches `ReadyForRun`, sends `Run` only
    /// while still in that state, and returns immediately. If the pipeline is
    /// already `Running`, this returns successfully without sending a duplicate
    /// command.
    ///
    /// If a finite flow reaches a terminal state before the post-readiness state
    /// check can send `Run`, this returns an error. Use `run()` or
    /// `run_with_metrics()` for finite flows that should be driven to completion.
    /// Intended for long-running/server flows where lifecycle is driven
    /// externally (e.g. via HTTP control API) rather than by awaiting
    /// `run()` to completion.
    pub async fn start(&self) -> Result<(), FlowError> {
        self.wait_for_ready().await?;
        let current_state = self.current_state();
        tracing::debug!(
            "FlowHandle::start() - Current pipeline state: {:?}",
            current_state
        );
        match current_state {
            PipelineState::ReadyForRun => {
                tracing::debug!("FlowHandle::start() - Sending PipelineEvent::Run to start flow");
                self.send_event(PipelineEvent::Run).await
            }
            PipelineState::Running => {
                tracing::debug!("FlowHandle::start() - Pipeline already running");
                Ok(())
            }
            PipelineState::Failed { reason, .. } => Err(FlowError::ExecutionFailed(Box::new(
                io::Error::other(reason),
            ))),
            PipelineState::AbortRequested { reason, .. } => Err(FlowError::ExecutionFailed(
                Box::new(io::Error::other(format!("{reason:?}"))),
            )),
            PipelineState::SourceCompleted
            | PipelineState::Draining
            | PipelineState::Drained
            | PipelineState::Created
            | PipelineState::Materializing
            | PipelineState::Materialized => Err(FlowError::ExecutionFailed(Box::new(
                io::Error::other(format!(
                    "Pipeline left readiness window before start command could be sent: {current_state:?}"
                )),
            ))),
        }
    }

    /// Run the pipeline and wait for completion
    ///
    /// This waits for `ReadyForRun` before sending `Run`. If the pipeline is
    /// already `Running`, it waits for completion without sending another `Run`.
    /// This is the primary method users should call after creating a flow.
    pub async fn run(self) -> Result<(), FlowError> {
        self.wait_for_ready().await?;
        let current_state = self.current_state();
        tracing::debug!(
            "FlowHandle::run() - Current pipeline state: {:?}",
            current_state
        );
        if matches!(current_state, PipelineState::ReadyForRun) {
            tracing::debug!("FlowHandle::run() - Sending PipelineEvent::Run to start flow");
            self.send_event(PipelineEvent::Run).await?;
        }
        tracing::debug!("FlowHandle::run() - Waiting for completion");

        // Capture state receiver before consuming self so we can inspect the terminal state
        let state_rx = self.state_receiver();

        // Now wait for it to complete
        let result = self.wait_for_completion().await;
        tracing::debug!(
            "FlowHandle::run() - wait_for_completion returned: {:?}",
            result
        );

        // Surface aborts/failures instead of letting the example print success on error
        if let Err(e) = result {
            tracing::error!("FlowHandle::run() failed: {}", e);
            return Err(e);
        }

        // Inspect final state to fail fast on pipeline aborts
        let final_state = state_rx.borrow().clone();
        match final_state {
            PipelineState::Failed { reason, .. } => Err(FlowError::ExecutionFailed(Box::new(
                io::Error::other(reason),
            ))),
            PipelineState::AbortRequested { reason, .. } => Err(FlowError::ExecutionFailed(
                Box::new(io::Error::other(format!("{reason:?}"))),
            )),
            _ => Ok(()),
        }
    }

    /// Run the pipeline and wait for completion, returning the metrics exporter
    /// Use this when you need to access metrics after the flow completes
    /// Typically used with finite sources (not infinite sources). This waits for
    /// `ReadyForRun` before sending `Run`; if the pipeline is already `Running`,
    /// it does not send a duplicate `Run`.
    pub async fn run_with_metrics(
        self,
    ) -> Result<Option<Arc<dyn obzenflow_core::metrics::MetricsExporter>>, FlowError> {
        self.wait_for_ready().await?;
        if matches!(self.current_state(), PipelineState::ReadyForRun) {
            self.send_event(PipelineEvent::Run).await?;
        }

        let state_rx = self.state_receiver();

        // Save metrics exporter before consuming self
        let metrics = self.metrics_exporter.clone();
        let system_journal = self.system_journal.clone();

        // Now wait for it to complete
        self.wait_for_completion().await?;

        let final_state = state_rx.borrow().clone();
        match final_state {
            PipelineState::Failed { reason, .. } => {
                return Err(FlowError::ExecutionFailed(Box::new(io::Error::other(
                    reason,
                ))));
            }
            PipelineState::AbortRequested { reason, .. } => {
                return Err(FlowError::ExecutionFailed(Box::new(io::Error::other(
                    format!("{reason:?}"),
                ))));
            }
            _ => {}
        }

        // Best-effort: wait for the metrics subsystem to complete its final export.
        //
        // Many tests (and UI clients) assume `/metrics` becomes accurate shortly after
        // pipeline completion; in practice, the metrics aggregator may still be draining.
        // We use the system journal's MetricsCoordination events as a synchronization point.
        if let Some(journal) = system_journal {
            use obzenflow_core::event::system_event::MetricsCoordinationEvent;
            use obzenflow_core::event::SystemEventType;
            use std::time::Duration;

            let deadline = std::time::Instant::now() + Duration::from_secs(10);
            while std::time::Instant::now() < deadline {
                match journal.read_last_n(256).await {
                    Ok(events) => {
                        let drained = events.iter().any(|envelope| {
                            matches!(
                                envelope.event.event,
                                SystemEventType::MetricsCoordination(
                                    MetricsCoordinationEvent::Drained
                                        | MetricsCoordinationEvent::Shutdown
                                )
                            )
                        });
                        if drained {
                            break;
                        }
                    }
                    Err(e) => {
                        tracing::warn!(
                            journal_error = %e,
                            "Failed to read system journal while waiting for metrics drain"
                        );
                        break;
                    }
                }

                tokio::time::sleep(Duration::from_millis(50)).await;
            }
        }

        Ok(metrics)
    }

    /// User-initiated stop request.
    ///
    /// This is distinct from `PipelineEvent::Shutdown` which represents natural
    /// source completion detected by the pipeline supervisor.
    pub async fn stop(&self) -> Result<(), FlowError> {
        self.stop_cancel().await
    }

    /// Stop as quickly as possible (Cancel semantics).
    pub async fn stop_cancel(&self) -> Result<(), FlowError> {
        // If the supervisor already terminated, treat Stop as an idempotent no-op.
        // This avoids surfacing "supervisor not running" as an error to callers
        // that may issue Stop more than once (e.g. UI retries).
        if !self.is_running() {
            return Ok(());
        }
        self.send_event(PipelineEvent::StopRequested {
            mode: FlowStopMode::Cancel,
            reason: None,
        })
        .await
    }

    /// Cancel due to a graceful stop timeout escalation (`stop_timeout`).
    ///
    /// This is primarily intended for process-level shutdown coordinators
    /// (e.g. SIGTERM handlers) that enforce a deadline and need terminal
    /// lifecycle observability to reflect that timeout.
    #[doc(hidden)]
    pub async fn stop_cancel_timeout(&self) -> Result<(), FlowError> {
        if !self.is_running() {
            return Ok(());
        }
        self.send_event(PipelineEvent::StopRequested {
            mode: FlowStopMode::Cancel,
            reason: Some(STOP_REASON_TIMEOUT.to_string()),
        })
        .await
    }

    /// Stop intake and attempt a bounded drain (GracefulStop semantics).
    ///
    /// On timeout expiry, the pipeline should escalate to Cancel.
    pub async fn stop_graceful(&self, timeout: Duration) -> Result<(), FlowError> {
        if !self.is_running() {
            return Ok(());
        }
        self.send_event(PipelineEvent::StopRequested {
            mode: FlowStopMode::Graceful { timeout },
            reason: None,
        })
        .await
    }

    /// Backwards-compatible alias for `stop()`.
    pub async fn shutdown(&self) -> Result<(), FlowError> {
        self.stop().await
    }

    /// Force shutdown by sending Error event to FSM
    pub async fn abort(&self, reason: &str) -> Result<(), FlowError> {
        self.send_event(PipelineEvent::Error {
            message: format!("Force abort: {reason}"),
        })
        .await
    }

    /// Check if the pipeline is still running
    pub fn is_running(&self) -> bool {
        self.handle.is_running()
    }

    /// Get a receiver for watching state changes
    pub fn state_receiver(&self) -> tokio::sync::watch::Receiver<PipelineState> {
        self.handle.state_receiver()
    }

    /// Get the latest observed pipeline supervisor state.
    pub fn current_state(&self) -> PipelineState {
        self.handle.current_state()
    }

    /// Get the metrics exporter for concurrent access during flow execution
    ///
    /// This allows starting a metrics server before running the flow,
    /// enabling real-time monitoring of long-running flows.
    /// The exporter is thread-safe and can be accessed concurrently.
    pub fn metrics_exporter(&self) -> Option<Arc<dyn obzenflow_core::metrics::MetricsExporter>> {
        self.metrics_exporter.clone()
    }

    /// Get the flow topology for visualization
    ///
    /// This provides access to the flow's structure (stages and connections)
    /// for visualization tools and monitoring dashboards.
    /// The topology is immutable and thread-safe.
    pub fn topology(&self) -> Option<Arc<Topology>> {
        self.topology.clone()
    }

    /// Get structural contract names per edge (for topology endpoint).
    ///
    /// FLOWIP-114b: middleware, join metadata, stage typing, and subgraph
    /// membership are now annotation fields on `topology()`; pull them
    /// from there. Contracts remain a side map because they are derived
    /// in `PipelineBuilder::build` from topology shape.
    pub fn contract_attachments(&self) -> Option<ContractAttachments> {
        self.contract_attachments.clone()
    }

    /// Get the system journal for lifecycle events (if available)
    pub fn system_journal(&self) -> Option<Arc<dyn Journal<SystemEvent>>> {
        self.system_journal.clone()
    }

    pub fn pipeline_writer_id(&self) -> WriterId {
        self.pipeline_writer_id
    }

    pub fn liveness_snapshots(&self) -> Option<LivenessSnapshots> {
        self.liveness_snapshots.clone()
    }

    /// Get the user-specified flow name from the flow! macro
    ///
    /// This returns the name provided in the `name:` field of the flow! macro,
    /// which may differ from the auto-generated topology-based name.
    pub fn flow_name(&self) -> &str {
        &self.flow_name
    }

    /// Render metrics based on the wrapped exporter's format
    pub async fn render_metrics(&self) -> Result<String, FlowError> {
        if let Some(ref exporter) = self.metrics_exporter {
            exporter.render_metrics().map_err(|e| {
                FlowError::ExecutionFailed(Box::new(std::io::Error::other(e.to_string())))
            })
        } else {
            Err(FlowError::ExecutionFailed(Box::new(std::io::Error::other(
                "No metrics exporter configured",
            ))))
        }
    }
}

// Custom implementation for SupervisorHandle trait to use FlowError
#[async_trait::async_trait]
impl SupervisorHandle for FlowHandle {
    type Event = PipelineEvent;
    type State = PipelineState;
    type Error = FlowError;

    async fn send_event(&self, event: Self::Event) -> Result<(), Self::Error> {
        self.handle.send_event(event).await.map_err(|e| match e {
            HandleError::SupervisorNotRunning => {
                FlowError::ExecutionFailed(Box::new(std::io::Error::new(
                    std::io::ErrorKind::BrokenPipe,
                    "Pipeline supervisor is not running",
                )))
            }
            HandleError::SupervisorFailed(msg) => {
                FlowError::ExecutionFailed(Box::new(std::io::Error::other(msg)))
            }
            HandleError::SupervisorPanicked(msg) => FlowError::ExecutionFailed(Box::new(
                std::io::Error::other(format!("Task panicked: {msg}")),
            )),
            _ => FlowError::ExecutionFailed(Box::new(std::io::Error::other(e.to_string()))),
        })
    }

    fn current_state(&self) -> Self::State {
        self.handle.current_state()
    }

    async fn wait_for_completion(self) -> Result<(), Self::Error> {
        self.handle
            .wait_for_completion()
            .await
            .map_err(|e| match e {
                HandleError::SupervisorNotRunning => {
                    FlowError::ExecutionFailed(Box::new(std::io::Error::new(
                        std::io::ErrorKind::BrokenPipe,
                        "Pipeline supervisor is not running",
                    )))
                }
                HandleError::SupervisorFailed(msg) => {
                    FlowError::ExecutionFailed(Box::new(std::io::Error::other(msg)))
                }
                HandleError::SupervisorPanicked(msg) => FlowError::ExecutionFailed(Box::new(
                    std::io::Error::other(format!("Task panicked: {msg}")),
                )),
                _ => FlowError::ExecutionFailed(Box::new(std::io::Error::other(e.to_string()))),
            })
    }

    async fn abort_and_wait(&self) -> Result<(), Self::Error> {
        self.handle.abort_and_wait().await.map_err(|error| {
            FlowError::ExecutionFailed(Box::new(std::io::Error::other(error.to_string())))
        })
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::supervised_base::{ChannelBuilder, EventReceiver, HandleBuilder};
    use obzenflow_core::event::types::ViolationCause;
    use std::error::Error;
    use tokio::sync::mpsc::error::TryRecvError;

    fn empty_extras() -> FlowHandleExtras {
        FlowHandleExtras {
            topology: None,
            flow_name: "test_flow".to_string(),
            contract_attachments: None,
            system_journal: None,
            pipeline_writer_id: WriterId::from(obzenflow_core::id::SystemId::new()),
            flow_effective_config: None,
            liveness_snapshots: None,
            run_substrate: RunSubstrateState::Ephemeral,
        }
    }

    fn flow_handle_that_finishes_in(final_state: PipelineState) -> FlowHandle {
        let (event_sender, mut event_receiver, state_watcher) =
            ChannelBuilder::<PipelineEvent, PipelineState>::new()
                .with_event_buffer(4)
                .build(PipelineState::ReadyForRun);

        let state_watcher_for_task = state_watcher.clone();
        let task = tokio::spawn(async move {
            match event_receiver.recv().await {
                Some(PipelineEvent::Run) => {
                    state_watcher_for_task
                        .update(final_state)
                        .map_err(|e| -> Box<dyn std::error::Error + Send + Sync> { Box::new(e) })?;
                    Ok(())
                }
                Some(event) => Err(format!("unexpected event: {event:?}").into()),
                None => Err("event channel closed before Run".into()),
            }
        });

        let handle = HandleBuilder::new()
            .with_event_sender(event_sender)
            .with_state_watcher(state_watcher)
            .with_supervisor_task(task)
            .build_standard()
            .expect("standard handle should build");

        FlowHandle::new(handle, None, empty_extras())
    }

    fn flow_handle_for_start_admission(
        initial_state: PipelineState,
    ) -> (FlowHandle, EventReceiver<PipelineEvent>) {
        let (event_sender, event_receiver, state_watcher) =
            ChannelBuilder::<PipelineEvent, PipelineState>::new()
                .with_event_buffer(4)
                .build(initial_state);

        let task = tokio::spawn(async { Ok::<(), Box<dyn std::error::Error + Send + Sync>>(()) });

        let handle = HandleBuilder::new()
            .with_event_sender(event_sender)
            .with_state_watcher(state_watcher)
            .with_supervisor_task(task)
            .build_standard()
            .expect("standard handle should build");

        (
            FlowHandle::new(handle, None, empty_extras()),
            event_receiver,
        )
    }

    #[tokio::test]
    async fn start_if_ready_now_dispatches_run_in_ready_for_run() {
        let (handle, mut event_receiver) =
            flow_handle_for_start_admission(PipelineState::ReadyForRun);

        let outcome = handle
            .start_if_ready_now()
            .await
            .expect("ReadyForRun admission should succeed");

        assert_eq!(
            outcome,
            FlowStartControlOutcome::Started {
                state: PipelineState::ReadyForRun
            }
        );
        assert!(
            matches!(event_receiver.try_recv(), Ok(PipelineEvent::Run)),
            "ReadyForRun admission should dispatch Run"
        );
    }

    #[tokio::test]
    async fn start_if_ready_now_accepts_running_without_dispatch() {
        let (handle, mut event_receiver) = flow_handle_for_start_admission(PipelineState::Running);

        let outcome = handle
            .start_if_ready_now()
            .await
            .expect("Running admission should succeed");

        assert_eq!(
            outcome,
            FlowStartControlOutcome::AlreadyRunning {
                state: PipelineState::Running
            }
        );
        assert!(
            matches!(event_receiver.try_recv(), Err(TryRecvError::Empty)),
            "Running admission must not dispatch duplicate Run"
        );
    }

    #[tokio::test]
    async fn start_if_ready_now_rejects_non_ready_states_without_dispatch() {
        let cases = [
            PipelineState::Created,
            PipelineState::Materializing,
            PipelineState::Materialized,
            PipelineState::SourceCompleted,
            PipelineState::AbortRequested {
                reason: ViolationCause::Other("abort".to_string()),
                upstream: None,
            },
            PipelineState::Draining,
            PipelineState::Drained,
            PipelineState::Failed {
                reason: "failed".to_string(),
                failure_cause: None,
            },
        ];

        for state in cases {
            let (handle, mut event_receiver) = flow_handle_for_start_admission(state.clone());

            let outcome = handle
                .start_if_ready_now()
                .await
                .expect("rejection should be reported as a control outcome");

            assert_eq!(
                outcome,
                FlowStartControlOutcome::Rejected {
                    state,
                    reason: "pipeline is not ready for run"
                }
            );
            assert!(
                matches!(event_receiver.try_recv(), Err(TryRecvError::Empty)),
                "rejected admission must not dispatch Run"
            );
        }
    }

    #[tokio::test]
    async fn wait_for_ready_returns_error_for_terminal_or_aborting_states() {
        let cases = [
            PipelineState::SourceCompleted,
            PipelineState::AbortRequested {
                reason: ViolationCause::Other("abort".to_string()),
                upstream: None,
            },
            PipelineState::Draining,
            PipelineState::Drained,
            PipelineState::Failed {
                reason: "failed".to_string(),
                failure_cause: None,
            },
        ];

        for state in cases {
            let (handle, _event_receiver) = flow_handle_for_start_admission(state);

            let result = handle.wait_for_ready().await;

            assert!(result.is_err(), "terminal state must not satisfy readiness");
        }
    }

    #[tokio::test]
    async fn start_accepts_coalesced_running_without_dispatch() {
        let (handle, mut event_receiver) = flow_handle_for_start_admission(PipelineState::Running);

        handle
            .start()
            .await
            .expect("Running should satisfy start without dispatching Run");

        assert!(
            matches!(event_receiver.try_recv(), Err(TryRecvError::Empty)),
            "start must not dispatch duplicate Run after observing Running"
        );
    }

    #[tokio::test]
    async fn run_with_metrics_returns_failed_terminal_state_as_error() {
        let handle = flow_handle_that_finishes_in(PipelineState::Failed {
            reason: "terminal failure".to_string(),
            failure_cause: None,
        });

        let result = handle.run_with_metrics().await;
        assert!(
            result.is_err(),
            "Failed terminal state must surface as an error"
        );
        let err = result.err().expect("error should be present");
        let source = err.source().expect("source error should be present");

        assert!(
            source.to_string().contains("terminal failure"),
            "unexpected source error: {source}"
        );
    }

    #[tokio::test]
    async fn run_with_metrics_returns_abort_terminal_state_as_error() {
        let handle = flow_handle_that_finishes_in(PipelineState::AbortRequested {
            reason: ViolationCause::Other("abort requested".to_string()),
            upstream: None,
        });

        let result = handle.run_with_metrics().await;
        assert!(
            result.is_err(),
            "AbortRequested terminal state must surface as an error"
        );
        let err = result.err().expect("error should be present");
        let source = err.source().expect("source error should be present");

        assert!(
            source.to_string().contains("abort requested"),
            "unexpected source error: {source}"
        );
    }

    #[tokio::test]
    async fn run_with_metrics_allows_successful_terminal_state() {
        let handle = flow_handle_that_finishes_in(PipelineState::Drained);

        let metrics = handle
            .run_with_metrics()
            .await
            .expect("Drained terminal state should remain successful");

        assert!(metrics.is_none());
    }
}