vinzenz/hyperhive
vinzenz/hyperhive
Watch
0
0
Fork
You've already forked hyperhive
 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions 2

Compare commits

base: vinzenz:ddc017f01b230a171268778553913765e9561e23
Branches Tags
vinzenz:main
..
compare: vinzenz:6f87821110af052bcf243dc7b2778ae2cc853c4a
Branches Tags
vinzenz:main
17 changed files with 880 additions and 791 deletions
Download patch file Download diff file Expand all files Collapse all files

8
docs/coordinator.md
View file

@ -106,7 +106,7 @@ subgraph each (independent roots, run concurrently on their own leases), not
N separate DAGs.
**These are built dynamically from each agent's live running state** (an
async `lifecycle::is_running` read), so they live in `job_queue/power.rs`,
async `lifecycle::is_running` read), so they live in `job_queue/submit.rs`,
not the pure/sync `templates.rs`. Per-agent shape rule: `stop`/`start` carry
a head `SetWanted` (intent) — `restart` does not; the tail `Reconcile`
(convergence guarantee — cheap, noops when already converged) is ALWAYS
@ -161,12 +161,12 @@ Notable collapses:
Per-agent power *intent* — `wanted: Up | Offline` — is durable as the
`agent_power` table in the coordinator DB (`hive-c0re/src/stores/power.rs`).
`container_view` remains the observed *status*; `Reconcile` nodes converge the
two. Setting `wanted` is never a queued node: the power layer
(`job_queue/power.rs`) writes the row synchronously, then inserts the DAG
two. Setting `wanted` is never a queued node: the submit layer
(`job_queue/submit.rs`) writes the row synchronously, then submits the DAG
whose `Reconcile` reads the fresh value — rapid toggles are last-writer-wins.
Power toggles never commit to the meta repo. Every operator power surface —
dashboard buttons, the MCP tools, and `hivectl stop/start/restart/kill` —
rides the queue through that power layer, so intent, lease serialization,
rides the queue through that submit layer, so intent, lease serialization,
and crash-watch suppression can't drift per surface; the only direct starts
left are the root-agent bootstrap and infra containers (no lease, no
harness). Cancelling a still-queued power DAG reverts `wanted` to the

52
hive-c0re/src/actions.rs
View file

@ -55,12 +55,13 @@ pub async fn approve(coord: Arc<Coordinator>, id: i64) -> Result<()> {
// nothing).
let inputs: Vec<String> =
serde_json::from_str(&approval.commit_ref).unwrap_or_default();
let inserted = coord.job_queue.insert_job(|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"));
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"));
}
coord.emit_rebuild_queue_snapshot();
Ok(())
@ -74,12 +75,13 @@ pub async fn approve(coord: Arc<Coordinator>, id: i64) -> Result<()> {
{
tracing::warn!(agent = %approval.agent, error = ?e, "agent_power: seed on spawn failed");
}
let inserted = coord.job_queue.insert_job(|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"));
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"));
}
coord.emit_rebuild_queue_snapshot();
Ok(())
@ -104,7 +106,12 @@ pub async fn approve(coord: Arc<Coordinator>, 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);
enqueue_approval_rebuild(
&coord,
approval.agent.as_str(),
id,
format!("approval #{id} merge config pr"),
);
Ok(())
}
}
@ -114,12 +121,19 @@ pub async fn approve(coord: Arc<Coordinator>, 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<Coordinator>, agent: &str, approval_id: i64) {
if let Err(e) = coord.job_queue.insert_job(|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");
fn enqueue_approval_rebuild(
coord: &Arc<Coordinator>,
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");
}
coord.emit_rebuild_queue_snapshot();
}

101
hive-c0re/src/dashboard/lifecycle_ops.rs
View file

@ -1,12 +1,11 @@
//! Container lifecycle endpoints for the dashboard.
//!
//! Rebuild / restart / start / stop (hard + graceful) / update-all all
//! insert DAGs into the job queue — the power ops via
//! [`crate::job_queue::power`], the static shapes straight through
//! `JobQueue::insert` — so each shows a visible queued→running transient on
//! the dashboard; a direct sub-second start/stop only flashed the badge.
//! Start/stop also persist the agent's `wanted` power intent before
//! inserting; the DAG's `Reconcile` converges to it. Destroy delegates to
//! submit DAGs to the job queue (`job_queue::submit`), so each shows a
//! visible queued→running transient on the dashboard — a direct
//! sub-second start/stop only flashed the badge. Start/stop also
//! persist the agent's `wanted` power intent before submitting; the
//! DAG's `Reconcile` converges to it. Destroy delegates to
//! `actions::destroy` (optionally purging).
use axum::{
@ -28,6 +27,7 @@ pub(super) struct GracefulParams {
}
use super::{AppState, Ident, error_response, guard_agent_name, strip_container_prefix};
use crate::job_queue::{Source, submit};
use crate::{actions, lifecycle};
/// Queue a rebuild DAG for `name`.
@ -50,13 +50,12 @@ pub(super) async fn post_rebuild(
if let Some(reject) = guard_agent_name(&state, &logical).await {
return reject;
}
if let Err(e) = state.coord.job_queue.insert_job(|b| {
crate::job_queue::templates::rebuild(b, &logical, true);
Vec::new()
}) {
tracing::error!(agent = %logical, error = ?e, "rebuild: insert failed");
}
state.coord.emit_rebuild_queue_snapshot();
submit::rebuild(
&state.coord,
&logical,
Source::Manual,
"manual via dashboard ↻ R3BU1LD button".to_owned(),
);
(StatusCode::OK, "ok").into_response()
}
@ -93,12 +92,13 @@ pub(super) async fn post_kill(
// timeout fallback to a hard stop). The agent's lifecycle
// lease keeps it from racing an in-flight rebuild for the same
// agent, and per-node progress surfaces on the queue snapshot.
if let Err(e) =
crate::job_queue::power::stop_many(&state.coord, std::slice::from_ref(&logical), true)
.await
{
tracing::error!(agent = %logical, error = ?e, "graceful stop: insert failed");
}
submit::graceful_stop(
&state.coord,
&logical,
Source::Manual,
"manual via dashboard graceful stop".to_owned(),
)
.await;
return (StatusCode::OK, "ok").into_response();
}
// Manager is stoppable from the dashboard like any other
@ -111,12 +111,13 @@ pub(super) async fn post_kill(
// `socket_server.rs::Request::Kill` stays in place: a
// manager calling Kill on its own container is self-suicide
// mid-call, not a legitimate operator action.
if let Err(e) =
crate::job_queue::power::stop_many(&state.coord, std::slice::from_ref(&logical), false)
.await
{
tracing::error!(agent = %logical, error = ?e, "stop: insert failed");
}
submit::stop(
&state.coord,
&logical,
Source::Manual,
"manual via dashboard stop".to_owned(),
)
.await;
(StatusCode::OK, "ok").into_response()
}
@ -148,23 +149,22 @@ pub(super) async fn post_restart(
return reject;
}
if params.graceful {
if let Err(e) = crate::job_queue::power::restart_many(
submit::graceful_restart(
&state.coord,
std::slice::from_ref(&logical),
true,
&logical,
Source::Manual,
"manual via dashboard graceful restart".to_owned(),
)
.await
{
tracing::error!(agent = %logical, error = ?e, "graceful restart: insert failed");
}
.await;
return (StatusCode::OK, "ok").into_response();
}
if let Err(e) =
crate::job_queue::power::restart_many(&state.coord, std::slice::from_ref(&logical), false)
.await
{
tracing::error!(agent = %logical, error = ?e, "restart: insert failed");
}
submit::restart(
&state.coord,
&logical,
Source::Manual,
"manual via dashboard ↺ R3START button".to_owned(),
)
.await;
(StatusCode::OK, "ok").into_response()
}
@ -226,11 +226,13 @@ pub(super) async fn post_start(
return (StatusCode::OK, "ok").into_response();
}
}
if let Err(e) =
crate::job_queue::power::start_many(&state.coord, std::slice::from_ref(&logical)).await
{
tracing::error!(agent = %logical, error = ?e, "start: insert failed");
}
submit::start(
&state.coord,
&logical,
Source::Manual,
"manual via dashboard start".to_owned(),
)
.await;
(StatusCode::OK, "ok").into_response()
}
@ -409,14 +411,13 @@ pub(super) async fn post_update_all(State(state): State<AppState>) -> Response {
else {
continue;
};
if let Err(e) = state.coord.job_queue.insert_job(|b| {
crate::job_queue::templates::rebuild(b, &logical, true);
Vec::new()
}) {
tracing::error!(agent = %logical, error = ?e, "update-all: insert failed");
}
submit::rebuild(
&state.coord,
&logical,
Source::Manual,
"manual via dashboard 🌀 UPDATE ALL".to_owned(),
);
}
state.coord.emit_rebuild_queue_snapshot();
(StatusCode::OK, "ok").into_response()
}

16
hive-c0re/src/dashboard/meta_inputs.rs
View file

@ -208,18 +208,16 @@ 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.
state
.coord
.job_queue
.insert_job(|b| {
crate::job_queue::templates::meta_update(b, inputs, None);
Vec::new()
})
.expect("template-declared shapes are acyclic");
state.coord.emit_rebuild_queue_snapshot();
crate::job_queue::submit::meta_update(
&state.coord,
inputs,
crate::job_queue::Source::Manual,
format!("meta-update via dashboard ({inputs_label})"),
);
(StatusCode::OK, "ok").into_response()
}

68
hive-c0re/src/dashboard/permissions.rs
View file

@ -155,21 +155,15 @@ 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.
state
.coord
.job_queue
.insert_job(|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();
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(),
},
);
tracing::info!(agent = %logical, groups = ?body.groups, "operator: set tool-groups via dashboard");
Ok((StatusCode::OK, "ok").into_response())
}
@ -270,21 +264,15 @@ 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.
state
.coord
.job_queue
.insert_job(|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();
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(),
},
);
tracing::info!(agent = %logical, caps = ?body.caps, "operator: set capabilities via dashboard");
Ok((StatusCode::OK, "ok").into_response())
}
@ -371,19 +359,13 @@ pub(super) async fn post_permissions(
}
// Phase 2 — submit one combined PermChange DAG per affected agent.
for (logical, groups, caps) in staged {
state
.coord
.job_queue
.insert_job(|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();
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 },
);
tracing::info!(agent = %logical, "operator: batch perm change via dashboard");
}
Ok((StatusCode::OK, "ok").into_response())

31
hive-c0re/src/dashboard/topology.rs
View file

@ -21,6 +21,7 @@ use utoipa::ToSchema;
use problem_details::ProblemDetails;
use super::{AppState, error_problem};
use crate::job_queue::{Source, submit};
/// `POST /api/topology/set-parent` body. `child` is required.
/// `new_parent` may be:
@ -93,15 +94,12 @@ pub(super) async fn post_set_parent(
new_parent = ?new_parent,
"operator: set-parent via dashboard"
);
state
.coord
.job_queue
.insert_job(|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();
submit::reparent(
&state.coord,
vec![(child, new_parent)],
Source::Manual,
"manual set-parent via dashboard".to_owned(),
);
Ok((StatusCode::OK, "ok").into_response())
}
@ -152,14 +150,11 @@ 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");
state
.coord
.job_queue
.insert_job(|b| {
crate::job_queue::templates::reparent(b, moves);
Vec::new()
})
.expect("template-declared shapes are acyclic");
state.coord.emit_rebuild_queue_snapshot();
submit::reparent(
&state.coord,
moves,
Source::Manual,
"manual set-parent-bulk via dashboard".to_owned(),
);
Ok((StatusCode::OK, "ok").into_response())
}

21
hive-c0re/src/job_queue/exec.rs
View file

@ -53,7 +53,7 @@ pub(super) async fn run_node(
kind: &NodeKind,
) -> (super::JobBuilder, Result<()>) {
// The agent this node targets rides the payload — empty for the agentless
// kinds (`MetaLock`, `Reparent`), which never read it.
// container kinds (`MetaLock`, `Dag`), which never read it.
let agent = kind.agent();
// Every arm is `Result<()>`; the three that grow work declare into `builder`
// *synchronously*, after their own awaits have finished. Borrowing `&builder`
@ -115,21 +115,26 @@ pub(super) async fn run_node(
run_finalize_deploy(coord, *approval_id).await
}
NodeKind::DeployTail { approval_id, .. } => {
run_deploy_tail(coord, coord.job_queue.root_of(id), agent, *approval_id).await
run_deploy_tail(coord, coord.job_queue.dag_of(id), agent, *approval_id).await
}
NodeKind::ResolveApproval {
approval_id,
outcome,
} => run_resolve_approval(coord, coord.job_queue.root_of(id), *approval_id, *outcome).await,
} => run_resolve_approval(coord, coord.job_queue.dag_of(id), *approval_id, *outcome).await,
NodeKind::EmitRebuilt { ok, .. } => {
run_emit_rebuilt(coord, agent, coord.job_queue.root_of(id), *ok).await;
run_emit_rebuilt(coord, agent, coord.job_queue.dag_of(id), *ok).await;
Ok(())
}
NodeKind::SetWanted { up, .. } => run_set_wanted(coord, agent, *up),
// Braces carry no work of their own; completing one lets it reach
// `Finishing` so the nodes under it start. What they declare stays held
// until their whole subtree settles.
NodeKind::DeployWindow { .. } | NodeKind::AgentWindow { .. } => Ok(()),
// The nodes that carry no work of their own; completing one lets it
// reach `Finishing` so the nodes under it start.
// - `Dag`: pure grouping container. The DAG's terminal side effect, if
// any, is its own tail node in the graph.
// - `DeployWindow` / `AgentWindow`: pure resource holders (braces) —
// what they declare stays held until their subtree settles.
NodeKind::Dag { .. } | NodeKind::DeployWindow { .. } | NodeKind::AgentWindow { .. } => {
Ok(())
}
};
(builder, result)
}

129
hive-c0re/src/job_queue/mod.rs
View file

@ -10,16 +10,17 @@
//! carries the agent it targets ([`NodeKind::agent`]); the two resource
//! classes are [`resource::Resource`] (`BuildSlot` node-held, `Agent` lease
//! subtree-held), declared per node at its construction site;
//! - **a job has no container node.** A template declares its nodes and names
//! the roots it wants back; `insert_job` returns those ids. Grouping is the
//! parent axis (a root's rolled-up state *is* its subtree's), so membership is
//! a graph walk with no host-side side-tables. The lease is owned by a subtree
//! root and borrowed by its descendants (continuity);
//! - terminal work is an ordinary **tail node**
//! - a **DAG is a single container node** ([`NodeKind::Dag`], `parent = None`)
//! carrying the group's metadata, with the work 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 is an ordinary **tail node**
//! ([`NodeKind::ResolveApproval`] / [`NodeKind::EmitRebuilt`]) that the builder
//! appends in [`templates`], edged onto the job's group roots by the outcome it
//! reports. Templates emit one tail per outcome and the graph runs exactly one,
//! so nothing branches at runtime.
//! appends in [`templates`], edged onto the DAG's other group roots by the
//! outcome it reports. Templates emit one tail per outcome and the graph runs
//! exactly one, so nothing branches at runtime.
//!
//! 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
@ -28,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;
@ -45,7 +46,7 @@ use hive_jobq_wire::{GraphNode, GraphWire};
use tokio::sync::Notify;
pub use hive_jobq::TerminalState;
pub use model::{NodeKind, PermPayload, State};
pub use model::{NodeKind, PermPayload, Source, State};
use resource::Resource;
/// A job under construction: `hive_jobq`'s builder over this queue's payload
@ -89,10 +90,12 @@ pub struct RunningTransient {
/// The crate scheduler, specialised to this host's node + resource types.
///
/// A job is **just its nodes** — no container, no grouping side-tables. A
/// root's rolled-up state is its subtree's, so membership is a graph walk and
/// "which job is this node in" is [`JobQueue::root_of`]. One shared crate
/// [`Graph`] holds every job's nodes.
/// A **DAG is a single container node** ([`NodeKind::Dag`], `parent = None`)
/// whose subtree is the DAG's work — so the container's `NodeId` is the DAG id,
/// its rolled-up state is the DAG state, and there are no grouping side-tables:
/// membership + meta are graph queries ([`container`] + the `hive_jobq::Graph`
/// accessors, with the meta read straight off the container's payload). One
/// shared crate [`Graph`] holds every DAG.
///
/// There is deliberately **no wrapper struct and no per-node side map**. The
/// last map held the `build_logs` row id; that link now lives on the log row
@ -137,12 +140,35 @@ fn outcome_of(result: Result<(), String>) -> Outcome {
}
}
// `insert_group` lived here: a `group_parent`-taking insert whose only
// remaining caller was the DAG container, everything under it. Runtime growth
// never went through it — an executor declares into the builder `hive_jobq`
// hands it, which parents the new work under the emitting node by
// construction. With no container to be the other kind of parent, the
// distinction it existed to express is gone.
/// Insert a declared `job` into the shared graph, returning the inserted ids.
///
/// A node that declared no parent hangs under `group_parent` — the DAG
/// container for a template, the emitting node for a runtime-appended
/// subgraph. Templates declare the parent axis + sibling ordering directly, so
/// there is no dep-on-root to drop and no lease to hoist: each node declares
/// its own resources, and the crate's borrow model keeps a resource continuous
/// across a subtree (a root owns it, descendants borrow it). Independent group
/// roots carry no cross-links, so a multi-agent DAG's per-agent subgraphs run
/// concurrently, each on its own lease.
///
/// # Errors
/// Propagates a crate graph-insert error (malformed dep/parent / dep-scope).
fn insert_group(
inner: &mut Sched,
declare: impl FnOnce(&JobBuilder),
group_parent: Option<NodeId>,
) -> anyhow::Result<()> {
inner
.insert_job(group_parent, |b| {
declare(b);
// c0re names no handles: a DAG is addressed by its container node,
// which `submit` inserts itself, and nothing downstream looks an
// individual step up by id.
Vec::new()
})
.map_err(|e| anyhow::anyhow!("job_queue: graph insert failed: {e}"))?;
Ok(())
}
impl JobQueue {
#[must_use]
@ -162,34 +188,40 @@ impl JobQueue {
self.sched.lock().expect("job_queue mutex poisoned")
}
/// Insert a job's nodes into the shared graph, then wake the run loop.
/// Submit a DAG: insert a [`NodeKind::Dag`] **container node** carrying the
/// group's metadata, then insert the template's nodes as its subtree (their
/// roots re-parented to the container). Returns the container's id as the
/// DAG id — its rolled-up state is the DAG state.
///
/// Deliberately named for the [`hive_jobq`] primitive it wraps, because
/// that is nearly all it is. **The wrapper earns its place on the wake**:
/// the crate is sync and runtime-free — it holds no `Notify` at all — so
/// the channel the run loop parks on belongs to the host, and something has
/// to ping it. Left to call sites, an insert whose ping was forgotten would
/// leave a correct DAG sitting unscheduled until an unrelated event
/// happened along; nothing would fail, and no test in isolation would see
/// it.
/// The container is an ordinary node: it declares no resources, so the
/// scheduler claims it on the next pass, runs its (empty) logic and parks
/// it in `Finishing`, at which point its children become runnable. Nothing
/// here completes it by hand — a node with no work of its own still goes
/// the way every other node goes.
///
/// Returns exactly what the primitive returns: the ids of the nodes the
/// template named, in the order it named them.
/// `source` and `reason` are the container node's own payload — they are
/// arguments here rather than fields of a spec struct because that is all
/// they ever were. `declare` is the recipe, taken by generic and run
/// against a builder `hive_jobq` owns: it goes from the template straight
/// into this call, so there is nothing to allocate for.
///
/// # Errors
/// Propagates a graph-insert error (dependencies that aren't
/// dependency-topological).
pub fn insert_job(
pub fn submit(
&self,
declare: impl FnOnce(&JobBuilder) -> Vec<hive_jobq::NodeGuid>,
) -> anyhow::Result<Vec<NodeId>> {
source: Source,
reason: String,
declare: impl FnOnce(&JobBuilder),
) -> anyhow::Result<u64> {
let mut inner = self.lock();
let named = inner
.insert_job(None, declare)
.map_err(|e| anyhow::anyhow!("job_queue: graph insert failed: {e}"))?;
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))?;
drop(inner);
self.notify.notify_one();
Ok(named)
Ok(container.get())
}
/// The scheduler itself, for `hive_jobq`'s run-loop seam
@ -203,16 +235,11 @@ impl JobQueue {
&self.sched
}
/// The id of the **group root** `node` belongs to, for log lines and the
/// dashboard. Derived from the graph rather than carried alongside the node
/// — the parent axis already knows it.
///
/// Was `dag_of`, when a job's nodes hung under a container node that *was*
/// the group. Without it the parent chain ends at whichever root the
/// template declared, so this answers "which root owns this node", not
/// "which DAG is this in" — there is no longer such a thing.
/// The DAG container id owning `node`, for log lines and the dashboard.
/// Derived from the graph rather than carried alongside the node — the
/// parent axis already knows it.
#[must_use]
pub fn root_of(&self, node: NodeId) -> Option<u64> {
pub fn dag_of(&self, node: NodeId) -> Option<u64> {
self.lock().graph().root_of(node).map(NodeId::get)
}
@ -397,9 +424,9 @@ fn find_node(sched: &Sched, id: u64) -> Option<NodeId> {
/// root, plus the newest [`MAX_HISTORY_DAGS`] settled ones.
///
/// Selected *structurally* — a root is a node with no parent. The typed
/// projection this replaced keyed on the since-removed container kind instead,
/// which made the visible set depend on one host node kind; nothing here knows
/// what a node means.
/// projection this replaced keyed on `NodeKind::Dag` instead, which made the
/// visible set depend on one host node kind; nothing here knows what a node
/// means.
///
/// **This bound is load-bearing, not tidiness.** Nothing ever removes a node
/// from the graph (bounded pruning is a Stage-C follow-up), so serving

46
hive-c0re/src/job_queue/model.rs
View file

@ -1,18 +1,18 @@
//! Data model for the generic job-DAG queue: the node kinds — the primitive
//! operations — and what each one carries. The `State` / `PermPayload` wire
//! enums live in `hive_host_sock::jobs` (they travel on the host admin
//! socket) and are re-exported here for the queue's internal use. The graph
//! itself is served through `hive_jobq_wire`'s generic projection — there is
//! no second, typed view of it any more.
//! Data model for the generic job-DAG queue: node kinds (the primitive
//! operations), dependency edges, and the runtime `Dag` / `Node` store.
//! The `Source` / `State` / `PermPayload` wire enums live in
//! `hive_host_sock::jobs` (they travel on the host admin socket) and are
//! re-exported here for the queue's internal use. The graph itself is
//! served through `hive_jobq_wire`'s generic projection — there is no
//! second, typed view of it any more.
//!
//! **One level, not two.** The node is the unit of everything: scheduling,
//! execution, build-log, cancel, and the dashboard group (a group root's
//! subtree *is* the group). A DAG used to be a second level above it, with
//! its own store and its own id; there is no container node any more, so a
//! job is exactly the nodes it declared. 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, and carries its own `agent` (a
//! DAG can span agents). See `docs/coordinator.md::Job queue` for the
//! full design.
pub use hive_host_sock::jobs::{PermPayload, State};
pub use hive_host_sock::jobs::{PermPayload, Source, State};
use serde::Serialize;
use hive_jobq::TerminalState;
@ -280,6 +280,18 @@ pub enum NodeKind {
/// `Prebuild`, but that's a no-op there — the agent is down, so prebuild
/// is skipped.)
SetWanted { agent: String, up: bool },
/// The **DAG container** node: one per submitted DAG, carrying the group's
/// domain metadata. Every node hangs *under* it (its subtree), so
/// the container's `NodeId` **is** the DAG id and its rolled-up state **is**
/// the DAG state. Pure grouping — lease- and
/// build-slot-exempt; the executor instant-completes it (`Done`) so it
/// reaches `Finishing` and its children start.
///
/// No `created_at` here: the graph stamps [`hive_jobq::Node::created_at`] on
/// every node at insert, so the container already has one. A second copy in
/// the payload would be the same instant recorded twice, with only this
/// variant's version reachable to a generic viewer.
Dag { source: Source, reason: String },
}
/// How a hive-c0re node describes itself to a generic graph viewer.
@ -350,13 +362,14 @@ impl NodeKind {
NodeKind::ResolveApproval { .. } => "resolve_approval",
NodeKind::EmitRebuilt { .. } => "emit_rebuilt",
NodeKind::SetWanted { .. } => "set_wanted",
NodeKind::Dag { .. } => "dag",
}
}
/// The agent this node targets, or `""` for agentless kinds
/// ([`NodeKind::MetaLock`] on the `hyperhive` pseudo-agent,
/// [`NodeKind::Reparent`] which can span multiple agents, and
/// [`NodeKind::ResolveApproval`] which acts on an approval row).
/// [`NodeKind::Reparent`] which can span multiple agents, and the
/// [`NodeKind::Dag`] container).
#[must_use]
pub fn agent(&self) -> &str {
match self {
@ -384,7 +397,8 @@ impl NodeKind {
| NodeKind::SetWanted { agent, .. } => agent,
NodeKind::MetaLock { .. }
| NodeKind::Reparent { .. }
| NodeKind::ResolveApproval { .. } => "",
| NodeKind::ResolveApproval { .. }
| NodeKind::Dag { .. } => "",
}
}

2
hive-c0re/src/job_queue/scheduler.rs
View file

@ -93,7 +93,7 @@ pub async fn run_worker(coord: Arc<Coordinator>) {
let coord = node_coord;
async move {
tracing::info!(
dag = coord.job_queue.root_of(id).unwrap_or_default(),
dag = coord.job_queue.dag_of(id).unwrap_or_default(),
node = id.get(),
kind = kind.as_str(),
agent = %kind.agent(),

316
hive-c0re/src/job_queue/power.rs → hive-c0re/src/job_queue/submit.rs
View file

@ -1,27 +1,59 @@
//! 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.
//! Request-level submit API — the surface the dashboard POST handlers,
//! the MCP socket handlers, and `hivectl` paths call.
//!
//! 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.
//! 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).
//!
//! 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.
//! 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.
use std::sync::Arc;
use hive_jobq::NodeId;
use super::model::NodeKind;
use super::resource::Resource;
use super::templates::rebuild_nodes;
use super::{JobBuilder, NodeKind};
use super::{JobBuilder, Source, templates};
use crate::coordinator::Coordinator;
use crate::lifecycle;
fn submit_and_emit(
coord: &Arc<Coordinator>,
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<Coordinator>, 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`)
@ -35,16 +67,7 @@ use crate::lifecycle;
/// actually running (nothing to drain on a down container). The `Reconcile`
/// stays even for a down agent so a race-up between the state read and exec
/// is still stopped in-DAG.
/// Returns the group root's guid, which is what the caller names so
/// `insert_job` hands its id back — that id is how `hivectl` polls this agent's
/// progress. A chain that returned nothing would insert correctly and leave the
/// caller with nothing to wait on.
fn stop_chain(
builder: &JobBuilder,
agent: &str,
graceful: bool,
running: bool,
) -> hive_jobq::NodeGuid {
fn stop_chain(builder: &JobBuilder, agent: &str, graceful: bool, running: bool) {
// `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).
@ -73,26 +96,13 @@ fn stop_chain(
.needs(Resource::Agent(a()))
.part_of(wanted);
}
wanted.guid()
}
/// One agent's **start** subgraph. `SetWanted(Up)` head; a down + stale-rev
/// agent gets the rebuild subgraph (its tail `Reconcile` starts it on
/// current derivations), otherwise a plain `Reconcile` (which starts a down
/// agent and noops an already-running one).
///
/// Returns **every** group root — see [`stop_chain`] for why they are named.
///
/// ⚠️ More than one in the stale branch: `rebuild_nodes` chains its roots
/// *behind* `SetWanted` with `after_ok`, it does **not** nest them under it. So
/// `SetWanted` rolls up only itself, and naming it alone would report the whole
/// start finished while the rebuild was still running.
fn start_chain(
builder: &JobBuilder,
agent: &str,
running: bool,
stale: bool,
) -> Vec<hive_jobq::NodeGuid> {
fn start_chain(builder: &JobBuilder, agent: &str, running: bool, stale: bool) {
let wanted = builder
.node(NodeKind::SetWanted {
agent: agent.to_owned(),
@ -100,16 +110,10 @@ fn start_chain(
})
.needs(Resource::Agent(agent.to_owned()));
if !running && stale {
// Rebuild subtree chained behind the `SetWanted` head. `MetaSync`, the
// `AgentWindow` brace and `Reconcile` are their own group roots
// (top-level, per `rebuild_nodes`) — hence all four names.
let roots = rebuild_nodes(builder, agent, true, Some(wanted));
vec![
wanted.guid(),
roots.meta_sync.guid(),
roots.agent_window.guid(),
roots.reconcile.guid(),
]
// Rebuild subtree chained behind the `SetWanted` head. `MetaSync`,
// `Prebuild` + `Reconcile` are their own group roots (top-level, per
// `rebuild_nodes`).
rebuild_nodes(builder, agent, true, Some(wanted));
} else {
let _ = builder
.node(NodeKind::Reconcile {
@ -117,7 +121,6 @@ fn start_chain(
})
.needs(Resource::Agent(agent.to_owned()))
.part_of(wanted);
vec![wanted.guid()]
}
}
@ -130,22 +133,14 @@ fn start_chain(
/// before `Reconcile`; a down agent gets just `Reconcile`, which
/// converges to intent — a stopped (`wanted = Off`) agent stays stopped,
/// a crashed (`wanted = Up`) agent comes back up.
///
/// Returns the group root's guid — see [`stop_chain`].
fn restart_chain(
builder: &JobBuilder,
agent: &str,
graceful: bool,
running: bool,
) -> hive_jobq::NodeGuid {
fn restart_chain(builder: &JobBuilder, agent: &str, graceful: bool, running: bool) {
let a = || agent.to_owned();
if !running {
// Nothing to bounce — a lone Reconcile converges to intent, and is
// itself the root.
return builder
// Nothing to bounce — a lone Reconcile converges to intent.
let _ = builder
.node(NodeKind::Reconcile { agent: a() })
.needs(Resource::Agent(a()))
.guid();
.needs(Resource::Agent(a()));
return;
}
// 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
@ -181,7 +176,6 @@ fn restart_chain(
.needs(Resource::Agent(a()))
.part_of(signal)
.after_ok(stop);
signal.guid()
} else {
let stop = builder
.node(NodeKind::StopForUpdate { agent: a() })
@ -190,7 +184,6 @@ fn restart_chain(
.node(NodeKind::Reconcile { agent: a() })
.needs(Resource::Agent(a()))
.part_of(stop);
stop.guid()
}
}
@ -208,15 +201,10 @@ fn restart_chain(
// subgraph's indices onto another's used to be a function.
/// Declare the stop DAG from explicit `(agent, running)` targets.
pub(crate) fn stop_nodes(
builder: &JobBuilder,
targets: &[(String, bool)],
graceful: bool,
) -> Vec<hive_jobq::NodeGuid> {
targets
.iter()
.map(|(agent, running)| stop_chain(builder, agent, graceful, *running))
.collect()
pub(crate) fn stop_nodes(builder: &JobBuilder, targets: &[(String, bool)], graceful: bool) {
for (agent, running) in targets {
stop_chain(builder, agent, graceful, *running);
}
}
/// Assemble the start DAG from explicit `(agent, running, stale)` targets.
@ -225,68 +213,76 @@ pub(crate) fn stop_nodes(
/// running under its lease, so a down+stale agent that grew a rebuild subgraph
/// reports `rebuilding` during its swap and `starting` at its reconcile,
/// without the DAG having to guess one label covering every target.
pub(crate) fn start_nodes(
builder: &JobBuilder,
targets: &[(String, bool, bool)],
) -> Vec<hive_jobq::NodeGuid> {
targets
.iter()
.flat_map(|(agent, running, stale)| start_chain(builder, agent, *running, *stale))
.collect()
pub(crate) fn start_nodes(builder: &JobBuilder, targets: &[(String, bool, bool)]) {
for (agent, running, stale) in targets {
start_chain(builder, agent, *running, *stale);
}
}
/// Declare the restart DAG from explicit `(agent, running)` targets.
pub(crate) fn restart_nodes(
builder: &JobBuilder,
targets: &[(String, bool)],
graceful: bool,
) -> Vec<hive_jobq::NodeGuid> {
targets
.iter()
.map(|(agent, running)| restart_chain(builder, agent, graceful, *running))
.collect()
pub(crate) fn restart_nodes(builder: &JobBuilder, targets: &[(String, bool)], graceful: bool) {
for (agent, running) in targets {
restart_chain(builder, agent, graceful, *running);
}
}
// ---- entry points ---------------------------------------------------------
//
// Each reads the live state its shape depends on, then declares + inserts.
/// Restart a single agent. Thin wrapper over [`restart_many`].
pub async fn restart(coord: &Arc<Coordinator>, agent: &str, source: Source, reason: String) -> u64 {
restart_many(coord, &[agent.to_owned()], false, source, reason).await
}
/// 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
/// 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<Coordinator>,
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
/// tail `Reconcile` converges each agent to its EXISTING intent — a
/// deliberately-stopped agent stays down.
///
/// # Errors
/// Propagates a graph-insert error.
/// deliberately-stopped agent stays down. The whole hive-wide
/// `hivectl restart` is one DAG.
pub async fn restart_many(
coord: &Arc<Coordinator>,
agents: &[String],
graceful: bool,
) -> anyhow::Result<Vec<NodeId>> {
source: Source,
reason: String,
) -> u64 {
let mut targets = Vec::with_capacity(agents.len());
for agent in agents {
targets.push((agent.clone(), lifecycle::is_running(agent).await));
}
let ids = coord
.job_queue
.insert_job(|b| restart_nodes(b, &targets, graceful))?;
coord.emit_rebuild_queue_snapshot();
Ok(ids)
submit_and_emit(coord, source, reason, |builder| {
restart_nodes(builder, &targets, graceful);
})
}
/// 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.
/// Start a single agent. Thin wrapper over [`start_many`].
pub async fn start(coord: &Arc<Coordinator>, 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<Coordinator>,
agents: &[String],
) -> anyhow::Result<Vec<NodeId>> {
source: Source,
reason: String,
) -> u64 {
let current = crate::auto_update::current_flake_rev(&coord.hyperhive_flake);
let mut targets = Vec::with_capacity(agents.len());
for agent in agents {
@ -300,30 +296,88 @@ pub async fn start_many(
}
targets.push((agent.clone(), running, stale));
}
let ids = coord.job_queue.insert_job(|b| start_nodes(b, &targets))?;
coord.emit_rebuild_queue_snapshot();
Ok(ids)
submit_and_emit(coord, source, reason, |builder| {
start_nodes(builder, &targets);
})
}
/// 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.
/// Hard stop a single agent. Thin wrapper over [`stop_many`].
pub async fn stop(coord: &Arc<Coordinator>, 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<Coordinator>,
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.
pub async fn stop_many(
coord: &Arc<Coordinator>,
agents: &[String],
graceful: bool,
) -> anyhow::Result<Vec<NodeId>> {
source: Source,
reason: String,
) -> u64 {
let mut targets = Vec::with_capacity(agents.len());
for agent in agents {
targets.push((agent.clone(), lifecycle::is_running(agent).await));
}
let ids = coord
.job_queue
.insert_job(|b| stop_nodes(b, &targets, graceful))?;
coord.emit_rebuild_queue_snapshot();
Ok(ids)
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<Coordinator>,
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<Coordinator>,
inputs: Vec<String>,
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<Coordinator>,
moves: Vec<(hive_types::Ident, Option<hive_types::Ident>)>,
source: Source,
reason: String,
) -> u64 {
submit_and_emit(coord, source, reason, |builder| {
templates::reparent(builder, moves);
})
}

35
hive-c0re/src/job_queue/templates.rs
View file

@ -17,8 +17,8 @@
//!
//! The hive-wide **power ops** (`stop` / `start` / `restart`) are NOT here:
//! their per-agent shape depends on live running state (an async
//! `lifecycle::is_running` read), so [`super::power`] assembles them out of
//! the primitives this module exports ([`rebuild_nodes`]).
//! `lifecycle::is_running` read), so `submit.rs` assembles them out of the
//! primitives this module exports ([`rebuild_nodes`]).
use hive_jobq::TerminalState;
@ -317,22 +317,13 @@ pub(crate) fn deploy_rebuild_nodes(builder: &JobBuilder, agent: &str, approval_i
/// children.
///
/// Closed by an [`NodeKind::EmitRebuilt`] tail edged onto all three group-roots
/// (`MetaSync`, the `AgentWindow` brace, `Reconcile`) — the brace's roll-up
/// carries the whole `StopForUpdate`→`Swap`→`RebuildBookkeeping` subtree, so
/// those three cover every node. Edging `Reconcile` alone would not do: it is
/// `AfterAny` the brace, so it reaches `Done` even after a failed swap and the
/// tail would report success.
///
/// Returns those three roots, so a caller that needs to wait on the rebuild can
/// name them.
pub fn rebuild(builder: &JobBuilder, agent: &str, relock: bool) -> Vec<hive_jobq::NodeGuid> {
/// (`MetaSync`, `Prebuild`, `Reconcile`) — `Prebuild`'s roll-up carries the
/// whole `StopForUpdate`→`Swap`→`RebuildBookkeeping` subtree, so those three cover every
/// node. Edging `Reconcile` alone would not do: it is `AfterAny` `Prebuild`, so
/// it reaches `Done` even after a failed swap and the tail would report success.
pub fn rebuild(builder: &JobBuilder, agent: &str, relock: bool) {
let roots = rebuild_nodes(builder, agent, relock, None);
emit_rebuilt_tails(builder, agent, &roots.all());
vec![
roots.meta_sync.guid(),
roots.agent_window.guid(),
roots.reconcile.guid(),
]
}
/// Approval-driven deploy (`MergeConfigPr`) as a phase subtree rather than the
@ -492,16 +483,10 @@ pub fn meta_update(builder: &JobBuilder, inputs: Vec<String>, approval_id: Optio
/// checks), so a parent move needs no container rebuild to take effect.
/// No transient pill either — the node is agentless (no lease to hang one
/// off of) and near-instant. No tail node: the write is the whole effect.
///
/// Returns the single node's guid so a caller can wait on it.
pub fn reparent(
builder: &JobBuilder,
moves: Vec<(hive_types::Ident, Option<hive_types::Ident>)>,
) -> hive_jobq::NodeGuid {
builder
pub fn reparent(builder: &JobBuilder, moves: Vec<(hive_types::Ident, Option<hive_types::Ident>)>) {
let _reparent = builder
.node(NodeKind::Reparent { moves })
.needs(Resource::MetaWindow)
.guid()
.needs(Resource::MetaWindow);
}
// The boot is assembled inline in `workers/auto_update.rs::submit_boot_tree`

592
hive-c0re/src/job_queue/tests.rs
View file

File diff suppressed because it is too large Load diff

142
hive-c0re/src/server.rs
View file

@ -180,19 +180,13 @@ async fn dispatch(req: &HostRequest, coord: Arc<Coordinator>) -> HostResponse {
// submit returns a DAG id immediately, the caller polls
// `QueueDag` (`hivectl`'s wait/progress loop) for the
// outcome instead of blocking here on the commit.
let inserted = coord.job_queue.insert_job(|b| {
vec![crate::job_queue::templates::reparent(
b,
vec![(child.clone(), new_parent.clone())],
)]
});
match inserted {
Ok(ids) => {
coord.emit_rebuild_queue_snapshot();
HostResponse::queued(ids.into_iter().map(hive_jobq::NodeId::get).collect())
}
Err(e) => HostResponse::error(format!("queue reparent: {e}")),
}
let id = crate::job_queue::submit::reparent(
&coord,
vec![(child.clone(), new_parent.clone())],
crate::job_queue::Source::Manual,
"manual set-parent via hivectl".to_owned(),
);
HostResponse::queued(vec![id])
}
HostRequest::SetResourceLimits {
name,
@ -783,36 +777,39 @@ enum Verb {
}
async fn submit_single(coord: &Arc<Coordinator>, name: &str, verb: Verb) -> HostResponse {
use crate::job_queue::power;
// A single-target op is the N-target one with N = 1 — the shapes are
// identical, so there is no separate builder to keep in sync.
let targets = [name.to_owned()];
let ids = match verb {
use crate::job_queue::{Source, submit};
let id = match verb {
Verb::Kill => {
tracing::info!(%name, "kill");
power::stop_many(coord, &targets, false).await
submit::stop(
coord,
name,
Source::Manual,
"manual kill via hivectl".to_owned(),
)
.await
}
Verb::Restart => {
tracing::info!(%name, "restart");
power::restart_many(coord, &targets, false).await
submit::restart(
coord,
name,
Source::Manual,
"manual restart via hivectl".to_owned(),
)
.await
}
Verb::Rebuild => {
tracing::info!(%name, "rebuild");
// Not a power op: a rebuild's shape doesn't depend on live state,
// so it is a plain template insert rather than a `*_many` gather.
let inserted = coord
.job_queue
.insert_job(|b| crate::job_queue::templates::rebuild(b, name, true));
if inserted.is_ok() {
coord.emit_rebuild_queue_snapshot();
}
inserted
submit::rebuild(
coord,
name,
Source::Manual,
"manual rebuild via hivectl".to_owned(),
)
}
};
match ids {
Ok(ids) => HostResponse::queued(ids.into_iter().map(hive_jobq::NodeId::get).collect()),
Err(e) => HostResponse::error(format!("queue insert failed: {e}")),
}
HostResponse::queued(vec![id])
}
/// Stop the given `agents` (resolved logical names) then `infra` containers
@ -840,17 +837,27 @@ async fn handle_stop(
let mut errors: Vec<String> = Vec::new();
let mut queued: Vec<u64> = Vec::new();
// One insert for all targeted agents — a per-agent stop subgraph each
// One DAG for all targeted agents — a per-agent stop subgraph each
// (`SetWanted(Offline) → [Signal → Drain →] Reconcile`), independent
// roots that run concurrently on their own leases.
// roots that run concurrently on their own leases. A hive-wide
// `hivectl stop` is now a single DAG, not N.
if !agents.is_empty() {
match crate::job_queue::power::stop_many(coord, agents, graceful).await {
Ok(ids) => {
queued.extend(ids.into_iter().map(hive_jobq::NodeId::get));
ok_items.extend(agents.iter().cloned());
}
Err(e) => errors.push(format!("queue stop: {e}")),
}
let reason = if graceful {
"manual via hivectl graceful stop"
} else {
"manual via hivectl stop"
};
queued.push(
crate::job_queue::submit::stop_many(
coord,
agents,
graceful,
crate::job_queue::Source::Manual,
reason.to_owned(),
)
.await,
);
ok_items.extend(agents.iter().cloned());
}
// Agents go down before infra so they're not mid-request against a
@ -937,13 +944,16 @@ async fn handle_start(
// hivectl's wait loop.
let mut queued: Vec<u64> = Vec::new();
if !agents.is_empty() {
match crate::job_queue::power::start_many(coord, agents).await {
Ok(ids) => {
queued.extend(ids.into_iter().map(hive_jobq::NodeId::get));
ok_items.extend(agents.iter().cloned());
}
Err(e) => errors.push(format!("queue start: {e}")),
}
queued.push(
crate::job_queue::submit::start_many(
coord,
agents,
crate::job_queue::Source::Manual,
"manual via hivectl start".to_owned(),
)
.await,
);
ok_items.extend(agents.iter().cloned());
}
let mut resp = finish_lifecycle(ok_items, &errors);
@ -972,20 +982,28 @@ async fn handle_restart_scoped(
let mut errors: Vec<String> = Vec::new();
let mut queued: Vec<u64> = Vec::new();
// One insert for all targeted agents — a per-agent restart subgraph each
// (`[Signal → Drain →] StopForUpdate → Reconcile`; no `SetWanted`, since a
// restart converges to the agent's *existing* intent rather than rewriting
// it), independent roots running concurrently on their own leases. No
// client-side stop-then-start composition — the whole restart survives a
// dropped connection because the graph owns it.
// One DAG for all targeted agents — a per-agent restart subgraph each
// (`SetWanted → [Signal → Drain →] StopForUpdate → Reconcile`),
// independent roots that run concurrently on their own leases. A
// hive-wide `hivectl restart` is now a single DAG, not N. No
// client-side stop-then-start composition — the whole restart survives
// a dropped connection because the DAG owns it.
if !agents.is_empty() {
match crate::job_queue::power::restart_many(coord, &agents, graceful).await {
Ok(ids) => {
queued.extend(ids.into_iter().map(hive_jobq::NodeId::get));
ok_items.extend(agents.iter().cloned());
}
Err(e) => errors.push(format!("queue restart: {e}")),
}
queued.push(
crate::job_queue::submit::restart_many(
coord,
&agents,
graceful,
crate::job_queue::Source::Manual,
if graceful {
"manual via hivectl restart --graceful".to_owned()
} else {
"manual restart via hivectl restart".to_owned()
},
)
.await,
);
ok_items.extend(agents.iter().cloned());
}
for &container in &infra {

33
hive-c0re/src/socket_server/lifecycle_handlers.rs
View file

@ -20,9 +20,13 @@ pub(super) async fn handle_start(coord: &Arc<Coordinator>, 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.
if let Err(e) = crate::job_queue::power::start_many(coord, &[name.to_owned()]).await {
tracing::error!(%agent, %name, error = ?e, "start: insert failed");
}
crate::job_queue::submit::start(
coord,
name,
crate::job_queue::Source::Manual,
format!("agent `{agent}` start tool"),
)
.await;
Response::Ok
}
@ -44,9 +48,13 @@ pub(super) async fn handle_restart(coord: &Arc<Coordinator>, agent: &str, name:
return err;
}
tracing::info!(%agent, %name, "submit restart");
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");
}
crate::job_queue::submit::restart(
coord,
name,
crate::job_queue::Source::Manual,
format!("agent `{agent}` restart tool"),
)
.await;
Response::Ok
}
@ -145,13 +153,12 @@ pub(super) fn handle_update(coord: &Arc<Coordinator>, agent: &str, name: &str) -
return err;
}
tracing::info!(%agent, %name, "submit rebuild");
if let Err(e) = coord.job_queue.insert_job(|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();
crate::job_queue::submit::rebuild(
coord,
name,
crate::job_queue::Source::Manual,
format!("agent `{agent}` update tool"),
);
Response::Ok
}

35
hive-c0re/src/workers/auto_update.rs
View file

@ -109,11 +109,12 @@ pub async fn ensure_root_agent(coord: &Arc<Coordinator>) -> Result<()> {
tracing::warn!(
"manager container exists but no applied flake — forcing rebuild to migrate"
);
if let Err(e) = coord.job_queue.insert_job(|b| {
crate::job_queue::templates::rebuild(b, MANAGER_NAME, true);
Vec::new()
}) {
tracing::warn!(error = ?e, "manager migration rebuild insert failed");
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");
}
} else {
tracing::debug!("manager container already present");
@ -377,19 +378,18 @@ fn submit_boot_tree(
n_deferred: usize,
n_skipped: usize,
) {
// Fully-quiet boot (nothing stale, nothing drifted) inserts nothing.
use crate::job_queue::Source;
// Fully-quiet boot (nothing stale, nothing drifted) submits nothing.
if !any_stale && drifted.is_empty() {
return;
}
// 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"
let reason = format!(
"boot: {} rebuild(s), {} reconcile(s), {} deferred (offline), {} up-to-date",
fanout.len(),
drifted.len(),
n_deferred,
n_skipped,
);
// The sweep's own rebuild subgraphs emit their `Rebuilt` events as they
@ -397,11 +397,10 @@ 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.insert_job(|b| {
if let Err(e) = coord.job_queue.submit(Source::AutoUpdate, reason, |b| {
boot_nodes(b, any_stale, fanout, drifted);
Vec::new()
}) {
tracing::warn!(error = ?e, "boot: sweep DAG insert failed");
tracing::warn!(error = ?e, "boot: sweep DAG submit failed");
}
coord.emit_rebuild_queue_snapshot();
}

42
hive-host-sock/src/jobs.rs
View file

@ -1,11 +1,6 @@
//! Vocabulary hive-c0re's job queue shares with its clients: what a
//! permission change carries ([`PermPayload`]) and the scheduler's
//! lifecycle [`State`].
//!
//! A `Source` enum lived here too — where a job came from, rendered as the
//! "why" chip. It was a field on the DAG container, and it went with it: a
//! job is its nodes now, and a node says what it does rather than who asked
//! for it.
//! Vocabulary hive-c0re's job queue shares with its clients: where a job
//! came from ([`Source`]), what a permission change carries
//! ([`PermPayload`]), and the scheduler's lifecycle [`State`].
//!
//! **The typed `DagView`/`NodeView` projection that used to live here is
//! gone.** One graph is served one way now — `hive_jobq_wire`'s generic
@ -17,6 +12,37 @@
use serde::{Deserialize, Serialize};
/// Where the submit request originated — drives the "why" chip on the
/// dashboard.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum Source {
/// Operator action (dashboard button, CLI, manager tool).
Manual,
/// Meta-update cascade rebuild (grown into the meta-update DAG).
MetaUpdate,
/// Boot-time submission (the boot sweep DAG + boot reconciles).
AutoUpdate,
/// Crash recovery path (future use).
CrashRecover,
/// Operator approved a pending `Approval` row; `approval_id` on
/// the DAG points back at the source row.
Approval,
}
impl Source {
#[must_use]
pub fn as_str(self) -> &'static str {
match self {
Source::Manual => "manual",
Source::MetaUpdate => "meta_update",
Source::AutoUpdate => "auto_update",
Source::CrashRecover => "crash_recover",
Source::Approval => "approval",
}
}
}
pub use hive_jobq::State;
/// Kind-specific payload for `Template::PermChange` DAGs.