swarmctl: agent watch follows an agent's terminal rows

Reads GET /api/agents/{name}/term/stream over the controller socket or the
gateway and prints one line per row. The gateway path uses a client without
the total-request timeout so the stream isn't cut after 30 seconds.

Closes #6160

docs/2-handbook/tools/swarmctl-cli.md

@@ -9,6 +9,7 @@ This document contains the help content for the `swarmctl` command-line program.
 * [`swarmctl agent create`↴](#swarmctl-agent-create)
 * [`swarmctl agent mint-forge-token`↴](#swarmctl-agent-mint-forge-token)
 * [`swarmctl agent set-state`↴](#swarmctl-agent-set-state)
+* [`swarmctl agent watch`↴](#swarmctl-agent-watch)
 * [`swarmctl agent inference`↴](#swarmctl-agent-inference)
 * [`swarmctl agent inference get`↴](#swarmctl-agent-inference-get)
 * [`swarmctl agent inference set`↴](#swarmctl-agent-inference-set)
@@ -74,6 +75,7 @@ Manage agents across the swarm
 * `create` — Queue creation of a new agent on a hive in this swarm
 * `mint-forge-token` — Check one agent's forge token, and mint it if it's missing or stale
 * `set-state` — Declare the state a hive converges an agent to: `up`, `offline`, `paused` or `destroyed`
+* `watch` — Follow an agent's terminal rows live: the newest retained rows, then each new one, one line per row, until Ctrl-C or the stream ends
 * `inference` — Show the model, effort, subagent models and subagent cap the swarm sets for an agent, or change the model, effort and subagent models
 * `permissions` — Show or change the permissions the swarm grants an agent
 * `agents-md` — Show an agent's AGENTS.md, or edit it with a diff
@@ -161,6 +163,26 @@ No approval gate guards this: the caller is root on the controller's host, or lo
 
 
 
+## `swarmctl agent watch`
+
+Follow an agent's terminal rows live: the newest retained rows, then each new one, one line per row, until Ctrl-C or the stream ends.
+
+The swarm-level counterpart of `hivectl agent watch`, read from the swarm-controller rather than from the agent's own web UI, so it works from any machine with `--swarm` and doesn't need the agent running to show what it last did. Fails when the swarm has never heard from the agent.
+
+**Usage:** `swarmctl agent watch [OPTIONS] <AGENT>`
+
+###### **Arguments:**
+
+* `<AGENT>` — Agent name
+
+###### **Options:**
+
+* `--controller-socket <PATH>` — swarm-controller's unix socket.
+
+   Supplied by the nix module that installs this binary, from the same `socketPath` option the daemon binds; falls back to `SWARM_CONTROLLER_SOCKET`.
+
+
+
 ## `swarmctl agent inference`
 
 Show the model, effort, subagent models and subagent cap the swarm sets for an agent, or change the model, effort and subagent models.

swarmctl/README.md

@@ -146,6 +146,20 @@ the swarm UI's state menu uses. The values are `swarm_wire::wanted::AgentState`'
 refuses an agent the swarm hasn't declared on that hive. `destroyed` can't be
 undone, so it needs `--yes`.
 
+## `agent watch`
+
+```console
+# swarmctl agent watch ruth
+watching ruth — Ctrl-C to stop
+🔧 Bash ls
+```
+
+`GET /api/agents/{name}/term/stream` on the swarm-controller: the agent's
+newest retained terminal rows, then each new one, as `<icon> <summary>` lines.
+It ends on Ctrl-C or when the stream closes. Over `--swarm` the body is read
+as it arrives, with no total-request timeout. The header-state stream
+(`/state/stream`) isn't read.
+
 ## `forge make-admin`
 
 ```console

swarmctl/src/agent.rs

@@ -286,7 +286,7 @@ async fn over_socket(
 /// throw it away. Falls back through `title` to the raw body so a response
 /// from something that is *not* the controller is still shown rather than
 /// reduced to a bare number.
-fn describe_error(status: hyper::StatusCode, body: &[u8]) -> String {
+pub(crate) fn describe_error(status: hyper::StatusCode, body: &[u8]) -> String {
     let text = String::from_utf8_lossy(body);
     let text = text.trim();
     let field = |v: &serde_json::Value, k: &str| {
@@ -313,7 +313,7 @@ fn describe_error(status: hyper::StatusCode, body: &[u8]) -> String {
 /// none of them: the daemon is off, it binds elsewhere, or this host is
 /// not the one running it at all. No group-membership case to mention
 /// here — `swarmctl` already runs as root.
-fn connect_hint(socket: &Path) -> String {
+pub(crate) fn connect_hint(socket: &Path) -> String {
     format!(
         "could not connect to swarm-controller at {} — is it running on this host? \
          (`systemctl status swarm-controller`). The path comes from \

swarmctl/src/main.rs

@@ -34,6 +34,7 @@ mod permissions;
 mod remote;
 mod schedule;
 mod users;
+mod watch;
 
 use std::fs::{self, File, Permissions};
 use std::io::{self, Write as _};
@@ -292,6 +293,15 @@ enum AgentVerb {
     /// No approval gate guards this: the caller is root on the controller's
     /// host, or logged in through the swarm domain's SSO gate.
     SetState(AgentSetStateArgs),
+    /// Follow an agent's terminal rows live: the newest retained rows, then
+    /// each new one, one line per row, until Ctrl-C or the stream ends.
+    ///
+    /// The swarm-level counterpart of `hivectl agent watch`, read from the
+    /// swarm-controller rather than from the agent's own web UI, so it works
+    /// from any machine with `--swarm` and doesn't need the agent running to
+    /// show what it last did. Fails when the swarm has never heard from the
+    /// agent.
+    Watch(AgentWatchArgs),
     /// Show the model, effort, subagent models and subagent cap the swarm sets
     /// for an agent, or change the model, effort and subagent models.
     ///
@@ -445,6 +455,14 @@ struct AgentSetStateArgs {
     socket: ControllerSocketArg,
 }
 
+#[derive(Args)]
+struct AgentWatchArgs {
+    /// Agent name.
+    agent: String,
+    #[command(flatten)]
+    socket: ControllerSocketArg,
+}
+
 #[derive(Args)]
 struct AgentCreateArgs {
     /// Name for the new agent: 1–63 characters of `[a-z0-9-]`.
@@ -726,6 +744,12 @@ fn main() -> Result<()> {
             let target = args.socket.resolve(&remote)?;
             agent::set_state(&target, &args.agent, &args.hive, args.state, args.yes)
         }
+        Verb::Agent {
+            command: AgentVerb::Watch(args),
+        } => {
+            let target = args.socket.resolve(&remote)?;
+            watch::watch(&target, &args.agent)
+        }
         Verb::Agent {
             command: AgentVerb::Inference { command },
         } => inference(&remote, command),

swarmctl/src/remote.rs

@@ -80,11 +80,24 @@ impl Swarm {
     /// An HTTP client that never follows a redirect: from the gateway one
     /// means "log in", and following it would land on the login page.
     pub(crate) fn http(&self) -> Result<reqwest::Client> {
-        let mut builder = bounded(
+        self.build_client(bounded(
             reqwest::Client::builder(),
             HTTP_CONNECT_TIMEOUT,
             HTTP_TIMEOUT,
-        );
+        ))
+    }
+
+    /// [`Swarm::http`] without the total-request bound, which would cut a
+    /// stream off after `HTTP_TIMEOUT`. Reaching the gateway is still bounded.
+    fn stream_http(&self) -> Result<reqwest::Client> {
+        self.build_client(
+            reqwest::Client::builder()
+                .redirect(reqwest::redirect::Policy::none())
+                .connect_timeout(HTTP_CONNECT_TIMEOUT),
+        )
+    }
+
+    fn build_client(&self, mut builder: reqwest::ClientBuilder) -> Result<reqwest::Client> {
         if let Some(path) = &self.ca_file {
             let pem = fs::read(path)
                 .with_context(|| format!("reading the CA certificate {}", path.display()))?;
@@ -125,6 +138,21 @@ impl Swarm {
             .await
     }
 
+    /// `GET uri` with the cached login, answered as soon as the headers arrive
+    /// so the caller can read the body as it comes. Auth failures are handled
+    /// as for [`Swarm::request`]; any other status is the caller's to judge.
+    pub(crate) async fn open_stream(&self, uri: &str, what: &str) -> Result<reqwest::Response> {
+        self.send_authed(
+            &self.stream_http()?,
+            hyper::Method::GET,
+            uri,
+            None,
+            what,
+            Some("text/event-stream"),
+        )
+        .await
+    }
+
     async fn request_with(
         &self,
         http: &reqwest::Client,
@@ -133,9 +161,29 @@ impl Swarm {
         body: Option<Vec<u8>>,
         what: &str,
     ) -> Result<(hyper::StatusCode, bytes::Bytes)> {
+        let resp = self
+            .send_authed(http, method, uri, body, what, None)
+            .await?;
+        let status = resp.status();
+        let body = resp.bytes().await.context("reading the response")?;
+        Ok((status, body))
+    }
+
+    async fn send_authed(
+        &self,
+        http: &reqwest::Client,
+        method: hyper::Method,
+        uri: &str,
+        body: Option<Vec<u8>>,
+        what: &str,
+        accept: Option<&str>,
+    ) -> Result<reqwest::Response> {
         let url = self.origin.join(uri).context("joining the request path")?;
         let send = |token: String| {
             let mut req = http.request(method.clone(), url.clone()).bearer_auth(token);
+            if let Some(accept) = accept {
+                req = req.header(reqwest::header::ACCEPT, accept);
+            }
             if let Some(body) = &body {
                 req = req
                     .header(reqwest::header::CONTENT_TYPE, "application/json")
@@ -165,8 +213,7 @@ impl Swarm {
                 self.origin
             );
         }
-        let body = resp.bytes().await.context("reading the response")?;
+        Ok(resp)
-        Ok((status, body))
     }
 
     /// The cached access token, refreshed first if it is about to expire or is
@@ -203,7 +250,7 @@ impl Swarm {
 }
 
 /// `builder` following no redirect, with every request bounded by `connect`
-/// and `total`. No caller streams a response, so a total bound fits them all.
+/// and `total`. Not for a streamed response: `stream_http` has no total bound.
 fn bounded(
     builder: reqwest::ClientBuilder,
     connect: Duration,

swarmctl/src/watch.rs

@@ -0,0 +1,368 @@
+//! `swarmctl agent watch` — follow one agent's terminal rows from the CLI.
+//!
+//! Reads `GET /api/agents/{name}/term/stream`, the SSE relay of the rows the
+//! agent publishes to the swarm queue: the newest retained rows first, then
+//! live ones. This is the swarm-level counterpart of `hivectl agent watch`
+//! and prints one line per row, `<icon> <summary>`, as that verb prints a
+//! tool call.
+//!
+//! The ordinary request path in [`crate::agent`] buffers a whole response, so
+//! the stream has its own: a unix-socket `hyper` request, or a bearer-
+//! authenticated gateway request whose body is read chunk by chunk.
+
+use anyhow::{Context as _, Result, bail};
+use http_body_util::{BodyExt as _, Empty};
+use hyper_util::rt::TokioIo;
+use tokio::net::UnixStream;
+
+use crate::agent::{connect_hint, describe_error, parse_ident};
+use crate::remote::Target;
+
+/// Run `swarmctl agent watch`. Returns when the stream ends or on Ctrl-C.
+pub(crate) fn watch(target: &Target, agent: &str) -> Result<()> {
+    let agent = parse_ident(agent, "agent name")?;
+    let rt = tokio::runtime::Builder::new_current_thread()
+        .enable_all()
+        .build()
+        .context("starting a tokio runtime for the controller request")?;
+    rt.block_on(async {
+        tokio::select! {
+            result = follow(target, &agent) => result,
+            _ = tokio::signal::ctrl_c() => Ok(()),
+        }
+    })
+}
+
+async fn follow(target: &Target, agent: &str) -> Result<()> {
+    let uri = format!("/api/agents/{agent}/term/stream");
+    let mut body = open(target, &uri).await?;
+    let mut out = std::io::stdout().lock();
+    if !emit(&mut out, &format!("watching {agent} — Ctrl-C to stop"))? {
+        return Ok(());
+    }
+    let mut frames = Frames::default();
+    while let Some(chunk) = body.next().await? {
+        for frame in frames.push(&chunk) {
+            if let Some(line) = render_frame(&frame)
+                && !emit(&mut out, &line)?
+            {
+                return Ok(());
+            }
+        }
+    }
+    emit(&mut out, "(stream closed by swarm-controller)")?;
+    Ok(())
+}
+
+/// Write one line and flush. `Ok(false)` means the reader went away (a closed
+/// pipe, as in `watch x | head`), which ends the watch quietly; any other
+/// write error is an error.
+fn emit(out: &mut impl std::io::Write, line: &str) -> Result<bool> {
+    match writeln!(out, "{line}").and_then(|()| out.flush()) {
+        Ok(()) => Ok(true),
+        Err(e) if e.kind() == std::io::ErrorKind::BrokenPipe => Ok(false),
+        Err(e) => Err(e).context("writing to stdout"),
+    }
+}
+
+/// Replace every control character except tab with U+FFFD. Rows are
+/// agent-authored, and an escape sequence printed as-is would act on the
+/// operator's terminal.
+fn sanitize(text: &str) -> String {
+    text.chars()
+        .map(|c| {
+            if c.is_control() && c != '\t' {
+                '\u{FFFD}'
+            } else {
+                c
+            }
+        })
+        .collect()
+}
+
+/// A response body read in pieces, whichever transport carried it.
+enum Body {
+    Socket(hyper::body::Incoming),
+    Gateway(reqwest::Response),
+}
+
+impl Body {
+    /// The next piece of data, or `None` at the end of the stream.
+    async fn next(&mut self) -> Result<Option<bytes::Bytes>> {
+        match self {
+            Self::Socket(body) => {
+                while let Some(frame) = body.frame().await {
+                    let frame = frame.context("reading swarm-controller's event stream")?;
+                    if let Ok(data) = frame.into_data() {
+                        return Ok(Some(data));
+                    }
+                }
+                Ok(None)
+            }
+            Self::Gateway(resp) => resp
+                .chunk()
+                .await
+                .context("reading swarm-controller's event stream"),
+        }
+    }
+}
+
+/// Open the stream, failing with the controller's own error on a non-2xx.
+async fn open(target: &Target, uri: &str) -> Result<Body> {
+    match target {
+        Target::Socket(socket) => {
+            let stream = UnixStream::connect(socket)
+                .await
+                .with_context(|| connect_hint(socket))?;
+            let (mut sender, conn) = hyper::client::conn::http1::handshake(TokioIo::new(stream))
+                .await
+                .with_context(|| format!("HTTP handshake with {}", socket.display()))?;
+            tokio::spawn(conn);
+            let req = hyper::Request::builder()
+                .method(hyper::Method::GET)
+                .uri(uri)
+                .header(hyper::header::HOST, "localhost")
+                .header(hyper::header::ACCEPT, "text/event-stream")
+                .body(Empty::<bytes::Bytes>::new())
+                .context("building the watch request")?;
+            let resp = sender
+                .send_request(req)
+                .await
+                .context("sending the watch request to swarm-controller")?;
+            let status = resp.status();
+            if !status.is_success() {
+                let body = resp
+                    .into_body()
+                    .collect()
+                    .await
+                    .context("reading swarm-controller's response")?
+                    .to_bytes();
+                bail!("{}", describe_error(status, &body));
+            }
+            Ok(Body::Socket(resp.into_body()))
+        }
+        Target::Swarm(swarm) => {
+            let resp = swarm.open_stream(uri, "watch").await?;
+            let status = resp.status();
+            if !status.is_success() {
+                let body = resp
+                    .bytes()
+                    .await
+                    .context("reading the gateway's response")?;
+                bail!("{}", describe_error(status, &body));
+            }
+            Ok(Body::Gateway(resp))
+        }
+    }
+}
+
+/// Splits a byte stream into SSE frames. A chunk boundary has no relation to
+/// a frame boundary, so bytes are held until the blank line that ends one.
+#[derive(Default)]
+struct Frames {
+    pending: Vec<u8>,
+}
+
+impl Frames {
+    /// Add `chunk`; returns every frame it completed, without the separator.
+    fn push(&mut self, chunk: &[u8]) -> Vec<String> {
+        self.pending.extend_from_slice(chunk);
+        let mut out = Vec::new();
+        while let Some(pos) = self.pending.windows(2).position(|w| w == b"\n\n") {
+            let frame: Vec<u8> = self.pending.drain(..pos + 2).take(pos).collect();
+            out.push(String::from_utf8_lossy(&frame).into_owned());
+        }
+        out
+    }
+}
+
+/// One frame as a terminal line: the row's `<icon> <summary>`. A frame with
+/// no `data:` line (a keep-alive comment) prints nothing; a payload that is
+/// not a row prints raw, so a shape this doesn't know stays visible.
+fn render_frame(frame: &str) -> Option<String> {
+    render_raw(frame).map(|line| sanitize(&line))
+}
+
+fn render_raw(frame: &str) -> Option<String> {
+    let data = frame
+        .lines()
+        .filter_map(|l| l.strip_prefix("data:"))
+        .map(str::trim_start)
+        .collect::<Vec<_>>()
+        .join("\n");
+    if data.is_empty() {
+        return None;
+    }
+    let Ok(row) = serde_json::from_str::<serde_json::Value>(&data) else {
+        return Some(format!("! {data}"));
+    };
+    let field = |k: &str| row.get(k).and_then(serde_json::Value::as_str);
+    Some(match (field("icon"), field("summary")) {
+        (Some(icon), Some(summary)) => format!("{icon} {summary}"),
+        (None, Some(summary)) => summary.to_owned(),
+        _ => format!("! {data}"),
+    })
+}
+
+#[cfg(test)]
+mod tests {
+    use super::{Frames, render_frame};
+
+    #[test]
+    fn a_frame_split_across_chunks_is_emitted_once_whole() {
+        let mut frames = Frames::default();
+        assert!(frames.push(b"id: 1\ndata: {\"summ").is_empty());
+        let done = frames.push(b"ary\":\"x\"}\n\nid: 2\nda");
+        assert_eq!(done, vec!["id: 1\ndata: {\"summary\":\"x\"}".to_owned()]);
+        let rest = frames.push(b"ta: y\n\n");
+        assert_eq!(rest, vec!["id: 2\ndata: y".to_owned()]);
+    }
+
+    #[test]
+    fn a_multibyte_char_split_across_chunks_survives() {
+        let bytes = "data: ✅ ok\n\n".as_bytes();
+        let mut frames = Frames::default();
+        assert!(frames.push(&bytes[..8]).is_empty());
+        assert_eq!(frames.push(&bytes[8..]), vec!["data: ✅ ok".to_owned()]);
+    }
+
+    #[test]
+    fn a_row_prints_icon_and_summary() {
+        let line = render_frame(
+            "id: 7\ndata: {\"ts\":\"t\",\"icon\":\"🔧\",\"level\":\"info\",\"summary\":\"Bash ls\"}",
+        );
+        assert_eq!(line.as_deref(), Some("🔧 Bash ls"));
+        let bare = render_frame("data: {\"summary\":\"plain\"}");
+        assert_eq!(bare.as_deref(), Some("plain"));
+    }
+
+    #[test]
+    fn keep_alives_print_nothing_and_unknown_payloads_print_raw() {
+        assert_eq!(render_frame(": keep-alive"), None);
+        assert_eq!(
+            render_frame("data: not json").as_deref(),
+            Some("! not json")
+        );
+        assert_eq!(
+            render_frame("data: {\"x\":1}").as_deref(),
+            Some("! {\"x\":1}")
+        );
+    }
+
+    #[test]
+    fn control_characters_in_a_row_never_reach_the_terminal() {
+        let line = render_frame("data: {\"icon\":\"\\u001b]0;x\\u0007\",\"summary\":\"a\\u001b[2Jb\\tc\\r\\u009bd\\u007f\"}")
+            .expect("row");
+        assert_eq!(
+            line,
+            "\u{FFFD}]0;x\u{FFFD} a\u{FFFD}[2Jb\tc\u{FFFD}\u{FFFD}d\u{FFFD}"
+        );
+        let raw = render_frame("data: \u{1b}[2J").expect("raw");
+        assert_eq!(raw, "! \u{FFFD}[2J");
+    }
+
+    struct Failing(std::io::ErrorKind);
+
+    impl std::io::Write for Failing {
+        fn write(&mut self, _: &[u8]) -> std::io::Result<usize> {
+            Err(self.0.into())
+        }
+
+        fn flush(&mut self) -> std::io::Result<()> {
+            Ok(())
+        }
+    }
+
+    #[test]
+    fn a_closed_pipe_ends_the_watch_quietly_and_other_write_errors_fail() {
+        let mut ok = Vec::new();
+        assert!(super::emit(&mut ok, "x").expect("writes"));
+        assert_eq!(ok, b"x\n");
+        let closed = super::emit(&mut Failing(std::io::ErrorKind::BrokenPipe), "x");
+        assert!(!closed.expect("broken pipe is not an error"));
+        let other = super::emit(&mut Failing(std::io::ErrorKind::PermissionDenied), "x");
+        assert!(other.is_err());
+    }
+
+    /// Serves one canned response on a fresh unix socket and returns its path.
+    fn serve_once(name: &str, response: Vec<&'static [u8]>) -> std::path::PathBuf {
+        use std::io::{Read as _, Write as _};
+        let dir = std::env::temp_dir().join(format!("swarmctl-{name}-{}", std::process::id()));
+        let _ = std::fs::remove_dir_all(&dir);
+        std::fs::create_dir_all(&dir).expect("dir");
+        let path = dir.join("controller.sock");
+        let listener = std::os::unix::net::UnixListener::bind(&path).expect("bind");
+        std::thread::spawn(move || {
+            let (mut conn, _) = listener.accept().expect("accept");
+            let mut seen = Vec::new();
+            let mut buf = [0u8; 512];
+            while !seen.windows(4).any(|w| w == b"\r\n\r\n") {
+                let n = conn.read(&mut buf).expect("read request");
+                seen.extend_from_slice(&buf[..n]);
+            }
+            for part in response {
+                conn.write_all(part).expect("write");
+                conn.flush().expect("flush");
+                std::thread::sleep(std::time::Duration::from_millis(50));
+            }
+        });
+        path
+    }
+
+    fn rt() -> tokio::runtime::Runtime {
+        tokio::runtime::Builder::new_current_thread()
+            .enable_all()
+            .build()
+            .expect("runtime")
+    }
+
+    /// Rows arrive as the server sends them, a frame split across two chunks
+    /// is rendered once, and the end of the stream ends the loop.
+    #[test]
+    fn rows_stream_over_the_socket_until_it_closes() {
+        let socket = serve_once(
+            "watch-ok",
+            vec![
+                b"HTTP/1.1 200 OK\r\ncontent-type: text/event-stream\r\ntransfer-encoding: chunked\r\n\r\n",
+                b"1b\r\nid: 1\ndata: {\"icon\":\"A\",\"su\r\n",
+                b"2e\r\nmmary\":\"one\"}\n\nid: 2\ndata: {\"summary\":\"two\"}\n\n\r\n",
+                b"0\r\n\r\n",
+            ],
+        );
+        let lines = rt().block_on(async {
+            let mut body = super::open(&crate::remote::Target::Socket(socket), "/x")
+                .await
+                .expect("opens");
+            let mut frames = Frames::default();
+            let mut lines = Vec::new();
+            while let Some(chunk) = body.next().await.expect("reads") {
+                lines.extend(frames.push(&chunk).iter().filter_map(|f| render_frame(f)));
+            }
+            lines
+        });
+        assert_eq!(lines, vec!["A one".to_owned(), "two".to_owned()]);
+    }
+
+    /// A non-2xx is an error carrying the controller's own sentence.
+    #[test]
+    fn a_refusal_surfaces_the_controllers_detail() {
+        let body: &'static [u8] =
+            b"{\"status\":404,\"title\":\"Not Found\",\"detail\":\"\\\"ghost\\\" has never reported\"}";
+        let socket = serve_once(
+            "watch-404",
+            vec![
+                b"HTTP/1.1 404 Not Found\r\ncontent-type: application/problem+json\r\nconnection: close\r\n\r\n",
+                body,
+            ],
+        );
+        let err = rt()
+            .block_on(super::open(&crate::remote::Target::Socket(socket), "/x"))
+            .err()
+            .expect("refused");
+        let msg = err.to_string();
+        assert!(
+            msg.contains("404") && msg.contains("has never reported"),
+            "{msg}"
+        );
+    }
+}