52037133e2
An agent's queue credential could subscribe to every subject in the swarm account, so any agent could read every other agent's terminal, turn state, operator input and request replies. Agents are the least trusted clients on the queue (mara's ruling on #4901), so their grants now carry a subscribe allow-list: - an agent's own credential (the harness and its subagent daemon): `_INBOX.agent.<agent>.>` and `$SWARM.input.<agent>` - the hive's shared agent client: `_INBOX.client.<client-id>.>` Hive and swarm-controller grants are unchanged (no `sub` claim, so still every subject). The clients connect with the matching inbox prefix, so the replies they rely on still arrive: the icon write's JetStream ack and the schedule DIRECT.GET. `connect_with_token` takes the agent name and sets the prefix itself; the hive-shared fallback uses the new `connect_with_inbox_prefix`. The harness no longer subscribes to its input subject when it fell back to the hive's shared client: that identity names no agent, so it cannot be granted one agent's input without all of them. It logs that and skips, the same as the schedule reader on that path. Refs #4901
438 lines
16 KiB
Rust
438 lines
16 KiB
Rust
//! Publishing each subagent's terminal rows onto the swarm queue, as this
|
|
//! agent.
|
|
//!
|
|
//! Every line a subagent writes passes through `session`'s sink, which hands
|
|
//! it to [`Publisher::offer`]. The sink runs synchronously on the subagent's
|
|
//! output reader, so `offer` only queues. One task drains the queue: it
|
|
//! classifies each `stream-json` event into the rows the agent's own terminal
|
|
//! carries (`hive_sh4re::stream_enrich`) and publishes them on
|
|
//! `$SWARM.term.<agent>.sub.<subagent>` (`swarm_queue_client::subagent_term`).
|
|
//!
|
|
//! **Output only.** Nothing here subscribes: a subagent takes no input from
|
|
//! the swarm.
|
|
//!
|
|
//! **As the agent, never as the hive.** The connection presents the agent's
|
|
//! own queue credential, read from the swarm secret store under the agent's
|
|
//! store identity: the same credential and identity the harness uses, since
|
|
//! both daemons run as the agent's user. The hive's shared client is never
|
|
//! tried, because the queue grants it nothing under these subjects.
|
|
//!
|
|
//! **Publish only.** The rows land in `term-sub-<agent>`, the stream
|
|
//! `swarm-controller` creates for this agent; this daemon never opens it. A
|
|
//! row published while that stream does not exist yet still reaches a live
|
|
//! subscriber, but the swarm does not list the subagent from it.
|
|
//!
|
|
//! **Best-effort.** A subagent's run never waits on the queue: no store, a
|
|
//! refused credential, a full queue and a failed publish are each a log
|
|
//! line, and the run goes on.
|
|
|
|
use std::collections::HashMap;
|
|
use std::sync::Arc;
|
|
use std::sync::atomic::{AtomicU64, Ordering};
|
|
use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH};
|
|
|
|
use anyhow::{Context as _, anyhow};
|
|
use hive_sh4re::term_msg::{ClassifyCtx, Level, TermMsg, fit, iso8601_utc};
|
|
use swarm_queue_client::subagent_term;
|
|
use swarm_secret_client::{
|
|
SecretStore,
|
|
client::{DEFAULT_CERT_MOUNT, ENV_ADDR, ENV_CACERT, Settings},
|
|
policy, queue,
|
|
};
|
|
use tokio::sync::mpsc;
|
|
|
|
/// Names the agent. The store role, the store path, the queue token and the
|
|
/// subjects are all built from it.
|
|
const ENV_AGENT_NAME: &str = "HIVE_AGENT_NAME";
|
|
|
|
/// Where the swarm queue listens, as this container reaches it.
|
|
const ENV_NATS_URL: &str = "HIVE_AGENT_NATS_URL";
|
|
|
|
/// Lines waiting for the publish task. Past this, `offer` drops the line
|
|
/// rather than block a subagent's output reader.
|
|
const QUEUE_DEPTH: usize = 4096;
|
|
|
|
/// How long one read of the store may take; it runs inside a queue
|
|
/// connection attempt.
|
|
const STORE_READ_TIMEOUT: Duration = Duration::from_secs(10);
|
|
|
|
/// How long after a failed connect the task tries again. Rows that arrive
|
|
/// before a first connect succeeds are dropped.
|
|
const RETRY_AFTER: Duration = Duration::from_mins(1);
|
|
|
|
/// Room left under the server's payload limit for subject, headers and
|
|
/// framing, as in `hive-agent`'s `swarm_term`.
|
|
const HEADROOM: usize = 1024;
|
|
|
|
/// One line of a subagent's output, as the runtime's sink hands it over.
|
|
pub enum Output {
|
|
/// A `stream-json` event.
|
|
Event(serde_json::Value),
|
|
/// A stdout line that was not JSON.
|
|
Stdout(String),
|
|
/// A stderr line.
|
|
Stderr(String),
|
|
}
|
|
|
|
struct Line {
|
|
subagent: String,
|
|
output: Output,
|
|
/// Unix seconds when the sink saw the line.
|
|
ts: i64,
|
|
}
|
|
|
|
/// The sink's handle on the publish task.
|
|
pub struct Publisher {
|
|
tx: mpsc::Sender<Line>,
|
|
dropped: Arc<AtomicU64>,
|
|
}
|
|
|
|
impl Publisher {
|
|
/// Start the publish task, or return `None` when this daemon has no
|
|
/// queue address or no store to read the agent's credential from. Logs
|
|
/// which, once.
|
|
#[must_use]
|
|
pub fn spawn() -> Option<Self> {
|
|
let target = match Target::from_lookup(|k| std::env::var(k).ok()) {
|
|
Ok(Some(target)) => target,
|
|
Ok(None) => {
|
|
tracing::info!(
|
|
"no swarm queue address or no secret store for this agent; subagent \
|
|
terminals are not published"
|
|
);
|
|
return None;
|
|
}
|
|
Err(e) => {
|
|
tracing::warn!(
|
|
error = format!("{e:#}"),
|
|
"this agent's secret store coordinates are incomplete; subagent \
|
|
terminals are not published"
|
|
);
|
|
return None;
|
|
}
|
|
};
|
|
swarm_queue_client::install_crypto_provider();
|
|
tracing::info!(
|
|
url = %target.url,
|
|
stream = %subagent_term::stream_name(&target.agent),
|
|
"publishing subagent terminals to the swarm queue as this agent"
|
|
);
|
|
let (publisher, rx) = Self::channel(QUEUE_DEPTH);
|
|
tokio::spawn(run(rx, target, Arc::clone(&publisher.dropped)));
|
|
Some(publisher)
|
|
}
|
|
|
|
fn channel(depth: usize) -> (Self, mpsc::Receiver<Line>) {
|
|
let (tx, rx) = mpsc::channel(depth);
|
|
let publisher = Self {
|
|
tx,
|
|
dropped: Arc::new(AtomicU64::new(0)),
|
|
};
|
|
(publisher, rx)
|
|
}
|
|
|
|
/// Queue `output` from `subagent` for publishing. Never blocks: a full
|
|
/// queue drops the line and counts it.
|
|
pub fn offer(&self, subagent: &str, output: Output) {
|
|
let ts = SystemTime::now()
|
|
.duration_since(UNIX_EPOCH)
|
|
.map_or(0, |d| i64::try_from(d.as_secs()).unwrap_or(i64::MAX));
|
|
let line = Line {
|
|
subagent: subagent.to_owned(),
|
|
output,
|
|
ts,
|
|
};
|
|
if self.tx.try_send(line).is_err() {
|
|
self.dropped.fetch_add(1, Ordering::Relaxed);
|
|
}
|
|
}
|
|
}
|
|
|
|
/// Where to publish, and what to connect with.
|
|
struct Target {
|
|
url: String,
|
|
agent: String,
|
|
store: StoreSource,
|
|
}
|
|
|
|
/// This agent's queue secret as the store holds it.
|
|
struct StoreSource {
|
|
settings: Settings,
|
|
role: String,
|
|
path: String,
|
|
}
|
|
|
|
impl Target {
|
|
/// `None` when the queue address or the store address is absent, which
|
|
/// is a container without a swarm queue or without a store.
|
|
///
|
|
/// # Errors
|
|
/// A store address without the agent name or the identity beside it.
|
|
fn from_lookup(get: impl Fn(&str) -> Option<String>) -> anyhow::Result<Option<Self>> {
|
|
let Some(url) = get(ENV_NATS_URL).filter(|v| !v.is_empty()) else {
|
|
return Ok(None);
|
|
};
|
|
if get(ENV_ADDR).is_none_or(|v| v.is_empty()) {
|
|
return Ok(None);
|
|
}
|
|
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")?;
|
|
let store = StoreSource {
|
|
settings,
|
|
role: policy::agent_object_name(&agent)?,
|
|
path: queue::agent_queue_path(&agent)?,
|
|
};
|
|
Ok(Some(Self { url, agent, store }))
|
|
}
|
|
|
|
fn subject(&self, subagent: &str) -> String {
|
|
subagent_term::subject(&self.agent, subagent)
|
|
}
|
|
}
|
|
|
|
/// Whether the CA bundle at `path` is a file with bytes in it. Absent means
|
|
/// the container's own trust store.
|
|
fn ca_is_usable(path: &str) -> bool {
|
|
std::fs::metadata(path).is_ok_and(|m| m.len() > 0)
|
|
}
|
|
|
|
impl StoreSource {
|
|
/// The stored secret, read with a fresh login. Errors name the path and
|
|
/// never the value.
|
|
async fn read(&self) -> anyhow::Result<String> {
|
|
let fetch = async {
|
|
let store = SecretStore::connect(&self.settings, &self.role, DEFAULT_CERT_MOUNT)
|
|
.await
|
|
.context("logging in to the swarm secret store as this agent")?;
|
|
let credential: Option<queue::AgentCredential> = store
|
|
.read_optional(&self.path)
|
|
.await
|
|
.with_context(|| format!("reading {} from the store", self.path))?;
|
|
credential
|
|
.map(|c| c.value.trim().to_owned())
|
|
.filter(|v| !v.is_empty())
|
|
.with_context(|| format!("nothing is stored at {}", self.path))
|
|
};
|
|
tokio::time::timeout(STORE_READ_TIMEOUT, fetch)
|
|
.await
|
|
.map_err(|_| anyhow!("the store did not answer within {STORE_READ_TIMEOUT:?}"))?
|
|
}
|
|
}
|
|
|
|
/// Connect presenting the agent's own token, read from the store on every
|
|
/// connection attempt so a re-minted secret is picked up on reconnect.
|
|
async fn connect(target: &Arc<Target>) -> anyhow::Result<async_nats::Client> {
|
|
let url = target.url.clone();
|
|
let target = Arc::clone(target);
|
|
let agent = target.agent.clone();
|
|
swarm_queue_client::connect_with_token(&url, None, &agent, move || {
|
|
let target = Arc::clone(&target);
|
|
// Spawned so the future handed back is only a `JoinHandle`, which is
|
|
// `Sync` as async-nats's auth callback requires; the store read is not.
|
|
let attempt = tokio::spawn(async move {
|
|
let value = target.store.read().await.map_err(|e| format!("{e:#}"))?;
|
|
swarm_queue_client::agent_token::format_agent_token(&target.agent, &value)
|
|
.map_err(|e| e.to_string())
|
|
});
|
|
async move { attempt.await.map_err(|e| e.to_string())? }
|
|
})
|
|
.await
|
|
.map_err(|e| anyhow!(swarm_queue_client::chain(&e)))
|
|
}
|
|
|
|
/// The connection, retried at most once per [`RETRY_AFTER`].
|
|
#[derive(Default)]
|
|
struct Queue {
|
|
client: Option<async_nats::Client>,
|
|
connect_after: Option<Instant>,
|
|
}
|
|
|
|
impl Queue {
|
|
/// A client to publish on, connecting first when a connect is due.
|
|
async fn ready(&mut self, target: &Arc<Target>) -> Option<async_nats::Client> {
|
|
let now = Instant::now();
|
|
if self.client.is_none() && self.connect_after.is_none_or(|t| now >= t) {
|
|
match connect(target).await {
|
|
Ok(client) => {
|
|
tracing::info!(url = %target.url, "subagent terminal: connected as this agent");
|
|
self.client = Some(client);
|
|
}
|
|
Err(e) => {
|
|
tracing::warn!(
|
|
error = format!("{e:#}"),
|
|
retry_in = ?RETRY_AFTER,
|
|
"subagent terminal: connecting with this agent's own queue credential \
|
|
failed; rows are dropped until a connect succeeds"
|
|
);
|
|
self.connect_after = Some(now + RETRY_AFTER);
|
|
}
|
|
}
|
|
}
|
|
self.client.clone()
|
|
}
|
|
}
|
|
|
|
async fn run(mut rx: mpsc::Receiver<Line>, target: Target, dropped: Arc<AtomicU64>) {
|
|
let target = Arc::new(target);
|
|
let mut ctxs: HashMap<String, ClassifyCtx> = HashMap::new();
|
|
let mut queue = Queue::default();
|
|
while let Some(line) = rx.recv().await {
|
|
let missed = dropped.swap(0, Ordering::Relaxed);
|
|
if missed > 0 {
|
|
tracing::warn!(missed, "subagent terminal: queue full, lines dropped");
|
|
}
|
|
let ctx = ctxs.entry(line.subagent.clone()).or_default();
|
|
let rows = classify(&line.output, line.ts, ctx);
|
|
if rows.is_empty() {
|
|
continue;
|
|
}
|
|
let Some(client) = queue.ready(&target).await else {
|
|
continue;
|
|
};
|
|
let subject = target.subject(&line.subagent);
|
|
for row in rows {
|
|
publish(&client, &subject, row).await;
|
|
}
|
|
}
|
|
}
|
|
|
|
/// The rows one line of output renders as, each stamped with `ts`.
|
|
fn classify(output: &Output, ts: i64, ctx: &mut ClassifyCtx) -> Vec<TermMsg> {
|
|
let rows = match output {
|
|
Output::Event(v) => hive_sh4re::stream_enrich::classify_stream_value(v, ctx),
|
|
Output::Stdout(line) => vec![TermMsg::new(Level::Debug, line.clone())],
|
|
Output::Stderr(line) => vec![TermMsg::new(Level::Warn, format!("stderr: {line}"))],
|
|
};
|
|
let ts = iso8601_utc(ts);
|
|
rows.into_iter().map(|m| m.at(ts.as_str())).collect()
|
|
}
|
|
|
|
/// Offer one row. Every failure is terminal for that row and for nothing else.
|
|
async fn publish(client: &async_nats::Client, subject: &str, msg: TermMsg) {
|
|
// Unconnected, the client buffers rather than fails, and reports the
|
|
// library's default payload limit rather than the server's.
|
|
if let Err(e) = swarm_queue_client::ensure_connected(client) {
|
|
tracing::warn!(error = %swarm_queue_client::chain(&e), "subagent terminal: publish skipped");
|
|
return;
|
|
}
|
|
let limit = swarm_queue_client::max_payload(client).saturating_sub(HEADROOM);
|
|
let summary = msg.summary.clone();
|
|
let Some(msg) = fit(msg, limit) else {
|
|
tracing::warn!(
|
|
summary,
|
|
limit,
|
|
"subagent terminal: row does not fit even without its body, dropped"
|
|
);
|
|
return;
|
|
};
|
|
let payload = match serde_json::to_vec(&msg) {
|
|
Ok(payload) => payload,
|
|
Err(e) => {
|
|
tracing::warn!(error = %e, "subagent terminal: serialising failed, row dropped");
|
|
return;
|
|
}
|
|
};
|
|
if let Err(e) = client.publish(subject.to_owned(), payload.into()).await {
|
|
tracing::warn!(error = %e, "subagent terminal: publish failed, row dropped");
|
|
}
|
|
}
|
|
|
|
#[cfg(test)]
|
|
mod tests {
|
|
use super::*;
|
|
|
|
fn env(pairs: &[(&str, &str)]) -> impl Fn(&str) -> Option<String> {
|
|
let pairs: Vec<(String, String)> = pairs
|
|
.iter()
|
|
.map(|(k, v)| ((*k).to_owned(), (*v).to_owned()))
|
|
.collect();
|
|
move |k| pairs.iter().find(|(n, _)| n == k).map(|(_, v)| v.clone())
|
|
}
|
|
|
|
const STORE: [(&str, &str); 3] = [
|
|
("BAO_ADDR", "https://bao.t.local:8200"),
|
|
("BAO_CLIENT_CERT", "/run/credentials/c"),
|
|
("BAO_CLIENT_KEY", "/run/credentials/k"),
|
|
];
|
|
|
|
fn full() -> Vec<(&'static str, &'static str)> {
|
|
let mut pairs = STORE.to_vec();
|
|
pairs.push((ENV_NATS_URL, "tls://nats.t.local:4222"));
|
|
pairs.push((ENV_AGENT_NAME, "iris"));
|
|
pairs
|
|
}
|
|
|
|
/// Each subagent publishes on its own subject under this agent's family,
|
|
/// the one the queue grants the agent.
|
|
#[test]
|
|
fn a_subagent_publishes_under_the_agents_own_family() {
|
|
let target = Target::from_lookup(env(&full()))
|
|
.expect("complete")
|
|
.expect("configured");
|
|
assert_eq!(target.subject("scout"), "$SWARM.term.iris.sub.scout");
|
|
assert_eq!(target.subject("w-2"), "$SWARM.term.iris.sub.w-2");
|
|
assert_eq!(target.store.path, "swarm/agents/iris/queue");
|
|
}
|
|
|
|
#[test]
|
|
fn without_a_queue_address_or_a_store_nothing_is_published() {
|
|
let no_url: Vec<_> = full()
|
|
.into_iter()
|
|
.filter(|(k, _)| *k != ENV_NATS_URL)
|
|
.collect();
|
|
assert!(Target::from_lookup(env(&no_url)).expect("legal").is_none());
|
|
let no_store: Vec<_> = full()
|
|
.into_iter()
|
|
.filter(|(k, _)| *k != "BAO_ADDR")
|
|
.collect();
|
|
assert!(
|
|
Target::from_lookup(env(&no_store))
|
|
.expect("legal")
|
|
.is_none()
|
|
);
|
|
}
|
|
|
|
#[test]
|
|
fn a_store_without_the_agent_name_is_an_error() {
|
|
let no_name: Vec<_> = full()
|
|
.into_iter()
|
|
.filter(|(k, _)| *k != ENV_AGENT_NAME)
|
|
.collect();
|
|
assert!(Target::from_lookup(env(&no_name)).is_err());
|
|
}
|
|
|
|
/// The sink must never wait on the publisher: a full queue drops the line
|
|
/// and counts it.
|
|
#[test]
|
|
fn a_full_queue_drops_and_counts_rather_than_blocks() {
|
|
let (publisher, _rx) = Publisher::channel(1);
|
|
publisher.offer("scout", Output::Stdout("one".to_owned()));
|
|
publisher.offer("scout", Output::Stdout("two".to_owned()));
|
|
assert_eq!(publisher.dropped.load(Ordering::Relaxed), 1);
|
|
}
|
|
|
|
#[test]
|
|
fn every_kind_of_line_renders_as_stamped_rows() {
|
|
let mut ctx = ClassifyCtx::default();
|
|
let ts = 1_789_302_903;
|
|
let event = serde_json::json!({
|
|
"type": "assistant",
|
|
"message": { "content": [{ "type": "text", "text": "looking" }] }
|
|
});
|
|
let rows = classify(&Output::Event(event), ts, &mut ctx);
|
|
assert_eq!(rows.len(), 1);
|
|
assert_eq!(rows[0].ts, "2026-09-13T12:35:03Z");
|
|
|
|
let rows = classify(&Output::Stderr("boom".to_owned()), ts, &mut ctx);
|
|
assert_eq!(rows[0].level, Level::Warn);
|
|
assert_eq!(rows[0].summary, "stderr: boom");
|
|
|
|
let rows = classify(&Output::Stdout("chatter".to_owned()), ts, &mut ctx);
|
|
assert_eq!(rows[0].level, Level::Debug);
|
|
assert_eq!(rows[0].ts, "2026-09-13T12:35:03Z");
|
|
}
|
|
}
|