From 2a59f2f5fcccacfa6e4c4930fb7026e340810eab Mon Sep 17 00:00:00 2001 From: atlas Date: Tue, 14 Jul 2026 21:42:26 +0200 Subject: [PATCH 1/3] refactor(#2441): move agent field from DAG onto Node; drop dedup MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Agent was a single field on Dag/DagSpec, making a DAG structurally one-agent — a multi-agent op could only ever be N separate DAGs. Move it onto Node/NodeSpec (and the NodeView wire type), drop it from Dag/DagSpec (and DagView): a DAG can now span agents. - lifecycle lease keys on the node's agent, still globally exclusive per agent across all DAGs (Inner.leases unchanged in shape). A DAG holds one lease per distinct agent it touches; settle() frees each at DAG-terminal (per-agent-subgraph early release is a follow-up, only observable with multi-agent DAGs). - transient guard keyed (dag_id, agent); cancel-revert + Rebuilt events walk TerminalDag.agents. - submit-time dedup removed (a multi-agent DAG has no single agent to key on); every submit enqueues a fresh DAG. Whether dedup needs reintroducing is tracked in a follow-up sub-issue. - templates gain a node(agent, kind, deps) helper stamping the agent onto every node; meta templates stamp "hyperhive". Templates stay single-agent in this PR — behaviour is unchanged, only the representation + wire shape. Multi-agent DAG emission (restart/restart-all/ broad stop+start as one DAG) and the SetWanted-as-a-node change are follow-ups off #2439. --- hive-c0re/src/job_queue/exec.rs | 53 +++++---- hive-c0re/src/job_queue/mod.rs | 105 +++++++++-------- hive-c0re/src/job_queue/model.rs | 65 +++++----- hive-c0re/src/job_queue/scheduler.rs | 18 ++- hive-c0re/src/job_queue/templates.rs | 170 ++++++++++----------------- hive-c0re/src/job_queue/tests.rs | 155 ++++-------------------- hive-sh4re/src/jobs.rs | 13 +- 7 files changed, 223 insertions(+), 356 deletions(-) diff --git a/hive-c0re/src/job_queue/exec.rs b/hive-c0re/src/job_queue/exec.rs index 8ccbc6d9..17749b20 100644 --- a/hive-c0re/src/job_queue/exec.rs +++ b/hive-c0re/src/job_queue/exec.rs @@ -514,12 +514,17 @@ pub(super) async fn on_dag_terminal(coord: &Arc, terminal: &Termina | Template::GracefulRestart ) { - let running = crate::lifecycle::is_running(&terminal.agent).await; - if let Err(e) = coord - .power - .set(&terminal.agent, crate::power::Wanted::from_running(running)) - { - tracing::warn!(agent = %terminal.agent, error = ?e, "agent_power: cancel revert failed"); + // Revert each targeted agent's power intent to its observed state — + // the operator's cancel means "don't do it". Single-agent power-op + // DAGs have one agent here. + for agent in &terminal.agents { + let running = crate::lifecycle::is_running(agent).await; + if let Err(e) = coord + .power + .set(agent, crate::power::Wanted::from_running(running)) + { + tracing::warn!(%agent, error = ?e, "agent_power: cancel revert failed"); + } } } if terminal.approval_id.is_some() { @@ -527,22 +532,26 @@ pub(super) async fn on_dag_terminal(coord: &Arc, terminal: &Termina return; } if matches!(terminal.template, Template::Rebuild | Template::PermChange) { - match terminal.state { - State::Done => coord.notify_manager(&hive_sh4re::HelperEvent::Rebuilt { - agent: terminal.agent.clone(), - ok: true, - note: None, - sha: None, - tag: None, - }), - State::Failed => coord.notify_manager(&hive_sh4re::HelperEvent::Rebuilt { - agent: terminal.agent.clone(), - ok: false, - note: terminal.error.clone(), - sha: None, - tag: None, - }), - _ => {} + // Rebuild / PermChange are single-agent; emit one `Rebuilt` per + // targeted agent (exactly one today). + for agent in &terminal.agents { + match terminal.state { + State::Done => coord.notify_manager(&hive_sh4re::HelperEvent::Rebuilt { + agent: agent.clone(), + ok: true, + note: None, + sha: None, + tag: None, + }), + State::Failed => coord.notify_manager(&hive_sh4re::HelperEvent::Rebuilt { + agent: agent.clone(), + ok: false, + note: terminal.error.clone(), + sha: None, + tag: None, + }), + _ => {} + } } } } diff --git a/hive-c0re/src/job_queue/mod.rs b/hive-c0re/src/job_queue/mod.rs index 27a968ee..db0c3e1e 100644 --- a/hive-c0re/src/job_queue/mod.rs +++ b/hive-c0re/src/job_queue/mod.rs @@ -7,10 +7,12 @@ //! Concurrency is gated by two resource classes: //! 1. **Build slots** — N permits (`services.hyperhive.c0re.buildSlots`, //! default 1) held by nix-heavy nodes for the node's duration. -//! 2. **Per-agent lifecycle lease** — DAG-scoped: acquired before the -//! DAG's first container-affecting node runs, held until the DAG is -//! terminal, so two lifecycle DAGs for one agent never interleave -//! their container ops. +//! 2. **Per-agent lifecycle lease** — keyed on the *node's* agent and +//! globally exclusive per agent across all DAGs: acquired before a +//! container-affecting node runs, held (by the owning DAG) until no +//! live node of that DAG still targets the agent, so two DAGs never +//! interleave container ops on the same agent. A DAG spanning multiple +//! agents holds one lease per agent it touches. //! //! The meta *repo* is serialized by `meta::META_LOCK` inside the //! executors themselves. Per-agent power *intent* (`wanted`) lives in @@ -57,15 +59,18 @@ pub struct Claim { pub dag_id: u64, pub node_id: NodeId, pub kind: NodeKind, + /// The agent this node targets (its own, not a DAG-level field). The + /// executor operates on this agent's container; the lease is keyed on + /// it. pub agent: String, pub template: Template, pub source: Source, pub approval_id: Option, pub inputs: Vec, pub perm_payload: Option, - /// True when claiming this node acquired the DAG's agent lease — - /// the scheduler creates the DAG-scoped transient guard on this - /// edge. + /// True when claiming this node newly acquired its agent's lease — + /// the scheduler creates the per-`(dag, agent)` transient guard on + /// this edge. pub lease_acquired: bool, /// Transient pill kind for the lease window (from the spec). pub transient: Option, @@ -78,7 +83,10 @@ pub struct Claim { pub struct TerminalDag { pub dag_id: u64, pub template: Template, - pub agent: String, + /// Distinct agents this DAG's nodes targeted (one for a single-agent + /// DAG). The cancel-revert hook walks these to snap each agent's + /// power intent back on a cancelled power-op DAG. + pub agents: Vec, pub approval_id: Option, pub state: State, /// First failed node's error when `state == Failed`. @@ -128,27 +136,17 @@ impl JobQueue { } } - /// Submit a DAG. Validates the spec (cycle rejection) and dedups - /// against non-started DAGs; returns the DAG id (newly-allocated, - /// or the existing DAG's id with the new reason appended). + /// Submit a DAG. Validates the spec (cycle rejection) and returns the + /// newly-allocated DAG id. /// - /// Dedup: a DAG whose roll-up is still `Queued` (no node started) - /// with the same `(template, agent, parent_id, approval_id)` — plus - /// `inputs` for `MetaUpdate` and the perm-type discriminant for - /// `PermChange` — swallows the repeat. `parent_id` is part of the - /// key so a meta-update cascade rebuild never collapses into a - /// standalone or sweep rebuild. Running / terminal DAGs never - /// dedup — operators are free to re-queue. + /// Submit-time dedup was removed with the agent-per-node refactor + /// (a multi-agent DAG has no single agent to key a dedup on) — every + /// submit now enqueues a fresh DAG. Whether any dedup needs + /// reintroducing (and in what form) is tracked as a follow-up; see the + /// dedup re-evaluation issue. pub fn submit(&self, spec: DagSpec) -> anyhow::Result { templates::validate(&spec)?; let mut inner = self.inner.lock().expect("job_queue mutex poisoned"); - if let Some(existing) = Self::dedup_target(&mut inner, &spec) { - if !existing.reason.contains(&spec.reason) { - use std::fmt::Write as _; - let _ = write!(existing.reason, "\nalso requested by: {}", spec.reason); - } - return Ok(existing.id); - } let id = Self::push_dag(&mut inner, spec); drop(inner); self.notify.notify_one(); @@ -156,8 +154,8 @@ impl JobQueue { } /// Append fan-out children under a parent DAG (meta-update / sweep - /// cascade). Applies the same dedup as [`Self::submit`]; returns - /// the child ids actually created or coalesced into. + /// cascade); returns the child ids created. No dedup (see + /// [`Self::submit`]). pub fn append_children(&self, specs: Vec) -> Vec { let mut ids = Vec::with_capacity(specs.len()); for spec in specs { @@ -183,9 +181,14 @@ impl JobQueue { pub fn append_node(&self, dag_id: u64, kind: NodeKind, dep_on: NodeId) -> Option { let mut inner = self.inner.lock().expect("job_queue mutex poisoned"); let dag = inner.dags.iter_mut().find(|d| d.id == dag_id)?; + // The appended sub-step targets the same agent as the node that + // emitted it (a `Reconcile` fanning out its `Start`/`Stop` acts on + // the same container), so inherit `dep_on`'s agent. + let agent = dag.node(dep_on)?.agent.clone(); let new_id: NodeId = u32::try_from(dag.nodes.len()).unwrap_or(u32::MAX); dag.nodes.push(Node { id: new_id, + agent, kind, deps: vec![model::Dep { on: dep_on, @@ -203,21 +206,6 @@ impl JobQueue { Some(new_id) } - fn dedup_target<'a>(inner: &'a mut Inner, spec: &DagSpec) -> Option<&'a mut Dag> { - inner.dags.iter_mut().find(|d| { - d.rollup() == State::Queued - && d.template == spec.template - && d.agent == spec.agent - && d.parent_id == spec.parent_id - && d.approval_id == spec.approval_id - && (d.template != Template::MetaUpdate || d.inputs == spec.inputs) - && model::perm_payload_same_type( - d.perm_payload.as_ref(), - spec.perm_payload.as_ref(), - ) - }) - } - fn push_dag(inner: &mut Inner, spec: DagSpec) -> u64 { inner.next_id += 1; let id = inner.next_id; @@ -227,6 +215,7 @@ impl JobQueue { .enumerate() .map(|(i, n)| Node { id: u32::try_from(i).unwrap_or(u32::MAX), + agent: n.agent, kind: n.kind, deps: n.deps, state: State::Queued, @@ -240,7 +229,6 @@ impl JobQueue { inner.dags.push_back(Dag { id, template: spec.template, - agent: spec.agent, source: spec.source, reason: spec.reason, parent_id: spec.parent_id, @@ -282,16 +270,23 @@ impl JobQueue { let dag = &inner.dags[di]; let node = dag.node(node_id).expect("node id from same dag"); let needs_slot = node.kind.needs_build_slot(); + let needs_lease = node.kind.needs_lease(); + // The lifecycle lease is keyed on the *node's* agent, held + // by this DAG (dag_id) — still globally exclusive per agent + // across all DAGs. A multi-agent DAG acquires one lease per + // agent it touches; each is released in `settle` once no + // live node of this DAG still targets that agent. + let node_agent = node.agent.clone(); if needs_slot && inner.slots_used >= inner.build_slots { continue; } let mut lease_acquired = false; - if node.kind.needs_lease() { - match inner.leases.get(dag.agent.as_str()) { + if needs_lease { + match inner.leases.get(node_agent.as_str()) { Some(&holder) if holder != dag_id => continue, Some(_) => {} None => { - inner.leases.insert(dag.agent.clone(), dag_id); + inner.leases.insert(node_agent.clone(), dag_id); lease_acquired = true; } } @@ -304,7 +299,7 @@ impl JobQueue { dag_id, node_id, kind: dag.node(node_id).expect("node").kind.clone(), - agent: dag.agent.clone(), + agent: node_agent, template: dag.template, source: dag.source, approval_id: dag.approval_id, @@ -425,13 +420,23 @@ impl JobQueue { continue; } dag.terminal_reported = true; - if inner.leases.get(dag.agent.as_str()) == Some(&dag.id) { - freed.push(dag.agent.clone()); + // Free every agent-lease this DAG holds (one per distinct + // agent it touched). Single-agent DAGs release their one lease; + // a future multi-agent DAG releases all of them at terminal. + // (Per-agent early release — freeing an agent's lease the moment + // that agent's subgraph is terminal rather than at whole-DAG + // terminal — is a refinement for the multi-agent-emission + // follow-up, where it actually matters.) + let agents = dag.agents(); + for agent in &agents { + if inner.leases.get(agent.as_str()) == Some(&dag.id) { + freed.push(agent.clone()); + } } reports.push(TerminalDag { dag_id: dag.id, template: dag.template, - agent: dag.agent.clone(), + agents, approval_id: dag.approval_id, state: dag.rollup(), error: dag.first_error().map(str::to_owned), diff --git a/hive-c0re/src/job_queue/model.rs b/hive-c0re/src/job_queue/model.rs index 7e474785..bf0e4c95 100644 --- a/hive-c0re/src/job_queue/model.rs +++ b/hive-c0re/src/job_queue/model.rs @@ -5,33 +5,15 @@ //! `hive_sh4re::jobs` (wire types belong to the shared crate) and are //! re-exported here for the queue's internal use. //! -//! Two levels: the **DAG** is the unit of dedup / cancel / -//! approval-resolution and the dashboard group; the **node** is the -//! unit of scheduling / execution / build-log / step label. See -//! `docs/coordinator.md::Job queue` for the full design. +//! Two levels: the **DAG** is the unit of cancel / approval-resolution +//! and the dashboard group; the **node** is the unit of scheduling / +//! execution / build-log / step label, and carries its own `agent` (a +//! DAG can span agents). See `docs/coordinator.md::Job queue` for the +//! full design. pub use hive_sh4re::jobs::{DagView, NodeId, NodeView, PermPayload, Source, State, Template}; use serde::Serialize; -/// Dedup compares the perm *type*, not the value — a tool-groups -/// change and a capabilities change for the same agent are distinct -/// operations that must not collapse. -pub(super) fn perm_payload_same_type(a: Option<&PermPayload>, b: Option<&PermPayload>) -> bool { - matches!( - (a, b), - ( - Some(PermPayload::ToolGroups { .. }), - Some(PermPayload::ToolGroups { .. }) - ) | ( - Some(PermPayload::Capabilities { .. }), - Some(PermPayload::Capabilities { .. }) - ) | ( - Some(PermPayload::Combined { .. }), - Some(PermPayload::Combined { .. }) - ) | (None, None) - ) -} - /// When a dependency edge is considered satisfied. #[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize)] #[serde(rename_all = "snake_case")] @@ -203,6 +185,11 @@ impl NodeKind { #[derive(Debug, Clone)] pub struct Node { pub id: NodeId, + /// The agent this node's work targets. Per-node so a single DAG can + /// span agents (e.g. a hive-wide restart); the lifecycle lease is + /// acquired against *this* agent (still globally exclusive per agent + /// across all DAGs). `"hyperhive"` for meta-level nodes. + pub agent: String, pub kind: NodeKind, pub deps: Vec, pub state: State, @@ -221,19 +208,23 @@ pub struct Node { /// Submit-time spec for one node. #[derive(Debug, Clone)] pub struct NodeSpec { + /// The agent this node targets (see [`Node::agent`]). Built by the + /// `templates.rs` `node` helper, which stamps the template's agent + /// onto every node. + pub agent: String, pub kind: NodeKind, pub deps: Vec, } /// Submit-time spec for a whole DAG. Built by `templates.rs`; validated -/// (cycle rejection) and dedup'd by `JobQueue::submit`. +/// (cycle rejection) by `JobQueue::submit`. No DAG-level `agent` — every +/// node carries its own (a DAG can span agents), and the queue derives +/// per-agent leasing from [`NodeSpec::agent`]. #[derive(Debug, Clone)] pub struct DagSpec { pub template: Template, - /// Primary target agent, or `"hyperhive"` for meta-level DAGs. - pub agent: String, pub source: Source, - /// Free-form "why"; dedup appends "also requested by …" lines. + /// Free-form "why". pub reason: String, /// Cascade grouping (meta-update / sweep children). pub parent_id: Option, @@ -250,12 +241,13 @@ pub struct DagSpec { pub nodes: Vec, } -/// A live DAG in the queue. +/// A live DAG in the queue. No DAG-level `agent`: agent is per-[`Node`], +/// so a DAG can span agents. Per-agent leasing is derived from the +/// nodes' agents. #[derive(Debug, Clone)] pub struct Dag { pub id: u64, pub template: Template, - pub agent: String, pub source: Source, pub reason: String, pub parent_id: Option, @@ -319,6 +311,19 @@ impl Dag { pub fn node_mut(&mut self, id: NodeId) -> Option<&mut Node> { self.nodes.iter_mut().find(|n| n.id == id) } + + /// Distinct agents this DAG's nodes target, in first-seen order. + /// Used for per-agent lease release and the terminal cancel-revert — + /// a single-agent DAG yields one, a multi-agent DAG yields several. + pub fn agents(&self) -> Vec { + let mut seen: Vec = Vec::new(); + for n in &self.nodes { + if !seen.iter().any(|a| a == &n.agent) { + seen.push(n.agent.clone()); + } + } + seen + } } impl Dag { @@ -331,7 +336,6 @@ impl Dag { }; DagView { id: self.id, - agent: self.agent.clone(), kind: self.template, state: self.rollup(), source: self.source, @@ -348,6 +352,7 @@ impl Dag { .iter() .map(|n| NodeView { id: n.id, + agent: n.agent.clone(), kind: n.kind.as_str().to_owned(), deps: n.deps.iter().map(|d| d.on).collect(), state: n.state, diff --git a/hive-c0re/src/job_queue/scheduler.rs b/hive-c0re/src/job_queue/scheduler.rs index 230f9d30..eb03e98c 100644 --- a/hive-c0re/src/job_queue/scheduler.rs +++ b/hive-c0re/src/job_queue/scheduler.rs @@ -35,8 +35,10 @@ struct NodeDone { pub async fn run_worker(coord: Arc) { let mut shutdown = coord.shutdown_rx(); let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel::(); - // DAG id → transient guard held for the lease window. - let mut transients: HashMap = HashMap::new(); + // (DAG id, agent) → transient guard held for that agent's lease + // window. Keyed per-agent so a multi-agent DAG shows one transient + // pill per agent it touches. + let mut transients: HashMap<(u64, String), crate::coordinator::TransientGuard> = HashMap::new(); loop { // Terminal roll-ups can appear without a node completion — // the cancel surfaces settle DAGs directly and wake this loop @@ -49,7 +51,10 @@ pub async fn run_worker(coord: Arc) { if claim.lease_acquired && let Some(kind) = claim.transient { - transients.insert(claim.dag_id, coord.transient_guard(&claim.agent, kind)); + transients.insert( + (claim.dag_id, claim.agent.clone()), + coord.transient_guard(&claim.agent, kind), + ); } tracing::info!( dag = claim.dag_id, @@ -88,7 +93,7 @@ pub async fn run_worker(coord: Arc) { async fn handle_completion( coord: &Arc, - transients: &mut HashMap, + transients: &mut HashMap<(u64, String), crate::coordinator::TransientGuard>, done: NodeDone, ) { let NodeDone { claim, result } = done; @@ -143,10 +148,11 @@ async fn handle_completion( /// `Rebuilt` events, cancelled-power-op intent revert). async fn process_terminals( coord: &Arc, - transients: &mut HashMap, + transients: &mut HashMap<(u64, String), crate::coordinator::TransientGuard>, ) { for terminal in coord.job_queue.drain_terminal() { - transients.remove(&terminal.dag_id); + // Drop every per-agent transient guard this DAG held. + transients.retain(|(dag_id, _), _| *dag_id != terminal.dag_id); exec::on_dag_terminal(coord, &terminal).await; } } diff --git a/hive-c0re/src/job_queue/templates.rs b/hive-c0re/src/job_queue/templates.rs index 29ffa8d3..48c3086c 100644 --- a/hive-c0re/src/job_queue/templates.rs +++ b/hive-c0re/src/job_queue/templates.rs @@ -3,6 +3,11 @@ //! confined to this validation; the runtime store stays the plain //! `Vec` + `deps`). //! +//! Every node carries its own `agent` (there is no DAG-level agent) — the +//! `node` helper stamps the template's agent onto each. Today's templates +//! are single-agent (every node shares one agent); a future multi-agent +//! template would stamp different agents per subgraph. +//! //! ```text //! rebuild(a): Prebuild(a) → StopForUpdate(a) → Swap(a) →(any) Reconcile(a) //! graceful-stop(a): [wanted=Offline] Signal(a) → Drain(a) → Reconcile(a) @@ -29,35 +34,42 @@ fn after_ok(on: u32) -> Vec { }] } +/// Build one node targeting `agent`. The single place templates stamp a +/// node's agent, so a whole template is single-agent by passing the same +/// `agent` to every call. +fn node(agent: &str, kind: NodeKind, deps: Vec) -> NodeSpec { + NodeSpec { + agent: agent.to_owned(), + kind, + deps, + } +} + /// The rebuild node chain. `Reconcile` deps on `Swap` with `AfterAny`: /// it must run even when the profile swap failed, so a previously-up /// agent comes back on its old config (today's recovery-start). This /// is the only `AfterAny` edge in v1. -fn rebuild_nodes(relock: bool, base: u32) -> Vec { +fn rebuild_nodes(agent: &str, relock: bool, base: u32) -> Vec { vec![ - NodeSpec { - kind: NodeKind::Prebuild { relock }, - deps: if base == 0 { + node( + agent, + NodeKind::Prebuild { relock }, + if base == 0 { Vec::new() } else { after_ok(base - 1) }, - }, - NodeSpec { - kind: NodeKind::StopForUpdate, - deps: after_ok(base), - }, - NodeSpec { - kind: NodeKind::Swap, - deps: after_ok(base + 1), - }, - NodeSpec { - kind: NodeKind::Reconcile, - deps: vec![Dep { + ), + node(agent, NodeKind::StopForUpdate, after_ok(base)), + node(agent, NodeKind::Swap, after_ok(base + 1)), + node( + agent, + NodeKind::Reconcile, + vec![Dep { on: base + 2, when: DepWhen::AfterAny, }], - }, + ), ] } @@ -75,7 +87,6 @@ pub fn rebuild( ) -> DagSpec { DagSpec { template: Template::Rebuild, - agent: agent.to_owned(), source, reason, parent_id, @@ -83,7 +94,7 @@ pub fn rebuild( inputs: Vec::new(), perm_payload: None, transient: Some(TransientKind::Rebuilding), - nodes: rebuild_nodes(relock, 0), + nodes: rebuild_nodes(agent, relock, 0), } } @@ -93,7 +104,6 @@ pub fn rebuild( pub fn approval_deploy(agent: &str, approval_id: i64, reason: String) -> DagSpec { DagSpec { template: Template::Rebuild, - agent: agent.to_owned(), source: Source::Approval, reason, parent_id: None, @@ -101,10 +111,7 @@ pub fn approval_deploy(agent: &str, approval_id: i64, reason: String) -> DagSpec inputs: Vec::new(), perm_payload: None, transient: Some(TransientKind::Rebuilding), - nodes: vec![NodeSpec { - kind: NodeKind::ApprovalDeploy, - deps: Vec::new(), - }], + nodes: vec![node(agent, NodeKind::ApprovalDeploy, Vec::new())], } } @@ -117,7 +124,6 @@ pub fn approval_deploy(agent: &str, approval_id: i64, reason: String) -> DagSpec pub fn graceful_stop(agent: &str, source: Source, reason: String) -> DagSpec { DagSpec { template: Template::GracefulStop, - agent: agent.to_owned(), source, reason, parent_id: None, @@ -126,18 +132,9 @@ pub fn graceful_stop(agent: &str, source: Source, reason: String) -> DagSpec { perm_payload: None, transient: Some(TransientKind::Stopping), nodes: vec![ - NodeSpec { - kind: NodeKind::Signal, - deps: Vec::new(), - }, - NodeSpec { - kind: NodeKind::Drain, - deps: after_ok(0), - }, - NodeSpec { - kind: NodeKind::Reconcile, - deps: after_ok(1), - }, + node(agent, NodeKind::Signal, Vec::new()), + node(agent, NodeKind::Drain, after_ok(0)), + node(agent, NodeKind::Reconcile, after_ok(1)), ], } } @@ -148,7 +145,6 @@ pub fn graceful_stop(agent: &str, source: Source, reason: String) -> DagSpec { pub fn restart(agent: &str, source: Source, reason: String) -> DagSpec { DagSpec { template: Template::Restart, - agent: agent.to_owned(), source, reason, parent_id: None, @@ -157,14 +153,8 @@ pub fn restart(agent: &str, source: Source, reason: String) -> DagSpec { perm_payload: None, transient: Some(TransientKind::Restarting), nodes: vec![ - NodeSpec { - kind: NodeKind::StopForUpdate, - deps: Vec::new(), - }, - NodeSpec { - kind: NodeKind::Reconcile, - deps: after_ok(0), - }, + node(agent, NodeKind::StopForUpdate, Vec::new()), + node(agent, NodeKind::Reconcile, after_ok(0)), ], } } @@ -180,7 +170,6 @@ pub fn restart(agent: &str, source: Source, reason: String) -> DagSpec { pub fn graceful_restart(agent: &str, source: Source, reason: String) -> DagSpec { DagSpec { template: Template::GracefulRestart, - agent: agent.to_owned(), source, reason, parent_id: None, @@ -189,22 +178,10 @@ pub fn graceful_restart(agent: &str, source: Source, reason: String) -> DagSpec perm_payload: None, transient: Some(TransientKind::Restarting), nodes: vec![ - NodeSpec { - kind: NodeKind::Signal, - deps: Vec::new(), - }, - NodeSpec { - kind: NodeKind::Drain, - deps: after_ok(0), - }, - NodeSpec { - kind: NodeKind::StopForUpdate, - deps: after_ok(1), - }, - NodeSpec { - kind: NodeKind::Reconcile, - deps: after_ok(2), - }, + node(agent, NodeKind::Signal, Vec::new()), + node(agent, NodeKind::Drain, after_ok(0)), + node(agent, NodeKind::StopForUpdate, after_ok(1)), + node(agent, NodeKind::Reconcile, after_ok(2)), ], } } @@ -220,7 +197,6 @@ pub fn reconcile_only( ) -> DagSpec { DagSpec { template, - agent: agent.to_owned(), source, reason, parent_id: None, @@ -228,10 +204,7 @@ pub fn reconcile_only( inputs: Vec::new(), perm_payload: None, transient, - nodes: vec![NodeSpec { - kind: NodeKind::Reconcile, - deps: Vec::new(), - }], + nodes: vec![node(agent, NodeKind::Reconcile, Vec::new())], } } @@ -242,7 +215,6 @@ pub fn reconcile_only( pub fn spawn(agent: &str, approval_id: i64, reason: String) -> DagSpec { DagSpec { template: Template::Spawn, - agent: agent.to_owned(), source: Source::Approval, reason, parent_id: None, @@ -251,22 +223,10 @@ pub fn spawn(agent: &str, approval_id: i64, reason: String) -> DagSpec { perm_payload: None, transient: Some(TransientKind::Spawning), nodes: vec![ - NodeSpec { - kind: NodeKind::Provision, - deps: Vec::new(), - }, - NodeSpec { - kind: NodeKind::Create, - deps: after_ok(0), - }, - NodeSpec { - kind: NodeKind::WriteDropin, - deps: after_ok(1), - }, - NodeSpec { - kind: NodeKind::Reconcile, - deps: after_ok(2), - }, + node(agent, NodeKind::Provision, Vec::new()), + node(agent, NodeKind::Create, after_ok(0)), + node(agent, NodeKind::WriteDropin, after_ok(1)), + node(agent, NodeKind::Reconcile, after_ok(2)), ], } } @@ -275,14 +235,10 @@ pub fn spawn(agent: &str, approval_id: i64, reason: String) -> DagSpec { /// the updated `HIVE_TOOL_GROUPS` / `HIVE_CAPABILITIES` env var takes /// effect in the container. pub fn perm_change(agent: &str, source: Source, reason: String, payload: PermPayload) -> DagSpec { - let mut nodes = vec![NodeSpec { - kind: NodeKind::WritePermFile, - deps: Vec::new(), - }]; - nodes.extend(rebuild_nodes(true, 1)); + let mut nodes = vec![node(agent, NodeKind::WritePermFile, Vec::new())]; + nodes.extend(rebuild_nodes(agent, true, 1)); DagSpec { template: Template::PermChange, - agent: agent.to_owned(), source, reason, parent_id: None, @@ -306,7 +262,6 @@ pub fn meta_update( ) -> DagSpec { DagSpec { template: Template::MetaUpdate, - agent: "hyperhive".to_owned(), source, reason, parent_id: None, @@ -314,19 +269,17 @@ pub fn meta_update( inputs, perm_payload: None, transient: None, - nodes: vec![NodeSpec { - kind: NodeKind::MetaLock { + nodes: vec![node( + "hyperhive", + NodeKind::MetaLock { sweep: false, fanout: None, }, - deps: Vec::new(), - }], + Vec::new(), + )], } } -/// Boot-time sweep parent: bump meta's hyperhive input (non-fatal), -/// then fan out `Rebuild` children for the precomputed stale agent -/// list (topology-sorted by the caller). /// Boot-time root anchor DAG: a single [`NodeKind::Noop`] node that groups /// this boot's `StartupSweep` + per-agent `Reconcile` child DAGs (linked via /// `parent_id`) into one tree so the dashboard renders the boot as one entry. @@ -336,7 +289,6 @@ pub fn meta_update( pub fn boot_root(reason: String) -> DagSpec { DagSpec { template: Template::Boot, - agent: "hyperhive".to_owned(), source: Source::AutoUpdate, reason, parent_id: None, @@ -344,17 +296,16 @@ pub fn boot_root(reason: String) -> DagSpec { inputs: Vec::new(), perm_payload: None, transient: None, - nodes: vec![NodeSpec { - kind: NodeKind::Noop, - deps: Vec::new(), - }], + nodes: vec![node("hyperhive", NodeKind::Noop, Vec::new())], } } +/// Boot-time sweep parent: bump meta's hyperhive input (non-fatal), +/// then fan out `Rebuild` children for the precomputed stale agent +/// list (topology-sorted by the caller). pub fn startup_sweep(reason: String, stale_agents: Vec) -> DagSpec { DagSpec { template: Template::StartupSweep, - agent: "hyperhive".to_owned(), source: Source::AutoUpdate, reason, parent_id: None, @@ -362,13 +313,14 @@ pub fn startup_sweep(reason: String, stale_agents: Vec) -> DagSpec { inputs: Vec::new(), perm_payload: None, transient: None, - nodes: vec![NodeSpec { - kind: NodeKind::MetaLock { + nodes: vec![node( + "hyperhive", + NodeKind::MetaLock { sweep: true, fanout: Some(stale_agents), }, - deps: Vec::new(), - }], + Vec::new(), + )], } } diff --git a/hive-c0re/src/job_queue/tests.rs b/hive-c0re/src/job_queue/tests.rs index 958040da..b969db92 100644 --- a/hive-c0re/src/job_queue/tests.rs +++ b/hive-c0re/src/job_queue/tests.rs @@ -1,4 +1,4 @@ -//! Queue-core unit tests: dedup, cycle rejection, resource +//! Queue-core unit tests: submit / no-dedup, cycle rejection, resource //! serialization (build slots / per-agent leases), lease-exempt //! overlap, FIFO fairness, cancel semantics, `AfterAny` failure //! routing, fan-out, and history retention. All synchronous — the @@ -35,7 +35,7 @@ fn state_of(q: &JobQueue, dag_id: u64) -> State { .state } -// ---- submit / dedup ---- +// ---- submit (dedup removed — every submit is a fresh DAG) ---- #[test] fn submit_assigns_distinct_ids() { @@ -46,20 +46,22 @@ fn submit_assigns_distinct_ids() { assert_eq!(q.snapshot().len(), 2); } +/// Submit-time dedup was removed with the agent-per-node refactor (a +/// multi-agent DAG has no single agent to key a dedup on), so an identical +/// resubmit — same template + agent, still queued — now enqueues a distinct +/// DAG instead of collapsing into the pending one. Whether any dedup needs +/// reintroducing is tracked as a follow-up. #[test] -fn dedup_pending_same_template_and_agent() { +fn identical_resubmit_is_a_distinct_dag() { let q = JobQueue::new(1); let a = submit(&q, rebuild("agent-a", "first")); - let b = submit(&q, rebuild("agent-a", "auto sweep")); - assert_eq!(a, b, "dedup should return existing id"); - let snap = q.snapshot(); - assert_eq!(snap.len(), 1); - assert!(snap[0].reason.contains("first")); - assert!(snap[0].reason.contains("auto sweep")); + let b = submit(&q, rebuild("agent-a", "again")); + assert_ne!(a, b, "no dedup: identical resubmit is a new DAG"); + assert_eq!(q.snapshot().len(), 2); } #[test] -fn dedup_does_not_apply_across_templates_or_agents() { +fn distinct_submits_never_collapse() { let q = JobQueue::new(1); let a = submit(&q, rebuild("agent-a", "r")); let b = submit(&q, rebuild("agent-b", "r")); @@ -73,7 +75,7 @@ fn dedup_does_not_apply_across_templates_or_agents() { } #[test] -fn dedup_skips_running_dags() { +fn resubmit_while_running_is_new_dag() { let q = JobQueue::new(1); let a = submit(&q, rebuild("agent-a", "first")); let claim = claim_one(&q); // Prebuild running @@ -84,126 +86,6 @@ fn dedup_skips_running_dags() { assert_eq!(q.snapshot().len(), 2); } -#[test] -fn meta_update_dedup_matches_inputs() { - let q = JobQueue::new(1); - let a = submit( - &q, - templates::meta_update( - vec!["nixpkgs".to_owned()], - Source::Manual, - "first".to_owned(), - None, - ), - ); - let b = submit( - &q, - templates::meta_update( - vec!["nixpkgs".to_owned()], - Source::Manual, - "duplicate click".to_owned(), - None, - ), - ); - assert_eq!(a, b, "identical-inputs meta-updates should dedup"); - let c = submit( - &q, - templates::meta_update( - vec!["agent-bitburner/bitburner-agent".to_owned()], - Source::Manual, - "bump agent".to_owned(), - None, - ), - ); - assert_ne!(a, c, "different-inputs meta-updates must NOT dedup"); - assert_eq!(q.snapshot().len(), 2); -} - -#[test] -fn approval_dags_dedup_only_on_matching_id() { - let q = JobQueue::new(1); - let a = submit( - &q, - templates::approval_deploy("agent-a", 1, "approval #1".to_owned()), - ); - let b = submit( - &q, - templates::approval_deploy("agent-a", 2, "approval #2".to_owned()), - ); - assert_ne!(a, b, "distinct approvals must not collapse"); - // Rapid double-click on the same approval IS a single op. - let c = submit( - &q, - templates::approval_deploy("agent-a", 1, "approval #1 (dup)".to_owned()), - ); - assert_eq!(a, c); - assert_eq!(q.snapshot().len(), 2); -} - -#[test] -fn perm_change_dedup_respects_perm_type() { - let q = JobQueue::new(1); - let groups = templates::perm_change( - "agent-a", - Source::Manual, - "groups".to_owned(), - PermPayload::ToolGroups { groups: vec![] }, - ); - let caps = templates::perm_change( - "agent-a", - Source::Manual, - "caps".to_owned(), - PermPayload::Capabilities { caps: vec![] }, - ); - let a = submit(&q, groups.clone()); - let b = submit(&q, caps); - assert_ne!(a, b, "tool-groups vs capabilities must not collapse"); - let c = submit(&q, groups); - assert_eq!(a, c, "same perm type dedups"); -} - -/// A `MetaUpdate` cascade `Rebuild` (with `parent_id = Some(meta_id)`) -/// must NOT dedup into a queued `Rebuild` with a different -/// `parent_id` (e.g. from a startup sweep) — without the guard the -/// cascade child would be swallowed and the agent never rebuilt -/// against the post-bump meta. -#[test] -fn dedup_respects_parent_id() { - let q = JobQueue::new(1); - let sweep = submit(&q, templates::startup_sweep("boot".to_owned(), vec![])); - let sweep_child = submit( - &q, - templates::rebuild( - "alice", - Source::StartupSweep, - "startup sweep".to_owned(), - Some(sweep), - true, - ), - ); - let meta = submit( - &q, - templates::meta_update(vec![], Source::Manual, "bump".to_owned(), None), - ); - let cascade_child = submit( - &q, - templates::rebuild( - "alice", - Source::MetaUpdate, - "meta-update cascade".to_owned(), - Some(meta), - false, - ), - ); - assert_ne!(sweep_child, cascade_child); - let rebuilds = q - .snapshot() - .iter() - .filter(|d| d.kind == Template::Rebuild && d.agent == "alice") - .count(); - assert_eq!(rebuilds, 2, "both rebuilds must be present"); -} - // ---- cycle rejection ---- #[test] @@ -213,6 +95,7 @@ fn cyclic_dag_is_rejected_at_submit() { // 0 → 1 → 0 cycle. spec.nodes = vec![ NodeSpec { + agent: "agent-a".to_owned(), kind: NodeKind::StopForUpdate, deps: vec![Dep { on: 1, @@ -220,6 +103,7 @@ fn cyclic_dag_is_rejected_at_submit() { }], }, NodeSpec { + agent: "agent-a".to_owned(), kind: NodeKind::Reconcile, deps: vec![Dep { on: 0, @@ -236,6 +120,7 @@ fn unknown_dep_is_rejected_at_submit() { let q = JobQueue::new(1); let mut spec = rebuild("agent-a", "bad dep"); spec.nodes = vec![NodeSpec { + agent: "agent-a".to_owned(), kind: NodeKind::Reconcile, deps: vec![Dep { on: 9, @@ -578,7 +463,7 @@ fn cancel_children_skips_running_child() { // ---- fan-out ---- #[test] -fn append_children_sets_parent_and_dedups() { +fn append_children_sets_parent() { let q = JobQueue::new(1); let meta = submit( &q, @@ -599,7 +484,7 @@ fn append_children_sets_parent_and_dedups() { Some(meta), false, ), - // Duplicate — must coalesce into the first alice child. + // No dedup: a second alice child is its own DAG now. templates::rebuild( "alice", Source::MetaUpdate, @@ -610,10 +495,10 @@ fn append_children_sets_parent_and_dedups() { ]; let ids = q.append_children(specs); assert_eq!(ids.len(), 3); - assert_eq!(ids[0], ids[2], "duplicate child dedups"); + assert_ne!(ids[0], ids[2], "no dedup: duplicate child is distinct"); let snap = q.snapshot(); let children: Vec<_> = snap.iter().filter(|d| d.parent_id == Some(meta)).collect(); - assert_eq!(children.len(), 2); + assert_eq!(children.len(), 3); } // ---- terminal reporting + lease release ---- diff --git a/hive-sh4re/src/jobs.rs b/hive-sh4re/src/jobs.rs index b576f3c2..b22b9b80 100644 --- a/hive-sh4re/src/jobs.rs +++ b/hive-sh4re/src/jobs.rs @@ -150,6 +150,11 @@ pub type NodeId = u32; #[derive(Debug, Clone, Serialize, Deserialize)] pub struct NodeView { pub id: NodeId, + /// The agent whose container (or meta repo, for `hyperhive` meta-level + /// nodes) this node operates on. Agent is per-node — a single DAG can + /// span multiple agents (e.g. a hive-wide restart), so there is no + /// DAG-level agent field; consumers group by this. + pub agent: String, /// Node primitive tag: `"prebuild"`, `"stop_for_update"`, /// `"swap"`, `"create"`, `"meta_lock"`, `"reconcile"`, `"signal"`, /// `"drain"`, `"write_dropin"`, `"write_perm_file"`, @@ -171,13 +176,13 @@ pub struct NodeView { pub error: Option, } -/// A queued/running/recent DAG. DAG-level fields mirror the pre-DAG -/// `QueueEntry` names (`kind` = template string, roll-up `state`); -/// everything per-node appears exactly once, inside `nodes`. +/// 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`. #[derive(Debug, Clone, Serialize, Deserialize)] pub struct DagView { pub id: u64, - pub agent: String, /// Template wire string — same values the old `kind` field used. pub kind: Template, /// Roll-up: `failed` if any node failed, else `running` / From e2b48d20143e56ab71f7ceaf990b7cded4b6db33 Mon Sep 17 00:00:00 2001 From: atlas Date: Tue, 14 Jul 2026 21:42:37 +0200 Subject: [PATCH 2/3] refactor(#2441): update queue consumers + docs for agent-per-node DagView no longer has a DAG-level agent, so consumers derive it from the per-node agents: - hivectl dag_progress.rs: dag_agents(d) helper (distinct node agents, comma-joined) in place of d.agent. - dashboard builds.js: entryAgents(entry) helper likewise for the rebuild-queue card + live-log header + cancel confirms. - docs/coordinator.md: lease prose (node-agent-keyed, global per agent), wire shape (NodeView.agent, no DagView.agent), and dropped the removed dedup section. --- docs/coordinator.md | 39 +++++++++++---------- frontend/packages/dashboard/src/builds.js | 21 +++++++++--- hive-c0re/src/bin/hivectl/dag_progress.rs | 42 ++++++++++++++++------- 3 files changed, 66 insertions(+), 36 deletions(-) diff --git a/docs/coordinator.md b/docs/coordinator.md index 1ac89697..acd1f84b 100644 --- a/docs/coordinator.md +++ b/docs/coordinator.md @@ -22,7 +22,7 @@ The **DAG** is the unit of dedup / cancel / approval-resolution and the dashboard group; the **node** is the unit of scheduling / execution / build-log / step label. Deps are intra-DAG edges only (`AfterOk` by default: the dep must succeed, a failed/cancelled dep cancels the dependent — -cancel-downstream). Cross-DAG ordering comes from the per-agent lease + dedup, +cancel-downstream). Cross-DAG ordering comes from the per-agent lease, never from edges between DAGs. Submit-time validation (petgraph `toposort`) rejects cyclic specs outright, fixing the old queue's "circular dep silently deadlocks" caveat. @@ -139,11 +139,13 @@ resources are free. Resources: 1. **Build slots** — `services.hyperhive.c0re.buildSlots` permits (default 1), held by nix-heavy nodes for the node's duration. -2. **Per-agent lifecycle lease** — DAG-scoped: acquired at the DAG's first - container-affecting node (`StopForUpdate`, `Swap`, `Signal`, `Drain`, - `Reconcile`, `WriteDropin`, `Create`, `ApprovalDeploy`), held until the DAG - is terminal, so two lifecycle DAGs for one agent never interleave their - container ops. **Lease-exempt**: `Prebuild`, `MetaLock`, `WritePermFile` — +2. **Per-agent lifecycle lease** — keyed on the **node's** agent (agent is + per-node; a DAG can span agents) and globally exclusive per agent across + all DAGs: acquired at a container-affecting node (`StopForUpdate`, `Swap`, + `Signal`, `Drain`, `Reconcile`, `WriteDropin`, `Create`, `ApprovalDeploy`), + held by the owning DAG until it's terminal, so two DAGs never interleave + container ops on the same agent. A DAG touching several agents holds one + lease per agent. **Lease-exempt**: `Prebuild`, `MetaLock`, `WritePermFile` — they touch the store / meta, not the running container, which is exactly why a stop can land while another DAG's prebuild is still building. @@ -156,14 +158,11 @@ The queue is in-memory only and lost on hive-c0re restart — deliberate: desired state is re-derived at boot from the DB + rev markers (see _Boot reconcile_), so there is no durable-recovery machinery to go wrong. -### Dedup, cancel, history +### Cancel, history -Dedup at **DAG granularity**: a repeat submit against a DAG whose roll-up is -still `Queued` with the same `(template, agent, parent_id, approval_id)` — -plus `inputs` for meta-updates and the perm-type discriminant for perm -changes — returns the existing id and appends an "also requested by …" line. -`parent_id` in the key keeps a cascade child from collapsing into a -standalone or sweep rebuild. Running/terminal DAGs never dedup. +Submit-time dedup was removed with the agent-per-node move (a multi-agent DAG +has no single agent to key a dedup on), so every submit enqueues a fresh DAG; +whether any dedup needs reintroducing is tracked as a follow-up. Cancel only applies to still-fully-queued DAGs (an in-flight nix build isn't interruptible); `cancel_children` cancels a parent's still-queued child DAGs. @@ -186,12 +185,14 @@ of dangling it). ### Wire shape `RebuildQueueChanged { seq, queue: [DagView…] }` (event name kept). Each -`DagView` carries the old entry-level fields (`id`, `kind` = template string, -roll-up `state`, `agent`, `source`, `parent_id`, `reason`, timestamps, -`inputs`, `approval_id`) plus `nodes: [NodeView…]` — per-node `kind`, `deps`, -`state`, `step`, `build_log_id`, timestamps, `error`. Step labels and build -logs are **per-node**; the dashboard renders the node chain on each queue -card and keys the live-log panel off the running node. +`DagView` carries the entry-level fields (`id`, `kind` = template string, +roll-up `state`, `source`, `parent_id`, `reason`, timestamps, `inputs`, +`approval_id`) plus `nodes: [NodeView…]` — per-node `agent`, `kind`, `deps`, +`state`, `step`, `build_log_id`, timestamps, `error`. There is **no +DAG-level `agent`** (agent is per-node, so a DAG can span agents); consumers +derive a DAG's agent(s) from its nodes. Step labels and build logs are +**per-node**; the dashboard renders the node chain on each queue card and +keys the live-log panel off the running node. --- diff --git a/frontend/packages/dashboard/src/builds.js b/frontend/packages/dashboard/src/builds.js index e56d5060..ac01cca2 100644 --- a/frontend/packages/dashboard/src/builds.js +++ b/frontend/packages/dashboard/src/builds.js @@ -165,11 +165,22 @@ function firstFailedNode(entry) { return (entry.nodes || []).find((n) => n.state === 'failed') || null; } +// Distinct agents across a DAG's nodes, comma-joined for display. Agent is +// per-node now (a DAG can span agents, e.g. a hive-wide restart), so there's +// no DAG-level `agent` field — derive it from the nodes. +function entryAgents(entry) { + const seen = []; + for (const n of entry.nodes || []) { + if (n.agent && !seen.includes(n.agent)) seen.push(n.agent); + } + return seen.join(','); +} + function rebuildQueueEntryFingerprint(entry, isChild) { return JSON.stringify({ state: entry.state, kind: entry.kind, - agent: entry.agent, + agent: entryAgents(entry), source: entry.source, started_at: entry.started_at, enqueued_at: entry.enqueued_at, @@ -265,7 +276,7 @@ function renderQueueEntry(entry, _byId, isChild) { el('span', { class: 'rqe-kind', title: entry.kind }, (QUEUE_KIND_GLYPH[entry.kind] || '?') + ' ' + entry.kind), ' ', - el('code', { class: 'rqe-agent' }, entry.agent), + el('code', { class: 'rqe-agent' }, entryAgents(entry)), ); li.append(' ', el('span', { class: 'rqe-source rqe-source-' + entry.source }, entry.source)); if (entry.state === 'queued') { @@ -331,14 +342,14 @@ function renderQueueEntry(entry, _byId, isChild) { class: 'inline rqe-cancel', 'data-async': '', 'data-confirm': - `cancel ${entry.kind} for \`${entry.agent}\` (queue id ${entry.id})? ` + + `cancel ${entry.kind} for \`${entryAgents(entry)}\` (queue id ${entry.id})? ` + `the row drops from the queue and never runs. running / done / failed entries can't be cancelled this way.`, }); cancelForm.append(el('button', { type: 'submit', class: 'rqe-cancel-btn', title: 'cancel this queued ' + entry.kind, - 'aria-label': 'cancel queued ' + entry.kind + ' for ' + entry.agent, + 'aria-label': 'cancel queued ' + entry.kind + ' for ' + entryAgents(entry), }, '✗')); li.append(cancelForm); } @@ -414,7 +425,7 @@ function renderRebuildLiveLog(queue) { const header = el('div', { class: 'rebuild-live-log-header' }, toggle, ' ', el('span', { class: 'rebuild-live-log-title' }, 'live build log — '), - el('code', { class: 'rqe-agent' }, running.agent), + el('code', { class: 'rqe-agent' }, entryAgents(running)), ' ', el('span', { class: 'rqe-kind' }, (running.kind || 'rebuild') + ' · ' + (NODE_KIND_LABEL[liveNode.kind] || liveNode.kind)), ' ', badge, ' ', diff --git a/hive-c0re/src/bin/hivectl/dag_progress.rs b/hive-c0re/src/bin/hivectl/dag_progress.rs index a50b7007..0875b814 100644 --- a/hive-c0re/src/bin/hivectl/dag_progress.rs +++ b/hive-c0re/src/bin/hivectl/dag_progress.rs @@ -68,7 +68,7 @@ async fn wait_for_dags_plain(socket: &Path, ids: Vec) -> Result<()> { // 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)); + failed.push(format!("{} {}", d.kind.as_str(), dag_agents(d))); } } else { all_terminal = false; @@ -138,7 +138,7 @@ async fn wait_for_dags_animated(socket: &Path, ids: Vec) -> Result<()> { "{} {} {} · {}", state_glyph(d.state), d.kind.as_str(), - d.agent, + dag_agents(d), fmt_dur(dag_elapsed(d, now)), )); for n in &d.nodes { @@ -166,7 +166,7 @@ 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(), d.agent)); + failed.push(format!("{} {}", d.kind.as_str(), dag_agents(d))); } } else { all_terminal = false; @@ -204,6 +204,19 @@ fn finish_wait(mut failed: Vec) -> Result<()> { } } +/// Distinct agents across a DAG's nodes, comma-joined for display — the +/// per-node replacement for the old DAG-level `agent` field. Single-agent +/// DAGs render one name; a hive-wide DAG lists each. +fn dag_agents(d: &hive_sh4re::jobs::DagView) -> String { + let mut seen: Vec<&str> = Vec::new(); + for n in &d.nodes { + if !seen.contains(&n.agent.as_str()) { + seen.push(&n.agent); + } + } + seen.join(",") +} + /// Current unix time in seconds (0 on the impossible pre-epoch error). fn now_unix() -> i64 { std::time::SystemTime::now() @@ -290,7 +303,7 @@ fn render_dag_line(d: &hive_sh4re::jobs::DagView) -> String { "{} {} {:<12}", state_glyph(d.state), d.kind.as_str(), - d.agent + dag_agents(d) ); for (i, n) in d.nodes.iter().enumerate() { let sep = if i == 0 { " " } else { " → " }; @@ -314,9 +327,10 @@ mod tests { use super::render_dag_line; - fn node(id: u32, kind: &str, state: State, step: Option<&str>) -> NodeView { + fn node(id: u32, agent: &str, kind: &str, state: State, step: Option<&str>) -> NodeView { NodeView { id, + agent: agent.to_owned(), kind: kind.to_owned(), deps: if id == 0 { vec![] } else { vec![id - 1] }, state, @@ -332,7 +346,6 @@ mod tests { 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, @@ -345,10 +358,16 @@ mod tests { 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), + 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), ], }; let line = render_dag_line(&dag); @@ -363,11 +382,10 @@ mod tests { #[test] fn render_dag_line_surfaces_first_node_error() { - let mut failed = node(0, "prebuild", State::Failed, None); + let mut failed = node(0, "bob", "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, From de69a9f02c61bccb2ebfba26ff5fcb5bcfc7cd41 Mon Sep 17 00:00:00 2001 From: atlas Date: Tue, 14 Jul 2026 21:56:02 +0200 Subject: [PATCH 3/3] refactor(#2441): live-log header shows the streaming node's agent, not the DAG's MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Review (iris/argus): the dashboard rebuild live-log header labels one specific node's log stream (liveNode, keyed by its build_log_id), so it should show that node's own agent — entryAgents(running) listed every agent in the DAG, which would mislabel a single agent's log once DAGs span multiple. The other entryAgents() sites (row label, cancel-confirm, fingerprint) are correct whole-DAG summaries and unchanged. --- frontend/packages/dashboard/src/builds.js | 6 +++++- 1 file changed, 5 insertions(+), 1 deletion(-) diff --git a/frontend/packages/dashboard/src/builds.js b/frontend/packages/dashboard/src/builds.js index ac01cca2..99381a07 100644 --- a/frontend/packages/dashboard/src/builds.js +++ b/frontend/packages/dashboard/src/builds.js @@ -425,7 +425,11 @@ function renderRebuildLiveLog(queue) { const header = el('div', { class: 'rebuild-live-log-header' }, toggle, ' ', el('span', { class: 'rebuild-live-log-title' }, 'live build log — '), - el('code', { class: 'rqe-agent' }, entryAgents(running)), + // This header labels one specific node's log stream (`liveNode`), not + // the DAG as a whole — so it's the node's own agent, not the DAG's + // full agent set (which would mislabel a single agent's log with every + // agent once DAGs span multiple). + el('code', { class: 'rqe-agent' }, liveNode.agent), ' ', el('span', { class: 'rqe-kind' }, (running.kind || 'rebuild') + ' · ' + (NODE_KIND_LABEL[liveNode.kind] || liveNode.kind)), ' ', badge, ' ',