diff --git a/docs/coordinator.md b/docs/coordinator.md index acd1f84b..1ac89697 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, +cancel-downstream). Cross-DAG ordering comes from the per-agent lease + dedup, 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,13 +139,11 @@ 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** — 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` — +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` — 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. @@ -158,11 +156,14 @@ 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. -### Cancel, history +### Dedup, cancel, history -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. +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. 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. @@ -185,14 +186,12 @@ of dangling it). ### Wire shape `RebuildQueueChanged { seq, queue: [DagView…] }` (event name kept). Each -`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. +`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. --- diff --git a/frontend/packages/dashboard/src/builds.js b/frontend/packages/dashboard/src/builds.js index 99381a07..e56d5060 100644 --- a/frontend/packages/dashboard/src/builds.js +++ b/frontend/packages/dashboard/src/builds.js @@ -165,22 +165,11 @@ 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: entryAgents(entry), + agent: entry.agent, source: entry.source, started_at: entry.started_at, enqueued_at: entry.enqueued_at, @@ -276,7 +265,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' }, entryAgents(entry)), + el('code', { class: 'rqe-agent' }, entry.agent), ); li.append(' ', el('span', { class: 'rqe-source rqe-source-' + entry.source }, entry.source)); if (entry.state === 'queued') { @@ -342,14 +331,14 @@ function renderQueueEntry(entry, _byId, isChild) { class: 'inline rqe-cancel', 'data-async': '', 'data-confirm': - `cancel ${entry.kind} for \`${entryAgents(entry)}\` (queue id ${entry.id})? ` + + `cancel ${entry.kind} for \`${entry.agent}\` (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 ' + entryAgents(entry), + 'aria-label': 'cancel queued ' + entry.kind + ' for ' + entry.agent, }, '✗')); li.append(cancelForm); } @@ -425,11 +414,7 @@ function renderRebuildLiveLog(queue) { const header = el('div', { class: 'rebuild-live-log-header' }, toggle, ' ', el('span', { class: 'rebuild-live-log-title' }, 'live build log — '), - // 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('code', { class: 'rqe-agent' }, running.agent), ' ', 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 0875b814..a50b7007 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(), dag_agents(d))); + failed.push(format!("{} {}", d.kind.as_str(), d.agent)); } } 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(), - dag_agents(d), + d.agent, 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(), dag_agents(d))); + failed.push(format!("{} {}", d.kind.as_str(), d.agent)); } } else { all_terminal = false; @@ -204,19 +204,6 @@ 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() @@ -303,7 +290,7 @@ fn render_dag_line(d: &hive_sh4re::jobs::DagView) -> String { "{} {} {:<12}", state_glyph(d.state), d.kind.as_str(), - dag_agents(d) + d.agent ); for (i, n) in d.nodes.iter().enumerate() { let sep = if i == 0 { " " } else { " → " }; @@ -327,10 +314,9 @@ mod tests { use super::render_dag_line; - fn node(id: u32, agent: &str, kind: &str, state: State, step: Option<&str>) -> NodeView { + fn node(id: u32, 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, @@ -346,6 +332,7 @@ 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, @@ -358,16 +345,10 @@ mod tests { approval_id: None, perm_payload: 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, "prebuild", State::Done, None), + node(1, "stop_for_update", State::Done, None), + node(2, "swap", State::Running, Some("nixos-container update")), + node(3, "reconcile", State::Queued, None), ], }; let line = render_dag_line(&dag); @@ -382,10 +363,11 @@ mod tests { #[test] fn render_dag_line_surfaces_first_node_error() { - let mut failed = node(0, "bob", "prebuild", State::Failed, None); + let mut failed = node(0, "prebuild", State::Failed, None); failed.error = Some("nix build exploded".to_owned()); let dag = DagView { id: 8, + agent: "bob".to_owned(), kind: Template::Rebuild, state: State::Failed, source: Source::Manual, diff --git a/hive-c0re/src/job_queue/exec.rs b/hive-c0re/src/job_queue/exec.rs index 17749b20..8ccbc6d9 100644 --- a/hive-c0re/src/job_queue/exec.rs +++ b/hive-c0re/src/job_queue/exec.rs @@ -514,17 +514,12 @@ pub(super) async fn on_dag_terminal(coord: &Arc, terminal: &Termina | Template::GracefulRestart ) { - // 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"); - } + 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"); } } if terminal.approval_id.is_some() { @@ -532,26 +527,22 @@ pub(super) async fn on_dag_terminal(coord: &Arc, terminal: &Termina return; } if matches!(terminal.template, Template::Rebuild | Template::PermChange) { - // 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, - }), - _ => {} - } + 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, + }), + _ => {} } } } diff --git a/hive-c0re/src/job_queue/mod.rs b/hive-c0re/src/job_queue/mod.rs index db0c3e1e..27a968ee 100644 --- a/hive-c0re/src/job_queue/mod.rs +++ b/hive-c0re/src/job_queue/mod.rs @@ -7,12 +7,10 @@ //! 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** — 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. +//! 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. //! //! The meta *repo* is serialized by `meta::META_LOCK` inside the //! executors themselves. Per-agent power *intent* (`wanted`) lives in @@ -59,18 +57,15 @@ 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 newly acquired its agent's lease — - /// the scheduler creates the per-`(dag, agent)` transient guard on - /// this edge. + /// True when claiming this node acquired the DAG's agent lease — + /// the scheduler creates the DAG-scoped transient guard on this + /// edge. pub lease_acquired: bool, /// Transient pill kind for the lease window (from the spec). pub transient: Option, @@ -83,10 +78,7 @@ pub struct Claim { pub struct TerminalDag { pub dag_id: u64, pub template: Template, - /// 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 agent: String, pub approval_id: Option, pub state: State, /// First failed node's error when `state == Failed`. @@ -136,17 +128,27 @@ impl JobQueue { } } - /// Submit a DAG. Validates the spec (cycle rejection) and returns the - /// newly-allocated DAG id. + /// 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-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. + /// 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. 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(); @@ -154,8 +156,8 @@ impl JobQueue { } /// Append fan-out children under a parent DAG (meta-update / sweep - /// cascade); returns the child ids created. No dedup (see - /// [`Self::submit`]). + /// cascade). Applies the same dedup as [`Self::submit`]; returns + /// the child ids actually created or coalesced into. pub fn append_children(&self, specs: Vec) -> Vec { let mut ids = Vec::with_capacity(specs.len()); for spec in specs { @@ -181,14 +183,9 @@ 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, @@ -206,6 +203,21 @@ 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; @@ -215,7 +227,6 @@ 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, @@ -229,6 +240,7 @@ 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, @@ -270,23 +282,16 @@ 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 needs_lease { - match inner.leases.get(node_agent.as_str()) { + if node.kind.needs_lease() { + match inner.leases.get(dag.agent.as_str()) { Some(&holder) if holder != dag_id => continue, Some(_) => {} None => { - inner.leases.insert(node_agent.clone(), dag_id); + inner.leases.insert(dag.agent.clone(), dag_id); lease_acquired = true; } } @@ -299,7 +304,7 @@ impl JobQueue { dag_id, node_id, kind: dag.node(node_id).expect("node").kind.clone(), - agent: node_agent, + agent: dag.agent.clone(), template: dag.template, source: dag.source, approval_id: dag.approval_id, @@ -420,23 +425,13 @@ impl JobQueue { continue; } dag.terminal_reported = true; - // 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()); - } + if inner.leases.get(dag.agent.as_str()) == Some(&dag.id) { + freed.push(dag.agent.clone()); } reports.push(TerminalDag { dag_id: dag.id, template: dag.template, - agents, + agent: dag.agent.clone(), 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 bf0e4c95..7e474785 100644 --- a/hive-c0re/src/job_queue/model.rs +++ b/hive-c0re/src/job_queue/model.rs @@ -5,15 +5,33 @@ //! `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 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. +//! 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. 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")] @@ -185,11 +203,6 @@ 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, @@ -208,23 +221,19 @@ 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) 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`]. +/// (cycle rejection) and dedup'd by `JobQueue::submit`. #[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". + /// Free-form "why"; dedup appends "also requested by …" lines. pub reason: String, /// Cascade grouping (meta-update / sweep children). pub parent_id: Option, @@ -241,13 +250,12 @@ pub struct DagSpec { pub nodes: Vec, } -/// 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. +/// A live DAG in the queue. #[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, @@ -311,19 +319,6 @@ 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 { @@ -336,6 +331,7 @@ impl Dag { }; DagView { id: self.id, + agent: self.agent.clone(), kind: self.template, state: self.rollup(), source: self.source, @@ -352,7 +348,6 @@ 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 eb03e98c..230f9d30 100644 --- a/hive-c0re/src/job_queue/scheduler.rs +++ b/hive-c0re/src/job_queue/scheduler.rs @@ -35,10 +35,8 @@ 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, 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(); + // DAG id → transient guard held for the lease window. + let mut transients: HashMap = HashMap::new(); loop { // Terminal roll-ups can appear without a node completion — // the cancel surfaces settle DAGs directly and wake this loop @@ -51,10 +49,7 @@ pub async fn run_worker(coord: Arc) { if claim.lease_acquired && let Some(kind) = claim.transient { - transients.insert( - (claim.dag_id, claim.agent.clone()), - coord.transient_guard(&claim.agent, kind), - ); + transients.insert(claim.dag_id, coord.transient_guard(&claim.agent, kind)); } tracing::info!( dag = claim.dag_id, @@ -93,7 +88,7 @@ pub async fn run_worker(coord: Arc) { async fn handle_completion( coord: &Arc, - transients: &mut HashMap<(u64, String), crate::coordinator::TransientGuard>, + transients: &mut HashMap, done: NodeDone, ) { let NodeDone { claim, result } = done; @@ -148,11 +143,10 @@ async fn handle_completion( /// `Rebuilt` events, cancelled-power-op intent revert). async fn process_terminals( coord: &Arc, - transients: &mut HashMap<(u64, String), crate::coordinator::TransientGuard>, + transients: &mut HashMap, ) { for terminal in coord.job_queue.drain_terminal() { - // Drop every per-agent transient guard this DAG held. - transients.retain(|(dag_id, _), _| *dag_id != terminal.dag_id); + transients.remove(&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 48c3086c..29ffa8d3 100644 --- a/hive-c0re/src/job_queue/templates.rs +++ b/hive-c0re/src/job_queue/templates.rs @@ -3,11 +3,6 @@ //! 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) @@ -34,42 +29,35 @@ 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(agent: &str, relock: bool, base: u32) -> Vec { +fn rebuild_nodes(relock: bool, base: u32) -> Vec { vec![ - node( - agent, - NodeKind::Prebuild { relock }, - if base == 0 { + NodeSpec { + kind: NodeKind::Prebuild { relock }, + deps: if base == 0 { Vec::new() } else { after_ok(base - 1) }, - ), - node(agent, NodeKind::StopForUpdate, after_ok(base)), - node(agent, NodeKind::Swap, after_ok(base + 1)), - node( - agent, - NodeKind::Reconcile, - vec![Dep { + }, + NodeSpec { + kind: NodeKind::StopForUpdate, + deps: after_ok(base), + }, + NodeSpec { + kind: NodeKind::Swap, + deps: after_ok(base + 1), + }, + NodeSpec { + kind: NodeKind::Reconcile, + deps: vec![Dep { on: base + 2, when: DepWhen::AfterAny, }], - ), + }, ] } @@ -87,6 +75,7 @@ pub fn rebuild( ) -> DagSpec { DagSpec { template: Template::Rebuild, + agent: agent.to_owned(), source, reason, parent_id, @@ -94,7 +83,7 @@ pub fn rebuild( inputs: Vec::new(), perm_payload: None, transient: Some(TransientKind::Rebuilding), - nodes: rebuild_nodes(agent, relock, 0), + nodes: rebuild_nodes(relock, 0), } } @@ -104,6 +93,7 @@ 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, @@ -111,7 +101,10 @@ pub fn approval_deploy(agent: &str, approval_id: i64, reason: String) -> DagSpec inputs: Vec::new(), perm_payload: None, transient: Some(TransientKind::Rebuilding), - nodes: vec![node(agent, NodeKind::ApprovalDeploy, Vec::new())], + nodes: vec![NodeSpec { + kind: NodeKind::ApprovalDeploy, + deps: Vec::new(), + }], } } @@ -124,6 +117,7 @@ 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, @@ -132,9 +126,18 @@ pub fn graceful_stop(agent: &str, source: Source, reason: String) -> DagSpec { perm_payload: None, transient: Some(TransientKind::Stopping), nodes: vec![ - node(agent, NodeKind::Signal, Vec::new()), - node(agent, NodeKind::Drain, after_ok(0)), - node(agent, NodeKind::Reconcile, after_ok(1)), + NodeSpec { + kind: NodeKind::Signal, + deps: Vec::new(), + }, + NodeSpec { + kind: NodeKind::Drain, + deps: after_ok(0), + }, + NodeSpec { + kind: NodeKind::Reconcile, + deps: after_ok(1), + }, ], } } @@ -145,6 +148,7 @@ 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, @@ -153,8 +157,14 @@ pub fn restart(agent: &str, source: Source, reason: String) -> DagSpec { perm_payload: None, transient: Some(TransientKind::Restarting), nodes: vec![ - node(agent, NodeKind::StopForUpdate, Vec::new()), - node(agent, NodeKind::Reconcile, after_ok(0)), + NodeSpec { + kind: NodeKind::StopForUpdate, + deps: Vec::new(), + }, + NodeSpec { + kind: NodeKind::Reconcile, + deps: after_ok(0), + }, ], } } @@ -170,6 +180,7 @@ 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, @@ -178,10 +189,22 @@ pub fn graceful_restart(agent: &str, source: Source, reason: String) -> DagSpec perm_payload: None, transient: Some(TransientKind::Restarting), nodes: vec![ - 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)), + 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), + }, ], } } @@ -197,6 +220,7 @@ pub fn reconcile_only( ) -> DagSpec { DagSpec { template, + agent: agent.to_owned(), source, reason, parent_id: None, @@ -204,7 +228,10 @@ pub fn reconcile_only( inputs: Vec::new(), perm_payload: None, transient, - nodes: vec![node(agent, NodeKind::Reconcile, Vec::new())], + nodes: vec![NodeSpec { + kind: NodeKind::Reconcile, + deps: Vec::new(), + }], } } @@ -215,6 +242,7 @@ 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, @@ -223,10 +251,22 @@ pub fn spawn(agent: &str, approval_id: i64, reason: String) -> DagSpec { perm_payload: None, transient: Some(TransientKind::Spawning), nodes: vec![ - 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)), + 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), + }, ], } } @@ -235,10 +275,14 @@ 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![node(agent, NodeKind::WritePermFile, Vec::new())]; - nodes.extend(rebuild_nodes(agent, true, 1)); + let mut nodes = vec![NodeSpec { + kind: NodeKind::WritePermFile, + deps: Vec::new(), + }]; + nodes.extend(rebuild_nodes(true, 1)); DagSpec { template: Template::PermChange, + agent: agent.to_owned(), source, reason, parent_id: None, @@ -262,6 +306,7 @@ pub fn meta_update( ) -> DagSpec { DagSpec { template: Template::MetaUpdate, + agent: "hyperhive".to_owned(), source, reason, parent_id: None, @@ -269,17 +314,19 @@ pub fn meta_update( inputs, perm_payload: None, transient: None, - nodes: vec![node( - "hyperhive", - NodeKind::MetaLock { + nodes: vec![NodeSpec { + kind: NodeKind::MetaLock { sweep: false, fanout: None, }, - Vec::new(), - )], + deps: 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. @@ -289,6 +336,7 @@ 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, @@ -296,16 +344,17 @@ pub fn boot_root(reason: String) -> DagSpec { inputs: Vec::new(), perm_payload: None, transient: None, - nodes: vec![node("hyperhive", NodeKind::Noop, Vec::new())], + nodes: vec![NodeSpec { + kind: NodeKind::Noop, + deps: 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, @@ -313,14 +362,13 @@ pub fn startup_sweep(reason: String, stale_agents: Vec) -> DagSpec { inputs: Vec::new(), perm_payload: None, transient: None, - nodes: vec![node( - "hyperhive", - NodeKind::MetaLock { + nodes: vec![NodeSpec { + kind: NodeKind::MetaLock { sweep: true, fanout: Some(stale_agents), }, - Vec::new(), - )], + deps: Vec::new(), + }], } } diff --git a/hive-c0re/src/job_queue/tests.rs b/hive-c0re/src/job_queue/tests.rs index b969db92..958040da 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: submit / no-dedup, cycle rejection, resource +//! Queue-core unit tests: 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 removed — every submit is a fresh DAG) ---- +// ---- submit / dedup ---- #[test] fn submit_assigns_distinct_ids() { @@ -46,22 +46,20 @@ 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 identical_resubmit_is_a_distinct_dag() { +fn dedup_pending_same_template_and_agent() { let q = JobQueue::new(1); let a = submit(&q, rebuild("agent-a", "first")); - 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); + 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")); } #[test] -fn distinct_submits_never_collapse() { +fn dedup_does_not_apply_across_templates_or_agents() { let q = JobQueue::new(1); let a = submit(&q, rebuild("agent-a", "r")); let b = submit(&q, rebuild("agent-b", "r")); @@ -75,7 +73,7 @@ fn distinct_submits_never_collapse() { } #[test] -fn resubmit_while_running_is_new_dag() { +fn dedup_skips_running_dags() { let q = JobQueue::new(1); let a = submit(&q, rebuild("agent-a", "first")); let claim = claim_one(&q); // Prebuild running @@ -86,6 +84,126 @@ fn resubmit_while_running_is_new_dag() { 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] @@ -95,7 +213,6 @@ 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, @@ -103,7 +220,6 @@ fn cyclic_dag_is_rejected_at_submit() { }], }, NodeSpec { - agent: "agent-a".to_owned(), kind: NodeKind::Reconcile, deps: vec![Dep { on: 0, @@ -120,7 +236,6 @@ 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, @@ -463,7 +578,7 @@ fn cancel_children_skips_running_child() { // ---- fan-out ---- #[test] -fn append_children_sets_parent() { +fn append_children_sets_parent_and_dedups() { let q = JobQueue::new(1); let meta = submit( &q, @@ -484,7 +599,7 @@ fn append_children_sets_parent() { Some(meta), false, ), - // No dedup: a second alice child is its own DAG now. + // Duplicate — must coalesce into the first alice child. templates::rebuild( "alice", Source::MetaUpdate, @@ -495,10 +610,10 @@ fn append_children_sets_parent() { ]; let ids = q.append_children(specs); assert_eq!(ids.len(), 3); - assert_ne!(ids[0], ids[2], "no dedup: duplicate child is distinct"); + assert_eq!(ids[0], ids[2], "duplicate child dedups"); let snap = q.snapshot(); let children: Vec<_> = snap.iter().filter(|d| d.parent_id == Some(meta)).collect(); - assert_eq!(children.len(), 3); + assert_eq!(children.len(), 2); } // ---- terminal reporting + lease release ---- diff --git a/hive-sh4re/src/jobs.rs b/hive-sh4re/src/jobs.rs index b22b9b80..b576f3c2 100644 --- a/hive-sh4re/src/jobs.rs +++ b/hive-sh4re/src/jobs.rs @@ -150,11 +150,6 @@ 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"`, @@ -176,13 +171,13 @@ pub struct NodeView { pub error: Option, } -/// 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/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`. #[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` /