feat(#2282): animated indicatif DAG progress render for hivectl wait

hive-c0re/src/bin/hivectl.rs

@@ -1459,9 +1459,22 @@ async fn agents_restart(socket: &Path, name: &str, no_wait: bool) -> Result<()>
 /// 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<u64>, 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<u64>) -> Result<()> {
     let mut pending: std::collections::BTreeSet<u64> = ids.into_iter().collect();
     let mut last: std::collections::HashMap<u64, String> = std::collections::HashMap::new();
     let mut failed: Vec<String> = Vec::new();
@@ -1490,10 +1503,12 @@ async fn wait_for_dags(socket: &Path, ids: Vec<u64>, no_wait: bool) -> Result<()
                 // 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.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;
-                } else if d.state == hive_sh4re::jobs::State::Failed {
-                    failed.push(format!("{} {}", d.kind.as_str(), d.agent));
                 }
             }
             if all_terminal {
@@ -1504,6 +1519,118 @@ async fn wait_for_dags(socket: &Path, ids: Vec<u64>, no_wait: bool) -> Result<()
             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<u64>) -> 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<u64, ProgressBar> =
+        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<u64> = ids.into_iter().collect();
+    let mut failed: Vec<String> = 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<String>) -> Result<()> {
     if failed.is_empty() {
         Ok(())
     } else {
@@ -1513,6 +1640,73 @@ async fn wait_for_dags(socket: &Path, ids: Vec<u64>, no_wait: bool) -> Result<()
     }
 }
 
+/// 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 => "⏸",