vinzenz/hyperhive
vinzenz/hyperhive
Watch
0
0
Fork
You've already forked hyperhive
 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions 1

Compare commits

base: vinzenz:ccf9951e5dc4ad599a8944a0da48ab44093dbc5c
Branches Tags
vinzenz:main
..
compare: vinzenz:5f197a38eeb90a80a6ebd9838c6d2f619ee8499a
Branches Tags
vinzenz:main
10 changed files with 57 additions and 418 deletions
Download patch file Download diff file Expand all files Collapse all files

6
hive-c0re/src/forge/mod.rs
View file

@ -17,9 +17,9 @@ pub use pr_merge::{
};
pub use reconcile::{reconcile_config_apply, reconcile_config_status};
pub use repos::{
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,
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};

62
hive-c0re/src/forge/repos.rs
View file

@ -593,68 +593,6 @@ 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.

23
hive-c0re/src/job_queue/templates.rs
View file

@ -452,15 +452,6 @@ 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() })
@ -479,20 +470,8 @@ pub(crate) fn spawn_nodes<'a>(builder: &'a JobBuilder, agent: &str) -> Handle<'a
.needs(Resource::Agent(a()))
.part_of(create)
.after_ok(dropin);
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()]
resolve_approval_tails(builder, approval_id, provision);
}
/// Teardown: `Stop` → `DestroyContainer` → (`PurgeState`) → `DestroyBookkeeping`.

56
hive-c0re/src/lifecycle/setup.rs
View file

@ -14,18 +14,12 @@ 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, 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.
/// 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.
///
/// **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
@ -40,9 +34,19 @@ 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()))?;
if !crate::forge::clone_config_into_proposed(proposed_dir, name).await {
seed_template(proposed_dir, name).await?;
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?;
}
// Idempotently wire the `applied` remote — purely for the
// manager's ergonomics. The URL is the path inside the manager
@ -52,30 +56,6 @@ 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()

4
hive-c0re/src/lifecycle/tests.rs
View file

@ -6,10 +6,6 @@ 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");

131
hive-c0re/src/swarm_status.rs
View file

@ -88,12 +88,7 @@ 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,
hive.clone(),
shutdown.clone(),
));
tokio::spawn(drain_swarm_events(client.clone(), coord, shutdown.clone()));
let mut health = SweepHealth::new("swarm_status_publish", "warn", FAILURES_BEFORE_BANNER);
loop {
@ -132,40 +127,39 @@ pub fn spawn(
});
}
/// Listen on the swarm's event subjects and act on what arrives.
/// Listen on the swarm's knowledge-event subject 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. 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.
/// 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.
///
/// 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.
/// # There is no payload, and that is deliberate
///
/// # A missed message costs the two events very differently
/// 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.
///
/// 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.
/// # What a missed message costs
///
/// ⚠️ **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.
/// 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.
///
/// ⚠️ **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 either event,
/// 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 knowledge
/// changes, 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
@ -185,20 +179,7 @@ async fn drain_swarm_events(
return;
}
};
// 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");
tracing::info!(%subject, "swarm events: listening");
loop {
tokio::select! {
@ -216,13 +197,6 @@ 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;
@ -231,67 +205,6 @@ 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

17
swarm-controller/src/forge.rs
View file

@ -380,17 +380,12 @@ 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) — 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.
/// 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.
///
/// Idempotent by construction rather than by catching a conflict: an
/// unconditional `repo_change_files` against an already-seeded repo

91
swarm-controller/src/main.rs
View file

@ -86,14 +86,6 @@ 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 {
@ -104,17 +96,15 @@ 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 {
// 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.
// 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.
match self {
SwarmNodeKind::CreateIdentity { agent }
| SwarmNodeKind::CreateRepo { agent }
@ -123,9 +113,6 @@ 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 })
}
}
}
}
@ -153,13 +140,6 @@ 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/
@ -246,51 +226,10 @@ 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
@ -912,22 +851,9 @@ async fn create_agent(
})
.after_ok(create_repo)
.after_ok(create_forge_user);
let init_config = b
.node(SwarmNodeKind::InitAgentConfigRepo {
agent: agent.clone(),
})
let _init_config = b
.node(SwarmNodeKind::InitAgentConfigRepo { agent })
.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| {
@ -1225,7 +1151,6 @@ 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(
@ -1540,7 +1465,6 @@ mod tests {
let deps = WorkerDeps {
auth: None,
forge: None,
queue: None,
};
let runner =
hive_jobq::scheduler::Scheduler::claim_next(&sched, move |id, kind, builder| {
@ -1592,7 +1516,6 @@ mod tests {
let deps = WorkerDeps {
auth: None,
forge: None,
queue: None,
};
let runner =
hive_jobq::scheduler::Scheduler::claim_next(&sched, move |id, kind, builder| {

45
swarm-nats-auth/src/policy.rs
View file

@ -278,14 +278,6 @@ 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
}
@ -381,43 +373,6 @@ 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

40
swarm-queue-client/src/lib.rs
View file

@ -173,46 +173,6 @@ 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.