From e0d5c98356ecc1bc7b9dc977d8f78b3a3ac486ef Mon Sep 17 00:00:00 2001 From: damocles Date: Fri, 10 Jul 2026 16:24:18 +0200 Subject: [PATCH] refactor(#2282): split dag_progress out of hivectl.rs --- hive-c0re/src/bin/hivectl.rs | 382 +-------------------- hive-c0re/src/bin/hivectl/dag_progress.rs | 386 ++++++++++++++++++++++ 2 files changed, 394 insertions(+), 374 deletions(-) create mode 100644 hive-c0re/src/bin/hivectl/dag_progress.rs diff --git a/hive-c0re/src/bin/hivectl.rs b/hive-c0re/src/bin/hivectl.rs index 7a1cf9d1..bde9f405 100644 --- a/hive-c0re/src/bin/hivectl.rs +++ b/hive-c0re/src/bin/hivectl.rs @@ -22,6 +22,14 @@ use anyhow::{Context as _, Result, bail}; use clap::{Args, Parser, Subcommand}; use hive_c0re::coordinator::Coordinator; +/// Rebuild-queue DAG progress rendering (`wait_for_dags` + the spinner / +/// plain renderers), split out to keep this file manageable. `#[path]` keeps +/// the file under `bin/hivectl/` (a subdir cargo won't treat as its own +/// binary) rather than the sibling `bin/dag_progress.rs` a bare `mod` maps to. +#[path = "hivectl/dag_progress.rs"] +mod dag_progress; +use dag_progress::wait_for_dags; + #[derive(Parser)] #[command( name = "hivectl", @@ -1449,301 +1457,6 @@ async fn agents_restart(socket: &Path, name: &str, no_wait: bool) -> Result<()> } } -// --------------------------------------------------------------------------- -// Job-queue wait/progress loop — shared by every verb that submits DAGs -// --------------------------------------------------------------------------- - -/// Poll the submitted DAG ids (`HostRequest::QueueDag`, ~1s interval) -/// and print a progress line whenever a DAG's rendered state changes — -/// including fan-out children that appear under a polled parent. Exits -/// non-zero when any DAG (or child) ends `failed`; a `cancelled` DAG -/// terminates the wait but is an operator action, not an error. -async fn wait_for_dags(socket: &Path, ids: Vec, no_wait: bool) -> Result<()> { - use std::io::IsTerminal as _; - if no_wait || ids.is_empty() { - return Ok(()); - } - // Animate only on a real terminal. Piped / CI output falls back to the - // plain line-on-change stream so logs stay free of spinner redraw noise. - if std::io::stderr().is_terminal() { - wait_for_dags_animated(socket, ids).await - } else { - wait_for_dags_plain(socket, ids).await - } -} - -/// Non-TTY progress: print a fresh line whenever a DAG's rendered state -/// changes. No cursor tricks, so it's clean in pipes and CI logs. -async fn wait_for_dags_plain(socket: &Path, ids: Vec) -> Result<()> { - let mut pending: std::collections::BTreeSet = ids.into_iter().collect(); - let mut last: std::collections::HashMap = std::collections::HashMap::new(); - let mut failed: Vec = Vec::new(); - while !pending.is_empty() { - for id in pending.clone() { - let resp = hive_c0re::client::request(socket, hive_sh4re::HostRequest::QueueDag { id }) - .await - .with_context(|| format!("connect to daemon socket {}", socket.display()))?; - let dags = resp.dags.unwrap_or_default(); - if dags.is_empty() { - // Evicted from the queue's history tail — it finished a - // while ago; nothing left to report on. - println!("job #{id}: gone from queue history"); - pending.remove(&id); - continue; - } - let mut all_terminal = true; - for d in &dags { - let line = render_dag_line(d); - if last.get(&d.id) != Some(&line) { - println!("{line}"); - last.insert(d.id, line); - } - // Node-level terminality, not the roll-up: a DAG rolls - // up `failed` the moment one node fails while its - // after-any recovery node (rebuild's tail Reconcile) - // may still be running — keep watching so the operator - // sees whether the agent came back. - if d.nodes.iter().all(|n| n.state.is_terminal()) { - if d.state == hive_sh4re::jobs::State::Failed { - failed.push(format!("{} {}", d.kind.as_str(), d.agent)); - } - } else { - all_terminal = false; - } - } - if all_terminal { - pending.remove(&id); - } - } - if !pending.is_empty() { - tokio::time::sleep(std::time::Duration::from_secs(1)).await; - } - } - finish_wait(failed) -} - -/// TTY progress: a live `indicatif` render — one braille-spinner line per -/// DAG node (grouped under a per-DAG header), each with its own elapsed -/// timer, plus an overall-elapsed footer. A node with more than one -/// dependency (fan-in) gets its own row with an `(after …)` marker rather -/// than being crammed onto a chain line. -async fn wait_for_dags_animated(socket: &Path, ids: Vec) -> Result<()> { - use indicatif::{MultiProgress, ProgressBar, ProgressStyle}; - - let spinner = ProgressStyle::with_template(" {spinner} {msg}") - .unwrap_or_else(|_| ProgressStyle::default_spinner()) - .tick_strings(&["⠋", "⠙", "⠹", "⠸", "⠼", "⠴", "⠦", "⠧", "⠇", "⠏", "·"]); - let plain = - ProgressStyle::with_template("{msg}").unwrap_or_else(|_| ProgressStyle::default_bar()); - - let mp = MultiProgress::new(); - let started = std::time::Instant::now(); - // Per-DAG header bars + per-node bars, keyed so we update in place. The - // insertion order groups a DAG's nodes right under its header. - let mut dag_bars: std::collections::HashMap = - std::collections::HashMap::new(); - let mut node_bars: std::collections::HashMap<(u64, hive_sh4re::jobs::NodeId), ProgressBar> = - std::collections::HashMap::new(); - let mut node_done: std::collections::HashSet<(u64, hive_sh4re::jobs::NodeId)> = - std::collections::HashSet::new(); - - let mut pending: std::collections::BTreeSet = ids.into_iter().collect(); - let mut failed: Vec = Vec::new(); - - while !pending.is_empty() { - let now = now_unix(); - for id in pending.clone() { - let resp = hive_c0re::client::request(socket, hive_sh4re::HostRequest::QueueDag { id }) - .await - .with_context(|| format!("connect to daemon socket {}", socket.display()))?; - let dags = resp.dags.unwrap_or_default(); - if dags.is_empty() { - mp.println(format!("job #{id}: gone from queue history")) - .ok(); - pending.remove(&id); - continue; - } - let mut all_terminal = true; - for d in &dags { - let hdr = dag_bars.entry(d.id).or_insert_with(|| { - let b = mp.add(ProgressBar::new_spinner()); - b.set_style(plain.clone()); - b - }); - hdr.set_message(format!( - "{} {} {} · {}", - state_glyph(d.state), - d.kind.as_str(), - d.agent, - fmt_dur(dag_elapsed(d, now)), - )); - for n in &d.nodes { - let key = (d.id, n.id); - if node_done.contains(&key) { - continue; - } - let bar = node_bars.entry(key).or_insert_with(|| { - let b = mp.add(ProgressBar::new_spinner()); - b.set_style(spinner.clone()); - b.enable_steady_tick(std::time::Duration::from_millis(120)); - b - }); - if n.state.is_terminal() { - bar.set_style(plain.clone()); - bar.finish_with_message(format!( - " {} {}", - state_glyph(n.state), - node_line(d, n, now) - )); - node_done.insert(key); - } else { - bar.set_message(node_line(d, n, now)); - } - } - if d.nodes.iter().all(|n| n.state.is_terminal()) { - if d.state == hive_sh4re::jobs::State::Failed { - failed.push(format!("{} {}", d.kind.as_str(), d.agent)); - } - } else { - all_terminal = false; - } - } - if all_terminal { - pending.remove(&id); - } - } - if !pending.is_empty() { - tokio::time::sleep(std::time::Duration::from_secs(1)).await; - } - } - // Leave the final node lines on screen; drop the still-spinning headers - // into their terminal state and print an overall-elapsed footer. - for hdr in dag_bars.values() { - hdr.finish(); - } - mp.println(format!( - "done · {} elapsed", - fmt_dur(started.elapsed().as_secs().try_into().unwrap_or(0)) - )) - .ok(); - finish_wait(failed) -} - -/// Shared tail: succeed when nothing failed, else bail listing the failures. -fn finish_wait(mut failed: Vec) -> Result<()> { - if failed.is_empty() { - Ok(()) - } else { - failed.sort(); - failed.dedup(); - bail!("queued job(s) failed: {}", failed.join(", ")) - } -} - -/// Current unix time in seconds (0 on the impossible pre-epoch error). -fn now_unix() -> i64 { - std::time::SystemTime::now() - .duration_since(std::time::UNIX_EPOCH) - .ok() - .and_then(|d| i64::try_from(d.as_secs()).ok()) - .unwrap_or(0) -} - -/// Elapsed seconds for a DAG: `started_at` (falling back to `enqueued_at`) -/// through `finished_at` or `now`. -fn dag_elapsed(d: &hive_sh4re::jobs::DagView, now: i64) -> i64 { - let start = d.started_at.unwrap_or(d.enqueued_at); - (d.finished_at.unwrap_or(now) - start).max(0) -} - -/// Elapsed seconds for a node: `started_at` → `finished_at`/`now`, or 0 -/// when it hasn't started. -fn node_elapsed(n: &hive_sh4re::jobs::NodeView, now: i64) -> i64 { - match n.started_at { - Some(start) => (n.finished_at.unwrap_or(now) - start).max(0), - None => 0, - } -} - -/// Compact duration: `45s` under a minute, else `1m03s`. -fn fmt_dur(secs: i64) -> String { - let s = secs.max(0); - if s < 60 { - format!("{s}s") - } else { - format!("{}m{:02}s", s / 60, s % 60) - } -} - -/// One animated node line: kind, live step, an `(after …)` marker for a -/// fan-in node (>1 dep), its elapsed timer, and a truncated error tail. -fn node_line(d: &hive_sh4re::jobs::DagView, n: &hive_sh4re::jobs::NodeView, now: i64) -> String { - use std::fmt::Write as _; - let mut s = n.kind.clone(); - if n.state == hive_sh4re::jobs::State::Running - && let Some(step) = &n.step - { - let _ = write!(s, " ({step})"); - } - if n.deps.len() > 1 { - let after: Vec<&str> = n - .deps - .iter() - .filter_map(|dep| d.nodes.iter().find(|m| m.id == *dep)) - .map(|m| m.kind.as_str()) - .collect(); - if !after.is_empty() { - let _ = write!(s, " (after {})", after.join(", ")); - } - } - let el = node_elapsed(n, now); - if el > 0 { - let _ = write!(s, " · {}", fmt_dur(el)); - } - if let Some(err) = &n.error { - let short: String = err.chars().take(100).collect(); - let _ = write!(s, " — {short}"); - } - s -} - -fn state_glyph(state: hive_sh4re::jobs::State) -> &'static str { - match state { - hive_sh4re::jobs::State::Queued => "⏸", - hive_sh4re::jobs::State::Running => "▶", - hive_sh4re::jobs::State::Done => "✔", - hive_sh4re::jobs::State::Failed => "✖", - hive_sh4re::jobs::State::Cancelled => "⊘", - } -} - -/// One progress line for a DAG: roll-up glyph, template, agent, then -/// the node chain with the running node's live step label — the CLI -/// twin of the dashboard's queue card. -fn render_dag_line(d: &hive_sh4re::jobs::DagView) -> String { - use std::fmt::Write as _; - let mut out = format!( - "{} {} {:<12}", - state_glyph(d.state), - d.kind.as_str(), - d.agent - ); - for (i, n) in d.nodes.iter().enumerate() { - let sep = if i == 0 { " " } else { " → " }; - let _ = write!(out, "{sep}{} {}", state_glyph(n.state), n.kind); - if n.state == hive_sh4re::jobs::State::Running - && let Some(step) = &n.step - { - let _ = write!(out, " ({step})"); - } - } - if let Some(err) = d.nodes.iter().find_map(|n| n.error.as_deref()) { - let short: String = err.chars().take(120).collect(); - let _ = write!(out, " — {short}"); - } - out -} - /// `hivectl agents list` — fetch the per-agent status roster from the /// daemon (`HostRequest::AgentStatus`) and render it as a padded table, /// or the raw JSON rows with `--json`. Reuses the dashboard's @@ -2021,82 +1734,3 @@ fn validate_htpasswd_username(username: &str) -> Result<()> { } Ok(()) } - -#[cfg(test)] -mod tests { - use hive_sh4re::jobs::{DagView, NodeView, Source, State, Template}; - - use super::render_dag_line; - - fn node(id: u32, kind: &str, state: State, step: Option<&str>) -> NodeView { - NodeView { - id, - kind: kind.to_owned(), - deps: if id == 0 { vec![] } else { vec![id - 1] }, - state, - step: step.map(str::to_owned), - build_log_id: None, - started_at: None, - finished_at: None, - error: None, - } - } - - #[test] - fn render_dag_line_shows_chain_and_running_step() { - let dag = DagView { - id: 7, - agent: "alice".to_owned(), - kind: Template::Rebuild, - state: State::Running, - source: Source::Manual, - parent_id: None, - reason: "manual".to_owned(), - enqueued_at: 0, - started_at: Some(1), - finished_at: None, - inputs: vec![], - approval_id: None, - perm_payload: None, - nodes: vec![ - node(0, "prebuild", State::Done, None), - node(1, "stop_for_update", State::Done, None), - node(2, "swap", State::Running, Some("nixos-container update")), - node(3, "reconcile", State::Queued, None), - ], - }; - let line = render_dag_line(&dag); - assert!(line.starts_with("▶ rebuild alice"), "{line}"); - assert!( - line.contains( - "✔ prebuild → ✔ stop_for_update → ▶ swap (nixos-container update) → ⏸ reconcile" - ), - "{line}" - ); - } - - #[test] - fn render_dag_line_surfaces_first_node_error() { - let mut failed = node(0, "prebuild", State::Failed, None); - failed.error = Some("nix build exploded".to_owned()); - let dag = DagView { - id: 8, - agent: "bob".to_owned(), - kind: Template::Rebuild, - state: State::Failed, - source: Source::Manual, - parent_id: None, - reason: "manual".to_owned(), - enqueued_at: 0, - started_at: Some(1), - finished_at: Some(2), - inputs: vec![], - approval_id: None, - perm_payload: None, - nodes: vec![failed], - }; - let line = render_dag_line(&dag); - assert!(line.contains("✖ rebuild"), "{line}"); - assert!(line.contains("— nix build exploded"), "{line}"); - } -} diff --git a/hive-c0re/src/bin/hivectl/dag_progress.rs b/hive-c0re/src/bin/hivectl/dag_progress.rs new file mode 100644 index 00000000..52fb3b11 --- /dev/null +++ b/hive-c0re/src/bin/hivectl/dag_progress.rs @@ -0,0 +1,386 @@ +//! `hivectl` rebuild-queue progress rendering. +//! +//! Split out of `hivectl.rs` (which is already large): everything that +//! polls the daemon's DAG queue (`HostRequest::QueueDag`) and renders the +//! per-DAG / per-node progress lives here. [`wait_for_dags`] is the entry +//! point the command handlers call; it dispatches to a live `indicatif` +//! animation on a TTY and a plain line-on-change stream otherwise. + +use std::path::Path; + +use anyhow::{Context as _, Result, bail}; + +/// Poll the submitted DAG ids (`HostRequest::QueueDag`, ~1s interval) and +/// render progress until they all reach a terminal state. Exits non-zero +/// (via the returned `Err`) when any DAG (or fan-out child) ends `failed`; +/// a `cancelled` DAG terminates the wait but is an operator action, not an +/// error. +/// +/// # Errors +/// +/// Returns an error if the daemon socket can't be reached, or if any polled +/// DAG finished in the `failed` state. +pub(crate) async fn wait_for_dags(socket: &Path, ids: Vec, no_wait: bool) -> Result<()> { + use std::io::IsTerminal as _; + if no_wait || ids.is_empty() { + return Ok(()); + } + // Animate only on a real terminal. Piped / CI output falls back to the + // plain line-on-change stream so logs stay free of spinner redraw noise. + if std::io::stderr().is_terminal() { + wait_for_dags_animated(socket, ids).await + } else { + wait_for_dags_plain(socket, ids).await + } +} + +/// Non-TTY progress: print a fresh line whenever a DAG's rendered state +/// changes. No cursor tricks, so it's clean in pipes and CI logs. +async fn wait_for_dags_plain(socket: &Path, ids: Vec) -> Result<()> { + let mut pending: std::collections::BTreeSet = ids.into_iter().collect(); + let mut last: std::collections::HashMap = std::collections::HashMap::new(); + let mut failed: Vec = Vec::new(); + while !pending.is_empty() { + for id in pending.clone() { + let resp = hive_c0re::client::request(socket, hive_sh4re::HostRequest::QueueDag { id }) + .await + .with_context(|| format!("connect to daemon socket {}", socket.display()))?; + let dags = resp.dags.unwrap_or_default(); + if dags.is_empty() { + // Evicted from the queue's history tail — it finished a + // while ago; nothing left to report on. + println!("job #{id}: gone from queue history"); + pending.remove(&id); + continue; + } + let mut all_terminal = true; + for d in &dags { + let line = render_dag_line(d); + if last.get(&d.id) != Some(&line) { + println!("{line}"); + last.insert(d.id, line); + } + // Node-level terminality, not the roll-up: a DAG rolls + // up `failed` the moment one node fails while its + // after-any recovery node (rebuild's tail Reconcile) + // may still be running — keep watching so the operator + // sees whether the agent came back. + if d.nodes.iter().all(|n| n.state.is_terminal()) { + if d.state == hive_sh4re::jobs::State::Failed { + failed.push(format!("{} {}", d.kind.as_str(), d.agent)); + } + } else { + all_terminal = false; + } + } + if all_terminal { + pending.remove(&id); + } + } + if !pending.is_empty() { + tokio::time::sleep(std::time::Duration::from_secs(1)).await; + } + } + finish_wait(failed) +} + +/// TTY progress: a live `indicatif` render — one braille-spinner line per +/// DAG node (grouped under a per-DAG header), each with its own elapsed +/// timer, plus an overall-elapsed footer. A node with more than one +/// dependency (fan-in) gets its own row with an `(after …)` marker rather +/// than being crammed onto a chain line. +async fn wait_for_dags_animated(socket: &Path, ids: Vec) -> Result<()> { + use indicatif::{MultiProgress, ProgressBar, ProgressStyle}; + + let spinner = ProgressStyle::with_template(" {spinner} {msg}") + .unwrap_or_else(|_| ProgressStyle::default_spinner()) + .tick_strings(&["⠋", "⠙", "⠹", "⠸", "⠼", "⠴", "⠦", "⠧", "⠇", "⠏", "·"]); + let plain = + ProgressStyle::with_template("{msg}").unwrap_or_else(|_| ProgressStyle::default_bar()); + + let mp = MultiProgress::new(); + let started = std::time::Instant::now(); + // Per-DAG header bars + per-node bars, keyed so we update in place. The + // insertion order groups a DAG's nodes right under its header. + let mut dag_bars: std::collections::HashMap = + std::collections::HashMap::new(); + let mut node_bars: std::collections::HashMap<(u64, hive_sh4re::jobs::NodeId), ProgressBar> = + std::collections::HashMap::new(); + let mut node_done: std::collections::HashSet<(u64, hive_sh4re::jobs::NodeId)> = + std::collections::HashSet::new(); + + let mut pending: std::collections::BTreeSet = ids.into_iter().collect(); + let mut failed: Vec = Vec::new(); + + while !pending.is_empty() { + let now = now_unix(); + for id in pending.clone() { + let resp = hive_c0re::client::request(socket, hive_sh4re::HostRequest::QueueDag { id }) + .await + .with_context(|| format!("connect to daemon socket {}", socket.display()))?; + let dags = resp.dags.unwrap_or_default(); + if dags.is_empty() { + mp.println(format!("job #{id}: gone from queue history")) + .ok(); + pending.remove(&id); + continue; + } + let mut all_terminal = true; + for d in &dags { + let hdr = dag_bars.entry(d.id).or_insert_with(|| { + let b = mp.add(ProgressBar::new_spinner()); + b.set_style(plain.clone()); + b + }); + hdr.set_message(format!( + "{} {} {} · {}", + state_glyph(d.state), + d.kind.as_str(), + d.agent, + fmt_dur(dag_elapsed(d, now)), + )); + for n in &d.nodes { + let key = (d.id, n.id); + if node_done.contains(&key) { + continue; + } + let bar = node_bars.entry(key).or_insert_with(|| { + let b = mp.add(ProgressBar::new_spinner()); + b.set_style(spinner.clone()); + b.enable_steady_tick(std::time::Duration::from_millis(120)); + b + }); + if n.state.is_terminal() { + bar.set_style(plain.clone()); + bar.finish_with_message(format!( + " {} {}", + state_glyph(n.state), + node_line(d, n, now) + )); + node_done.insert(key); + } else { + bar.set_message(node_line(d, n, now)); + } + } + if d.nodes.iter().all(|n| n.state.is_terminal()) { + if d.state == hive_sh4re::jobs::State::Failed { + failed.push(format!("{} {}", d.kind.as_str(), d.agent)); + } + } else { + all_terminal = false; + } + } + if all_terminal { + pending.remove(&id); + } + } + if !pending.is_empty() { + tokio::time::sleep(std::time::Duration::from_secs(1)).await; + } + } + // Leave the final node lines on screen; drop the still-spinning headers + // into their terminal state and print an overall-elapsed footer. + for hdr in dag_bars.values() { + hdr.finish(); + } + mp.println(format!( + "done · {} elapsed", + fmt_dur(started.elapsed().as_secs().try_into().unwrap_or(0)) + )) + .ok(); + finish_wait(failed) +} + +/// Shared tail: succeed when nothing failed, else bail listing the failures. +fn finish_wait(mut failed: Vec) -> Result<()> { + if failed.is_empty() { + Ok(()) + } else { + failed.sort(); + failed.dedup(); + bail!("queued job(s) failed: {}", failed.join(", ")) + } +} + +/// Current unix time in seconds (0 on the impossible pre-epoch error). +fn now_unix() -> i64 { + std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .ok() + .and_then(|d| i64::try_from(d.as_secs()).ok()) + .unwrap_or(0) +} + +/// Elapsed seconds for a DAG: `started_at` (falling back to `enqueued_at`) +/// through `finished_at` or `now`. +fn dag_elapsed(d: &hive_sh4re::jobs::DagView, now: i64) -> i64 { + let start = d.started_at.unwrap_or(d.enqueued_at); + (d.finished_at.unwrap_or(now) - start).max(0) +} + +/// Elapsed seconds for a node: `started_at` → `finished_at`/`now`, or 0 +/// when it hasn't started. +fn node_elapsed(n: &hive_sh4re::jobs::NodeView, now: i64) -> i64 { + match n.started_at { + Some(start) => (n.finished_at.unwrap_or(now) - start).max(0), + None => 0, + } +} + +/// Compact duration: `45s` under a minute, else `1m03s`. +fn fmt_dur(secs: i64) -> String { + let s = secs.max(0); + if s < 60 { + format!("{s}s") + } else { + format!("{}m{:02}s", s / 60, s % 60) + } +} + +/// One animated node line: kind, live step, an `(after …)` marker for a +/// fan-in node (>1 dep), its elapsed timer, and a truncated error tail. +fn node_line(d: &hive_sh4re::jobs::DagView, n: &hive_sh4re::jobs::NodeView, now: i64) -> String { + use std::fmt::Write as _; + let mut s = n.kind.clone(); + if n.state == hive_sh4re::jobs::State::Running + && let Some(step) = &n.step + { + let _ = write!(s, " ({step})"); + } + if n.deps.len() > 1 { + let after: Vec<&str> = n + .deps + .iter() + .filter_map(|dep| d.nodes.iter().find(|m| m.id == *dep)) + .map(|m| m.kind.as_str()) + .collect(); + if !after.is_empty() { + let _ = write!(s, " (after {})", after.join(", ")); + } + } + let el = node_elapsed(n, now); + if el > 0 { + let _ = write!(s, " · {}", fmt_dur(el)); + } + if let Some(err) = &n.error { + let short: String = err.chars().take(100).collect(); + let _ = write!(s, " — {short}"); + } + s +} + +fn state_glyph(state: hive_sh4re::jobs::State) -> &'static str { + match state { + hive_sh4re::jobs::State::Queued => "⏸", + hive_sh4re::jobs::State::Running => "▶", + hive_sh4re::jobs::State::Done => "✔", + hive_sh4re::jobs::State::Failed => "✖", + hive_sh4re::jobs::State::Cancelled => "⊘", + } +} + +/// One progress line for a DAG: roll-up glyph, template, agent, then +/// the node chain with the running node's live step label — the CLI +/// twin of the dashboard's queue card. Used by the plain (non-TTY) path. +fn render_dag_line(d: &hive_sh4re::jobs::DagView) -> String { + use std::fmt::Write as _; + let mut out = format!( + "{} {} {:<12}", + state_glyph(d.state), + d.kind.as_str(), + d.agent + ); + for (i, n) in d.nodes.iter().enumerate() { + let sep = if i == 0 { " " } else { " → " }; + let _ = write!(out, "{sep}{} {}", state_glyph(n.state), n.kind); + if n.state == hive_sh4re::jobs::State::Running + && let Some(step) = &n.step + { + let _ = write!(out, " ({step})"); + } + } + if let Some(err) = d.nodes.iter().find_map(|n| n.error.as_deref()) { + let short: String = err.chars().take(120).collect(); + let _ = write!(out, " — {short}"); + } + out +} + +#[cfg(test)] +mod tests { + use hive_sh4re::jobs::{DagView, NodeView, Source, State, Template}; + + use super::render_dag_line; + + fn node(id: u32, kind: &str, state: State, step: Option<&str>) -> NodeView { + NodeView { + id, + kind: kind.to_owned(), + deps: if id == 0 { vec![] } else { vec![id - 1] }, + state, + step: step.map(str::to_owned), + build_log_id: None, + started_at: None, + finished_at: None, + error: None, + } + } + + #[test] + fn render_dag_line_shows_chain_and_running_step() { + let dag = DagView { + id: 7, + agent: "alice".to_owned(), + kind: Template::Rebuild, + state: State::Running, + source: Source::Manual, + parent_id: None, + reason: "manual".to_owned(), + enqueued_at: 0, + started_at: Some(1), + finished_at: None, + inputs: vec![], + approval_id: None, + perm_payload: None, + nodes: vec![ + node(0, "prebuild", State::Done, None), + node(1, "stop_for_update", State::Done, None), + node(2, "swap", State::Running, Some("nixos-container update")), + node(3, "reconcile", State::Queued, None), + ], + }; + let line = render_dag_line(&dag); + assert!(line.starts_with("▶ rebuild alice"), "{line}"); + assert!( + line.contains( + "✔ prebuild → ✔ stop_for_update → ▶ swap (nixos-container update) → ⏸ reconcile" + ), + "{line}" + ); + } + + #[test] + fn render_dag_line_surfaces_first_node_error() { + let mut failed = node(0, "prebuild", State::Failed, None); + failed.error = Some("nix build exploded".to_owned()); + let dag = DagView { + id: 8, + agent: "bob".to_owned(), + kind: Template::Rebuild, + state: State::Failed, + source: Source::Manual, + parent_id: None, + reason: "manual".to_owned(), + enqueued_at: 0, + started_at: Some(1), + finished_at: Some(2), + inputs: vec![], + approval_id: None, + perm_payload: None, + nodes: vec![failed], + }; + let line = render_dag_line(&dag); + assert!(line.contains("✖ rebuild"), "{line}"); + assert!(line.contains("— nix build exploded"), "{line}"); + } +}