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/docs/swarm/README.md b/docs/swarm/README.md index e67bd379..c09e404e 100644 --- a/docs/swarm/README.md +++ b/docs/swarm/README.md @@ -389,6 +389,20 @@ how it gets there. A hive that hasn't been given 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. +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. The `` is the one the agent's client id names, +which is the same string the broker builds its grant from. 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. + +A row too large for the queue's `max_payload` would be refused outright and +take the connection down with it, so the harness drops such a row's body before +sending and leaves a marker in its place; the summary, level and icon still +arrive. A row that's too large even without its body is logged and skipped. + ### Swarm-wide forge webhooks At startup the controller registers two Forgejo hooks pointing at 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)); + } +} diff --git a/hive-c0re/src/swarm_agent_status.rs b/hive-c0re/src/swarm_agent_status.rs index 2514927c..54dd1777 100644 --- a/hive-c0re/src/swarm_agent_status.rs +++ b/hive-c0re/src/swarm_agent_status.rs @@ -35,7 +35,7 @@ const FAILURES_BEFORE_BANNER: u32 = 3; /// /// Shares its connect gate with [`crate::swarm_status::spawn`] — both read /// `crate::swarm_queue::client()`, which connects once per process — so a -/// hive with no queue configured pays for this decision once, not twice. +/// hive with no queue coordinates pays for this decision once, not twice. /// /// Takes no `Coordinator` handle, unlike its sibling: this task only reads /// agent state that already exists on disk / in the container runtime diff --git a/hive-c0re/src/swarm_notices.rs b/hive-c0re/src/swarm_notices.rs index d2b9ceb4..914fcd70 100644 --- a/hive-c0re/src/swarm_notices.rs +++ b/hive-c0re/src/swarm_notices.rs @@ -13,8 +13,8 @@ //! [`swarm_queue_client::notices`] for the stream this publishes into. //! //! **Best-effort, never fatal to the caller.** A hive with no queue -//! configured is a silent no-op (the ordinary case). A hive whose queue -//! is unreachable loses the swarm's visibility of the notice, not the +//! coordinates is a silent no-op (the ordinary case). A hive that cannot +//! reach the queue loses the swarm's visibility of the notice, not the //! host's — `warn!` fires on every failed attempt regardless, and the //! dashboard banners only after [`FAILURES_BEFORE_BANNER`] consecutive //! misses, the same debounce shape [`crate::swarm_status`] uses and for diff --git a/nix/host-modules/swarm-nats.nix b/nix/host-modules/swarm-nats.nix index 341ebb6a..4541fef2 100644 --- a/nix/host-modules/swarm-nats.nix +++ b/nix/host-modules/swarm-nats.nix @@ -677,10 +677,10 @@ in # The most a single publish may be, in bytes, before the # server answers `-ERR 'Maximum Payload Violation'` and # closes the connection — not a truncation, a dropped row - # and a reconnect. This queue is about to carry agent - # terminal rows published whole rather than split, so - # upstream's own default is sized to lose one of those - # rather than merely shorten it. Bounded from above by + # and a reconnect. This queue carries agent terminal rows + # published whole rather than split, so upstream's own + # default is sized to lose one of those rather than merely + # shorten it. Bounded from above by # `max_pending`: nats-server refuses to start once this # exceeds it, and widening that ceiling instead costs # memory per connection, so this stays comfortably under it. @@ -783,6 +783,16 @@ in "--hive-client-prefix ${lib.escapeShellArg autheliaCfg.hiveClientPrefix}" "--agent-client-suffix ${lib.escapeShellArg autheliaCfg.agentClientSuffix}" "--reader-client ${lib.escapeShellArg controllerCfg.queueClientId}" + # What an agent may publish to, `{hive}` standing for the + # hive its client id names. Without it the responder has no + # agent grant to hand out and refuses every agent at CONNECT, + # so the terminal stream below depends on this line existing. + # + # `$$`, not `$`: systemd substitutes `$NAME` in `ExecStart` + # whether or not it is quoted, so a single dollar reaches the + # responder as the empty expansion of an unset `SWARM` and the + # grant silently becomes `.term.{hive}.>`. + "--agent-publish-subject ${lib.escapeShellArg "\$\$SWARM.term.{hive}.>"}" ]; # Every credential arrives by `LoadCredential` and is named # on the command line only as a **path** — `argv` is diff --git a/nix/module-eval.nix b/nix/module-eval.nix index 4cc04261..a754f806 100644 --- a/nix/module-eval.nix +++ b/nix/module-eval.nix @@ -773,6 +773,24 @@ let name = "the queue's payload ceiling is set, not inherited from the server's default"; ok = natsOldPath.containers.swarm-nats.config.services.nats.settings.max_payload == 8388608; } + { + # Reads the RENDERED unit text, not the module's source, because the + # failure this defends against renders perfectly: systemd substitutes + # `$NAME` in `ExecStart` regardless of quoting, so a single dollar + # here hands the responder `.term.{hive}.>` — a grant that parses, is + # accepted, and matches nothing an agent ever publishes to. Asserting + # the doubled dollar is the only way to tell the two apart before + # deploy. The flag's presence is asserted separately so that dropping + # the grant entirely fails as its own arm rather than as an escaping + # complaint. + name = "the responder grants agents their hive's terminal subject, and the dollar survives systemd"; + ok = + let + exec = + natsOldPath.containers.swarm-nats.config.systemd.services.swarm-nats-auth.serviceConfig.ExecStart; + in + lib.hasInfix "--agent-publish-subject " exec && lib.hasInfix "$$SWARM.term.{hive}.>" exec; + } { # Reads the host's tmpfiles rules, not the options: the socket directory # nginx and the container share is created there, so a rename that diff --git a/swarm-queue-client/src/lib.rs b/swarm-queue-client/src/lib.rs index 809304ff..0d353be3 100644 --- a/swarm-queue-client/src/lib.rs +++ b/swarm-queue-client/src/lib.rs @@ -634,6 +634,26 @@ pub fn ensure_connected(client: &async_nats::Client) -> Result<(), Error> { Ok(()) } +/// The largest payload this connection's server will accept, as the server +/// itself announced it at CONNECT. +/// +/// Read it from here rather than from a constant: the limit is set in the +/// queue's own NixOS module, so a client that hardcoded it would be asserting +/// a number it does not own and would keep asserting it after the operator +/// raised it. A publisher that exceeds the limit does not get a shortened +/// message — the server answers `Maximum Payload Violation` and closes the +/// connection, costing the row and a reconnect — so a publisher of +/// variable-size rows must size them against this before handing them over. +/// +/// ⚠️ The value is per connection. Before the first successful CONNECT the +/// client reports the library's own default, which is smaller than any +/// deployment sets; call this after [`ensure_connected`], and the answer is +/// the one this server actually announced. +#[must_use] +pub fn max_payload(client: &async_nats::Client) -> usize { + client.server_info().max_payload +} + /// Connect to the swarm queue, presenting a token on each connection attempt /// and minting a fresh one only when the cached one is near expiry. ///