hivectl: add agent <name> watch to follow live events from the CLI

hivectl/src/agents.rs

@@ -1,6 +1,6 @@
 //! `hivectl agent <name>` — everything scoped to ONE managed agent:
 //! container lifecycle over the host admin socket (restart/pause/resume/
-//! spawn/kill/destroy/rebuild/set-parent/set-limits/choom), plus the
+//! spawn/kill/destroy/rebuild/set-parent/set-limits/choom/watch), plus the
 //! `quota` and `subvol` groups, whose handlers live in their own modules.
 //! `agents_list` (`hivectl list-agents`) is the one genuinely hive-wide
 //! read that lives in this module too since it shares the same daemon
@@ -210,6 +210,7 @@ pub(crate) async fn run_agent(socket: &Path, name: &str, cmd: AgentCmd) -> Resul
         AgentCmd::Choom { resume_session } => {
             crate::choom::choom(socket, name, resume_session.as_deref()).await
         }
+        AgentCmd::Watch => crate::watch::watch(socket, name).await,
         // `quota` and `subvol` keep their own modules — these arms are
         // just the reparenting glue that hoists `name` in from the
         // parent `agent <name>` command.

hivectl/src/cli.rs

@@ -534,6 +534,14 @@ pub enum AgentCmd {
         #[arg(long = "resume", value_name = "SESSION")]
         resume_session: Option<String>,
     },
+    /// Follow this agent's live turn/tool-call event stream from the CLI.
+    ///
+    /// Dials the same unix socket the gateway's nginx already proxies
+    /// through (no gateway hop, no daemon round-trip for the stream
+    /// itself) and prints one compact line per event. `Ctrl-C` to stop.
+    /// Requires the agent to be running (its harness must have bound the
+    /// web UI socket).
+    Watch,
     /// This agent's disk accounting + optional quota via btrfs qgroups.
     ///
     /// Needs `hivectl quota-enable` run once hive-wide first. No-op on

hivectl/src/main.rs

@@ -42,6 +42,7 @@ mod wg;
 use wg::{peer_config, require_hive_domain, wg_init, wg_peer, wg_status};
 mod choom;
 mod github;
+mod watch;
 use github::github_set_token;
 mod forge;
 use forge::{forge_create_user, forge_reconcile_config};

hivectl/src/watch.rs

@@ -0,0 +1,135 @@
+//! `hivectl agent <name> watch` — follow an agent's live event stream from
+//! the CLI. Dials the same unix socket the gateway's nginx already
+//! `proxy_pass`es through (`hive_host_sock::agent_web_socket`) directly and
+//! speaks a bare HTTP/1.1 request for the existing `/events/stream` SSE
+//! endpoint over it — no gateway hop, no daemon round-trip for the stream
+//! itself. Same client technique `hive-agent/src/web_ui/proxy.rs` uses for
+//! the reverse direction (an agent proxying out to a unix-socket upstream).
+//!
+//! The daemon socket is only used for the "does this agent exist"
+//! pre-flight — see `choom`'s doc comment for why that check has to be
+//! remote rather than a local stat.
+
+use std::path::Path;
+
+use anyhow::{Context as _, Result, bail};
+use http_body_util::{BodyExt, Empty};
+use hyper_util::rt::TokioIo;
+use tokio::net::UnixStream;
+
+use crate::util::agent_exists;
+
+pub(crate) async fn watch(socket: &Path, name: &str) -> Result<()> {
+    if !agent_exists(socket, name).await? {
+        bail!(
+            "no such agent: '{name}' (no state dir under {}/)",
+            hive_host_sock::AGENTS_ROOT
+        );
+    }
+    let ident = crate::util::parse_ident(name)?;
+    let sock_path = hive_host_sock::agent_web_socket(&ident);
+
+    let stream = UnixStream::connect(&sock_path).await.with_context(|| {
+        format!(
+            "connect to {name}'s web UI socket at {} — is the agent running? \
+             (`hivectl list-agents`; a container that just spawned or rebuilt may \
+             not have bound its socket yet)",
+            sock_path.display()
+        )
+    })?;
+    let io = TokioIo::new(stream);
+    let (mut sender, conn) = hyper::client::conn::http1::handshake(io)
+        .await
+        .with_context(|| format!("HTTP handshake with {name}'s web UI socket"))?;
+    // Drives the connection's I/O in the background; a one-shot GET has
+    // nothing useful to do with the join handle, and a closed connection
+    // surfaces to the caller anyway via the next `frame()` returning `None`
+    // or an error.
+    tokio::spawn(conn);
+
+    let req = hyper::Request::builder()
+        .method(hyper::Method::GET)
+        .uri("/events/stream")
+        .header(hyper::header::HOST, "localhost")
+        .header(hyper::header::ACCEPT, "text/event-stream")
+        .body(Empty::<bytes::Bytes>::new())
+        .context("build /events/stream request")?;
+
+    let resp = sender
+        .send_request(req)
+        .await
+        .with_context(|| format!("request {name}'s /events/stream"))?;
+    if !resp.status().is_success() {
+        bail!(
+            "{name}'s web UI returned {} for /events/stream",
+            resp.status()
+        );
+    }
+
+    println!("watching {name} — Ctrl-C to stop");
+    let mut body = resp.into_body();
+    // SSE frames are separated by a blank line; a chunk boundary from the
+    // socket has no relation to a frame boundary, so buffer raw text until
+    // a full frame (up to the `\n\n` separator) has arrived before parsing.
+    let mut pending = String::new();
+    while let Some(frame) = body.frame().await {
+        let frame = frame.with_context(|| format!("reading {name}'s event stream"))?;
+        let Some(chunk) = frame.data_ref() else {
+            continue;
+        };
+        pending.push_str(&String::from_utf8_lossy(chunk));
+        while let Some(pos) = pending.find("\n\n") {
+            let frame_text = pending[..pos].to_owned();
+            pending.drain(..=pos + 1);
+            render_frame(&frame_text);
+        }
+    }
+    println!("(stream closed by {name})");
+    Ok(())
+}
+
+/// Render one SSE frame (everything up to the blank-line separator,
+/// already stripped) as a single compact terminal line. Not an attempt at
+/// the web UI's full row taxonomy (`docs/terminal-rendering.md`) — that's
+/// presentation for a browser, not a `tail -f`. Falls back to the raw
+/// payload for anything unrecognised rather than silently dropping it, so
+/// a shape this doesn't know about is still visible.
+fn render_frame(frame: &str) {
+    let data: String = frame
+        .lines()
+        .filter_map(|l| l.strip_prefix("data:"))
+        .map(str::trim_start)
+        .collect::<Vec<_>>()
+        .join("\n");
+    if data.is_empty() {
+        return;
+    }
+    let Ok(v) = serde_json::from_str::<serde_json::Value>(&data) else {
+        println!("! {data}");
+        return;
+    };
+    let str_field = |k: &str| v.get(k).and_then(serde_json::Value::as_str);
+    let kind = str_field("kind").unwrap_or("?");
+    let line = match kind {
+        "turn_start" => format!("◆ TURN ← {}", str_field("from").unwrap_or("?")),
+        "turn_end" => {
+            if v.get("ok").and_then(serde_json::Value::as_bool) == Some(true) {
+                "✅ turn ok".to_owned()
+            } else {
+                format!("❌ turn fail — {}", str_field("note").unwrap_or(""))
+            }
+        }
+        "note" => format!("· {}", str_field("text").unwrap_or("")),
+        "status_changed" => format!("· status: {}", str_field("status").unwrap_or("")),
+        // Claude stream-json lines, enriched server-side
+        // (`hive-agent/src/stream_enrich.rs`) with `_icon`/`_summary` so
+        // clients don't need to reimplement the web UI's dispatch logic.
+        "stream" => match (str_field("_icon"), str_field("_summary")) {
+            (Some(icon), Some(summary)) => format!("{icon} {summary}"),
+            (None, Some(summary)) => summary.to_owned(),
+            _ => format!("· {}", str_field("type").unwrap_or("stream")),
+        },
+        other => format!("· {other}"),
+    };
+    println!("{line}");
+}