diff --git a/docs/swarm/README.md b/docs/swarm/README.md index 26f30b7a..298c7408 100644 --- a/docs/swarm/README.md +++ b/docs/swarm/README.md @@ -131,14 +131,15 @@ it's stored in the queue, not in the daemon. ### Giving the agents the queue too -An agent authenticates as its own client, not as its hive, so it needs its -own coordinates. Two of them are options on the hive; the other two arrive -with the credential itself and aren't configurable. +An agent connects to the queue as itself, with a credential `swarm-controller` +mints for it when it creates the agent (`swarm/agents//queue` in the +secret store, see [`credentials.md`](credentials.md)). The harness reads it from +the store under the agent's own certificate before each connection attempt and +holds it in memory only. The one thing to configure is where the queue is: -| option | what to set it to | -| ------------------------------------------- | -------------------------------------------------------------- | -| `deploy.hive-controller.queue.agentNatsUrl` | where the queue listens, as an agent **container** reaches it | -| `swarm.statusPublish.tokenEndpoint` | the swarm IdP's `/api/oidc/token` — the same one the hive uses | +| option | what to set it to | +| ------------------------------------------- | ------------------------------------------------------------- | +| `deploy.hive-controller.queue.agentNatsUrl` | where the queue listens, as an agent **container** reaches it | On every hive, `agentNatsUrl` defaults to the same `tls://:` as the hive's own. On the queue's @@ -147,26 +148,33 @@ firewall opens the port; on any other hive it resolves through the host's DNS, like the store's name. ⚠️ **Never a loopback address here**: an agent has its own network namespace, so `127.0.0.1` reaches the agent. -The harness sees four variables, and treats them as all-or-none: -`HIVE_AGENT_NATS_URL` and `HIVE_AGENT_OIDC_TOKEN_ENDPOINT` from the two options -above, plus `HIVE_AGENT_OIDC_CLIENT_SECRET_FILE` and -`HIVE_AGENT_OIDC_CLIENT_ID_FILE`, which point into the unit's own credentials -directory. The last two come from the delivered credential rather than from -config — see [`secrets.md`](secrets.md#hive-level--one-of-each-per-hive) for -how it gets there. A hive lacking the queue's address for its -agents sets none of the four and each agent logs that it has none; a half-set -environment logs an error and the harness keeps serving. +**An agent can't run without the connection.** Without it the swarm can neither +show the agent nor send it anything, so a harness without one stops: + +- A harness started without the queue's address (`HIVE_AGENT_NATS_URL`) or the + store's coordinates (`BAO_ADDR`, `HIVE_AGENT_NAME` and the agent's + certificate) exits at boot with an error naming what's missing. +- A harness whose first connect fails tries again after 5 seconds, doubling the + wait each time, and logs a warning for each attempt that fails. After the + seventh, about five minutes in, it logs an error and exits. + +Either way the `hive-agent` unit fails, and systemd starts it again 30 seconds +later, with no start limit, until the queue accepts it. Once connected, the +client reconnects by itself after a dropped connection, reading the credential +from the store again for each attempt. + +The hive's shared queue client is still delivered into the harness unit's +credentials directory, and `swarm.statusPublish.tokenEndpoint` still reaches +the container, but only for `swarm-logs`: the harness never connects with +them.
What an agent publishes over the queue What an agent does with that connection is publish its terminal. Every row its own web UI renders also goes to `$SWARM.term.`, one subject per agent, so a swarm-level terminal can follow one agent without subscribing to the swarm's -whole traffic. An agent connected with its own queue credential gets that -subject. An agent without one, or whose own credential the queue -refused, connects with its hive's shared client and publishes to -`$SWARM.term..` instead, the `` being the one that client id -names. The swarm controller relays both. Publishing only: an +whole traffic. The queue grants that subject to the agent's own credential, +and the swarm controller relays it. Publishing only: an agent talks about itself here and reads nothing. Rows aren't retained — a subscriber that wasn't listening missed them, the same as on the agent's own live stream. @@ -192,11 +200,11 @@ sending and leaves a marker in its place; the summary, level and icon still arrive. The harness logs and skips a row that's too large even without its body. The second thing an agent publishes is its **turn-state header**, on -`$SWARM.agent-state.` (or `$SWARM.agent-state..`) — same shape of subject, same grant -mechanics, same lack of retention. It carries what a header bar wants: what the -turn loop is doing (`turn_state`, plus `turn_state_since` as an ISO 8601 UTC -stamp), which model (`model` and the resolved id the last turn actually ran on), -the context budget and the last turn's context and cost token blocks, and +`$SWARM.agent-state.` — same shape of subject, same grant mechanics, +same lack of retention. It carries what a header bar wants: what the turn loop +is doing (`turn_state`, plus `turn_state_since` as an ISO 8601 UTC stamp), +which model (`model` and the resolved id the last turn actually ran on), the +context budget and the last turn's context and cost token blocks, and `agent_state`. `agent_state` reuses the swarm's own wanted-state vocabulary @@ -224,24 +232,24 @@ The payload passes through opaquely — the controller never parses a header. The third thing an agent publishes is its **icon**, the same SVG its own `GET /icon` serves. It goes into the `agent-icons` KV bucket under the key ``, with no hive in it, so the swarm can show the icon of an agent that's -stopped or has moved hives. Only an agent connected with its own queue -credential publishes it: the queue grants that credential -`$KV.agent-icons.` and no other key, and grants the hive's shared client -none of the bucket. The harness writes once per start, because a config change -reaches an agent by restarting its container. An agent with no icon deletes its -key. Swarm-side, `GET /api/agents//icon` serves the stored bytes, and 404 -means the agent has no icon. swarm-ui's agent cards load it as an `` and show the -dimmed hyperhive mark for an agent without one, as the hive dashboard does. +stopped or has moved hives. The queue grants the agent's own credential +`$KV.agent-icons.` and no other key. The harness writes once per start, +because a config change reaches an agent by restarting its container. An agent +with no icon deletes its key. Swarm-side, `GET /api/agents//icon` serves +the stored bytes, and 404 means the agent has no icon. swarm-ui's agent cards +load it as an `` and show the dimmed hyperhive mark for an agent without +one, as the hive dashboard does. The queue also limits what an agent may **subscribe** to, because agents are the least trusted clients on the queue. An agent's own queue credential, which its subagent daemon also presents, may subscribe to its reply inbox `_INBOX.agent..>` and its operator-input subject `$SWARM.input.`, and nothing else. The hive's shared agent client may subscribe to its reply -inbox `_INBOX.client..>` only. Each connects with its prefix as its -inbox prefix, so the replies to the icon write and the schedule read arrive. -Neither can read another agent's terminal, state, input or replies. Hives and -the swarm controller may still subscribe to any subject in the account. +inbox `_INBOX.client..>` only. The harness and the subagent daemon +connect with `_INBOX.agent.` as their inbox prefix, so the replies to +the icon write and the schedule read arrive. Neither credential can read +another agent's terminal, state, input or replies. Hives and the swarm +controller may still subscribe to any subject in the account.
diff --git a/docs/swarm/schedules.md b/docs/swarm/schedules.md index 45c7de1d..30dba13b 100644 --- a/docs/swarm/schedules.md +++ b/docs/swarm/schedules.md @@ -76,14 +76,11 @@ or within a minute if the controller's first attempt to pass it on failed. ## When a prompt doesn't fire -- **The agent has no queue credential of its own.** The agent reads its - prompts with its own swarm-queue credential. An agent connected with its - hive's shared client logs `connected with the hive's shared client, which -has no read grant on the agent-schedules bucket` and fires nothing. -- **The agent's first connect with its own credential failed.** The harness - keeps the outcome of its first connect for the rest of its run: it falls - back to the hive's shared client (the case above) or stays disconnected, - and reads no prompts until it restarts. +- **The agent isn't connected to the swarm queue.** It reads its prompts with + its own swarm-queue credential. A harness that can't connect retries for + about five minutes, then exits and restarts 30 seconds later; it reads no + prompts until a connect succeeds. See + [Giving the agents the queue too](README.md#giving-the-agents-the-queue-too). - **The agent hasn't read its prompts since it started.** It fires from the last list it read; until the first read succeeds it fires nothing. A read that fails later keeps the previous list. diff --git a/docs/swarm/secrets.md b/docs/swarm/secrets.md index 62a01a02..6a15ffea 100644 --- a/docs/swarm/secrets.md +++ b/docs/swarm/secrets.md @@ -216,11 +216,11 @@ The **second reader** is the agent containers' queue credential: `glue-queue-agent-credential.nix` lands it as two files, the client secret and the client id it authenticates, because that's how a queue client reads them — the secret by path, the id by value. No local fallback exists, and none is -possible, so absent files mean this hive's agents don't connect, which is what -a swarm looks like before the publisher on the authelia host has run. The -reader runs before `hive-c0re.service`, which wants it (doesn't require it), so -an agent container never renders ahead of the credential; an unreachable store -delays the daemon's start rather than failing it. +possible, so absent files mean this hive's agents have no `swarm-logs`, which +is what a swarm looks like before the publisher on the authelia host has run. +The reader runs before `hive-c0re.service`, which wants it (doesn't require +it), so an agent container never renders ahead of the credential; an +unreachable store delays the daemon's start rather than failing it. That credential still has one hop left, because the reader of it lives inside an agent container. **It crosses as a systemd credential, not as a bind @@ -229,13 +229,13 @@ runs as its own unprivileged agent user, so a bind would deliver a file that user can't open. `hive-c0re` stats the two files — the directory is `0755`, so it needs no read access to either — and hands them to `systemd-nspawn --load-credential` through `hive-priv`, which runs as root. -Inside, `hive-agent.service` names the same two ids in `LoadCredential=` and -reads them out of its own `$CREDENTIALS_DIRECTORY`, owned by the agent user -and by nobody else. `hive-c0re` never reads the bytes at any point: it runs as -`hive-core`, which is the reason a copy wasn't an option either. When the -files aren't there the daemon forwards nothing and says so in its journal, and -the harness logs that it has no queue coordinates — the same absent-and-legal state, twice, -rather than a container that refuses to start. +Inside, `hive-agent.service` names the same two ids in `LoadCredential=`, which +lands them in its own `$CREDENTIALS_DIRECTORY`, owned by the agent user and by +nobody else, where the agent's `swarm-logs` reads them. The harness itself +doesn't: it connects to the queue with the agent's own credential. `hive-c0re` +never reads the bytes at any point: it runs as `hive-core`, which is the reason +a copy wasn't an option either. When the files aren't there the daemon forwards +nothing and says so in its journal, and the container starts without them. The **third reader** is Grafana's OIDC client secret, and it's the first one that's a _swarm service's_ own credential rather than a hive's. It has exactly diff --git a/docs/swarm/ui.md b/docs/swarm/ui.md index 1a910e9e..87fd4865 100644 --- a/docs/swarm/ui.md +++ b/docs/swarm/ui.md @@ -78,11 +78,10 @@ request, replacing any the browser sent. `POST /api/agents//input` takes that header as the sender and sends `{from, body}` as a NATS request on `$SWARM.input.`. swarm-controller is the only queue client allowed to publish there. The agent's harness subscribes to its own subject and queues -each message for its turn loop with its reminders. Only an agent connected with -its own queue credential may subscribe there; an agent on its hive's shared -client takes no input this way, and the request gets the 503 below. The harness doesn't reply, -so swarm-controller answers 202 when the request times out after two seconds, -and 503 when the queue reports no subscriber. +each message for its turn loop with its reminders. Only the agent's own queue +credential may subscribe there. The harness doesn't reply, so swarm-controller +answers 202 when the request times out after two seconds, and 503 when the +queue reports no subscriber. diff --git a/hive-agent/src/main.rs b/hive-agent/src/main.rs index c749f43c..a52ff3f3 100644 --- a/hive-agent/src/main.rs +++ b/hive-agent/src/main.rs @@ -512,9 +512,11 @@ async fn serve_main(socket: &Path, poll_ms: u64) -> Result<()> { let claude_dir = login::default_dir(); let initial = LoginState::from_dir(&claude_dir); tracing::info!(state = ?initial, claude_dir = %claude_dir.display(), "harness boot"); - // Resolve the swarm-queue credential while the boot log is still the - // thing an operator is reading. Nothing connects yet. - swarm_queue::init(); + // The queue connection is required: missing coordinates fail the boot, + // and a connect that keeps failing exits the process. Connecting here, + // not on a publisher's first call, holds that whichever publishers start. + swarm_queue::init()?; + tokio::spawn(swarm_queue::client()); // Config fact, stamped once — see `harness_state::write_api_key_mode`'s // doc for why hive-c0re needs this to stop reporting `needs_login` for // an agent whose `~/.claude/` is empty by design. diff --git a/hive-agent/src/schedule_timer.rs b/hive-agent/src/schedule_timer.rs index bada2353..a6f4f4c2 100644 --- a/hive-agent/src/schedule_timer.rs +++ b/hive-agent/src/schedule_timer.rs @@ -23,8 +23,7 @@ //! paused is recorded as if fired, so resuming does not fire it. //! //! The list itself is held in memory only: until the first successful read -//! after a start, nothing fires. Reading needs this agent's own credential; -//! on the hive's shared client the task logs that and stops. +//! after a start, nothing fires. use std::path::Path; use std::sync::{Arc, Mutex}; @@ -35,7 +34,6 @@ use rusqlite::{Connection, OptionalExtension as _, params}; use swarm_queue_client::agent_schedules::{AgentSchedules, Read, Schedule, read_own}; use tokio::sync::Notify; -use crate::swarm_queue::{Connection as QueueConnection, Presented}; use crate::todos::Todos; /// How often the key is read and the list checked for due slots. Also the @@ -179,12 +177,9 @@ fn push_todo(todos: &Todos, wake: &Notify, schedule: &Schedule) -> bool { } } -/// Run the ticker for the harness's lifetime, if this agent has a queue -/// credential and a label to name its key with. +/// Run the ticker for the harness's lifetime, if this agent has a label to +/// name its key with. pub async fn run(todos: Arc, wake: Arc) { - if !crate::swarm_queue::configured() { - return; - } let agent = crate::identity::label(); if agent.is_empty() { tracing::warn!("this agent has no label; not reading its swarm schedules"); @@ -197,16 +192,7 @@ pub async fn run(todos: Arc, wake: Arc) { return; } }; - let Some(QueueConnection { client, presented }) = crate::swarm_queue::client().await else { - return; - }; - if presented != Presented::Agent { - tracing::warn!( - "connected with the hive's shared client, which has no read grant on the \ - agent-schedules bucket; swarm schedules are not read" - ); - return; - } + let client = crate::swarm_queue::client().await; tracing::info!(%agent, "reading swarm schedules from the swarm queue"); let mut list: Option = None; loop { diff --git a/hive-agent/src/swarm_agent_icon.rs b/hive-agent/src/swarm_agent_icon.rs index 3a421ba2..45f8d458 100644 --- a/hive-agent/src/swarm_agent_icon.rs +++ b/hive-agent/src/swarm_agent_icon.rs @@ -10,9 +10,7 @@ //! An agent with no icon deletes its key, so an icon removed from the config //! also disappears from the swarm. //! -//! **Only under the agent's own credential.** That credential is granted -//! exactly this agent's key. The hive's shared client is granted none of the -//! bucket, so under it this publishes nothing. +//! The agent's own queue credential is granted exactly this agent's key. //! //! The write is a `JetStream` publish straight to the key's subject, the same //! message `kv::Store::put` sends, without opening the bucket first. That @@ -25,8 +23,6 @@ use std::time::Duration; use async_nats::jetstream::context::PublishErrorKind; -use crate::swarm_queue::{Connection, Presented}; - /// Where the agent's NixOS config puts the icon /// (`services.hyperhive.agent.icon`), and the file `GET /icon` serves. Absent /// when none is configured. @@ -63,20 +59,8 @@ fn read_icon(path: &Path) -> std::io::Result { } } -/// The subject this agent's icon is written to, or `None` under a credential -/// that is not granted it. -fn subject(presented: &Presented, agent: &str) -> Option { - match presented { - Presented::Agent => Some(swarm_queue_client::agent_icon::subject(agent)), - Presented::Hive { .. } => None, - } -} - -/// Start the publish task, if this agent has a queue credential and a label. +/// Start the publish task, if this agent has a label. pub fn spawn() { - if !crate::swarm_queue::configured() { - return; - } let agent = crate::identity::label(); if agent.is_empty() { tracing::warn!("this agent has no label; not publishing its icon upward"); @@ -86,16 +70,8 @@ pub fn spawn() { } async fn run(agent: String) { - let Some(Connection { client, presented }) = crate::swarm_queue::client().await else { - return; - }; - let Some(subject) = subject(&presented, &agent) else { - tracing::info!( - "connected with the hive's shared client, which may not write an agent's \ - icon; not publishing it upward" - ); - return; - }; + let client = crate::swarm_queue::client().await; + let subject = swarm_queue_client::agent_icon::subject(&agent); let icon = match read_icon(Path::new(ICON_PATH)) { Ok(icon) => icon, Err(e) => { @@ -168,25 +144,7 @@ async fn write( #[cfg(test)] mod tests { - use super::{Icon, Presented, read_icon, subject}; - - #[test] - fn under_its_own_credential_an_agent_writes_its_own_key() { - assert_eq!( - subject(&Presented::Agent, "atlas").as_deref(), - Some("$KV.agent-icons.atlas") - ); - } - - /// The hive's shared client is granted no icon key, so a publish under it - /// would be refused. - #[test] - fn under_the_hives_shared_client_nothing_is_written() { - let hive = Presented::Hive { - client_id: "hive-alpha-agent".to_owned(), - }; - assert_eq!(subject(&hive, "atlas"), None); - } + use super::{Icon, read_icon}; #[test] fn a_configured_icon_is_published_byte_for_byte() { diff --git a/hive-agent/src/swarm_agent_state.rs b/hive-agent/src/swarm_agent_state.rs index f4d48d77..a9a9cb16 100644 --- a/hive-agent/src/swarm_agent_state.rs +++ b/hive-agent/src/swarm_agent_state.rs @@ -24,10 +24,9 @@ //! alternative is a KV bucket written on every turn-state flip, and the //! swarm already has one of those at the resolution it can afford. //! -//! **Best-effort in every direction.** The turn loop and the web UI must not -//! notice whether the queue exists, so the bus receiver is the only thing -//! this task blocks on; a failed publish is a log line and the next -//! transition is still attempted. +//! **Best-effort per transition.** The turn loop and the web UI never wait on +//! the queue, so the bus receiver is the only thing this task blocks on; a +//! failed publish is a log line and the next transition is still attempted. use serde::Serialize; use tokio::sync::broadcast; @@ -35,7 +34,6 @@ use tokio::sync::broadcast; use swarm_queue_client::wanted::AgentState; use crate::events::{Bus, BusEvent, LiveEvent, TurnState}; -use crate::swarm_queue::{Connection, Presented}; use crate::term_msg::iso8601_utc; /// Subject family carrying agent turn-state headers, the swarm-wide @@ -44,18 +42,6 @@ use crate::term_msg::iso8601_utc; /// agent without filtering the swarm's whole header traffic. const SUBJECT_PREFIX: &str = "$SWARM.agent-state"; -/// What the queue's client id looks like either side of the hive's own name. -/// -/// These mirror the auth-callout responder's `--hive-client-prefix` and -/// `--agent-client-suffix`, which is where the authoritative pair lives: the -/// responder parses the hive back out of the presented client id and grants -/// exactly the configured subjects for *that* hive. The agent is not told -/// those flags, so it restates their defaults — a deployment that retunes -/// either one has to change them here too, and the symptom of not doing so -/// is every publish refused rather than a wrong subject being accepted. -const CLIENT_ID_PREFIX: &str = "hive-"; -const CLIENT_ID_SUFFIX: &str = "-agent"; - /// How much room to leave under the announced limit for everything the /// publish adds around the payload — subject, headers, protocol framing. /// Same margin and same reasoning as `swarm_term::HEADROOM`. @@ -108,25 +94,6 @@ struct AgentStateMsg { cost_usage: Option, } -/// The hive a queue client id names. -/// -/// The hive in the subject has to be the string the **responder** parses out -/// of this same client id, because that is what its grant is built from. The -/// harness knows a hive display name too, from a different source and with -/// no rule tying the two together — deriving the subject from that one -/// instead produces a publish the broker refuses, which surfaces as a header -/// that is simply never populated. -/// -/// `None` for an id that does not have the expected shape: a publisher that -/// guessed at a subject would be asking for a grant it cannot have, so the -/// caller disables itself instead. -fn hive_from_client_id(client_id: &str) -> Option<&str> { - let hive = client_id - .strip_prefix(CLIENT_ID_PREFIX)? - .strip_suffix(CLIENT_ID_SUFFIX)?; - (!hive.is_empty()).then_some(hive) -} - /// What this process can honestly report about the agent-state axis. /// /// Only two of [`AgentState`]'s four variants are knowable from inside the @@ -172,37 +139,13 @@ fn snapshot(bus: &Bus) -> AgentStateMsg { } } -/// The subject this agent's header goes to under the credential it connected -/// with, or `None` when a hive client id names no hive. -fn subject(presented: &Presented, agent: &str) -> Option { - match presented { - Presented::Agent => Some(format!("{SUBJECT_PREFIX}.{agent}")), - Presented::Hive { client_id } => { - let Some(hive) = hive_from_client_id(client_id) else { - tracing::warn!( - %client_id, - expected = format!("{CLIENT_ID_PREFIX}{CLIENT_ID_SUFFIX}"), - "queue client id does not name a hive; not publishing turn state upward" - ); - return None; - }; - Some(format!("{SUBJECT_PREFIX}.{hive}.{agent}")) - } - } +/// The subject this agent's header goes to. +fn subject(agent: &str) -> String { + format!("{SUBJECT_PREFIX}.{agent}") } -/// Start the publish task, if this agent has a queue credential and a label -/// to name its subject with. -/// -/// Returns without spawning in every other case — no queue or no label — each -/// of which is a legal state for an agent rather than an error, and each -/// logged once here rather than per transition. +/// Start the publish task, if this agent has a label to name its subject with. pub fn spawn(bus: &Bus) { - if !crate::swarm_queue::configured() { - // `swarm_queue::init` already said why at boot; repeating it here - // would be the same fact logged twice. - return; - } let agent = crate::identity::label(); if agent.is_empty() { tracing::warn!("this agent has no label; not publishing turn state upward"); @@ -230,12 +173,8 @@ pub fn spawn(bus: &Bus) { /// any added later, funnels into the comparison and costs nothing when it /// changes nothing. async fn run(mut rx: broadcast::Receiver, bus: Bus, agent: String) { - let Some(Connection { client, presented }) = crate::swarm_queue::client().await else { - return; - }; - let Some(subject) = subject(&presented, &agent) else { - return; - }; + let client = crate::swarm_queue::client().await; + let subject = subject(&agent); tracing::info!(subject, "publishing agent turn state to the swarm queue"); // The last payload actually sent, so a rebuild that changed nothing is @@ -321,7 +260,7 @@ async fn publish( #[cfg(test)] mod tests { - use super::{AgentStateMsg, Presented, hive_from_client_id, subject}; + use super::{AgentStateMsg, subject}; use crate::events::TurnState; use swarm_queue_client::wanted::AgentState; @@ -367,63 +306,13 @@ mod tests { } } - #[test] - fn a_client_id_names_the_hive_between_the_prefix_and_the_suffix() { - assert_eq!(hive_from_client_id("hive-alpha-agent"), Some("alpha")); - // A hive whose own name contains the suffix still resolves: only the - // trailing one is stripped, so the responder and this agree. - assert_eq!( - hive_from_client_id("hive-alpha-agent-agent"), - Some("alpha-agent") - ); - } - - /// Each way an id can fail to name a hive. A guessed subject would be - /// refused by the broker, and a refusal reaches an operator as a header - /// that never fills in rather than as an error, so none of these may - /// fall back. - #[test] - fn an_id_of_another_shape_names_no_hive() { - for id in [ - // The hive's own id, not an agent's. - "hive-alpha", - // Missing the prefix the responder keys on. - "alpha-agent", - // Prefix and suffix but nothing between them: the subject would - // be `$SWARM.agent-state..`, whose empty token matches no - // grant. - "hive--agent", - // Prefix and suffix overlapping with no hive at all. - "hive-agent", - "", - ] { - assert_eq!(hive_from_client_id(id), None, "{id} must name no hive"); - } - } - - /// The subject the swarm side subscribes to, built end to end from the - /// two things that identify this publisher. `swarm-controller`'s relay + /// The subject the swarm side subscribes to. `swarm-controller`'s relay /// spells the same prefix in its own constant and the two never meet, so - /// this pins the half that lives here. + /// this pins the half that lives here. The agent's own credential is + /// granted the subject naming the agent alone, with no hive in it. #[test] - fn the_subject_is_the_prefix_then_the_hive_then_the_agent() { - let presented = Presented::Hive { - client_id: "hive-alpha-agent".to_owned(), - }; - assert_eq!( - subject(&presented, "mara").as_deref(), - Some("$SWARM.agent-state.alpha.mara") - ); - } - - /// An agent that connected with its own credential publishes on the - /// subject that credential is granted, which names no hive. - #[test] - fn an_agent_on_its_own_credential_publishes_on_its_hive_free_subject() { - assert_eq!( - subject(&Presented::Agent, "mara").as_deref(), - Some("$SWARM.agent-state.mara") - ); + fn the_subject_is_the_prefix_then_the_agent() { + assert_eq!(subject("mara"), "$SWARM.agent-state.mara"); } /// The published contract, asserted on the JSON a subscriber parses diff --git a/hive-agent/src/swarm_input.rs b/hive-agent/src/swarm_input.rs index 238dc5b1..75acbe04 100644 --- a/hive-agent/src/swarm_input.rs +++ b/hive-agent/src/swarm_input.rs @@ -13,8 +13,7 @@ //! the sender from the swarm gateway's authelia answer. //! //! **Own credential only.** The queue grants subscribe on this subject to the -//! agent's own credential alone; the hive's shared client names no agent, so -//! it cannot be granted one agent's input without every agent's. +//! agent's own credential alone. //! //! **Core NATS, no reply.** A message sent while this task is not subscribed //! is gone; the controller sends it as a request so the operator gets a 503 @@ -25,14 +24,8 @@ use futures_util::{Stream, StreamExt as _}; use hive_sh4re::inbox::DeliveredMessage; use tokio::sync::mpsc::UnboundedSender; -use crate::swarm_queue::{Connection, Presented}; - -/// Start the subscriber, if this agent has a queue credential and a label to -/// name its subject with. +/// Start the subscriber, if this agent has a label to name its subject with. pub fn spawn(tx: UnboundedSender) { - if !crate::swarm_queue::configured() { - return; - } let agent = crate::identity::label(); if agent.is_empty() { tracing::warn!("this agent has no label; not taking input from the swarm queue"); @@ -42,16 +35,7 @@ pub fn spawn(tx: UnboundedSender) { } async fn run(tx: UnboundedSender, agent: String) { - let Some(Connection { client, presented }) = crate::swarm_queue::client().await else { - return; - }; - if presented != Presented::Agent { - tracing::warn!( - "connected with the hive's shared client, which may not subscribe to an \ - agent's input subject; not taking input from the swarm queue" - ); - return; - } + let client = crate::swarm_queue::client().await; let subject = swarm_queue_client::agent_input_subject(&agent); let subscriber = match client.subscribe(subject.clone()).await { Ok(subscriber) => subscriber, diff --git a/hive-agent/src/swarm_queue.rs b/hive-agent/src/swarm_queue.rs index 607eb1cb..b81206c2 100644 --- a/hive-agent/src/swarm_queue.rs +++ b/hive-agent/src/swarm_queue.rs @@ -1,54 +1,29 @@ -//! This agent's swarm-queue credentials, resolved once at boot. +//! This agent's swarm-queue connection, made with the agent's own credential. //! -//! Three of the four coordinates arrive as environment variables the meta -//! flake renders into the harness unit. The fourth — the OIDC client id — -//! arrives as a *file*, delivered beside the secret as one systemd -//! credential pair (`nix/agent-modules/queue.nix`). That is the whole reason -//! this module exists rather than a bare [`QueueConfig::from_env`] call: the -//! id rides with the secret so a reader never has to spell `hive--agent` -//! a second time, and `from_env` wants it as a value. +//! The credential is minted per agent at swarm level. This process reads it +//! from the swarm secret store under the agent's own certificate and holds it +//! in memory only ([`AgentSecret`]), and publishes on the agent's own +//! hive-free subjects. //! -//! Reading the file and assigning the variable would put the same rule back -//! in `from_env`'s hands, but `std::env::set_var` is unsound in a process that -//! has already spawned threads, and this one has. So the rule is restated -//! here over the inputs this consumer actually has, and [`decide`] is the -//! single place it lives. -//! -//! Beside all four sits a fifth coordinate, resolved by [`store_source`] and -//! not part of their group. The four are the *hive's* — one OIDC client -//! shared by every container on it — so at the queue's auth callout they say -//! which hive is connecting and never which agent. The fifth is this agent's -//! own, minted per agent at swarm level. This process reads it from the swarm -//! secret store under the agent's own certificate and holds it in memory only -//! ([`AgentSecret`]). -//! -//! When the store holds one, [`client`] connects with it first, and the agent -//! publishes on its own hive-free subjects. When it holds none, or the queue -//! refuses it, the first connect falls back to the hive's shared client and -//! the hive-scoped subjects. [`Connection::presented`] says which, so a -//! publisher builds the subject that credential is granted. +//! **The connection is required.** [`init`] fails when the queue's address or +//! the store's coordinates are missing, and [`client`] retries the first +//! connect on [`CONNECT_BACKOFF`] and then exits the process, so the harness +//! unit fails and systemd restarts it. -use std::path::{Path, PathBuf}; use std::sync::{Arc, Mutex, OnceLock}; use std::time::Duration; use anyhow::{Context as _, anyhow}; use tokio::sync::OnceCell; -use swarm_queue_client::{ClientSecret, QueueConfig}; use swarm_secret_client::{ SecretStore, client::{DEFAULT_CERT_MOUNT, ENV_ADDR, ENV_CACERT, Settings}, policy, queue, }; -/// Variable prefix for this agent's coordinates. Distinct from `HIVE_C0RE`'s -/// on purpose: an agent authenticates as its own client, not as its hive. -const ENV_PREFIX: &str = "HIVE_AGENT"; - -/// Resolved once at boot, so the publisher that connects to the queue reads -/// one answer rather than re-deriving it per call. -static CONFIG: OnceLock> = OnceLock::new(); +/// Where the swarm queue listens, as this container reaches it. +const ENV_NATS_URL: &str = "HIVE_AGENT_NATS_URL"; /// Names the agent whose queue secret this process reads. The cert-auth role /// and the store path are both built from it by `swarm_secret_client`. @@ -59,19 +34,29 @@ const ENV_AGENT_NAME: &str = "HIVE_AGENT_NAME"; /// every reconnect behind it. const STORE_READ_TIMEOUT: Duration = Duration::from_secs(10); -/// Where to present this agent's own credential, resolved once at boot. -static AGENT: OnceLock> = OnceLock::new(); +/// The waits between first-connect attempts: 5s doubling, about five minutes +/// in all. After the last attempt fails the process exits. +const CONNECT_BACKOFF: [Duration; 6] = [ + Duration::from_secs(5), + Duration::from_secs(10), + Duration::from_secs(20), + Duration::from_secs(40), + Duration::from_secs(80), + Duration::from_secs(160), +]; + +/// Where to connect, resolved once at boot by [`init`]. +static AGENT: OnceLock = OnceLock::new(); /// The one connection every publisher in this process shares. Separate from -/// [`CONFIG`] because resolving the coordinates is synchronous boot work and +/// [`AGENT`] because resolving the coordinates is synchronous boot work and /// connecting is not — see [`client`]. -static CLIENT: OnceCell> = OnceCell::const_new(); +static CLIENT: OnceCell = OnceCell::const_new(); /// What connecting with this agent's own credential needs. #[derive(Debug, PartialEq, Eq)] struct AgentPath { url: String, - ca_file: Option, store: Arc, } @@ -150,8 +135,8 @@ impl AgentSecret { .unwrap_or_else(std::sync::PoisonError::into_inner) } - /// Read the secret before the first connection attempt. An `Err` here is - /// what sends [`connect_once`] to the hive's shared client. + /// Read the secret before the first connection attempt. An `Err` here + /// fails that attempt. async fn prime(&self) -> anyhow::Result<()> { let value = self .read() @@ -216,109 +201,31 @@ impl AgentSecret { } } -/// Which credential the shared connection presented. -#[derive(Debug, Clone, PartialEq, Eq)] -pub enum Presented { - /// This agent's own, granted `.`. - Agent, - /// The hive's shared client, granted `..>` for the hive the - /// client id names. - Hive { client_id: String }, -} - -/// The shared queue connection and the credential it was made with. -#[derive(Clone)] -pub struct Connection { - pub client: async_nats::Client, - pub presented: Presented, -} - -/// The four variables the harness unit sets, before the client-id file is -/// read. Collected into a struct so [`decide`] is pure over them and the -/// process env is touched in exactly one place. -struct QueueEnv { - nats_url: Option, - token_endpoint: Option, - client_id_file: Option, - client_secret_file: Option, - /// Independent of the all-or-none group below, exactly as in - /// `QueueConfig::from_env`: a swarm behind a publicly-trusted - /// certificate needs no extra anchor. Nothing in this tree sets it for - /// an agent — a container already trusts the swarm root, which - /// `hive_c0re::meta` embeds at build time — so it is here for a - /// deployment that needs a different one, not for ours. - ca_file: Option, -} - -impl QueueEnv { - fn from_env() -> Self { - let var = |suffix: &str| std::env::var(format!("{ENV_PREFIX}_{suffix}")).ok(); - Self { - nats_url: var("NATS_URL"), - token_endpoint: var("OIDC_TOKEN_ENDPOINT"), - client_id_file: var("OIDC_CLIENT_ID_FILE"), - client_secret_file: var("OIDC_CLIENT_SECRET_FILE"), - ca_file: var("OIDC_CA_FILE"), - } - } -} - -/// What the environment plus the client-id file add up to. -enum Resolution { - /// Everything is here; the agent can reach the queue. - Configured(Box), - /// No queue coordinates for this agent, and that is a legal state — carries why. - Absent(&'static str), - /// Some of the environment, not all of it. A deployment bug rather than - /// an absent integration, so it is reported and then survived. - Partial, -} - -/// Read the client id out of the file the credential landed at. -/// -/// `None` for a missing or empty file, which is the ordinary state of a hive -/// whose secret store has nothing published yet — the credential simply is -/// not there, and nspawn forwards nothing. Trailing newline stripped: the -/// reader unit writes one and an id with a newline in it authenticates as -/// nobody. -fn read_client_id(path: &Path) -> Option { - match std::fs::read_to_string(path) { - Ok(raw) => { - let id = raw.trim(); - (!id.is_empty()).then(|| id.to_owned()) - } - Err(e) if e.kind() == std::io::ErrorKind::NotFound => None, - Err(e) => { - tracing::warn!(path = %path.display(), error = %e, "reading the queue client id failed"); - None - } - } -} - -/// Where this agent reads its own queue secret from, or `None` when this -/// container was given no store. +/// Where this agent reads its own queue secret from. /// /// Taken through a lookup so it can be asserted without mutating the process /// environment. The shape follows `hive_matrix_mcp::credential`'s /// `store_settings`, which reads the same variables for the same identity. /// /// # Errors -/// A store address without the agent name or the identity beside it, which -/// is a half-delivered container rather than one without a store. -fn store_source(get: impl Fn(&str) -> Option) -> anyhow::Result> { +/// No store address, or one without the agent name or the identity beside it. +fn store_source(get: impl Fn(&str) -> Option) -> anyhow::Result { if get(ENV_ADDR).is_none_or(|v| v.is_empty()) { - return Ok(None); + anyhow::bail!( + "{ENV_ADDR} is unset or empty, so there is no secret store to read this agent's \ + queue credential from" + ); } let agent = get(ENV_AGENT_NAME) .filter(|v| !v.is_empty()) .with_context(|| format!("{ENV_AGENT_NAME} is unset or empty"))?; let settings = Settings::from_lookup(|k| get(k).filter(|v| k != ENV_CACERT || ca_is_usable(v))) .context("reading the swarm secret store's coordinates from the environment")?; - Ok(Some(StoreSource { + Ok(StoreSource { settings, role: policy::agent_object_name(&agent)?, path: queue::agent_queue_path(&agent)?, - })) + }) } /// Whether the CA bundle at `path` is a file with bytes in it. `BAO_CACERT` @@ -328,196 +235,105 @@ fn ca_is_usable(path: &str) -> bool { std::fs::metadata(path).is_ok_and(|m| m.len() > 0) } -/// Decide whether this agent can connect with its own credential: it needs the -/// queue's address and a store to read its secret from, and nothing of the -/// hive's client. -fn decide_agent_path(env: &QueueEnv, store: Option) -> Option { - Some(AgentPath { - url: env.nats_url.clone()?, - ca_file: env.ca_file.as_ref().map(Into::into), - store: Arc::new(store?), +/// What connecting needs: the queue's address and a store to read this agent's +/// secret from. +/// +/// # Errors +/// Either one missing. +fn agent_path(get: impl Fn(&str) -> Option) -> anyhow::Result { + let url = get(ENV_NATS_URL) + .filter(|v| !v.is_empty()) + .with_context(|| format!("{ENV_NATS_URL} is unset or empty"))?; + Ok(AgentPath { + url, + store: Arc::new(store_source(&get)?), }) } -/// Decide what this agent's queue configuration is, given the environment and -/// whatever the client-id file held. +/// Resolve this agent's queue coordinates and record them for [`client`]. /// -/// All four variables or none, for the reason `QueueConfig::from_env` gives: -/// a half-set environment produces a process that comes up fine and never -/// connects. The client id is graded separately from the four because its -/// file has a legal absence the variables do not — the unit names the path a -/// credential *would* arrive at whether or not one has been published yet. -fn decide(env: &QueueEnv, client_id: Option) -> Resolution { - match ( - env.nats_url.as_ref(), - env.token_endpoint.as_ref(), - env.client_id_file.as_ref(), - env.client_secret_file.as_ref(), - ) { - (None, None, None, None) => Resolution::Absent("this hive has no swarm queue configured"), - (Some(url), Some(token_endpoint), Some(_), Some(secret)) => match client_id { - Some(client_id) => Resolution::Configured(Box::new(QueueConfig { - url: url.clone(), - token_endpoint: token_endpoint.clone(), - client_id, - client_secret: ClientSecret::File(secret.into()), - ca_file: env.ca_file.as_ref().map(Into::into), - })), - None => { - Resolution::Absent("the queue credential has not been published to this hive yet") - } - }, - _ => Resolution::Partial, - } -} - -/// Resolve this agent's queue configuration and record it for later use. -/// -/// Never fails: a harness that cannot reach the queue still serves its -/// operator, its web UI and its turn loop, so every outcome here is a log -/// line and not an exit. -pub fn init() { - let env = QueueEnv::from_env(); - let client_id = env - .client_id_file - .as_deref() - .map(Path::new) - .and_then(read_client_id); - let resolved = match decide(&env, client_id) { - Resolution::Configured(cfg) => { - // url + client id only. The secret is a path in this struct and - // stays one: neither it nor its contents belong in a log. - tracing::info!(url = %cfg.url, client_id = %cfg.client_id, "swarm queue configured"); - Some(*cfg) - } - Resolution::Absent(why) => { - tracing::info!(why, "no swarm queue coordinates for this agent"); - None - } - Resolution::Partial => { - tracing::error!( - prefix = ENV_PREFIX, - "swarm queue half-configured: {ENV_PREFIX}_NATS_URL, \ - {ENV_PREFIX}_OIDC_TOKEN_ENDPOINT, {ENV_PREFIX}_OIDC_CLIENT_ID_FILE and \ - {ENV_PREFIX}_OIDC_CLIENT_SECRET_FILE are set together or not at all — \ - this agent will not connect" - ); - None - } - }; - let _ = CONFIG.set(resolved); - - // Independent of everything above: this agent may hold its own secret on - // a hive whose shared client is not published yet, or the shared client - // and no secret of its own. - let store = store_source(|k| std::env::var(k).ok()).unwrap_or_else(|e| { - tracing::warn!( - error = format!("{e:#}"), - "this agent's secret store coordinates are incomplete, so it cannot read \ - its own swarm queue credential" - ); - None - }); - if let Some(store) = &store { - tracing::info!( - path = %store.path, - "this agent reads its own swarm queue credential from the secret store" - ); - } else { - tracing::info!( - "no secret store for this agent; it is known to the queue by its hive's \ - shared client" - ); - } - let _ = AGENT.set(decide_agent_path(&env, store)); -} - -/// Whether this agent has any credential to reach the queue with. `false` -/// before [`init`] has run. -pub fn configured() -> bool { - config().is_some() || AGENT.get().is_some_and(Option::is_some) -} - -/// What [`init`] resolved, or `None` when this agent has no queue. -/// -/// Borrowed from the `OnceLock` rather than cloned: a caller wants the client -/// id to derive its subject from, and handing out an owned copy of a struct -/// holding a credential path invites it being stored somewhere with a -/// different lifetime than the one place that owns it. -/// -/// `None` before [`init`] has run, which is the same answer as "no queue" and -/// deliberately not a panic — the ordering is a boot detail, and a harness -/// that reordered its boot should lose the queue, not die. -pub fn config() -> Option<&'static QueueConfig> { - CONFIG.get()?.as_ref() +/// # Errors +/// A coordinate is missing. The harness exits on it: an agent without a queue +/// connection is invisible to the operator at swarm level. +pub fn init() -> anyhow::Result<()> { + let agent = agent_path(|k| std::env::var(k).ok()) + .context("resolving this agent's swarm queue coordinates")?; + tracing::info!( + url = %agent.url, + path = %agent.store.path, + "this agent connects to the swarm queue with its own credential from the secret store" + ); + let _ = AGENT.set(agent); + Ok(()) } /// The shared queue connection, made on first call and memoized for the rest /// of the process. /// /// One connection per process, not per publisher: the agent authenticates as -/// one client, so a second `connect` would be a second token mint and a second -/// live connection for the same identity rather than a second credential. +/// one client, so a second `connect` would be a second live connection for the +/// same identity rather than a second credential. /// -/// `None` covers both "no queue coordinates" and "configured but the connect -/// failed" — a caller does nothing differently between them, since either way -/// there is nothing to publish onto. Connecting is lazy so that an agent on a -/// hive with no queue pays nothing at boot. -pub async fn client() -> Option { - CLIENT.get_or_init(connect_once).await.clone() +/// The first call retries the connect on [`CONNECT_BACKOFF`], and every other +/// caller waits on it. When no attempt succeeds the process exits with status +/// 1, so this never returns without a client. +pub async fn client() -> async_nats::Client { + CLIENT.get_or_init(connect_or_exit).await.clone() } -/// This agent's own credential first, then the hive's. Each path taken is -/// logged once, here. -async fn connect_once() -> Option { - if let Some(agent) = AGENT.get().and_then(Option::as_ref) { - let secret = AgentSecret::new(Arc::clone(&agent.store)); - let name = crate::identity::label(); - match connect_as_agent(&agent.url, agent.ca_file.as_deref(), name, secret).await { - Ok(client) => { - tracing::info!( - url = %agent.url, - "connected to the swarm queue with this agent's own credential" - ); - return Some(Connection { - client, - presented: Presented::Agent, - }); - } - Err(e) => tracing::warn!( - error = format!("{e:#}"), - "connecting with this agent's own credential failed; falling back to \ - its hive's shared client" - ), - } - } - let cfg = config()?; - let inbox = swarm_queue_client::client_inbox_prefix(&cfg.client_id); - match swarm_queue_client::connect_with_inbox_prefix(cfg.clone(), inbox).await { - Ok(client) => { - tracing::info!( - url = %cfg.url, - client_id = %cfg.client_id, - "connecting to the swarm queue with the hive's shared client; the \ - client retries in the background until the queue accepts it" - ); - Some(Connection { - client, - presented: Presented::Hive { - client_id: cfg.client_id.clone(), - }, - }) - } - Err(e) => { - // `chain`, not `{:#}`: this is `swarm_queue_client::Error`, whose - // `Display` ignores the alternate flag, so `{:#}` renders the - // headline and drops the cause that says which half failed. - tracing::warn!( - error = %swarm_queue_client::chain(&e), - "swarm queue connect failed; this agent publishes nothing upward" - ); - None - } +async fn connect_or_exit() -> async_nats::Client { + let connected = match AGENT.get() { + Some(agent) => retry_first_connect(|| connect_once(agent), tokio::time::sleep).await, + None => Err(anyhow!("swarm_queue::init has not run")), + }; + connected.unwrap_or_else(|e| { + tracing::error!( + error = format!("{e:#}"), + "no swarm queue connection; exiting, since this agent cannot run without one" + ); + std::process::exit(1) + }) +} + +/// One connection attempt with this agent's own credential. +async fn connect_once(agent: &AgentPath) -> anyhow::Result { + let secret = AgentSecret::new(Arc::clone(&agent.store)); + let name = crate::identity::label(); + let client = connect_as_agent(&agent.url, name, secret).await?; + tracing::info!( + url = %agent.url, + "connected to the swarm queue with this agent's own credential" + ); + Ok(client) +} + +/// Run `connect` until it succeeds, sleeping each wait in [`CONNECT_BACKOFF`] +/// between attempts, and return the last error once every attempt has failed. +async fn retry_first_connect(mut connect: C, mut sleep: S) -> anyhow::Result +where + C: FnMut() -> CFut, + CFut: Future>, + S: FnMut(Duration) -> SFut, + SFut: Future, +{ + let mut waits = CONNECT_BACKOFF.iter(); + let mut attempt = 1; + loop { + let e = match connect().await { + Ok(client) => return Ok(client), + Err(e) => e, + }; + let Some(&wait) = waits.next() else { + return Err(e.context(format!("{attempt} connection attempts failed"))); + }; + tracing::warn!( + error = format!("{e:#}"), + attempt, + retry_in = ?wait, + "connecting to the swarm queue failed" + ); + sleep(wait).await; + attempt += 1; } } @@ -525,18 +341,19 @@ async fn connect_once() -> Option { /// /// The secret is read before the first attempt, and any failure of that read /// or of the first attempt, a refusal included, is returned for -/// [`connect_once`] to fall back on. Once connected, each reconnect attempt +/// [`retry_first_connect`] to retry. Once connected, each reconnect attempt /// takes its secret from [`AgentSecret::for_attempt`]. async fn connect_as_agent( url: &str, - ca_file: Option<&Path>, name: String, secret: AgentSecret, ) -> anyhow::Result { secret.prime().await?; let secret = Arc::new(secret); let agent = name.clone(); - swarm_queue_client::connect_with_token(url, ca_file, &agent, move || { + // No extra trust anchor: the container trusts the swarm root, which + // `hive_c0re::meta` embeds at build time. + swarm_queue_client::connect_with_token(url, None, &agent, move || { let secret = Arc::clone(&secret); let name = name.clone(); // Spawned so the future handed back is only a `JoinHandle`, which is @@ -551,6 +368,7 @@ async fn connect_as_agent( .await .map_err(|e| anyhow!(swarm_queue_client::chain(&e))) } + #[cfg(test)] mod tests { use std::collections::VecDeque; @@ -561,127 +379,25 @@ mod tests { use anyhow::anyhow; use super::{ - AgentPath, AgentSecret, ClientSecret, QueueEnv, Resolution, SecretSource, StoreSource, - connect_as_agent, decide, decide_agent_path, read_client_id, store_source, + AgentPath, AgentSecret, CONNECT_BACKOFF, SecretSource, StoreSource, agent_path, + connect_as_agent, retry_first_connect, store_source, }; - fn env(parts: [Option<&str>; 4]) -> QueueEnv { - let [nats_url, token_endpoint, client_id_file, client_secret_file] = parts; - QueueEnv { - nats_url: nats_url.map(str::to_owned), - token_endpoint: token_endpoint.map(str::to_owned), - client_id_file: client_id_file.map(str::to_owned), - client_secret_file: client_secret_file.map(str::to_owned), - ca_file: None, - } - } - - fn full() -> QueueEnv { - env([ - Some("nats://10.42.0.1:4222"), - Some("https://auth.t.local/api/oidc/token"), - Some("/run/credentials/hive-agent.service/hive-queue-agent-client-id"), - Some("/run/credentials/hive-agent.service/hive-queue-agent-secret"), - ]) - } - - /// The reader unit writes the id with a trailing newline; an id carrying - /// one is a client authelia has never heard of. - #[test] - fn a_client_id_file_is_read_without_its_newline() { - let dir = tempfile::tempdir().expect("tempdir"); - let path = dir.path().join("client_id"); - std::fs::write(&path, "hive-h1-agent\n").expect("write"); - assert_eq!(read_client_id(&path).as_deref(), Some("hive-h1-agent")); - } - - /// Both shapes of "no credential here": never delivered, or delivered - /// empty. Neither is an id, and treating an empty string as one would - /// authenticate as the anonymous client rather than failing. - #[test] - fn a_missing_or_empty_client_id_file_reads_as_no_id() { - let dir = tempfile::tempdir().expect("tempdir"); - let missing = dir.path().join("client_id"); - assert_eq!(read_client_id(&missing), None); - std::fs::write(&missing, "\n").expect("write"); - assert_eq!(read_client_id(&missing), None); - } - - #[test] - fn a_complete_environment_with_a_published_id_configures_the_queue() { - let e = full(); - let Resolution::Configured(cfg) = decide(&e, Some("hive-h1-agent".to_owned())) else { - panic!("expected a configured queue"); - }; - assert_eq!(cfg.url, "nats://10.42.0.1:4222"); - assert_eq!(cfg.client_id, "hive-h1-agent"); - let ClientSecret::File(path) = &cfg.client_secret else { - panic!("the secret stays a path: {:?}", cfg.client_secret); - }; - assert!( - path.ends_with("hive-queue-agent-secret"), - "the secret stays a path: {}", - path.display() - ); - } - - /// A hive that has not been given its queue's coordinates at all. Silence - /// here is correct, and it has to be distinguishable from the half-set - /// case below — that distinction is the only thing that makes the error - /// branch worth logging. - #[test] - fn an_empty_environment_is_no_queue_rather_than_an_error() { - let e = env([None, None, None, None]); - assert!(matches!(decide(&e, None), Resolution::Absent(_))); - } - - /// The credential's own absence. The unit names the path unconditionally - /// once the hive has a queue, so this is the state of every agent on a - /// swarm whose publisher has not run — legal, and not the error branch. - #[test] - fn a_complete_environment_with_no_published_id_is_no_queue() { - let e = full(); - assert!(matches!(decide(&e, None), Resolution::Absent(_))); - } - - /// Each single-variable omission, because the failure a partial set - /// produces is a harness that looks healthy and publishes nothing. - #[test] - fn any_missing_variable_is_a_partial_configuration() { - for drop in 0..4 { - let mut parts = [ - Some("nats://10.42.0.1:4222"), - Some("https://auth.t.local/api/oidc/token"), - Some("/run/credentials/hive-agent.service/hive-queue-agent-client-id"), - Some("/run/credentials/hive-agent.service/hive-queue-agent-secret"), - ]; - parts[drop] = None; - let e = env(parts); - assert!( - matches!( - decide(&e, Some("hive-h1-agent".to_owned())), - Resolution::Partial - ), - "dropping variable {drop} must report a partial configuration" - ); - } - } - - /// A lookup standing in for a container that was handed a store. + /// A lookup standing in for a container that was handed a store and the + /// queue's address. fn with_store(k: &str) -> Option { match k { "BAO_ADDR" => Some("https://bao.t.local:8200".to_owned()), "BAO_CLIENT_CERT" => Some("/run/credentials/x/cert".to_owned()), "BAO_CLIENT_KEY" => Some("/run/credentials/x/key".to_owned()), "HIVE_AGENT_NAME" => Some("a1".to_owned()), + "HIVE_AGENT_NATS_URL" => Some("nats://10.42.0.1:4222".to_owned()), _ => None, } } fn source() -> StoreSource { - store_source(with_store) - .expect("a complete store environment") - .expect("a store address was given") + store_source(with_store).expect("a complete store environment") } #[test] @@ -691,20 +407,19 @@ mod tests { assert_eq!(s.role, "hive-agent-a1"); } - /// No address: this container was given no store, so there is nothing to - /// read from and nothing wrong. #[test] - fn no_store_address_is_no_store() { - let none = store_source(|k| if k == "BAO_ADDR" { None } else { with_store(k) }); - assert!(none.expect("absent is not an error").is_none()); - let empty = store_source(|k| { - if k == "BAO_ADDR" { - Some(String::new()) - } else { - with_store(k) - } - }); - assert!(empty.expect("empty is absent").is_none()); + fn a_missing_store_address_is_an_error() { + for addr in [None, Some(String::new())] { + let e = store_source(|k| { + if k == "BAO_ADDR" { + addr.clone() + } else { + with_store(k) + } + }) + .expect_err("no store to read the credential from"); + assert!(format!("{e:#}").contains("BAO_ADDR"), "{e:#}"); + } } #[test] @@ -727,37 +442,94 @@ mod tests { with_store(k) } }) - .expect("complete") - .expect("given a store"); + .expect("complete"); assert_eq!(with_missing_ca, source()); } - /// The per-agent secret is resolved by a separate switch from the hive's - /// client: an agent whose hive has no queue client can still read its own - /// swarm-minted secret, and needs only the queue's address for it. #[test] - fn an_agent_with_a_store_and_the_queue_address_connects_as_itself() { - let url_only = env([Some("nats://10.42.0.1:4222"), None, None, None]); - assert!(!matches!( - decide(&url_only, None), - Resolution::Configured(_) - )); + fn a_store_and_the_queue_address_are_all_an_agent_connects_with() { assert_eq!( - decide_agent_path(&url_only, Some(source())), - Some(AgentPath { + agent_path(with_store).expect("complete"), + AgentPath { url: "nats://10.42.0.1:4222".to_owned(), - ca_file: None, store: Arc::new(source()), - }) + } ); } #[test] - fn without_a_store_or_the_queue_address_an_agent_does_not_connect_as_itself() { - assert_eq!(decide_agent_path(&full(), None), None); + fn without_a_store_or_the_queue_address_an_agent_has_nothing_to_connect_with() { + for var in ["HIVE_AGENT_NATS_URL", "BAO_ADDR"] { + let e = agent_path(|k| if k == var { None } else { with_store(k) }) + .expect_err("a coordinate is missing"); + assert!(format!("{e:#}").contains(var), "dropping {var}: {e:#}"); + } + } + + /// Runs `retry_first_connect` against a connect that fails `failures` + /// times and then succeeds, recording each wait instead of sleeping it. + async fn retry_with(failures: usize) -> (anyhow::Result, usize, Vec) { + let mut calls = 0; + let mut waits = Vec::new(); + let result = retry_first_connect( + || { + calls += 1; + let n = calls; + async move { + if n > failures { + Ok(n) + } else { + Err(anyhow!("attempt {n} refused")) + } + } + }, + |wait| { + waits.push(wait); + async {} + }, + ) + .await; + (result, calls, waits) + } + + #[tokio::test] + async fn a_first_connect_that_succeeds_waits_for_nothing() { + let (result, calls, waits) = retry_with(0).await; + assert_eq!(result.expect("connected"), 1); + assert_eq!(calls, 1); + assert!(waits.is_empty(), "{waits:?}"); + } + + /// 5s doubling, one wait per failed attempt, and the attempt after the + /// last wait is the one that connects. + #[tokio::test] + async fn a_failed_connect_is_retried_on_a_doubling_backoff() { + let (result, calls, waits) = retry_with(3).await; + assert_eq!(result.expect("connected on the fourth attempt"), 4); + assert_eq!(calls, 4); assert_eq!( - decide_agent_path(&env([None, None, None, None]), Some(source())), - None + waits, + [5, 10, 20].map(Duration::from_secs), + "each wait doubles the one before" + ); + } + + /// About five minutes of waits, then the last attempt's error rather than + /// another wait. + #[tokio::test] + async fn the_retry_gives_up_with_the_last_error_after_about_five_minutes() { + let (result, calls, waits) = retry_with(usize::MAX).await; + let e = result.expect_err("every attempt failed"); + assert_eq!(calls, CONNECT_BACKOFF.len() + 1); + assert_eq!(waits, CONNECT_BACKOFF); + let total: Duration = waits.iter().sum(); + assert!( + (Duration::from_mins(4)..=Duration::from_mins(6)).contains(&total), + "{total:?}" + ); + assert!( + format!("{e:#}").contains(&format!("attempt {calls} refused")), + "{e:#}" ); } @@ -875,18 +647,17 @@ mod tests { .expect_err("the dropped value does not come back"); } - /// (c) Nothing to connect with before the first attempt is an error, which - /// is what sends `connect_once` to the hive's shared client. The queue is - /// never dialled: nothing listens at the address given. + /// (c) Nothing to connect with before the first attempt fails that attempt. + /// The queue is never dialled: nothing listens at the address given. #[tokio::test] - async fn no_secret_before_the_first_connect_falls_back() { + async fn no_secret_before_the_first_connect_fails_the_attempt() { for (answer, says) in [ (Answer::Fails, "unreachable"), (Answer::Absent, "nothing is stored"), (Answer::Value(" \n"), "nothing is stored"), ] { let secret = AgentSecret::new(scripted([answer])); - let e = connect_as_agent("nats://127.0.0.1:9", None, "a1".to_owned(), secret) + let e = connect_as_agent("nats://127.0.0.1:9", "a1".to_owned(), secret) .await .expect_err("no secret to present"); assert!(format!("{e:#}").contains(says), "{e:#}"); @@ -957,7 +728,7 @@ mod tests { ]); let secret = AgentSecret::new(Arc::clone(&store)); let url = format!("nats://127.0.0.1:{port}"); - let _client = connect_as_agent(&url, None, "a1".to_owned(), secret) + let _client = connect_as_agent(&url, "a1".to_owned(), secret) .await .expect("the first connect is accepted"); diff --git a/hive-agent/src/swarm_term.rs b/hive-agent/src/swarm_term.rs index 4a8c686f..b6222ae8 100644 --- a/hive-agent/src/swarm_term.rs +++ b/hive-agent/src/swarm_term.rs @@ -19,15 +19,13 @@ //! same way it would have missed it on the agent's own SSE stream. That also //! keeps the agent's grant to a plain publish. //! -//! **Best-effort in every direction.** The turn loop and the web UI must not -//! notice whether the queue exists, so the bus receiver is the only thing this -//! task blocks on; a failed publish is a log line and the next row is still -//! attempted. +//! **Best-effort per row.** The turn loop and the web UI never wait on the +//! queue, so the bus receiver is the only thing this task blocks on; a failed +//! publish is a log line and the next row is still attempted. use tokio::sync::broadcast; use crate::events::BusEvent; -use crate::swarm_queue::{Connection, Presented}; use crate::term_msg::{ClassifyCtx, TermMsg, classify}; use hive_sh4re::term_msg::fit; @@ -36,18 +34,6 @@ use hive_sh4re::term_msg::fit; /// can follow one agent without filtering the swarm's whole terminal traffic. const SUBJECT_PREFIX: &str = "$SWARM.term"; -/// What the queue's client id looks like either side of the hive's own name. -/// -/// These mirror the auth-callout responder's `--hive-client-prefix` and -/// `--agent-client-suffix`, which is where the authoritative pair lives: the -/// responder parses the hive back out of the presented client id and grants -/// exactly `$SWARM.term..>`. The agent is not told those flags, so -/// it restates their defaults — a deployment that retunes either one has to -/// change them here too, and the symptom of not doing so is every publish -/// refused rather than a wrong subject being accepted. -const CLIENT_ID_PREFIX: &str = "hive-"; -const CLIENT_ID_SUFFIX: &str = "-agent"; - /// How much room to leave under the announced limit for everything the /// publish adds around the payload — subject, headers, protocol framing. /// @@ -57,56 +43,13 @@ const CLIENT_ID_SUFFIX: &str = "-agent"; /// out of eight megabytes buys that margin for nothing. const HEADROOM: usize = 1024; -/// The hive a queue client id names. -/// -/// The hive in the subject has to be the string the **responder** parses out -/// of this same client id, because that is what its grant is built from. The -/// harness knows a hive display name too, from a different source and with no -/// rule tying the two together — deriving the subject from that one instead -/// produces a publish the broker refuses, which surfaces as a terminal that is -/// simply empty at the swarm. -/// -/// `None` for an id that does not have the expected shape: a publisher that -/// guessed at a subject would be asking for a grant it cannot have, so the -/// caller disables itself instead. -fn hive_from_client_id(client_id: &str) -> Option<&str> { - let hive = client_id - .strip_prefix(CLIENT_ID_PREFIX)? - .strip_suffix(CLIENT_ID_SUFFIX)?; - (!hive.is_empty()).then_some(hive) +/// The subject this agent's rows go to. +fn subject(agent: &str) -> String { + format!("{SUBJECT_PREFIX}.{agent}") } -/// The subject this agent's rows go to under the credential it connected -/// with, or `None` when a hive client id names no hive. -fn subject(presented: &Presented, agent: &str) -> Option { - match presented { - Presented::Agent => Some(format!("{SUBJECT_PREFIX}.{agent}")), - Presented::Hive { client_id } => { - let Some(hive) = hive_from_client_id(client_id) else { - tracing::warn!( - %client_id, - expected = format!("{CLIENT_ID_PREFIX}{CLIENT_ID_SUFFIX}"), - "queue client id does not name a hive; not publishing the terminal upward" - ); - return None; - }; - Some(format!("{SUBJECT_PREFIX}.{hive}.{agent}")) - } - } -} - -/// Start the publish task, if this agent has a queue credential and a label -/// to name its subject with. -/// -/// Returns without spawning in every other case — no queue or no label — each -/// of which is a legal state for an agent rather than an error, and each -/// logged once here rather than per row. +/// Start the publish task, if this agent has a label to name its subject with. pub fn spawn(rx: broadcast::Receiver) { - if !crate::swarm_queue::configured() { - // `swarm_queue::init` already said why at boot; repeating it here - // would be the same fact logged twice. - return; - } let agent = crate::identity::label(); if agent.is_empty() { tracing::warn!("this agent has no label; not publishing the terminal upward"); @@ -116,12 +59,8 @@ pub fn spawn(rx: broadcast::Receiver) { } async fn run(mut rx: broadcast::Receiver, agent: String) { - let Some(Connection { client, presented }) = crate::swarm_queue::client().await else { - return; - }; - let Some(subject) = subject(&presented, &agent) else { - return; - }; + let client = crate::swarm_queue::client().await; + let subject = subject(&agent); tracing::info!(subject, "publishing the agent terminal to the swarm queue"); // One context for the task's whole life — see the module doc. @@ -180,7 +119,7 @@ async fn publish(client: &async_nats::Client, subject: &str, msg: TermMsg) { #[cfg(test)] mod tests { - use super::{Presented, hive_from_client_id, subject}; + use super::subject; use crate::events::LiveEvent; use crate::term_msg::{ClassifyCtx, classify}; use hive_sh4re::term_msg::fit; @@ -188,57 +127,11 @@ mod tests { /// The time `classify` renders for the event the test below classifies. const ROW_TS: &str = "2026-09-13T12:35:03Z"; + /// The agent's own credential is granted the subject naming the agent + /// alone, with no hive in it. #[test] - fn a_client_id_names_the_hive_between_the_prefix_and_the_suffix() { - assert_eq!(hive_from_client_id("hive-alpha-agent"), Some("alpha")); - // A hive whose own name contains the suffix still resolves: only the - // trailing one is stripped, so the responder and this agree. - assert_eq!( - hive_from_client_id("hive-alpha-agent-agent"), - Some("alpha-agent") - ); - } - - /// Each way an id can fail to name a hive. A guessed subject would be - /// refused by the broker, and a refusal reaches an operator as an empty - /// terminal rather than as an error, so none of these may fall back. - #[test] - fn an_id_of_another_shape_names_no_hive() { - for id in [ - // The hive's own id, not an agent's. - "hive-alpha", - // Missing the prefix the responder keys on. - "alpha-agent", - // Prefix and suffix but nothing between them: the subject would - // be `$SWARM.term..`, whose empty token matches no grant. - "hive--agent", - // Prefix and suffix overlapping with no hive at all. - "hive-agent", - "", - ] { - assert_eq!(hive_from_client_id(id), None, "{id} must name no hive"); - } - } - - /// Each credential publishes on the subject it is granted: its own under - /// the agent's credential, its hive's under the shared one. - #[test] - fn the_subject_follows_the_credential_the_agent_connected_with() { - assert_eq!( - subject(&Presented::Agent, "mara").as_deref(), - Some("$SWARM.term.mara") - ); - let hive = Presented::Hive { - client_id: "hive-alpha-agent".to_owned(), - }; - assert_eq!( - subject(&hive, "mara").as_deref(), - Some("$SWARM.term.alpha.mara") - ); - let unparseable = Presented::Hive { - client_id: "hive-alpha".to_owned(), - }; - assert_eq!(subject(&unparseable, "mara"), None); + fn the_subject_is_the_prefix_then_the_agent() { + assert_eq!(subject("mara"), "$SWARM.term.mara"); } /// What a subscriber actually receives: the payload this module hands diff --git a/hive-c0re/src/meta.rs b/hive-c0re/src/meta.rs index 45aca967..998e667b 100644 --- a/hive-c0re/src/meta.rs +++ b/hive-c0re/src/meta.rs @@ -530,11 +530,10 @@ const FORWARDED_VARS: &[&str] = &[ /// answer. Half-wiring this map is not a missing nicety; it is a value that /// evaluates fine and is silently wrong. const FORWARDED_VAR_OPTIONS: &[(&str, &str)] = &[ - // Read at build time to decide whether the harness unit declares the - // queue credential at all — see `nix/agent-modules/queue.nix`. An agent - // whose option says no queue coordinates and whose env says otherwise logs the - // partial-config error rather than half-connecting, which is the same - // failure this map's doc describes and the reason both halves are wired. + // Read at build time to decide whether the harness unit imports the hive's + // queue credential at all — see `nix/agent-modules/queue.nix`. Both halves + // wired for this map's own reason: an agent whose option says no queue and + // whose env says otherwise imports no credential for `swarm-logs`. ( "HIVE_AGENT_NATS_URL", "services.hyperhive.agent.queue.natsUrl", diff --git a/nix/agent-modules/agent-service.nix b/nix/agent-modules/agent-service.nix index 9609e2b2..a8254a28 100644 --- a/nix/agent-modules/agent-service.nix +++ b/nix/agent-modules/agent-service.nix @@ -466,6 +466,10 @@ in description = "${binary} harness"; wantedBy = [ "multi-user.target" ]; after = [ "network.target" ]; + # No start limit: a harness waiting out a queue outage must keep + # restarting however long it lasts, not stop at `start-limit-hit`. + # `[Unit]` setting, hence here and not in `serviceConfig`. + startLimitIntervalSec = 0; # `/run/wrappers` before `/run/current-system/sw` so setuid # `sudo` resolves first. Passing the bare prefixes (no trailing # `/bin`) is intentional — see docs pointer above. @@ -536,8 +540,12 @@ in # Pin the journal identity to the binary name (otherwise systemd # derives SyslogIdentifier from the ExecStart basename). SyslogIdentifier = binary; + # The harness exits when it cannot reach the swarm queue + # (`hive-agent`'s `swarm_queue`), and a swarm-wide outage does that to + # every agent at once: 30s keeps those restarts from hammering the + # store and the queue. Restart = "on-failure"; - RestartSec = 2; + RestartSec = 30; # A LOWER OOMScoreAdjust means LESS likely to be killed, so this # negative value puts the harness — and the agent's own `claude`, # which it spawns as a child and which therefore inherits the diff --git a/nix/agent-modules/queue.nix b/nix/agent-modules/queue.nix index f59a1813..782cedc1 100644 --- a/nix/agent-modules/queue.nix +++ b/nix/agent-modules/queue.nix @@ -1,15 +1,17 @@ -# Swarm-queue coordinates for this agent's harness. +# The hive's shared swarm-queue client credential, imported into the harness +# unit's credentials directory for `swarm-logs` (./logs.nix) to authenticate +# with. The harness itself connects to the queue with the agent's own +# credential (./queue-identity.nix) and reads neither file. # -# Three of the four the harness needs are addresses (this file's two options -# plus the secret's path); the fourth, the client id, arrives as a file beside -# the secret so a reader never spells `hive--agent` a second time. +# The credential is two files, the secret and the client id beside it, so a +# reader never spells `hive--agent` a second time. # # ⚠️ The credential arrives as a systemd credential and NOT as a bind mount, -# and the mode is why: the host file is `root:0600` and this unit runs as the +# and the mode is why: the host file is `root:0600` and the reader runs as the # unprivileged agent user. nspawn's `--load-credential` (written by # `hive_c0re::lifecycle::host_config`) is read by the container manager as # root and re-exposed under this unit's own `User=`; a bind would deliver a -# file the harness cannot open. +# file the agent user cannot open. { lib, config, @@ -24,10 +26,9 @@ let secretCredential = "hive-queue-agent-secret"; clientIdCredential = "hive-queue-agent-client-id"; - # Where systemd materialises this unit's credentials. `%d` above is the - # same directory, but `%d` only expands *inside* a unit — a consumer that - # is not one (the `swarm-logs` wrapper, see ./logs.nix) needs the literal, - # and deriving it from the unit name here is what keeps the two from + # Where systemd materialises this unit's credentials. `%d` only expands + # *inside* a unit, so `swarm-logs` (./logs.nix), which is not one, needs the + # literal; deriving it from the unit name here is what keeps the two from # disagreeing about which unit's credentials they mean. credentialsDir = "/run/credentials/hive-agent.service"; in @@ -46,9 +47,10 @@ in container `127.0.0.1` is the agent itself. `null` means this hive has not been given the queue's address for its - agents, and the harness then declares no credential and logs that it - has none. It is deliberately not defaulted to anything: a guessed - address builds fine and talks to the wrong machine. + agents: the harness unit imports no credential, and the harness fails + at boot, since it cannot run without a queue connection. It is + deliberately not defaulted to anything: a guessed address builds fine + and talks to the wrong machine. ''; }; @@ -57,13 +59,12 @@ in default = null; example = "https://auth.example.com/api/oidc/token"; description = '' - The swarm IdP's OAuth2 token endpoint. The harness mints a - `client_credentials` token there and presents it to the queue, which - authenticates it as the client named in the delivered credential. + The swarm IdP's OAuth2 token endpoint, where `swarm-logs` mints a + `client_credentials` token for the client named in the delivered + credential. - Set together with {option}`services.hyperhive.agent.queue.natsUrl` or not at all — - the harness treats a half-set pair as a deployment bug rather than as - "no queue coordinates". + The harness unit imports that credential only when this and + {option}`services.hyperhive.agent.queue.natsUrl` are both set. ''; }; @@ -106,31 +107,11 @@ in # inherits a credential the service *manager* received, which is what # the container manager passed in. `man systemd.exec` also makes that # form non-fatal when the credential is absent, which is exactly the - # behaviour a hive whose publisher has not run yet needs — the unit - # starts, finds no id, and says so. + # behaviour a hive whose publisher has not run yet needs. serviceConfig.LoadCredential = [ secretCredential clientIdCredential ]; - environment = { - # `%d` is `$CREDENTIALS_DIRECTORY`, per-unit and owned by `User=`. - # Same shape hive-c0re's own queue client is handed its secret in - # (`nix/host-modules/hive-c0re/environment.nix`) — the harness reads - # a path and never a value. - HIVE_AGENT_OIDC_CLIENT_SECRET_FILE = "%d/${secretCredential}"; - # The id is public (it is sent to the token endpoint on every - # connection) but still arrives as a path, because it arrives *with* - # the secret. `QueueConfig::from_env` wants it as a value, so the - # harness reads this file itself — see `hive-agent`'s `swarm_queue`. - HIVE_AGENT_OIDC_CLIENT_ID_FILE = "%d/${clientIdCredential}"; - }; - # No `HIVE_AGENT_OIDC_CA_FILE`. The hive's own client needs one because - # the host does not trust the swarm's CA, but an agent does: the meta - # flake embeds the hive CA and the swarm root it is issued under into - # this container's `security.pki.certificateFiles` at build time, and - # reqwest's rustls backend verifies against the system store. A path - # here would need the bundle delivered as a third credential to say - # nothing new. }; }; } diff --git a/nix/module-eval/agent-queue-bao.nix b/nix/module-eval/agent-queue-bao.nix index 3e117387..f529e7fe 100644 --- a/nix/module-eval/agent-queue-bao.nix +++ b/nix/module-eval/agent-queue-bao.nix @@ -40,19 +40,11 @@ let agentNoBao = agentWith { }; - # Both at once, for the property that only exists when both are configured: - # the store certificate never reaches a queue variable. - agentQueueBao = agentWith { - hyperhive.queue.natsUrl = "nats://10.42.0.1:4222"; - hyperhive.queue.tokenEndpoint = "https://auth.t.local/api/oidc/token"; - services.hyperhive.agent.bao.addr = "https://bao.t.local:8200"; - }; - agentBaoIdentity = machine: machine.systemd.services.hive-agent-bao-identity; cases = [ { # Both ids or neither: the secret authenticates nobody without the id it - # belongs to, and the harness refuses to treat one of the two as a queue. + # belongs to, and `swarm-logs` reads both. name = "an agent with queue coordinates imports both halves of its credential"; ok = let @@ -60,31 +52,25 @@ let in builtins.elem "hive-queue-agent-secret" c && builtins.elem "hive-queue-agent-client-id" c; } - { - # `%d` and not a path under the state dir: the host file is `0600` - # root-owned, so the only copy this unprivileged unit can open is the - # one systemd puts in its own credentials directory. - name = "the harness reads its queue credential out of the credentials directory"; - ok = - let - e = (agentHarness agentQueue).environment; - in - e.HIVE_AGENT_OIDC_CLIENT_SECRET_FILE == "%d/hive-queue-agent-secret" - && e.HIVE_AGENT_OIDC_CLIENT_ID_FILE == "%d/hive-queue-agent-client-id"; - } { # An agent built before its hive was handed the queue's address. It # must declare nothing rather than name a credential that never - # arrives — and the harness then reports "no queue coordinates" - # instead of a half-set environment. + # arrives. name = "an agent with no queue coordinates declares no credential"; + ok = !((agentHarness agentNoQueue).serviceConfig ? LoadCredential); + } + { + # The harness exits when it cannot reach the queue. It has to come back + # however long the outage lasts, and slowly enough that every agent of a + # swarm restarting at once does not hammer the store and the queue. + name = "the harness restarts every 30s with no start limit"; ok = let - u = agentHarness agentNoQueue; + u = agentHarness agentQueue; in - !(u.serviceConfig ? LoadCredential) - && !(u.environment ? HIVE_AGENT_OIDC_CLIENT_SECRET_FILE) - && !(u.environment ? HIVE_AGENT_OIDC_CLIENT_ID_FILE); + u.serviceConfig.Restart == "on-failure" + && u.serviceConfig.RestartSec == 30 + && u.startLimitIntervalSec == 0; } { # The three ids `hive_c0re::lifecycle::agent_identity` forwards under. @@ -101,11 +87,11 @@ let && builtins.elem "hive-agent-bao-server-ca" c; } { - # `%d` and not a path under the agent's state dir, for the reason the - # queue arm above gives: the host file is `0600` to the hive daemon, so - # the only copy this unprivileged unit can open is the one systemd puts - # in its own credentials directory. The address is the option's value - # rather than a literal that agrees with it today. + # `%d` and not a path under the agent's state dir: the host file is + # `0600` to the hive daemon, so the only copy this unprivileged unit can + # open is the one systemd puts in its own credentials directory. The + # address is the option's value rather than a literal that agrees with + # it today. name = "the identity check presents its certificate out of the credentials directory"; ok = let @@ -204,20 +190,6 @@ let !(agentBao.systemd.services ? hive-agent-queue-credential) && !((agentHarness agentBao).environment ? HIVE_AGENT_QUEUE_AGENT_SECRET_FILE); } - { - # ⛔ The store certificate authenticates against the store and nothing - # else. With both the store and the queue configured, no queue variable - # may name one of its paths. - name = "the harness never points the queue at the store certificate"; - ok = - let - e = (agentHarness agentQueueBao).environment; - queueVars = lib.filterAttrs ( - n: _: lib.hasPrefix "HIVE_AGENT_OIDC_" n || n == "HIVE_AGENT_NATS_URL" - ) e; - in - queueVars != { } && !(lib.any (lib.hasInfix "hive-agent-bao") (builtins.attrValues queueVars)); - } { # The absence arm. An agent whose swarm gave it no store has nothing to # log in with, so the harness is handed no store coordinates at all. diff --git a/swarm-queue-client/src/lib.rs b/swarm-queue-client/src/lib.rs index e6f6d918..9e17485e 100644 --- a/swarm-queue-client/src/lib.rs +++ b/swarm-queue-client/src/lib.rs @@ -771,22 +771,6 @@ pub fn install_crypto_provider() { /// token expires, the controller keeps serving, its status data quietly stops /// updating, and nothing says so until someone reads a dashboard. pub async fn connect(cfg: QueueConfig) -> Result { - connect_inner(cfg, None).await -} - -/// [`connect`], with reply inboxes under `inbox_prefix`. For a client whose -/// grant limits subscribe to that prefix, such as [`client_inbox_prefix`]. -pub async fn connect_with_inbox_prefix( - cfg: QueueConfig, - inbox_prefix: String, -) -> Result { - connect_inner(cfg, Some(inbox_prefix)).await -} - -async fn connect_inner( - cfg: QueueConfig, - inbox_prefix: Option, -) -> Result { install_crypto_provider(); // Trusts `cfg.ca_file` when set (the swarm's own CA, when the token // endpoint is signed by it) — see `build_http_client`. A timeout too, @@ -871,9 +855,6 @@ async fn connect_inner( if let Some(path) = ca_file { options = options.add_root_certificates(path); } - if let Some(prefix) = inbox_prefix { - options = options.custom_inbox_prefix(prefix); - } let client = options .connect(&url) @@ -896,7 +877,7 @@ async fn connect_inner( /// /// Unlike [`connect`], the first attempt must succeed: a refused or failed /// connect is returned rather than retried in the background, so the caller -/// can fall back to another credential. `ca_file` is the queue's trust anchor, +/// sees it and decides how to retry. `ca_file` is the queue's trust anchor, /// as [`QueueConfig::ca_file`]. pub async fn connect_with_token( url: &str,