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` /