diff --git a/hive-c0re/src/forge/mod.rs b/hive-c0re/src/forge/mod.rs index 5dfdb81e..67b8ad73 100644 --- a/hive-c0re/src/forge/mod.rs +++ b/hive-c0re/src/forge/mod.rs @@ -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}; diff --git a/hive-c0re/src/forge/repos.rs b/hive-c0re/src/forge/repos.rs index ed80387c..74a7252e 100644 --- a/hive-c0re/src/forge/repos.rs +++ b/hive-c0re/src/forge/repos.rs @@ -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/` 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/` 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 ` 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. diff --git a/hive-c0re/src/job_queue/templates.rs b/hive-c0re/src/job_queue/templates.rs index 94d7366a..881b52b6 100644 --- a/hive-c0re/src/job_queue/templates.rs +++ b/hive-c0re/src/job_queue/templates.rs @@ -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 { + vec![spawn_nodes(builder, agent).guid()] } /// Teardown: `Stop` → `DestroyContainer` → (`PurgeState`) → `DestroyBookkeeping`. diff --git a/hive-c0re/src/lifecycle/setup.rs b/hive-c0re/src/lifecycle/setup.rs index e61b8950..b695c23d 100644 --- a/hive-c0re/src/lifecycle/setup.rs +++ b/hive-c0re/src/lifecycle/setup.rs @@ -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/` 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() diff --git a/hive-c0re/src/lifecycle/tests.rs b/hive-c0re/src/lifecycle/tests.rs index 81e925eb..c9734efc 100644 --- a/hive-c0re/src/lifecycle/tests.rs +++ b/hive-c0re/src/lifecycle/tests.rs @@ -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"); diff --git a/hive-c0re/src/swarm_status.rs b/hive-c0re/src/swarm_status.rs index 6f71fc03..4f17bb1c 100644 --- a/hive-c0re/src/swarm_status.rs +++ b/hive-c0re/src/swarm_status.rs @@ -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.`, 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, + hive: String, mut shutdown: tokio::sync::watch::Receiver, ) { // 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, + 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 diff --git a/swarm-controller/src/forge.rs b/swarm-controller/src/forge.rs index 74f38b25..ca4271a9 100644 --- a/swarm-controller/src/forge.rs +++ b/swarm-controller/src/forge.rs @@ -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 diff --git a/swarm-controller/src/main.rs b/swarm-controller/src/main.rs index 1b4477f6..7a82c0ef 100644 --- a/swarm-controller/src/main.rs +++ b/swarm-controller/src/main.rs @@ -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>, forge: Option>, + /// 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, } /// 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| { diff --git a/swarm-nats-auth/src/policy.rs b/swarm-nats-auth/src/policy.rs index 8c1fea29..ae5900c3 100644 --- a/swarm-nats-auth/src/policy.rs +++ b/swarm-nats-auth/src/policy.rs @@ -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 diff --git a/swarm-queue-client/src/lib.rs b/swarm-queue-client/src/lib.rs index c7a9143f..0d561af8 100644 --- a/swarm-queue-client/src/lib.rs +++ b/swarm-queue-client/src/lib.rs @@ -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.