diff --git a/hive-c0re/src/job_queue/exec.rs b/hive-c0re/src/job_queue/exec.rs index d554b9a8..ef24cb9b 100644 --- a/hive-c0re/src/job_queue/exec.rs +++ b/hive-c0re/src/job_queue/exec.rs @@ -10,7 +10,7 @@ use std::sync::Arc; use anyhow::{Context as _, Result}; -use super::model::{NodeKind, State, Template}; +use super::model::{NodeKind, NodeSpec, State, Template}; use super::{Claim, TerminalDag}; use crate::coordinator::Coordinator; use crate::power::{ReconcileAction, reconcile_action}; @@ -38,6 +38,16 @@ pub struct NodeOutput { /// these *before* the emitting node's completion so the DAG never /// rolls terminal with the appended work still pending. pub append_nodes: Vec, + /// Whole per-agent *subgraphs* to append into *this same* DAG at + /// runtime — the multi-node generalisation of `append_nodes`. Each + /// inner `Vec` is one independent subgraph whose `deps` are + /// local (0-based within that subgraph); the scheduler appends each via + /// [`JobQueue::append_subgraph`], which rebases the deps onto the DAG's + /// node-id space and roots the subgraph on the emitting node. The + /// startup sweep's `MetaLock` uses this to grow one stale-agent rebuild + /// subgraph per agent into the same boot DAG instead of fanning out + /// child DAGs. Same before-completion ordering as `append_nodes`. + pub append_subgraph: Vec>, } /// Step-label + build-log sink for one claimed node. @@ -276,8 +286,17 @@ async fn run_meta_lock( if let Err(e) = crate::meta::lock_update_hyperhive().await { tracing::warn!(error = ?e, "startup sweep: meta lock_update_hyperhive failed"); } + // Grow one rebuild subgraph per stale agent into *this* boot DAG + // (rooted on this `MetaLock`, so they build against the post-bump + // lock), rather than fanning out child DAGs. `relock = true` — a + // boot sweep relocks per-agent like a manual rebuild. + let append_subgraph = fanout + .unwrap_or_default() + .iter() + .map(|agent| super::templates::rebuild_nodes(agent, true, 0)) + .collect(); return Ok(NodeOutput { - fanout: fanout.unwrap_or_default(), + append_subgraph, ..Default::default() }); } diff --git a/hive-c0re/src/job_queue/mod.rs b/hive-c0re/src/job_queue/mod.rs index db0c3e1e..eddf4f81 100644 --- a/hive-c0re/src/job_queue/mod.rs +++ b/hive-c0re/src/job_queue/mod.rs @@ -34,7 +34,8 @@ use hive_sh4re::wire_time::now_unix; use tokio::sync::Notify; pub use model::{ - Dag, DagSpec, DagView, DepWhen, Node, NodeId, NodeKind, PermPayload, Source, State, Template, + Dag, DagSpec, DagView, DepWhen, Node, NodeId, NodeKind, NodeSpec, PermPayload, Source, State, + Template, }; /// How many terminal DAGs (`Done` / `Failed` / `Cancelled`) to retain @@ -206,6 +207,66 @@ impl JobQueue { Some(new_id) } + /// Append a whole *subgraph* into a live (non-terminal) DAG at runtime + /// — the multi-node, multi-agent generalisation of [`Self::append_node`]. + /// Each [`NodeSpec`] carries its own `agent` and subgraph-relative `deps` + /// (indices into `nodes`); this rebases those onto the DAG's node-id + /// space (`id == index`, an invariant `append_node` also maintains) and + /// attaches every subgraph *root* — a node with no internal deps — to + /// `dep_on` with an `AfterOk` edge. Used by the startup sweep's + /// `MetaLock` to grow per-agent rebuild subgraphs into the same boot DAG + /// instead of fanning out child DAGs. Same call-*before*-`complete_node` + /// contract as `append_node` (so the DAG can't roll terminal with the + /// appended work still pending). Returns the new node ids; empty if the + /// DAG is gone or `nodes` is empty. + pub fn append_subgraph(&self, dag_id: u64, nodes: Vec, dep_on: NodeId) -> Vec { + if nodes.is_empty() { + return Vec::new(); + } + let mut inner = self.inner.lock().expect("job_queue mutex poisoned"); + let Some(dag) = inner.dags.iter_mut().find(|d| d.id == dag_id) else { + return Vec::new(); + }; + let base: NodeId = u32::try_from(dag.nodes.len()).unwrap_or(u32::MAX); + let mut new_ids = Vec::with_capacity(nodes.len()); + for (i, spec) in nodes.into_iter().enumerate() { + let new_id: NodeId = base + u32::try_from(i).unwrap_or(u32::MAX); + // Subgraph roots (no internal deps) hang off the emitting node; + // internal deps rebase from subgraph-relative onto the DAG id + // space (both start at `base`). + let deps = if spec.deps.is_empty() { + vec![model::Dep { + on: dep_on, + when: DepWhen::AfterOk, + }] + } else { + spec.deps + .into_iter() + .map(|d| model::Dep { + on: base + d.on, + when: d.when, + }) + .collect() + }; + dag.nodes.push(Node { + id: new_id, + agent: spec.agent, + kind: spec.kind, + deps, + state: State::Queued, + step: None, + build_log_id: None, + started_at: None, + finished_at: None, + error: None, + }); + new_ids.push(new_id); + } + drop(inner); + self.notify.notify_one(); + new_ids + } + fn push_dag(inner: &mut Inner, spec: DagSpec) -> u64 { inner.next_id += 1; let id = inner.next_id; diff --git a/hive-c0re/src/job_queue/scheduler.rs b/hive-c0re/src/job_queue/scheduler.rs index eb03e98c..90e3bda6 100644 --- a/hive-c0re/src/job_queue/scheduler.rs +++ b/hive-c0re/src/job_queue/scheduler.rs @@ -116,6 +116,15 @@ async fn handle_completion( .job_queue .append_node(claim.dag_id, kind, claim.node_id); } + // Same before-completion ordering as `append_nodes`, but for + // whole per-agent subgraphs (the startup sweep's rebuild + // subgraphs growing into the boot DAG) — each an independent + // subgraph rooted on this node. + for subgraph in output.append_subgraph { + coord + .job_queue + .append_subgraph(claim.dag_id, subgraph, claim.node_id); + } coord .job_queue .complete_node(claim.dag_id, claim.node_id, Ok(()));