hivectl agent watch: strip control chars, exit quietly on a broken pipe

hivectl/src/watch.rs

@@ -17,6 +17,8 @@ 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<()> {
@@ -66,7 +68,10 @@ pub(crate) async fn watch(socket: &Path, name: &str) -> Result<()> {
         );
     }
 
-    println!("watching {name} — Ctrl-C to stop");
+    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
@@ -78,23 +83,42 @@ pub(crate) async fn watch(socket: &Path, name: &str) -> Result<()> {
             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);
+        if !drain_frames(&mut pending, &mut out)? {
+            return Ok(());
         }
     }
-    println!("(stream closed by {name})");
+    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) {
+fn render_frame(frame: &str) -> Option<String> {
     let data: String = frame
         .lines()
         .filter_map(|l| l.strip_prefix("data:"))
@@ -102,11 +126,10 @@ fn render_frame(frame: &str) {
         .collect::<Vec<_>>()
         .join("\n");
     if data.is_empty() {
-        return;
+        return None;
     }
     let Ok(v) = serde_json::from_str::<serde_json::Value>(&data) else {
-        println!("! {data}");
-        return;
+        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("?");
@@ -131,5 +154,58 @@ fn render_frame(frame: &str) {
         },
         other => format!("· {other}"),
     };
-    println!("{line}");
+    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());
+    }
 }