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

hive-agent: publish the agent terminal to the swarm queue

Browse source 
The harness has had its queue coordinates since the credential reached
the container, but nothing used them. This offers each terminal row
upward on `$SWARM.term.<hive>.<agent>`, so a swarm-level terminal can
render an agent without reaching into the hive that hosts it.

It publishes the same `TermMsg` the web UI is handed rather than a
second model of the same events, so a new tool or a reclassified event
changes both surfaces together. It subscribes to the event bus rather
than to the SSE handler: the handler classifies per connected browser,
so hanging this off it would mean an agent nobody is watching publishes
nothing. That also means its own long-lived `ClassifyCtx`, since a
publisher restarting its correlation state would lose the `tool_use` →
name mapping a `tool_result` needs to render.

The hive in the subject is derived from the queue client id, not from
the harness's hive display name. Those come from different sources with
no rule tying them together, and the responder builds its grant from the
client id — so deriving it from the display name yields a publish the
broker refuses, reaching an operator as a terminal that is merely empty.
The prefix and suffix that bracket the hive are the responder's flags,
which the agent is not told; it restates their defaults, and the symptom
of a deployment retuning one without changing this is every publish
refused rather than a wrong subject accepted.

Oversize rows degrade in the publisher. Exceeding `max_payload` is not a
truncation: the server refuses the message and closes the connection, so
an oversize publish costs the row, the connection, and the rows racing
behind it through the reconnect. The body is the only unbounded field —
summaries are already trimmed at classification — so it is the field
spent, and the row keeps its icon, level, summary and coalesce key. A
row that does not fit even then is logged and dropped rather than sent.
The limit is read off the connection, so `8388608` stays spelled once in
the queue's own module; size is measured by serializing, because JSON
escaping separates character count from wire length by an unbounded
factor on exactly the rows already near the limit.

Best-effort throughout: no queue, an unparseable client id and a failed
connect each disable the publisher with one log line, and a failed
publish loses its row and nothing else. The turn loop and the web UI
never block on the queue.

Refs #3805
This commit is contained in:
atlas 2026-09-13 11:59:55 +02:00
commit 2cdd7f2ff1
5 changed files with 375 additions and 2 deletions
Download patch file Download diff file Expand all files Collapse all files

1
Cargo.lock generated
View file

@ -1646,6 +1646,7 @@ name = "hive-agent"
version = "0.1.0"
dependencies = [
"anyhow",
"async-nats",
"axum",
"chrono",
"clap",

8
hive-agent/Cargo.toml
View file

@ -9,6 +9,9 @@ workspace = true
[dependencies]
anyhow.workspace = true
# Named directly only for the client type the terminal publisher holds; the
# connect and the credential handling live in `swarm-queue-client` below.
async-nats.workspace = true
axum.workspace = true
chrono.workspace = true
reqwest.workspace = true
@ -31,8 +34,9 @@ rusqlite.workspace = true
schemars.workspace = true
serde.workspace = true
serde_json.workspace = true
# Bare: `kv`/`notices` name buckets and streams this harness opens neither
# end of. All it wants from the crate is `QueueConfig` and, next, a connect.
# Bare: `kv`/`notices` name buckets and streams this harness opens neither end
# of. The terminal publisher is a plain core-subject publish, so it needs the
# connect and the payload limit and nothing from JetStream.
swarm-queue-client.workspace = true
tokio.workspace = true
tokio-stream.workspace = true

4
hive-agent/src/main.rs
View file

@ -29,6 +29,7 @@ mod state_entry_watch;
mod stats;
mod stream_enrich;
mod swarm_queue;
mod swarm_term;
mod term_msg;
mod todo_server;
mod todos;
@ -505,6 +506,9 @@ async fn serve_main<S: Surface>(socket: &Path, poll_ms: u64) -> Result<()> {
harness_state::write_api_key_mode(login::using_api_key());
let login_state = Arc::new(Mutex::new(initial));
let bus = Bus::new();
// Subscribes before anything is emitted, so the swarm sees this agent's
// terminal from its first row rather than from whenever a reader attached.
swarm_term::spawn(bus.subscribe());
// Set by the web UI's `/api/cancel` on a successful SIGINT, read-and-
// cleared by `handle_turn` before building the next wake prompt — see
// `hive_sh4re::inbox::INTERRUPTED_HINT`. Shared between the web server task and

53
hive-agent/src/swarm_queue.rs
View file

@ -17,6 +17,8 @@
use std::path::Path;
use std::sync::OnceLock;
use tokio::sync::OnceCell;
use swarm_queue_client::QueueConfig;
/// Variable prefix for this agent's coordinates. Distinct from `HIVE_C0RE`'s
@ -27,6 +29,11 @@ const ENV_PREFIX: &str = "HIVE_AGENT";
/// one answer rather than re-deriving it per call.
static CONFIG: OnceLock<Option<QueueConfig>> = OnceLock::new();
/// The one connection every publisher in this process shares. Separate from
/// [`CONFIG`] because resolving the coordinates is synchronous boot work and
/// connecting is not — see [`client`].
static CLIENT: OnceCell<Option<async_nats::Client>> = OnceCell::const_new();
/// 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.
@ -158,6 +165,52 @@ pub fn init() {
let _ = CONFIG.set(resolved);
}
/// 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()
}
/// 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.
///
/// `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<async_nats::Client> {
CLIENT.get_or_init(connect_once).await.clone()
}
async fn connect_once() -> Option<async_nats::Client> {
let cfg = config()?;
match swarm_queue_client::connect(cfg.clone()).await {
Ok(client) => Some(client),
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
}
}
}
#[cfg(test)]
mod tests {
use super::{QueueEnv, Resolution, decide, read_client_id};

311
hive-agent/src/swarm_term.rs Normal file
View file

@ -0,0 +1,311 @@
//! Publishing this agent's terminal rows onto the swarm queue.
//!
//! The per-agent web UI already classifies the event bus into [`TermMsg`]
//! rows; this offers the same rows upward so a swarm-level terminal can
//! render an agent without reaching into its hive. It publishes the value the
//! web UI is handed rather than a second model, so a new tool or a
//! reclassified event changes both surfaces at once.
//!
//! **Subscribes to the bus, not to a browser.** The SSE handler builds its
//! rows per connected client, so hanging this off it would mean an agent
//! whose terminal nobody has open publishes nothing. This is its own
//! subscriber with its own long-lived [`ClassifyCtx`] — a publisher that
//! restarted its correlation state per reader would lose the `tool_use` →
//! name mapping a `tool_result` needs.
//!
//! **A core subject, not `JetStream`.** A terminal row has no current value a
//! late reader can ask for, and nothing here is worth the durability of the
//! notices stream: a subscriber that was not listening missed the row, the
//! 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.
use tokio::sync::broadcast;
use crate::events::BusEvent;
use crate::term_msg::{ClassifyCtx, TermMsg, classify};
/// Subject family carrying agent terminal rows, the swarm-wide agreement this
/// publisher holds up its end of. One leaf subject per agent, so a subscriber
/// 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.<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";
/// What a row that lost its body to the payload limit carries instead.
const DROPPED_BODY: &str = "[body dropped: over the queue's payload limit]";
/// How much room to leave under the announced limit for everything the
/// publish adds around the payload — subject, headers, protocol framing.
///
/// The server measures the payload alone, so this is not required for
/// correctness; it is here because the cost of being one byte over is not one
/// lost byte but a refused message and a dropped connection, and a kilobyte
/// 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)
}
/// Bring `msg` under `limit` serialized bytes, or report that it cannot be.
///
/// The body is the only field that carries arbitrary length — a diff, a whole
/// tool result — so it is the only one worth spending: replacing it keeps the
/// row's identity, level and summary, which is what makes the row readable at
/// all, and a reader sees that something was there rather than seeing nothing.
///
/// `None` means even the degraded row does not fit, so the caller reports it
/// instead of publishing. That matters more than it looks: an oversize publish
/// is not truncated by the server, it is refused and the connection is closed,
/// which costs the row *and* every row racing behind it through the reconnect.
fn fit(msg: TermMsg, limit: usize) -> Option<TermMsg> {
if serialized_len(&msg)? <= limit {
return Some(msg);
}
// Rebuilt rather than mutated in place: `body_format` describes the body,
// and a marker left tagged `Diff` renders as a broken diff downstream.
let degraded = TermMsg {
body: Some(DROPPED_BODY.to_owned()),
body_format: None,
..msg
};
(serialized_len(&degraded)? <= limit).then_some(degraded)
}
/// Serialized size of a row, or `None` if it does not serialize at all.
///
/// Measured by serializing rather than estimated from field lengths: the
/// payload is what the server measures, and JSON escaping makes the two differ
/// by an unbounded factor on exactly the rows that are already near the limit.
fn serialized_len(msg: &TermMsg) -> Option<usize> {
serde_json::to_vec(msg).ok().map(|v| v.len())
}
/// Start the publish task, if this agent has both queue coordinates and an
/// identity the subject can be derived from.
///
/// Returns without spawning in every other case — no queue, an unparseable
/// client id, 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.
pub fn spawn(rx: broadcast::Receiver<BusEvent>) {
let Some(cfg) = crate::swarm_queue::config() else {
// `swarm_queue::init` already said why at boot; repeating it here
// would be the same fact logged twice.
return;
};
let Some(hive) = hive_from_client_id(&cfg.client_id) else {
tracing::warn!(
client_id = %cfg.client_id,
expected = format!("{CLIENT_ID_PREFIX}<hive>{CLIENT_ID_SUFFIX}"),
"queue client id does not name a hive; not publishing the terminal upward"
);
return;
};
let agent = crate::identity::label();
if agent.is_empty() {
tracing::warn!("this agent has no label; not publishing the terminal upward");
return;
}
let subject = format!("{SUBJECT_PREFIX}.{hive}.{agent}");
tokio::spawn(run(rx, subject));
}
async fn run(mut rx: broadcast::Receiver<BusEvent>, subject: String) {
let Some(client) = crate::swarm_queue::client().await else {
return;
};
tracing::info!(subject, "publishing the agent terminal to the swarm queue");
// One context for the task's whole life — see the module doc.
let mut ctx = ClassifyCtx::default();
loop {
let event = match rx.recv().await {
Ok(event) => event,
// The bus drops events for a subscriber that falls behind. The
// terminal is a tail rather than a log, so the answer is to say
// how many were missed and keep reading, exactly as the web UI's
// own subscriber does.
Err(broadcast::error::RecvError::Lagged(missed)) => {
tracing::warn!(missed, "swarm terminal: lagged, rows skipped");
continue;
}
// Every sender is gone, so the harness is shutting down.
Err(broadcast::error::RecvError::Closed) => return,
};
for msg in classify(&event.event, &mut ctx) {
publish(&client, &subject, msg).await;
}
}
}
/// 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) {
// An unconnected client does not fail a publish, it buffers it — and the
// limit read below is the library's pre-connect default until the server
// has announced its own, which is smaller than any deployment sets and
// would degrade rows that would have fit.
if let Err(e) = swarm_queue_client::ensure_connected(client) {
tracing::warn!(error = %swarm_queue_client::chain(&e), "swarm 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,
"swarm 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, "swarm terminal: serialising failed, row dropped");
return;
}
};
if let Err(e) = client.publish(subject.to_owned(), payload.into()).await {
tracing::warn!(error = %e, "swarm terminal: publish failed, row dropped");
}
}
#[cfg(test)]
mod tests {
use super::{DROPPED_BODY, fit, hive_from_client_id};
use crate::term_msg::{BodyFormat, Level, TermMsg};
#[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..<agent>`, 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");
}
}
#[test]
fn a_row_that_already_fits_is_published_unchanged() {
let msg = TermMsg::new(Level::Info, "turn ok")
.icon("✅")
.body("a short body", Some(BodyFormat::Markdown));
let fitted = fit(msg, 4096).expect("a small row fits");
assert_eq!(fitted.body.as_deref(), Some("a short body"));
assert_eq!(fitted.body_format, Some(BodyFormat::Markdown));
}
/// The case the degrade exists for: a body larger than the limit costs the
/// body and nothing else, and what comes back is actually under the limit
/// rather than merely smaller.
#[test]
fn an_oversize_body_is_replaced_and_the_result_fits() {
let limit = 512;
let msg = TermMsg::new(Level::Info, "Edit(src/main.rs)")
.icon("🔧")
.body("x".repeat(limit * 4), Some(BodyFormat::Diff))
.coalesce("tool-1");
let fitted = fit(msg, limit).expect("dropping the body brings this under the limit");
assert_eq!(fitted.body.as_deref(), Some(DROPPED_BODY));
// The tag describes a body that is no longer there; left set, a reader
// renders the marker as a diff.
assert_eq!(fitted.body_format, None);
// The fields that make the row readable survive.
assert_eq!(fitted.summary, "Edit(src/main.rs)");
assert_eq!(fitted.icon.as_deref(), Some("🔧"));
assert_eq!(fitted.coalesce_key.as_deref(), Some("tool-1"));
assert!(
serde_json::to_vec(&fitted).expect("serialises").len() <= limit,
"the degraded row must be under the limit, not merely smaller"
);
}
/// A row whose summary alone exceeds the limit cannot be degraded into
/// one, and publishing it anyway would cost the connection rather than the
/// row. Reported by the caller, never sent.
#[test]
fn a_row_too_large_even_without_its_body_is_refused() {
let limit = 256;
let msg = TermMsg::new(Level::Warn, "s".repeat(limit * 4)).body("x".repeat(limit), None);
assert!(fit(msg, limit).is_none());
}
/// Control for the test above: the same oversize summary with a body that
/// would fit still refuses, so the refusal is about total size and not
/// about the body having been present.
#[test]
fn the_refusal_is_about_size_rather_than_the_body_being_present() {
let limit = 256;
let msg = TermMsg::new(Level::Warn, "s".repeat(limit * 4));
assert!(fit(msg, limit).is_none());
}
/// JSON escaping is why the size is measured by serializing: a body of
/// quotes serializes to twice its own length, so a row that fits by
/// character count can still be refused on the wire.
#[test]
fn the_limit_is_measured_on_the_serialized_bytes() {
let body = "\"".repeat(200);
let msg = TermMsg::new(Level::Info, "quotes").body(body.clone(), None);
let limit = body.len() + 64;
// Well under the limit as characters, over it once escaped.
assert!(
serde_json::to_vec(&msg).expect("serialises").len() > limit,
"this fixture must be oversize only after escaping"
);
let fitted = fit(msg, limit).expect("dropping the body fits");
assert_eq!(fitted.body.as_deref(), Some(DROPPED_BODY));
}
}