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

refactor(#2439): build hive-wide stop/start/restart DAGs dynamically

Browse source 
Hive-wide `stop` / `start` / `restart` emit ONE DAG with a per-agent
subgraph each (concurrent on their own leases) instead of N DAGs — and each
subgraph is now built dynamically from the agent's live running state rather
than a fixed template shape:

- online agent: the full stop→reconcile (restart: stop-for-update→reconcile)
  chain; `graceful` prepends signal→drain.
- offline agent: just `SetWanted → Reconcile` (nothing to quiesce/stop; a
  restart of a down agent is really a start).

The head `SetWanted` (intent) and tail `Reconcile` (convergence guarantee)
are always present; only the mechanical `Signal`/`Drain`/`StopForUpdate`
nodes are state-conditional. Keeping `Reconcile` in every shape closes the
TOCTOU window — a race-up between the `is_running` read and node exec is
still converged in-DAG (with `StopForUpdate`-noop as the backstop) — with no
reliance on an external reconcile sweep.

The state-aware assembly needs an async `is_running` read, so it moves out
of the pure/sync `templates.rs` into `submit.rs`, layered as pure
`*_chain(running)` → pure `*_spec(targets)` (the unit-test seam) → async
`*_many` (reads live state + submits). `templates.rs` keeps only the shared
pure primitives (`node`/`after_ok`/`rebuild_nodes`).

Callers await the now-async submit fns (server, dashboard, socket_server).
Tests exercise both the online and offline shapes via the pure `*_spec`
seam. docs/coordinator.md shapes updated.
This commit is contained in:
atlas 2026-07-14 23:34:56 +02:00
commit 860484a193
8 changed files with 574 additions and 335 deletions
Download patch file Download diff file Expand all files Collapse all files

33
docs/coordinator.md
View file

@ -76,19 +76,32 @@ container build:
The power ops write the durable `wanted` intent via a head `SetWanted`
node (not a pre-submit side effect) — it holds the agent lease, so
intent-write + reconcile is atomic per-agent. `restart` takes an agent
*list*: a hive-wide `hivectl restart` / `restart-all` is ONE DAG with a
per-agent restart subgraph each (independent roots, run concurrently on
their own leases), not N separate DAGs.
intent-write + reconcile is atomic per-agent. The hive-wide power ops —
`restart`, `stop`, and `start` — take an agent *list*: a hive-wide `hivectl
restart` / `stop` / `start` is ONE DAG with a per-agent 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/submit.rs`,
not the pure/sync `templates.rs`. Per-agent shape rule: the head `SetWanted`
(intent) and the tail `Reconcile` (convergence guarantee — cheap, noops when
already converged) are 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 exec, the tail `Reconcile` still converges it in-DAG (with
`StopForUpdate`-noop as the backstop) — no reliance on an external reconcile
sweep. `start` folds the per-agent stale-rev upgrade in (a *down + stale*
agent's subgraph is a rebuild-then-start).
```text
rebuild(a): Prebuild(a) → StopForUpdate(a) → Swap(a) →(after-any) Reconcile(a)
graceful-stop(a): SetWanted(a,Off) → Signal(a) → Drain(a) → Reconcile(a)
restart(a..): per agent: SetWanted(a,Up) → StopForUpdate(a) → Reconcile(a) (N subgraphs, 1 DAG)
graceful-restart(a): SetWanted(a,Up) → Signal(a) → Drain(a) → StopForUpdate(a) → Reconcile(a)
start(a): SetWanted(a,Up) → Reconcile(a) (stale rev ⇒ stale-start below)
stale-start(a): SetWanted(a,Up) → «rebuild subgraph» (prebuild noops — agent is down)
stop(a): SetWanted(a,Off) → Reconcile(a)
stop(a..): online a: SetWanted(a,Off) → [Signal→Drain→ if graceful] Reconcile(a)
offline a: SetWanted(a,Off) → Reconcile(a) (N subgraphs, 1 DAG)
restart(a..): online a: SetWanted(a,Up) → [Signal→Drain→ if graceful] StopForUpdate(a) → Reconcile(a)
offline a: SetWanted(a,Up) → Reconcile(a) (nothing to stop — it's a start)
start(a..): a: SetWanted(a,Up) → Reconcile(a) (down+stale ⇒ SetWanted(a,Up) → «rebuild subgraph»)
spawn(a): [wanted=Up at approve] Create(a) → WriteDropin(a) → Reconcile(a)
perm-change(a): WritePermFile(a) → «rebuild subgraph»
meta-update(inp): MetaLock(inp) → «fan-out rebuild(a) per affected agent»

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

@ -65,7 +65,8 @@ pub(super) async fn post_kill(
&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
@ -83,7 +84,8 @@ pub(super) async fn post_kill(
&logical,
Source::Manual,
"manual via dashboard stop".to_owned(),
);
)
.await;
(StatusCode::OK, "ok").into_response()
}
@ -100,7 +102,8 @@ pub(super) async fn post_restart(
&logical,
Source::Manual,
"manual via dashboard ↺ R3START button".to_owned(),
);
)
.await;
(StatusCode::OK, "ok").into_response()
}
@ -117,7 +120,8 @@ pub(super) async fn post_start(
&logical,
Source::Manual,
"manual via dashboard start".to_owned(),
);
)
.await;
(StatusCode::OK, "ok").into_response()
}

355
hive-c0re/src/job_queue/submit.rs
View file

@ -1,15 +1,32 @@
//! Request-level submit API — the surface the dashboard POST handlers,
//! the MCP socket handlers, and `hivectl` paths call. The durable
//! `wanted` power intent is now written by a `SetWanted` DAG node at the
//! head of each power-op template (not a pre-submit side effect); the
//! only submit-time logic left is the stale-start *shape* decision
//! (`start` vs `stale_start`). Every helper emits a fresh queue snapshot
//! so the dashboard shows the new DAG immediately.
//! the MCP socket handlers, and `hivectl` paths call.
//!
//! 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
//! (`templates::{node, after_ok, rebuild_nodes}`), and concatenate them into
//! ONE DAG (independent per-agent roots, concurrent on their own leases).
//!
//! Dynamic shape rule: the head `SetWanted(w)` (durable intent) and the tail
//! `Reconcile` (the convergence guarantee — cheap, noops when already
//! converged) are 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 super::model::{DagSpec, Dep, NodeKind, NodeSpec, Template};
use super::templates::{after_ok, node, rebuild_nodes};
use super::{Source, templates};
use crate::coordinator::Coordinator;
use crate::coordinator::{Coordinator, TransientKind};
use crate::lifecycle;
fn submit_and_emit(coord: &Arc<Coordinator>, spec: super::DagSpec) -> u64 {
let id = coord
@ -27,73 +44,305 @@ pub fn rebuild(coord: &Arc<Coordinator>, agent: &str, source: Source, reason: St
submit_and_emit(coord, templates::rebuild(agent, source, reason, None, true))
}
/// Restart a single agent: mechanical stop + converge to `wanted = Up`.
/// The intent write is the template's head `SetWanted(Up)` node — it
/// matters when `wanted` drifted `Offline` under a running agent (an
/// operator asking for a restart plainly wants it running, not a stop).
/// Thin wrapper over [`restart_many`] with a one-agent slice.
pub fn restart(coord: &Arc<Coordinator>, agent: &str, source: Source, reason: String) -> u64 {
restart_many(coord, &[agent.to_owned()], false, source, reason)
// ---- dynamic power-op DAG assembly ----------------------------------------
//
// The pure per-agent chain builders below take `running` (and `stale`)
// explicitly so they stay pure + unit-testable without a live container;
// the async `*_many` fns read the real state via `lifecycle::is_running`
// then hand it in. Each chain uses LOCAL (0-based) deps; `concat_subgraphs`
// rebases them into one DAG.
/// One agent's **stop** subgraph. `SetWanted(Off)` head + `Reconcile` tail
/// always; the graceful `Signal → Drain` quiesce only when the agent is
/// 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.
fn stop_chain(agent: &str, graceful: bool, running: bool) -> Vec<NodeSpec> {
let mut n = vec![node(agent, NodeKind::SetWanted { up: false }, Vec::new())];
if graceful && running {
n.push(node(agent, NodeKind::Signal, after_ok(0)));
n.push(node(agent, NodeKind::Drain, after_ok(1)));
n.push(node(agent, NodeKind::Reconcile, after_ok(2)));
} else {
n.push(node(agent, NodeKind::Reconcile, after_ok(0)));
}
n
}
/// 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).
fn start_chain(agent: &str, running: bool, stale: bool) -> Vec<NodeSpec> {
let mut n = vec![node(agent, NodeKind::SetWanted { up: true }, Vec::new())];
if !running && stale {
// Rebuild subgraph rooted at the SetWanted head (base = 1, so
// `Prebuild` deps `after_ok(0)` = the head).
n.extend(rebuild_nodes(agent, true, 1));
} else {
n.push(node(agent, NodeKind::Reconcile, after_ok(0)));
}
n
}
/// One agent's **restart** subgraph. `SetWanted(Up)` head + `Reconcile`
/// tail; the stop portion (`Signal → Drain` when graceful, then
/// `StopForUpdate`) only when the agent is running — a restart of a down
/// agent is really a start (`SetWanted(Up) → Reconcile`).
fn restart_chain(agent: &str, graceful: bool, running: bool) -> Vec<NodeSpec> {
let mut n = vec![node(agent, NodeKind::SetWanted { up: true }, Vec::new())];
if running {
let mut prev = 0u32;
if graceful {
n.push(node(agent, NodeKind::Signal, after_ok(prev)));
prev += 1;
n.push(node(agent, NodeKind::Drain, after_ok(prev)));
prev += 1;
}
n.push(node(agent, NodeKind::StopForUpdate, after_ok(prev)));
prev += 1;
n.push(node(agent, NodeKind::Reconcile, after_ok(prev)));
} else {
n.push(node(agent, NodeKind::Reconcile, after_ok(0)));
}
n
}
/// Concatenate per-agent subgraphs (each with LOCAL 0-based deps) into one
/// node list, rebasing each subgraph's internal deps by its offset. A
/// subgraph root (empty deps — the `SetWanted` head) stays a root, so the
/// per-agent subgraphs are independent and run concurrently, each on its
/// own lease.
fn concat_subgraphs(chains: Vec<Vec<NodeSpec>>) -> Vec<NodeSpec> {
let mut out: Vec<NodeSpec> = Vec::new();
for chain in chains {
let base = u32::try_from(out.len()).unwrap_or(u32::MAX);
for spec in chain {
let deps = spec
.deps
.into_iter()
.map(|d| Dep {
on: base + d.on,
when: d.when,
})
.collect();
out.push(NodeSpec {
agent: spec.agent,
kind: spec.kind,
deps,
});
}
}
out
}
/// Wrap assembled power-op `nodes` in a `DagSpec`.
fn power_dag(
template: Template,
transient: TransientKind,
source: Source,
reason: String,
nodes: Vec<NodeSpec>,
) -> DagSpec {
DagSpec {
template,
source,
reason,
parent_id: None,
approval_id: None,
inputs: Vec::new(),
perm_payload: None,
transient: Some(transient),
nodes,
}
}
// The `*_spec` builders below are the PURE core the async `*_many` fns call
// after reading live state — they take the per-agent running (and stale)
// flags explicitly, so unit tests exercise the online/offline shapes without
// a live container. `*_many` = gather state + call `*_spec` + submit.
/// Assemble the stop DAG from explicit `(agent, running)` targets.
pub(crate) fn stop_spec(
targets: &[(String, bool)],
graceful: bool,
source: Source,
reason: String,
) -> DagSpec {
let chains = targets
.iter()
.map(|(agent, running)| stop_chain(agent, graceful, *running))
.collect();
let template = if graceful {
Template::GracefulStop
} else {
Template::Stop
};
power_dag(
template,
TransientKind::Stopping,
source,
reason,
concat_subgraphs(chains),
)
}
/// Assemble the start DAG from explicit `(agent, running, stale)` targets.
/// Transient is `Rebuilding` when any down+stale agent grew a rebuild
/// subgraph (crash-watch suppression during its Swap), else `Starting`;
/// applied per-agent at claim time, so each agent still shows its own pill.
pub(crate) fn start_spec(
targets: &[(String, bool, bool)],
source: Source,
reason: String,
) -> DagSpec {
let chains = targets
.iter()
.map(|(agent, running, stale)| start_chain(agent, *running, *stale))
.collect();
let any_rebuild = targets.iter().any(|(_, running, stale)| !running && *stale);
let transient = if any_rebuild {
TransientKind::Rebuilding
} else {
TransientKind::Starting
};
power_dag(
Template::Start,
transient,
source,
reason,
concat_subgraphs(chains),
)
}
/// Assemble the restart DAG from explicit `(agent, running)` targets.
pub(crate) fn restart_spec(
targets: &[(String, bool)],
graceful: bool,
source: Source,
reason: String,
) -> DagSpec {
let chains = targets
.iter()
.map(|(agent, running)| restart_chain(agent, graceful, *running))
.collect();
let template = if graceful {
Template::GracefulRestart
} else {
Template::Restart
};
power_dag(
template,
TransientKind::Restarting,
source,
reason,
concat_subgraphs(chains),
)
}
/// 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
}
/// 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, running concurrently on their own leases. `graceful`
/// prepends the signal→drain quiesce per agent. The whole hive-wide
/// `hivectl restart` / `restart-all` is now one DAG instead of N.
pub fn restart_many(
/// 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
/// `SetWanted(Up) → Reconcile` (there's nothing to stop). The whole
/// hive-wide `hivectl restart` / `restart-all` is one DAG.
pub async fn restart_many(
coord: &Arc<Coordinator>,
agents: &[String],
graceful: bool,
source: Source,
reason: String,
) -> u64 {
submit_and_emit(coord, templates::restart(agents, graceful, source, reason))
}
/// Start: `SetWanted(Up)` (a DAG node now) then reconcile. A stale rev
/// marker upgrades the start to a rebuild-then-start (`stale_start`, whose
/// tail `Reconcile` does the start) so the container always comes up on
/// current derivations — the old fast-lane `run_start` upgrade, still a
/// submit-time *shape* decision (which template), while the intent write
/// itself is now the template's head node.
pub fn start(coord: &Arc<Coordinator>, agent: &str, source: Source, reason: String) -> u64 {
let stored = std::fs::read_to_string(crate::paths::applied_rev_marker(agent)).ok();
let stale = crate::auto_update::current_flake_rev(&coord.hyperhive_flake)
.is_some_and(|rev| stored.as_deref() != Some(rev.as_str()));
if stale {
tracing::info!(%agent, "start: rev stale — upgrading to rebuild+start");
return submit_and_emit(
coord,
templates::stale_start(agent, source, format!("{reason} (stale — rebuild+start)")),
);
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, templates::start(agent, source, reason))
submit_and_emit(coord, restart_spec(&targets, graceful, source, reason))
}
/// Hard stop: `SetWanted(Offline)` (a DAG node now) then reconcile (kill +
/// unregister + `Killed` event).
pub fn stop(coord: &Arc<Coordinator>, agent: &str, source: Source, reason: String) -> u64 {
submit_and_emit(coord, templates::stop(agent, source, reason))
/// 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
}
/// Graceful stop: signal → drain → reconcile (the actual stop). The head
/// `SetWanted(Offline)` node writes the intent as part of the DAG.
pub fn graceful_stop(coord: &Arc<Coordinator>, agent: &str, source: Source, reason: String) -> u64 {
submit_and_emit(coord, templates::graceful_stop(agent, source, reason))
/// 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],
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 {
let running = lifecycle::is_running(agent).await;
let stored = std::fs::read_to_string(crate::paths::applied_rev_marker(agent)).ok();
let stale = current
.as_ref()
.is_some_and(|rev| stored.as_deref() != Some(rev.as_str()));
if !running && stale {
tracing::info!(%agent, "start: rev stale + agent down — rebuild-then-start");
}
targets.push((agent.clone(), running, stale));
}
submit_and_emit(coord, start_spec(&targets, source, reason))
}
/// Graceful restart of a single agent: signal → drain → mechanical stop →
/// reconcile (starts it back up) — one atomic DAG, no client-side "await
/// the stop DAG then submit a start DAG" split. The head `SetWanted(Up)`
/// node writes the intent as part of the DAG. Thin wrapper over
/// [`restart_many`] with `graceful = true`.
pub fn graceful_restart(
/// 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 {
restart_many(coord, &[agent.to_owned()], true, source, reason)
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,
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));
}
submit_and_emit(coord, stop_spec(&targets, graceful, source, reason))
}
/// Perm change: commit the JSON file(s) then rebuild.

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

@ -4,47 +4,45 @@
//! `Vec<Node>` + `deps`).
//!
//! Every node carries its own `agent` (there is no DAG-level agent) — the
//! `node` helper stamps the template's agent onto each. `restart` takes an
//! agent *list* and stamps each agent onto its own subgraph, so a hive-wide
//! `hivectl restart` is ONE DAG with N independent per-agent subgraphs
//! (each a root chain, run concurrently on its own lease) rather than N
//! separate DAGs. The other templates are still single-agent.
//!
//! The power ops write the durable `wanted` intent via a head
//! `SetWanted(w)` node (not a pre-submit side effect); it holds the agent
//! lease so intent+reconcile is atomic per-agent.
//! `node` helper stamps each node's agent. This module holds the *pure*
//! shape builders (no I/O). The hive-wide **power ops** (`stop` / `start` /
//! `restart`) are NOT here: their per-agent shape depends on each agent's
//! live running state (an async `lifecycle::is_running` read), so they are
//! assembled dynamically in `submit.rs` out of the shared pure primitives
//! this module exports (`node`, `after_ok`, `rebuild_nodes`) — one
//! independent per-agent subgraph each, concurrent on its own lease, ONE
//! DAG for the whole hive-wide op. Power ops write the durable `wanted`
//! intent via a head `SetWanted(w)` node (holding the agent lease, so
//! intent+reconcile is atomic per-agent).
//!
//! ```text
//! rebuild(a): Prebuild(a) → StopForUpdate(a) → Swap(a) →(any) Reconcile(a)
//! graceful-stop(a): SetWanted(a,Off) → Signal(a) → Drain(a) → Reconcile(a)
//! restart(a): SetWanted(a,Up) → StopForUpdate(a) → Reconcile(a)
//! graceful-restart(a): SetWanted(a,Up) → Signal(a) → Drain(a) → StopForUpdate(a) → Reconcile(a)
//! start(a): SetWanted(a,Up) → Reconcile(a)
//! stop(a): SetWanted(a,Off) → Reconcile(a)
//! stale-start(a): SetWanted(a,Up) → «rebuild subgraph» (rev stale; prebuild noops, agent down)
//! spawn(a): Provision(a) → Create(a) → WriteDropin(a) → Reconcile(a) [wanted=Up at approve]
//! perm-change(a): WritePermFile(a) → «rebuild subgraph»
//! meta-update(inp): MetaLock(inp) → «fan-out rebuild(a) per affected a»
//! startup sweep: MetaLock(hyperhive, non-fatal) → «fan-out rebuild(stale a)»
//! ```
//!
//! For the dynamic power-op shapes (`stop` / `start` / `restart`, built from
//! live online/offline state), see `submit.rs`.
use anyhow::{Result, bail};
use super::model::{DagSpec, Dep, DepWhen, NodeKind, NodeSpec, PermPayload, Source, Template};
use crate::coordinator::TransientKind;
/// After-ok edge on the previous node — the common chain link.
fn after_ok(on: u32) -> Vec<Dep> {
/// After-ok edge on the previous node — the common chain link. Shared with
/// the async power-op builders in `submit.rs` (which assemble per-agent
/// chains dynamically from live container state).
pub(crate) fn after_ok(on: u32) -> Vec<Dep> {
vec![Dep {
on,
when: DepWhen::AfterOk,
}]
}
/// Build one node targeting `agent`. The single place templates stamp a
/// node's agent, so a whole template is single-agent by passing the same
/// `agent` to every call.
fn node(agent: &str, kind: NodeKind, deps: Vec<Dep>) -> NodeSpec {
/// Build one node targeting `agent`. The single place a node's agent is
/// stamped. Shared with `submit.rs`'s dynamic power-op builders.
pub(crate) fn node(agent: &str, kind: NodeKind, deps: Vec<Dep>) -> NodeSpec {
NodeSpec {
agent: agent.to_owned(),
kind,
@ -56,7 +54,7 @@ fn node(agent: &str, kind: NodeKind, deps: Vec<Dep>) -> NodeSpec {
/// it must run even when the profile swap failed, so a previously-up
/// agent comes back on its old config (today's recovery-start). This
/// is the only `AfterAny` edge in v1.
fn rebuild_nodes(agent: &str, relock: bool, base: u32) -> Vec<NodeSpec> {
pub(crate) fn rebuild_nodes(agent: &str, relock: bool, base: u32) -> Vec<NodeSpec> {
vec![
node(
agent,
@ -122,73 +120,6 @@ pub fn approval_deploy(agent: &str, approval_id: i64, reason: String) -> DagSpec
}
}
/// Graceful stop: cheap `Signal` fires immediately (no build slot), the
/// `Drain` awaits the harness checkpoint (bounded), and the tail
/// `Reconcile` performs the actual container stop. The head `SetWanted`
/// node writes `wanted = Offline` as part of the DAG (was a pre-submit
/// side effect). A whole-hive graceful stop therefore signals every agent
/// up front and overlaps every drain, replacing the old detached-watcher
/// thread structurally.
pub fn graceful_stop(agent: &str, source: Source, reason: String) -> DagSpec {
DagSpec {
template: Template::GracefulStop,
source,
reason,
parent_id: None,
approval_id: None,
inputs: Vec::new(),
perm_payload: None,
transient: Some(TransientKind::Stopping),
nodes: vec![
node(agent, NodeKind::SetWanted { up: false }, Vec::new()),
node(agent, NodeKind::Signal, after_ok(0)),
node(agent, NodeKind::Drain, after_ok(1)),
node(agent, NodeKind::Reconcile, after_ok(2)),
],
}
}
/// Restart one or more agents in a **single** DAG. Each agent gets an
/// independent subgraph — a head `SetWanted(Up)` (a root: no cross-agent
/// dep) then its restart chain — so all agents' restarts run concurrently,
/// each taking its own agent lease. `graceful` inserts `Signal → Drain`
/// before the mechanical `StopForUpdate` (each agent's subgraph is still one
/// atomic restart). One agent = the ordinary single-agent restart; many =
/// a hive-wide `hivectl restart` as one DAG instead of N separate ones.
pub fn restart(agents: &[String], graceful: bool, source: Source, reason: String) -> DagSpec {
let mut nodes = Vec::new();
for agent in agents {
let base = u32::try_from(nodes.len()).unwrap_or(u32::MAX);
// Head of this agent's subgraph — a root (empty deps), so the N
// per-agent subgraphs are independent and run concurrently.
nodes.push(node(agent, NodeKind::SetWanted { up: true }, Vec::new()));
if graceful {
nodes.push(node(agent, NodeKind::Signal, after_ok(base)));
nodes.push(node(agent, NodeKind::Drain, after_ok(base + 1)));
nodes.push(node(agent, NodeKind::StopForUpdate, after_ok(base + 2)));
nodes.push(node(agent, NodeKind::Reconcile, after_ok(base + 3)));
} else {
nodes.push(node(agent, NodeKind::StopForUpdate, after_ok(base)));
nodes.push(node(agent, NodeKind::Reconcile, after_ok(base + 1)));
}
}
DagSpec {
template: if graceful {
Template::GracefulRestart
} else {
Template::Restart
},
source,
reason,
parent_id: None,
approval_id: None,
inputs: Vec::new(),
perm_payload: None,
transient: Some(TransientKind::Restarting),
nodes,
}
}
/// Boot-time reconcile: a single `Reconcile` node that converges observed
/// power state to the persisted intent — `wanted` is untouched (no
/// `SetWanted`), unlike the operator `start`/`stop` templates. Used only
@ -213,68 +144,6 @@ pub fn reconcile_only(
}
}
/// Start: write `wanted = Up` (head `SetWanted`), then reconcile (which
/// starts the container). The intent write is a DAG node now, not a
/// pre-submit side effect.
pub fn start(agent: &str, source: Source, reason: String) -> DagSpec {
DagSpec {
template: Template::Start,
source,
reason,
parent_id: None,
approval_id: None,
inputs: Vec::new(),
perm_payload: None,
transient: Some(TransientKind::Starting),
nodes: vec![
node(agent, NodeKind::SetWanted { up: true }, Vec::new()),
node(agent, NodeKind::Reconcile, after_ok(0)),
],
}
}
/// Stop: write `wanted = Offline` (head `SetWanted`), then reconcile
/// (kill + unregister + `Killed` event).
pub fn stop(agent: &str, source: Source, reason: String) -> DagSpec {
DagSpec {
template: Template::Stop,
source,
reason,
parent_id: None,
approval_id: None,
inputs: Vec::new(),
perm_payload: None,
transient: Some(TransientKind::Stopping),
nodes: vec![
node(agent, NodeKind::SetWanted { up: false }, Vec::new()),
node(agent, NodeKind::Reconcile, after_ok(0)),
],
}
}
/// Stale start: a `start` whose rev marker is stale, so it rebuilds
/// before coming up — `SetWanted(Up)` → «rebuild subgraph» → `Reconcile`
/// (the rebuild's tail `Reconcile` starts it, since `wanted = Up`). The
/// rebuild nodes are the same `rebuild_nodes` chain a manual rebuild uses
/// (reused, not a variant); only the leading `SetWanted(Up)` intent
/// differs. Shows as a `Rebuild` on the dashboard like the old
/// submit-time upgrade did.
pub fn stale_start(agent: &str, source: Source, reason: String) -> DagSpec {
let mut nodes = vec![node(agent, NodeKind::SetWanted { up: true }, Vec::new())];
nodes.extend(rebuild_nodes(agent, true, 1));
DagSpec {
template: Template::Rebuild,
source,
reason,
parent_id: None,
approval_id: None,
inputs: Vec::new(),
perm_payload: None,
transient: Some(TransientKind::Rebuilding),
nodes,
}
}
/// First-deploy spawn (approval-driven): `Provision` (proposed/applied
/// repos, state subvolume, meta registration) then `Create`
/// (`nixos-container create`), drop-in write, then `Reconcile` starts

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

@ -16,6 +16,22 @@ fn rebuild(agent: &str, reason: &str) -> DagSpec {
templates::rebuild(agent, Source::Manual, reason.to_owned(), None, true)
}
/// Restart DAG spec with every agent treated as **running** — the online
/// shape (`SetWanted → [Signal→Drain→] StopForUpdate → Reconcile`) most
/// queue-mechanics tests assume. Mirrors the pre-dynamic `templates::restart`
/// (which is now the state-aware `submit::restart_spec`).
fn restart_online(agents: &[&str], graceful: bool, reason: &str) -> DagSpec {
let targets: Vec<(String, bool)> = agents.iter().map(|a| ((*a).to_owned(), true)).collect();
submit::restart_spec(&targets, graceful, Source::Manual, reason.to_owned())
}
/// Stop DAG spec with every agent treated as **running** — the online shape
/// (`SetWanted → [Signal→Drain→](graceful) Reconcile`).
fn stop_online(agents: &[&str], graceful: bool, reason: &str) -> DagSpec {
let targets: Vec<(String, bool)> = agents.iter().map(|a| ((*a).to_owned(), true)).collect();
submit::stop_spec(&targets, graceful, Source::Manual, reason.to_owned())
}
/// Claim helper asserting exactly one node comes back.
fn claim_one(q: &JobQueue) -> Claim {
let mut claims = q.claim_ready();
@ -65,15 +81,7 @@ fn distinct_submits_never_collapse() {
let q = JobQueue::new(1);
let a = submit(&q, rebuild("agent-a", "r"));
let b = submit(&q, rebuild("agent-b", "r"));
let c = submit(
&q,
templates::restart(
&["agent-a".to_owned()],
false,
Source::Manual,
"r".to_owned(),
),
);
let c = submit(&q, restart_online(&["agent-a"], false, "r"));
assert_ne!(a, b);
assert_ne!(a, c);
assert_eq!(q.snapshot().len(), 3);
@ -203,15 +211,7 @@ fn fifo_fairness_for_the_slot() {
#[test]
fn lease_serializes_two_lifecycle_dags_for_same_agent() {
let q = JobQueue::new(4);
let restart = submit(
&q,
templates::restart(
&["agent-a".to_owned()],
false,
Source::Manual,
"restart".to_owned(),
),
);
let restart = submit(&q, restart_online(&["agent-a"], false, "restart"));
let stop = submit(
&q,
templates::reconcile_only(
@ -292,24 +292,8 @@ fn lease_exempt_prebuild_overlaps_other_dag_on_same_agent() {
#[test]
fn agents_do_not_contend_on_each_others_leases() {
let q = JobQueue::new(4);
submit(
&q,
templates::restart(
&["agent-a".to_owned()],
false,
Source::Manual,
"r".to_owned(),
),
);
submit(
&q,
templates::restart(
&["agent-b".to_owned()],
false,
Source::Manual,
"r".to_owned(),
),
);
submit(&q, restart_online(&["agent-a"], false, "r"));
submit(&q, restart_online(&["agent-b"], false, "r"));
let claims = q.claim_ready();
assert_eq!(claims.len(), 2, "different agents run concurrently");
}
@ -319,12 +303,7 @@ fn multi_agent_restart_is_one_dag_with_concurrent_per_agent_subgraphs() {
let q = JobQueue::new(4);
let id = submit(
&q,
templates::restart(
&["agent-a".to_owned(), "agent-b".to_owned()],
false,
Source::Manual,
"hive-wide".to_owned(),
),
restart_online(&["agent-a", "agent-b"], false, "hive-wide"),
);
// A hive-wide restart is ONE DAG, not one-per-agent.
assert_eq!(q.snapshot().len(), 1);
@ -348,6 +327,129 @@ fn multi_agent_restart_is_one_dag_with_concurrent_per_agent_subgraphs() {
);
}
#[test]
fn multi_agent_stop_is_one_dag_with_concurrent_per_agent_subgraphs() {
let q = JobQueue::new(4);
let id = submit(
&q,
stop_online(&["agent-a", "agent-b"], false, "hive-wide stop"),
);
// A hive-wide stop is ONE DAG, not one-per-agent.
assert_eq!(q.snapshot().len(), 1);
let claims = q.claim_ready();
assert!(claims.iter().all(|c| c.dag_id == id));
let mut heads: Vec<(&str, &str, bool)> = claims
.iter()
.map(|c| (c.agent.as_str(), c.kind.as_str(), c.lease_acquired))
.collect();
heads.sort_unstable();
assert_eq!(
heads,
vec![
("agent-a", "set_wanted", true),
("agent-b", "set_wanted", true),
],
"both per-agent stop subgraphs start concurrently, each on its own lease"
);
}
#[test]
fn multi_agent_start_one_dag_folds_per_agent_stale_rebuild() {
let q = JobQueue::new(4);
let id = submit(
&q,
// fresh: offline + not stale → SetWanted → Reconcile.
// stale: offline + stale → SetWanted → «rebuild subgraph».
submit::start_spec(
&[
("fresh".to_owned(), false, false),
("stale".to_owned(), false, true),
],
Source::Manual,
"hive-wide start".to_owned(),
),
);
// One DAG spanning both agents.
assert_eq!(q.snapshot().len(), 1);
// Both subgraph heads (SetWanted(Up)) are roots — claimable at once,
// each acquiring its own agent lease.
let heads = q.claim_ready();
assert!(
heads
.iter()
.all(|c| c.dag_id == id && c.kind.as_str() == "set_wanted")
);
let mut head_agents: Vec<&str> = heads.iter().map(|c| c.agent.as_str()).collect();
head_agents.sort_unstable();
assert_eq!(head_agents, vec!["fresh", "stale"]);
// Complete both heads; the fresh agent then reconciles directly while
// the stale agent's subgraph is the rebuild chain (prebuild first).
for c in &heads {
q.complete_node(id, c.node_id, Ok(()));
}
let next = q.claim_ready();
let mut kinds: Vec<(&str, &str)> = next
.iter()
.map(|c| (c.agent.as_str(), c.kind.as_str()))
.collect();
kinds.sort_unstable();
assert_eq!(
kinds,
vec![("fresh", "reconcile"), ("stale", "prebuild")],
"fresh agent starts directly; stale agent rebuilds first, all in one DAG"
);
}
#[test]
fn offline_agents_skip_mechanical_nodes_but_keep_reconcile() {
// The dynamic build skips Signal/Drain/StopForUpdate for a down agent
// (nothing to quiesce/stop) but ALWAYS keeps the Reconcile tail — the
// convergence guarantee that catches a race-up between the is_running
// read and node exec.
let q = JobQueue::new(4);
// Offline graceful stop → SetWanted(Off) → Reconcile (no Signal/Drain).
let stop = submit(
&q,
submit::stop_spec(
&[("down".to_owned(), false)],
true,
Source::Manual,
"stop down".to_owned(),
),
);
// Offline restart → SetWanted(Up) → Reconcile (no StopForUpdate): a
// restart of a down agent is really a start.
let restart = submit(
&q,
submit::restart_spec(
&[("down2".to_owned(), false)],
true,
Source::Manual,
"restart down".to_owned(),
),
);
let shape = |id: u64| -> Vec<String> {
q.snapshot()
.iter()
.find(|d| d.id == id)
.expect("dag")
.nodes
.iter()
.map(|n| n.kind.clone())
.collect()
};
assert_eq!(
shape(stop),
vec!["set_wanted".to_owned(), "reconcile".to_owned()],
"offline graceful stop skips the signal/drain quiesce, keeps Reconcile"
);
assert_eq!(
shape(restart),
vec!["set_wanted".to_owned(), "reconcile".to_owned()],
"offline restart skips StopForUpdate, keeps Reconcile (it's a start)"
);
}
// ---- failure: cancel-downstream + AfterAny ----
#[test]
@ -566,15 +668,7 @@ fn append_children_sets_parent() {
#[test]
fn terminal_dag_reported_exactly_once_and_lease_released() {
let q = JobQueue::new(1);
let id = submit(
&q,
templates::restart(
&["agent-a".to_owned()],
false,
Source::Manual,
"r".to_owned(),
),
);
let id = submit(&q, restart_online(&["agent-a"], false, "r"));
// restart = SetWanted → StopForUpdate → Reconcile; not terminal until
// the last node completes.
let set_wanted = claim_one(&q);
@ -681,8 +775,8 @@ fn trim_keeps_terminal_parent_with_live_children() {
);
let lock = claim_one(&q);
q.complete_node(meta, lock.node_id, Ok(()));
let mut child_spec = templates::restart(
&["agent-x".to_owned()],
let mut child_spec = submit::restart_spec(
&[("agent-x".to_owned(), true)],
false,
Source::MetaUpdate,
"cascade".to_owned(),
@ -798,10 +892,7 @@ fn error_is_truncated() {
#[test]
fn graceful_stop_shape_signal_drain_reconcile() {
let q = JobQueue::new(1);
let id = submit(
&q,
templates::graceful_stop("agent-a", Source::Manual, "graceful".to_owned()),
);
let id = submit(&q, stop_online(&["agent-a"], true, "graceful"));
for expected in ["set_wanted", "signal", "drain", "reconcile"] {
let c = claim_one(&q);
assert_eq!(c.kind.as_str(), expected);
@ -816,14 +907,8 @@ fn graceful_signal_and_drain_hold_no_build_slot() {
// buildSlots = 1 while a rebuild hogs the slot.
let q = JobQueue::new(1);
submit(&q, rebuild("builder", "slot hog"));
submit(
&q,
templates::graceful_stop("agent-a", Source::Manual, "g".to_owned()),
);
submit(
&q,
templates::graceful_stop("agent-b", Source::Manual, "g".to_owned()),
);
submit(&q, stop_online(&["agent-a"], true, "g"));
submit(&q, stop_online(&["agent-b"], true, "g"));
let claims = q.claim_ready();
let kinds: Vec<&str> = claims.iter().map(|c| c.kind.as_str()).collect();
assert_eq!(

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

@ -97,8 +97,8 @@ async fn dispatch(req: &HostRequest, coord: Arc<Coordinator>) -> HostResponse {
tracing::info!(%id, %name, "spawn approval queued");
HostResponse::success()
}
HostRequest::Kill { name } => submit_single(&coord, name, Verb::Kill),
HostRequest::Restart { name } => submit_single(&coord, name, Verb::Restart),
HostRequest::Kill { name } => submit_single(&coord, name, Verb::Kill).await,
HostRequest::Restart { name } => submit_single(&coord, name, Verb::Restart).await,
HostRequest::RestartAll => handle_restart_all(&coord).await?,
HostRequest::RestartScoped { scope, graceful } => {
handle_restart_scoped(&coord, scope, *graceful).await?
@ -143,7 +143,7 @@ async fn dispatch(req: &HostRequest, coord: Arc<Coordinator>) -> HostResponse {
actions::destroy(&coord, name, *purge).await?;
HostResponse::success()
}
HostRequest::Rebuild { name } => submit_single(&coord, name, Verb::Rebuild),
HostRequest::Rebuild { name } => submit_single(&coord, name, Verb::Rebuild).await,
HostRequest::QueueDag { id } => {
// The polled DAG first, then its live fan-out children.
let dags = coord
@ -558,7 +558,7 @@ enum Verb {
Rebuild,
}
fn submit_single(coord: &Arc<Coordinator>, name: &str, verb: Verb) -> HostResponse {
async fn submit_single(coord: &Arc<Coordinator>, name: &str, verb: Verb) -> HostResponse {
use crate::job_queue::{Source, submit};
let id = match verb {
Verb::Kill => {
@ -569,6 +569,7 @@ fn submit_single(coord: &Arc<Coordinator>, name: &str, verb: Verb) -> HostRespon
Source::Manual,
"manual kill via hivectl".to_owned(),
)
.await
}
Verb::Restart => {
tracing::info!(%name, "restart");
@ -578,6 +579,7 @@ fn submit_single(coord: &Arc<Coordinator>, name: &str, verb: Verb) -> HostRespon
Source::Manual,
"manual restart via hivectl".to_owned(),
)
.await
}
Verb::Rebuild => {
tracing::info!(%name, "rebuild");
@ -606,13 +608,16 @@ async fn handle_restart_all(coord: &Arc<Coordinator>) -> Result<HostResponse> {
let queued = if agents.is_empty() {
Vec::new()
} else {
vec![crate::job_queue::submit::restart_many(
coord,
&agents,
false,
crate::job_queue::Source::Manual,
"manual restart via hivectl restart-all".to_owned(),
)]
vec![
crate::job_queue::submit::restart_many(
coord,
&agents,
false,
crate::job_queue::Source::Manual,
"manual restart via hivectl restart-all".to_owned(),
)
.await,
]
};
let mut resp = HostResponse::list(agents);
resp.queued_dags = Some(queued);
@ -644,29 +649,27 @@ async fn handle_stop(
let mut errors: Vec<String> = Vec::new();
let mut queued: Vec<u64> = Vec::new();
for agent in agents {
// 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. A hive-wide
// `hivectl stop` is now a single DAG, not N.
if !agents.is_empty() {
let reason = if graceful {
"manual via hivectl graceful stop"
} else {
"manual via hivectl stop"
};
let id = if graceful {
crate::job_queue::submit::graceful_stop(
queued.push(
crate::job_queue::submit::stop_many(
coord,
agent,
agents,
graceful,
crate::job_queue::Source::Manual,
reason.to_owned(),
)
} else {
crate::job_queue::submit::stop(
coord,
agent,
crate::job_queue::Source::Manual,
reason.to_owned(),
)
};
queued.push(id);
ok_items.push(agent.clone());
.await,
);
ok_items.extend(agents.iter().cloned());
}
// Agents go down before infra so they're not mid-request against a
@ -739,18 +742,25 @@ async fn handle_start(
}
}
// One DAG for all targeted agents — a per-agent start subgraph each
// (`SetWanted(Up) → Reconcile`, or a rebuild-then-start for a stale
// rev), independent roots that run concurrently on their own leases. A
// hive-wide `hivectl start` is now a single DAG, not N. Through the
// queue: persists `wanted = Up`, per-agent stale-rev upgrade to a full
// rebuild, serializes on each agent's lease. The id rides back for
// hivectl's wait loop.
let mut queued: Vec<u64> = Vec::new();
for agent in agents {
// Through the queue: persists `wanted = Up`, upgrades a
// stale-rev start to a full rebuild, and serializes on the
// agent's lease. Ids ride back for hivectl's wait loop.
queued.push(crate::job_queue::submit::start(
coord,
agent,
crate::job_queue::Source::Manual,
"manual via hivectl start".to_owned(),
));
ok_items.push(agent.clone());
if !agents.is_empty() {
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);
@ -786,17 +796,20 @@ async fn handle_restart_scoped(
// client-side stop-then-start composition — the whole restart survives
// a dropped connection because the DAG owns it.
if !agents.is_empty() {
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()
},
));
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());
}

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

@ -12,7 +12,11 @@ use crate::coordinator::Coordinator;
/// `Start` — start a container, kicking its next turn. The caller must be an
/// ancestor of `name` in the topology (the root covers every agent).
pub(super) fn handle_start(coord: &Arc<Coordinator>, agent: &str, name: &str) -> AgentResponse {
pub(super) async fn handle_start(
coord: &Arc<Coordinator>,
agent: &str,
name: &str,
) -> AgentResponse {
if let Some(err) = require_descendant(agent, name, "start") {
return err;
}
@ -25,7 +29,8 @@ pub(super) fn handle_start(coord: &Arc<Coordinator>, agent: &str, name: &str) ->
name,
crate::job_queue::Source::Manual,
format!("agent `{agent}` start tool"),
);
)
.await;
AgentResponse::Ok
}
@ -56,7 +61,8 @@ pub(super) async fn handle_restart(
name,
crate::job_queue::Source::Manual,
format!("agent `{agent}` restart tool"),
);
)
.await;
AgentResponse::Ok
}

2
hive-c0re/src/socket_server/mod.rs
View file

@ -557,7 +557,7 @@ async fn dispatch(req: &AgentRequest, agent: &str, coord: &Arc<Coordinator>) ->
match req {
// Lifecycle + config: caller must be an ancestor of the target
// (a parent owns its whole subtree; the root covers every agent).
AgentRequest::Start { name } => handle_start(coord, agent, name),
AgentRequest::Start { name } => handle_start(coord, agent, name).await,
AgentRequest::Restart { name } => handle_restart(coord, agent, name).await,
AgentRequest::Kill { name } => handle_kill(coord, agent, name).await,
AgentRequest::Update { name } => handle_update(coord, agent, name),