553ed8376c
The hive reads the agent-archetype key when it provisions and rebuilds an agent and keeps it in swarm-agent-archetypes.json. For an archetype agent the meta flake has no agent-<name> input and mkAgent imports hyperhive.hyperhiveArchetypes.<name>.module; provisioning, rebuilds, the meta-input cascade, a deploy's rev and the startup migration skip the config repos. The harness gets HIVE_ARCHETYPE and the system prompt says the agent has no config repo. The create form sends the picked archetype. Adds the android-dev built-in. Refs #5386
1003 lines
46 KiB
Rust
1003 lines
46 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, Mutex, OnceLock};
|
|
|
|
use anyhow::{Context as _, Result};
|
|
|
|
use hive_jobq::NodeId;
|
|
|
|
use super::model::NodeKind;
|
|
use crate::coordinator::Coordinator;
|
|
use crate::power::{ReconcileAction, reconcile_action};
|
|
use crate::stats::sweep_health::{self, SweepHealth};
|
|
|
|
/// Max time `Drain` waits for the harness to run its stop-checkpoint
|
|
/// turn before falling back to the hard stop. Generous — a checkpoint
|
|
/// turn can take a while — but bounded so a wedged agent never blocks
|
|
/// the stop indefinitely.
|
|
pub const GRACEFUL_STOP_TIMEOUT: std::time::Duration = std::time::Duration::from_mins(3);
|
|
|
|
/// Max time `PauseDrain` waits for the harness to report
|
|
/// `PauseAcknowledged` before giving up and resolving anyway (the
|
|
/// marker itself — not this node — is what actually gates the turn
|
|
/// loop, so "giving up" costs nothing but a slightly-late dashboard
|
|
/// badge). Same ceiling as `GRACEFUL_STOP_TIMEOUT` — no reason for the
|
|
/// two to diverge yet, but aliased under its own name so a future
|
|
/// change to one doesn't silently retune the other.
|
|
const PAUSE_ACK_TIMEOUT: std::time::Duration = GRACEFUL_STOP_TIMEOUT;
|
|
|
|
/// 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`), 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 { .. } => {
|
|
run_destroy_bookkeeping(coord, agent).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::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::AgentWindow { .. } => Ok(()),
|
|
NodeKind::ForgeSweep => run_forge_sweep().await,
|
|
NodeKind::MatrixSweep => run_matrix_sweep().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(())
|
|
}
|
|
|
|
/// Matrix user/space sweep as a DAG node — see [`NodeKind::MatrixSweep`].
|
|
/// Every pass reports here, whoever submitted it: as the node's own error on
|
|
/// the dashboard, and into the debounced banner.
|
|
async fn run_matrix_sweep() -> Result<()> {
|
|
let ok = tokio::time::timeout(MATRIX_SWEEP_DEADLINE, crate::matrix::ensure_all())
|
|
.await
|
|
.unwrap_or_else(|_| {
|
|
tracing::warn!(deadline = ?MATRIX_SWEEP_DEADLINE, "matrix: sweep timed out");
|
|
false
|
|
});
|
|
let mut health = matrix_sweep_health()
|
|
.lock()
|
|
.unwrap_or_else(std::sync::PoisonError::into_inner);
|
|
if ok {
|
|
health.record_ok();
|
|
Ok(())
|
|
} else {
|
|
health.record_err(matrix_sweep_banner);
|
|
anyhow::bail!("matrix ensure_all: one or more agents failed sync (see logs)")
|
|
}
|
|
}
|
|
|
|
/// A pass still running holds [`Resource::MatrixSweep`] against every later
|
|
/// one. Each HTTP call has its own 10 s timeout, but the call count grows with
|
|
/// the agent count and `lifecycle::list()` has none. A pass dropped here leaves
|
|
/// at worst a created but unpersisted room, which the next pass adopts by name.
|
|
///
|
|
/// [`Resource::MatrixSweep`]: super::resource::Resource::MatrixSweep
|
|
const MATRIX_SWEEP_DEADLINE: std::time::Duration = std::time::Duration::from_mins(10);
|
|
|
|
/// Banner for a matrix sweep that keeps failing. Two misses in a row, so one
|
|
/// bad pass (homeserver mid-restart, a transient HTTP blip) doesn't flap the
|
|
/// dashboard; cleared by the next clean pass.
|
|
fn matrix_sweep_health() -> &'static Mutex<SweepHealth> {
|
|
static HEALTH: OnceLock<Mutex<SweepHealth>> = OnceLock::new();
|
|
HEALTH.get_or_init(|| Mutex::new(SweepHealth::new("matrix_ensure_all", "warn", 2)))
|
|
}
|
|
|
|
fn matrix_sweep_banner(ctx: sweep_health::SweepFailure) -> String {
|
|
let age = ctx.since_last_ok.map_or_else(
|
|
|| "no success this session".to_owned(),
|
|
|d| format!("last ok {} ago", sweep_health::fmt_age(d)),
|
|
);
|
|
format!(
|
|
"matrix user/space sweep failing ({} consecutive, {age}) \
|
|
— some agents may be missing matrix accounts, space membership, \
|
|
or chat-room invites",
|
|
ctx.consecutive
|
|
)
|
|
}
|
|
|
|
/// `/knowledge` pull as a DAG node — see [`NodeKind::KnowledgePull`]. Every
|
|
/// pass reports here, whoever submitted it: as the node's own error on the
|
|
/// dashboard, and into the debounced banner.
|
|
async fn run_knowledge_pull(coord: &Arc<Coordinator>) -> Result<()> {
|
|
let result = tokio::time::timeout(
|
|
KNOWLEDGE_PULL_DEADLINE,
|
|
crate::workers::knowledge::pull(coord),
|
|
)
|
|
.await
|
|
.unwrap_or_else(|_| {
|
|
Err(anyhow::anyhow!(
|
|
"knowledge pull timed out after {KNOWLEDGE_PULL_DEADLINE:?}"
|
|
))
|
|
});
|
|
let mut health = knowledge_pull_health()
|
|
.lock()
|
|
.unwrap_or_else(std::sync::PoisonError::into_inner);
|
|
match &result {
|
|
Ok(()) => health.record_ok(),
|
|
Err(e) => {
|
|
let err = format!("{e:#}");
|
|
health.record_err(|ctx| {
|
|
let age = ctx.since_last_ok.map_or_else(
|
|
|| "no success this session".to_owned(),
|
|
|d| format!("last ok {} ago", sweep_health::fmt_age(d)),
|
|
);
|
|
format!(
|
|
"knowledge repo pull failing ({} consecutive, {age}) \
|
|
— /knowledge is stale until it recovers: {err}",
|
|
ctx.consecutive
|
|
)
|
|
});
|
|
}
|
|
}
|
|
result
|
|
}
|
|
|
|
/// A pull still running holds [`Resource::KnowledgeTree`] against every later
|
|
/// one. The `reset` / `clean` / `pull` git children are `kill_on_drop`, so the
|
|
/// deadline kills them.
|
|
///
|
|
/// [`Resource::KnowledgeTree`]: super::resource::Resource::KnowledgeTree
|
|
const KNOWLEDGE_PULL_DEADLINE: std::time::Duration = std::time::Duration::from_mins(5);
|
|
|
|
/// Banner for a `/knowledge` pull that keeps failing: three misses in a row,
|
|
/// cleared by the next successful pull.
|
|
fn knowledge_pull_health() -> &'static Mutex<SweepHealth> {
|
|
static HEALTH: OnceLock<Mutex<SweepHealth>> = OnceLock::new();
|
|
HEALTH.get_or_init(|| Mutex::new(SweepHealth::new("knowledge_pull", "warn", 3)))
|
|
}
|
|
|
|
/// Boot-time swarm wanted-state pull as a DAG node — see
|
|
/// [`NodeKind::WantedPull`]. "The controller has declared nothing" is a
|
|
/// successful outcome the worker logs, so an error reaching here means the
|
|
/// read itself failed and this hive is running blind to its declaration —
|
|
/// which is exactly the state worth seeing on the dashboard.
|
|
async fn run_wanted_pull(coord: &Arc<Coordinator>) -> Result<()> {
|
|
crate::workers::wanted::pull(coord).await
|
|
}
|
|
|
|
/// 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. A destroy that leaves the
|
|
/// container in place fails this node before un-registration, and its failure
|
|
/// is what keeps the `after_ok` purge and bookkeeping from running.
|
|
async fn run_destroy_container(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) {
|
|
// 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");
|
|
}
|
|
// 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;
|
|
}
|
|
|
|
/// 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)?;
|
|
// A failed read keeps the archetype last stored, so the agent is built as
|
|
// it was.
|
|
if let Err(e) = crate::archetypes::refresh(name).await {
|
|
tracing::warn!(%name, error = ?e, "rebuild: reading the swarm's archetype failed; keeping the stored one");
|
|
}
|
|
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?;
|
|
// An archetype agent has no `agent-<name>` input to relock.
|
|
if relock && crate::archetypes::of(name)?.is_none() {
|
|
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 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 sync on every successful rebuild so the rebuild path is
|
|
// equivalent to the startup sweep: meta access, the `meta` remote and the
|
|
// `internal/docs` grant all recover without a hive-c0re restart. (No
|
|
// matrix step: the swarm mints an agent's matrix account, not this hive.)
|
|
crate::forge::sync_agent(name, crate::forge::core_token().as_deref()).await;
|
|
// No `kick_agent` here any more — reporting from `hive-agent`'s own
|
|
// boot instead avoids the duplicate-restart-message problem this used
|
|
// to cause: the container is still stopped at this point (`swap_update` requires it stopped and
|
|
// never restarts it — the DAG's tail `Reconcile`/`Start` does), so a
|
|
// host-side "you were rebuilt" broker message would just queue
|
|
// alongside `run_start`'s own kick for the same boot. `hive-agent` now
|
|
// self-announces once on its own startup instead — one message per
|
|
// actual boot, not one per DAG node that happens to run during it.
|
|
// Rescan so dashboards drop the "needs update" chip; lock bump →
|
|
// meta-inputs re-render.
|
|
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();
|
|
// Resolved before the lock bump so a failed container list aborts
|
|
// the update before anything is committed.
|
|
let cascade = match fanout {
|
|
Some(list) => validate_agent_names(list),
|
|
None => meta_update_cascade_agents(inputs).await?,
|
|
};
|
|
crate::meta::lock_update(inputs).await?;
|
|
// Lock file changed — meta-inputs panel re-renders.
|
|
crate::dashboard::emit_meta_inputs_snapshot(coord);
|
|
// 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.
|
|
// An archetype agent has no config repo and no `agent-<name>` input, so
|
|
// it is left out here; it is still rebuilt with the cascade.
|
|
let archetypes = crate::archetypes::read_at(&crate::archetypes::path())?;
|
|
let custom: Vec<String> = cascade
|
|
.iter()
|
|
.filter(|name| !archetypes.contains_key(*name))
|
|
.cloned()
|
|
.collect();
|
|
catch_up_cascade_agents(
|
|
&custom,
|
|
crate::forge::fast_forward_applied_main,
|
|
|agent_inputs| async move { crate::meta::lock_update(&agent_inputs).await },
|
|
)
|
|
.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).
|
|
Ok(cascade)
|
|
}
|
|
|
|
/// Fast-forward each cascade agent's `applied/<name>/main` to its forge
|
|
/// config main, then relock every `agent-<name>` input in one call.
|
|
///
|
|
/// The declared `agent-<name>` input is the `applied/<name>` mirror itself,
|
|
/// and the cascade's rebuilds (`relock = false`) build whatever this relock
|
|
/// pins, so the mirror has to move first. The fast-forward is best-effort
|
|
/// per agent: a failed one leaves that agent relocked against whatever its
|
|
/// mirror already has. The relock is one combined call, so one broken input
|
|
/// fails it for the whole cascade.
|
|
///
|
|
/// The fast-forward and the relock are passed in so the order is testable
|
|
/// without nix or a forge.
|
|
async fn catch_up_cascade_agents<'a, FF, Relock>(
|
|
cascade: &'a [String],
|
|
mut fast_forward: impl FnMut(&'a str) -> FF,
|
|
relock: impl FnOnce(Vec<String>) -> Relock,
|
|
) -> Result<()>
|
|
where
|
|
FF: std::future::Future<Output = bool>,
|
|
Relock: std::future::Future<Output = Result<()>>,
|
|
{
|
|
if cascade.is_empty() {
|
|
return Ok(());
|
|
}
|
|
for name in cascade {
|
|
if !fast_forward(name).await {
|
|
tracing::warn!(
|
|
%name,
|
|
"meta-update cascade: applied/main fast-forward skipped or failed"
|
|
);
|
|
}
|
|
}
|
|
relock(cascade.iter().map(|name| format!("agent-{name}")).collect()).await
|
|
}
|
|
|
|
/// 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)?;
|
|
// No `kick_agent` here any more — `hive-agent` pushes its own "you
|
|
// were just (re)started" todo on boot now, so this node doesn't need
|
|
// to tell it. See the matching comment in `run_rebuild_bookkeeping`.
|
|
coord.rescan_containers_and_emit().await;
|
|
Ok(())
|
|
}
|
|
|
|
/// Mechanical container stop — the sub-step a `Reconcile` planner fans
|
|
/// out when it observes `wanted = Offline` and the container up.
|
|
async fn run_stop(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(())
|
|
}
|
|
|
|
/// 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.
|
|
/// Sorted by name — the parent field the old depth sort keyed on is
|
|
/// gone, and with every agent a root that sort already reduced to
|
|
/// this.
|
|
///
|
|
/// `inputs` is the caller-supplied flake-input-name list (the dashboard's
|
|
/// `POST /api/meta-update` form field, operator-supplied 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.
|
|
///
|
|
/// # Errors
|
|
///
|
|
/// The every-container branch could not list the containers.
|
|
pub async fn meta_update_cascade_agents(inputs: &[String]) -> Result<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::agent_names(crate::lifecycle::list().await)
|
|
.context("listing containers for the meta-update cascade")?
|
|
} else {
|
|
touched_agents
|
|
};
|
|
names.sort();
|
|
Ok(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
|
|
/// an `agent-configs/<name>` forge 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::{catch_up_cascade_agents, parse_agent_input_names, validate_agent_names};
|
|
use std::cell::RefCell;
|
|
use std::collections::BTreeMap;
|
|
|
|
/// Runs [`catch_up_cascade_agents`] over `cascade` against a model of
|
|
/// the applied mirrors: every mirror starts at `"old"`, a fast-forward
|
|
/// moves it to `"new"` unless the agent is in `failing`. Returns the
|
|
/// mirror head the relock saw for each input it was given.
|
|
async fn relocked_heads(cascade: &[&str], failing: &[&str]) -> Vec<(String, &'static str)> {
|
|
let cascade: Vec<String> = cascade.iter().map(|&n| n.to_owned()).collect();
|
|
let heads = RefCell::new(
|
|
cascade
|
|
.iter()
|
|
.map(|n| (n.clone(), "old"))
|
|
.collect::<BTreeMap<_, _>>(),
|
|
);
|
|
let relocked = RefCell::new(Vec::new());
|
|
let fast_forward = |name: &str| {
|
|
let ok = !failing.contains(&name);
|
|
if ok {
|
|
heads.borrow_mut().insert(name.to_owned(), "new");
|
|
}
|
|
async move { ok }
|
|
};
|
|
let relock = |inputs: Vec<String>| {
|
|
for input in inputs {
|
|
let name = input.strip_prefix("agent-").expect("agent input");
|
|
let head = heads.borrow()[name];
|
|
relocked.borrow_mut().push((input, head));
|
|
}
|
|
async { anyhow::Ok(()) }
|
|
};
|
|
catch_up_cascade_agents(&cascade, fast_forward, relock)
|
|
.await
|
|
.expect("catch up");
|
|
relocked.into_inner()
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn the_cascade_relock_pins_the_fast_forwarded_head() {
|
|
assert_eq!(
|
|
relocked_heads(&["iris", "atlas"], &[]).await,
|
|
vec![
|
|
("agent-iris".to_owned(), "new"),
|
|
("agent-atlas".to_owned(), "new"),
|
|
]
|
|
);
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn a_failed_fast_forward_still_relocks_that_agent() {
|
|
assert_eq!(
|
|
relocked_heads(&["iris", "atlas"], &["iris"]).await,
|
|
vec![
|
|
("agent-iris".to_owned(), "old"),
|
|
("agent-atlas".to_owned(), "new"),
|
|
]
|
|
);
|
|
}
|
|
|
|
/// A relock with no inputs is a bare `nix flake update`, which bumps
|
|
/// every meta input.
|
|
#[tokio::test]
|
|
async fn an_empty_cascade_moves_nothing_and_relocks_nothing() {
|
|
let fast_forwarded = std::cell::Cell::new(false);
|
|
let relocked = std::cell::Cell::new(false);
|
|
let outcome = catch_up_cascade_agents(
|
|
&[],
|
|
|_| {
|
|
fast_forwarded.set(true);
|
|
async { true }
|
|
},
|
|
|_| {
|
|
relocked.set(true);
|
|
async { anyhow::Ok(()) }
|
|
},
|
|
)
|
|
.await;
|
|
outcome.expect("an empty cascade is not an error");
|
|
assert!(!fast_forwarded.get(), "nothing to fast-forward");
|
|
assert!(!relocked.get(), "an empty relock would bump every input");
|
|
}
|
|
|
|
#[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"]);
|
|
}
|
|
}
|