Compare commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
ccf9951e5d | ||
|
|
bfae9aa51a | ||
|
|
b004ba3dc5 | ||
|
|
7519d9b904 | ||
|
|
d4eb62434d | ||
|
|
38ccef987e | ||
|
|
93c7454bf5 |
10 changed files with 418 additions and 57 deletions
|
|
@ -17,9 +17,9 @@ pub use pr_merge::{
|
|||
};
|
||||
pub use reconcile::{reconcile_config_apply, reconcile_config_status};
|
||||
pub use repos::{
|
||||
create_agent_repo, ensure_config_repo, ensure_knowledge_repo, ensure_meta_remote, ensure_repo,
|
||||
ensure_shared_docs_repo, fetch_config_main_into_applied, meta_read_access, push_config,
|
||||
push_meta, shared_docs_access,
|
||||
clone_config_into_proposed, create_agent_repo, ensure_config_repo, ensure_knowledge_repo,
|
||||
ensure_meta_remote, ensure_repo, ensure_shared_docs_repo, fetch_config_main_into_applied,
|
||||
meta_read_access, push_config, push_meta, shared_docs_access,
|
||||
};
|
||||
pub use users::{core_token, ensure_user_for, provision_user_token};
|
||||
|
||||
|
|
|
|||
|
|
@ -593,6 +593,68 @@ pub async fn fetch_config_main_into_applied(name: &str) -> bool {
|
|||
true
|
||||
}
|
||||
|
||||
/// Seed an agent's `proposed` repo from `agent-configs/<name>` on the forge.
|
||||
///
|
||||
/// When the swarm creates an agent it writes that agent's config to the forge
|
||||
/// before any hive is told to deploy it, so the hive's job here is to **take**
|
||||
/// that config rather than author a second one beside it. The local template
|
||||
/// is the fallback, not the default.
|
||||
///
|
||||
/// Best-effort like [`fetch_config_main_into_applied`], and `false` is an
|
||||
/// ordinary answer rather than an error: the forge is absent, the core token
|
||||
/// is not minted yet, or `agent-configs/<name>` has no `main` — the last of
|
||||
/// which is the *normal* case for the hive-level create flow, where the repo
|
||||
/// is only created after the first spawn (`actions::forge_after_first_spawn`).
|
||||
///
|
||||
/// `clone` rather than `init` + `fetch`: git removes a directory it created
|
||||
/// when the clone fails, and cloning into an existing empty dir leaves no
|
||||
/// `.git` behind either — so the caller's "does this repo already exist"
|
||||
/// check reads exactly the same before and after a failed attempt.
|
||||
pub async fn clone_config_into_proposed(dir: &Path, name: &str) -> bool {
|
||||
if !is_present().await {
|
||||
return false;
|
||||
}
|
||||
let Some(token) = core_token() else {
|
||||
return false;
|
||||
};
|
||||
let url = forge_git_url(&format!("{CONFIG_ORG}/{name}"));
|
||||
// `--branch main` fails outright against a repo with no commits, so an
|
||||
// empty repo reads as "nothing to take" instead of cloning to an unborn
|
||||
// HEAD that would then look like a seeded checkout.
|
||||
let out = crate::lifecycle::git_command_authed(&core_auth_header(&token))
|
||||
.args([
|
||||
"clone",
|
||||
"--branch",
|
||||
"main",
|
||||
&url,
|
||||
&dir.display().to_string(),
|
||||
])
|
||||
.output()
|
||||
.await;
|
||||
match out {
|
||||
Ok(o) if o.status.success() => {
|
||||
tracing::info!(%name, "forge: seeded proposed repo from agent-configs main");
|
||||
true
|
||||
}
|
||||
Ok(o) => {
|
||||
// `info`, not `warn`: on the hive-level create flow every first
|
||||
// spawn lands here and nothing is wrong. Logged at all because
|
||||
// "where did this agent's config come from" is worth answering
|
||||
// from the journal, and proposed is seeded exactly once.
|
||||
tracing::info!(
|
||||
%name,
|
||||
stderr = %String::from_utf8_lossy(&o.stderr).trim(),
|
||||
"forge: no config to clone for this agent; seeding the template"
|
||||
);
|
||||
false
|
||||
}
|
||||
Err(e) => {
|
||||
tracing::warn!(%name, error = ?e, "forge: proposed clone failed to run");
|
||||
false
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// Run a single `git push <url> <refspec>` in the applied repo `dir` and
|
||||
/// return the raw output for the caller to classify. Split out so
|
||||
/// [`push_config`] can push tags and `main` as independent pushes.
|
||||
|
|
|
|||
|
|
@ -452,6 +452,15 @@ pub fn approval_deploy(builder: &JobBuilder, agent: &str, approval_id: i64) {
|
|||
/// `AfterAny` onto `Provision` — the DAG's only other group-root, so its roll-up
|
||||
/// already carries the whole cascade.
|
||||
pub fn spawn(builder: &JobBuilder, agent: &str, approval_id: i64) {
|
||||
let provision = spawn_nodes(builder, agent);
|
||||
resolve_approval_tails(builder, approval_id, provision);
|
||||
}
|
||||
|
||||
/// The spawn subgraph with no tail, returning its group root.
|
||||
///
|
||||
/// Split out for the same reason [`rebuild_nodes`] is: two callers want the
|
||||
/// same four nodes and disagree only about what closes them.
|
||||
pub(crate) fn spawn_nodes<'a>(builder: &'a JobBuilder, agent: &str) -> Handle<'a> {
|
||||
let a = || agent.to_owned();
|
||||
let provision = builder
|
||||
.node(NodeKind::Provision { agent: a() })
|
||||
|
|
@ -470,8 +479,20 @@ pub fn spawn(builder: &JobBuilder, agent: &str, approval_id: i64) {
|
|||
.needs(Resource::Agent(a()))
|
||||
.part_of(create)
|
||||
.after_ok(dropin);
|
||||
provision
|
||||
}
|
||||
|
||||
resolve_approval_tails(builder, approval_id, provision);
|
||||
/// First deploy of an agent this hive has never seen, asked for by the swarm.
|
||||
///
|
||||
/// [`spawn`] without the approval tail, and the absence is the point rather
|
||||
/// than an omission: that flow exists because an operator used to approve the
|
||||
/// spawn *at the hive*. When the swarm asks, the operator has already clicked
|
||||
/// create at swarm level — the deploy request carries that authorisation, and a
|
||||
/// second gate here would be asking the same person the same question twice.
|
||||
///
|
||||
/// Returns the group root so a caller can wait on the whole subtree.
|
||||
pub fn first_deploy(builder: &JobBuilder, agent: &str) -> Vec<hive_jobq::NodeGuid> {
|
||||
vec![spawn_nodes(builder, agent).guid()]
|
||||
}
|
||||
|
||||
/// Teardown: `Stop` → `DestroyContainer` → (`PurgeState`) → `DestroyBookkeeping`.
|
||||
|
|
|
|||
|
|
@ -14,12 +14,18 @@ use super::git::{
|
|||
git, git_command, git_commit, git_read_tree_reset, git_rev_parse, git_root_commit, git_tag,
|
||||
};
|
||||
|
||||
/// Initialize an agent's config repo. Seeds two tracked files:
|
||||
/// `agent.nix` (the agent's own module) and `flake.nix` (the
|
||||
/// boilerplate that lets the meta flake import this repo as an input —
|
||||
/// meta locks at a specific sha and reads `nixosModules.default`, so
|
||||
/// `flake.nix` must be in the commit). `flake.nix` isn't meant to be
|
||||
/// edited, but it's tracked so it can be read.
|
||||
/// Initialize an agent's config repo, preferring the one the forge
|
||||
/// already has. `agent-configs/<name>` is authored by whoever created
|
||||
/// the agent — at swarm level that is the controller, which writes the
|
||||
/// config before any hive is told to deploy it — so this **clones**
|
||||
/// when there is something to clone and seeds a template only when
|
||||
/// there is not. Two authors for one file is the failure mode being
|
||||
/// avoided: whichever wrote second would win, and neither knows about
|
||||
/// the other.
|
||||
///
|
||||
/// The template ([`seed_template`]) is what the hive-level create flow
|
||||
/// still lands on, since that flow creates the forge repo *after* the
|
||||
/// first spawn.
|
||||
///
|
||||
/// **Seeding is the whole of hive-c0re's write.** Changes to the repo
|
||||
/// arrive as PRs from a clone, via the forge, like any other code
|
||||
|
|
@ -34,19 +40,9 @@ pub async fn setup_proposed(proposed_dir: &Path, name: &str) -> Result<()> {
|
|||
if fresh {
|
||||
std::fs::create_dir_all(proposed_dir)
|
||||
.with_context(|| format!("create {}", proposed_dir.display()))?;
|
||||
let agent_path = proposed_dir.join("agent.nix");
|
||||
if !agent_path.exists() {
|
||||
std::fs::write(&agent_path, initial_agent_nix(name))
|
||||
.with_context(|| format!("write {}", agent_path.display()))?;
|
||||
if !crate::forge::clone_config_into_proposed(proposed_dir, name).await {
|
||||
seed_template(proposed_dir, name).await?;
|
||||
}
|
||||
let flake_path = proposed_dir.join("flake.nix");
|
||||
if !flake_path.exists() {
|
||||
std::fs::write(&flake_path, initial_flake_nix())
|
||||
.with_context(|| format!("write {}", flake_path.display()))?;
|
||||
}
|
||||
git(proposed_dir, &["init", "--initial-branch=main"]).await?;
|
||||
git(proposed_dir, &["add", "agent.nix", "flake.nix"]).await?;
|
||||
git_commit(proposed_dir, "hive-c0re init").await?;
|
||||
}
|
||||
// Idempotently wire the `applied` remote — purely for the
|
||||
// manager's ergonomics. The URL is the path inside the manager
|
||||
|
|
@ -56,6 +52,30 @@ pub async fn setup_proposed(proposed_dir: &Path, name: &str) -> Result<()> {
|
|||
ensure_applied_remote(proposed_dir, name).await
|
||||
}
|
||||
|
||||
/// Author the initial config in place: `agent.nix` (the agent's own module)
|
||||
/// and `flake.nix` (the boilerplate that lets the meta flake import this repo
|
||||
/// as an input — meta locks at a specific sha and reads
|
||||
/// `nixosModules.default`, so `flake.nix` must be in the commit). `flake.nix`
|
||||
/// isn't meant to be edited, but it's tracked so it can be read.
|
||||
///
|
||||
/// The fallback half of [`setup_proposed`]: reached only when the forge has
|
||||
/// no config for this agent to take.
|
||||
async fn seed_template(proposed_dir: &Path, name: &str) -> Result<()> {
|
||||
let agent_path = proposed_dir.join("agent.nix");
|
||||
if !agent_path.exists() {
|
||||
std::fs::write(&agent_path, initial_agent_nix(name))
|
||||
.with_context(|| format!("write {}", agent_path.display()))?;
|
||||
}
|
||||
let flake_path = proposed_dir.join("flake.nix");
|
||||
if !flake_path.exists() {
|
||||
std::fs::write(&flake_path, initial_flake_nix())
|
||||
.with_context(|| format!("write {}", flake_path.display()))?;
|
||||
}
|
||||
git(proposed_dir, &["init", "--initial-branch=main"]).await?;
|
||||
git(proposed_dir, &["add", "agent.nix", "flake.nix"]).await?;
|
||||
git_commit(proposed_dir, "hive-c0re init").await
|
||||
}
|
||||
|
||||
async fn ensure_applied_remote(proposed_dir: &Path, name: &str) -> Result<()> {
|
||||
let want = format!("/applied/{name}/.git");
|
||||
let existing = git_command()
|
||||
|
|
|
|||
|
|
@ -6,6 +6,10 @@ use super::*;
|
|||
/// Regression test: `setup_proposed` must seed both agent.nix and flake.nix
|
||||
/// in the initial commit. Before commit 5b5a93e flake.nix was missing from
|
||||
/// the scaffold, requiring manual creation (seen with the damocles agent).
|
||||
///
|
||||
/// Exercises the template arm: there is no forge to clone from in a test
|
||||
/// (`forge::is_present` needs the priv socket), so `setup_proposed` falls
|
||||
/// through to `seed_template` — which is the arm this asserts about.
|
||||
#[tokio::test]
|
||||
async fn setup_proposed_seeds_flake_nix() {
|
||||
let dir = tempfile::tempdir().expect("tempdir");
|
||||
|
|
|
|||
|
|
@ -88,7 +88,12 @@ pub fn spawn(
|
|||
// paths independent reconnect state, so one could be serving while
|
||||
// the other was still down. `async_nats::Client` is a handle, so the
|
||||
// clone is cheap.
|
||||
tokio::spawn(drain_swarm_events(client.clone(), coord, shutdown.clone()));
|
||||
tokio::spawn(drain_swarm_events(
|
||||
client.clone(),
|
||||
coord,
|
||||
hive.clone(),
|
||||
shutdown.clone(),
|
||||
));
|
||||
|
||||
let mut health = SweepHealth::new("swarm_status_publish", "warn", FAILURES_BEFORE_BANNER);
|
||||
loop {
|
||||
|
|
@ -127,39 +132,40 @@ pub fn spawn(
|
|||
});
|
||||
}
|
||||
|
||||
/// Listen on the swarm's knowledge-event subject and act on what arrives.
|
||||
/// Listen on the swarm's event subjects and act on what arrives.
|
||||
///
|
||||
/// The controller decides *what a forge delivery means* and addresses the
|
||||
/// result here; this end does not know a forge exists. Today the one event is
|
||||
/// **the knowledge repository changed**, and the response is the pull this
|
||||
/// daemon already runs at boot.
|
||||
/// result here; this end does not know a forge exists. Two events today:
|
||||
/// **the knowledge repository changed**, answered by the pull this daemon
|
||||
/// already runs at boot, and **deploy this agent**, answered by the same
|
||||
/// rebuild insert the operator's own verb makes.
|
||||
///
|
||||
/// # There is no payload, and that is deliberate
|
||||
/// Only the deploy event carries a payload, and only the agent name: its
|
||||
/// subject already names the hive, so this one listens on its own rather than
|
||||
/// filtering a swarm-wide feed. Even then it is a trigger, never the config,
|
||||
/// which git owns.
|
||||
///
|
||||
/// The event carries nothing. The webhook handler this replaces read two
|
||||
/// fields from Forgejo and used neither — both were filters — then ran
|
||||
/// `git pull`, which re-derives everything from the repository. So it is an
|
||||
/// edge trigger, and reading a body here would invent a contract nobody owes.
|
||||
/// # A missed message costs the two events very differently
|
||||
///
|
||||
/// # What a missed message costs
|
||||
/// Core NATS, so delivery is at-most-once. Harmless for knowledge: this
|
||||
/// daemon pulls at startup regardless, and the webhook it replaced was lost
|
||||
/// identically when a hive was down.
|
||||
///
|
||||
/// Core NATS, so delivery is at-most-once: a hive that is down when the
|
||||
/// controller publishes never hears it, and its knowledge stays as of its last
|
||||
/// pull until it next boots. **That is not a regression** — a webhook delivery
|
||||
/// to a hive that is down is lost identically, and this daemon pulls at startup
|
||||
/// regardless. `JetStream` would require this end to *publish* to
|
||||
/// `$JS.API.CONSUMER.CREATE.<stream>`, which the callout policy does not grant,
|
||||
/// so durability would cost a grant on both sides to remove a failure the boot
|
||||
/// pull already covers.
|
||||
/// ⚠️ **The deploy event has no such second path** — nothing else would ever
|
||||
/// tell this hive to build that agent. Its backstop is the hive-side reconcile
|
||||
/// loop ("hives pull and self-update"); until that exists this is a nudge with
|
||||
/// no safety net. Not papered over with `JetStream`: durability on one subject
|
||||
/// looks like a fix while the desired state still lives only in a message.
|
||||
///
|
||||
/// ⚠️ **A refused subscription is indistinguishable from a quiet one.** NATS
|
||||
/// reports an authorization violation asynchronously on the connection, not as
|
||||
/// an error from `subscribe`, so this task cannot tell "no events published"
|
||||
/// from "not allowed to hear them". If a hive stops picking up knowledge
|
||||
/// changes, the server log is the thing that knows why — nothing here will say.
|
||||
/// from "not allowed to hear them". If a hive stops picking up either event,
|
||||
/// the server log is the thing that knows why — nothing here will say.
|
||||
async fn drain_swarm_events(
|
||||
client: async_nats::Client,
|
||||
coord: std::sync::Arc<crate::coordinator::Coordinator>,
|
||||
hive: String,
|
||||
mut shutdown: tokio::sync::watch::Receiver<bool>,
|
||||
) {
|
||||
// One subject for the whole swarm, so this hive's own name never enters
|
||||
|
|
@ -179,7 +185,20 @@ async fn drain_swarm_events(
|
|||
return;
|
||||
}
|
||||
};
|
||||
tracing::info!(%subject, "swarm events: listening");
|
||||
// This hive's own deploy subject, not a swarm-wide one: the controller
|
||||
// addresses each hive, so nothing arrives here that is not for us.
|
||||
let deploy_subject = swarm_queue_client::deploy_subject(&hive);
|
||||
let mut deploy_sub = match client.subscribe(deploy_subject.clone()).await {
|
||||
Ok(sub) => sub,
|
||||
Err(e) => {
|
||||
tracing::warn!(
|
||||
subject = %deploy_subject, error = %e,
|
||||
"swarm events: subscribe failed; this hive will not hear deploy requests"
|
||||
);
|
||||
return;
|
||||
}
|
||||
};
|
||||
tracing::info!(%subject, %deploy_subject, "swarm events: listening");
|
||||
|
||||
loop {
|
||||
tokio::select! {
|
||||
|
|
@ -197,6 +216,13 @@ async fn drain_swarm_events(
|
|||
tracing::warn!(error = ?e, "swarm events: knowledge pull failed");
|
||||
}
|
||||
}
|
||||
msg = deploy_sub.next() => {
|
||||
let Some(msg) = msg else {
|
||||
tracing::warn!(subject = %deploy_subject, "swarm events: deploy subscription closed");
|
||||
return;
|
||||
};
|
||||
handle_deploy_request(&coord, &msg.payload).await;
|
||||
}
|
||||
_ = shutdown.changed() => {
|
||||
tracing::info!("swarm events: shutdown signal received");
|
||||
return;
|
||||
|
|
@ -205,6 +231,67 @@ async fn drain_swarm_events(
|
|||
}
|
||||
}
|
||||
|
||||
/// Act on one deploy request. Everything that arrives on this hive's own
|
||||
/// deploy subject is for this hive, so there is nothing to filter.
|
||||
///
|
||||
/// A payload that will not decode is worth a `warn`: the controller and this
|
||||
/// end share one type, so a decode failure means they disagree about it.
|
||||
async fn handle_deploy_request(
|
||||
coord: &std::sync::Arc<crate::coordinator::Coordinator>,
|
||||
payload: &[u8],
|
||||
) {
|
||||
let request: swarm_queue_client::DeployRequest = match serde_json::from_slice(payload) {
|
||||
Ok(request) => request,
|
||||
Err(e) => {
|
||||
tracing::warn!(error = %e, "swarm events: undecodable deploy request");
|
||||
return;
|
||||
}
|
||||
};
|
||||
let agent = request.agent;
|
||||
|
||||
// Which of the two meanings this request has is decided here, and the
|
||||
// predicate is "does a container exist", not "is one running":
|
||||
// `agents_for_meta_listing` is `nixos-container list`, so a stopped agent
|
||||
// still counts. `Coordinator::list_agents` would *look* right and is the
|
||||
// registered-MCP-socket set — a stopped agent is absent from it, and this
|
||||
// would then try to create a container that already exists.
|
||||
let known = match crate::lifecycle::agents_for_meta_listing().await {
|
||||
Ok(agents) => agents.iter().any(|spec| spec.name == agent),
|
||||
Err(e) => {
|
||||
// Fail closed: without the list this cannot tell first deploy from
|
||||
// rebuild, and guessing wrong in the "new" direction tries to
|
||||
// create over an existing container.
|
||||
tracing::warn!(%agent, error = ?e, "swarm events: cannot enumerate agents; deploy dropped");
|
||||
return;
|
||||
}
|
||||
};
|
||||
|
||||
let inserted = if known {
|
||||
// The same insert the operator's own `rebuild` verb makes, relock and
|
||||
// all: "deploy this agent" means here exactly what it already meant, and
|
||||
// a swarm-triggered rebuild that quietly did something narrower would be
|
||||
// a second definition of the word.
|
||||
coord
|
||||
.job_queue
|
||||
.insert_job(|b| crate::job_queue::templates::rebuild(b, &agent, true))
|
||||
} else {
|
||||
// First deploy. The swarm has already created the identity, the forge
|
||||
// repo and its config; what is left is hive-local, and there is no
|
||||
// approval to wait on because the operator's click at swarm level is
|
||||
// the authorisation.
|
||||
coord
|
||||
.job_queue
|
||||
.insert_job(|b| crate::job_queue::templates::first_deploy(b, &agent))
|
||||
};
|
||||
match inserted {
|
||||
Ok(_) => {
|
||||
tracing::info!(%agent, known, "swarm events: deploy requested by the swarm, queued");
|
||||
coord.emit_rebuild_queue_snapshot();
|
||||
}
|
||||
Err(e) => tracing::warn!(%agent, error = %e, "swarm events: queueing the deploy failed"),
|
||||
}
|
||||
}
|
||||
|
||||
/// Offer one snapshot: this hive's current readiness, under its own key.
|
||||
async fn publish(client: &async_nats::Client, hive: &str) -> Result<()> {
|
||||
// An unconnected client does not fail a JetStream request, it hangs
|
||||
|
|
|
|||
|
|
@ -380,12 +380,17 @@ impl Client {
|
|||
/// Seed `repo` with the two files every agent config repo needs:
|
||||
/// `agent.nix` (the agent's own module) and `flake.nix` (the
|
||||
/// boilerplate that lets the meta flake import this repo as a flake
|
||||
/// input) — same content `hive-c0re::lifecycle::setup::setup_proposed`
|
||||
/// writes at the per-hive level, committed here in one atomic
|
||||
/// `repo_change_files` call instead of a local `git commit` (this
|
||||
/// process has no working tree to commit from — it only ever talks to
|
||||
/// the forge over HTTP). The whole job of the `InitAgentConfigRepo`
|
||||
/// node.
|
||||
/// input) — committed here in one atomic `repo_change_files` call
|
||||
/// instead of a local `git commit` (this process has no working tree to
|
||||
/// commit from — it only ever talks to the forge over HTTP). The whole
|
||||
/// job of the `InitAgentConfigRepo` node.
|
||||
///
|
||||
/// **This is the agent's config, not a copy of it.** A hive told to
|
||||
/// deploy the agent clones this repo
|
||||
/// (`hive-c0re::forge::clone_config_into_proposed`) rather than writing
|
||||
/// the same files again locally; the byte-identical template in
|
||||
/// `hive-c0re::lifecycle::setup::seed_template` is only reached when no
|
||||
/// repo exists here to clone.
|
||||
///
|
||||
/// Idempotent by construction rather than by catching a conflict: an
|
||||
/// unconditional `repo_change_files` against an already-seeded repo
|
||||
|
|
|
|||
|
|
@ -86,6 +86,14 @@ enum SwarmNodeKind {
|
|||
/// swarm routes a deploy message to — it belongs on the node that
|
||||
/// sends that message, not on this one.
|
||||
InitAgentConfigRepo { agent: String },
|
||||
/// Tell `hive` to rebuild `agent`, by publishing on the swarm's deploy
|
||||
/// subject. The one node kind whose effect leaves this host.
|
||||
///
|
||||
/// Carries the hive, unlike every variant above — this is the node
|
||||
/// `InitAgentConfigRepo`'s doc points at when it says the hive belongs
|
||||
/// on the node that sends the deploy message. It names the subject the
|
||||
/// message goes to, one per hive, so no other hive is woken by it.
|
||||
TriggerDeploy { hive: String, agent: String },
|
||||
}
|
||||
|
||||
impl hive_jobq_wire::WireNode for SwarmNodeKind {
|
||||
|
|
@ -96,15 +104,17 @@ impl hive_jobq_wire::WireNode for SwarmNodeKind {
|
|||
SwarmNodeKind::CreateForgeUser { .. } => "create_forge_user".to_owned(),
|
||||
SwarmNodeKind::AddRepoMember { .. } => "add_repo_member".to_owned(),
|
||||
SwarmNodeKind::InitAgentConfigRepo { .. } => "init_agent_config_repo".to_owned(),
|
||||
SwarmNodeKind::TriggerDeploy { .. } => "trigger_deploy".to_owned(),
|
||||
}
|
||||
}
|
||||
|
||||
fn data(&self, _id: hive_jobq_wire::WireId) -> serde_json::Value {
|
||||
// Every node in this graph is about exactly one agent, so they all
|
||||
// render the same and `label()` is what distinguishes them. Spelled
|
||||
// out as an or-pattern rather than a catch-all on purpose: a fifth
|
||||
// variant then fails to compile here instead of silently rendering
|
||||
// as an agent name.
|
||||
// Spelled out as an or-pattern rather than a catch-all on purpose: a
|
||||
// new variant then fails to compile here instead of silently
|
||||
// rendering as an agent name. That gate has now fired once —
|
||||
// `TriggerDeploy` is the first node that is about an agent *and a
|
||||
// hive*, and a catch-all would have dropped the hive from the
|
||||
// viewer without anyone noticing.
|
||||
match self {
|
||||
SwarmNodeKind::CreateIdentity { agent }
|
||||
| SwarmNodeKind::CreateRepo { agent }
|
||||
|
|
@ -113,6 +123,9 @@ impl hive_jobq_wire::WireNode for SwarmNodeKind {
|
|||
| SwarmNodeKind::InitAgentConfigRepo { agent } => {
|
||||
serde_json::json!({ "agent": agent })
|
||||
}
|
||||
SwarmNodeKind::TriggerDeploy { hive, agent } => {
|
||||
serde_json::json!({ "agent": agent, "hive": hive })
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
@ -140,6 +153,13 @@ impl hive_jobq_wire::WireResource for SwarmResourceKind {
|
|||
struct WorkerDeps {
|
||||
auth: Option<std::sync::Arc<auth::AuthBridge>>,
|
||||
forge: Option<std::sync::Arc<forge::Client>>,
|
||||
/// The swarm queue connection, for nodes whose effect is a published
|
||||
/// event rather than an API call. A handle rather than the status
|
||||
/// reader it is cloned from: a node that announces a deploy has no
|
||||
/// business reading hive status, and `status.rs`'s own doc says the
|
||||
/// connection living there is an accident of construction order, not a
|
||||
/// claim that events are a kind of status.
|
||||
queue: Option<async_nats::Client>,
|
||||
}
|
||||
|
||||
/// Run a claimed node's actual work. Mirrors `hive-c0re/src/job_queue/
|
||||
|
|
@ -226,10 +246,51 @@ async fn run_swarm_node(
|
|||
Err(e) => Outcome::Failed(format!("{e:#}")),
|
||||
},
|
||||
},
|
||||
SwarmNodeKind::TriggerDeploy { hive, agent } => match deps.queue {
|
||||
None => Outcome::Failed(
|
||||
"no swarm queue is configured on this host, so no hive can be told to deploy"
|
||||
.to_owned(),
|
||||
),
|
||||
Some(client) => publish_deploy(&client, &hive, &agent).await,
|
||||
},
|
||||
};
|
||||
(builder, outcome)
|
||||
}
|
||||
|
||||
/// Publish one deploy request and report whether it left this process.
|
||||
///
|
||||
/// The flush is not belt-and-braces: `publish` hands the message to the
|
||||
/// client's write buffer and returns, so a node that reported `Done` on
|
||||
/// that alone would be claiming a delivery it has no evidence for — the
|
||||
/// same ordering `webhook::announce_knowledge_change` documents.
|
||||
async fn publish_deploy(
|
||||
client: &async_nats::Client,
|
||||
hive: &str,
|
||||
agent: &str,
|
||||
) -> hive_jobq::scheduler::Outcome {
|
||||
use hive_jobq::scheduler::Outcome;
|
||||
|
||||
// One subject per hive, so the other hives are never woken by this.
|
||||
let subject = swarm_queue_client::deploy_subject(hive);
|
||||
let request = swarm_queue_client::DeployRequest {
|
||||
agent: agent.to_owned(),
|
||||
};
|
||||
let payload = match serde_json::to_vec(&request) {
|
||||
Ok(payload) => payload,
|
||||
Err(e) => return Outcome::Failed(format!("encoding the deploy request failed: {e}")),
|
||||
};
|
||||
if let Err(e) = client.publish(subject.clone(), payload.into()).await {
|
||||
return Outcome::Failed(format!("publishing to {subject} failed: {e}"));
|
||||
}
|
||||
if let Err(e) = client.flush().await {
|
||||
return Outcome::Failed(format!(
|
||||
"flushing the deploy event to {subject} failed: {e}"
|
||||
));
|
||||
}
|
||||
tracing::info!(%subject, %hive, %agent, "swarm jobq: deploy requested");
|
||||
Outcome::Done
|
||||
}
|
||||
|
||||
/// Spawn the swarm-level job-graph scheduler loop. Mirrors `hive-c0re/src/
|
||||
/// job_queue/scheduler.rs::run_worker`'s shape: claim one runnable node,
|
||||
/// spawn the future that runs + completes it, loop again immediately if
|
||||
|
|
@ -851,9 +912,22 @@ async fn create_agent(
|
|||
})
|
||||
.after_ok(create_repo)
|
||||
.after_ok(create_forge_user);
|
||||
let _init_config = b
|
||||
.node(SwarmNodeKind::InitAgentConfigRepo { agent })
|
||||
let init_config = b
|
||||
.node(SwarmNodeKind::InitAgentConfigRepo {
|
||||
agent: agent.clone(),
|
||||
})
|
||||
.after_ok(create_repo);
|
||||
// Last, and specifically after the config repo is seeded: the
|
||||
// hive deploys by reading that repo, so a deploy asked for any
|
||||
// earlier would find nothing to build. This is the edge that
|
||||
// makes creating an agent at swarm level actually put it on a
|
||||
// hive, rather than leaving a provisioned name nobody runs.
|
||||
let _trigger_deploy = b
|
||||
.node(SwarmNodeKind::TriggerDeploy {
|
||||
hive: hive.clone(),
|
||||
agent,
|
||||
})
|
||||
.after_ok(init_config);
|
||||
vec![create_identity.guid()]
|
||||
})
|
||||
.map_err(|e| {
|
||||
|
|
@ -1151,6 +1225,7 @@ async fn main() -> Result<()> {
|
|||
let deps = WorkerDeps {
|
||||
auth: auth.clone(),
|
||||
forge: forge_client.clone(),
|
||||
queue: status.as_ref().map(|s| s.queue_client()),
|
||||
};
|
||||
|
||||
let jobq = Arc::new(Mutex::new(hive_jobq::scheduler::Scheduler::new(
|
||||
|
|
@ -1465,6 +1540,7 @@ mod tests {
|
|||
let deps = WorkerDeps {
|
||||
auth: None,
|
||||
forge: None,
|
||||
queue: None,
|
||||
};
|
||||
let runner =
|
||||
hive_jobq::scheduler::Scheduler::claim_next(&sched, move |id, kind, builder| {
|
||||
|
|
@ -1516,6 +1592,7 @@ mod tests {
|
|||
let deps = WorkerDeps {
|
||||
auth: None,
|
||||
forge: None,
|
||||
queue: None,
|
||||
};
|
||||
let runner =
|
||||
hive_jobq::scheduler::Scheduler::claim_next(&sched, move |id, kind, builder| {
|
||||
|
|
|
|||
|
|
@ -278,6 +278,14 @@ impl Policy {
|
|||
// about a change, with nothing in the controller's log to say a
|
||||
// permission was the reason.
|
||||
swarm_queue_client::KNOWLEDGE_SUBJECT.to_owned(),
|
||||
// The deploy events: one subject per hive, so a hive is not woken
|
||||
// by a deploy meant for another. Granted as a wildcard because
|
||||
// this responder has no roster — it cannot enumerate hives, and a
|
||||
// grant that had to track one would be a second place to get the
|
||||
// list wrong (see `hive_name`'s doc for the same argument about
|
||||
// admission). Same failure mode as the knowledge event above: a
|
||||
// refused publish reaches the client as a timeout.
|
||||
swarm_queue_client::DEPLOY_SUBJECT_WILDCARD.to_owned(),
|
||||
]);
|
||||
subjects
|
||||
}
|
||||
|
|
@ -373,6 +381,43 @@ mod tests {
|
|||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn a_reader_may_publish_a_deploy_event_to_any_hive() {
|
||||
let p = policy().permissions("swarm-controller").expect("a reader");
|
||||
assert!(
|
||||
p.publish
|
||||
.contains(&swarm_queue_client::DEPLOY_SUBJECT_WILDCARD.to_owned()),
|
||||
"the controller is the only publisher of these events; without the \
|
||||
grant its publish is refused, and a refusal arrives as a timeout"
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn a_hive_may_not_publish_a_deploy_event_to_anyone_including_itself() {
|
||||
// Same arm as the knowledge event's, and it matters more here: a forged
|
||||
// knowledge event makes a hive re-read a repo, while a forged deploy
|
||||
// event makes it rebuild and restart a named agent.
|
||||
//
|
||||
// Both directions asserted, because splitting the subject per hive
|
||||
// makes them separate strings for the first time: a hive must reach
|
||||
// neither another hive's deploy subject nor its own. Nothing in the
|
||||
// grant should mention this family at all.
|
||||
let p = policy()
|
||||
.permissions("hive-alpha")
|
||||
.expect("a hive is admitted");
|
||||
for forbidden in [
|
||||
swarm_queue_client::deploy_subject("beta"),
|
||||
swarm_queue_client::deploy_subject("alpha"),
|
||||
swarm_queue_client::DEPLOY_SUBJECT_WILDCARD.to_owned(),
|
||||
] {
|
||||
assert!(
|
||||
!p.publish.contains(&forbidden),
|
||||
"a hive must not publish {forbidden}: {:?}",
|
||||
p.publish
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn a_hive_grant_never_includes_the_jetstream_wildcard() {
|
||||
// `$JS.API.>` also covers `$JS.API.STREAM.DELETE.KV_hive-status`, with
|
||||
|
|
|
|||
|
|
@ -173,6 +173,46 @@ pub mod status;
|
|||
/// permitted at all — speaks neither `jetstream` nor `kv`.
|
||||
pub const KNOWLEDGE_SUBJECT: &str = "$SWARM.knowledge";
|
||||
|
||||
/// The subject the swarm controller publishes on to ask `hive` to rebuild one
|
||||
/// of its agents. Lives here for the same three-crate reason as
|
||||
/// [`KNOWLEDGE_SUBJECT`], but unlike it this is a **family, one subject per
|
||||
/// hive** — a hive subscribes to its own and is never woken by a deploy meant
|
||||
/// for someone else.
|
||||
///
|
||||
/// ⚠️ **That split is about noise, not confidentiality.** The auth-callout
|
||||
/// responder scopes *publish* only and leaves `sub` unrestricted, so a hive
|
||||
/// that wanted another's messages could still subscribe to them. What the
|
||||
/// family buys is that it does not receive them by default.
|
||||
#[must_use]
|
||||
pub fn deploy_subject(hive: &str) -> String {
|
||||
format!("{DEPLOY_SUBJECT_PREFIX}.{hive}")
|
||||
}
|
||||
|
||||
/// The publish grant covering every [`deploy_subject`], for the one client
|
||||
/// that may send them. A wildcard rather than a subject per hive because the
|
||||
/// responder has no roster — it cannot enumerate hives, and a grant it had to
|
||||
/// keep in step with one would be a second place to get the list wrong.
|
||||
pub const DEPLOY_SUBJECT_WILDCARD: &str = "$SWARM.deploy.*";
|
||||
|
||||
/// Shared by [`deploy_subject`] and [`DEPLOY_SUBJECT_WILDCARD`] so the two
|
||||
/// cannot drift into naming different families.
|
||||
const DEPLOY_SUBJECT_PREFIX: &str = "$SWARM.deploy";
|
||||
|
||||
/// What a [`deploy_subject`] message carries.
|
||||
///
|
||||
/// Only the agent: the subject already names the hive, and repeating it here
|
||||
/// would be two places stating one fact, free to disagree.
|
||||
///
|
||||
/// ⚠️ **A trigger, not the config.** The hive already tracks the agent's
|
||||
/// config repo; putting desired state on the wire would make this message a
|
||||
/// second source of truth for something git already owns, and a hive that
|
||||
/// missed a message would then be wrong rather than merely late.
|
||||
#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
|
||||
pub struct DeployRequest {
|
||||
/// The agent to rebuild, as the swarm knows it.
|
||||
pub agent: String,
|
||||
}
|
||||
|
||||
/// The hive-notices stream, shared by the hive that publishes and
|
||||
/// whatever eventually consumes it. Behind the `notices` feature, same
|
||||
/// reason `status` is behind `kv` — see the module doc.
|
||||
|
|
|
|||
Loading…
Reference in a new issue