swarm: present tense + no-queue-coordinates wording

The queue's payload ceiling was justified by what the queue was about to
carry; it carries it now, so the comment says so.

The other two sites say "a hive with no queue configured". The swarm has
exactly one queue and a hive cannot lack it — only its coordinates, its
credential, or its ability to reach it. That wording is already used
everywhere else the absence is named; these two predate it.

The docs section on the agents' queue coordinates stopped at delivering
them and never said what the connection is for. It now names the subject
and the degrade rule, which is the part an operator reading an agent's
terminal at the swarm needs.

Refs #3805

Cargo.lock

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

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.<hive>.<agent>`, one subject per
+agent, so a swarm-level terminal can follow one agent without subscribing to
+the swarm's whole traffic. The `<hive>` 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

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

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<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

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<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};

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.<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));
+    }
+}

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

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

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

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

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.
 ///