//! Request-level submit API — the surface the dashboard POST handlers, //! 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 //! (`JobBuilder::node` + `templates::rebuild_nodes`), all declaring into ONE job //! (independent per-agent roots, concurrent on their own leases). //! //! 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 super::model::NodeKind; use super::resource::Resource; use super::templates::rebuild_nodes; use super::{JobBuilder, Source, templates}; use crate::coordinator::Coordinator; use crate::lifecycle; fn submit_and_emit( coord: &Arc, source: Source, reason: String, declare: impl FnOnce(&JobBuilder), ) -> u64 { let id = coord .job_queue .submit(source, reason, declare) .expect("template-declared shapes are acyclic"); coord.emit_rebuild_queue_snapshot(); id } /// Manual/approval-independent rebuild (always relocks the agent's /// meta input — the meta-update cascade grows its own rebuild subgraphs /// in-DAG instead of going through this surface). pub fn rebuild(coord: &Arc, agent: &str, source: Source, reason: String) -> u64 { submit_and_emit(coord, source, reason, |builder| { templates::rebuild(builder, agent, true); }) } // ---- dynamic power-op DAG assembly ---------------------------------------- // // The pure per-agent chain builders below take `running` (and `stale`) // 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 { let signal = builder .node(NodeKind::Signal { agent: a() }) .needs(Resource::Agent(a())) .part_of(wanted); let drain = builder .node(NodeKind::Drain { agent: a() }) .needs(Resource::Agent(a())) .part_of(wanted) .after_ok(signal); 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. 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); } } /// Restart a single agent. Thin wrapper over [`restart_many`]. pub async fn restart(coord: &Arc, agent: &str, source: Source, reason: String) -> u64 { restart_many(coord, &[agent.to_owned()], false, source, reason).await } /// Graceful restart of a single agent (signal → drain → stop → reconcile, /// when running). Thin wrapper over [`restart_many`] with `graceful = true`. pub async fn graceful_restart( coord: &Arc, agent: &str, source: Source, reason: String, ) -> u64 { restart_many(coord, &[agent.to_owned()], true, source, reason).await } /// Restart `agents` (one or many) in a **single** DAG — one per-agent /// subgraph each, built dynamically from live running state and run /// concurrently on their own leases. A running agent gets the stop→reconcile /// chain (`graceful` prepends signal→drain); a down agent gets just a lone /// `Reconcile` (nothing to stop). Restart never writes `wanted`, so the /// tail `Reconcile` converges each agent to its EXISTING intent — a /// deliberately-stopped agent stays down. The whole hive-wide /// `hivectl restart` is one DAG. pub async fn restart_many( coord: &Arc, 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, source, reason, |builder| { restart_nodes(builder, &targets, graceful); }) } /// Start a single agent. Thin wrapper over [`start_many`]. pub async fn start(coord: &Arc, agent: &str, source: Source, reason: String) -> u64 { start_many(coord, &[agent.to_owned()], source, reason).await } /// Start `agents` (one or many) in a **single** DAG — one per-agent subgraph /// each, built dynamically from live state and run concurrently on their own /// leases. A down agent gets `SetWanted(Up) → Reconcile` (or, rev stale, a /// rebuild-then-start so it comes up on current derivations); an already- /// running agent gets `SetWanted(Up) → Reconcile` (the reconcile noops). The /// whole hive-wide `hivectl start` is one DAG. pub async fn start_many( coord: &Arc, agents: &[String], source: Source, reason: String, ) -> u64 { 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, source, reason, |builder| { start_nodes(builder, &targets); }) } /// Hard stop a single agent. Thin wrapper over [`stop_many`]. pub async fn stop(coord: &Arc, agent: &str, source: Source, reason: String) -> u64 { stop_many(coord, &[agent.to_owned()], false, source, reason).await } /// Graceful stop of a single agent (signal → drain → reconcile, when /// running). Thin wrapper over [`stop_many`] with `graceful = true`. pub async fn graceful_stop( coord: &Arc, agent: &str, source: Source, reason: String, ) -> u64 { stop_many(coord, &[agent.to_owned()], true, source, reason).await } /// Stop `agents` (one or many) in a **single** DAG — one per-agent subgraph /// each, built dynamically from live state and run concurrently on their own /// leases. A running agent gets `SetWanted(Off) → [Signal → Drain →](graceful) /// Reconcile`; a down agent gets just `SetWanted(Off) → Reconcile` (skips the /// pointless quiesce, keeps the Reconcile as the race-up backstop). The whole /// hive-wide `hivectl stop` is one DAG. pub async fn stop_many( coord: &Arc, 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, source, reason, |builder| { stop_nodes(builder, &targets, graceful); }) } /// Perm change: commit the JSON file(s) then rebuild. pub fn perm_change( coord: &Arc, agent: &str, source: Source, reason: String, payload: super::PermPayload, ) -> u64 { submit_and_emit(coord, source, reason, |builder| { templates::perm_change(builder, agent, payload); }) } /// Meta-input lock bump; cascade rebuilds fan out on completion. pub fn meta_update( coord: &Arc, inputs: Vec, source: Source, reason: String, ) -> u64 { submit_and_emit(coord, source, reason, |builder| { templates::meta_update(builder, inputs, None); }) } /// Topology move(s) as a queue DAG. `moves` is `(child, new_parent)` pairs — /// one entry for `set-parent`, N for `set-parent-bulk`. Fire-and-forget like /// everything else in this module: submits and returns a DAG id /// immediately, the caller learns the outcome async (dashboard job view / /// `hivectl`'s `QueueDag` poll). Wired from `server.rs`'s `HostRequest:: /// SetParent` (hivectl) and `dashboard/topology.rs`'s `set-parent`/ /// `set-parent-bulk` handlers. pub fn reparent( coord: &Arc, moves: Vec<(hive_types::Ident, Option)>, source: Source, reason: String, ) -> u64 { submit_and_emit(coord, source, reason, |builder| { templates::reparent(builder, moves); }) }