diff --git a/Cargo.lock b/Cargo.lock index f1de027e..03443d9a 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -609,6 +609,19 @@ dependencies = [ "crossbeam-utils", ] +[[package]] +name = "console" +version = "0.15.11" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "054ccb5b10f9f2cbf51eb355ca1d05c2d279ce1804688d0db74b4733a5aeafd8" +dependencies = [ + "encode_unicode", + "libc", + "once_cell", + "unicode-width", + "windows-sys 0.59.0", +] + [[package]] name = "const-oid" version = "0.9.6" @@ -935,6 +948,12 @@ version = "1.16.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "91622ff5e7162018101f2fea40d6ebf4a78bbe5a49736a2020649edf9693679e" +[[package]] +name = "encode_unicode" +version = "1.0.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "34aa73646ffb006b8f5147f3dc182bd4bcb190227ce861fc4a4844bf8e3cb2c0" + [[package]] name = "encoding_rs" version = "0.8.35" @@ -1433,6 +1452,7 @@ dependencies = [ "clap_complete", "forgejo-api", "hive-sh4re", + "indicatif", "libc", "listenfd", "petgraph", @@ -1901,6 +1921,19 @@ dependencies = [ "serde_core", ] +[[package]] +name = "indicatif" +version = "0.17.11" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "183b3088984b400f4cfac3620d5e076c84da5364016b4f49473de574b2586235" +dependencies = [ + "console", + "number_prefix", + "portable-atomic", + "unicode-width", + "web-time", +] + [[package]] name = "inout" version = "0.1.4" @@ -2520,6 +2553,12 @@ dependencies = [ "libc", ] +[[package]] +name = "number_prefix" +version = "0.4.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "830b246a0e5f20af87141b25c173cd1b609bd7779a4617d6ec582abaf90870f3" + [[package]] name = "oauth2" version = "5.0.0" @@ -2697,6 +2736,12 @@ dependencies = [ "universal-hash", ] +[[package]] +name = "portable-atomic" +version = "1.13.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c33a9471896f1c69cecef8d20cbe2f7accd12527ce60845ff44c153bb2a21b49" + [[package]] name = "potential_utf" version = "0.1.5" @@ -4328,6 +4373,12 @@ dependencies = [ "tinyvec", ] +[[package]] +name = "unicode-width" +version = "0.2.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b4ac048d71ede7ee76d585517add45da530660ef4390e49b098733c6e897f254" + [[package]] name = "unicode-xid" version = "0.2.6" @@ -4717,6 +4768,15 @@ dependencies = [ "windows-targets 0.52.6", ] +[[package]] +name = "windows-sys" +version = "0.59.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1e38bc4d79ed67fd075bcc251a1c39b32a1776bbe92e5bef1f0bf1f8c531853b" +dependencies = [ + "windows-targets 0.52.6", +] + [[package]] name = "windows-sys" version = "0.60.2" diff --git a/Cargo.toml b/Cargo.toml index 8208a293..b06e72c9 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -35,6 +35,7 @@ chrono = { version = "0.4", default-features = false, features = [ ] } clap = { version = "4", features = ["derive"] } clap_complete = "4" +indicatif = "0.17" hive-sh4re = { path = "hive-sh4re" } hive-claude = { path = "hive-claude" } thiserror = "2" diff --git a/hive-c0re/Cargo.toml b/hive-c0re/Cargo.toml index de3e6b77..5f9fdfe2 100644 --- a/hive-c0re/Cargo.toml +++ b/hive-c0re/Cargo.toml @@ -17,6 +17,7 @@ url.workspace = true clap.workspace = true clap_complete.workspace = true clap-markdown = "0.1" +indicatif.workspace = true hive-sh4re.workspace = true libc.workspace = true listenfd = "1" diff --git a/hive-c0re/src/bin/hivectl.rs b/hive-c0re/src/bin/hivectl.rs index 5f4398b7..7a1cf9d1 100644 --- a/hive-c0re/src/bin/hivectl.rs +++ b/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, 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(); @@ -1490,10 +1503,12 @@ async fn wait_for_dags(socket: &Path, ids: Vec, 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, 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) -> 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 { @@ -1513,6 +1640,73 @@ async fn wait_for_dags(socket: &Path, ids: Vec, 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 => "⏸",