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

swarm: add a declared "paused" agent wanted state

Browse source 
mara (#4170): swarm-ui's wanted-state dropdown could only ever declare
up/offline/destroy, with no way to swarm-declare the existing hive-local
turn-loop pause (`hivectl agent pause|resume`).

`AgentState::Paused` is not a fifth peer of Up/Offline/Destroyed on the
power axis this enum otherwise answers — it's Up plus an orthogonal
turn-loop pause. `hive-c0re`'s `workers::wanted` reconcile loop now
decides the two axes independently (`decide` for power, the new
`decide_pause` for the marker), so a stopped agent declared Paused
converges with both a Start and a Pause in the same pass.

Known, deliberate limitation: a Paused declaration on an agent this
hive has never deployed only reaches Deploy this pass — writing the
pause marker into a harness dir that may not exist yet was judged not
worth the risk, so it converges on the next pass once the agent is
present instead.

swarm-ui's WantedMenu gains a fourth "paused" option (warning-tone
badge). No separate "resume" entry — selecting "up" from a paused row
already clears the marker via the same decide_pause path.

Pause/resume marker writes go through one shared
Coordinator::set_paused_by_name helper, used by both the interactive
dashboard pause/resume handlers and this reconcile loop, instead of
each duplicating the parse-name/write-marker/track-rescan shape.
swarm-ui's "offline" and "paused" confirm dialogs share one
confirmTarget state and one ConfirmDialog instead of two near-identical
copies.

Closes #4170
This commit is contained in:
iris 2026-09-11 00:22:08 +02:00 • committed by mara
commit 513554fe9a
6 changed files with 483 additions and 55 deletions
Download patch file Download diff file Expand all files Collapse all files

42
hive-c0re/src/coordinator.rs
View file

@ -440,6 +440,26 @@ pub struct ApprovalAdded<'a> {
pub pr_number: Option<u64>,
}
/// The two ways [`Coordinator::set_paused_by_name`] can fail. See its own
/// doc comment for why this is a distinct type rather than one
/// `anyhow::Error` (`Write` already wraps one, `BadName` never needs to).
#[derive(Debug)]
pub enum SetPausedByNameError {
/// `name` did not parse as a [`hive_types::Ident`] — the parse error's
/// own message.
BadName(&'static str),
Write(anyhow::Error),
}
impl std::fmt::Display for SetPausedByNameError {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
match self {
SetPausedByNameError::BadName(msg) => write!(f, "bad agent name: {msg}"),
SetPausedByNameError::Write(e) => write!(f, "{e}"),
}
}
}
impl Coordinator {
pub fn open(
db_path: &Path,
@ -1449,6 +1469,28 @@ impl Coordinator {
crate::priv_client::set_agent_paused(name.as_str(), paused).await
}
/// [`Self::set_paused`] from a plain agent-name string, parsing it
/// first — the exact shape both `dashboard::lifecycle_ops`'s
/// `/api/pause`/`/api/resume` handlers and `workers::wanted`'s
/// reconcile loop need, previously duplicated between them (mara, PR
/// review: "could this be a bit less duplicated code?"). A distinct
/// error variant per failure mode, not a single
/// `anyhow::Error`, because the two callers report the two cases
/// differently: a bad name is the caller's mistake (400 there, a
/// warned skip in the reconcile loop), a write failure is this
/// host's (500 there, also a warned skip here).
///
/// # Errors
///
/// `BadName` when `name` is not a legal [`hive_types::Ident`];
/// `Write` for everything [`Self::set_paused`] itself can fail with.
pub async fn set_paused_by_name(name: &str, paused: bool) -> Result<(), SetPausedByNameError> {
let id = hive_types::Ident::parse(name).map_err(SetPausedByNameError::BadName)?;
Self::set_paused(&id, paused)
.await
.map_err(SetPausedByNameError::Write)
}
/// Enumerate names that have a persistent state dir under
/// `/var/lib/hyperhive/agents/` (i.e. config / claude creds /
/// notes survive). Includes both currently-existing containers and

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

@ -262,12 +262,8 @@ pub(super) async fn post_pause(
if let Some(reject) = guard_agent_name(&state, &logical).await {
return reject;
}
let ident = match Ident::parse(&logical) {
Ok(i) => i,
Err(e) => return (StatusCode::BAD_REQUEST, format!("bad agent name: {e}")).into_response(),
};
if let Err(e) = crate::coordinator::Coordinator::set_paused(&ident, true).await {
return error_response(&format!("pause {logical}: {e}"));
if let Err(e) = crate::coordinator::Coordinator::set_paused_by_name(&logical, true).await {
return set_paused_error_response(&logical, &e);
}
state.coord.rescan_containers_and_emit().await;
(StatusCode::OK, "ok").into_response()
@ -297,17 +293,30 @@ pub(super) async fn post_resume(
if let Some(reject) = guard_agent_name(&state, &logical).await {
return reject;
}
let ident = match Ident::parse(&logical) {
Ok(i) => i,
Err(e) => return (StatusCode::BAD_REQUEST, format!("bad agent name: {e}")).into_response(),
};
if let Err(e) = crate::coordinator::Coordinator::set_paused(&ident, false).await {
return error_response(&format!("resume {logical}: {e}"));
if let Err(e) = crate::coordinator::Coordinator::set_paused_by_name(&logical, false).await {
return set_paused_error_response(&logical, &e);
}
state.coord.rescan_containers_and_emit().await;
(StatusCode::OK, "ok").into_response()
}
/// Render a [`crate::coordinator::SetPausedByNameError`] as the response
/// `post_pause`/`post_resume` both need: 400 for a bad name (the caller's
/// mistake), 500 for a write failure (this host's).
fn set_paused_error_response(
logical: &str,
e: &crate::coordinator::SetPausedByNameError,
) -> Response {
match e {
crate::coordinator::SetPausedByNameError::BadName(_) => {
(StatusCode::BAD_REQUEST, e.to_string()).into_response()
}
crate::coordinator::SetPausedByNameError::Write(_) => {
error_response(&format!("{logical}: {e}"))
}
}
}
/// Form fields for `post_resource_limits`. Both fields are optional strings;
/// an empty value clears the per-agent override for that field, falling back
/// to the hive-wide default.

301
hive-c0re/src/workers/wanted.rs
View file

@ -206,13 +206,34 @@ async fn converge(coord: &Arc<Coordinator>, declared: &HiveWanted) -> Result<()>
}
}
let plan = plan(declared, &present, &intents);
// The pause marker is a stat, not a store read — it cannot fail the way
// `coord.power.get` above can, so there is no "unreadable, skip" branch
// to mirror. A name that does not parse as an `Ident` is skipped with a
// warning instead, the same defensive shape `crash_watch` uses for the
// same reason: a declaration is attacker-adjacent input (published by
// the swarm controller, not typed by an operator at this hive), so a
// malformed name should not panic the convergence loop.
let mut paused = BTreeMap::new();
for agent in declared.agents.keys() {
match hive_types::Ident::parse(agent) {
Ok(id) => {
paused.insert(agent.clone(), Coordinator::is_paused(&id));
}
Err(e) => {
tracing::warn!(%agent, error = %e, "wanted state: invalid agent name; skipping pause check");
}
}
}
let plan = plan(declared, &present, &intents, &paused);
tracing::info!(
declared = declared.agents.len(),
deploy = plan.deploy.len(),
start = plan.start.len(),
stop = plan.stop.len(),
destroy = plan.destroy.len(),
pause = plan.pause.len(),
resume = plan.resume.len(),
"wanted state: converging"
);
@ -247,16 +268,46 @@ async fn converge(coord: &Arc<Coordinator>, declared: &HiveWanted) -> Result<()>
// The power ops and destroy emit their own; the first deploys do not.
coord.emit_rebuild_queue_snapshot();
}
// `set_paused_by_name` — same helper `dashboard::lifecycle_ops`'s
// `/api/pause`/`/api/resume` use, not the job-queue DAG those
// name-check in their own doc comments — this loop already has its own
// per-agent retry story (redeclare-or-rewatch), so there is nothing
// here to wait on an acknowledgement for. One loop over both lists
// paired with their target `paused` value, not two near-identical
// copies (argus, PR review).
let mut rescan = false;
for (agents, paused) in [(&plan.pause, true), (&plan.resume, false)] {
for agent in agents {
match Coordinator::set_paused_by_name(agent, paused).await {
Ok(()) => rescan = true,
Err(e) => {
tracing::warn!(%agent, paused, error = %e, "wanted state: set-paused failed");
}
}
}
}
if rescan {
// Same call `post_pause`/`post_resume` make, so the dashboard's
// "paused" badge flips without waiting for the next periodic sweep.
coord.rescan_containers_and_emit().await;
}
Ok(())
}
/// Everything one declaration asks this hive to queue.
///
/// `pause`/`resume` are a second, independent axis from the other four —
/// see [`decide_pause`]'s own comment for why an agent can land in one of
/// the power-axis lists *and* one of these in the same pass.
#[derive(Debug, Default, PartialEq, Eq)]
struct Plan {
deploy: Vec<String>,
start: Vec<String>,
stop: Vec<String>,
destroy: Vec<String>,
pause: Vec<String>,
resume: Vec<String>,
}
/// Turn one declaration into that plan — pure, so what the loop does with a
@ -274,13 +325,26 @@ fn plan(
declared: &HiveWanted,
present: &BTreeSet<String>,
intents: &BTreeMap<String, Option<Wanted>>,
paused: &BTreeMap<String, bool>,
) -> Plan {
let mut plan = Plan::default();
for (agent, decl) in &declared.agents {
// The pause axis does not gate on a readable power intent the way
// the match below does — it has no intent row to fail reading, only
// a stat that always answers — so it is evaluated unconditionally,
// even for an agent this pass otherwise skips.
let is_present = present.contains(agent);
let is_paused = paused.get(agent).copied().unwrap_or(false);
match decide_pause(decl.state, is_present, is_paused) {
PauseConverge::Pause => plan.pause.push(agent.clone()),
PauseConverge::Resume => plan.resume.push(agent.clone()),
PauseConverge::Nothing => {}
}
let Some(intent) = intents.get(agent) else {
continue;
};
match decide(decl.state, present.contains(agent), *intent) {
match decide(decl.state, is_present, *intent) {
Converge::Deploy => plan.deploy.push(agent.clone()),
Converge::Start => plan.start.push(agent.clone()),
Converge::Stop => plan.stop.push(agent.clone()),
@ -310,10 +374,14 @@ enum Converge {
/// The whole decision, pure: no queue, no store, no container. The three
/// inputs are exactly what [`converge`] reads per agent, so the table below is
/// the behaviour rather than a model of it.
///
/// `Paused` shares every one of `Up`'s arms here — on the power axis a
/// paused agent is still a running one, just with its turn loop parked.
/// [`decide_pause`] is where the two diverge.
fn decide(state: AgentState, present: bool, intent: Option<Wanted>) -> Converge {
match state {
AgentState::Up if !present => Converge::Deploy,
AgentState::Up => {
AgentState::Up | AgentState::Paused if !present => Converge::Deploy,
AgentState::Up | AgentState::Paused => {
if intent == Some(Wanted::Up) {
Converge::Nothing
} else {
@ -339,11 +407,52 @@ fn decide(state: AgentState, present: bool, intent: Option<Wanted>) -> Converge
}
}
/// What the loop will do about one declared agent's turn-loop pause state.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum PauseConverge {
Pause,
Resume,
/// Either the pause marker already matches the declaration, or there is
/// no container yet to hold one — see this function's own comment on
/// `!present`.
Nothing,
}
/// The pause-axis half of [`decide`], deliberately **not** folded into it:
/// the two vary independently (an agent can need `Converge::Start` and
/// `PauseConverge::Pause` in the same pass — declared `Paused`, currently
/// stopped), so a single combined enum would need one variant per
/// combination rather than the two small ones here.
///
/// `present` gates this exactly like [`decide`] gates deploy-vs-repair on
/// the power axis: a `Paused` declaration on an absent agent reaches
/// `Converge::Deploy` above, and the pause marker converges on the *next*
/// pass once that deploy has made the agent present, not this one — writing
/// a pause marker into a harness dir that may not exist yet is the same
/// "repair against a guess" this module's `Offline`/`Destroyed` present-gate
/// already refuses on the power axis.
fn decide_pause(state: AgentState, present: bool, currently_paused: bool) -> PauseConverge {
if !present {
return PauseConverge::Nothing;
}
match state {
AgentState::Paused if currently_paused => PauseConverge::Nothing,
AgentState::Paused => PauseConverge::Pause,
AgentState::Up if currently_paused => PauseConverge::Resume,
// `Up` already unpaused: agreement. `Offline`/`Destroyed`: the pause
// marker is moot either way (no turn loop runs on a stopped or gone
// container to park), so it is left as-is rather than cleared —
// whatever it says gets re-evaluated the moment either state moves
// back to `Up` or `Paused`.
AgentState::Up | AgentState::Offline | AgentState::Destroyed => PauseConverge::Nothing,
}
}
#[cfg(test)]
mod tests {
use std::collections::{BTreeMap, BTreeSet};
use super::{Converge, Plan, carries_a_declaration, decide, plan};
use super::{Converge, PauseConverge, Plan, carries_a_declaration, decide, decide_pause, plan};
use crate::power::Wanted;
use swarm_queue_client::wanted::{AgentState, AgentWanted, HiveWanted};
@ -367,6 +476,13 @@ mod tests {
.collect()
}
fn paused(entries: &[(&str, bool)]) -> BTreeMap<String, bool> {
entries
.iter()
.map(|(name, is_paused)| ((*name).to_owned(), *is_paused))
.collect()
}
/// The property the whole migration rides on: this hive's own agents are
/// not converged, they are not looked at. `adopted` is the control — the
/// same call must still act on what the declaration *does* name, or this
@ -378,6 +494,7 @@ mod tests {
&declared,
&set(&["adopted", "legacy"]),
&intents(&[("adopted", Some(Wanted::Offline)), ("legacy", None)]),
&paused(&[]),
);
assert_eq!(
planned,
@ -386,6 +503,8 @@ mod tests {
start: vec!["adopted".to_owned()],
stop: vec![],
destroy: vec![],
pause: vec![],
resume: vec![],
}
);
}
@ -399,11 +518,73 @@ mod tests {
&declared,
&set(&["unreadable", "readable"]),
&intents(&[("readable", None)]),
&paused(&[]),
);
assert_eq!(planned.start, vec!["readable".to_owned()]);
assert!(planned.deploy.is_empty() && planned.stop.is_empty());
}
/// The pause axis does not gate on a readable intent row at all — a
/// `paused` agent this hive has no power-intent row for still gets its
/// pause marker converged, unlike the power axis right above.
#[test]
fn pause_converges_even_when_the_power_intent_is_unreadable() {
let declared = declaration(&[("unreadable", AgentState::Paused)]);
let planned = plan(
&declared,
&set(&["unreadable"]),
&intents(&[]),
&paused(&[("unreadable", false)]),
);
assert_eq!(planned.pause, vec!["unreadable".to_owned()]);
assert!(planned.start.is_empty() && planned.deploy.is_empty());
}
/// A declared-paused agent that is present, running and already carries
/// the marker needs nothing on either axis.
#[test]
fn an_already_paused_agent_converges_to_nothing() {
let declared = declaration(&[("parked", AgentState::Paused)]);
let planned = plan(
&declared,
&set(&["parked"]),
&intents(&[("parked", Some(Wanted::Up))]),
&paused(&[("parked", true)]),
);
assert_eq!(planned, Plan::default());
}
/// The case `decide_pause`'s own comment calls out: declared `Paused` but
/// currently stopped needs a power `Start` *and* a `Pause`, in the same
/// pass — this is the test that would fail if the two axes were folded
/// into one enum instead of decided independently.
#[test]
fn a_stopped_agent_declared_paused_gets_both_a_start_and_a_pause() {
let declared = declaration(&[("parked", AgentState::Paused)]);
let planned = plan(
&declared,
&set(&["parked"]),
&intents(&[("parked", Some(Wanted::Offline))]),
&paused(&[("parked", false)]),
);
assert_eq!(planned.start, vec!["parked".to_owned()]);
assert_eq!(planned.pause, vec!["parked".to_owned()]);
}
/// A declared-`Up` agent whose marker is still set from an earlier
/// `Paused` declaration gets resumed.
#[test]
fn an_up_agent_still_carrying_the_marker_is_resumed() {
let declared = declaration(&[("parked", AgentState::Up)]);
let planned = plan(
&declared,
&set(&["parked"]),
&intents(&[("parked", Some(Wanted::Up))]),
&paused(&[("parked", true)]),
);
assert_eq!(planned.resume, vec!["parked".to_owned()]);
}
#[test]
fn a_declared_agent_this_hive_does_not_have_is_deployed() {
assert_eq!(decide(AgentState::Up, false, None), Converge::Deploy);
@ -477,6 +658,85 @@ mod tests {
);
}
/// `Paused` shares every `Up` arm on the power axis — same deploy-when-
/// absent, same start-when-stopped behaviour.
#[test]
fn paused_converges_on_the_power_axis_exactly_like_up() {
for present in [true, false] {
for intent in [None, Some(Wanted::Up), Some(Wanted::Offline)] {
assert_eq!(
decide(AgentState::Paused, present, intent),
decide(AgentState::Up, present, intent),
"present={present:?} intent={intent:?}"
);
}
}
}
#[test]
fn decide_pause_sets_the_marker_for_a_present_unpaused_paused_declaration() {
assert_eq!(
decide_pause(AgentState::Paused, true, false),
PauseConverge::Pause
);
}
#[test]
fn decide_pause_is_idempotent_once_the_marker_already_matches() {
assert_eq!(
decide_pause(AgentState::Paused, true, true),
PauseConverge::Nothing
);
assert_eq!(
decide_pause(AgentState::Up, true, false),
PauseConverge::Nothing
);
}
#[test]
fn decide_pause_clears_the_marker_for_a_declared_up_agent() {
assert_eq!(
decide_pause(AgentState::Up, true, true),
PauseConverge::Resume
);
}
/// No container, nothing to write a marker into — the same present-gate
/// [`decide`] applies to `Offline`/`Destroyed`, mirrored here for the
/// pause axis regardless of declared state.
#[test]
fn decide_pause_does_nothing_for_an_absent_agent() {
for state in [
AgentState::Up,
AgentState::Paused,
AgentState::Offline,
AgentState::Destroyed,
] {
for currently_paused in [true, false] {
assert_eq!(
decide_pause(state, false, currently_paused),
PauseConverge::Nothing,
"state={state:?} currently_paused={currently_paused:?}"
);
}
}
}
/// `Offline`/`Destroyed` never touch the marker either way — it is left
/// exactly as found, whatever that is.
#[test]
fn decide_pause_leaves_offline_and_destroyed_alone() {
for state in [AgentState::Offline, AgentState::Destroyed] {
for currently_paused in [true, false] {
assert_eq!(
decide_pause(state, true, currently_paused),
PauseConverge::Nothing,
"state={state:?} currently_paused={currently_paused:?}"
);
}
}
}
/// The watch's half of "absence is not a deletion order". `Put` is the
/// control: without it this would pass on a function that refused
/// everything, which would silently stop the fast path converging at all.
@ -501,12 +761,39 @@ mod tests {
/// pass forever without measuring anything.
#[test]
fn every_state_this_build_knows_is_covered_above() {
for state in [AgentState::Up, AgentState::Offline, AgentState::Destroyed] {
for state in [
AgentState::Up,
AgentState::Offline,
AgentState::Paused,
AgentState::Destroyed,
] {
let seen = [true, false].iter().any(|present| {
decide(state, *present, None) != Converge::Nothing
|| decide(state, *present, Some(Wanted::Up)) != Converge::Nothing
});
assert!(seen, "{state:?} produces no action in any combination");
assert!(
seen,
"{state:?} produces no power action in any combination"
);
}
}
/// The pause-axis equivalent of the test above: every state this build
/// knows produces a non-`Nothing` pause decision in at least one
/// combination, so a state that silently fell out of `decide_pause`
/// would be caught here rather than passing forever unmeasured. `Up` and
/// `Paused` are the two expected to; `Offline`/`Destroyed` are asserted
/// separately above as deliberately inert on this axis.
#[test]
fn up_and_paused_both_produce_a_pause_action_in_some_combination() {
for state in [AgentState::Up, AgentState::Paused] {
let seen = [true, false].iter().any(|currently_paused| {
decide_pause(state, true, *currently_paused) != PauseConverge::Nothing
});
assert!(
seen,
"{state:?} produces no pause action in any combination"
);
}
}
}