nornir 0.5.4

//! **f5 — a build/pipeline is a root jera job whose body is a DAG of sub-jobs.**
//!
//! A build is not one job: it is a **root** [`crate::jobs`] job (kind
//! [`crate::jobs::kind::CI_BUILD`]) that **owns a [`PlanDag`] of sub-jobs**, and
//! **every pipeline step is a tracked sub-job** in its own right — its own
//! `job_id`, its own `queued → running → done|failed` lifecycle on the ledger, and
//! (via slice 2) its own build-info sub-stream filtered by `step_id`. Edges are
//! `needs` dependencies (a step runs only once every step it needs is `done`).
//!
//! ## The DAG is data — reuse, don't re-roll
//!
//! The graph vocabulary is **funnel-core's [`PlanDag`]** (`topo_order` /
//! `critical_path` / `depths` = the ready-wave layering). We do NOT invent a graph
//! type: a [`PipelineSpec`] projects onto a `PlanDag` ([`PipelineSpec::to_plan_dag`])
//! where **nodes = sub-jobs** and **edges = `needs` dependencies** ([`EdgeKind::Dep`]),
//! so `critical_path()` gives the bottleneck lineage and `depths()` the scheduling
//! wave for free.
//!
//! ## The ledger is the sub-job hierarchy — reuse, don't re-roll
//!
//! The lifecycle rides the **nornir-jobs** ledger ([`JobStore`]): the root is a
//! normal top-level job; each step is a **child** [`JobRecord`] (`parent_id` = the
//! root's `job_id`), admitted `queued` up front (the root literally *owns* the DAG
//! of sub-jobs), promoted `running` when its wave is ready, then `done|failed`. A
//! step whose `needs` did not all reach `done` is left blocked → `failed` without
//! running. The viz Jobs tree already folds children under their parent by
//! `parent_id`, so no viz change is needed to see the DAG per-step state.
//!
//! ## One engine for nornir AND dwarves
//!
//! This is the single `root job → pipeline DAG → sub-jobs` model the design (§f5)
//! calls for; dwarves' pipeline steps and nornir's build waves are the same shape,
//! not two schedulers.
//!
//! ## Additive
//!
//! The existing single-job build path is unchanged — it is simply a one-step
//! "DAG" (one node, zero edges, one sub-job), see
//! [`tests::single_step_is_a_one_node_dag_additive`].

use std::collections::BTreeMap;

use anyhow::{bail, Result};
use serde::{Deserialize, Serialize};

use crate::funnel::{DagStatus, EdgeKind, PlanDag, PlanDagEdge, PlanDagNode};
use crate::jobs::{kind, status, JobRecord, JobStore};
use nornir_build_thing::build_info::BuildInfoRow;

/// One declarative pipeline step = one sub-job to run + the step ids it `needs`.
///
/// `id` is the **logical, author-given** step id (stable across runs); it is what
/// [`PlanDag`] edges reference and what a caller passes to
/// [`PipelineOutcome::build_info_for`]. It is distinct from the ledger `job_id` the
/// run mints for the sub-job (a fresh uuid) — the build-info stream's `step_id`
/// column carries that `job_id` (see [`PipelineOutcome::job_id_for`]).
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct PipelineStep {
    /// The stable logical step id (the [`PlanDag`] node id / edge endpoint).
    pub id: String,
    /// A human title for the sub-job (the ledger `target`).
    pub name: String,
    /// The step ids this step depends on — it runs only once all of them are `done`.
    #[serde(default)]
    pub needs: Vec<String>,
    /// The ledger [`crate::jobs::kind`] for this sub-job; empty → [`kind::CI_BUILD`].
    #[serde(default)]
    pub kind: String,
}

impl PipelineStep {
    /// A step `id` titled `name` that `needs` the given step ids (kind defaults to
    /// [`kind::CI_BUILD`]).
    pub fn new(id: &str, name: &str, needs: &[&str]) -> Self {
        PipelineStep {
            id: id.to_string(),
            name: name.to_string(),
            needs: needs.iter().map(|s| s.to_string()).collect(),
            kind: String::new(),
        }
    }

    /// The effective ledger kind for this sub-job.
    fn job_kind(&self) -> &str {
        if self.kind.is_empty() { kind::CI_BUILD } else { &self.kind }
    }
}

/// A declarative build **pipeline**: a DAG of steps. The DAG *is data* — it
/// projects onto a funnel-core [`PlanDag`] ([`Self::to_plan_dag`]) for all graph
/// reasoning and runs onto the job ledger ([`Self::run_on_ledger`]) as a root job
/// owning one sub-job per step.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct PipelineSpec {
    /// The pipeline name (the root job's `target` + the DAG's `sel_plan` scope).
    pub name: String,
    /// The steps, in any order — dependencies come from `needs`, not list position.
    pub steps: Vec<PipelineStep>,
}

impl PipelineSpec {
    /// Project this spec onto a funnel-core [`PlanDag`] — the DAG vocabulary the
    /// whole slice reuses. **Nodes = sub-jobs** (one per step), **edges = `needs`
    /// dependencies** ([`EdgeKind::Dep`], `need → step`). All steps start
    /// [`DagStatus::Idle`] (i.e. queued); `fanout` is the number of steps that need
    /// this one. `critical_path()` / `depths()` / `topo_order()` then work verbatim.
    #[must_use]
    pub fn to_plan_dag(&self) -> PlanDag {
        // fan-out = how many steps depend ON this step.
        let mut needed_by: BTreeMap<&str, u64> = BTreeMap::new();
        for s in &self.steps {
            for n in &s.needs {
                *needed_by.entry(n.as_str()).or_default() += 1;
            }
        }
        let nodes: Vec<PlanDagNode> = self
            .steps
            .iter()
            .map(|s| PlanDagNode {
                id: s.id.clone(),
                title: s.name.clone(),
                group: Some(self.name.clone()),
                status: DagStatus::Idle,
                is_bug: false,
                staleness_hours: 0.0,
                weight_bytes: 0,
                fanout: needed_by.get(s.id.as_str()).copied().unwrap_or(0),
            })
            .collect();
        let mut edges: Vec<PlanDagEdge> = Vec::new();
        for s in &self.steps {
            for n in &s.needs {
                edges.push(PlanDagEdge { from: n.clone(), to: s.id.clone(), kind: EdgeKind::Dep });
            }
        }
        edges.sort_by(|a, b| (&a.from, &a.to).cmp(&(&b.from, &b.to)));
        PlanDag { nodes, edges, sel_plan: Some(self.name.clone()) }
    }

    /// Validate the spec: non-empty, unique step ids, every `needs` resolves, and
    /// the graph is acyclic (a build pipeline must be a DAG).
    pub fn validate(&self) -> Result<()> {
        if self.steps.is_empty() {
            bail!("pipeline `{}` has no steps", self.name);
        }
        let mut seen: BTreeMap<&str, ()> = BTreeMap::new();
        for s in &self.steps {
            if seen.insert(s.id.as_str(), ()).is_some() {
                bail!("pipeline `{}` has a duplicate step id `{}`", self.name, s.id);
            }
        }
        for s in &self.steps {
            for n in &s.needs {
                if !seen.contains_key(n.as_str()) {
                    bail!("step `{}` needs unknown step `{}`", s.id, n);
                }
            }
        }
        self.to_plan_dag()
            .topo_order()
            .map_err(|c| anyhow::anyhow!("pipeline `{}` has a dependency cycle around {c:?}", self.name))?;
        Ok(())
    }

    /// **Materialize + run** this pipeline on the ledger: a root [`kind::CI_BUILD`]
    /// job that OWNS a [`PlanDag`] of sub-jobs. Each step becomes a child sub-job
    /// with its own `job_id` and a `queued → running → done|failed` lifecycle;
    /// steps run in dependency (ready-wave) order, and a step whose `needs` did not
    /// all reach `done` is left blocked and marked `failed` without running.
    ///
    /// `run_step` is the executor seam: `Ok(())` = the step succeeded (`done`),
    /// `Err` = it failed (`failed`, error carried into the sub-job's `detail_json`).
    /// The build itself is not run here — this is the DAG/ledger wiring; a real
    /// caller passes a closure that drives `build_variant_with_info` (slice 2) per
    /// step, tagging its build-info rows with the sub-job's `job_id` as `step_id`.
    ///
    /// Returns the [`PipelineOutcome`]: the root id, the projected DAG, and every
    /// sub-job's `(step_id, job_id, status)` — all robot-observable via
    /// [`PipelineOutcome::state_json`].
    pub fn run_on_ledger(
        &self,
        store: &JobStore,
        workspace: &str,
        mut run_step: impl FnMut(&PipelineStep) -> Result<()>,
    ) -> Result<PipelineOutcome> {
        self.validate()?;
        let dag = self.to_plan_dag();
        // Acyclic guaranteed by validate(); topo order = dependency-before-dependent.
        let order = dag
            .topo_order()
            .map_err(|c| anyhow::anyhow!("pipeline `{}` cycle around {c:?}", self.name))?;

        // The root job OWNS the DAG.
        let root = store.start(
            kind::CI_BUILD,
            &self.name,
            workspace,
            serde_json::json!({
                "pipeline": self.name,
                "steps": self.steps.len(),
                "critical_path": dag.critical_path(),
                "critical_path_len": dag.critical_path_len(),
            }),
        );
        let root_id = root.job_id().to_string();

        // Admit every step as a QUEUED sub-job up front (the root owns the whole DAG
        // of sub-jobs; they light up wave by wave).
        let mut job_id_of: BTreeMap<String, String> = BTreeMap::new();
        let mut rec_of: BTreeMap<String, JobRecord> = BTreeMap::new();
        for s in &self.steps {
            let rec = new_child_record(&root_id, s, workspace, status::QUEUED);
            store.submit(&rec)?;
            job_id_of.insert(s.id.clone(), rec.job_id.clone());
            rec_of.insert(s.id.clone(), rec);
        }

        // Ready-wave execution: topo order visits dependencies before dependents,
        // so by the time we reach a step every `need` has a final status.
        let step_by_id: BTreeMap<&str, &PipelineStep> =
            self.steps.iter().map(|s| (s.id.as_str(), s)).collect();
        let mut final_status: BTreeMap<String, String> = BTreeMap::new();
        for sid in &order {
            let Some(step) = step_by_id.get(sid.as_str()) else { continue };
            let rec = rec_of.get_mut(sid).expect("every step has a queued record");

            // Blocked if any need did not reach `done`.
            let blocked = step
                .needs
                .iter()
                .any(|n| final_status.get(n).map(|st| st != status::DONE).unwrap_or(true));
            if blocked {
                let blockers: Vec<&String> = step
                    .needs
                    .iter()
                    .filter(|n| final_status.get(n.as_str()).map(|st| st != status::DONE).unwrap_or(true))
                    .collect();
                terminate(
                    rec,
                    status::FAILED,
                    serde_json::json!({ "step_id": step.id, "needs": step.needs, "blocked_by": blockers }),
                );
                store.submit(rec)?;
                final_status.insert(sid.clone(), status::FAILED.to_string());
                continue;
            }

            // queued → running.
            promote_running(rec);
            store.submit(rec)?;
            let st = match run_step(step) {
                Ok(()) => {
                    terminate(rec, status::DONE, serde_json::json!({ "step_id": step.id, "needs": step.needs }));
                    status::DONE
                }
                Err(e) => {
                    terminate(
                        rec,
                        status::FAILED,
                        serde_json::json!({ "step_id": step.id, "needs": step.needs, "error": format!("{e:#}") }),
                    );
                    status::FAILED
                }
            };
            store.submit(rec)?;
            final_status.insert(sid.clone(), st.to_string());
        }

        let sub_jobs: Vec<SubJob> = self
            .steps
            .iter()
            .map(|s| SubJob {
                step_id: s.id.clone(),
                job_id: job_id_of[&s.id].clone(),
                status: final_status.get(&s.id).cloned().unwrap_or_else(|| status::QUEUED.to_string()),
                needs: s.needs.clone(),
            })
            .collect();
        let ok = sub_jobs.iter().all(|s| s.status == status::DONE);

        if ok {
            root.finish(serde_json::json!({ "steps": self.steps.len(), "outcome": "done" }), "");
        } else {
            let failed: Vec<&String> = sub_jobs.iter().filter(|s| s.status != status::DONE).map(|s| &s.step_id).collect();
            root.fail_with_detail(serde_json::json!({ "failed_steps": failed }));
        }

        Ok(PipelineOutcome { root_id, dag, sub_jobs, ok })
    }
}

/// One sub-job's recorded outcome — the tie between a logical `step_id`, its ledger
/// `job_id`, and its final lifecycle `status`.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct SubJob {
    /// The logical, author-given step id ([`PipelineStep::id`]).
    pub step_id: String,
    /// The ledger `job_id` minted for this sub-job — the build-info stream tags its
    /// rows' `step_id` column with THIS value (see [`PipelineOutcome::build_info_for`]).
    pub job_id: String,
    /// The final ledger [`crate::jobs::status`] (`done` / `failed`; `queued` only if
    /// the run never reached it).
    pub status: String,
    /// The step ids this sub-job depended on (the `needs` edges).
    pub needs: Vec<String>,
}

/// The result of a [`PipelineSpec::run_on_ledger`]: the root job id, the projected
/// [`PlanDag`], and every sub-job.
#[derive(Debug, Clone)]
pub struct PipelineOutcome {
    /// The root job's `job_id` — the build-info stream's `build_id`.
    pub root_id: String,
    /// The DAG that was owned/run (funnel-core [`PlanDag`]).
    pub dag: PlanDag,
    /// One entry per step, in spec order.
    pub sub_jobs: Vec<SubJob>,
    /// Whether every step reached `done`.
    pub ok: bool,
}

impl PipelineOutcome {
    /// The ledger `job_id` a logical `step_id` was run as — i.e. the value the
    /// build-info stream carries in its `step_id` column for that sub-job.
    #[must_use]
    pub fn job_id_for(&self, step_id: &str) -> Option<&str> {
        self.sub_jobs.iter().find(|s| s.step_id == step_id).map(|s| s.job_id.as_str())
    }

    /// **Filter a slice-2 build-info stream to ONE sub-job.** Returns the rows whose
    /// `build_id` is this run's root AND whose `step_id` is the sub-job's ledger
    /// `job_id` — the per-step feed the design (§f5) calls for ("each step's
    /// log/metric stream is the f3 feed filtered by `step_id`"). Unknown `step_id`
    /// → empty.
    #[must_use]
    pub fn build_info_for<'a>(&self, rows: &'a [BuildInfoRow], step_id: &str) -> Vec<&'a BuildInfoRow> {
        let Some(job_id) = self.job_id_for(step_id) else { return Vec::new() };
        rows.iter().filter(|r| r.build_id == self.root_id && r.step_id == job_id).collect()
    }

    /// The robot-observable introspection blob (LAW 6): the root id, the DAG
    /// vocabulary (critical path etc.), and every sub-job's `(step_id, job_id,
    /// status, needs)`.
    #[must_use]
    pub fn state_json(&self) -> serde_json::Value {
        serde_json::json!({
            "root_id": self.root_id,
            "ok": self.ok,
            "sub_job_count": self.sub_jobs.len(),
            "critical_path": self.dag.critical_path(),
            "critical_path_len": self.dag.critical_path_len(),
            "dep_edges": self.dag.dep_edges(),
            "sub_jobs": self.sub_jobs.iter().map(|s| serde_json::json!({
                "step_id": s.step_id,
                "job_id": s.job_id,
                "status": s.status,
                "needs": s.needs,
            })).collect::<Vec<_>>(),
        })
    }
}

fn now_micros() -> i64 {
    chrono::Utc::now().timestamp_micros()
}

/// A fresh child [`JobRecord`] for `step` under `root_id`, in `initial_status`. The
/// `needs` edges are stashed into `detail_json` so the ledger row itself carries the
/// dependency set (the DAG is recoverable from the ledger, not only the projection).
fn new_child_record(root_id: &str, step: &PipelineStep, ws: &str, initial_status: &str) -> JobRecord {
    JobRecord {
        job_id: uuid::Uuid::new_v4().to_string(),
        kind: step.job_kind().to_string(),
        target: step.name.clone(),
        workspace: ws.to_string(),
        status: initial_status.to_string(),
        ts_start_micros: now_micros(),
        ts_end_micros: None,
        elapsed_ms: None,
        detail_json: serde_json::json!({ "step_id": step.id, "needs": step.needs }).to_string(),
        result_ref: String::new(),
        parent_id: Some(root_id.to_string()),
        node: None,
    }
}

/// Promote a queued record to `running`, re-stamping `ts_start` so `elapsed_ms`
/// measures actual work, not the queue wait (mirrors the ledger's own scheduler).
fn promote_running(rec: &mut JobRecord) {
    rec.status = status::RUNNING.to_string();
    rec.ts_start_micros = now_micros();
}

/// Write a terminal status onto a record (mirrors [`crate::jobs::JobHandle`]'s
/// private `terminate`, which we cannot reach for a caller-owned record).
fn terminate(rec: &mut JobRecord, st: &str, detail: serde_json::Value) {
    let end = now_micros();
    rec.status = st.to_string();
    rec.ts_end_micros = Some(end);
    rec.elapsed_ms = Some((end - rec.ts_start_micros) / 1000);
    rec.detail_json = detail.to_string();
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::jobs::JobSelector;
    use nornir_build_thing::build_info::{BuildInfoKind, BuildInfoRow};

    fn tmpdir(tag: &str) -> std::path::PathBuf {
        std::env::temp_dir().join(format!("nornir-pipeline-{tag}-{}-{}", std::process::id(), now_micros()))
    }

    /// The canonical 3-node build DAG: fetch → compile → test.
    fn three_node() -> PipelineSpec {
        PipelineSpec {
            name: "build".into(),
            steps: vec![
                PipelineStep::new("fetch", "fetch sources", &[]),
                PipelineStep::new("compile", "cargo build", &["fetch"]),
                PipelineStep::new("test", "cargo test", &["compile"]),
            ],
        }
    }

    /// A build-info row tagged for a given `(build_id, step_id)`.
    fn row(build_id: &str, step_id: &str, kind: BuildInfoKind, name: &str) -> BuildInfoRow {
        let mut r = BuildInfoRow::new(build_id, 1, kind);
        r.step_id = step_id.to_string();
        r.name = name.to_string();
        r
    }

    /// **RED-when-broken (the slice's headline):** a 3-node DAG creates a root +
    /// three sub-jobs on the ledger, with the correct `needs` edges and per-step
    /// status.
    #[test]
    fn three_node_dag_makes_root_plus_three_subjobs_with_needs_and_status() {
        let dir = tmpdir("3node");
        let store = JobStore::open(&dir).unwrap();
        let spec = three_node();

        // The DAG vocabulary is funnel-core's PlanDag (reuse, not re-rolled).
        let dag = spec.to_plan_dag();
        assert_eq!(dag.node_count(), 3, "one node per step");
        assert_eq!(dag.dep_edges(), 2, "fetch→compile, compile→test");
        assert_eq!(dag.critical_path(), vec!["fetch", "compile", "test"]);
        assert_eq!(dag.critical_path_len(), 2, "two sequential waves define the schedule");

        let out = spec.run_on_ledger(&store, "nordisk", |_step| Ok(())).unwrap();
        assert!(out.ok, "all steps ran green");

        let all = store.list(&JobSelector::All).unwrap();
        assert_eq!(all.len(), 4, "one root + three sub-jobs on the ledger");

        // The root: top-level CI_BUILD, done.
        let root = all.iter().find(|r| r.job_id == out.root_id).expect("root on ledger");
        assert_eq!(root.kind, kind::CI_BUILD);
        assert_eq!(root.parent_id, None, "root is top-level");
        assert_eq!(root.status, status::DONE);

        // Exactly three children, all done, all folded under the root by parent_id.
        let children: Vec<&JobRecord> =
            all.iter().filter(|r| r.parent_id.as_deref() == Some(out.root_id.as_str())).collect();
        assert_eq!(children.len(), 3, "three sub-jobs under the root");
        assert!(children.iter().all(|r| r.status == status::DONE), "every sub-job done");

        // The `needs` edges are recorded on each sub-job's ledger row.
        let compile = children.iter().find(|r| r.detail_json.contains("\"step_id\":\"compile\"")).unwrap();
        assert!(compile.detail_json.contains("fetch"), "compile's needs edge (fetch) is on the ledger row");
        let fetch = children.iter().find(|r| r.detail_json.contains("\"step_id\":\"fetch\"")).unwrap();
        assert!(fetch.detail_json.contains("\"needs\":[]"), "fetch has no needs");

        // Per-step status via the outcome.
        assert!(out.sub_jobs.iter().all(|s| s.status == status::DONE));
        assert_eq!(out.sub_jobs.len(), 3);

        nornir_testmatrix::functional_status(
            "build-pipeline-subjobs",
            "three_node_root_plus_subjobs",
            all.len() == 4 && children.len() == 3 && dag.dep_edges() == 2,
            "3-node DAG → 1 root + 3 sub-jobs with needs edges + per-step done",
        );

        std::fs::remove_dir_all(&dir).ok();
    }

    /// A failed step blocks its dependents and fails the root: compile fails → test
    /// is blocked (never runs) → both non-done, root failed. RED-when-broken.
    #[test]
    fn failed_step_blocks_dependents_and_fails_root() {
        let dir = tmpdir("fail");
        let store = JobStore::open(&dir).unwrap();
        let spec = three_node();

        let out = spec
            .run_on_ledger(&store, "ws", |s| if s.id == "compile" { bail!("compile boom") } else { Ok(()) })
            .unwrap();
        assert!(!out.ok, "a failed step fails the pipeline");

        let by = |id: &str| out.sub_jobs.iter().find(|s| s.step_id == id).unwrap();
        assert_eq!(by("fetch").status, status::DONE, "fetch ran green");
        assert_eq!(by("compile").status, status::FAILED, "compile failed");
        assert_eq!(by("test").status, status::FAILED, "test is blocked by the failed compile");

        // The blocked step records WHY (blocked_by), and the failed step its error.
        let all = store.list(&JobSelector::All).unwrap();
        let test = all.iter().find(|r| r.job_id == by("test").job_id).unwrap();
        assert!(test.detail_json.contains("blocked_by"), "test row says it was blocked");
        assert!(test.detail_json.contains("compile"), "blocked by compile");
        let compile = all.iter().find(|r| r.job_id == by("compile").job_id).unwrap();
        assert!(compile.detail_json.contains("compile boom"), "compile carries its error");
        let root = all.iter().find(|r| r.job_id == out.root_id).unwrap();
        assert_eq!(root.status, status::FAILED, "root fails when a step fails");

        nornir_testmatrix::functional_status(
            "build-pipeline-subjobs",
            "failed_step_blocks_dependents",
            !out.ok && by("test").status == status::FAILED && root.status == status::FAILED,
            "compile fails → test blocked → root failed",
        );

        std::fs::remove_dir_all(&dir).ok();
    }

    /// Additive: the existing single-job build path is a one-step "DAG" — one node,
    /// zero edges, root + one sub-job, both done.
    #[test]
    fn single_step_is_a_one_node_dag_additive() {
        let dir = tmpdir("single");
        let store = JobStore::open(&dir).unwrap();
        let spec = PipelineSpec {
            name: "single".into(),
            steps: vec![PipelineStep::new("build", "cargo build", &[])],
        };
        let dag = spec.to_plan_dag();
        assert_eq!(dag.node_count(), 1);
        assert_eq!(dag.dep_edges(), 0, "a lone step has no needs edges");
        assert_eq!(dag.critical_path_len(), 0, "one node = zero waves of dependency");

        let out = spec.run_on_ledger(&store, "ws", |_| Ok(())).unwrap();
        assert!(out.ok);
        let all = store.list(&JobSelector::All).unwrap();
        assert_eq!(all.len(), 2, "root + one sub-job");
        assert!(all.iter().all(|r| r.status == status::DONE));

        std::fs::remove_dir_all(&dir).ok();
    }

    /// The build-info stream (slice 2) filters per `step_id` to each sub-job: rows
    /// tagged with a sub-job's ledger `job_id` come back only for that step.
    #[test]
    fn build_info_stream_filters_per_step_id_to_each_subjob() {
        let dir = tmpdir("binfo");
        let store = JobStore::open(&dir).unwrap();
        let spec = three_node();
        let out = spec.run_on_ledger(&store, "ws", |_| Ok(())).unwrap();

        let fetch_job = out.job_id_for("fetch").unwrap().to_string();
        let compile_job = out.job_id_for("compile").unwrap().to_string();
        assert_ne!(fetch_job, compile_job, "each sub-job has its own id");

        // A mixed stream for the whole build (build_id = root), tagged per sub-job.
        let rows = vec![
            row(&out.root_id, &fetch_job, BuildInfoKind::Log, "fetching"),
            row(&out.root_id, &compile_job, BuildInfoKind::Log, "compiling"),
            row(&out.root_id, &compile_job, BuildInfoKind::Metric, "warnings"),
            // A row for a different build entirely — must NOT leak in.
            row("other-build", &compile_job, BuildInfoKind::Log, "elsewhere"),
        ];

        let fetch_rows = out.build_info_for(&rows, "fetch");
        assert_eq!(fetch_rows.len(), 1, "one row for fetch's sub-job");
        assert_eq!(fetch_rows[0].name, "fetching");

        let compile_rows = out.build_info_for(&rows, "compile");
        assert_eq!(compile_rows.len(), 2, "compile's own two rows, not the other build's");
        assert!(compile_rows.iter().all(|r| r.build_id == out.root_id));

        // Unknown step → empty (not a panic).
        assert!(out.build_info_for(&rows, "nope").is_empty());

        nornir_testmatrix::functional_status(
            "build-pipeline-subjobs",
            "build_info_filtered_per_step",
            fetch_rows.len() == 1 && compile_rows.len() == 2,
            "slice-2 stream filtered by sub-job step_id",
        );

        std::fs::remove_dir_all(&dir).ok();
    }

    /// A cyclic spec is rejected (a build pipeline must be a DAG).
    #[test]
    fn cyclic_pipeline_is_rejected() {
        let dir = tmpdir("cycle");
        let store = JobStore::open(&dir).unwrap();
        let spec = PipelineSpec {
            name: "loop".into(),
            steps: vec![
                PipelineStep::new("a", "a", &["b"]),
                PipelineStep::new("b", "b", &["a"]),
            ],
        };
        assert!(spec.validate().is_err(), "a cycle is invalid");
        assert!(spec.run_on_ledger(&store, "ws", |_| Ok(())).is_err(), "run refuses a cyclic spec");
        // And a needs pointing at a non-existent step is rejected too.
        let bad = PipelineSpec { name: "x".into(), steps: vec![PipelineStep::new("a", "a", &["ghost"])] };
        assert!(bad.validate().is_err(), "dangling needs is invalid");

        std::fs::remove_dir_all(&dir).ok();
    }

    /// The DAG projection reuses funnel-core's critical_path across a branchy DAG:
    /// a diamond fetch → (build_x ‖ build_y) → link picks a 2-wave critical path.
    #[test]
    fn diamond_dag_reuses_funnel_core_critical_path() {
        let spec = PipelineSpec {
            name: "diamond".into(),
            steps: vec![
                PipelineStep::new("fetch", "fetch", &[]),
                PipelineStep::new("build_x", "build x", &["fetch"]),
                PipelineStep::new("build_y", "build y", &["fetch"]),
                PipelineStep::new("link", "link", &["build_x", "build_y"]),
            ],
        };
        let dag = spec.to_plan_dag();
        assert_eq!(dag.dep_edges(), 4);
        assert_eq!(dag.critical_path_len(), 2, "fetch → build_* → link = 2 waves");
        // fetch has fan-out 2 (both builds need it); link has fan-out 0.
        let fetch = dag.nodes.iter().find(|n| n.id == "fetch").unwrap();
        assert_eq!(fetch.fanout, 2, "two steps need fetch");
        let s = out_state(&spec);
        assert_eq!(s["dep_edges"], 4);
    }

    fn out_state(spec: &PipelineSpec) -> serde_json::Value {
        let dir = tmpdir("state");
        let store = JobStore::open(&dir).unwrap();
        let out = spec.run_on_ledger(&store, "ws", |_| Ok(())).unwrap();
        let s = out.state_json();
        std::fs::remove_dir_all(&dir).ok();
        s
    }
}