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

//! Node executors — one async fn per [`NodeKind`], each a thin wrapper
//! over existing `lifecycle.rs` / `meta_rev.rs` / `actions.rs` code. Node
//! executors keep their own internal error handling where it exists
//! today (cold-start fallback inside `Reconcile`, warn-only forge sync in
//! the `Swap` tail); DAG-level failure handling is cancel-downstream in
//! the queue.

use std::path::PathBuf;
use std::sync::{Arc, Mutex, OnceLock};

use anyhow::{Context as _, Result};
use swarm_wire::agent_closures::AgentClosure;

use hive_jobq::NodeId;

use super::model::NodeKind;
use crate::coordinator::{AgentPaths, Coordinator, HiveEnv};
use crate::meta_rev::MetaBuild;
use crate::power::{ReconcileAction, reconcile_action};
use crate::stats::sweep_health::{self, SweepHealth};

/// Max time `Drain` waits for the harness to run its stop-checkpoint
/// turn before falling back to the hard stop. Generous — a checkpoint
/// turn can take a while — but bounded so a wedged agent never blocks
/// the stop indefinitely.
pub const GRACEFUL_STOP_TIMEOUT: std::time::Duration = std::time::Duration::from_mins(3);

/// Max time `PauseDrain` waits for the harness to report
/// `PauseAcknowledged` before giving up and resolving anyway (the
/// marker itself — not this node — is what actually gates the turn
/// loop, so "giving up" costs nothing but a slightly-late dashboard
/// badge). Same ceiling as `GRACEFUL_STOP_TIMEOUT` — no reason for the
/// two to diverge yet, but aliased under its own name so a future
/// change to one doesn't silently retune the other.
const PAUSE_ACK_TIMEOUT: std::time::Duration = GRACEFUL_STOP_TIMEOUT;

/// The container-state calls the executors below make on the host: systemd,
/// `hive-priv`, the swarm's closure record, the `internal/meta` record. Production uses
/// [`RealHost`], which forwards to `lifecycle` / `meta_rev` / `Coordinator`
/// unchanged; tests substitute a recording fake.
pub(super) trait Host: Sync {
    /// Whether the agent's container unit is active.
    async fn is_running(&self, name: &str) -> bool;
    /// Stop the agent's container.
    async fn kill(&self, name: &str) -> Result<()>;
    /// The `internal/meta` commit `MetaSync` recorded for the agent.
    fn recorded_rev(&self, name: &str) -> Result<String>;
    /// The record the swarm's builder wrote for the agent; `None` when it wrote
    /// none.
    async fn closure_record(&self, name: &str) -> Result<Option<AgentClosure>>;
    /// The pins `internal/meta` at the commit `rev` holds for the agent.
    async fn pins_at(&self, name: &str, rev: &str) -> Result<MetaBuild>;
    /// Download `store_path` from the swarm's cache into the store, never
    /// building it.
    async fn realise_closure(&self, name: &str, store_path: &str) -> Result<()>;
    /// Re-apply drop-ins, then point the container at the closure `toplevel`.
    async fn swap_update(
        &self,
        name: &str,
        hive: &HiveEnv,
        paths: &AgentPaths,
        toplevel: &str,
        node_id: u64,
    ) -> Result<()>;
    /// Write or remove the agent's pause marker.
    async fn set_paused(&self, agent: &hive_types::Ident, paused: bool) -> Result<()>;
    /// Publish a lifecycle notice for the hive under the `core` subsystem.
    async fn notify(&self, key: String, summary: String);
    /// Re-read the agent's archetype from the swarm.
    async fn refresh_archetype(&self, name: &str) -> Result<()>;
    /// Rebuild preamble: port check, applied repo and state dirs.
    async fn prepare_rebuild_dirs(&self, name: &str, paths: &AgentPaths) -> Result<()>;
    /// Rewrite the host-side files that follow the agent set.
    async fn refresh_agent_set(&self) -> Result<()>;
    /// Pin the `internal/meta` commit the agent builds from.
    async fn resolve_meta(&self, name: &str) -> Result<MetaBuild>;
    /// Record `build` as the commit `Prebuild` and `Swap` expect the swarm's
    /// closure to be built from.
    fn record_meta(&self, name: &str, build: &MetaBuild) -> Result<()>;
    /// The build `MetaSync` recorded, if any.
    fn recorded_build(&self, name: &str) -> Option<MetaBuild>;
    /// Mark the agent as running the pins of `build`.
    fn mark_deployed(&self, name: &str, build: &MetaBuild) -> Result<()>;
    /// First-spawn provisioning: repos, state subvolume, agent-set refresh.
    async fn provision(&self, name: &str, paths: &AgentPaths) -> Result<()>;
    /// `nixos-container create` from the closure `toplevel`.
    async fn create_only(&self, name: &str, toplevel: &str) -> Result<()>;
    /// Write the nspawn and resource-limit drop-ins, then reload systemd.
    async fn write_dropins(&self, name: &str, hive: &HiveEnv, paths: &AgentPaths) -> Result<()>;
    /// Start preamble, then start with the cold-start fallback.
    async fn start(&self, name: &str, hive: &HiveEnv, paths: &AgentPaths) -> Result<()>;
    /// Bind the agent's MCP listener.
    fn register_agent(&self, coord: &Arc<Coordinator>, name: &str) -> Result<()>;
    /// `nixos-container destroy`.
    async fn destroy(&self, name: &str) -> Result<()>;
    /// The agent's ephemeral runtime dir.
    fn runtime_dir(&self, name: &str) -> PathBuf;
    /// Delete the agent's state subvolume through `hive-priv`.
    async fn delete_state_subvolume(&self, name: &str) -> Result<()>;
    /// The persistent directories a purge removes.
    fn purge_dirs(&self, name: &str) -> Vec<PathBuf>;
    /// Publish the hive's forge credential for nix.
    async fn forge_ensure_all(&self);
    /// One `/knowledge` pull.
    async fn knowledge_pull(&self, coord: &Arc<Coordinator>) -> Result<()>;
    /// One swarm wanted-state pull.
    async fn wanted_pull(&self, coord: &Arc<Coordinator>) -> Result<()>;
}

/// [`Host`] over the real system.
pub(super) struct RealHost;

impl Host for RealHost {
    async fn is_running(&self, name: &str) -> bool {
        crate::lifecycle::is_running(name).await
    }

    async fn kill(&self, name: &str) -> Result<()> {
        crate::lifecycle::kill(name).await
    }

    fn recorded_rev(&self, name: &str) -> Result<String> {
        Ok(crate::meta_rev::recorded_or_err(name)?.rev)
    }

    async fn closure_record(&self, name: &str) -> Result<Option<AgentClosure>> {
        crate::attic::closure::record(name).await
    }

    async fn pins_at(&self, name: &str, rev: &str) -> Result<MetaBuild> {
        crate::meta_rev::resolve_at(name, rev).await
    }

    async fn realise_closure(&self, name: &str, store_path: &str) -> Result<()> {
        crate::priv_client::realise_closure(name, store_path).await
    }

    async fn swap_update(
        &self,
        name: &str,
        hive: &HiveEnv,
        paths: &AgentPaths,
        toplevel: &str,
        node_id: u64,
    ) -> Result<()> {
        crate::lifecycle::swap_update(name, hive, paths, toplevel, Some(node_id)).await
    }

    async fn set_paused(&self, agent: &hive_types::Ident, paused: bool) -> Result<()> {
        Coordinator::set_paused(agent, paused).await
    }

    async fn notify(&self, key: String, summary: String) {
        crate::swarm_notices::notify("core", Some(key), summary, None).await;
    }

    async fn refresh_archetype(&self, name: &str) -> Result<()> {
        crate::archetypes::refresh(name).await
    }

    async fn prepare_rebuild_dirs(&self, name: &str, paths: &AgentPaths) -> Result<()> {
        crate::lifecycle::prepare_rebuild_dirs(name, paths).await
    }

    async fn refresh_agent_set(&self) -> Result<()> {
        let agents = crate::lifecycle::agents_for_meta_listing().await?;
        crate::agent_set::refresh(&agents).await
    }

    async fn resolve_meta(&self, name: &str) -> Result<MetaBuild> {
        crate::meta_rev::resolve(name).await
    }

    fn record_meta(&self, name: &str, build: &MetaBuild) -> Result<()> {
        crate::meta_rev::record(name, build)
    }

    fn recorded_build(&self, name: &str) -> Option<MetaBuild> {
        crate::meta_rev::recorded(name)
    }

    fn mark_deployed(&self, name: &str, build: &MetaBuild) -> Result<()> {
        crate::meta_rev::mark_deployed(name, build)
    }

    async fn provision(&self, name: &str, paths: &AgentPaths) -> Result<()> {
        crate::lifecycle::provision_container(name, paths).await
    }

    async fn create_only(&self, name: &str, toplevel: &str) -> Result<()> {
        crate::lifecycle::create_only(name, toplevel).await
    }

    async fn write_dropins(&self, name: &str, hive: &HiveEnv, paths: &AgentPaths) -> Result<()> {
        crate::lifecycle::write_dropins(name, hive, paths).await
    }

    async fn start(&self, name: &str, hive: &HiveEnv, paths: &AgentPaths) -> Result<()> {
        let token = crate::lifecycle::converge_start_preamble(name, hive, paths).await?;
        crate::lifecycle::start_with_fallback(token).await
    }

    fn register_agent(&self, coord: &Arc<Coordinator>, name: &str) -> Result<()> {
        coord.register_agent(name).map(drop)
    }

    async fn destroy(&self, name: &str) -> Result<()> {
        crate::lifecycle::destroy(name).await
    }

    fn runtime_dir(&self, name: &str) -> PathBuf {
        crate::paths::agent_runtime_dir(name)
    }

    async fn delete_state_subvolume(&self, name: &str) -> Result<()> {
        crate::priv_client::delete_agent_subvolume(name).await
    }

    fn purge_dirs(&self, name: &str) -> Vec<PathBuf> {
        // A malformed name can't have a persistent state tree (the state dir is
        // only ever created under a validated Ident), so only the applied dir
        // is left to clear.
        hive_types::Ident::parse(name)
            .ok()
            .map(|id| crate::paths::agent_state_dir(&id))
            .into_iter()
            .chain([crate::paths::applied_dir(name)])
            .collect()
    }

    async fn forge_ensure_all(&self) {
        crate::forge::ensure_all().await;
    }

    async fn knowledge_pull(&self, coord: &Arc<Coordinator>) -> Result<()> {
        crate::workers::knowledge::pull(coord).await
    }

    async fn wanted_pull(&self, coord: &Arc<Coordinator>) -> Result<()> {
        crate::workers::wanted::pull(coord).await
    }
}

/// Run one claimed node to completion. Called from a task the
/// scheduler spawns per claim; the `Result` (stringified) becomes the
/// node's terminal state.
///
/// `builder` is the node's own growth channel: an executor that decides more work
/// is needed declares it here, and the scheduler inserts it under this node
/// when the node completes. Most executors never touch it. Nothing is inserted
/// while the node runs — the builder is local state, so this stays outside the
/// queue's lock for the whole (often multi-minute) execution.
///
/// ⚠️ Taken **by value and handed back**, not by reference. A `JobBuilder` is
/// `RefCell`-backed: owned it is `Send`, but `&JobBuilder` is not (a shared ref
/// is `Send` only if the referent is `Sync`, and `RefCell` never is). A `&JobBuilder`
/// parameter would be live across every `.await` in this fn and make the whole
/// future non-`Send`, which the scheduler's `tokio::spawn` rejects. So the
/// growth executors below return *what to grow* and the declaration happens
/// here, synchronously, between awaits.
///
/// The node is identified by its own id + payload rather than by a `Claim`
/// side-struct: `kind` already carries the agent, and the DAG id is a
/// derived read (`JobQueue::dag_of`) the three arms that need it take
/// themselves. Nothing here needs a claim to exist as a type.
pub(super) async fn run_node(
    coord: &Arc<Coordinator>,
    builder: super::JobBuilder,
    id: NodeId,
    kind: &NodeKind,
) -> (super::JobBuilder, Result<()>) {
    dispatch(&RealHost, coord, builder, id, kind).await
}

/// [`run_node`] over any [`Host`].
async fn dispatch(
    host: &impl Host,
    coord: &Arc<Coordinator>,
    builder: super::JobBuilder,
    id: NodeId,
    kind: &NodeKind,
) -> (super::JobBuilder, Result<()>) {
    // The agent this node targets rides the payload — empty for the agentless
    // kinds (the sweeps), which never read it.
    let agent = kind.agent();
    // Every arm is `Result<()>`; the one that grows work declares into `builder`
    // *synchronously*, after its own awaits have finished. Borrowing `&builder`
    // inside an `.await` would make this future non-`Send` (see above), so the
    // growth executors return what to grow rather than taking the builder.
    let result = match kind {
        NodeKind::MetaSync { .. } => run_meta_sync(host, agent).await,
        NodeKind::Prebuild { .. } => run_prebuild(host, agent).await,
        NodeKind::Swap { .. } => run_swap(host, coord, agent, id).await,
        NodeKind::RebuildBookkeeping { .. } => run_rebuild_bookkeeping(host, coord, agent).await,
        NodeKind::Provision { .. } => run_provision(host, agent).await,
        NodeKind::Create { .. } => run_create(host, agent).await,
        NodeKind::DestroyContainer { .. } => run_destroy_container(host, coord, agent).await,
        NodeKind::PurgeState { .. } => {
            run_purge_state(host, agent).await;
            Ok(())
        }
        NodeKind::DestroyBookkeeping { .. } => {
            run_destroy_bookkeeping(host, coord, agent).await;
            Ok(())
        }

        NodeKind::Reconcile { .. } => run_reconcile(host, coord, agent).await.map(|sub| {
            if let Some(kind) = sub {
                super::templates::fanned_out_mechanical(&builder, kind);
            }
        }),
        NodeKind::Start { .. } => run_start(host, coord, agent).await,
        NodeKind::Stop { .. } => run_stop(host, coord, agent).await,
        NodeKind::StopForUpdate { .. } => run_stop_for_update(host, coord, agent).await,
        NodeKind::Signal { .. } => {
            run_signal(coord, agent);
            Ok(())
        }
        NodeKind::Drain { .. } => run_drain(coord, agent, GRACEFUL_STOP_TIMEOUT).await,
        NodeKind::PauseSignal { .. } => run_pause_signal(host, coord, agent).await,
        NodeKind::PauseDrain { .. } => run_pause_drain(coord, agent, PAUSE_ACK_TIMEOUT).await,
        NodeKind::WriteDropin { .. } => run_write_dropin(host, coord, agent).await,
        // The payload rides the node and is destructured here, so the executor
        // takes it directly instead of re-matching the kind behind a `bail!`
        // that could never fire.
        NodeKind::EmitRebuilt { ok, .. } => {
            run_emit_rebuilt(host, coord, agent, coord.job_queue.root_of(id), *ok).await;
            Ok(())
        }
        NodeKind::SetWanted { up, .. } => run_set_wanted(coord, agent, *up),
        // Braces carry no work of their own; completing one lets it reach
        // `Finishing` so the nodes under it start. What they declare stays held
        // until their whole subtree settles.
        NodeKind::AgentWindow { .. } => Ok(()),
        NodeKind::ForgeSweep => run_forge_sweep(host).await,
        NodeKind::KnowledgePull => run_knowledge_pull(host, coord, KNOWLEDGE_PULL_DEADLINE).await,
        NodeKind::WantedPull => run_wanted_pull(host, coord).await,
    };
    (builder, result)
}

/// Boot-time publish of the hive's forge credential for nix, as a DAG node —
/// see [`NodeKind::ForgeSweep`]. `forge::ensure_all` logs its own outcome and
/// is best-effort (a build re-publishes the credential before it fetches), so
/// this wrapper has nothing left to report; it exists to make the step a
/// visible unit of work.
async fn run_forge_sweep(host: &impl Host) -> Result<()> {
    host.forge_ensure_all().await;
    Ok(())
}

/// `/knowledge` pull as a DAG node — see [`NodeKind::KnowledgePull`]. Every
/// pass reports here, whoever submitted it: as the node's own error on the
/// dashboard, and into the debounced warning.
async fn run_knowledge_pull(
    host: &impl Host,
    coord: &Arc<Coordinator>,
    deadline: std::time::Duration,
) -> Result<()> {
    let result = tokio::time::timeout(deadline, host.knowledge_pull(coord))
        .await
        .unwrap_or_else(|_| {
            Err(anyhow::anyhow!(
                "knowledge pull timed out after {deadline:?}"
            ))
        });
    let mut health = knowledge_pull_health()
        .lock()
        .unwrap_or_else(std::sync::PoisonError::into_inner);
    match &result {
        Ok(()) => health.record_ok(),
        Err(e) => {
            let err = format!("{e:#}");
            health.record_err(|ctx| {
                let age = ctx.since_last_ok.map_or_else(
                    || "no success this session".to_owned(),
                    |d| format!("last ok {} ago", sweep_health::fmt_age(d)),
                );
                format!(
                    "knowledge repo pull failing ({} consecutive, {age}) \
                     — /knowledge is stale until it recovers: {err}",
                    ctx.consecutive
                )
            });
        }
    }
    result
}

/// A pull still running holds [`Resource::KnowledgeTree`] against every later
/// one. The `reset` / `clean` / `pull` git children are `kill_on_drop`, so the
/// deadline kills them.
///
/// [`Resource::KnowledgeTree`]: super::resource::Resource::KnowledgeTree
const KNOWLEDGE_PULL_DEADLINE: std::time::Duration = std::time::Duration::from_mins(5);

/// Warning for a `/knowledge` pull that keeps failing: three misses in a row,
/// cleared by the next successful pull.
fn knowledge_pull_health() -> &'static Mutex<SweepHealth> {
    static HEALTH: OnceLock<Mutex<SweepHealth>> = OnceLock::new();
    HEALTH.get_or_init(|| Mutex::new(SweepHealth::new("knowledge_pull", "warn", 3)))
}

/// Boot-time swarm wanted-state pull as a DAG node — see
/// [`NodeKind::WantedPull`]. "The controller has declared nothing" is a
/// successful outcome the worker logs, so an error reaching here means the
/// read itself failed and this hive is running blind to its declaration —
/// which is exactly the state worth seeing on the dashboard.
async fn run_wanted_pull(host: &impl Host, coord: &Arc<Coordinator>) -> Result<()> {
    host.wanted_pull(coord).await
}

/// `nixos-container destroy`, then drop the agent from the roster and clear
/// its ephemeral runtime dir (the mcp socket, which does not survive a restart
/// anyway).
///
/// The only fallible step is the destroy itself: once the container is gone the
/// un-registration cannot meaningfully fail, and returning early would strand
/// the roster claiming an agent that no longer exists. A destroy that leaves the
/// container in place fails this node before un-registration, and its failure
/// is what keeps the `after_ok` purge and bookkeeping from running.
async fn run_destroy_container(
    host: &impl Host,
    coord: &Arc<Coordinator>,
    agent: &str,
) -> Result<()> {
    host.destroy(agent).await?;
    coord.unregister_agent(agent);
    let runtime = host.runtime_dir(agent);
    if runtime.exists() {
        let _ = std::fs::remove_dir_all(&runtime);
    }
    Ok(())
}

/// The `purge = true` half: wipe the agent's persistent trees.
///
/// Every step is best-effort-with-a-warning rather than fatal, and that is
/// deliberate — the container is already destroyed by the time this runs, so
/// failing the node would leave the operator with a half-purged agent and a red
/// DAG, when what they can actually act on is the log line naming the path.
async fn run_purge_state(host: &impl Host, agent: &str) {
    // The state root may be a btrfs subvolume: a subvolume root can't be
    // removed with rmdir/`remove_dir_all`, so delete it via hive-priv (root)
    // first. No-op for plain-dir agents — the loop below then handles the
    // plain-dir state root plus the applied dir.
    if let Err(e) = host.delete_state_subvolume(agent).await {
        tracing::warn!(error = ?e, %agent, "purge: delete state subvolume failed");
    }
    for dir in host.purge_dirs(agent) {
        if dir.exists()
            && let Err(e) = std::fs::remove_dir_all(&dir)
        {
            tracing::warn!(error = ?e, dir = %dir.display(), "purge: remove failed");
        }
    }
}

/// Post-destroy bookkeeping. Infallible by construction: every step is
/// warn-and-continue, because the destroy it follows has already succeeded and
/// none of this is undoable — a failed agent-set refresh or power-store write is a
/// bookkeeping drift to log, not a reason to red a DAG whose container is
/// already gone.
async fn run_destroy_bookkeeping(host: &impl Host, coord: &Arc<Coordinator>, agent: &str) {
    // Drop the agent from the topology and the gateway files.
    if let Err(e) = host.refresh_agent_set().await {
        tracing::warn!(error = ?e, %agent, "agent_set refresh after destroy failed");
    }
    // Drop the durable power intent — a future agent of the same name seeds
    // fresh from its observed state.
    if let Err(e) = coord.power.remove(agent) {
        tracing::warn!(%agent, error = ?e, "agent_power: remove on destroy failed");
    }
    host.notify(
        format!("destroyed:{agent}"),
        format!("agent '{agent}' destroyed"),
    )
    .await;
    // Container row disappeared — rescan so the dashboard fires
    // `ContainerRemoved` for the gone row, then emit the tombstones snapshot
    // (gained one on destroy, lost one on purge — recompute either way).
    coord.rescan_containers_and_emit().await;
    crate::dashboard::emit_tombstones_snapshot(coord).await;
}

/// Emit this agent's rebuild-complete todo. `ok` is not computed — it is which
/// of the tail pair the graph let run. The failure note comes from the DAG's
/// first failing node, since the branch knows *that* it failed but not *why*.
async fn run_emit_rebuilt(
    host: &impl Host,
    coord: &Arc<Coordinator>,
    agent: &str,
    dag_id: Option<u64>,
    ok: bool,
) {
    let note = (!ok)
        .then(|| dag_id.and_then(|dag| coord.job_queue.first_error(dag)))
        .flatten();
    let summary = crate::coordinator::rebuilt_todo_summary(agent, ok, note.as_deref(), None, None);
    host.notify(format!("rebuilt:{agent}"), summary).await;
}

/// Write the agent's durable power intent — the DAG-node form of the old
/// pre-submit `set_wanted` side effect. Store-only (no container touch), so
/// build-slot-exempt; but it declares the agent's lifecycle lease
/// (`Resource::Agent`) so the whole power-op DAG is atomic per-agent.
/// The downstream `Reconcile` reads the intent this writes. Unlike the old
/// warn-and-continue write, a failed write fails the node (cancel-downstream
/// cancels the `Reconcile`) rather than letting it converge to a stale
/// intent — that atomicity is the point of moving it into the DAG.
fn run_set_wanted(coord: &Arc<Coordinator>, agent: &str, up: bool) -> Result<()> {
    let wanted = if up {
        crate::power::Wanted::Up
    } else {
        crate::power::Wanted::Offline
    };
    coord
        .power
        .set(agent, wanted)
        .with_context(|| format!("set wanted={} for agent {agent}", <&str>::from(wanted)))?;
    Ok(())
}

/// The rebuild's preamble: runtime-dir prep, the agent-set refresh, and the
/// `internal/meta` commit `Prebuild` and `Swap` expect the swarm's closure to
/// be built from.
///
/// Fails when `internal/meta` does not pin the agent's current config yet:
/// there is no commit to deploy.
async fn run_meta_sync(host: &impl Host, name: &str) -> Result<()> {
    // Runs while the agent is still up — the runtime dir and MCP listener
    // already exist. Use the pure path accessor; no need to re-register the
    // listener (event-driven: registered at start/create).
    let agent_dir = crate::paths::agent_runtime_dir(name);
    let paths = Coordinator::agent_paths(name, agent_dir)?;
    // A failed read keeps the archetype last stored, so the agent is built as
    // it was.
    if let Err(e) = host.refresh_archetype(name).await {
        tracing::warn!(%name, error = ?e, "rebuild: reading the swarm's archetype failed; keeping the stored one");
    }
    host.prepare_rebuild_dirs(name, &paths).await?;
    host.refresh_agent_set().await?;
    // Resolved once here so `Prebuild` and `Swap` expect the same commit.
    let build = host.resolve_meta(name).await?;
    host.record_meta(name, &build)?;
    Ok(())
}

/// Download the swarm's closure for the agent while the container keeps
/// serving, so a rebuild with no closure fails here, before `StopForUpdate`
/// takes a healthy container down. `Swap` downloads the same closure again
/// (a no-op once it is in the store) and applies it. Skipped when the container
/// is already down: there is no uptime to protect, and `Swap` fails the same way.
async fn run_prebuild(host: &impl Host, name: &str) -> Result<()> {
    if host.is_running(name).await {
        let rev = host.recorded_rev(name)?;
        let closure = closure_for(host, name, &rev, "prebuild").await?;
        tracing::info!(%name, %rev, "prebuild: substituted {}", closure.path);
    }
    Ok(())
}

/// Why the swarm's closure for an agent cannot be deployed.
enum NoClosure {
    /// The builder has recorded no build of the agent.
    NoRecord,
    /// A build is recorded, but its closure cannot be had: the record could not
    /// be read, the builder did not push the closure, or the download failed.
    Unpullable(String),
}

/// The closure the swarm built for an agent, in the store.
struct Closure {
    path: String,
    /// The `internal/meta` commit the builder built it from.
    meta_rev: String,
}

/// The closure the swarm built for `name`, downloaded into the store. The
/// deploy applies the swarm's closure whatever commit it was built from: when
/// that is not the `rev` this hive chose, it is logged, and the caller records
/// the pins of the commit actually deployed ([`deployed_pins`]).
///
/// # Errors
/// [`without_closure`]'s refusal, after logging it.
async fn closure_for(host: &impl Host, name: &str, rev: &str, step: &str) -> Result<Closure> {
    pull_closure(host, name, rev, step)
        .await
        .map_err(|why| without_closure(name, rev, step, &why))
}

async fn pull_closure(
    host: &impl Host,
    name: &str,
    rev: &str,
    step: &str,
) -> Result<Closure, NoClosure> {
    let record = host
        .closure_record(name)
        .await
        .map_err(|e| NoClosure::Unpullable(format!("{e:#}")))?
        .ok_or(NoClosure::NoRecord)?;
    if !record.cached {
        return Err(NoClosure::Unpullable(
            "the builder did not push the closure".to_owned(),
        ));
    }
    host.realise_closure(name, &record.store_path)
        .await
        .map_err(|e| NoClosure::Unpullable(format!("{e:#}")))?;
    if record.meta_rev != rev {
        tracing::warn!(
            %name, %rev, built_from = %record.meta_rev,
            "{step}: the swarm built {} from internal/meta {}, not the {rev} this hive chose; deploying the swarm's",
            record.store_path, record.meta_rev
        );
    }
    Ok(Closure {
        path: record.store_path,
        meta_rev: record.meta_rev,
    })
}

/// The pins to record for a deploy of `closure` when this hive chose the commit
/// `rev`: those of the commit the closure was built from, so the markers say
/// what runs. `None` when they are the same commit.
///
/// # Errors
/// The pins of the closure's commit could not be read. The caller fails before
/// applying anything.
async fn deployed_pins(
    host: &impl Host,
    name: &str,
    closure: &Closure,
    rev: &str,
) -> Result<Option<MetaBuild>> {
    if closure.meta_rev == rev {
        return Ok(None);
    }
    let pins = host
        .pins_at(name, &closure.meta_rev)
        .await
        .with_context(|| format!("{name}: reading the pins of the commit the swarm built from"))?;
    Ok(Some(pins))
}

/// What a rebuild or create does when there is no closure to deploy: it
/// fails, and nothing changes, so a running container keeps running as it is
/// and a create makes none. The failure is logged at error level with the
/// agent's name.
///
/// This is the one place that decides what an unpullable closure (the cache
/// unreachable, a TLS error, a failed download) costs. A rebuild never builds
/// locally on the hive.
fn without_closure(name: &str, rev: &str, step: &str, why: &NoClosure) -> anyhow::Error {
    let reason = match why {
        NoClosure::NoRecord => "the swarm has recorded no build of this agent".to_owned(),
        NoClosure::Unpullable(e) => format!("the swarm's closure cannot be pulled: {e}"),
    };
    tracing::error!(%name, %rev, "{step}: {reason}; nothing was changed");
    anyhow::anyhow!("{name}: {reason}")
}

/// Profile-swap: re-apply drop-ins (rebuild is the reconcile verb), point the
/// container at the swarm's closure, then the post-rebuild bookkeeping tail
/// (rev marker, `Rebuilt` event, forge sync, kick, rescan).
/// The recovery-start on failure is NOT here — the DAG's tail
/// `Reconcile` runs after this node terminal ok *or* fail.
async fn run_swap(
    host: &impl Host,
    coord: &Arc<Coordinator>,
    name: &str,
    id: NodeId,
) -> Result<()> {
    // Swap runs on an already-existing (stopped) container — runtime dir
    // and listener were created earlier. Pure path accessor suffices.
    let agent_dir = crate::paths::agent_runtime_dir(name);
    let hive = coord.hive_env();
    let paths = Coordinator::agent_paths(name, agent_dir)?;
    let rev = host.recorded_rev(name)?;
    let closure = closure_for(host, name, &rev, "swap").await?;
    let deployed = deployed_pins(host, name, &closure, &rev).await?;
    tracing::info!(%name, %rev, "swap: applying {}", closure.path);
    let result = host
        .swap_update(name, &hive, &paths, &closure.path, id.get())
        .await;
    // The bookkeeping node marks what the record holds, so it holds what was
    // applied.
    if result.is_ok()
        && let Some(deployed) = &deployed
    {
        host.record_meta(name, deployed)?;
    }
    // On success the Ok-only bookkeeping tail (rev marker, forge
    // sync, kick, rescan, snapshot) runs in the sibling `RebuildBookkeeping` node,
    // which deps `AfterOk(Swap)`. On failure `RebuildBookkeeping` is cancel-cascaded
    // and the tail `Reconcile` (`AfterAny(RebuildBookkeeping)`) handles recovery; here
    // we only refresh the observed state so dashboards reflect the failed
    // swap immediately. The `Rebuilt { ok: false }` manager event is emitted by
    // the DAG's `EmitRebuilt` tail (any node may be the one that failed).
    if result.is_err() {
        coord.rescan_containers_and_emit().await;
    }
    result
}

/// The post-`Swap` bookkeeping tail, split into its own node for dashboard
/// visibility + retry granularity. Deps `AfterOk(Swap)`, so reaching here
/// means the profile swap succeeded. Store/forge work only — no nix
/// build (build-slot-exempt); the agent lease taken at `Swap` is still held
/// (the whole chain up to `Reconcile` is one agent's subgraph).
async fn run_rebuild_bookkeeping(
    host: &impl Host,
    coord: &Arc<Coordinator>,
    name: &str,
) -> Result<()> {
    if let Some(build) = host.recorded_build(name)
        && let Err(e) = host.mark_deployed(name, &build)
    {
        tracing::warn!(%name, error = ?e, "write rev markers failed");
    }
    // The `Rebuilt` manager event is emitted exactly once per agent by the DAG's
    // `EmitRebuilt` tail — emitting ok here and letting a failed tail `Reconcile`
    // add a contradictory !ok would double-report the same rebuild.

    // No `kick_agent` here any more — reporting from `hive-agent`'s own
    // boot instead avoids the duplicate-restart-message problem this used
    // to cause: the container is still stopped at this point (`swap_update` requires it stopped and
    // never restarts it — the DAG's tail `Reconcile`/`Start` does), so a
    // host-side "you were rebuilt" broker message would just queue
    // alongside `run_start`'s own kick for the same boot. `hive-agent` now
    // self-announces once on its own startup instead — one message per
    // actual boot, not one per DAG node that happens to run during it.
    // Rescan so dashboards drop the "needs update" chip.
    coord.rescan_containers_and_emit().await;
    Ok(())
}

/// First-spawn pre-create provisioning: proposed/applied repos, state
/// subvolume, and the agent-set refresh.
async fn run_provision(host: &impl Host, name: &str) -> Result<()> {
    let agent_dir = crate::paths::agent_runtime_dir(name);
    let paths = Coordinator::agent_paths(name, agent_dir)?;
    host.provision(name, &paths).await?;
    Ok(())
}

/// `nixos-container create` proper — the upstream `Provision` node
/// already registered the agent, so this only reads the store. Runtime
/// dir creation and MCP listener registration are deferred to the tail
/// `Reconcile` (`converge_start_preamble` + `register_agent`) so this
/// node stays purely "create", not "create + start". Marks the agent as
/// running the pins it was created from.
///
/// The container is created from the swarm's closure for the resolved commit;
/// with none to deploy the create fails and makes no container.
async fn run_create(host: &impl Host, name: &str) -> Result<()> {
    let build = host.resolve_meta(name).await?;
    let closure = closure_for(host, name, &build.rev, "create").await?;
    let build = deployed_pins(host, name, &closure, &build.rev)
        .await?
        .unwrap_or(build);
    tracing::info!(%name, rev = %build.rev, "create: substituted {}", closure.path);
    host.create_only(name, &closure.path).await?;
    host.mark_deployed(name, &build)?;
    Ok(())
}

/// Idempotent power-converge *planner*: compare `wanted` (durable
/// intent) against observed state and, when they diverge, fan the
/// mechanical `Start` / `Stop` out as a first-class node appended to
/// *this* DAG (a single node declared into `builder`, rooted on this node).
/// Does no container work itself — the sub-step becomes visible in the
/// DAG and the lease-window transient (or the sub-step's own node-local
/// guard) rides across it.
/// Returns the mechanical node to fan out (`None` on a noop) rather than
/// declaring it — the declaration has to happen outside any `.await`, see
/// [`run_node`]. `NodeKind` carries the agent it targets, so this node's agent
/// is stamped into the fanned-out kind here.
async fn run_reconcile(
    host: &impl Host,
    coord: &Arc<Coordinator>,
    name: &str,
) -> Result<Option<NodeKind>> {
    let running = host.is_running(name).await;
    let wanted = coord.power.get_or_seed(name, running)?;
    Ok(match reconcile_action(wanted, running) {
        ReconcileAction::Start => Some(NodeKind::Start {
            agent: name.to_owned(),
        }),
        ReconcileAction::Stop => Some(NodeKind::Stop {
            agent: name.to_owned(),
        }),
        ReconcileAction::Noop => {
            tracing::debug!(%name, wanted = <&str>::from(wanted), running, "reconcile: noop");
            None
        }
    })
}

/// Mechanical container start — the sub-step a `Reconcile` planner fans
/// out when it observes `wanted = Up` and the container down.
async fn run_start(host: &impl Host, coord: &Arc<Coordinator>, name: &str) -> Result<()> {
    // No node-local transient guard: the pill is derived from the running node
    // set, and `Start` reports `Starting` via `NodeKind::transient_kind`. This
    // used to take one "only when the DAG holds none", which was a second
    // derivation covering the gap left by a DAG-level declaration that couldn't
    // describe a sub-step.
    // Run the typed start preamble: ensures the runtime dir exists and
    // writes the nspawn/resource-limits drop-ins. The returned
    // StartableAgent token is the only way to call start_with_fallback —
    // omitting this becomes a compile error.
    let agent_dir = crate::paths::agent_runtime_dir(name);
    let hive = coord.hive_env();
    let paths = Coordinator::agent_paths(name, agent_dir)?;
    host.start(name, &hive, &paths).await?;
    // Bind the MCP listener immediately after starting the container.
    // The preamble created the runtime dir; the container is now coming
    // up and will connect to this socket on its first turn. Event-driven
    // (no background poll) — c0re owns the listener lifecycle, so
    // register here rather than waiting for a sweep.
    host.register_agent(coord, name)?;
    // No `kick_agent` here any more — `hive-agent` pushes its own "you
    // were just (re)started" todo on boot now, so this node doesn't need
    // to tell it. See the matching comment in `run_rebuild_bookkeeping`.
    coord.rescan_containers_and_emit().await;
    Ok(())
}

/// Mechanical container stop — the sub-step a `Reconcile` planner fans
/// out when it observes `wanted = Offline` and the container up.
async fn run_stop(host: &impl Host, coord: &Arc<Coordinator>, name: &str) -> Result<()> {
    // See `run_start`: no node-local guard — `Stop` reports `Stopping` from its
    // own kind now.
    host.kill(name).await?;
    coord.unregister_agent(name);
    host.notify(format!("killed:{name}"), format!("agent '{name}' killed"))
        .await;
    coord.rescan_containers_and_emit().await;
    Ok(())
}

/// Mechanical stop for the profile swap. Never *changes* `wanted`;
/// noop when already stopped.
async fn run_stop_for_update(host: &impl Host, coord: &Arc<Coordinator>, name: &str) -> Result<()> {
    if host.is_running(name).await {
        // Seed a missing agent_power row from the PRE-stop observation
        // — the DAG's tail `Reconcile` observes only the mechanically
        // stopped state and would otherwise seed a running-but-unknown
        // agent as `Offline`, stranding it down after its own rebuild.
        if let Err(e) = coord.power.get_or_seed(name, true) {
            tracing::warn!(%name, error = ?e, "agent_power: pre-stop seed failed");
        }
        host.kill(name).await?;
        coord.rescan_containers_and_emit().await;
    }
    Ok(())
}

/// Set the graceful fence + kick so the harness sees it promptly and
/// runs its one stop-checkpoint turn.
///
/// Skipped entirely for a paused agent: its loop parks on the pause
/// marker without polling the broker, so it would never observe the
/// fence and the downstream drain would just burn
/// `GRACEFUL_STOP_TIMEOUT`. Safe because the harness tests the marker at
/// the top of its loop — a paused agent has no turn in flight, so there
/// is nothing to checkpoint.
fn run_signal(coord: &Arc<Coordinator>, name: &str) {
    if hive_types::Ident::parse(name).is_ok_and(|a| Coordinator::is_paused(&a)) {
        return;
    }
    coord.mark_graceful_stop(name);
    coord.kick_agent(name, "graceful stop requested");
}

/// Await the harness clearing the fence (`GracefulStopComplete`) or
/// the timeout — either way the downstream `Reconcile` proceeds with
/// the actual stop.
async fn run_drain(
    coord: &Arc<Coordinator>,
    name: &str,
    timeout: std::time::Duration,
) -> Result<()> {
    let deadline = std::time::Instant::now() + timeout;
    while coord.is_graceful_stop_pending(name) {
        if std::time::Instant::now() >= deadline {
            tracing::warn!(agent = %name, "graceful stop: drain timed out — hard-stopping");
            break;
        }
        tokio::time::sleep(std::time::Duration::from_millis(500)).await;
    }
    coord.clear_graceful_stop(name);
    Ok(())
}

/// Write the pause marker + mark `pause_pending`, no kick (unlike
/// `run_signal`) — the harness's own between-turns poll (`PAUSE_POLL`,
/// 1s default) is already responsive enough, and `run_signal`'s kick
/// message ("you were just (re)started") would be actively misleading
/// here.
///
/// Skips marking `pause_pending` (the marker write still happens,
/// harmlessly idempotent either way) if the agent is already paused:
/// the harness reports `PauseAcknowledged` only on the marker's
/// `false → true` edge, so re-pausing an already-paused agent produces
/// no edge for `run_pause_drain` to observe — marking pending here
/// would just burn its timeout every time an operator re-confirms a
/// pause that already took.
async fn run_pause_signal(host: &impl Host, coord: &Arc<Coordinator>, name: &str) -> Result<()> {
    let agent = hive_types::Ident::parse(name)
        .map_err(|e| anyhow::anyhow!("invalid agent name for pause {name:?}: {e}"))?;
    let already_paused = Coordinator::is_paused(&agent);
    host.set_paused(&agent, true).await?;
    if !already_paused {
        coord.mark_pause_pending(name);
    }
    // Same pattern `run_start`/`run_stop` use: refresh the dashboard's
    // view right after the state change so the paused badge flips
    // immediately instead of waiting on the next periodic rescan.
    coord.rescan_containers_and_emit().await;
    Ok(())
}

/// Await the harness reporting `PauseAcknowledged`, bounded by
/// `PAUSE_ACK_TIMEOUT`. Resolves ok either way, mirroring `run_drain` —
/// pausing is best-effort from the queue's perspective; the marker
/// (not this node) is what actually gates the harness's turn loop, so
/// a timed-out wait doesn't leave the agent un-paused, just leaves the
/// dashboard's "pausing…" badge running a little longer than it needed
/// to.
async fn run_pause_drain(
    coord: &Arc<Coordinator>,
    name: &str,
    timeout: std::time::Duration,
) -> Result<()> {
    let deadline = std::time::Instant::now() + timeout;
    while coord.is_pause_pending(name) {
        if std::time::Instant::now() >= deadline {
            tracing::warn!(agent = %name, "pause: ack wait timed out — marker is set regardless");
            break;
        }
        tokio::time::sleep(std::time::Duration::from_millis(500)).await;
    }
    coord.clear_pause_pending(name);
    Ok(())
}

/// `set_nspawn_flags` + `set_resource_limits` + daemon-reload.
async fn run_write_dropin(host: &impl Host, coord: &Arc<Coordinator>, name: &str) -> Result<()> {
    // write_dropins only needs the path value to build AgentPaths; the
    // dir doesn't need to exist at this point (created by ensure_agent_runtime_dir
    // on the upstream Prebuild/Start node).
    let agent_dir = crate::paths::agent_runtime_dir(name);
    let hive = coord.hive_env();
    let paths = Coordinator::agent_paths(name, agent_dir)?;
    host.write_dropins(name, &hive, &paths).await?;
    Ok(())
}

#[cfg(test)]
mod tests;