acp rpc: log a read error on the agent's stdout/stderr before treating the stream as closed

The reader loops used while-let Ok(Some(line)), so a read error ended them
exactly like EOF with nothing logged. next_line now warns on Err, naming
the stream, and returns None; the close path is unchanged.

hive-runtime/src/acp/rpc.rs

@@ -8,7 +8,7 @@ use std::sync::atomic::{AtomicU64, Ordering};
 use std::sync::{Arc, Mutex, PoisonError};
 
 use serde_json::{Value, json};
-use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader};
+use tokio::io::{AsyncBufRead, AsyncBufReadExt, AsyncWriteExt, BufReader, Lines};
 use tokio::process::{Child, ChildStdin, Command};
 use tokio::sync::{mpsc, oneshot};
 
@@ -85,7 +85,7 @@ impl Connection {
         let stderr_tx = tx.clone();
         tokio::spawn(async move {
             let mut lines = BufReader::new(stderr).lines();
-            while let Ok(Some(line)) = lines.next_line().await {
+            while let Some(line) = next_line(&mut lines, "stderr").await {
                 if stderr_tx.send(Incoming::Stderr(line)).is_err() {
                     break;
                 }
@@ -101,7 +101,7 @@ impl Connection {
         };
         tokio::spawn(async move {
             let mut lines = BufReader::new(stdout).lines();
-            while let Ok(Some(line)) = lines.next_line().await {
+            while let Some(line) = next_line(&mut lines, "stdout").await {
                 reader.dispatch(&line).await;
             }
             // Fail every request still waiting: no reply can come now.
@@ -169,6 +169,21 @@ impl Connection {
     }
 }
 
+/// The next line of the agent's `stream`; `None` at EOF or after a read error,
+/// which is logged. Either way the stream is finished.
+async fn next_line<R: AsyncBufRead + Unpin>(
+    lines: &mut Lines<R>,
+    stream: &'static str,
+) -> Option<String> {
+    match lines.next_line().await {
+        Ok(line) => line,
+        Err(e) => {
+            tracing::warn!(error = %e, stream, "ACP agent stream read failed; treating it as closed");
+            None
+        }
+    }
+}
+
 /// The process group a leader with `pid` heads, as `kill(2)` takes it.
 /// Never 0: `kill(-0, …)` signals the caller's own group.
 fn group_id(pid: Option<u32>) -> Option<i32> {
@@ -338,7 +353,7 @@ pub(super) fn permission_outcome(
 
 #[cfg(test)]
 mod tests {
-    use super::{group_id, permission_outcome};
+    use super::{group_id, next_line, permission_outcome};
     use crate::acp::{PermissionAsk, PermissionPolicy};
     use serde_json::json;
     use std::sync::Arc;
@@ -443,4 +458,33 @@ mod tests {
         assert_eq!(group_id(Some(u32::MAX)), None);
         assert_eq!(group_id(None), None);
     }
+
+    /// Yields one line, then a read error.
+    struct FailingRead(bool);
+
+    impl tokio::io::AsyncRead for FailingRead {
+        fn poll_read(
+            mut self: std::pin::Pin<&mut Self>,
+            _: &mut std::task::Context<'_>,
+            buf: &mut tokio::io::ReadBuf<'_>,
+        ) -> std::task::Poll<std::io::Result<()>> {
+            if self.0 {
+                return std::task::Poll::Ready(Err(std::io::Error::other("boom")));
+            }
+            self.0 = true;
+            buf.put_slice(b"hello\n");
+            std::task::Poll::Ready(Ok(()))
+        }
+    }
+
+    #[tokio::test]
+    async fn a_read_error_ends_the_stream_like_eof() {
+        use tokio::io::{AsyncBufReadExt, BufReader};
+        let mut lines = BufReader::new(FailingRead(false)).lines();
+        assert_eq!(
+            next_line(&mut lines, "stdout").await.as_deref(),
+            Some("hello")
+        );
+        assert_eq!(next_line(&mut lines, "stdout").await, None);
+    }
 }