//! `hivectl agent 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::::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::>() .join("\n"); if data.is_empty() { return; } let Ok(v) = serde_json::from_str::(&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}"); }