//! Power ops (`stop` / `start` / `restart`) — the DAG shapes whose per-agent //! form depends on **live** container state, so they cannot be static //! [`super::templates`] entries. //! //! The 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. //! //! Each entry point declares its group and inserts it. There is no metadata //! parameter and no container node: attribution is not something every caller //! has to invent, and a DAG is addressed by the nodes a template names. use std::sync::Arc; use hive_jobq::NodeId; use super::resource::Resource; use super::templates::rebuild_nodes; use super::{JobBuilder, NodeKind}; use crate::coordinator::Coordinator; use crate::lifecycle; // ---- 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 declares into the shared job it is handed, and // names the nodes it depends on — so there is nothing to rebase. /// 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(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). let a = || agent.to_owned(); let wanted = builder .node(NodeKind::SetWanted { agent: a(), up: false, }) .needs(Resource::Agent(a())); // Declaration order is dependency order: the quiesce steps come first so // the `Reconcile` that waits on them can name them. if graceful && running { // `SetWanted` is the brace here, so the quiesce pair borrows its grant // rather than declaring the lease itself — same shape the rebuild // template uses, one definition. let drain = super::templates::quiesce(builder, agent, wanted); let _ = builder .node(NodeKind::Reconcile { agent: a() }) .needs(Resource::Agent(a())) .part_of(wanted) .after_ok(drain); } else { let _ = builder .node(NodeKind::Reconcile { agent: a() }) .needs(Resource::Agent(a())) .part_of(wanted); } } /// 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(builder: &JobBuilder, agent: &str, running: bool, stale: bool) { let wanted = builder .node(NodeKind::SetWanted { agent: agent.to_owned(), up: true, }) .needs(Resource::Agent(agent.to_owned())); if !running && stale { // 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 { agent: agent.to_owned(), }) .needs(Resource::Agent(agent.to_owned())) .part_of(wanted); } } /// One agent's **restart** subgraph. Restart NEVER rewrites `wanted` /// intent (no `SetWanted` head, unlike stop/start): it bounces the /// container and lets the tail `Reconcile` converge to the agent's /// EXISTING intent, so a deliberately-stopped (`wanted = Off`) agent is /// not forced back up by a hive-wide restart. A running agent gets the /// mechanical stop (`Signal → Drain` when graceful, then `StopForUpdate`) /// 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. 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. let _ = builder .node(NodeKind::Reconcile { agent: a() }) .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 // children (borrow the lease, dep-ordered), so the bounce holds one // continuous lease and `Reconcile` cancel-cascades if a stop step fails. // // `Reconcile` gates on the last mechanical step. For a non-graceful bounce // that step *is* the root, and the parent gate already orders it — a child // must NOT dep on its own parent (dep-scope), so it takes no sibling edge. // // ⚠️ This is the one quiesce site that does NOT use `templates::quiesce`. // The helper needs a brace holding the lease above the pair; here `Signal` // *is* the holder, and the nesting under it is what keeps the grant // continuous across the bounce. Giving this chain its own brace would // unify all three sites — at the cost of one extra no-op node on every // graceful restart, which an operator would see. Deliberately not done as // a side effect of a rebuild-shape change. if graceful { let signal = builder .node(NodeKind::Signal { agent: a() }) .needs(Resource::Agent(a())); let drain = builder .node(NodeKind::Drain { agent: a() }) .needs(Resource::Agent(a())) .part_of(signal); let stop = builder .node(NodeKind::StopForUpdate { agent: a() }) .needs(Resource::Agent(a())) .part_of(signal) .after_ok(drain); let _ = builder .node(NodeKind::Reconcile { agent: a() }) .needs(Resource::Agent(a())) .part_of(signal) .after_ok(stop); } else { let stop = builder .node(NodeKind::StopForUpdate { agent: a() }) .needs(Resource::Agent(a())); let _ = builder .node(NodeKind::Reconcile { agent: a() }) .needs(Resource::Agent(a())) .part_of(stop); } } // The `*_nodes` declarers 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 + declare + submit. // // A power op has no tail node: its effect is its nodes (`SetWanted` + // `Reconcile`), with nothing left to do once they settle. // // There is no concatenation step either: every chain declares into the same // builder and each keeps its own root, so the per-agent subgraphs are // independent and run concurrently, each on its own lease. Rebasing one // 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) { for (agent, running) in targets { stop_chain(builder, agent, graceful, *running); } } /// Assemble the start DAG from explicit `(agent, running, stale)` targets. /// /// No DAG-level pill: each agent's dashboard label is derived from the node /// 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)]) { 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) { 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 `agents` in a **single** DAG — one per-agent subgraph each, built /// from live running state and run concurrently on their own leases. A running /// agent gets the stop→reconcile chain (`graceful` prepends signal→drain); a /// down agent gets a lone `Reconcile`. Restart never writes `wanted`, so the /// tail `Reconcile` converges each agent to its EXISTING intent — a /// deliberately-stopped agent stays down. /// /// # Errors /// Propagates a graph-insert error. pub async fn restart_many( coord: &Arc, agents: &[String], graceful: bool, ) -> anyhow::Result> { 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(|b| { restart_nodes(b, &targets, graceful); Vec::new() })?; coord.emit_rebuild_queue_snapshot(); Ok(ids) } /// Start `agents` in a **single** DAG. A down agent gets /// `SetWanted(Up) → Reconcile` (or, rev stale, a rebuild-then-start so it comes /// up on current derivations); an already-running agent gets the same shape /// with the reconcile noop'ing. /// /// # Errors /// Propagates a graph-insert error. pub async fn start_many(coord: &Arc, agents: &[String]) -> anyhow::Result> { let current = crate::auto_update::current_flake_rev(&coord.hyperhive_flake); let mut targets = Vec::with_capacity(agents.len()); for agent in agents { 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)); } let ids = coord.job_queue.insert(|b| { start_nodes(b, &targets); Vec::new() })?; coord.emit_rebuild_queue_snapshot(); Ok(ids) } /// Stop `agents` in a **single** DAG. A running agent gets /// `SetWanted(Off) → [Signal → Drain →](graceful) Reconcile`; a down agent /// skips the pointless quiesce but keeps the `Reconcile` as the race-up /// backstop. /// /// # Errors /// Propagates a graph-insert error. pub async fn stop_many( coord: &Arc, agents: &[String], graceful: bool, ) -> anyhow::Result> { 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(|b| { stop_nodes(b, &targets, graceful); Vec::new() })?; coord.emit_rebuild_queue_snapshot(); Ok(ids) }