From 30a61d96dd5809d5e3b46698a52fc108cff90c0d Mon Sep 17 00:00:00 2001 From: atlas Date: Sat, 7 Nov 2026 19:24:22 +0100 Subject: [PATCH] hivectl agent watch: strip control chars, exit quietly on a broken pipe --- hive-types/src/lib.rs | 3 +- hive-types/src/term_out.rs | 67 ++++++++++++++++++++++++++ hivectl/src/watch.rs | 98 +++++++++++++++++++++++++++++++++----- swarmctl/src/watch.rs | 51 ++------------------ 4 files changed, 160 insertions(+), 59 deletions(-) create mode 100644 hive-types/src/term_out.rs diff --git a/hive-types/src/lib.rs b/hive-types/src/lib.rs index fc184260..fca0bd9d 100644 --- a/hive-types/src/lib.rs +++ b/hive-types/src/lib.rs @@ -1,5 +1,5 @@ //! Foundational shared types for the hyperhive workspace: [`Ident`], [`Permission`], [`NixBuilds`], -//! the [`term_msg`] row markers and the [`review`] rule. +//! the [`term_msg`] row markers, the [`term_out`] terminal printing helpers and the [`review`] rule. //! //! A zero-dependency (bar `serde`) leaf crate so every wire-type crate //! (`hive-sh4re`, `hive-host-sock`, `hive-core-agent-sock`) and both binaries @@ -15,6 +15,7 @@ pub use permission::{Permission, deserialize_known}; pub mod review; pub mod term_msg; +pub mod term_out; /// The environment variable nix uses to hand this process the blacklist of /// names that already mean something to the message layer. diff --git a/hive-types/src/term_out.rs b/hive-types/src/term_out.rs new file mode 100644 index 00000000..b4ea4f0a --- /dev/null +++ b/hive-types/src/term_out.rs @@ -0,0 +1,67 @@ +//! Printing agent-authored text to an operator's terminal: `hivectl agent +//! watch` and `swarmctl agent watch` share these two helpers. + +use std::io::{ErrorKind, Write}; + +/// Replace every control character except tab with U+FFFD. Rows are +/// agent-authored, and an escape sequence printed as-is would act on the +/// operator's terminal. +#[must_use] +pub fn sanitize(text: &str) -> String { + text.chars() + .map(|c| { + if c.is_control() && c != '\t' { + '\u{FFFD}' + } else { + c + } + }) + .collect() +} + +/// Write one line and flush. `Ok(false)` means the reader went away (a closed +/// pipe, as in `watch x | head`), which ends a watch quietly. +/// +/// # Errors +/// Any write or flush error other than `BrokenPipe`. +pub fn emit(out: &mut impl Write, line: &str) -> std::io::Result { + match writeln!(out, "{line}").and_then(|()| out.flush()) { + Ok(()) => Ok(true), + Err(e) if e.kind() == ErrorKind::BrokenPipe => Ok(false), + Err(e) => Err(e), + } +} + +#[cfg(test)] +mod tests { + use super::*; + + struct Failing(ErrorKind); + + impl Write for Failing { + fn write(&mut self, _: &[u8]) -> std::io::Result { + Err(self.0.into()) + } + + fn flush(&mut self) -> std::io::Result<()> { + Ok(()) + } + } + + #[test] + fn sanitize_keeps_tab_and_text_and_replaces_other_controls() { + assert_eq!( + sanitize("a\u{1b}[2Jb\tc\r\u{9b}d\u{7f}é"), + "a\u{FFFD}[2Jb\tc\u{FFFD}\u{FFFD}d\u{FFFD}é" + ); + } + + #[test] + fn a_closed_pipe_ends_quietly_and_other_write_errors_fail() { + let mut ok = Vec::new(); + assert!(emit(&mut ok, "x").expect("writes")); + assert_eq!(ok, b"x\n"); + assert!(!emit(&mut Failing(ErrorKind::BrokenPipe), "x").expect("not an error")); + assert!(emit(&mut Failing(ErrorKind::PermissionDenied), "x").is_err()); + } +} diff --git a/hivectl/src/watch.rs b/hivectl/src/watch.rs index a6fa17b9..8446e5e7 100644 --- a/hivectl/src/watch.rs +++ b/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 { + 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 { + 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 { let data: String = frame .lines() .filter_map(|l| l.strip_prefix("data:")) @@ -102,11 +126,10 @@ fn render_frame(frame: &str) { .collect::>() .join("\n"); if data.is_empty() { - return; + return None; } let Ok(v) = serde_json::from_str::(&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 { + 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()); + } } diff --git a/swarmctl/src/watch.rs b/swarmctl/src/watch.rs index 58ba3ba4..80be7f6e 100644 --- a/swarmctl/src/watch.rs +++ b/swarmctl/src/watch.rs @@ -54,30 +54,10 @@ async fn follow(target: &Target, agent: &str) -> Result<()> { Ok(()) } -/// Write one line and flush. `Ok(false)` means the reader went away (a closed -/// pipe, as in `watch x | head`), which ends the watch quietly; any other -/// write error is an error. +/// 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 { - match writeln!(out, "{line}").and_then(|()| out.flush()) { - Ok(()) => Ok(true), - Err(e) if e.kind() == std::io::ErrorKind::BrokenPipe => Ok(false), - Err(e) => Err(e).context("writing to stdout"), - } -} - -/// Replace every control character except tab with U+FFFD. Rows are -/// agent-authored, and an escape sequence printed as-is would act on the -/// operator's terminal. -fn sanitize(text: &str) -> String { - text.chars() - .map(|c| { - if c.is_control() && c != '\t' { - '\u{FFFD}' - } else { - c - } - }) - .collect() + hive_types::term_out::emit(out, line).context("writing to stdout") } /// A response body read in pieces, whichever transport carried it. @@ -180,7 +160,7 @@ impl Frames { /// no `data:` line (a keep-alive comment) prints nothing; a payload that is /// not a row prints raw, so a shape this doesn't know stays visible. fn render_frame(frame: &str) -> Option { - render_raw(frame).map(|line| sanitize(&line)) + render_raw(frame).map(|line| hive_types::term_out::sanitize(&line)) } fn render_raw(frame: &str) -> Option { @@ -261,29 +241,6 @@ mod tests { assert_eq!(raw, "! \u{FFFD}[2J"); } - struct Failing(std::io::ErrorKind); - - impl std::io::Write for Failing { - fn write(&mut self, _: &[u8]) -> std::io::Result { - Err(self.0.into()) - } - - fn flush(&mut self) -> std::io::Result<()> { - Ok(()) - } - } - - #[test] - fn a_closed_pipe_ends_the_watch_quietly_and_other_write_errors_fail() { - let mut ok = Vec::new(); - assert!(super::emit(&mut ok, "x").expect("writes")); - assert_eq!(ok, b"x\n"); - let closed = super::emit(&mut Failing(std::io::ErrorKind::BrokenPipe), "x"); - assert!(!closed.expect("broken pipe is not an error")); - let other = super::emit(&mut Failing(std::io::ErrorKind::PermissionDenied), "x"); - assert!(other.is_err()); - } - /// Serves one canned response on a fresh unix socket and returns its path. fn serve_once(name: &str, response: Vec<&'static [u8]>) -> std::path::PathBuf { use std::io::{Read as _, Write as _};