hyperhive/hive-c0re/src/job_queue/power.rs

//! 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.
/// 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 {
    // `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);
    }
    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> {
    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`, 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(),
        ]
    } else {
        let _ = builder
            .node(NodeKind::Reconcile {
                agent: agent.to_owned(),
            })
            .needs(Resource::Agent(agent.to_owned()))
            .part_of(wanted);
        vec![wanted.guid()]
    }
}

/// 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.
///
/// Returns the group root's guid — see [`stop_chain`].
fn restart_chain(
    builder: &JobBuilder,
    agent: &str,
    graceful: bool,
    running: bool,
) -> hive_jobq::NodeGuid {
    let a = || agent.to_owned();
    if !running {
        // Nothing to bounce — a lone Reconcile converges to intent, and is
        // itself the root.
        return builder
            .node(NodeKind::Reconcile { agent: a() })
            .needs(Resource::Agent(a()))
            .guid();
    }
    // 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);
        signal.guid()
    } 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);
        stop.guid()
    }
}

// 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,
) -> Vec<hive_jobq::NodeGuid> {
    targets
        .iter()
        .map(|(agent, running)| stop_chain(builder, agent, graceful, *running))
        .collect()
}

/// 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)],
) -> Vec<hive_jobq::NodeGuid> {
    targets
        .iter()
        .flat_map(|(agent, running, stale)| start_chain(builder, agent, *running, *stale))
        .collect()
}

/// 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()
}

// ---- 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<Coordinator>,
    agents: &[String],
    graceful: bool,
) -> anyhow::Result<Vec<NodeId>> {
    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)
}

/// 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<Coordinator>,
    agents: &[String],
) -> anyhow::Result<Vec<NodeId>> {
    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_job(|b| start_nodes(b, &targets))?;
    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<Coordinator>,
    agents: &[String],
    graceful: bool,
) -> anyhow::Result<Vec<NodeId>> {
    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)
}

/// Declare the pause DAG for `agents` — one `PauseSignal → PauseDrain`
/// pair (see [`super::templates::pause_quiesce`]) per agent, independent
/// roots on their own agent lease so a whole-hive pause overlaps rather
/// than serialising.
pub(crate) fn pause_nodes(builder: &JobBuilder, agents: &[String]) -> Vec<hive_jobq::NodeGuid> {
    agents
        .iter()
        .map(|agent| super::templates::pause_quiesce(builder, agent).guid())
        .collect()
}

/// Pause `agents` in a **single** DAG. Unlike `stop`/`start`/`restart`,
/// this needs no live-state read first — `Coordinator::set_paused`
/// works on a stopped container too (the marker is sticky), so there's
/// no `running`/`stale` branch to resolve, and nothing here actually
/// awaits anything (`insert_job` and `emit_rebuild_queue_snapshot` are
/// both sync) — plain `fn`, not `async fn`, unlike its siblings.
///
/// # Errors
/// Propagates a graph-insert error.
pub fn pause_many(coord: &Arc<Coordinator>, agents: &[String]) -> anyhow::Result<Vec<NodeId>> {
    let targets: Vec<String> = agents.to_vec();
    let ids = coord.job_queue.insert_job(|b| pause_nodes(b, &targets))?;
    coord.emit_rebuild_queue_snapshot();
    Ok(ids)
}