diff --git a/hive-c0re/src/dashboard/build_logs.rs b/hive-c0re/src/dashboard/build_logs.rs index b0771608..657557b5 100644 --- a/hive-c0re/src/dashboard/build_logs.rs +++ b/hive-c0re/src/dashboard/build_logs.rs @@ -90,6 +90,42 @@ pub(super) async fn get_build_log_full( } } +/// `GET /api/build-log/{node_id}` — the build log for a **queue node**, +/// resolved node id → log-row id → full log. Same `BuildLogFull` JSON +/// (`stdout` / `stderr` + header) as `get_build_log_full`; HTTP 404 when the +/// node has no linked log (the client gates the request on `NodeView.has_log`, +/// but a vacuum race can still 404). This is the on-demand log fetch the +/// raw-graph dashboard uses instead of an inline `build_log_id` on the wire. +pub(super) async fn get_build_log_for_node( + State(state): State, + AxumPath(node_id): AxumPath, +) -> Response { + match state.coord.job_queue.build_log_id_of(node_id) { + Some(log_id) => get_build_log_full(State(state), AxumPath(log_id)).await, + None => ( + StatusCode::NOT_FOUND, + format!("node #{node_id} has no build log"), + ) + .into_response(), + } +} + +/// `GET /api/build-log/{node_id}/raw` — the node's build log as `text/plain` +/// for download (delegates to `get_build_log_raw` after resolving the node id). +pub(super) async fn get_build_log_raw_for_node( + State(state): State, + AxumPath(node_id): AxumPath, +) -> Response { + match state.coord.job_queue.build_log_id_of(node_id) { + Some(log_id) => get_build_log_raw(State(state), AxumPath(log_id)).await, + None => ( + StatusCode::NOT_FOUND, + format!("node #{node_id} has no build log"), + ) + .into_response(), + } +} + /// JSON frame sent on the `/api/build-logs/id/{id}/stream` SSE channel. /// `stdout_append` / `stderr_append` carry only the new bytes since the /// last frame; `done = true` means the build finished and the stream diff --git a/hive-c0re/src/dashboard/mod.rs b/hive-c0re/src/dashboard/mod.rs index 580feeba..7334a7d5 100644 --- a/hive-c0re/src/dashboard/mod.rs +++ b/hive-c0re/src/dashboard/mod.rs @@ -100,6 +100,14 @@ pub async fn serve( ) .route("/api/audit-log", get(misc_api::api_audit_log)) .route("/api/build-logs", get(build_logs::get_build_logs_all)) + .route( + "/api/build-log/{node_id}", + get(build_logs::get_build_log_for_node), + ) + .route( + "/api/build-log/{node_id}/raw", + get(build_logs::get_build_log_raw_for_node), + ) .route( "/api/build-logs/{agent}", get(build_logs::get_build_logs_agent), diff --git a/hive-c0re/src/job_queue/mod.rs b/hive-c0re/src/job_queue/mod.rs index 2e5a987a..b549d8d2 100644 --- a/hive-c0re/src/job_queue/mod.rs +++ b/hive-c0re/src/job_queue/mod.rs @@ -38,6 +38,7 @@ mod tests; use std::collections::HashMap; use std::sync::Mutex; +use chrono::{DateTime, Utc}; use hive_jobq::resources::ResourceTable; use hive_jobq::scheduler::{Outcome, Scheduler}; use hive_jobq::{Dep, DepWhen as JobDepWhen, Graph, NodeId, State as JobState}; @@ -96,15 +97,14 @@ pub struct TerminalDag { pub error: Option, } -/// Per-node runtime metadata the crate graph doesn't carry (kind + agent live -/// in the node payload; state lives in the node). +/// Per-node runtime metadata the crate graph doesn't carry. Lifecycle +/// (`started_at` / `finished_at` / `error`) lives on the `hive_jobq::Node` +/// itself now, so only the two host-side extras remain: the live sub-step +/// label and the build-log row link (the client fetches the log by node id). #[derive(Debug, Default, Clone)] struct NodeRuntime { step: Option, build_log_id: Option, - started_at: Option, - finished_at: Option, - error: Option, } /// An owned read-view of a DAG container's carried metadata ([`NodeKind::Dag`]). @@ -362,7 +362,6 @@ impl JobQueue { let mut inner = self.lock(); let inner = &mut *inner; let started = inner.sched.settle(); - let now = now_unix(); let mut claims = Vec::with_capacity(started.len()); for id in started { let Some(node) = inner.sched.graph().node(id) else { @@ -386,9 +385,8 @@ impl JobQueue { inputs: meta.inputs, transient: meta.transient, }); - if let Some(rt) = inner.node_rt.get_mut(&id) { - rt.started_at = Some(now); - } + // `started_at` is stamped on the graph `Node` by the scheduler's + // transition to `Running` — no host-side copy needed. } claims } @@ -405,23 +403,15 @@ impl JobQueue { result: Result<(), String>, ) -> Option { let mut inner = self.lock(); - let now = now_unix(); - let (error, outcome) = match result { - Ok(()) => (None, Outcome::Done), - Err(e) => { - // The reason rides the crate `Outcome::Failed` (stamped onto the - // graph `Node`); the `node_rt` copy stays for now until the wire - // reads it off the node directly. - let msg = truncate_error(&e); - (Some(msg.clone()), Outcome::Failed(msg)) - } + // The failure reason + `finished_at` are stamped onto the graph `Node` + // by the scheduler (the reason rides `Outcome::Failed`); no host-side + // copy. We only clear the live sub-step label here. + let outcome = match result { + Ok(()) => Outcome::Done, + Err(e) => Outcome::Failed(truncate_error(&e)), }; if let Some(rt) = inner.node_rt.get_mut(&node_id) { - rt.finished_at = Some(now); rt.step = None; - if let Some(e) = error { - rt.error = Some(e); - } } let container = inner.dag_of(node_id); inner.sched.complete(node_id, outcome); @@ -521,6 +511,21 @@ impl JobQueue { true } + /// The `build_logs` row id linked to the wire node id `node_id`, if any — + /// the lookup behind the `GET /api/build-log/` query endpoint (the + /// client fetches a node's captured build output on demand rather than + /// receiving it inline). Takes the raw wire `u64` (the endpoint's path + /// param); `node_rt` is keyed by the opaque `NodeId`, so this scans for the + /// matching id — the map is small (live + recently-terminal nodes). + #[must_use] + pub fn build_log_id_of(&self, node_id: u64) -> Option { + self.lock() + .node_rt + .iter() + .find(|(nid, _)| nid.get() == node_id) + .and_then(|(_, rt)| rt.build_log_id) + } + /// A DAG's terminal roll-up summary, computed on demand from its container. /// `None` if the DAG id is unknown. Test-only — production reads the summary /// `complete_node` returns when the container rolls up terminal. @@ -721,15 +726,13 @@ impl QueueInner { seen } - /// First failed work node's stored error, for the roll-up `error` field. + /// First failed work node's error (read off the graph `Node`), for the + /// terminal roll-up summary the inline hook consumes. fn dag_first_error(&self, container: NodeId) -> Option { for id in self.subtree(container) { - if self - .sched - .graph() - .node(id) - .is_some_and(|n| n.state == JobState::Failed) - && let Some(e) = self.node_rt.get(&id).and_then(|r| r.error.clone()) + if let Some(n) = self.sched.graph().node(id) + && n.state == JobState::Failed + && let Some(e) = n.error.clone() { return Some(e); } @@ -749,19 +752,35 @@ impl QueueInner { }) } - /// Rebuild the wire [`DagView`] for a DAG from its container metadata + work - /// nodes + per-node runtime. + /// Project a DAG into its wire [`DagView`]: a near-raw view of the + /// container's work nodes, with `Done` nodes excluded. Lifecycle + /// (`state` / `started_at` / `finished_at` / `error`) is read straight + /// off each `hive_jobq::Node`; the client derives the DAG label, roll-up + /// state, and DAG timestamps from the node set. Non-derivable per-node + /// payload (`approval_id`, meta `inputs`) rides the owning node. Returns + /// `None` when every work node is `Done` — a fully-completed DAG drops + /// out of the snapshot entirely (a `Failed` one lingers until aged out). fn dag_view(&self, container: NodeId) -> Option { let meta = self.dag_meta(container)?; - let node_ids = self.subtree(container); - let mut nodes = Vec::with_capacity(node_ids.len()); - let mut started: Vec = Vec::new(); - let mut finished: Vec = Vec::new(); - for &id in &node_ids { + let mut nodes = Vec::new(); + // DAG-level timestamps are taken over *all* subtree nodes (including the + // `Done` ones excluded from the wire) — the client can't derive them + // from a `Done`-filtered node set, so the host computes them here. + let mut started: Vec> = Vec::new(); + let mut finished: Vec> = Vec::new(); + for id in self.subtree(container) { let Some(node) = self.sched.graph().node(id) else { continue; }; - let rt = self.node_rt.get(&id); + if let Some(s) = node.started_at { + started.push(s); + } + if let Some(f) = node.finished_at { + finished.push(f); + } + if node.state == JobState::Done { + continue; + } let deps: Vec = node .deps .iter() @@ -770,51 +789,54 @@ impl QueueInner { Dep::Resource { .. } => None, }) .collect(); - if let Some(s) = rt.and_then(|r| r.started_at) { - started.push(s); - } - if let Some(fin) = rt.and_then(|r| r.finished_at) { - finished.push(fin); - } + // Non-derivable per-node payload rides the node that owns it. + let approval_id = matches!(node.payload, NodeKind::ApprovalDeploy { .. }) + .then_some(meta.approval_id) + .flatten(); + let inputs = if matches!(node.payload, NodeKind::MetaLock { .. }) { + meta.inputs.clone() + } else { + Vec::new() + }; + let has_log = self.node_rt.get(&id).and_then(|r| r.build_log_id).is_some(); nodes.push(NodeView { id: id.get(), agent: node.payload.agent().to_owned(), kind: node.payload.as_str().to_owned(), deps, state: to_wire_state(node.state), - step: rt.and_then(|r| r.step.clone()), - build_log_id: rt.and_then(|r| r.build_log_id), - started_at: rt.and_then(|r| r.started_at), - finished_at: rt.and_then(|r| r.finished_at), - error: rt.and_then(|r| r.error.clone()), + started_at: node.started_at, + finished_at: node.finished_at, + error: node.error.clone(), + approval_id, + inputs, + has_log, }); } + if nodes.is_empty() { + return None; + } let is_terminal = self.dag_is_terminal(container); Some(DagView { id: container.get(), - kind: meta.template, - state: self.dag_rollup(container), source: meta.source, reason: meta.reason.clone(), - enqueued_at: meta.created_at, + created_at: hive_sh4re::wire_time::from_secs(meta.created_at), started_at: started.into_iter().min(), - finished_at: if is_terminal { - finished.into_iter().max() - } else { - None - }, - inputs: meta.inputs.clone(), - approval_id: meta.approval_id, + finished_at: is_terminal.then(|| finished.into_iter().max()).flatten(), nodes, }) } /// When a DAG's work node finishes on `finished_at` — the max over its - /// subtree, for the history cap ordering. + /// subtree (read off the graph `Node`, as unix seconds), for the history + /// cap ordering. fn dag_finished_at(&self, container: NodeId) -> i64 { self.subtree(container) .iter() - .filter_map(|id| self.node_rt.get(id).and_then(|r| r.finished_at)) + .filter_map(|id| self.sched.graph().node(*id)) + .filter_map(|n| n.finished_at) + .map(|t| t.timestamp()) .max() .unwrap_or(0) } diff --git a/hive-c0re/src/job_queue/model.rs b/hive-c0re/src/job_queue/model.rs index bfc66aa0..192ea881 100644 --- a/hive-c0re/src/job_queue/model.rs +++ b/hive-c0re/src/job_queue/model.rs @@ -11,11 +11,62 @@ //! DAG can span agents). See `docs/coordinator.md::Job queue` for the //! full design. -pub use hive_sh4re::jobs::{DagView, NodeId, PermPayload, Source, State, Template}; +pub use hive_sh4re::jobs::{DagView, NodeId, PermPayload, Source, State}; use serde::Serialize; use crate::coordinator::TransientKind; +/// What a DAG *means* — the request-level shape. Internal to the queue now: +/// it drives the terminal-hook dispatch ([`crate::job_queue`]'s `dag_hook`) +/// and the meta-update dedup key, and is **no longer sent on the wire** — the +/// dashboard derives a DAG's label from its node kinds (see `DagView`). The +/// `NodeKind::Dag` container carries it in its payload. +#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, Serialize)] +#[serde(rename_all = "snake_case")] +pub enum Template { + /// Rebuild one agent's container (prebuild → stop → profile-swap → + /// reconcile). + Rebuild, + /// Bump meta flake locks; grows a rebuild subgraph per affected + /// agent into the same DAG on completion. + MetaUpdate, + /// First-deploy spawn (approval-driven). + Spawn, + /// Mechanical stop + converge to `wanted = Up` (a restart). + Restart, + /// Signal → drain → mechanical stop → converge to `wanted = Up` — a + /// graceful restart as one atomic DAG. + GracefulRestart, + /// Perm-file commit followed by the rebuild subgraph. + PermChange, + /// Quiesce the harness, drain, then stop (`wanted = Offline`). + GracefulStop, + /// Converge to `wanted = Up`. + Start, + /// Converge to `wanted = Offline`. + Stop, + /// Boot-time config sweep as one DAG. + Boot, +} + +impl Template { + #[must_use] + pub fn as_str(self) -> &'static str { + match self { + Template::Rebuild => "rebuild", + Template::MetaUpdate => "meta_update", + Template::Spawn => "spawn", + Template::Restart => "restart", + Template::GracefulRestart => "graceful_restart", + Template::PermChange => "perm_change", + Template::GracefulStop => "graceful_stop", + Template::Start => "start", + Template::Stop => "stop", + Template::Boot => "boot", + } + } +} + /// When a dependency edge is considered satisfied. #[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize)] #[serde(rename_all = "snake_case")] diff --git a/hive-c0re/src/job_queue/tests.rs b/hive-c0re/src/job_queue/tests.rs index b385e420..6adac2e9 100644 --- a/hive-c0re/src/job_queue/tests.rs +++ b/hive-c0re/src/job_queue/tests.rs @@ -45,11 +45,14 @@ fn claim_one(q: &JobQueue) -> Claim { } fn state_of(q: &JobQueue, dag_id: u64) -> State { + // A fully-`Done` DAG drops out of the snapshot (its nodes are all + // excluded) — absence is the completion signal, so map it to `Done`. + // Otherwise derive the roll-up from the node set, exactly as every wire + // consumer does. q.snapshot() .iter() .find(|d| d.id == dag_id) - .expect("dag present") - .state + .map_or(State::Done, DagView::rollup_state) } // ---- submit (dedup removed — every submit is a fresh DAG) ---- @@ -664,7 +667,7 @@ fn failed_node_cancels_downstream_but_afterany_reconcile_runs() { q.complete_node(id, reconcile.node_id, Ok(())); let snap = q.snapshot(); let dag = snap.iter().find(|d| d.id == id).expect("dag"); - assert_eq!(dag.state, State::Failed, "roll-up failed"); + assert_eq!(dag.rollup_state(), State::Failed, "roll-up failed"); let by_kind = |k: &str| { dag.nodes .iter() @@ -676,7 +679,12 @@ fn failed_node_cancels_downstream_but_afterany_reconcile_runs() { assert_eq!(by_kind("stop_for_update"), State::Cancelled); assert_eq!(by_kind("swap"), State::Cancelled); assert_eq!(by_kind("post_swap"), State::Cancelled); - assert_eq!(by_kind("reconcile"), State::Done); + // The AfterAny reconcile ran (claimed + completed Ok above) → it's `Done`, + // and `Done` nodes are excluded from the wire, so it's absent here. + assert!( + dag.nodes.iter().all(|n| n.kind != "reconcile"), + "the completed (Done) reconcile is filtered off the wire" + ); assert_eq!( dag.nodes .iter() @@ -872,10 +880,10 @@ fn set_step_only_on_running_and_signals_change() { ); assert!(q.set_step(id, c.node_id, "next phase")); assert!(q.set_step_running(id, "via running lookup")); + // `step` is host-side only now (off the wire); completion clears it + // internally, but there's no wire field to observe — the return-value + // contract above (running-gating + change signalling) is the behaviour. q.complete_node(id, c.node_id, Ok(())); - let snap = q.snapshot(); - let node = &snap.iter().find(|d| d.id == id).expect("dag").nodes[0]; - assert_eq!(node.step, None, "step cleared on completion"); } #[test] @@ -890,9 +898,13 @@ fn set_build_log_id_links_running_node() { assert!(q.set_build_log_id(id, c.node_id, 42)); assert!(q.set_build_log_id_running(id, 43)); q.complete_node(id, c.node_id, Ok(())); - let snap = q.snapshot(); - let node = &snap.iter().find(|d| d.id == id).expect("dag").nodes[0]; - assert_eq!(node.build_log_id, Some(43), "log id survives completion"); + // The log id is fetched by node id (the `GET /api/build-log/` lookup), + // not carried on the wire — it survives completion in the node runtime. + assert_eq!( + q.build_log_id_of(c.node_id.get()), + Some(43), + "log id survives completion" + ); } #[test] @@ -910,9 +922,12 @@ fn history_evicts_old_terminals_per_template() { ), ); let c = claim_one(&q); - // Completing the single work node rolls the container up terminal (its - // inline hook fires off the returned summary — no terminal-hook node). - q.complete_node(id, c.node_id, Ok(())); + // Fail the single work node so the DAG *lingers*: a fully-`Done` DAG + // drops off the wire entirely, but a `Failed` one is retained (+ + // history-capped) so the operator can still triage it. Completing the + // node rolls the container up terminal (its inline hook fires off the + // returned summary — no terminal-hook node). + q.complete_node(id, c.node_id, Err("boom".to_owned())); } // Fresh terminals are inside the grace window: nothing evicts yet, // so a ~1s QueueDag poller can still observe every terminal state diff --git a/hive-c0re/src/server.rs b/hive-c0re/src/server.rs index 4bfb3bf4..d117f977 100644 --- a/hive-c0re/src/server.rs +++ b/hive-c0re/src/server.rs @@ -795,9 +795,14 @@ async fn await_dags(coord: &Arc, ids: &[u64], timeout: std::time::D let deadline = std::time::Instant::now() + timeout; loop { let snap = coord.job_queue.snapshot(); - let pending = ids - .iter() - .any(|id| snap.iter().any(|d| d.id == *id && !d.state.is_terminal())); + // A DAG has settled when it's either gone from the snapshot (fully + // `Done` DAGs drop out) or still present but with every node terminal + // (a `Failed`/`Cancelled` DAG lingers). It's pending only while it has + // a non-terminal node. + let pending = ids.iter().any(|id| { + snap.iter() + .any(|d| d.id == *id && d.nodes.iter().any(|n| !n.state.is_terminal())) + }); if !pending { return; } diff --git a/hive-sh4re/src/jobs.rs b/hive-sh4re/src/jobs.rs index a429b72b..fc854fa0 100644 --- a/hive-sh4re/src/jobs.rs +++ b/hive-sh4re/src/jobs.rs @@ -6,68 +6,9 @@ //! live in `hive-c0re::job_queue`; these are the serialized views it //! produces. Semantics: `docs/coordinator.md::Job queue`. +use chrono::{DateTime, Utc}; use serde::{Deserialize, Serialize}; -/// What a DAG *means* — the request-level shape. Wire strings match -/// the pre-DAG queue's `kind` values so dashboards key off the same -/// tags. -#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, Serialize, Deserialize)] -#[serde(rename_all = "snake_case")] -pub enum Template { - /// Rebuild one agent's container (prebuild → stop → profile-swap → - /// reconcile). - Rebuild, - /// Bump meta flake locks; grows a rebuild subgraph per affected - /// agent into the same DAG on completion. - MetaUpdate, - /// First-deploy spawn (approval-driven). - Spawn, - /// Reserved for a future destroy integration. - Destroy, - /// Mechanical stop + converge to `wanted = Up` (a restart). - Restart, - /// Signal → drain → mechanical stop → converge to `wanted = Up` — a - /// graceful restart as one atomic DAG (drains the harness before the - /// stop, same as `GracefulStop`, but then reconciles back up instead - /// of staying down). - GracefulRestart, - /// Perm-file commit followed by the rebuild subgraph. - PermChange, - /// Quiesce the harness, drain, then stop (`wanted = Offline`). - GracefulStop, - /// Converge to `wanted = Up`. - Start, - /// Converge to `wanted = Offline`. - Stop, - /// Bare converge of observed power state to the persisted intent - /// (boot reconcile). - Reconcile, - /// Boot-time config sweep as one DAG: a hyperhive lock bump that grows - /// a rebuild subgraph per stale agent, plus a `Reconcile` per drifted - /// agent — all in a single DAG (no anchor node, no child DAGs). - Boot, -} - -impl Template { - #[must_use] - pub fn as_str(self) -> &'static str { - match self { - Template::Rebuild => "rebuild", - Template::MetaUpdate => "meta_update", - Template::Spawn => "spawn", - Template::Destroy => "destroy", - Template::Restart => "restart", - Template::GracefulRestart => "graceful_restart", - Template::PermChange => "perm_change", - Template::GracefulStop => "graceful_stop", - Template::Start => "start", - Template::Stop => "stop", - Template::Reconcile => "reconcile", - Template::Boot => "boot", - } - } -} - /// Where the submit request originated — drives the "why" chip on the /// dashboard. #[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)] @@ -140,9 +81,12 @@ pub enum PermPayload { /// so the widening from the old dag-local `u32` is transparent. pub type NodeId = u64; -/// One node of a queued DAG, as serialized. Step labels, build-log -/// links, errors, and timestamps are per-node; the DAG-level `state` -/// is a roll-up. +/// One node of a queued DAG, serialized near-raw from the scheduler +/// graph. Lifecycle (`state` / `started_at` / `finished_at` / `error`) +/// comes straight off the `hive_jobq::Node`. The client derives DAG-level +/// roll-ups (label, state, timestamps) from the node set — nothing is +/// rolled up host-side. Build logs are fetched on demand by node id +/// (`GET /api/build-log/`), not carried inline. #[derive(Debug, Clone, Serialize, Deserialize)] pub struct NodeView { pub id: NodeId, @@ -156,44 +100,124 @@ pub struct NodeView { /// `"drain"`, `"write_dropin"`, `"write_perm_file"`, /// `"approval_deploy"`. pub kind: String, - /// Ids of the nodes this one waits for. + /// Ids of the nodes this one waits for. May reference an already-`Done` + /// node that's been filtered out of the wire — the client treats a dep + /// on an absent node as satisfied. #[serde(default)] pub deps: Vec, pub state: State, #[serde(default, skip_serializing_if = "Option::is_none")] - pub step: Option, + pub started_at: Option>, #[serde(default, skip_serializing_if = "Option::is_none")] - pub build_log_id: Option, - #[serde(default, skip_serializing_if = "Option::is_none")] - pub started_at: Option, - #[serde(default, skip_serializing_if = "Option::is_none")] - pub finished_at: Option, + pub finished_at: Option>, #[serde(default, skip_serializing_if = "Option::is_none")] pub error: Option, + /// Approval-queue row id — present only on the `approval_deploy` node. + /// The client links a DAG to its pending approval through this (it is + /// not derivable from the graph, so it rides the node that owns it). + #[serde(default, skip_serializing_if = "Option::is_none")] + pub approval_id: Option, + /// Meta-flake inputs being bumped — present only on the `meta_lock` + /// node. Display-only payload, not derivable from the graph. + #[serde(default, skip_serializing_if = "Vec::is_empty")] + pub inputs: Vec, + /// Whether this node has a captured build log fetchable at + /// `GET /api/build-log/`. Only the nix-heavy nodes that stream build + /// output set one; the client gates its log link on this so lock / noop / + /// store-only nodes don't render a link that 404s. + #[serde(default)] + pub has_log: bool, } -/// A queued/running/recent DAG. `kind` = template string, roll-up -/// `state`; everything per-node appears exactly once, inside `nodes`. -/// There is no DAG-level `agent` — a DAG can span agents, so agent lives -/// on each [`NodeView`]; consumers group nodes by `NodeView::agent`. +/// A queued / running / failed DAG — a thin projection of one container +/// node plus its (non-`Done`) subtree from the scheduler graph. Only +/// non-derivable facts live here: `id`, `source`, `reason`, `created_at`, +/// and the node set. The client derives the card label, roll-up state, and +/// DAG timestamps from `nodes` (per-node `kind` + lifecycle) — nothing is +/// rolled up host-side. There is no DAG-level `agent`: a DAG can span +/// agents, so agent is per-[`NodeView`]; consumers group by `NodeView::agent`. #[derive(Debug, Clone, Serialize, Deserialize)] pub struct DagView { pub id: u64, - /// Template wire string — same values the old `kind` field used. - pub kind: Template, - /// Roll-up: `failed` if any node failed, else `running` / - /// `queued` / `cancelled` / `done`. - pub state: State, pub source: Source, pub reason: String, - pub enqueued_at: i64, + /// When the DAG was enqueued. + pub created_at: DateTime, + /// When the DAG's first node started (min over *all* its nodes) — computed + /// host-side, **not** derived on the client: `Done` nodes are excluded from + /// `nodes` below, so the earliest-started node is usually absent from the + /// wire and the client can't take the min itself. `None` until a node runs. #[serde(default, skip_serializing_if = "Option::is_none")] - pub started_at: Option, + pub started_at: Option>, + /// When the DAG finished (max `finished_at` over all its nodes), set only + /// once the DAG has settled terminal. Host-computed for the same reason as + /// `started_at`. `None` while the DAG is still live. #[serde(default, skip_serializing_if = "Option::is_none")] - pub finished_at: Option, - #[serde(default, skip_serializing_if = "Vec::is_empty")] - pub inputs: Vec, - #[serde(default, skip_serializing_if = "Option::is_none")] - pub approval_id: Option, + pub finished_at: Option>, + /// Nodes of this DAG with `Done` ones excluded. A DAG whose nodes are + /// all `Done` is omitted from the snapshot entirely; a `Failed` DAG + /// lingers until aged out by the history cap. pub nodes: Vec, } + +impl DagView { + /// Roll-up state derived from the node set — the shared derivation every + /// Rust consumer (hivectl, the wait loops, tests) uses so the dashboard's + /// JS render and the host agree: `Failed` if any node failed, else + /// `Running` if any running, else `Queued` if any queued, else + /// `Cancelled` if any cancelled, else `Done`. `Done` nodes are excluded + /// from the wire, so a DAG that is *entirely* done isn't sent at all — + /// its absence from the snapshot is what signals completion. + #[must_use] + pub fn rollup_state(&self) -> State { + let mut any_running = false; + let mut any_queued = false; + let mut any_cancelled = false; + for n in &self.nodes { + match n.state { + State::Failed => return State::Failed, + State::Running => any_running = true, + State::Queued => any_queued = true, + State::Cancelled => any_cancelled = true, + State::Done => {} + } + } + if any_running { + State::Running + } else if any_queued { + State::Queued + } else if any_cancelled { + State::Cancelled + } else { + State::Done + } + } + + /// A short human label for the DAG, derived from its node kinds — the + /// shared replacement for the old `Template` wire string now that the kind + /// isn't sent. Priority-ordered so the most distinctive node wins (an + /// approval deploy reads as "deploy" even though it also rebuilds). Purely + /// cosmetic (progress/queue display); consumers that need exactness inspect + /// the node kinds directly. + #[must_use] + pub fn label(&self) -> &'static str { + let has = |k: &str| self.nodes.iter().any(|n| n.kind == k); + if has("approval_deploy") { + "deploy" + } else if has("meta_lock") { + "meta-update" + } else if has("create") || has("provision") { + "spawn" + } else if has("swap") || has("prebuild") || has("post_swap") { + "rebuild" + } else if has("write_perm_file") { + "perm-change" + } else if has("signal") || has("drain") { + "graceful" + } else if has("set_wanted") || has("stop_for_update") { + "power" + } else { + "reconcile" + } + } +} diff --git a/hivectl/src/dag_progress.rs b/hivectl/src/dag_progress.rs index c726773f..89ff638c 100644 --- a/hivectl/src/dag_progress.rs +++ b/hivectl/src/dag_progress.rs @@ -66,8 +66,8 @@ async fn wait_for_dags_plain(socket: &Path, ids: Vec) -> Result<()> { // 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(), dag_agents(d))); + if d.rollup_state() == hive_sh4re::jobs::State::Failed { + failed.push(format!("{} {}", d.label(), dag_agents(d))); } } else { all_terminal = false; @@ -134,8 +134,8 @@ async fn wait_for_dags_animated(socket: &Path, ids: Vec) -> Result<()> { }); hdr.set_message(format!( "{} {} {} · {}", - state_glyph(d.state), - d.kind.as_str(), + state_glyph(d.rollup_state()), + d.label(), dag_agents(d), fmt_dur(dag_elapsed(d, now)), )); @@ -163,8 +163,8 @@ async fn wait_for_dags_animated(socket: &Path, ids: Vec) -> Result<()> { } } 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(), dag_agents(d))); + if d.rollup_state() == hive_sh4re::jobs::State::Failed { + failed.push(format!("{} {}", d.label(), dag_agents(d))); } } else { all_terminal = false; @@ -224,18 +224,21 @@ fn now_unix() -> i64 { .unwrap_or(0) } -/// Elapsed seconds for a DAG: `started_at` (falling back to `enqueued_at`) -/// through `finished_at` or `now`. +/// Elapsed seconds for a DAG: `started_at` (falling back to `created_at`) +/// through `finished_at` or `now`. The wire carries these as RFC3339 +/// `DateTime`; compare in unix seconds against `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) + let start = d + .started_at + .map_or_else(|| d.created_at.timestamp(), |t| t.timestamp()); + (d.finished_at.map_or(now, |t| t.timestamp()) - 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), + Some(start) => (n.finished_at.map_or(now, |t| t.timestamp()) - start.timestamp()).max(0), None => 0, } } @@ -250,16 +253,11 @@ fn fmt_dur(secs: i64) -> String { } } -/// 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. +/// One animated node line: kind, 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 @@ -292,25 +290,21 @@ fn state_glyph(state: hive_sh4re::jobs::State) -> &'static str { } } -/// 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. +/// One progress line for a DAG: roll-up glyph, derived label, agents, then +/// the node chain — the CLI twin of the dashboard's queue card. Both the +/// roll-up state and the label are derived from the node set (the wire no +/// longer carries them). 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(), + state_glyph(d.rollup_state()), + d.label(), dag_agents(d) ); 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(); @@ -321,76 +315,65 @@ fn render_dag_line(d: &hive_sh4re::jobs::DagView) -> String { #[cfg(test)] mod tests { - use hive_sh4re::jobs::{DagView, NodeView, Source, State, Template}; + use hive_sh4re::jobs::{DagView, NodeView, Source, State}; + use hive_sh4re::wire_time::from_secs; use super::render_dag_line; - fn node(id: u64, agent: &str, kind: &str, state: State, step: Option<&str>) -> NodeView { + fn node(id: u64, agent: &str, kind: &str, state: State) -> NodeView { NodeView { id, agent: agent.to_owned(), 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, + approval_id: None, + inputs: vec![], + has_log: false, } } #[test] - fn render_dag_line_shows_chain_and_running_step() { + fn render_dag_line_shows_chain_with_derived_label_and_state() { + // Roll-up state (Running) + label ("rebuild") are derived from the node + // set — the wire no longer carries them. (`Done` nodes are included + // here to exercise glyph rendering; production filters them off.) let dag = DagView { id: 7, - kind: Template::Rebuild, - state: State::Running, source: Source::Manual, reason: "manual".to_owned(), - enqueued_at: 0, - started_at: Some(1), + created_at: from_secs(0), + started_at: Some(from_secs(1)), finished_at: None, - inputs: vec![], - approval_id: None, nodes: vec![ - node(0, "alice", "prebuild", State::Done, None), - node(1, "alice", "stop_for_update", State::Done, None), - node( - 2, - "alice", - "swap", - State::Running, - Some("nixos-container update"), - ), - node(3, "alice", "reconcile", State::Queued, None), + node(0, "alice", "prebuild", State::Done), + node(1, "alice", "stop_for_update", State::Done), + node(2, "alice", "swap", State::Running), + node(3, "alice", "reconcile", State::Queued), ], }; 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.contains("✔ prebuild → ✔ stop_for_update → ▶ swap → ⏸ reconcile"), "{line}" ); } #[test] fn render_dag_line_surfaces_first_node_error() { - let mut failed = node(0, "bob", "prebuild", State::Failed, None); + let mut failed = node(0, "bob", "prebuild", State::Failed); failed.error = Some("nix build exploded".to_owned()); let dag = DagView { id: 8, - kind: Template::Rebuild, - state: State::Failed, source: Source::Manual, reason: "manual".to_owned(), - enqueued_at: 0, - started_at: Some(1), - finished_at: Some(2), - inputs: vec![], - approval_id: None, + created_at: from_secs(0), + started_at: Some(from_secs(1)), + finished_at: Some(from_secs(2)), nodes: vec![failed], }; let line = render_dag_line(&dag);