fabro/lib/crates/fabro-workflows/src/lib.rs

#![cfg_attr(
    test,
    allow(
        clippy::absolute_paths,
        clippy::get_unwrap,
        clippy::large_futures,
        clippy::needless_borrows_for_generic_args,
        clippy::option_option,
        clippy::ptr_as_ptr,
        clippy::ref_as_ptr,
        clippy::cast_ptr_alignment,
        clippy::uninlined_format_args,
        clippy::unnecessary_literal_bound
    )
)]

use std::collections::HashMap;
use std::sync::Arc;

use fabro_retro::retro::CompletedStage;
use fabro_store::EventEnvelope;
use serde::de::DeserializeOwned;

/// Callback invoked when a workflow node starts executing.
pub type OnNodeCallback = Option<Arc<dyn Fn(&str) + Send + Sync>>;

/// Convert a Duration's milliseconds to u64, saturating on overflow.
pub(crate) fn millis_u64(d: std::time::Duration) -> u64 {
    u64::try_from(d.as_millis()).unwrap_or(u64::MAX)
}

/// Save a value as pretty-printed JSON to a file.
pub(crate) fn save_json<T: serde::Serialize>(
    value: &T,
    path: &std::path::Path,
    label: &str,
) -> error::Result<()> {
    let json = serde_json::to_string_pretty(value)
        .map_err(|e| error::FabroError::Checkpoint(format!("{label} serialize failed: {e}")))?;
    std::fs::write(path, json)?;
    Ok(())
}

/// Load a value from a JSON file.
pub(crate) fn load_json<T: DeserializeOwned>(
    path: &std::path::Path,
    label: &str,
) -> error::Result<T> {
    let data = std::fs::read_to_string(path)?;
    serde_json::from_str(&data)
        .map_err(|e| error::FabroError::Checkpoint(format!("{label} deserialize failed: {e}")))
}

/// Build `Vec<CompletedStage>` from a `Checkpoint`, mapping workflow-engine
/// types into the flat struct expected by `fabro_retro::retro::derive_retro`.
pub fn build_completed_stages(cp: &records::Checkpoint, run_failed: bool) -> Vec<CompletedStage> {
    use outcome::{OutcomeExt, StageStatus};

    let mut stages = Vec::new();
    let mut any_stage_failed = false;

    for node_id in &cp.completed_nodes {
        let outcome = cp.node_outcomes.get(node_id);
        let retries = cp.node_retries.get(node_id).copied().unwrap_or(0);

        let status = outcome.map_or_else(|| "unknown".to_string(), |o| o.status.to_string());

        let succeeded = matches!(
            outcome.map(|o| &o.status),
            Some(StageStatus::Success | StageStatus::PartialSuccess)
        );
        let failed = matches!(outcome.map(|o| &o.status), Some(StageStatus::Fail));
        if failed {
            any_stage_failed = true;
        }

        stages.push(CompletedStage {
            node_id: node_id.clone(),
            status,
            succeeded,
            failed,
            retries,
            cost: outcome.and_then(|o| o.usage.as_ref()).and_then(|u| u.cost),
            notes: outcome.and_then(|o| o.notes.clone()),
            failure_reason: outcome.and_then(|o| o.failure_reason().map(String::from)),
            files_touched: outcome.map(|o| o.files_touched.clone()).unwrap_or_default(),
        });
    }

    // If run failed with an error not captured in stages, mark the last stage
    if run_failed && !any_stage_failed {
        if let Some(last) = stages.last_mut() {
            last.failed = true;
        } else {
            stages.push(CompletedStage {
                node_id: "unknown".to_string(),
                status: "fail".to_string(),
                succeeded: false,
                failed: true,
                retries: 0,
                cost: None,
                notes: None,
                failure_reason: None,
                files_touched: vec![],
            });
        }
    }

    stages
}

pub fn extract_stage_durations_from_events(events: &[EventEnvelope]) -> HashMap<String, u64> {
    let mut durations = HashMap::new();
    for envelope in events {
        let value = envelope.payload.as_value();
        if value.get("event").and_then(serde_json::Value::as_str) != Some("StageCompleted") {
            continue;
        }
        let Some(node_id) = value.get("node_id").and_then(serde_json::Value::as_str) else {
            continue;
        };
        let Some(duration_ms) = value.get("duration_ms").and_then(serde_json::Value::as_u64) else {
            continue;
        };
        durations.insert(node_id.to_string(), duration_ms);
    }
    durations
}

#[doc(hidden)]
pub mod artifact;
pub mod asset_snapshot;
pub mod assets;
pub(crate) mod condition;
pub mod context;
pub mod devcontainer_bridge;
pub mod error;
pub mod event;
pub mod git;
pub(crate) mod graph;
pub mod handler;
mod hook_context;
#[allow(dead_code)]
pub(crate) mod lifecycle;
pub(crate) mod node_handler;
pub mod operations;
pub mod outcome;
pub mod pipeline;
pub mod pull_request;
pub mod records;
mod retry;
pub(crate) mod run_dir;
pub mod run_lookup;
pub mod run_options;
pub mod run_status;
pub mod sandbox_git;
#[doc(hidden)]
pub mod test_support;
#[doc(hidden)]
pub mod transforms;

// Re-export aliases (back-compat with `fabro_workflows::transform::*` imports)
#[doc(hidden)]
pub mod transform {
    pub use crate::transforms::*;
}
#[doc(hidden)]
pub mod vars {
    pub use crate::transforms::variable_expansion::*;
}
#[doc(hidden)]
pub use transforms::stylesheet;