diff --git a/hive-c0re/src/job_queue/exec.rs b/hive-c0re/src/job_queue/exec.rs index 83bb8014..86ab6994 100644 --- a/hive-c0re/src/job_queue/exec.rs +++ b/hive-c0re/src/job_queue/exec.rs @@ -82,24 +82,24 @@ pub(super) async fn run_node(coord: &Arc, claim: &Claim) -> Result< node_id: claim.node_id, }; match &claim.kind { - NodeKind::Prebuild { relock } => run_prebuild(coord, claim, &ctx, *relock).await, - NodeKind::Swap => run_swap(coord, claim, &ctx).await, - NodeKind::PostSwap => run_post_swap(coord, claim, &ctx).await, - NodeKind::Provision => run_provision(coord, claim, &ctx).await, - NodeKind::Create => run_create(claim, &ctx).await, + NodeKind::Prebuild { relock, .. } => run_prebuild(coord, claim, &ctx, *relock).await, + NodeKind::Swap { .. } => run_swap(coord, claim, &ctx).await, + NodeKind::PostSwap { .. } => run_post_swap(coord, claim, &ctx).await, + NodeKind::Provision { .. } => run_provision(coord, claim, &ctx).await, + NodeKind::Create { .. } => run_create(claim, &ctx).await, NodeKind::MetaLock { sweep, fanout } => { run_meta_lock(coord, claim, &ctx, *sweep, fanout.clone()).await } - NodeKind::Reconcile => run_reconcile(coord, claim).await, - NodeKind::Start => run_start(coord, claim, &ctx).await, - NodeKind::Stop => run_stop(coord, claim, &ctx).await, - NodeKind::StopForUpdate => run_stop_for_update(coord, claim, &ctx).await, - NodeKind::Signal => Ok(run_signal(coord, claim, &ctx)), - NodeKind::Drain => run_drain(coord, claim, &ctx).await, - NodeKind::WriteDropin => run_write_dropin(coord, claim).await, - NodeKind::WritePermFile => run_write_perm_file(coord, claim, &ctx).await, - NodeKind::ApprovalDeploy => run_approval_deploy(coord, claim).await, - NodeKind::SetWanted { up } => run_set_wanted(coord, claim, *up), + NodeKind::Reconcile { .. } => run_reconcile(coord, claim).await, + NodeKind::Start { .. } => run_start(coord, claim, &ctx).await, + NodeKind::Stop { .. } => run_stop(coord, claim, &ctx).await, + NodeKind::StopForUpdate { .. } => run_stop_for_update(coord, claim, &ctx).await, + NodeKind::Signal { .. } => Ok(run_signal(coord, claim, &ctx)), + NodeKind::Drain { .. } => run_drain(coord, claim, &ctx).await, + NodeKind::WriteDropin { .. } => run_write_dropin(coord, claim).await, + NodeKind::WritePermFile { .. } => run_write_perm_file(coord, claim, &ctx).await, + NodeKind::ApprovalDeploy { .. } => run_approval_deploy(coord, claim).await, + NodeKind::SetWanted { up, .. } => run_set_wanted(coord, claim, *up), // Pure grouping container — no work; completing it lets it reach // `Finishing` so its child template nodes start. The DAG's terminal // hook fires (inline, via `run_terminal_hook`) when the container itself @@ -398,14 +398,17 @@ async fn run_reconcile(coord: &Arc, claim: &Claim) -> Result sub(NodeKind::Start), - ReconcileAction::Stop => sub(NodeKind::Stop), + ReconcileAction::Start => sub(NodeKind::Start { + agent: name.clone(), + }), + ReconcileAction::Stop => sub(NodeKind::Stop { + agent: name.clone(), + }), ReconcileAction::Noop => { tracing::debug!(%name, wanted = wanted.as_str(), running, "reconcile: noop"); Vec::new() diff --git a/hive-c0re/src/job_queue/mod.rs b/hive-c0re/src/job_queue/mod.rs index b6abc7f6..bf6158be 100644 --- a/hive-c0re/src/job_queue/mod.rs +++ b/hive-c0re/src/job_queue/mod.rs @@ -4,22 +4,22 @@ //! watcher, deferred-start follow-up, meta-update cascade) collapse into DAG //! *shapes* over the shared node primitives ([`model::NodeKind`]). //! -//! [`hive_jobq`] owns the graph, the two-class resource pool, and the settle -//! loop; this module maps hive-c0re's concepts onto it: -//! - `NodeKind` + agent → the crate payload [`resource::JobPayload`]; -//! - the two resource classes → [`resource::Resource`] -//! ([`resource::Resource::BuildSlot`] node-held, [`resource::Resource::Agent`] -//! subtree-held), derived per node by [`resource::JobPayload::resource_deps`]; -//! - a host **DAG id** groups a set of crate nodes; a node declares only its -//! `deps` (chain edges + resource deps), and the crate scheduler infers lease -//! re-entrancy from the [`Dep::Node`] graph — a node needing an agent lease a -//! node it depends on already holds re-enters it, with no parent annotation; -//! - per-DAG terminal work is a *focused* graph node per concern — `ResolveApproval` -//! (approval DAGs), `EmitRebuilt` (rebuild / perm-change), `RevertIntent` -//! (cancelled power-ops) — weak-depending on the DAG's tail nodes (extended onto -//! any runtime-appended subgraph's tail), so it runs once everything settles. -//! Hook-less DAGs (meta-update / boot / reconcile) get none. No drained event -//! stream, and no one node branching on DAG metadata. +//! [`hive_jobq`] owns the graph, the two-class resource pool, and the roll-up +//! settle loop; this module maps hive-c0re's concepts onto it: +//! - [`model::NodeKind`] **is** the crate payload `N` directly — each variant +//! carries the agent it targets ([`NodeKind::agent`]); the two resource +//! classes are [`resource::Resource`] (`BuildSlot` node-held, `Agent` lease +//! subtree-held), derived per node by [`NodeKind::resource_deps`]; +//! - a **DAG is a single container node** ([`NodeKind::Dag`], `parent = None`) +//! carrying the group's metadata, with the template's nodes hung under it as +//! its subtree (the **parent axis** groups; `deps` order). So the container's +//! `NodeId` is the DAG id, its rolled-up state is the DAG state, and membership +//! is a graph walk — there are no host grouping side-tables. The lease is owned +//! by a subtree root and borrowed by its descendants (continuity); +//! - per-DAG terminal work runs **inline** ([`exec::run_terminal_hook`]) when the +//! container rolls up terminal — dispatched off its template +//! ([`terminal_hook`]): approval-resolve, `Rebuilt`-emit, or power-intent +//! revert. No terminal-hook node, no drained event stream. //! //! The queue is runtime-only (no persistence): an empty graph on boot; desired //! state is re-derived by the reconcile sweep. A single scheduler task @@ -49,7 +49,7 @@ use crate::coordinator::TransientKind; pub use model::{ DagSpec, DagView, DepWhen, NodeKind, NodeSpec, PermPayload, Source, State, Template, }; -use resource::{JobPayload, Resource}; +use resource::Resource; /// How many terminal DAGs (`Done` / `Failed` / `Cancelled`) to retain per /// template in the snapshot, matching the old per-kind history cap. @@ -130,7 +130,7 @@ struct DagMeta { /// are graph queries ([`QueueInner::container`] / [`QueueInner::subtree`] / /// [`QueueInner::dag_meta`]). One shared crate [`Graph`] holds every DAG. struct QueueInner { - sched: Scheduler, + sched: Scheduler, /// Per-node runtime metadata (build-log id, step, timestamps, error) — /// mutable after insert, so it can't ride the immutable node payload. node_rt: HashMap, @@ -232,10 +232,7 @@ fn insert_group( ) -> anyhow::Result> { let mut ids: Vec = Vec::with_capacity(nodes.len()); for ns in nodes { - let payload = JobPayload { - kind: ns.kind.clone(), - agent: ns.agent.clone(), - }; + let payload = ns.kind.clone(); let mut deps = payload.resource_deps(); for d in &ns.deps { deps.push(Dep::Node { @@ -293,18 +290,15 @@ impl JobQueue { let container = inner .sched .append( - JobPayload { - kind: NodeKind::Dag { - template: spec.template, - source: spec.source, - reason: spec.reason, - transient: spec.transient, - approval_id: spec.approval_id, - inputs: spec.inputs, - perm_payload: spec.perm_payload, - created_at: now_unix(), - }, - agent: String::new(), + NodeKind::Dag { + template: spec.template, + source: spec.source, + reason: spec.reason, + transient: spec.transient, + approval_id: spec.approval_id, + inputs: spec.inputs, + perm_payload: spec.perm_payload, + created_at: now_unix(), }, Vec::new(), None, @@ -377,8 +371,8 @@ impl JobQueue { let Some(node) = inner.sched.graph().node(id) else { continue; }; - let kind = node.payload.kind.clone(); - let agent = node.payload.agent.clone(); + let kind = node.payload.clone(); + let agent = node.payload.agent().to_owned(); let Some(container) = inner.dag_of(id) else { continue; }; @@ -605,7 +599,7 @@ impl QueueInner { self.sched.graph().nodes().find_map(|n| { (n.parent.is_none() && n.id.get() == dag_id - && matches!(n.payload.kind, NodeKind::Dag { .. })) + && matches!(n.payload, NodeKind::Dag { .. })) .then_some(n.id) }) } @@ -646,7 +640,7 @@ impl QueueInner { inputs, perm_payload, created_at, - } = &self.sched.graph().node(container)?.payload.kind + } = &self.sched.graph().node(container)?.payload else { return None; }; @@ -714,9 +708,9 @@ impl QueueInner { let mut seen: Vec = Vec::new(); for id in self.subtree(container) { if let Some(n) = self.sched.graph().node(id) { - let agent = &n.payload.agent; + let agent = n.payload.agent(); if !agent.is_empty() && !seen.iter().any(|s| s == agent) { - seen.push(agent.clone()); + seen.push(agent.to_owned()); } } } @@ -780,8 +774,8 @@ impl QueueInner { } nodes.push(NodeView { id: id.get(), - agent: node.payload.agent.clone(), - kind: node.payload.kind.as_str().to_owned(), + agent: node.payload.agent().to_owned(), + kind: node.payload.as_str().to_owned(), deps, state: to_wire_state(node.state), step: rt.and_then(|r| r.step.clone()), @@ -827,7 +821,7 @@ impl QueueInner { self.sched .graph() .nodes() - .filter(|n| n.parent.is_none() && matches!(n.payload.kind, NodeKind::Dag { .. })) + .filter(|n| n.parent.is_none() && matches!(n.payload, NodeKind::Dag { .. })) .map(|n| n.id) .collect() } diff --git a/hive-c0re/src/job_queue/model.rs b/hive-c0re/src/job_queue/model.rs index 7365ce89..9c4da05c 100644 --- a/hive-c0re/src/job_queue/model.rs +++ b/hive-c0re/src/job_queue/model.rs @@ -56,12 +56,12 @@ pub enum NodeKind { /// skipped when the container is already down — it only exists to /// shrink the swap's downtime, which a stopped agent doesn't need /// (the sync + dir prep still run; `Swap` builds inline). - Prebuild { relock: bool }, + Prebuild { agent: String, relock: bool }, /// `nixos-container update` profile-swap (requires the container /// stopped). Re-applies nspawn flags + resource limits first — /// rebuild is the reconcile verb. The post-rebuild bookkeeping tail /// lives in the sibling `PostSwap` node. - Swap, + Swap { agent: String }, /// The post-`Swap` bookkeeping tail as a first-class node: rev marker, /// forge + matrix sync, manager kick, container rescan, meta-inputs /// snapshot. Split out of `Swap` for dashboard visibility + retry @@ -71,16 +71,16 @@ pub enum NodeKind { /// recovery still runs. Store/forge/matrix work only — no nix build, so /// build-slot-exempt; the agent lease taken at `Swap` is held across the /// whole chain until `Reconcile` settles, so it's not re-declared here. - PostSwap, + PostSwap { agent: String }, /// First-spawn pre-create provisioning: proposed/applied repos, /// state subvolume, and meta registration (`sync_agents`). Runs /// ahead of `Create` so the `nixos-container create --flake /// meta#` ref resolves. Store/meta-only — no container yet — /// so it's lease- and build-slot-exempt like `Prebuild`. - Provision, + Provision { agent: String }, /// First-spawn `nixos-container create` proper. Assumes the /// upstream `Provision` node already registered the agent in meta. - Create, + Create { agent: String }, /// Meta flake lock bump. `sweep = false`: `meta::lock_update` /// (commit fused, under `META_LOCK`) with the DAG's `inputs`; /// `sweep = true`: `meta::lock_update_hyperhive`, *non-fatal* (a @@ -97,36 +97,36 @@ pub enum NodeKind { /// `Offline` & up, else noop). The mechanical work is not done in /// this node — it fans a child [`NodeKind::Start`] / [`NodeKind::Stop`] /// DAG out at runtime so the sub-step is a first-class DAG node. - Reconcile, + Reconcile { agent: String }, /// Mechanical container start: the start preamble (runtime dir + /// drop-ins), `start_with_fallback`, MCP listener registration, and /// the manager kick. Fanned out by a [`NodeKind::Reconcile`] that /// observed `wanted = Up` and the container down. - Start, + Start { agent: String }, /// Mechanical container stop: `nixos-container` kill, MCP listener /// unregister, and the `Killed` manager notify. Fanned out by a /// [`NodeKind::Reconcile`] that observed `wanted = Offline` and up. - Stop, + Stop { agent: String }, /// Mechanical `nixos-container stop` for the profile swap. Never /// touches `wanted`. Noop if already stopped. - StopForUpdate, + StopForUpdate { agent: String }, /// Set the graceful-stop fence + kick the harness so it runs one /// stop-checkpoint turn. - Signal, + Signal { agent: String }, /// Await the harness clearing the fence, bounded by /// `GRACEFUL_STOP_TIMEOUT`. Resolves ok either way — the /// downstream `Reconcile` performs the actual stop. - Drain, + Drain { agent: String }, /// `set_nspawn_flags` + `set_resource_limits` + daemon-reload. - WriteDropin, + WriteDropin { agent: String }, /// Commit `tool-groups.json` / `capabilities.json` per the DAG's /// `perm_payload` (commit fused under `META_LOCK`). - WritePermFile, + WritePermFile { agent: String }, /// Opaque approval deploy pipeline (`MergeConfigPr`): the two-phase /// prepare/finalize/abort meta deploy stays inside `actions.rs` in v1 — /// deliberately not /// modeled as scheduler nodes (see the design doc §9). - ApprovalDeploy, + ApprovalDeploy { agent: String }, /// Write the agent's durable power intent (`wanted = Up` when `up`, else /// `Offline`) as a first-class DAG node, at the head of a power-op /// template so the downstream `Reconcile` reads it. Replaces the old @@ -140,7 +140,7 @@ pub enum NodeKind { /// the DAG. (In `stale_start` the lease is thus held across the head /// `Prebuild`, but that's a no-op there — the agent is down, so prebuild /// is skipped.) - SetWanted { up: bool }, + SetWanted { agent: String, up: bool }, /// The **DAG container** node: one per submitted DAG, carrying the group's /// domain metadata. Every template node hangs *under* it (its subtree), so /// the container's `NodeId` **is** the DAG id, its rolled-up state **is** the @@ -166,35 +166,60 @@ impl NodeKind { pub fn as_str(&self) -> &'static str { match self { NodeKind::Prebuild { .. } => "prebuild", - NodeKind::Swap => "swap", - NodeKind::PostSwap => "post_swap", - NodeKind::Provision => "provision", - NodeKind::Create => "create", + NodeKind::Swap { .. } => "swap", + NodeKind::PostSwap { .. } => "post_swap", + NodeKind::Provision { .. } => "provision", + NodeKind::Create { .. } => "create", NodeKind::MetaLock { .. } => "meta_lock", - NodeKind::Reconcile => "reconcile", - NodeKind::Start => "start", - NodeKind::Stop => "stop", - NodeKind::StopForUpdate => "stop_for_update", - NodeKind::Signal => "signal", - NodeKind::Drain => "drain", - NodeKind::WriteDropin => "write_dropin", - NodeKind::WritePermFile => "write_perm_file", - NodeKind::ApprovalDeploy => "approval_deploy", + NodeKind::Reconcile { .. } => "reconcile", + NodeKind::Start { .. } => "start", + NodeKind::Stop { .. } => "stop", + NodeKind::StopForUpdate { .. } => "stop_for_update", + NodeKind::Signal { .. } => "signal", + NodeKind::Drain { .. } => "drain", + NodeKind::WriteDropin { .. } => "write_dropin", + NodeKind::WritePermFile { .. } => "write_perm_file", + NodeKind::ApprovalDeploy { .. } => "approval_deploy", NodeKind::SetWanted { .. } => "set_wanted", NodeKind::Dag { .. } => "dag", } } + /// The agent this node targets, or `""` for agentless kinds + /// ([`NodeKind::MetaLock`] on the `hyperhive` pseudo-agent, and the + /// [`NodeKind::Dag`] container). + #[must_use] + pub fn agent(&self) -> &str { + match self { + NodeKind::Prebuild { agent, .. } + | NodeKind::Swap { agent } + | NodeKind::PostSwap { agent } + | NodeKind::Provision { agent } + | NodeKind::Create { agent } + | NodeKind::Reconcile { agent } + | NodeKind::Start { agent } + | NodeKind::Stop { agent } + | NodeKind::StopForUpdate { agent } + | NodeKind::Signal { agent } + | NodeKind::Drain { agent } + | NodeKind::WriteDropin { agent } + | NodeKind::WritePermFile { agent } + | NodeKind::ApprovalDeploy { agent } + | NodeKind::SetWanted { agent, .. } => agent, + NodeKind::MetaLock { .. } | NodeKind::Dag { .. } => "", + } + } + /// Nix-heavy kinds hold one of the `buildSlots` semaphore permits /// for the node's duration. pub fn needs_build_slot(&self) -> bool { matches!( self, NodeKind::Prebuild { .. } - | NodeKind::Swap - | NodeKind::Create + | NodeKind::Swap { .. } + | NodeKind::Create { .. } | NodeKind::MetaLock { .. } - | NodeKind::ApprovalDeploy + | NodeKind::ApprovalDeploy { .. } ) } @@ -209,14 +234,14 @@ impl NodeKind { pub fn needs_lease(&self) -> bool { matches!( self, - NodeKind::Swap - | NodeKind::Create - | NodeKind::Reconcile - | NodeKind::StopForUpdate - | NodeKind::Signal - | NodeKind::Drain - | NodeKind::WriteDropin - | NodeKind::ApprovalDeploy + NodeKind::Swap { .. } + | NodeKind::Create { .. } + | NodeKind::Reconcile { .. } + | NodeKind::StopForUpdate { .. } + | NodeKind::Signal { .. } + | NodeKind::Drain { .. } + | NodeKind::WriteDropin { .. } + | NodeKind::ApprovalDeploy { .. } | NodeKind::SetWanted { .. } ) } @@ -225,9 +250,9 @@ impl NodeKind { /// Submit-time spec for one node. #[derive(Debug, Clone)] pub struct NodeSpec { - /// The agent this node's work targets. Built by the `templates.rs` `node` - /// helper, which stamps the template's agent onto every node. - pub agent: String, + /// The node's payload — [`NodeKind`] is the queue's payload type directly, + /// and each variant carries the agent it targets (a DAG can span agents; + /// the queue derives per-agent leasing from [`NodeKind::agent`]). pub kind: NodeKind, pub deps: Vec, /// The **structural parent** axis — the spec-local index of this node's diff --git a/hive-c0re/src/job_queue/resource.rs b/hive-c0re/src/job_queue/resource.rs index 9358af5f..a59a428f 100644 --- a/hive-c0re/src/job_queue/resource.rs +++ b/hive-c0re/src/job_queue/resource.rs @@ -1,8 +1,8 @@ -//! The concrete resource + payload types the rebuild queue schedules over — -//! the bridge from hive-c0re's [`NodeKind`] onto the domain-agnostic -//! `hive-jobq` crate. `hive-jobq` is generic over a resource type -//! `R: Clone + Eq + Hash` and a node payload `N`; here `R` is [`Resource`] and -//! `N` is [`JobPayload`]. +//! The concrete resource type the rebuild queue schedules over — the bridge +//! from hive-c0re's [`NodeKind`] onto the domain-agnostic `hive-jobq` crate. +//! `hive-jobq` is generic over a resource type `R: Clone + Eq + Hash` and a node +//! payload `N`; here `R` is [`Resource`] and `N` is [`NodeKind`] directly (each +//! variant carries the agent it targets). use hive_jobq::Dep; @@ -24,16 +24,7 @@ pub enum Resource { Agent(String), } -/// A schedulable node's payload — the crate's generic `N`. Carries the -/// primitive operation and the agent it targets. The agent is per-node (a DAG -/// spans agents), and the lease [`Resource::Agent`] is keyed on it. -#[derive(Debug, Clone)] -pub struct JobPayload { - pub kind: NodeKind, - pub agent: String, -} - -impl JobPayload { +impl NodeKind { /// The [`Dep::Resource`] edges this node must acquire to run, derived from /// its kind + agent: a build slot for nix-heavy kinds /// ([`NodeKind::needs_build_slot`]) and the agent lease for @@ -43,15 +34,15 @@ impl JobPayload { /// `Agent` lock through the crate's recursive re-entrancy. pub fn resource_deps(&self) -> Vec> { let mut deps = Vec::new(); - if self.kind.needs_build_slot() { + if self.needs_build_slot() { deps.push(Dep::Resource { name: Resource::BuildSlot, count: 1, }); } - if self.kind.needs_lease() { + if self.needs_lease() { deps.push(Dep::Resource { - name: Resource::Agent(self.agent.clone()), + name: Resource::Agent(self.agent().to_owned()), count: 1, }); } diff --git a/hive-c0re/src/job_queue/submit.rs b/hive-c0re/src/job_queue/submit.rs index 1a741813..5c0ddf4b 100644 --- a/hive-c0re/src/job_queue/submit.rs +++ b/hive-c0re/src/job_queue/submit.rs @@ -63,13 +63,20 @@ fn stop_chain(agent: &str, graceful: bool, running: bool) -> Vec { // `SetWanted` is the group root and owns the agent lease; the mechanical // steps are its children (borrow the lease, run once it reaches `Finishing`, // dep-ordered among themselves). - let mut n = vec![node(agent, NodeKind::SetWanted { up: false }, Vec::new())]; + let a = || agent.to_owned(); + let mut n = vec![node( + NodeKind::SetWanted { + agent: a(), + up: false, + }, + Vec::new(), + )]; if graceful && running { - n.push(child(0, agent, NodeKind::Signal, Vec::new())); - n.push(child(0, agent, NodeKind::Drain, after_ok(1))); - n.push(child(0, agent, NodeKind::Reconcile, after_ok(2))); + n.push(child(0, NodeKind::Signal { agent: a() }, Vec::new())); + n.push(child(0, NodeKind::Drain { agent: a() }, after_ok(1))); + n.push(child(0, NodeKind::Reconcile { agent: a() }, after_ok(2))); } else { - n.push(child(0, agent, NodeKind::Reconcile, Vec::new())); + n.push(child(0, NodeKind::Reconcile { agent: a() }, Vec::new())); } n } @@ -79,14 +86,26 @@ fn stop_chain(agent: &str, graceful: bool, running: bool) -> Vec { /// current derivations), otherwise a plain `Reconcile` (which starts a down /// agent and noops an already-running one). fn start_chain(agent: &str, running: bool, stale: bool) -> Vec { - let mut n = vec![node(agent, NodeKind::SetWanted { up: true }, Vec::new())]; + let mut n = vec![node( + NodeKind::SetWanted { + agent: agent.to_owned(), + up: true, + }, + Vec::new(), + )]; if !running && stale { // Rebuild subtree after the SetWanted head (base = 1, so the rebuild's // `Prebuild` root deps `after_ok(0)` = the head). `Prebuild` + // `Reconcile` are their own group roots (top-level, per `rebuild_nodes`). n.extend(rebuild_nodes(agent, true, 1)); } else { - n.push(child(0, agent, NodeKind::Reconcile, Vec::new())); + n.push(child( + 0, + NodeKind::Reconcile { + agent: agent.to_owned(), + }, + Vec::new(), + )); } n } @@ -101,22 +120,27 @@ fn start_chain(agent: &str, running: bool, stale: bool) -> Vec { /// converges to intent — a stopped (`wanted = Off`) agent stays stopped, /// a crashed (`wanted = Up`) agent comes back up. fn restart_chain(agent: &str, graceful: bool, running: bool) -> Vec { + let a = || agent.to_owned(); if !running { // Nothing to bounce — a lone Reconcile converges to intent. - return vec![node(agent, NodeKind::Reconcile, Vec::new())]; + return vec![node(NodeKind::Reconcile { agent: a() }, Vec::new())]; } // Running: mechanical stop then Reconcile. The first stop node is the group // ROOT (no SetWanted head) and owns the agent lease; the rest are its // children (borrow the lease, dep-ordered), so the bounce holds one // continuous lease and `Reconcile` cancel-cascades if a stop step fails. let mut n = vec![if graceful { - node(agent, NodeKind::Signal, Vec::new()) + node(NodeKind::Signal { agent: a() }, Vec::new()) } else { - node(agent, NodeKind::StopForUpdate, Vec::new()) + node(NodeKind::StopForUpdate { agent: a() }, Vec::new()) }]; if graceful { - n.push(child(0, agent, NodeKind::Drain, Vec::new())); - n.push(child(0, agent, NodeKind::StopForUpdate, after_ok(1))); + n.push(child(0, NodeKind::Drain { agent: a() }, Vec::new())); + n.push(child( + 0, + NodeKind::StopForUpdate { agent: a() }, + after_ok(1), + )); } // `Reconcile` gates on the last mechanical step. When the only step is the // root itself (non-graceful, `StopForUpdate` == index 0), the parent gate @@ -128,7 +152,7 @@ fn restart_chain(agent: &str, graceful: bool, running: bool) -> Vec { } else { Vec::new() }; - n.push(child(0, agent, NodeKind::Reconcile, deps)); + n.push(child(0, NodeKind::Reconcile { agent: a() }, deps)); n } @@ -151,7 +175,6 @@ fn concat_subgraphs(chains: Vec>) -> Vec { }) .collect(); out.push(NodeSpec { - agent: spec.agent, kind: spec.kind, deps, // Rebase the structural parent by the same offset (a subgraph diff --git a/hive-c0re/src/job_queue/templates.rs b/hive-c0re/src/job_queue/templates.rs index 61852e1f..5c739962 100644 --- a/hive-c0re/src/job_queue/templates.rs +++ b/hive-c0re/src/job_queue/templates.rs @@ -42,14 +42,14 @@ pub(crate) fn after_ok(on: u64) -> Vec { }] } -/// Build one **top-level (group-root)** node targeting `agent` — `parent = -/// None`. The single place a node's agent is stamped. Shared with `submit.rs`'s -/// dynamic power-op builders. A root owns whatever resource it declares for its -/// whole subtree; its descendants borrow it (agent-lease / build-slot -/// continuity). Ordering vs other nodes is `deps`; grouping is `parent`. -pub(crate) fn node(agent: &str, kind: NodeKind, deps: Vec) -> NodeSpec { +/// Build one **top-level (group-root)** node — `parent = None`. `kind` carries +/// the agent it targets ([`NodeKind`] is the payload directly). Shared with +/// `submit.rs`'s dynamic power-op builders. A root owns whatever resource it +/// declares for its whole subtree; its descendants borrow it (agent-lease / +/// build-slot continuity). Ordering vs other nodes is `deps`; grouping is +/// `parent`. +pub(crate) fn node(kind: NodeKind, deps: Vec) -> NodeSpec { NodeSpec { - agent: agent.to_owned(), kind, deps, parent: None, @@ -60,9 +60,8 @@ pub(crate) fn node(agent: &str, kind: NodeKind, deps: Vec) -> NodeSpec { /// child runs once its parent reaches `Finishing` (the parent gate), so it must /// NOT `deps` on `parent` (dep-scope validation rejects a dep on one's own /// parent). `deps` here order the child against its *siblings* only. -pub(crate) fn child(parent: u64, agent: &str, kind: NodeKind, deps: Vec) -> NodeSpec { +pub(crate) fn child(parent: u64, kind: NodeKind, deps: Vec) -> NodeSpec { NodeSpec { - agent: agent.to_owned(), kind, deps, parent: Some(parent), @@ -87,22 +86,25 @@ pub(crate) fn child(parent: u64, agent: &str, kind: NodeKind, deps: Vec) -> /// (recovery-start invariant). It takes a fresh lease; the tiny gap is /// harmless — `Reconcile` converges to the persisted `wanted` idempotently. pub(crate) fn rebuild_nodes(agent: &str, relock: bool, base: u64) -> Vec { + let a = || agent.to_owned(); vec![ node( - agent, - NodeKind::Prebuild { relock }, + NodeKind::Prebuild { agent: a(), relock }, if base == 0 { Vec::new() } else { after_ok(base - 1) }, ), - child(base, agent, NodeKind::StopForUpdate, Vec::new()), - child(base + 1, agent, NodeKind::Swap, Vec::new()), - child(base + 1, agent, NodeKind::PostSwap, after_ok(base + 2)), + child(base, NodeKind::StopForUpdate { agent: a() }, Vec::new()), + child(base + 1, NodeKind::Swap { agent: a() }, Vec::new()), + child( + base + 1, + NodeKind::PostSwap { agent: a() }, + after_ok(base + 2), + ), node( - agent, - NodeKind::Reconcile, + NodeKind::Reconcile { agent: a() }, vec![Dep { on: base, when: DepWhen::AfterAny, @@ -141,7 +143,12 @@ 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![node( + NodeKind::ApprovalDeploy { + agent: agent.to_owned(), + }, + Vec::new(), + )], } } @@ -166,7 +173,12 @@ pub fn reconcile_only( inputs: Vec::new(), perm_payload: None, transient, - nodes: vec![node(agent, NodeKind::Reconcile, Vec::new())], + nodes: vec![node( + NodeKind::Reconcile { + agent: agent.to_owned(), + }, + Vec::new(), + )], } } @@ -188,12 +200,15 @@ pub fn spawn(agent: &str, approval_id: i64, reason: String) -> DagSpec { inputs: Vec::new(), perm_payload: None, transient: Some(TransientKind::Spawning), - nodes: vec![ - node(agent, NodeKind::Provision, Vec::new()), - child(0, agent, NodeKind::Create, Vec::new()), - child(1, agent, NodeKind::WriteDropin, Vec::new()), - child(1, agent, NodeKind::Reconcile, after_ok(2)), - ], + nodes: { + let a = || agent.to_owned(); + vec![ + node(NodeKind::Provision { agent: a() }, Vec::new()), + child(0, NodeKind::Create { agent: a() }, Vec::new()), + child(1, NodeKind::WriteDropin { agent: a() }, Vec::new()), + child(1, NodeKind::Reconcile { agent: a() }, after_ok(2)), + ] + }, } } @@ -201,7 +216,12 @@ 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())]; + let mut nodes = vec![node( + NodeKind::WritePermFile { + agent: agent.to_owned(), + }, + Vec::new(), + )]; nodes.extend(rebuild_nodes(agent, true, 1)); DagSpec { template: Template::PermChange, @@ -240,7 +260,6 @@ pub fn meta_update( perm_payload: None, transient: Some(TransientKind::Rebuilding), nodes: vec![node( - "hyperhive", NodeKind::MetaLock { sweep: false, fanout: None, diff --git a/hive-c0re/src/job_queue/tests.rs b/hive-c0re/src/job_queue/tests.rs index 0063b7fc..b70fc195 100644 --- a/hive-c0re/src/job_queue/tests.rs +++ b/hive-c0re/src/job_queue/tests.rs @@ -109,8 +109,9 @@ fn cyclic_dag_is_rejected_at_submit() { // 0 → 1 → 0 cycle. spec.nodes = vec![ NodeSpec { - agent: "agent-a".to_owned(), - kind: NodeKind::StopForUpdate, + kind: NodeKind::StopForUpdate { + agent: "agent-a".to_owned(), + }, deps: vec![Dep { on: 1, when: DepWhen::AfterOk, @@ -118,8 +119,9 @@ fn cyclic_dag_is_rejected_at_submit() { parent: None, }, NodeSpec { - agent: "agent-a".to_owned(), - kind: NodeKind::Reconcile, + kind: NodeKind::Reconcile { + agent: "agent-a".to_owned(), + }, deps: vec![Dep { on: 0, when: DepWhen::AfterOk, @@ -136,8 +138,9 @@ 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, + kind: NodeKind::Reconcile { + agent: "agent-a".to_owned(), + }, deps: vec![Dep { on: 9, when: DepWhen::AfterOk, @@ -154,8 +157,9 @@ fn invalid_parent_is_rejected_at_submit() { // A forward/out-of-bounds parent index must be refused at validate, not // panic in `insert_group`. spec.nodes = vec![NodeSpec { - agent: "agent-a".to_owned(), - kind: NodeKind::Reconcile, + kind: NodeKind::Reconcile { + agent: "agent-a".to_owned(), + }, deps: Vec::new(), parent: Some(3), }]; @@ -560,7 +564,6 @@ fn append_subgraph_roots_on_emitter_and_rebases_local_deps() { perm_payload: None, transient: None, nodes: vec![NodeSpec { - agent: "hyperhive".to_owned(), kind: NodeKind::MetaLock { sweep: true, fanout: None, diff --git a/hive-c0re/src/workers/auto_update.rs b/hive-c0re/src/workers/auto_update.rs index c95ab682..2ed340e1 100644 --- a/hive-c0re/src/workers/auto_update.rs +++ b/hive-c0re/src/workers/auto_update.rs @@ -325,7 +325,6 @@ fn submit_boot_tree( // MetaLock into `run_meta_lock`, which appends the rebuild subgraphs. if any_stale { nodes.push(NodeSpec { - agent: "hyperhive".to_owned(), kind: NodeKind::MetaLock { sweep: true, fanout: Some(fanout), @@ -337,8 +336,7 @@ fn submit_boot_tree( // One boot Reconcile per drifted agent — independent roots. for name in drifted { nodes.push(NodeSpec { - agent: name, - kind: NodeKind::Reconcile, + kind: NodeKind::Reconcile { agent: name }, deps: Vec::new(), parent: None, });