hyperhive/hivectl/src/watch.rs

//! `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}");
}