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 hive_types::term_out::sanitize;

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()
        );
    }

    let mut out = std::io::stdout().lock();
    if !emit(&mut out, &format!("watching {name} — Ctrl-C to stop"))? {
        return Ok(());
    }
    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));
        if !drain_frames(&mut pending, &mut out)? {
            return Ok(());
        }
    }
    emit(&mut out, &format!("(stream closed by {name})"))?;
    Ok(())
}

/// Write one line; `Ok(false)` means the reader went away (a closed pipe,
/// as in `watch x | head`), which ends the watch quietly.
fn emit(out: &mut impl std::io::Write, line: &str) -> Result<bool> {
    hive_types::term_out::emit(out, line).context("writing to stdout")
}

/// Print every complete frame in `pending` and remove it from the buffer.
/// `Ok(false)` means the reader went away.
fn drain_frames(pending: &mut String, out: &mut impl std::io::Write) -> Result<bool> {
    while let Some(pos) = pending.find("\n\n") {
        let frame_text = pending[..pos].to_owned();
        pending.drain(..=pos + 1);
        if let Some(line) = render_frame(&frame_text)
            && !emit(out, &line)?
        {
            return Ok(false);
        }
    }
    Ok(true)
}

/// 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/4-implementation/web-ui/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) -> Option<String> {
    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 None;
    }
    let Ok(v) = serde_json::from_str::<serde_json::Value>(&data) else {
        return Some(format!("! {}", sanitize(&data)));
    };
    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-sh4re/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}"),
    };
    Some(sanitize(&line))
}

#[cfg(test)]
mod tests {
    use super::*;

    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 control_characters_in_a_row_never_reach_the_terminal() {
        let row = render_frame(
            "data: {\"kind\":\"stream\",\"_icon\":\"i\",\"_summary\":\"a\\u001b]0;x\\u0007b\\tc\"}",
        )
        .expect("row");
        assert_eq!(row, "i a\u{FFFD}]0;x\u{FFFD}b\tc");
        let note = render_frame("data: {\"kind\":\"note\",\"text\":\"\\u001b[2J\"}").expect("note");
        assert_eq!(note, "· \u{FFFD}[2J");
        let raw = render_frame("data: \u{1b}[2J").expect("raw");
        assert_eq!(raw, "! \u{FFFD}[2J");
    }

    #[test]
    fn complete_frames_are_printed_and_a_partial_one_waits() {
        let mut pending = "data: {\"kind\":\"note\",\"text\":\"a\"}\n\ndata: {\"ki".to_owned();
        let mut out = Vec::new();
        assert!(drain_frames(&mut pending, &mut out).expect("writes"));
        assert_eq!(out, "· a\n".as_bytes());
        assert_eq!(pending, "data: {\"ki");
    }

    #[test]
    fn a_closed_pipe_ends_the_watch_quietly_and_other_write_errors_fail() {
        let frames = "data: {\"kind\":\"note\",\"text\":\"a\"}\n\n";
        let mut pending = frames.to_owned();
        let closed = drain_frames(&mut pending, &mut Failing(std::io::ErrorKind::BrokenPipe));
        assert!(!closed.expect("broken pipe is not an error"));
        let mut pending = frames.to_owned();
        let other = drain_frames(
            &mut pending,
            &mut Failing(std::io::ErrorKind::PermissionDenied),
        );
        assert!(other.is_err());
    }
}