b87eac0a61
append_subgraph is the multi-node/multi-agent generalisation of the single-node append_node, so the two in-DAG-growth channels collapse to one: the Reconcile planner now emits its mechanical Start/Stop as a single-node append_subgraph rooted on the reconcile node (stamping claim.agent on the NodeSpec, which append_node inherited implicitly). Removes NodeOutput.append_nodes + its scheduler drain loop and JobQueue::append_node. No behaviour change — a channel unification.
629 lines
28 KiB
Rust
629 lines
28 KiB
Rust
//! 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::model::{NodeKind, NodeSpec, State, Template};
|
||
use super::{Claim, TerminalDag};
|
||
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
|
||
/// [`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::Prebuild { relock } => run_prebuild(coord, claim, &ctx, *relock).await,
|
||
NodeKind::Swap => run_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::ApprovalDeploy => run_approval_deploy(coord, claim).await,
|
||
// A pure grouping anchor (boot root): no work, completes immediately so
|
||
// its child DAGs settle it and the boot tree resolves.
|
||
NodeKind::Noop => Ok(NodeOutput::default()),
|
||
NodeKind::SetWanted { up } => run_set_wanted(coord, claim, *up),
|
||
}
|
||
}
|
||
|
||
/// 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())
|
||
}
|
||
|
||
/// Out-of-band toplevel build while the container keeps serving: meta
|
||
/// sync + optional per-agent relock, then warm
|
||
/// `system.build.toplevel` so the later `Swap` hits cache and skips
|
||
/// straight to the profile-swap. 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. The meta sync +
|
||
/// dir prep still run unconditionally (the `Swap` depends on them).
|
||
async fn run_prebuild(
|
||
coord: &Arc<Coordinator>,
|
||
claim: &Claim,
|
||
ctx: &Ctx<'_>,
|
||
relock: bool,
|
||
) -> Result<NodeOutput> {
|
||
let name = &claim.agent;
|
||
// Prebuild 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. Both run under
|
||
// the deploy-window gate so they can never land inside another
|
||
// node's staged prepare→finalize window; the gate drops before the
|
||
// (long) toplevel build, which only reads the store.
|
||
{
|
||
let _window = crate::meta::exclusive().await;
|
||
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?;
|
||
}
|
||
}
|
||
// 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;
|
||
match &result {
|
||
Ok(()) => {
|
||
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);
|
||
}
|
||
Err(_) => {
|
||
// The `Rebuilt { ok: false }` manager event fires once per
|
||
// DAG from the terminal hook (any node may be the one that
|
||
// failed); here only refresh the observed state.
|
||
coord.rescan_containers_and_emit().await;
|
||
}
|
||
}
|
||
result.map(|()| NodeOutput::default())
|
||
}
|
||
|
||
/// First-spawn pre-create provisioning: proposed/applied repos, state
|
||
/// subvolume, and the meta `sync_agents` registration. Holds the
|
||
/// deploy-window gate for the commit so it can't land inside another
|
||
/// node's staged deploy window — same discipline as `Prebuild`, which
|
||
/// commits under the gate then drops it before the (store-only) build.
|
||
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");
|
||
let _window = crate::meta::exclusive().await;
|
||
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");
|
||
let _window = crate::meta::exclusive().await;
|
||
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");
|
||
{
|
||
let _window = crate::meta::exclusive().await;
|
||
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. The old
|
||
// `append_node` inherited the emitter's agent implicitly; `append_subgraph`
|
||
// carries it on the `NodeSpec`, so stamp `claim.agent` explicitly (same
|
||
// effect, one in-DAG-growth channel instead of two).
|
||
let sub = |kind| vec![vec![super::templates::node(name, kind, Vec::new())]];
|
||
let append_subgraph = match reconcile_action(wanted, running) {
|
||
ReconcileAction::Start => sub(NodeKind::Start),
|
||
ReconcileAction::Stop => sub(NodeKind::Stop),
|
||
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.
|
||
fn run_signal(coord: &Arc<Coordinator>, claim: &Claim, ctx: &Ctx<'_>) -> NodeOutput {
|
||
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;
|
||
ctx.step("writing + committing perm file");
|
||
// Deploy-window gate: 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).
|
||
let _window = crate::meta::exclusive().await;
|
||
match &claim.perm_payload {
|
||
Some(PermPayload::ToolGroups { groups }) => {
|
||
crate::meta::commit_tool_groups(name, groups)
|
||
.await
|
||
.with_context(|| format!("commit tool-groups for {name}"))?;
|
||
coord.emit_tool_groups_snapshot();
|
||
}
|
||
Some(PermPayload::Capabilities { caps }) => {
|
||
crate::meta::commit_capabilities(name, caps)
|
||
.await
|
||
.with_context(|| format!("commit capabilities for {name}"))?;
|
||
coord.emit_capabilities_snapshot();
|
||
}
|
||
Some(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();
|
||
}
|
||
}
|
||
None => anyhow::bail!(
|
||
"perm_change dag {} for {name} is missing perm_payload",
|
||
claim.dag_id
|
||
),
|
||
}
|
||
Ok(NodeOutput::default())
|
||
}
|
||
|
||
/// Opaque approval deploy pipeline: `ApplyCommit` and `MergeConfigPr`
|
||
/// both end in a container rebuild; branch on the approval row's kind
|
||
/// (the authoritative source). The two-phase prepare/finalize/abort
|
||
/// meta deploy — and the approval resolution — stay inside
|
||
/// `actions.rs` in v1 (design doc §9).
|
||
async fn run_approval_deploy(coord: &Arc<Coordinator>, claim: &Claim) -> Result<NodeOutput> {
|
||
let approval_id = claim
|
||
.approval_id
|
||
.with_context(|| format!("approval_deploy dag {} has no approval_id", claim.dag_id))?;
|
||
// Hold the deploy-window gate for the whole prepare→finalize span:
|
||
// `prepare_deploy` stages `flake.lock` uncommitted for the entire
|
||
// container build, and no other meta mutation may land inside that
|
||
// window (it would sweep the staged lock and neuter `abort_deploy`).
|
||
let _window = crate::meta::exclusive().await;
|
||
let kind = coord
|
||
.approvals
|
||
.get(approval_id)
|
||
.ok()
|
||
.flatten()
|
||
.map(|a| a.kind);
|
||
let result = if kind == Some(hive_sh4re::ApprovalKind::MergeConfigPr) {
|
||
crate::actions::run_approval_merge_config_pr(coord, Some(claim.dag_id), approval_id).await
|
||
} else {
|
||
crate::actions::run_approval_apply_commit(coord, Some(claim.dag_id), approval_id).await
|
||
};
|
||
result.map(|()| NodeOutput::default())
|
||
}
|
||
|
||
/// Terminal-roll-up hook, fired exactly once per DAG (node completion
|
||
/// and cancel paths alike — the queue buffers roll-ups and the
|
||
/// scheduler drains them). Three concerns:
|
||
/// - approval DAGs resolve their approval row (except the opaque
|
||
/// deploy pipeline, which resolves inside its node — unless it was
|
||
/// cancelled while still queued and the node never ran);
|
||
/// - non-approval rebuild-shaped DAGs emit exactly one `Rebuilt`
|
||
/// manager event: ok on `Done`, !ok on `Failed`, none on cancel;
|
||
/// - a cancelled power-op DAG reverts the `wanted` intent its submit
|
||
/// wrote: the operator's cancel means "don't do it", so intent
|
||
/// snaps back to the observed state instead of the flip executing
|
||
/// as a surprise side effect of some later reconcile.
|
||
pub(super) async fn on_dag_terminal(coord: &Arc<Coordinator>, terminal: &TerminalDag) {
|
||
if terminal.state == State::Cancelled
|
||
&& matches!(
|
||
terminal.template,
|
||
Template::Start
|
||
| Template::Stop
|
||
| Template::GracefulStop
|
||
| Template::Restart
|
||
| Template::GracefulRestart
|
||
)
|
||
{
|
||
// Revert each targeted agent's power intent to its observed state —
|
||
// the operator's cancel means "don't do it". Single-agent power-op
|
||
// DAGs have one agent here.
|
||
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");
|
||
}
|
||
}
|
||
}
|
||
if terminal.approval_id.is_some() {
|
||
crate::actions::resolve_approval_dag(coord, terminal).await;
|
||
return;
|
||
}
|
||
if matches!(terminal.template, Template::Rebuild | Template::PermChange) {
|
||
// Rebuild / PermChange are single-agent; emit one `Rebuilt` per
|
||
// targeted agent (exactly one today).
|
||
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,
|
||
}),
|
||
_ => {}
|
||
}
|
||
}
|
||
}
|
||
}
|
||
|
||
/// 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
|
||
}
|