swarm-logs: an agent's CLI for the swarm log store

An agent can reach VictoriaLogs only through the gateway, and since the
machine query route landed the way to read it has been to hand-roll a
client_credentials token request and a curl, per query. This is the CLI
that closes that: `swarm-logs query '<LogsQL>'`, matched log lines on
stdout, so the answer pipes into grep like any other command's.

Built to the plan posted on the tracker thread: own crate, own
docs/tools reference generated off the clap tree, `query` as the one
verb, and the JSON error body surfaced on a non-200 rather than
swallowed. No `tail`: streaming is a different endpoint with a different
response shape, and folding it in here would be a fatter scope than the
ask.

Minting the token is NOT implemented here — swarm-queue-client already
owns the client_credentials request, its error type and its CA handling,
and a token-endpoint fix has to be findable in one place. What this crate
adds is the agent-shaped half: the client id arrives as a *file* beside
the secret, so nothing outside nix/agent-modules/queue.nix spells
`hive-<name>-agent` twice. That is the same problem hive-agent's
swarm_queue module solves, and swarm-logs/src/auth.rs is its `decide`
restated over this binary's inputs.

⚠️ The plan named one thing to verify empirically before calling the auth
settled: whether authelia's bearer policy for the logs vhost accepts the
agent client's audience. Measured from inside a container: it does not.
The client minted a token fine but with `aud: []` and `scp: []`, asking
for the logs URL as an audience answered `invalid_target`, and presenting
the audience-less token to the gateway answered a bare 401. So
swarm-authelia.nix's agentClients gains `authelia.bearer.authz` and the
query URL as a second audience — authelia authorises a bearer token by
the URL being requested, and that URL is now one binding read by three
places rather than three spellings of one address.

The URL reaches an agent the same way its queue coordinates do: computed
on the host (a container cannot derive a gateway address), forwarded by
hive_c0re::meta into the container's option set, and consumed by a new
agent module that installs the binary *wrapped* with its coordinates —
the shape swarm-controller.nix installs swarmctl in. Gated on the queue
credential as well as on the URL: a binary that can only answer 401 is
worse than no binary, because an agent reads a 401 as "no logs", which is
the exact confusion the store's machine route was added to end.

swarm-logs/src/query.rs

@@ -0,0 +1,215 @@
+//! The one request this binary makes: a LogsQL query against the store's
+//! machine route on the gateway.
+//!
+//! The route is `^~ /select/logsql/`, added by `swarm-victorialogs.nix`
+//! specifically so a machine caller is answered with a **401 it cannot
+//! mistake for an empty result**. The operator's `/` route ends in
+//! `error_page 401 =302` and hands an unauthenticated caller authelia's login
+//! page as a 200 with an HTML body — which is why [`run`] below reports a
+//! non-200 with its body rather than printing nothing and exiting 0. That
+//! failure mode is the reason the route exists; reproducing it in the client
+//! would be the whole bug again one layer up.
+
+use std::io::{BufRead as _, BufReader, Write};
+
+use anyhow::{Context as _, Result, bail};
+
+use crate::auth::Config;
+
+/// How the matched records are written to stdout.
+#[derive(Clone, Copy, Debug, PartialEq, Eq, clap::ValueEnum)]
+pub enum Format {
+    /// One log message per line — the `_msg` field and nothing else.
+    ///
+    /// The default because the ask this CLI answers was to *pipe and grep*
+    /// rather than hand-roll curl, and a grep pattern is written against the
+    /// message. A record with no `_msg` is passed through as its own JSON
+    /// object rather than dropped: silently losing a row would make an
+    /// incomplete answer indistinguishable from a complete one.
+    Message,
+    /// The store's response body, unmodified, one JSON object per line.
+    ///
+    /// For everything [`Format::Message`] leaves out — `_time`, `_stream`
+    /// and whatever fields the record carries — and for feeding `jq`.
+    Json,
+}
+
+/// Run `query`, writing matched records to `out`.
+///
+/// Streams rather than collecting: the route applies no scoping, so a loose
+/// query matches the whole swarm's logs and buffering the answer to print it
+/// is an out-of-memory waiting for a bad `LogsQL`.
+pub fn run(
+    cfg: &Config,
+    logsql: &str,
+    limit: Option<u64>,
+    format: Format,
+    out: &mut impl Write,
+) -> Result<()> {
+    // Minted per invocation, with the query URL as the audience. Both halves
+    // are load-bearing: authelia refuses a token carrying no audience at the
+    // authz endpoint the gateway's `auth_request` calls, and the audience it
+    // checks is the URL being requested — so the string sent here and the
+    // string requested below must be one binding, which is why `Config` holds
+    // exactly one.
+    let token = swarm_queue_client::mint_token_for_blocking(&cfg.queue, Some(&cfg.query_url))
+        .map_err(|e| anyhow::anyhow!("{}", swarm_queue_client::chain(&e)))
+        .context("minting an access token for the swarm log store")?;
+
+    let http = build_http_client(cfg)?;
+
+    let mut form = vec![("query", logsql.to_owned())];
+    if let Some(limit) = limit {
+        form.push(("limit", limit.to_string()));
+    }
+
+    let response = http
+        .post(&cfg.query_url)
+        .bearer_auth(&token)
+        .form(&form)
+        .send()
+        .with_context(|| format!("querying the swarm log store at {}", cfg.query_url))?;
+
+    let status = response.status();
+    if !status.is_success() {
+        // Surfaced, not swallowed. A 401 here is the ordinary shape of "this
+        // client's identity is not authorised for this route", and the body
+        // is the half that says which — nginx answers its own HTML for an
+        // `auth_request` denial, the store answers JSON for a bad LogsQL.
+        // Truncated because the login-page case is a full HTML document and
+        // an operator needs the first screen of it, not all of it.
+        let body = response.text().unwrap_or_default();
+        let body = body.trim();
+        let shown: String = body.chars().take(2000).collect();
+        bail!(
+            "the swarm log store refused the query ({status}): {shown}{}",
+            if shown.len() < body.chars().count() {
+                " […]"
+            } else {
+                ""
+            }
+        );
+    }
+
+    let mut reader = BufReader::new(response);
+    let mut line = String::new();
+    loop {
+        line.clear();
+        let read = reader
+            .read_line(&mut line)
+            .context("reading the query response from the swarm log store")?;
+        if read == 0 {
+            break;
+        }
+        let line = line.trim_end_matches(['\r', '\n']);
+        if line.is_empty() {
+            continue;
+        }
+        writeln!(out, "{}", render(line, format)).context("writing to stdout")?;
+    }
+    Ok(())
+}
+
+/// Render one NDJSON record from the store as the caller asked for it.
+///
+/// Pure over the line so the projection is testable without a store: what
+/// goes wrong here is a record shape, not a request.
+fn render(line: &str, format: Format) -> String {
+    match format {
+        Format::Json => line.to_owned(),
+        Format::Message => serde_json::from_str::<serde_json::Value>(line)
+            .ok()
+            .and_then(|v| v.get("_msg")?.as_str().map(str::to_owned))
+            // Anything without a string `_msg` — including a line that is not
+            // JSON at all — falls back to the line itself. See
+            // [`Format::Message`]: dropping it would make a partial answer
+            // look like a whole one.
+            .unwrap_or_else(|| line.to_owned()),
+    }
+}
+
+/// The HTTP client for the query itself.
+///
+/// Separate from the one `swarm-queue-client` builds for the token endpoint
+/// because they are two hosts — `auth.<swarm>` and `logs.<swarm>` — and the
+/// CA-trust knob is declared per endpoint. Same optional anchor, same reason:
+/// without it a swarm using its own CA fails at TLS with
+/// `invalid peer certificate: UnknownIssuer`, having never looked at the
+/// anchor sitting on the host.
+///
+/// No timeout, deliberately, where the token client has one: a query over a
+/// long retention window legitimately takes longer than any number that would
+/// be right for a token request, and a client-side cut would truncate a
+/// correct answer into a shorter one with no marker saying so.
+fn build_http_client(cfg: &Config) -> Result<reqwest::blocking::Client> {
+    let mut builder = reqwest::blocking::Client::builder();
+    if let Some(path) = &cfg.queue.ca_file {
+        let pem = std::fs::read(path).with_context(|| {
+            format!(
+                "reading the log-store CA certificate from {}",
+                path.display()
+            )
+        })?;
+        let cert = reqwest::Certificate::from_pem(&pem).with_context(|| {
+            format!(
+                "parsing the log-store CA certificate from {} as PEM",
+                path.display()
+            )
+        })?;
+        builder = builder.add_root_certificate(cert);
+    }
+    builder
+        .build()
+        .context("building the log-store HTTP client")
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn json_passes_the_record_through_unmodified() {
+        let line = r#"{"_time":"2026-09-16T14:14:43Z","_msg":"hello","_stream":"{}"}"#;
+        assert_eq!(render(line, Format::Json), line);
+    }
+
+    #[test]
+    fn message_projects_just_the_log_line() {
+        let line = r#"{"_time":"2026-09-16T14:14:43Z","_msg":"hello there"}"#;
+        assert_eq!(render(line, Format::Message), "hello there");
+    }
+
+    /// The grep case this CLI exists for: a marker inside `_msg` has to
+    /// survive the projection intact, punctuation and all.
+    #[test]
+    fn message_keeps_a_marker_intact() {
+        let line = r#"{"_msg":"probe marker-1789568083-6970 emitted"}"#;
+        assert!(render(line, Format::Message).contains("marker-1789568083-6970"));
+    }
+
+    /// A record the projection cannot understand is still a match, and a
+    /// caller counting lines must not get a different answer than one reading
+    /// them. Dropping it would make an incomplete answer look complete.
+    #[test]
+    fn a_record_without_a_message_falls_back_to_the_whole_line() {
+        let line = r#"{"_time":"2026-09-16T14:14:43Z","_stream":"{}"}"#;
+        assert_eq!(render(line, Format::Message), line);
+    }
+
+    #[test]
+    fn a_non_json_line_falls_back_to_itself() {
+        assert_eq!(
+            render("not json at all", Format::Message),
+            "not json at all"
+        );
+    }
+
+    /// `_msg` holding a non-string is the same "shape I cannot project" case
+    /// as a missing one, and must take the same fallback rather than
+    /// rendering Rust's debug form of a number.
+    #[test]
+    fn a_non_string_message_falls_back_to_the_whole_line() {
+        let line = r#"{"_msg":42}"#;
+        assert_eq!(render(line, Format::Message), line);
+    }
+}