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

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

use std::sync::Arc;

use anyhow::{Context as _, Result};

use super::Claim;
use super::model::{NodeKind, NodeSpec, State};
use crate::coordinator::Coordinator;
use crate::power::{ReconcileAction, reconcile_action};

/// 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. Drains hold no build slot, so a whole-hive
/// graceful stop overlaps every agent's drain instead of serialising
/// N × this timeout.
pub const GRACEFUL_STOP_TIMEOUT: std::time::Duration = std::time::Duration::from_mins(3);

/// Extra signal an executor hands back to the scheduler alongside
/// success.
#[derive(Debug, Default)]
pub struct NodeOutput {
    /// Whole per-agent *subgraphs* to append into *this same* DAG at
    /// runtime — the single in-DAG-growth channel. Each inner
    /// `Vec<NodeSpec>` is one independent subgraph whose `deps` are local
    /// (0-based within that subgraph); the scheduler appends each via
    /// [`super::JobQueue::append_subgraph`], which rebases the deps onto the DAG's
    /// node-id space and roots the subgraph on the emitting node. Used both
    /// for the multi-node case (`MetaLock` growing one rebuild subgraph per
    /// agent — the startup sweep's stale agents, the meta-update cascade's
    /// affected agents) and the single-node case (a `Reconcile` planner
    /// emitting its mechanical `Start` / `Stop` as a one-node subgraph). The
    /// scheduler applies these *before* the emitting node's completion so the
    /// DAG never rolls terminal with the appended work still pending — keeping
    /// the lease-window transient held across the sub-step.
    pub append_subgraph: Vec<Vec<NodeSpec>>,
}

/// Step-label + build-log sink for one claimed node.
struct Ctx<'a> {
    coord: &'a Arc<Coordinator>,
    dag_id: u64,
    node_id: super::NodeId,
}

impl Ctx<'_> {
    fn step(&self, step: &str) {
        if self
            .coord
            .job_queue
            .set_step(self.dag_id, self.node_id, step)
        {
            self.coord.emit_rebuild_queue_snapshot();
        }
    }

    fn build_log(&self, log_id: i64) {
        if self
            .coord
            .job_queue
            .set_build_log_id(self.dag_id, self.node_id, log_id)
        {
            self.coord.emit_rebuild_queue_snapshot();
        }
    }
}

/// Run one claimed node to completion. Called from a task the
/// scheduler spawns per claim; the `Result` (stringified) becomes the
/// node's terminal state.
pub(super) async fn run_node(coord: &Arc<Coordinator>, claim: &Claim) -> Result<NodeOutput> {
    let ctx = Ctx {
        coord,
        dag_id: claim.dag_id,
        node_id: claim.node_id,
    };
    match &claim.kind {
        NodeKind::MetaSync { relock, .. } => run_meta_sync(coord, claim, *relock).await,
        NodeKind::Prebuild { .. } => run_prebuild(claim, &ctx).await,
        NodeKind::Swap { .. } => run_swap(coord, claim, &ctx).await,
        NodeKind::PostSwap { .. } => run_post_swap(coord, claim, &ctx).await,
        NodeKind::Provision { .. } => run_provision(coord, claim, &ctx).await,
        NodeKind::Create { .. } => run_create(claim, &ctx).await,
        NodeKind::MetaLock { sweep, fanout } => {
            run_meta_lock(coord, claim, &ctx, *sweep, fanout.clone()).await
        }
        NodeKind::Reconcile { .. } => run_reconcile(coord, claim).await,
        NodeKind::Start { .. } => run_start(coord, claim, &ctx).await,
        NodeKind::Stop { .. } => run_stop(coord, claim, &ctx).await,
        NodeKind::StopForUpdate { .. } => run_stop_for_update(coord, claim, &ctx).await,
        NodeKind::Signal { .. } => Ok(run_signal(coord, claim, &ctx)),
        NodeKind::Drain { .. } => run_drain(coord, claim, &ctx).await,
        NodeKind::WriteDropin { .. } => run_write_dropin(coord, claim).await,
        NodeKind::WritePermFile { .. } => run_write_perm_file(coord, claim, &ctx).await,
        NodeKind::DeployWindow { .. } => run_deploy_window(claim),
        NodeKind::MergeVerify { .. } => run_merge_verify(coord, claim).await,
        NodeKind::DeployApply { .. } => run_deploy_apply(coord, claim).await,
        NodeKind::FinalizeDeploy { .. } => run_finalize_deploy(coord, claim).await,
        NodeKind::DeployTail { .. } => run_deploy_tail(coord, claim).await,
        NodeKind::SetWanted { up, .. } => run_set_wanted(coord, claim, *up),
        // Pure grouping container — no work; completing it lets it reach
        // `Finishing` so its child template nodes start. The DAG's terminal
        // hook fires (inline, via `run_terminal_hook`) when the container itself
        // rolls up terminal — not as a scheduled node.
        NodeKind::Dag { .. } => Ok(NodeOutput::default()),
    }
}

/// Run a settled DAG's inline terminal hook, dispatched off its rolled-up
/// summary — the container-terminal replacement for the old per-DAG hook node.
/// Always best-effort: a hook failure is logged inside, never surfaced.
pub(crate) async fn run_terminal_hook(coord: &Arc<Coordinator>, terminal: &super::TerminalDag) {
    match super::terminal_hook(terminal.template, terminal.approval_id) {
        Some(super::HookKind::ResolveApproval) => {
            crate::actions::resolve_approval_dag(coord, terminal).await;
        }
        Some(super::HookKind::EmitRebuilt) => emit_rebuilt(coord, terminal),
        Some(super::HookKind::RevertIntent) => revert_intent(coord, terminal).await,
        None => {}
    }
}

/// Rebuild / perm-change hook: emit one `Rebuilt` manager event per targeted
/// agent — `ok` on `Done`, `!ok` on `Failed`, none on cancel.
fn emit_rebuilt(coord: &Arc<Coordinator>, terminal: &super::TerminalDag) {
    for agent in &terminal.agents {
        match terminal.state {
            State::Done => coord.notify_manager(&hive_sh4re::HelperEvent::Rebuilt {
                agent: agent.clone(),
                ok: true,
                note: None,
                sha: None,
                tag: None,
            }),
            State::Failed => coord.notify_manager(&hive_sh4re::HelperEvent::Rebuilt {
                agent: agent.clone(),
                ok: false,
                note: terminal.error.clone(),
                sha: None,
                tag: None,
            }),
            _ => {}
        }
    }
}

/// Power-op hook: on a *cancelled* DAG, revert each targeted agent's `wanted`
/// intent to its observed state — the operator's cancel means "don't do it", so
/// the intent snaps back instead of the flip executing as a surprise side effect
/// of some later reconcile. Noop on any non-cancelled outcome.
async fn revert_intent(coord: &Arc<Coordinator>, terminal: &super::TerminalDag) {
    if terminal.state != State::Cancelled {
        return;
    }
    for agent in &terminal.agents {
        let running = crate::lifecycle::is_running(agent).await;
        if let Err(e) = coord
            .power
            .set(agent, crate::power::Wanted::from_running(running))
        {
            tracing::warn!(%agent, error = ?e, "agent_power: cancel revert failed");
        }
    }
}

/// 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 takes the agent's lifecycle lease (see
/// `NodeKind::needs_lease`) 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>, claim: &Claim, up: bool) -> Result<NodeOutput> {
    let wanted = if up {
        crate::power::Wanted::Up
    } else {
        crate::power::Wanted::Offline
    };
    coord
        .power
        .set(&claim.agent, wanted)
        .with_context(|| format!("set wanted={} for agent {}", wanted.as_str(), claim.agent))?;
    Ok(NodeOutput::default())
}

/// The rebuild's meta preamble: runtime-dir prep, an idempotent meta
/// `sync_agents`, and the optional per-agent relock. Runs under the deploy
/// window (`NodeKind::needs_meta_window`, held by the scheduler for this
/// node) so its commits can never land inside another node's staged
/// prepare→finalize window.
///
/// Deliberately a separate node from the [`run_prebuild`] it feeds: that
/// build takes minutes and only *reads* the store, so keeping the global
/// window off it is what lets rebuilds of different agents overlap.
async fn run_meta_sync(
    coord: &Arc<Coordinator>,
    claim: &Claim,
    relock: bool,
) -> Result<NodeOutput> {
    let name = &claim.agent;
    // 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 hive = coord.hive_env();
    let paths = Coordinator::agent_paths(name, agent_dir);
    crate::lifecycle::prepare_rebuild_dirs(name, &paths).await?;
    // Idempotent meta sync so a manual rebuild can also recover from a
    // divergent meta repo; then bump just this agent's input. `relock =
    // false` only for meta-update cascade children, where re-locking
    // would revert the bump the cascade just committed.
    let agents = crate::lifecycle::agents_for_meta_listing().await?;
    crate::meta::sync_agents(&hive, &agents).await?;
    if relock {
        crate::meta::lock_update_for_rebuild(name).await?;
    }
    Ok(NodeOutput::default())
}

/// Out-of-band toplevel build while the container keeps serving: warm
/// `system.build.toplevel` so the later `Swap` hits cache and skips
/// straight to the profile-swap, against a meta repo the upstream
/// `MetaSync` node has already synced. The warm build is skipped when the
/// container is already down: its only purpose is to shrink the swap's
/// downtime window, so a stopped agent (no uptime to preserve) doesn't
/// pay the double eval — `Swap` builds inline instead.
async fn run_prebuild(claim: &Claim, ctx: &Ctx<'_>) -> Result<NodeOutput> {
    let name = &claim.agent;
    // Warm the toplevel build only when the container is up — the whole
    // point of prebuild is to shrink the swap's downtime window. A
    // stopped agent has no uptime to preserve, so skip the (expensive)
    // eval and let the downstream `Swap` build inline.
    if crate::lifecycle::is_running(name).await {
        ctx.step("nix build");
        let flake_ref = format!("{}#{name}", crate::paths::meta_root().display());
        crate::lifecycle::prebuild_toplevel(name, &flake_ref, &|log_id| ctx.build_log(log_id))
            .await?;
    } else {
        ctx.step("skipped prebuild (container down)");
    }
    Ok(NodeOutput::default())
}

/// Profile-swap: re-apply drop-ins (rebuild is the reconcile verb),
/// `nixos-container update`, then the post-rebuild bookkeeping tail
/// (rev marker, `Rebuilt` event, forge/matrix 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(coord: &Arc<Coordinator>, claim: &Claim, ctx: &Ctx<'_>) -> Result<NodeOutput> {
    let name = &claim.agent;
    // 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 result =
        crate::lifecycle::swap_update(name, &hive, &paths, &|step| ctx.step(step), &|log_id| {
            ctx.build_log(log_id);
        })
        .await;
    // On success the Ok-only bookkeeping tail (rev marker, forge/matrix
    // sync, kick, rescan, snapshot) runs in the sibling `PostSwap` node,
    // which deps `AfterOk(Swap)`. On failure `PostSwap` is cancel-cascaded
    // and the tail `Reconcile` (`AfterAny(PostSwap)`) handles recovery; here
    // we only refresh the observed state so dashboards reflect the failed
    // swap immediately. The `Rebuilt { ok: false }` manager event fires once
    // per DAG from the terminal hook (any node may be the one that failed).
    if result.is_err() {
        coord.rescan_containers_and_emit().await;
    }
    result.map(|()| NodeOutput::default())
}

/// 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/matrix 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_post_swap(
    coord: &Arc<Coordinator>,
    claim: &Claim,
    ctx: &Ctx<'_>,
) -> Result<NodeOutput> {
    let name = &claim.agent;
    if let Some(rev) = crate::auto_update::current_flake_rev(&coord.hyperhive_flake)
        && let Err(e) = std::fs::write(crate::paths::applied_rev_marker(name), rev)
    {
        tracing::warn!(%name, error = ?e, "write rev marker failed");
    }
    // The `Rebuilt` manager event fires exactly once per DAG from the
    // terminal hook — emitting ok here and letting a failed tail `Reconcile`
    // add a contradictory !ok would double-report the same rebuild.
    ctx.step("forge sync");
    // Full forge + matrix sync on every successful rebuild so the rebuild
    // path is equivalent to the startup sweep: tokens, config-repo mirror,
    // meta access all recover without a hive-c0re restart.
    crate::forge::sync_agent(name, crate::forge::core_token().as_deref()).await;
    crate::matrix::sync_agent_standalone(name).await;
    // Wake the agent on its next turn so claude sees a "you were rebuilt"
    // hint; rescan so dashboards drop the "needs update" chip; lock bump →
    // meta-inputs re-render.
    coord.kick_agent(name, "container rebuilt");
    coord.rescan_containers_and_emit().await;
    crate::dashboard::emit_meta_inputs_snapshot(coord);
    Ok(NodeOutput::default())
}

/// First-spawn pre-create provisioning: proposed/applied repos, state
/// subvolume, and the meta `sync_agents` registration. Runs under the
/// deploy window (`NodeKind::needs_meta_window`) so its commit can't
/// land inside another node's staged deploy window.
async fn run_provision(
    coord: &Arc<Coordinator>,
    claim: &Claim,
    ctx: &Ctx<'_>,
) -> Result<NodeOutput> {
    let name = &claim.agent;
    let agent_dir = crate::paths::agent_runtime_dir(name);
    let hive = coord.hive_env();
    let paths = Coordinator::agent_paths(name, agent_dir);
    ctx.step("provisioning");
    crate::lifecycle::provision_container(name, &hive, &paths).await?;
    Ok(NodeOutput::default())
}

/// `nixos-container create` proper — the upstream `Provision` node
/// already registered the agent in meta, so this only reads the store
/// (no deploy-window gate needed, mirroring `Prebuild`'s build). 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".
async fn run_create(claim: &Claim, ctx: &Ctx<'_>) -> Result<NodeOutput> {
    ctx.step("nixos-container create");
    crate::lifecycle::create_only(&claim.agent).await?;
    Ok(NodeOutput::default())
}

/// Meta flake lock bump. Boot-sweep flavour is non-fatal (a failed
/// bump must not cancel the fan-out rebuilds — they proceed against
/// the current lock, exactly like today's sweep); the meta-update
/// flavour propagates errors, and a failed bump fans out nothing.
async fn run_meta_lock(
    coord: &Arc<Coordinator>,
    claim: &Claim,
    ctx: &Ctx<'_>,
    sweep: bool,
    fanout: Option<Vec<String>>,
) -> Result<NodeOutput> {
    if sweep {
        ctx.step("nix flake update hyperhive");
        if let Err(e) = crate::meta::lock_update_hyperhive().await {
            tracing::warn!(error = ?e, "startup sweep: meta lock_update_hyperhive failed");
        }
        // Grow one rebuild subgraph per stale agent into *this* boot DAG
        // (rooted on this `MetaLock`, so they build against the post-bump
        // lock), rather than fanning out child DAGs. `relock = true` — a
        // boot sweep relocks per-agent like a manual rebuild.
        let append_subgraph = fanout
            .unwrap_or_default()
            .iter()
            .map(|agent| super::templates::rebuild_nodes(agent, true, 0))
            .collect();
        return Ok(NodeOutput { append_subgraph });
    }
    let _progress = coord.meta_update_guard();
    ctx.step("nix flake update");
    crate::meta::lock_update(&claim.inputs).await?;
    // Lock file changed — meta-inputs panel re-renders.
    crate::dashboard::emit_meta_inputs_snapshot(coord);
    let cascade = match fanout {
        Some(list) => list,
        None => meta_update_cascade_agents(&claim.inputs).await,
    };
    // Grow one rebuild subgraph per affected agent into *this* meta-update
    // DAG (rooted on this `MetaLock`, so they build against the post-bump
    // lock), rather than fanning out child DAGs. `relock = false` — the
    // cascade children must NOT re-lock, which would revert the bump this
    // node just committed (the property the old `fanout_specs` meta-update
    // branch encoded).
    let append_subgraph = cascade
        .iter()
        .map(|agent| super::templates::rebuild_nodes(agent, false, 0))
        .collect();
    Ok(NodeOutput { append_subgraph })
}

/// 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 `NodeOutput::append_subgraph` 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.
async fn run_reconcile(coord: &Arc<Coordinator>, claim: &Claim) -> Result<NodeOutput> {
    let name = &claim.agent;
    let running = crate::lifecycle::is_running(name).await;
    let wanted = coord.power.get_or_seed(name, running)?;
    // One node targeting this agent, rooted on this reconcile node. `NodeKind`
    // carries the agent it targets, so stamp `claim.agent` into the fanned-out
    // Start/Stop kind (one in-DAG-growth channel).
    let sub = |kind| vec![vec![super::templates::node(kind, Vec::new())]];
    let append_subgraph = match reconcile_action(wanted, running) {
        ReconcileAction::Start => sub(NodeKind::Start {
            agent: name.clone(),
        }),
        ReconcileAction::Stop => sub(NodeKind::Stop {
            agent: name.clone(),
        }),
        ReconcileAction::Noop => {
            tracing::debug!(%name, wanted = wanted.as_str(), running, "reconcile: noop");
            Vec::new()
        }
    };
    Ok(NodeOutput { append_subgraph })
}

/// Mechanical container start — the sub-step a `Reconcile` planner fans
/// out when it observes `wanted = Up` and the container down.
async fn run_start(coord: &Arc<Coordinator>, claim: &Claim, ctx: &Ctx<'_>) -> Result<NodeOutput> {
    let name = &claim.agent;
    // Node-local transient only when the DAG holds none (the
    // boot-reconcile template); a rebuild/spawn/etc. DAG's lease-window
    // transient already covers this node.
    let _guard = claim
        .transient
        .is_none()
        .then(|| coord.transient_guard(name, crate::coordinator::TransientKind::Starting));
    // 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);
    let token = crate::lifecycle::converge_start_preamble(name, &hive, &paths).await?;
    ctx.step("nixos-container start");
    crate::lifecycle::start_with_fallback(token).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.
    coord.register_agent(name)?;
    coord.kick_agent(name, "container started");
    coord.rescan_containers_and_emit().await;
    Ok(NodeOutput::default())
}

/// Mechanical container stop — the sub-step a `Reconcile` planner fans
/// out when it observes `wanted = Offline` and the container up.
async fn run_stop(coord: &Arc<Coordinator>, claim: &Claim, ctx: &Ctx<'_>) -> Result<NodeOutput> {
    let name = &claim.agent;
    let _guard = claim
        .transient
        .is_none()
        .then(|| coord.transient_guard(name, crate::coordinator::TransientKind::Stopping));
    ctx.step("nixos-container stop");
    crate::lifecycle::kill(name).await?;
    coord.unregister_agent(name);
    coord.notify_manager(&hive_sh4re::HelperEvent::Killed {
        agent: name.clone(),
    });
    coord.rescan_containers_and_emit().await;
    Ok(NodeOutput::default())
}

/// Mechanical stop for the profile swap. Never *changes* `wanted`;
/// noop when already stopped.
async fn run_stop_for_update(
    coord: &Arc<Coordinator>,
    claim: &Claim,
    ctx: &Ctx<'_>,
) -> Result<NodeOutput> {
    let name = &claim.agent;
    if crate::lifecycle::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");
        }
        ctx.step("nixos-container stop");
        crate::lifecycle::kill(name).await?;
        coord.rescan_containers_and_emit().await;
    }
    Ok(NodeOutput::default())
}

/// 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>, claim: &Claim, ctx: &Ctx<'_>) -> NodeOutput {
    if hive_types::Ident::parse(&claim.agent).is_ok_and(|a| Coordinator::is_paused(&a)) {
        ctx.step("graceful stop: agent paused, nothing to drain");
        return NodeOutput::default();
    }
    ctx.step("graceful stop: signalling agent");
    coord.mark_graceful_stop(&claim.agent);
    coord.kick_agent(&claim.agent, "graceful stop requested");
    NodeOutput::default()
}

/// 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>, claim: &Claim, ctx: &Ctx<'_>) -> Result<NodeOutput> {
    let name = &claim.agent;
    ctx.step("graceful stop: draining");
    let deadline = std::time::Instant::now() + GRACEFUL_STOP_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(NodeOutput::default())
}

/// `set_nspawn_flags` + `set_resource_limits` + daemon-reload.
async fn run_write_dropin(coord: &Arc<Coordinator>, claim: &Claim) -> Result<NodeOutput> {
    let name = &claim.agent;
    // 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);
    crate::lifecycle::write_dropins(name, &hive, &paths).await?;
    Ok(NodeOutput::default())
}

/// Write + commit the perm file(s) (fused under `META_LOCK` so the
/// working tree is never left dirty), then emit the P3RM1SS10NS-tab
/// snapshots so the dashboard reflects the new assignment.
async fn run_write_perm_file(
    coord: &Arc<Coordinator>,
    claim: &Claim,
    ctx: &Ctx<'_>,
) -> Result<NodeOutput> {
    use super::model::PermPayload;
    let name = &claim.agent;
    // The perm file payload rides the node itself (the only consumer).
    let NodeKind::WritePermFile { payload, .. } = &claim.kind else {
        anyhow::bail!("run_write_perm_file on a non-WritePermFile node");
    };
    ctx.step("writing + committing perm file");
    // Runs under the deploy window (`NodeKind::needs_meta_window`): a
    // perm commit landing inside another node's staged prepare→finalize
    // window would sweep the staged deploy lock into its commit (the
    // commits are also path-limited in meta.rs — belt and braces).
    match payload {
        PermPayload::ToolGroups { groups } => {
            crate::meta::commit_tool_groups(name, groups)
                .await
                .with_context(|| format!("commit tool-groups for {name}"))?;
            coord.emit_tool_groups_snapshot();
        }
        PermPayload::Capabilities { caps } => {
            crate::meta::commit_capabilities(name, caps)
                .await
                .with_context(|| format!("commit capabilities for {name}"))?;
            coord.emit_capabilities_snapshot();
        }
        PermPayload::Combined { groups, caps } => {
            crate::meta::commit_perms(name, groups.as_deref(), caps.as_deref())
                .await
                .with_context(|| format!("commit perms for {name}"))?;
            if groups.is_some() {
                coord.emit_tool_groups_snapshot();
            }
            if caps.is_some() {
                coord.emit_capabilities_snapshot();
            }
        }
    }
    Ok(NodeOutput::default())
}

/// The approval id every deploy phase re-reads its approval row by. Fails the
/// node when the DAG carries none, which would mean a `MergeConfigPr` DAG was
/// built without going through `templates::approval_deploy`.
fn deploy_approval_id(claim: &Claim) -> Result<i64> {
    claim
        .approval_id
        .with_context(|| format!("approval deploy dag {} has no approval_id", claim.dag_id))
}

/// The deploy subtree's root: pure resource holder, no work of its own.
///
/// It exists so the global meta window (plus the agent lease and a build slot)
/// is held continuously across every phase below it. `prepare_deploy` leaves
/// `flake.lock` staged-uncommitted for the whole container build, and any other
/// meta mutation landing inside that span would sweep the staged lock into its
/// own commit and neuter `abort_deploy` — so the window has to outlive any one
/// node, which the `MutexGuard` this replaced could not do.
///
/// Completing immediately moves it to `Finishing`, which is what starts the
/// children; the resources stay held until the whole subtree settles.
fn run_deploy_window(claim: &Claim) -> Result<NodeOutput> {
    deploy_approval_id(claim)?;
    Ok(NodeOutput::default())
}

/// Deploy phase 1 — drift gate, fetch, eval-verify. Mutates nothing, so a
/// failure here cancel-cascades the rest of the subtree with the forge and the
/// applied repo exactly as they were.
async fn run_merge_verify(coord: &Arc<Coordinator>, claim: &Claim) -> Result<NodeOutput> {
    crate::actions::run_deploy_merge_verify(coord, Some(claim.dag_id), deploy_approval_id(claim)?)
        .await
        .map(|()| NodeOutput::default())
}

/// Deploy phase 2 — the irreversible half: ff-merge, then phase 1 of the
/// two-phase meta deploy.
///
/// On success it grows the ordinary rebuild subgraph (plus its closing
/// `FinalizeDeploy`) into this DAG rooted on *this* node — which is what puts
/// the appended nodes inside the `DeployWindow`'s subtree, so the `MetaWindow`
/// their `MetaSync` declares is re-entered rather than deadlocked against the
/// ancestor already holding it. On failure nothing is appended and the tail
/// compensates, exactly as before.
async fn run_deploy_apply(coord: &Arc<Coordinator>, claim: &Claim) -> Result<NodeOutput> {
    crate::actions::run_deploy_apply(coord, Some(claim.dag_id), deploy_approval_id(claim)?).await?;
    Ok(NodeOutput {
        append_subgraph: vec![super::templates::deploy_rebuild_nodes(claim.kind.agent())],
    })
}

/// Deploy phase 3 — close the staged-lock window once the appended rebuild has
/// come up clean: drop the rollback ref, plant the `deployed/<id>` tag, commit
/// the staged lock.
async fn run_finalize_deploy(coord: &Arc<Coordinator>, claim: &Claim) -> Result<NodeOutput> {
    crate::actions::run_finalize_deploy(coord, Some(claim.dag_id), deploy_approval_id(claim)?)
        .await
        .map(|()| NodeOutput::default())
}

/// Deploy compensation + bookkeeping tail. `AfterAny` the apply node, so it
/// runs on every outcome; it is deliberately infallible (see
/// [`crate::actions::run_deploy_tail`]) — a failing tail must not flip an
/// otherwise-successful deploy's DAG state.
///
/// Takes the agent from the node payload so the tail can still compensate when
/// the approval row is gone (deny race, purge).
async fn run_deploy_tail(coord: &Arc<Coordinator>, claim: &Claim) -> Result<NodeOutput> {
    crate::actions::run_deploy_tail(
        coord,
        Some(claim.dag_id),
        claim.kind.agent(),
        deploy_approval_id(claim)?,
    )
    .await;
    Ok(NodeOutput::default())
}

/// Compute which agents a `nix flake update <inputs>` on the meta
/// flake affects — the fan-out set for `MetaUpdate` DAGs. Empty
/// `inputs` or any input under `hyperhive` → every container;
/// otherwise just the agents named by `agent-<name>` inputs.
/// Topology-sorted so parents rebuild before their children.
pub async fn meta_update_cascade_agents(inputs: &[String]) -> Vec<String> {
    let touched_hyperhive = inputs
        .iter()
        .any(|i| i == "hyperhive" || i.starts_with("hyperhive/"));
    let touched_agents: Vec<String> = inputs
        .iter()
        .filter_map(|i| i.strip_prefix("agent-"))
        .map(|rest| rest.split('/').next().unwrap_or(rest).to_owned())
        .collect();
    let mut names = if touched_hyperhive || inputs.is_empty() {
        crate::lifecycle::list()
            .await
            .unwrap_or_default()
            .into_iter()
            .filter_map(|c| {
                c.strip_prefix(crate::lifecycle::AGENT_PREFIX)
                    .map(str::to_owned)
            })
            .collect()
    } else {
        touched_agents
    };
    let topo = crate::topology::read();
    crate::auto_update::topology_sort(&mut names, &topo);
    names
}