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 hive_jobq::{NodeId, TerminalState};

use super::model::NodeKind;
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.
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;

/// 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<()>) {
    // The agent this node targets rides the payload — empty for the agentless
    // kinds (`MetaLock`, `Reparent`), which never read it.
    let agent = kind.agent();
    // Every arm is `Result<()>`; the three that grow work declare into `builder`
    // *synchronously*, after their 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 { relock, .. } => run_meta_sync(coord, agent, *relock).await,
        NodeKind::Prebuild { .. } => run_prebuild(agent, id).await,
        NodeKind::Swap { .. } => run_swap(coord, agent, id).await,
        NodeKind::RebuildBookkeeping { .. } => run_rebuild_bookkeeping(coord, agent).await,
        NodeKind::Provision { .. } => run_provision(coord, agent).await,
        NodeKind::Create { .. } => run_create(agent).await,
        NodeKind::DestroyContainer { .. } => run_destroy_container(coord, agent).await,
        NodeKind::PurgeState { .. } => {
            run_purge_state(agent).await;
            Ok(())
        }
        NodeKind::DestroyBookkeeping { purge, .. } => {
            run_destroy_bookkeeping(coord, agent, *purge).await;
            Ok(())
        }
        NodeKind::MetaLock {
            sweep,
            fanout,
            inputs,
        } => run_meta_lock(coord, *sweep, fanout.clone(), inputs)
            .await
            .map(|agents| {
                // `sweep` is the whole difference: it relocks per-agent like a
                // manual rebuild, and it drains agents that were mid-turn when
                // the host came up. A cascade does neither. Decided here rather
                // than returned, since `run_meta_lock` would only be deriving
                // it from the `sweep` this call site already holds.
                if *sweep {
                    super::templates::grown_graceful_rebuilds(&builder, &agents, true);
                } else {
                    super::templates::grown_rebuilds(&builder, &agents, false);
                }
            }),
        NodeKind::Reconcile { .. } => run_reconcile(coord, agent).await.map(|sub| {
            if let Some(kind) = sub {
                super::templates::fanned_out_mechanical(&builder, kind);
            }
        }),
        NodeKind::Start { .. } => run_start(coord, agent).await,
        NodeKind::Stop { .. } => run_stop(coord, agent).await,
        NodeKind::StopForUpdate { .. } => run_stop_for_update(coord, agent).await,
        NodeKind::Signal { .. } => {
            run_signal(coord, agent);
            Ok(())
        }
        NodeKind::Drain { .. } => run_drain(coord, agent).await,
        NodeKind::PauseSignal { .. } => run_pause_signal(coord, agent).await,
        NodeKind::PauseDrain { .. } => run_pause_drain(coord, agent).await,
        NodeKind::WriteDropin { .. } => run_write_dropin(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::WritePermFile { payload, .. } => run_write_perm_file(coord, agent, payload).await,
        NodeKind::Reparent { moves } => run_reparent(coord, moves).await,
        NodeKind::MergeVerify { approval_id, .. } => {
            run_merge_verify(coord, *approval_id, id).await
        }
        NodeKind::DeployApply { approval_id, .. } => {
            run_deploy_apply(coord, *approval_id, id).await.map(|()| {
                super::templates::deploy_rebuild_nodes(&builder, agent, *approval_id);
            })
        }
        NodeKind::FinalizeDeploy { approval_id, .. } => {
            run_finalize_deploy(coord, *approval_id).await
        }
        NodeKind::DeployTail { approval_id, .. } => {
            run_deploy_tail(coord, coord.job_queue.root_of(id), agent, *approval_id).await
        }
        NodeKind::ResolveApproval {
            approval_id,
            outcome,
        } => run_resolve_approval(coord, coord.job_queue.root_of(id), *approval_id, *outcome).await,
        NodeKind::EmitRebuilt { ok, .. } => {
            run_emit_rebuilt(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::DeployWindow { .. } | NodeKind::AgentWindow { .. } => Ok(()),
        NodeKind::ForgeSweep => run_forge_sweep().await,
        NodeKind::MatrixSweep => run_matrix_sweep().await,
        NodeKind::WebhookRegister => run_webhook_register().await,
        NodeKind::KnowledgePull => run_knowledge_pull(coord).await,
        NodeKind::WantedPull => run_wanted_pull(coord).await,
    };
    (builder, result)
}

/// Boot-time forge user/token sweep as a DAG node — see
/// [`NodeKind::ForgeSweep`]. `forge::ensure_all` already does its own
/// per-step error handling and boot-warning banners internally (it's
/// best-effort by design), so this wrapper has nothing left to report;
/// it exists purely to make the sweep a visible unit of work.
async fn run_forge_sweep() -> Result<()> {
    crate::forge::ensure_all().await;
    Ok(())
}

/// Boot-time matrix user/space sweep as a DAG node — see
/// [`NodeKind::MatrixSweep`]. Reports failure as the node's own error so a
/// failed boot sweep is visible on the dashboard; the debounced
/// `sweep_health`-driven warning banner is a separate concern owned by the
/// periodic loop in `main.rs`, unaffected by this node's own outcome.
async fn run_matrix_sweep() -> Result<()> {
    if crate::matrix::ensure_all().await {
        Ok(())
    } else {
        anyhow::bail!("matrix ensure_all: one or more agents failed sync (see logs)")
    }
}

/// Boot-time Forgejo webhook management as a DAG node — see
/// [`NodeKind::WebhookRegister`]. Mirrors the guard chain the
/// `tokio::spawn` block it replaced used: no-op (not an error) when the
/// HMAC secret, core token, or hive domain aren't available yet.
///
/// The node now does one of each: it still registers the config-PR hook,
/// and it *removes* the knowledge one. A knowledge push is delivered to
/// the swarm controller, which addresses an event to each hive over the
/// queue — so a hive holding its own registration is holding a shared
/// resource only one party can own. The removal runs every boot rather
/// than behind a marker because it is already idempotent: it is a no-op
/// the moment the hook is gone.
async fn run_webhook_register() -> Result<()> {
    let Ok(webhook_secret) = crate::webhook_secret::load_or_generate() else {
        tracing::debug!("webhook secret unavailable; skipping hook registration");
        return Ok(());
    };
    let Some(token) = crate::forge::core_token() else {
        return Ok(());
    };
    let domain = std::env::var("HYPERHIVE_HIVE_DOMAIN")
        .ok()
        .filter(|v| !v.is_empty());
    let Some(domain) = domain else {
        tracing::debug!("HYPERHIVE_HIVE_DOMAIN unset; skipping webhook registration");
        return Ok(());
    };
    if let Err(e) = crate::workers::knowledge::remove_webhook(&token, &domain).await {
        tracing::warn!(error = ?e, "knowledge: remove_webhook failed");
    }
    if let Err(e) = crate::forge::ensure_config_pr_webhook(&token, &domain, &webhook_secret).await {
        tracing::warn!(error = ?e, "forge: ensure_config_pr_webhook failed");
    }
    Ok(())
}

/// Boot-time `/knowledge` pull as a DAG node — see
/// [`NodeKind::KnowledgePull`]. Unlike the `main.rs` periodic loop's
/// startup call, a failure here is *not* swallowed to debug level: the node
/// exists so a failed boot pull is visible on the dashboard rather than
/// only in the journal.
async fn run_knowledge_pull(coord: &Arc<Coordinator>) -> Result<()> {
    crate::workers::knowledge::pull(coord).await
}

/// 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(coord: &Arc<Coordinator>) -> Result<()> {
    crate::workers::wanted::pull(coord).await
}

/// Resolve the DAG's approval row the way this node's own `outcome` says.
///
/// Nothing is inspected: a template emits one of these per outcome, each edged to
/// accept only that one, so *which* node the scheduler let run already is the
/// answer. Best-effort — a resolution failure is logged inside
/// [`crate::actions::resolve_approval_dag`], never surfaced as a node failure,
/// since the work already happened and failing the tail would only misreport it.
async fn run_resolve_approval(
    coord: &Arc<Coordinator>,
    dag_id: Option<u64>,
    approval_id: i64,
    outcome: TerminalState,
) -> Result<()> {
    let reason = (outcome == TerminalState::Failed)
        .then(|| dag_id.and_then(|dag| coord.job_queue.first_error(dag)))
        .flatten();
    crate::actions::resolve_approval_dag(coord, approval_id, outcome, reason.as_deref()).await;
    Ok(())
}

/// Rerender the meta flake from whatever containers still exist on disk.
/// Idempotent — a no-op when nothing changed. Lives here because the destroy
/// tail is its only caller; it moved with `destroy` when that became a DAG.
async fn sync_meta_after_lifecycle(coord: &Coordinator) -> Result<()> {
    let agents = crate::lifecycle::agents_for_meta_listing().await?;
    crate::meta::sync_agents(&coord.hive_env(), &agents).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.
async fn run_destroy_container(coord: &Arc<Coordinator>, agent: &str) -> Result<()> {
    crate::lifecycle::destroy(agent).await?;
    coord.unregister_agent(agent);
    let runtime = crate::paths::agent_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(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) = crate::priv_client::delete_agent_subvolume(agent).await {
        tracing::warn!(error = ?e, %agent, "purge: delete state subvolume failed");
    }
    // A malformed name can't have a persistent state tree (the state dir is
    // only ever created under a validated Ident), so its removal is a no-op —
    // skip the state-dir sweep and just clear the applied dir.
    let state_dir = hive_types::Ident::parse(agent)
        .ok()
        .map(|id| crate::paths::agent_state_dir(&id));
    for dir in state_dir
        .into_iter()
        .chain([crate::paths::applied_dir(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 meta sync 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(coord: &Arc<Coordinator>, agent: &str, purge: bool) {
    // Meta flake: drop the agent's input + nixosConfiguration so a future spawn
    // under the same name re-seeds cleanly, and so the meta lock doesn't
    // reference a vanished applied repo.
    if let Err(e) = sync_meta_after_lifecycle(coord).await {
        tracing::warn!(error = ?e, %agent, "meta sync after destroy failed");
    }
    let _ = coord.approvals.fail_pending_for_agent(
        agent,
        if purge {
            "agent purged"
        } else {
            "agent destroyed"
        },
    );
    // 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");
    }
    crate::swarm_notices::notify(
        "core",
        Some(format!("destroyed:{agent}")),
        format!("agent '{agent}' destroyed"),
        None,
    )
    .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;
    // Re-emit the schedules snapshot: the rescan above refreshed the live
    // roster, so any schedule that still targets the just-destroyed agent now
    // drops that ghost column live (no page reload needed).
    coord.emit_schedules_snapshot();
    // Update tmpfiles.d to remove the destroyed agent's dirs from the boot-time
    // pre-creation list. Best-effort: failure is logged only.
    tokio::spawn(crate::lifecycle::sync_tmpfiles());
}

/// 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(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);
    crate::swarm_notices::notify("core", Some(format!("rebuilt:{agent}")), summary, None).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 meta preamble: runtime-dir prep, an idempotent meta
/// `sync_agents`, and the optional per-agent relock. Runs under the deploy
/// window (`Resource::MetaWindow`, 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>, name: &str, relock: bool) -> 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 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(())
}

/// 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(name: &str, id: NodeId) -> Result<()> {
    // 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 {
        let flake_ref = format!("{}#{name}", crate::paths::meta_root().display());
        crate::lifecycle::prebuild_toplevel(name, &flake_ref, Some(id.get())).await?;
    }
    Ok(())
}

/// 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>, 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 result = crate::lifecycle::swap_update(name, &hive, &paths, Some(id.get())).await;
    // On success the Ok-only bookkeeping tail (rev marker, forge/matrix
    // 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/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_rebuild_bookkeeping(coord: &Arc<Coordinator>, name: &str) -> Result<()> {
    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 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.
    // 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(())
}

/// First-spawn pre-create provisioning: proposed/applied repos, state
/// subvolume, and the meta `sync_agents` registration. Runs under the
/// deploy window (it declares `Resource::MetaWindow`) so its commit can't
/// land inside another node's staged deploy window.
async fn run_provision(coord: &Arc<Coordinator>, name: &str) -> Result<()> {
    let agent_dir = crate::paths::agent_runtime_dir(name);
    let hive = coord.hive_env();
    let paths = Coordinator::agent_paths(name, agent_dir);
    crate::lifecycle::provision_container(name, &hive, &paths).await?;
    Ok(())
}

/// `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(name: &str) -> Result<()> {
    crate::lifecycle::create_only(name).await?;
    Ok(())
}

/// 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.
/// Returns the agents whose rebuild subgraphs the caller should grow into this
/// node, rather than declaring them here — the declaration has to happen outside
/// any `.await` (see [`run_node`]).
///
/// Only the agent list: *which* rebuild flavour to grow is a pure function of
/// `sweep`, which the caller passed in, so returning it too would be a round
/// trip rather than a decision.
async fn run_meta_lock(
    coord: &Arc<Coordinator>,
    sweep: bool,
    fanout: Option<Vec<String>>,
    inputs: &[String],
) -> Result<Vec<String>> {
    if sweep {
        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. The caller grows them
        // with the graceful flavour.
        return Ok(fanout.unwrap_or_default());
    }
    let _progress = coord.meta_update_guard();
    crate::meta::lock_update(inputs).await?;
    // Lock file changed — meta-inputs panel re-renders.
    crate::dashboard::emit_meta_inputs_snapshot(coord);
    let cascade = match fanout {
        Some(list) => validate_agent_names(list),
        None => meta_update_cascade_agents(inputs).await,
    };
    // Pull each cascade agent's own input too — an agent's config-repo main
    // is trusted, so a meta-input bump is a reasonable place to also catch
    // it up. Without this, a meta-input bump rebuilds every affected agent
    // against whatever `applied/<name>` already happened to be locked to —
    // normally current, but silently stale forever if a past deploy failed
    // and nothing since retried it. Relocks against each agent's *declared*
    // input (the forge URL, not the local `applied/<name>` mirror
    // `prepare_deploy` uses for a reviewed deploy) — this path has no PR to
    // review, so there's nothing to gate. One combined call, not per-agent:
    // simpler, at the cost of one unreachable/broken agent repo failing the
    // whole cascade relock rather than just that agent.
    if !cascade.is_empty() {
        let agent_inputs: Vec<String> =
            cascade.iter().map(|name| format!("agent-{name}")).collect();
        crate::meta::lock_update(&agent_inputs).await?;
        // The relock above is a one-shot effect on *this* rebuild only —
        // `applied/<name>` itself hasn't moved, so the next `relock = true`
        // rebuild trigger (the boot sweep, most notably) re-locks against
        // `applied/<name>` and reverts straight back to whatever it was
        // stuck on. Fast-forward each cascade agent's own applied mirror
        // too so a later rebuild through a different trigger doesn't undo
        // this one. Best-effort per agent (unlike the relock above): one
        // agent's forge repo being briefly unreachable here shouldn't be
        // fatal to the cascade that already relocked its input above.
        for name in &cascade {
            if !crate::forge::fast_forward_applied_main(name).await {
                tracing::warn!(
                    %name,
                    "meta-update cascade: applied/main fast-forward skipped or failed"
                );
            }
        }
    }
    // 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).
    Ok(cascade)
}

/// 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(coord: &Arc<Coordinator>, name: &str) -> Result<Option<NodeKind>> {
    let running = crate::lifecycle::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(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);
    let token = crate::lifecycle::converge_start_preamble(name, &hive, &paths).await?;
    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(())
}

/// 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>, name: &str) -> Result<()> {
    // See `run_start`: no node-local guard — `Stop` reports `Stopping` from its
    // own kind now.
    crate::lifecycle::kill(name).await?;
    coord.unregister_agent(name);
    crate::swarm_notices::notify(
        "core",
        Some(format!("killed:{name}")),
        format!("agent '{name}' killed"),
        None,
    )
    .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(coord: &Arc<Coordinator>, name: &str) -> Result<()> {
    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");
        }
        crate::lifecycle::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) -> Result<()> {
    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(())
}

/// 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(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);
    Coordinator::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) -> Result<()> {
    let deadline = std::time::Instant::now() + PAUSE_ACK_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(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);
    crate::lifecycle::write_dropins(name, &hive, &paths).await?;
    Ok(())
}

/// 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>,
    name: &str,
    payload: &super::model::PermPayload,
) -> Result<()> {
    use super::model::PermPayload;
    // Runs under the deploy window (it declares `Resource::MetaWindow`): 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(())
}

/// Apply the node's `(child, new_parent)` moves as one `META_LOCK`-fused
/// commit (`Coordinator::reparent_bulk_with_notify`, which already handles
/// both the single- and bulk-move case, sends the per-agent move
/// notifications, and rescans + diff-emits the container tree). Runs under
/// the deploy window (it declares `Resource::MetaWindow`), same reasoning as
/// `run_write_perm_file`: a topology commit landing inside another node's
/// staged deploy window would sweep the staged lock into its commit.
async fn run_reparent(
    coord: &Arc<Coordinator>,
    moves: &[(hive_types::Ident, Option<hive_types::Ident>)],
) -> Result<()> {
    let refs: Vec<(&str, Option<&str>)> = moves
        .iter()
        .map(|(child, parent)| {
            (
                child.as_str(),
                parent.as_ref().map(hive_types::Ident::as_str),
            )
        })
        .collect();
    coord
        .reparent_bulk_with_notify(&refs)
        .await
        .map_err(|e| anyhow::anyhow!(e))?;
    Ok(())
}

/// 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>, approval_id: i64, id: NodeId) -> Result<()> {
    crate::actions::run_deploy_merge_verify(coord, approval_id, Some(id.get())).await
}

/// 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>, approval_id: i64, id: NodeId) -> Result<()> {
    crate::actions::run_deploy_apply(coord, approval_id, Some(id.get())).await
}

/// 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>, approval_id: i64) -> Result<()> {
    crate::actions::run_finalize_deploy(coord, approval_id).await
}

/// 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>,
    dag_id: Option<u64>,
    agent: &str,
    approval_id: i64,
) -> Result<()> {
    crate::actions::run_deploy_tail(coord, dag_id, agent, approval_id).await;
    Ok(())
}

/// 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.
///
/// `inputs` is the caller-supplied flake-input-name list
/// (`RequestUpdateMetaInputs`'s `inputs` field, operator-approved but not
/// otherwise validated) — the `agent-<name>` branch parses agent names
/// straight out of it, so each is validated through [`hive_types::Ident`]
/// before it ever reaches a filesystem path or a forge URL built from a
/// cascade agent's name. The `touched_hyperhive` branch's names need no
/// such filter: they come from `lifecycle::list()`, already real
/// container names by construction, not parsed out of caller input.
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 = parse_agent_input_names(inputs);
    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
}

/// Parse `agent-<name>` input strings into validated agent names. Pure and
/// synchronous (no `lifecycle::list()` call) so the validation — the part
/// that actually matters for safety, since these names end up in
/// filesystem paths and forge URLs — is unit-testable without touching
/// live container state. Delegates the actual filtering to
/// [`validate_agent_names`] after stripping the `agent-` prefix.
fn parse_agent_input_names(inputs: &[String]) -> Vec<String> {
    let candidates = inputs
        .iter()
        .filter_map(|i| i.strip_prefix("agent-"))
        .map(|rest| rest.split('/').next().unwrap_or(rest).to_owned());
    validate_agent_names(candidates)
}

/// Drop any name that isn't a well-formed [`hive_types::Ident`] — the same
/// validating parser every other agent-name wire field in this codebase
/// goes through, so a name that reaches
/// [`fast_forward_applied_main`](crate::forge::fast_forward_applied_main) or
/// a `git+http://.../agent-configs/<name>.git` URL built off it has already
/// passed the same bar as one that arrived over a socket. Not just a
/// helper for [`parse_agent_input_names`]: `run_meta_lock`'s `fanout`
/// parameter carries pre-computed agent names too, and while today's only
/// caller of that branch (the boot sweep, `sweep = true`) returns before
/// ever reaching the filesystem/git code below, nothing in the type
/// enforces that pairing — so this same filter also guards the
/// `Some(list)` arm, closing the gap for good rather than relying on it
/// staying incidental.
fn validate_agent_names(names: impl IntoIterator<Item = String>) -> Vec<String> {
    names
        .into_iter()
        .filter(|name| {
            let valid = hive_types::Ident::parse(name).is_ok();
            if !valid {
                tracing::warn!(%name, "meta-update cascade: dropping malformed agent name");
            }
            valid
        })
        .collect()
}

#[cfg(test)]
mod cascade_agent_tests {
    use super::{parse_agent_input_names, validate_agent_names};

    #[test]
    fn well_formed_agent_inputs_pass_through() {
        let inputs = vec!["agent-iris".to_owned(), "agent-atlas".to_owned()];
        assert_eq!(parse_agent_input_names(&inputs), vec!["iris", "atlas"]);
    }

    #[test]
    fn path_traversal_in_an_agent_input_is_dropped() {
        let inputs = vec!["agent-../../etc".to_owned(), "agent-iris".to_owned()];
        assert_eq!(parse_agent_input_names(&inputs), vec!["iris"]);
    }

    #[test]
    fn non_agent_inputs_are_ignored_not_misparsed() {
        let inputs = vec!["hyperhive".to_owned(), "nixpkgs".to_owned()];
        assert!(parse_agent_input_names(&inputs).is_empty());
    }

    #[test]
    fn oversized_or_uppercase_names_are_dropped() {
        let too_long = format!("agent-{}", "a".repeat(64));
        let inputs = vec![too_long, "agent-Iris".to_owned(), "agent-iris".to_owned()];
        assert_eq!(parse_agent_input_names(&inputs), vec!["iris"]);
    }

    #[test]
    fn fanout_supplied_names_are_validated_too() {
        // `run_meta_lock`'s `Some(list)` arm feeds `fanout` straight into
        // this filter — pin that a bogus name in that pre-computed list
        // gets dropped exactly like a parsed `agent-<name>` input would.
        let names = vec![
            "iris".to_owned(),
            "../../etc".to_owned(),
            "atlas".to_owned(),
        ];
        assert_eq!(validate_agent_names(names), vec!["iris", "atlas"]);
    }
}