diff --git a/Cargo.lock b/Cargo.lock index 6b2e9f51..aa828666 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1646,6 +1646,7 @@ name = "hive-agent" version = "0.1.0" dependencies = [ "anyhow", + "async-nats", "axum", "chrono", "clap", diff --git a/hive-agent/Cargo.toml b/hive-agent/Cargo.toml index 205fba07..14478708 100644 --- a/hive-agent/Cargo.toml +++ b/hive-agent/Cargo.toml @@ -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 diff --git a/hive-agent/src/main.rs b/hive-agent/src/main.rs index 28d371b6..55f99144 100644 --- a/hive-agent/src/main.rs +++ b/hive-agent/src/main.rs @@ -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(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 diff --git a/hive-agent/src/swarm_queue.rs b/hive-agent/src/swarm_queue.rs index ab93fac7..c5d6250a 100644 --- a/hive-agent/src/swarm_queue.rs +++ b/hive-agent/src/swarm_queue.rs @@ -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> = 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> = 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 { + CLIENT.get_or_init(connect_once).await.clone() +} + +async fn connect_once() -> Option { + 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}; diff --git a/hive-agent/src/swarm_term.rs b/hive-agent/src/swarm_term.rs new file mode 100644 index 00000000..187eabaa --- /dev/null +++ b/hive-agent/src/swarm_term.rs @@ -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..>`. 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 { + 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(°raded)? <= 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 { + 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) { + 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}{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, 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..`, 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)); + } +}