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

//! 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
//! (`templates::{node, after_ok, rebuild_nodes}`), and concatenate them into
//! ONE DAG (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::{DagSpec, Dep, NodeKind, NodeSpec, Template};
use super::templates::{after_ok, child, node, rebuild_nodes};
use super::{Source, templates};
use crate::coordinator::{Coordinator, TransientKind};
use crate::lifecycle;

fn submit_and_emit(coord: &Arc<Coordinator>, spec: super::DagSpec) -> u64 {
    let id = coord
        .job_queue
        .submit(spec)
        .expect("template-built dag specs 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<Coordinator>, agent: &str, source: Source, reason: String) -> u64 {
    submit_and_emit(coord, templates::rebuild(agent, source, reason, 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 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> {
    // `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 mut n = vec![node(
        NodeKind::SetWanted {
            agent: a(),
            up: false,
        },
        Vec::new(),
    )];
    if graceful && running {
        n.push(child(0, NodeKind::Signal { agent: a() }, Vec::new()));
        n.push(child(0, NodeKind::Drain { agent: a() }, after_ok(1)));
        n.push(child(0, NodeKind::Reconcile { agent: a() }, after_ok(2)));
    } else {
        n.push(child(0, NodeKind::Reconcile { agent: a() }, Vec::new()));
    }
    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(
        NodeKind::SetWanted {
            agent: agent.to_owned(),
            up: true,
        },
        Vec::new(),
    )];
    if !running && stale {
        // Rebuild subtree after the SetWanted head (base = 1, so the rebuild's
        // `MetaSync` root deps `after_ok(0)` = the head). `MetaSync`,
        // `Prebuild` + `Reconcile` are their own group roots (top-level, per
        // `rebuild_nodes`).
        n.extend(rebuild_nodes(agent, true, 1));
    } else {
        n.push(child(
            0,
            NodeKind::Reconcile {
                agent: agent.to_owned(),
            },
            Vec::new(),
        ));
    }
    n
}

/// 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(agent: &str, graceful: bool, running: bool) -> Vec<NodeSpec> {
    let a = || agent.to_owned();
    if !running {
        // Nothing to bounce — a lone Reconcile converges to intent.
        return vec![node(NodeKind::Reconcile { agent: a() }, Vec::new())];
    }
    // 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.
    let mut n = vec![if graceful {
        node(NodeKind::Signal { agent: a() }, Vec::new())
    } else {
        node(NodeKind::StopForUpdate { agent: a() }, Vec::new())
    }];
    if graceful {
        n.push(child(0, NodeKind::Drain { agent: a() }, Vec::new()));
        n.push(child(
            0,
            NodeKind::StopForUpdate { agent: a() },
            after_ok(1),
        ));
    }
    // `Reconcile` gates on the last mechanical step. When the only step is the
    // root itself (non-graceful, `StopForUpdate` == index 0), the parent gate
    // already orders `Reconcile` after it — a child must NOT dep on its own
    // parent (dep-scope). So the sibling dep is added only for a graceful
    // bounce, where the last step is a sibling child.
    let deps = if n.len() > 1 {
        after_ok(u64::try_from(n.len() - 1).unwrap_or(0))
    } else {
        Vec::new()
    };
    n.push(child(0, NodeKind::Reconcile { agent: a() }, deps));
    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 = u64::try_from(out.len()).unwrap_or(u64::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 {
                kind: spec.kind,
                deps,
                // Rebase the structural parent by the same offset (a subgraph
                // root keeps `parent = None`, so the per-agent groups stay
                // independent + concurrent).
                parent: spec.parent.map(|p| base + p),
            });
        }
    }
    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,
        approval_id: None,
        inputs: Vec::new(),
        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, 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` / `restart-all` is one DAG.
pub async fn restart_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, restart_spec(&targets, graceful, 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
}

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

/// 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 {
    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.
pub fn perm_change(
    coord: &Arc<Coordinator>,
    agent: &str,
    source: Source,
    reason: String,
    payload: super::PermPayload,
) -> u64 {
    submit_and_emit(
        coord,
        templates::perm_change(agent, source, reason, payload),
    )
}

/// Meta-input lock bump; cascade rebuilds fan out on completion.
pub fn meta_update(
    coord: &Arc<Coordinator>,
    inputs: Vec<String>,
    source: Source,
    reason: String,
) -> u64 {
    submit_and_emit(coord, templates::meta_update(inputs, source, reason, 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<Coordinator>,
    moves: Vec<(hive_types::Ident, Option<hive_types::Ident>)>,
    source: Source,
    reason: String,
) -> u64 {
    submit_and_emit(coord, templates::reparent(moves, source, reason))
}