diff --git a/hive-c0re/src/actions.rs b/hive-c0re/src/actions.rs index 7333f387..8c531076 100644 --- a/hive-c0re/src/actions.rs +++ b/hive-c0re/src/actions.rs @@ -55,13 +55,12 @@ pub async fn approve(coord: Arc, id: i64) -> Result<()> { // nothing). let inputs: Vec = serde_json::from_str(&approval.commit_ref).unwrap_or_default(); - let submitted = coord.job_queue.submit( - crate::job_queue::Source::Approval, - format!("approval #{id} meta input update"), - |b| crate::job_queue::templates::meta_update(b, inputs, Some(id)), - ); - if let Err(e) = submitted { - return Err(e.context("submit meta-update dag")); + let inserted = coord.job_queue.insert(|b| { + crate::job_queue::templates::meta_update(b, inputs, Some(id)); + Vec::new() + }); + if let Err(e) = inserted { + return Err(e.context("insert meta-update dag")); } coord.emit_rebuild_queue_snapshot(); Ok(()) @@ -75,13 +74,12 @@ pub async fn approve(coord: Arc, id: i64) -> Result<()> { { tracing::warn!(agent = %approval.agent, error = ?e, "agent_power: seed on spawn failed"); } - let submitted = coord.job_queue.submit( - crate::job_queue::Source::Approval, - format!("approval #{id} spawn"), - |b| crate::job_queue::templates::spawn(b, approval.agent.as_str(), id), - ); - if let Err(e) = submitted { - return Err(e.context("submit spawn dag")); + let inserted = coord.job_queue.insert(|b| { + crate::job_queue::templates::spawn(b, approval.agent.as_str(), id); + Vec::new() + }); + if let Err(e) = inserted { + return Err(e.context("insert spawn dag")); } coord.emit_rebuild_queue_snapshot(); Ok(()) @@ -106,12 +104,7 @@ pub async fn approve(coord: Arc, id: i64) -> Result<()> { // `run_deploy_apply` (ff-merge, then grows the rebuild subgraph), // `run_finalize_deploy` (deploy tag + lock commit) and // `run_deploy_tail` (compensation + forge mirror). - enqueue_approval_rebuild( - &coord, - approval.agent.as_str(), - id, - format!("approval #{id} merge config pr"), - ); + enqueue_approval_rebuild(&coord, approval.agent.as_str(), id); Ok(()) } } @@ -121,19 +114,12 @@ pub async fn approve(coord: Arc, id: i64) -> Result<()> { /// dispatch arm — the work ends in a container rebuild routed through the /// queue. See [`crate::job_queue::templates::approval_deploy`] for the node /// shape; the executor dispatches each node to the `run_deploy_*` bodies below. -fn enqueue_approval_rebuild( - coord: &Arc, - agent: &str, - approval_id: i64, - reason: String, -) { - if let Err(e) = coord - .job_queue - .submit(crate::job_queue::Source::Approval, reason, |b| { - crate::job_queue::templates::approval_deploy(b, agent, approval_id); - }) - { - tracing::error!(%agent, approval_id, error = ?e, "submit approval deploy dag failed"); +fn enqueue_approval_rebuild(coord: &Arc, agent: &str, approval_id: i64) { + if let Err(e) = coord.job_queue.insert(|b| { + crate::job_queue::templates::approval_deploy(b, agent, approval_id); + Vec::new() + }) { + tracing::error!(%agent, approval_id, error = ?e, "insert approval deploy dag failed"); } coord.emit_rebuild_queue_snapshot(); } diff --git a/hive-c0re/src/dashboard/meta_inputs.rs b/hive-c0re/src/dashboard/meta_inputs.rs index 3efe6b74..da1f3baa 100644 --- a/hive-c0re/src/dashboard/meta_inputs.rs +++ b/hive-c0re/src/dashboard/meta_inputs.rs @@ -208,16 +208,18 @@ pub(super) async fn post_meta_update( if inputs.is_empty() { return error_response("meta-update: no inputs selected"); } - let inputs_label = inputs.join(", "); // Cascade rebuild children fan out from the MetaLock node when the // lock bump lands — appended by the scheduler so they build against // the post-bump lock, and a failed bump simply fans out nothing. - crate::job_queue::submit::meta_update( - &state.coord, - inputs, - crate::job_queue::Source::Manual, - format!("meta-update via dashboard ({inputs_label})"), - ); + state + .coord + .job_queue + .insert(|b| { + crate::job_queue::templates::meta_update(b, inputs, None); + Vec::new() + }) + .expect("template-declared shapes are acyclic"); + state.coord.emit_rebuild_queue_snapshot(); (StatusCode::OK, "ok").into_response() } diff --git a/hive-c0re/src/dashboard/permissions.rs b/hive-c0re/src/dashboard/permissions.rs index 601ca400..11f920d3 100644 --- a/hive-c0re/src/dashboard/permissions.rs +++ b/hive-c0re/src/dashboard/permissions.rs @@ -155,15 +155,21 @@ pub(super) async fn post_tool_groups( // META_LOCK inside the WritePermFile node, so concurrent // batch-apply actions for different agents never race on the // shared tool-groups.json. - crate::job_queue::submit::perm_change( - &state.coord, - &logical, - crate::job_queue::Source::Manual, - "tool-group change via permissions UI".to_owned(), - crate::job_queue::PermPayload::ToolGroups { - groups: body.groups.clone(), - }, - ); + state + .coord + .job_queue + .insert(|b| { + crate::job_queue::templates::perm_change( + b, + &logical, + crate::job_queue::PermPayload::ToolGroups { + groups: body.groups.clone(), + }, + ); + Vec::new() + }) + .expect("template-declared shapes are acyclic"); + state.coord.emit_rebuild_queue_snapshot(); tracing::info!(agent = %logical, groups = ?body.groups, "operator: set tool-groups via dashboard"); Ok((StatusCode::OK, "ok").into_response()) } @@ -264,15 +270,21 @@ pub(super) async fn post_capabilities( // META_LOCK inside the WritePermFile node, so concurrent // batch-apply actions for different agents never race on the // shared capabilities.json. - crate::job_queue::submit::perm_change( - &state.coord, - &logical, - crate::job_queue::Source::Manual, - "capability change via dashboard".to_owned(), - crate::job_queue::PermPayload::Capabilities { - caps: body.caps.clone(), - }, - ); + state + .coord + .job_queue + .insert(|b| { + crate::job_queue::templates::perm_change( + b, + &logical, + crate::job_queue::PermPayload::Capabilities { + caps: body.caps.clone(), + }, + ); + Vec::new() + }) + .expect("template-declared shapes are acyclic"); + state.coord.emit_rebuild_queue_snapshot(); tracing::info!(agent = %logical, caps = ?body.caps, "operator: set capabilities via dashboard"); Ok((StatusCode::OK, "ok").into_response()) } @@ -359,13 +371,19 @@ pub(super) async fn post_permissions( } // Phase 2 — submit one combined PermChange DAG per affected agent. for (logical, groups, caps) in staged { - crate::job_queue::submit::perm_change( - &state.coord, - &logical, - crate::job_queue::Source::Manual, - "batch permission change via permissions UI".to_owned(), - crate::job_queue::PermPayload::Combined { groups, caps }, - ); + state + .coord + .job_queue + .insert(|b| { + crate::job_queue::templates::perm_change( + b, + &logical, + crate::job_queue::PermPayload::Combined { groups, caps }, + ); + Vec::new() + }) + .expect("template-declared shapes are acyclic"); + state.coord.emit_rebuild_queue_snapshot(); tracing::info!(agent = %logical, "operator: batch perm change via dashboard"); } Ok((StatusCode::OK, "ok").into_response()) diff --git a/hive-c0re/src/dashboard/topology.rs b/hive-c0re/src/dashboard/topology.rs index 8e8422fb..d007f446 100644 --- a/hive-c0re/src/dashboard/topology.rs +++ b/hive-c0re/src/dashboard/topology.rs @@ -94,12 +94,15 @@ pub(super) async fn post_set_parent( new_parent = ?new_parent, "operator: set-parent via dashboard" ); - submit::reparent( - &state.coord, - vec![(child, new_parent)], - Source::Manual, - "manual set-parent via dashboard".to_owned(), - ); + state + .coord + .job_queue + .insert(|b| { + crate::job_queue::templates::reparent(b, vec![(child, new_parent)]); + Vec::new() + }) + .expect("template-declared shapes are acyclic"); + state.coord.emit_rebuild_queue_snapshot(); Ok((StatusCode::OK, "ok").into_response()) } @@ -150,11 +153,14 @@ pub(super) async fn post_set_parent_bulk( .map_err(|e| error_problem(&e))?; let names: Vec<&str> = body.iter().map(|e| e.child.as_str()).collect(); tracing::info!(agents = ?names, "operator: set-parent-bulk via dashboard"); - submit::reparent( - &state.coord, - moves, - Source::Manual, - "manual set-parent-bulk via dashboard".to_owned(), - ); + state + .coord + .job_queue + .insert(|b| { + crate::job_queue::templates::reparent(b, moves); + Vec::new() + }) + .expect("template-declared shapes are acyclic"); + state.coord.emit_rebuild_queue_snapshot(); Ok((StatusCode::OK, "ok").into_response()) } diff --git a/hive-c0re/src/job_queue/mod.rs b/hive-c0re/src/job_queue/mod.rs index b9727493..655ad636 100644 --- a/hive-c0re/src/job_queue/mod.rs +++ b/hive-c0re/src/job_queue/mod.rs @@ -29,9 +29,9 @@ pub mod exec; pub mod model; +pub mod power; pub mod resource; pub mod scheduler; -pub mod submit; pub mod templates; #[cfg(test)] mod tests; @@ -208,20 +208,17 @@ impl JobQueue { /// # Errors /// Propagates a graph-insert error (dependencies that aren't /// dependency-topological). - pub fn submit( + pub fn insert( &self, - source: Source, - reason: String, - declare: impl FnOnce(&JobBuilder), - ) -> anyhow::Result { + declare: impl FnOnce(&JobBuilder) -> Vec, + ) -> anyhow::Result> { let mut inner = self.lock(); - let container = inner - .append(NodeKind::Dag { source, reason }, Vec::new(), None) - .map_err(|e| anyhow::anyhow!("job_queue: container insert failed: {e}"))?; - insert_group(&mut inner, declare, Some(container))?; + let named = inner + .insert_job(None, declare) + .map_err(|e| anyhow::anyhow!("job_queue: graph insert failed: {e}"))?; drop(inner); self.notify.notify_one(); - Ok(container.get()) + Ok(named) } /// The scheduler itself, for `hive_jobq`'s run-loop seam diff --git a/hive-c0re/src/job_queue/submit.rs b/hive-c0re/src/job_queue/power.rs similarity index 56% rename from hive-c0re/src/job_queue/submit.rs rename to hive-c0re/src/job_queue/power.rs index 3ec3e11b..1bb26b95 100644 --- a/hive-c0re/src/job_queue/submit.rs +++ b/hive-c0re/src/job_queue/power.rs @@ -1,59 +1,26 @@ -//! Request-level submit API — the surface the dashboard POST handlers, -//! the MCP socket handlers, and `hivectl` paths call. +//! Power ops (`stop` / `start` / `restart`) — the DAG shapes whose per-agent +//! form depends on **live** container state, so they cannot be static +//! [`super::templates`] entries. //! -//! The **power ops** (`stop` / `start` / `restart`) are built here, not in -//! `templates.rs`: each agent's subgraph shape depends on its *live* running -//! state, which needs an async `lifecycle::is_running` read that a pure/sync -//! template can't do. So these fns are async — they read each agent's state, -//! assemble a per-agent subgraph out of the shared pure primitives -//! (`JobBuilder::node` + `templates::rebuild_nodes`), all declaring into ONE job -//! (independent per-agent roots, concurrent on their own leases). +//! The split is the purity line, not the subject matter: the `*_chain` / +//! `*_nodes` builders below are pure (they take `running` / `stale` as +//! parameters, which is what keeps them unit-testable without a container), +//! and only the `*_many` entry points do the async `lifecycle::is_running` +//! read that produces those parameters. //! -//! Dynamic shape rule: `stop`/`start` carry a head `SetWanted(w)` (durable -//! intent write) — `restart` does NOT (it bounces the container but leaves -//! `wanted` untouched, so a deliberately-stopped agent isn't forced up). The -//! tail `Reconcile` (the convergence guarantee — cheap, noops when already -//! converged) is ALWAYS present; only the *mechanical* nodes -//! (`Signal`/`Drain`/`StopForUpdate`) are state-conditional (skipped for a -//! down agent — nothing to quiesce/stop). Keeping `Reconcile` in every shape -//! closes the TOCTOU window: if an agent flips state between the `is_running` -//! read and node execution, the tail `Reconcile` still converges it in-DAG, -//! with `StopForUpdate`-noop as the backstop — no reliance on an external -//! reconcile sweep. Every helper emits a fresh queue snapshot so the -//! dashboard shows the new DAG immediately. +//! Each entry point declares its group and inserts it. There is no metadata +//! parameter and no container node: attribution is not something every caller +//! has to invent, and a DAG is addressed by the nodes a template names. use std::sync::Arc; -use super::model::NodeKind; -use super::resource::Resource; -use super::templates::rebuild_nodes; -use super::{JobBuilder, Source, templates}; +use hive_jobq::NodeId; + +use super::JobBuilder; +use super::templates; use crate::coordinator::Coordinator; use crate::lifecycle; -fn submit_and_emit( - coord: &Arc, - source: Source, - reason: String, - declare: impl FnOnce(&JobBuilder), -) -> u64 { - let id = coord - .job_queue - .submit(source, reason, declare) - .expect("template-declared shapes are acyclic"); - coord.emit_rebuild_queue_snapshot(); - id -} - -/// Manual/approval-independent rebuild (always relocks the agent's -/// meta input — the meta-update cascade grows its own rebuild subgraphs -/// in-DAG instead of going through this surface). -pub fn rebuild(coord: &Arc, agent: &str, source: Source, reason: String) -> u64 { - submit_and_emit(coord, source, reason, |builder| { - templates::rebuild(builder, agent, true); - }) -} - // ---- dynamic power-op DAG assembly ---------------------------------------- // // The pure per-agent chain builders below take `running` (and `stale`) @@ -226,63 +193,44 @@ pub(crate) fn restart_nodes(builder: &JobBuilder, targets: &[(String, bool)], gr } } -/// Restart a single agent. Thin wrapper over [`restart_many`]. -pub async fn restart(coord: &Arc, agent: &str, source: Source, reason: String) -> u64 { - restart_many(coord, &[agent.to_owned()], false, source, reason).await -} +// ---- entry points --------------------------------------------------------- +// +// Each reads the live state its shape depends on, then declares + inserts. -/// Graceful restart of a single agent (signal → drain → stop → reconcile, -/// when running). Thin wrapper over [`restart_many`] with `graceful = true`. -pub async fn graceful_restart( - coord: &Arc, - agent: &str, - source: Source, - reason: String, -) -> u64 { - restart_many(coord, &[agent.to_owned()], true, source, reason).await -} - -/// Restart `agents` (one or many) in a **single** DAG — one per-agent -/// subgraph each, built dynamically from live running state and run -/// concurrently on their own leases. A running agent gets the stop→reconcile -/// chain (`graceful` prepends signal→drain); a down agent gets just a lone -/// `Reconcile` (nothing to stop). Restart never writes `wanted`, so the +/// Restart `agents` in a **single** DAG — one per-agent subgraph each, built +/// from live running state and run concurrently on their own leases. A running +/// agent gets the stop→reconcile chain (`graceful` prepends signal→drain); a +/// down agent gets a lone `Reconcile`. Restart never writes `wanted`, so the /// tail `Reconcile` converges each agent to its EXISTING intent — a -/// deliberately-stopped agent stays down. The whole hive-wide -/// `hivectl restart` is one DAG. +/// deliberately-stopped agent stays down. +/// +/// # Errors +/// Propagates a graph-insert error. pub async fn restart_many( coord: &Arc, agents: &[String], graceful: bool, - source: Source, - reason: String, -) -> u64 { +) -> anyhow::Result> { let mut targets = Vec::with_capacity(agents.len()); for agent in agents { targets.push((agent.clone(), lifecycle::is_running(agent).await)); } - submit_and_emit(coord, source, reason, |builder| { - restart_nodes(builder, &targets, graceful); - }) + let ids = coord.job_queue.insert(|b| { + restart_nodes(b, &targets, graceful); + Vec::new() + })?; + coord.emit_rebuild_queue_snapshot(); + Ok(ids) } -/// Start a single agent. Thin wrapper over [`start_many`]. -pub async fn start(coord: &Arc, agent: &str, source: Source, reason: String) -> u64 { - start_many(coord, &[agent.to_owned()], source, reason).await -} - -/// Start `agents` (one or many) in a **single** DAG — one per-agent subgraph -/// each, built dynamically from live state and run concurrently on their own -/// leases. A down agent gets `SetWanted(Up) → Reconcile` (or, rev stale, a -/// rebuild-then-start so it comes up on current derivations); an already- -/// running agent gets `SetWanted(Up) → Reconcile` (the reconcile noops). The -/// whole hive-wide `hivectl start` is one DAG. -pub async fn start_many( - coord: &Arc, - agents: &[String], - source: Source, - reason: String, -) -> u64 { +/// Start `agents` in a **single** DAG. A down agent gets +/// `SetWanted(Up) → Reconcile` (or, rev stale, a rebuild-then-start so it comes +/// up on current derivations); an already-running agent gets the same shape +/// with the reconcile noop'ing. +/// +/// # Errors +/// Propagates a graph-insert error. +pub async fn start_many(coord: &Arc, agents: &[String]) -> anyhow::Result> { let current = crate::auto_update::current_flake_rev(&coord.hyperhive_flake); let mut targets = Vec::with_capacity(agents.len()); for agent in agents { @@ -296,88 +244,34 @@ pub async fn start_many( } targets.push((agent.clone(), running, stale)); } - submit_and_emit(coord, source, reason, |builder| { - start_nodes(builder, &targets); - }) + let ids = coord.job_queue.insert(|b| { + start_nodes(b, &targets); + Vec::new() + })?; + coord.emit_rebuild_queue_snapshot(); + Ok(ids) } -/// Hard stop a single agent. Thin wrapper over [`stop_many`]. -pub async fn stop(coord: &Arc, agent: &str, source: Source, reason: String) -> u64 { - stop_many(coord, &[agent.to_owned()], false, source, reason).await -} - -/// Graceful stop of a single agent (signal → drain → reconcile, when -/// running). Thin wrapper over [`stop_many`] with `graceful = true`. -pub async fn graceful_stop( - coord: &Arc, - agent: &str, - source: Source, - reason: String, -) -> u64 { - stop_many(coord, &[agent.to_owned()], true, source, reason).await -} - -/// Stop `agents` (one or many) in a **single** DAG — one per-agent subgraph -/// each, built dynamically from live state and run concurrently on their own -/// leases. A running agent gets `SetWanted(Off) → [Signal → Drain →](graceful) -/// Reconcile`; a down agent gets just `SetWanted(Off) → Reconcile` (skips the -/// pointless quiesce, keeps the Reconcile as the race-up backstop). The whole -/// hive-wide `hivectl stop` is one DAG. +/// Stop `agents` in a **single** DAG. A running agent gets +/// `SetWanted(Off) → [Signal → Drain →](graceful) Reconcile`; a down agent +/// skips the pointless quiesce but keeps the `Reconcile` as the race-up +/// backstop. +/// +/// # Errors +/// Propagates a graph-insert error. pub async fn stop_many( coord: &Arc, agents: &[String], graceful: bool, - source: Source, - reason: String, -) -> u64 { +) -> anyhow::Result> { let mut targets = Vec::with_capacity(agents.len()); for agent in agents { targets.push((agent.clone(), lifecycle::is_running(agent).await)); } - submit_and_emit(coord, source, reason, |builder| { - stop_nodes(builder, &targets, graceful); - }) -} - -/// Perm change: commit the JSON file(s) then rebuild. -pub fn perm_change( - coord: &Arc, - agent: &str, - source: Source, - reason: String, - payload: super::PermPayload, -) -> u64 { - submit_and_emit(coord, source, reason, |builder| { - templates::perm_change(builder, agent, payload); - }) -} - -/// Meta-input lock bump; cascade rebuilds fan out on completion. -pub fn meta_update( - coord: &Arc, - inputs: Vec, - source: Source, - reason: String, -) -> u64 { - submit_and_emit(coord, source, reason, |builder| { - templates::meta_update(builder, inputs, None); - }) -} - -/// Topology move(s) as a queue DAG. `moves` is `(child, new_parent)` pairs — -/// one entry for `set-parent`, N for `set-parent-bulk`. Fire-and-forget like -/// everything else in this module: submits and returns a DAG id -/// immediately, the caller learns the outcome async (dashboard job view / -/// `hivectl`'s `QueueDag` poll). Wired from `server.rs`'s `HostRequest:: -/// SetParent` (hivectl) and `dashboard/topology.rs`'s `set-parent`/ -/// `set-parent-bulk` handlers. -pub fn reparent( - coord: &Arc, - moves: Vec<(hive_types::Ident, Option)>, - source: Source, - reason: String, -) -> u64 { - submit_and_emit(coord, source, reason, |builder| { - templates::reparent(builder, moves); - }) + let ids = coord.job_queue.insert(|b| { + stop_nodes(b, &targets, graceful); + Vec::new() + })?; + coord.emit_rebuild_queue_snapshot(); + Ok(ids) } diff --git a/hive-c0re/src/job_queue/tests.rs b/hive-c0re/src/job_queue/tests.rs index 4bc199a9..87d17dca 100644 --- a/hive-c0re/src/job_queue/tests.rs +++ b/hive-c0re/src/job_queue/tests.rs @@ -36,17 +36,17 @@ fn rebuild(builder: &JobBuilder, agent: &str) { /// Restart shape with every agent treated as **running** — the online /// shape (`[Signal→Drain→] StopForUpdate → Reconcile`, no `SetWanted` head) /// most queue-mechanics tests assume. Mirrors the pre-dynamic -/// `templates::restart` (which is now the state-aware `submit::restart_nodes`). +/// `templates::restart` (which is now the state-aware `power::restart_nodes`). fn restart_online(builder: &JobBuilder, agents: &[&str], graceful: bool) { let targets: Vec<(String, bool)> = agents.iter().map(|a| ((*a).to_owned(), true)).collect(); - submit::restart_nodes(builder, &targets, graceful); + power::restart_nodes(builder, &targets, graceful); } /// Stop shape with every agent treated as **running** — the online shape /// (`SetWanted → [Signal→Drain→](graceful) Reconcile`). fn stop_online(builder: &JobBuilder, agents: &[&str], graceful: bool) { let targets: Vec<(String, bool)> = agents.iter().map(|a| ((*a).to_owned(), true)).collect(); - submit::stop_nodes(builder, &targets, graceful); + power::stop_nodes(builder, &targets, graceful); } // `Claimed` / `ClaimReady` / `CompleteNode` lived here: a claim snapshot type @@ -682,7 +682,7 @@ fn multi_agent_start_one_dag_folds_per_agent_stale_rebuild() { // fresh: offline + not stale → SetWanted → Reconcile. // stale: offline + stale → SetWanted → «rebuild subgraph». let id = submit(&q, "hive-wide start", |builder| { - submit::start_nodes( + power::start_nodes( builder, &[ ("fresh".to_owned(), false, false), @@ -742,13 +742,13 @@ fn offline_agents_skip_mechanical_nodes_but_keep_reconcile() { let q = JobQueue::new(4); // Offline graceful stop → SetWanted(Off) → Reconcile (no Signal/Drain). let stop = submit(&q, "stop down", |builder| { - submit::stop_nodes(builder, &[("down".to_owned(), false)], true); + power::stop_nodes(builder, &[("down".to_owned(), false)], true); }); // Offline restart → a lone Reconcile (no SetWanted, no StopForUpdate): // nothing to bounce, and restart never rewrites intent, so the tail // Reconcile converges the down agent to its existing `wanted`. let restart = submit(&q, "restart down", |builder| { - submit::restart_nodes(builder, &[("down2".to_owned(), false)], true); + power::restart_nodes(builder, &[("down2".to_owned(), false)], true); }); let shape = |id: u64| -> Vec { // The group's work nodes: its subtree minus the container itself, @@ -1153,19 +1153,19 @@ fn cancelled_power_op_runs_no_compensating_node() { let q = JobQueue::new(1); let id = submit(&q, "bounce", |builder| { - submit::restart_nodes(builder, &targets, graceful); + power::restart_nodes(builder, &targets, graceful); }); assert_cancels_clean(&q, id, false, &format!("restart {case}")); let q = JobQueue::new(1); let id = submit(&q, "stop", |builder| { - submit::stop_nodes(builder, &targets, graceful); + power::stop_nodes(builder, &targets, graceful); }); assert_cancels_clean(&q, id, true, &format!("stop {case}")); let q = JobQueue::new(1); let id = submit(&q, "start", |builder| { - submit::start_nodes(builder, &[("agent-a".to_owned(), running, false)]); + power::start_nodes(builder, &[("agent-a".to_owned(), running, false)]); }); assert_cancels_clean(&q, id, true, &format!("start {case}")); } diff --git a/hive-c0re/src/socket_server/lifecycle_handlers.rs b/hive-c0re/src/socket_server/lifecycle_handlers.rs index 46446c12..968d31a3 100644 --- a/hive-c0re/src/socket_server/lifecycle_handlers.rs +++ b/hive-c0re/src/socket_server/lifecycle_handlers.rs @@ -20,13 +20,9 @@ pub(super) async fn handle_start(coord: &Arc, agent: &str, name: &s // Persist `wanted = Up` and submit the Start DAG; the submit layer // upgrades a stale-rev start to a full rebuild so the container // runs current nix derivations before it starts. - crate::job_queue::submit::start( - coord, - name, - crate::job_queue::Source::Manual, - format!("agent `{agent}` start tool"), - ) - .await; + if let Err(e) = crate::job_queue::power::start_many(coord, &[name.to_owned()]).await { + tracing::error!(%agent, %name, error = ?e, "start: insert failed"); + } Response::Ok } @@ -48,13 +44,9 @@ pub(super) async fn handle_restart(coord: &Arc, agent: &str, name: return err; } tracing::info!(%agent, %name, "submit restart"); - crate::job_queue::submit::restart( - coord, - name, - crate::job_queue::Source::Manual, - format!("agent `{agent}` restart tool"), - ) - .await; + if let Err(e) = crate::job_queue::power::restart_many(coord, &[name.to_owned()], false).await { + tracing::error!(%agent, %name, error = ?e, "restart: insert failed"); + } Response::Ok } @@ -153,12 +145,13 @@ pub(super) fn handle_update(coord: &Arc, agent: &str, name: &str) - return err; } tracing::info!(%agent, %name, "submit rebuild"); - crate::job_queue::submit::rebuild( - coord, - name, - crate::job_queue::Source::Manual, - format!("agent `{agent}` update tool"), - ); + if let Err(e) = coord.job_queue.insert(|b| { + crate::job_queue::templates::rebuild(b, name, true); + Vec::new() + }) { + tracing::error!(%agent, %name, error = ?e, "update: insert failed"); + } + coord.emit_rebuild_queue_snapshot(); Response::Ok } diff --git a/hive-c0re/src/workers/auto_update.rs b/hive-c0re/src/workers/auto_update.rs index 8132f0b7..a629e73f 100644 --- a/hive-c0re/src/workers/auto_update.rs +++ b/hive-c0re/src/workers/auto_update.rs @@ -109,12 +109,11 @@ pub async fn ensure_root_agent(coord: &Arc) -> Result<()> { tracing::warn!( "manager container exists but no applied flake — forcing rebuild to migrate" ); - if let Err(e) = coord.job_queue.submit( - crate::job_queue::Source::AutoUpdate, - "manager migration: no applied flake".to_owned(), - |b| crate::job_queue::templates::rebuild(b, MANAGER_NAME, true), - ) { - tracing::warn!(error = ?e, "manager migration rebuild submit failed"); + if let Err(e) = coord.job_queue.insert(|b| { + crate::job_queue::templates::rebuild(b, MANAGER_NAME, true); + Vec::new() + }) { + tracing::warn!(error = ?e, "manager migration rebuild insert failed"); } } else { tracing::debug!("manager container already present"); @@ -378,18 +377,19 @@ fn submit_boot_tree( n_deferred: usize, n_skipped: usize, ) { - use crate::job_queue::Source; - - // Fully-quiet boot (nothing stale, nothing drifted) submits nothing. + // Fully-quiet boot (nothing stale, nothing drifted) inserts nothing. if !any_stale && drifted.is_empty() { return; } - let reason = format!( - "boot: {} rebuild(s), {} reconcile(s), {} deferred (offline), {} up-to-date", - fanout.len(), - drifted.len(), - n_deferred, - n_skipped, + // The summary the sweep used to hand the container as its `reason` is a log + // line now: it was only ever stored on a node nobody read, and the counts + // are worth having where they can actually be seen. + tracing::info!( + rebuilds = fanout.len(), + reconciles = drifted.len(), + deferred = n_deferred, + up_to_date = n_skipped, + "boot: sweep" ); // The sweep's own rebuild subgraphs emit their `Rebuilt` events as they @@ -397,10 +397,11 @@ fn submit_boot_tree( // The subgraphs also carry their own per-agent crash-watch suppression // during their `Swap` (applied at claim time); a reconcile-only boot needs // no transient. - if let Err(e) = coord.job_queue.submit(Source::AutoUpdate, reason, |b| { + if let Err(e) = coord.job_queue.insert(|b| { boot_nodes(b, any_stale, fanout, drifted); + Vec::new() }) { - tracing::warn!(error = ?e, "boot: sweep DAG submit failed"); + tracing::warn!(error = ?e, "boot: sweep DAG insert failed"); } coord.emit_rebuild_queue_snapshot(); }