hive-c0re: keep agent and manager listeners alive across accept errors

Both accept loops returned on the first accept() error. The AgentSocket
stayed in Coordinator.agents, so mcp_sockets::sync_on_start skipped the
agent, and only a Start node or a hive-c0re restart bound it again. The
unit sets no LimitNOFILE, so a single EMFILE at the 1024-fd soft limit
cut the agent off from send/recv/ack with one warn line as the only
trace.

Both loops now share accept_until_fatal. Connection errors
(ECONNABORTED/ECONNRESET/ECONNREFUSED) retry at once, any other error
retries after 1 s, following axum's serve loop that the dashboard
already runs. Errors that leave the listening fd unusable (EBADF,
EFAULT, EINVAL, ENOTSOCK, EOPNOTSUPP) log at error and exit the
process: nothing re-binds a listener while hive-c0re runs, the manager
listener has no re-bind path at all, and the unit's Restart=on-failure
restart runs start_manager and sync_on_start, which re-bind every
socket.

mcp_sockets.rs now states that invariant instead of asserting that a
listener can only disappear on restart.

LimitNOFILE is left unset: per-agent fd use is a listener plus a
Recv long-poll plus short-lived requests, and a higher limit would raise
the memory ceiling of the per-line bound (bound x open connections).

Closes #4721

hive-c0re/src/socket_server/mod.rs

@@ -15,7 +15,7 @@ use anyhow::{Context, Result};
 use hive_core_agent_sock::{Request, Response};
 use hive_sh4re::inbox::Message;
 use hive_sh4re::manager::MANAGER_AGENT;
-use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader};
+use tokio::io::{AsyncBufRead, AsyncBufReadExt, AsyncReadExt, AsyncWriteExt, BufReader};
 use tokio::net::{UnixListener, UnixStream};
 use tokio::task::JoinHandle;
 
@@ -63,23 +63,21 @@ pub fn start(agent: &str, socket_path: &Path, coord: Arc<Coordinator>) -> Result
 
     let path = socket_path.to_path_buf();
     let handle = tokio::spawn(async move {
-        loop {
+        let err = accept_until_fatal(
-            match listener.accept().await {
+            &agent,
-                Ok((stream, _)) => {
+            || async { listener.accept().await.map(|(stream, _)| stream) },
-                    let agent = agent.clone();
+            |stream| {
-                    let coord = coord.clone();
+                let agent = agent.clone();
-                    tokio::spawn(async move {
+                let coord = coord.clone();
-                        if let Err(e) = serve(stream, agent, coord).await {
+                tokio::spawn(async move {
-                            tracing::warn!(error = ?e, "agent connection failed");
+                    if let Err(e) = serve(stream, agent, coord).await {
-                        }
+                        tracing::warn!(error = ?e, "agent connection failed");
-                    });
+                    }
-                }
+                });
-                Err(e) => {
+            },
-                    tracing::warn!(error = ?e, "agent listener accept failed; exiting");
+        )
-                    return;
+        .await;
-                }
+        exit_on_dead_listener(&agent, &err);
-            }
-        }
     });
     Ok(AgentSocket { path, handle })
 }
@@ -107,47 +105,153 @@ pub fn start_manager(coord: Arc<Coordinator>) -> Result<()> {
     tracing::info!(socket = %socket.display(), "manager socket listening");
 
     tokio::spawn(async move {
-        loop {
+        let err = accept_until_fatal(
-            match listener.accept().await {
+            MANAGER_AGENT,
-                Ok((stream, _)) => {
+            || async { listener.accept().await.map(|(stream, _)| stream) },
-                    let coord = coord.clone();
+            |stream| {
-                    tokio::spawn(async move {
+                let coord = coord.clone();
-                        // Pure transport: serve as `ruth`, no privilege grant.
+                tokio::spawn(async move {
-                        if let Err(e) = serve(stream, MANAGER_AGENT.to_owned(), coord).await {
+                    // Pure transport: serve as `ruth`, no privilege grant.
-                            tracing::warn!(error = ?e, "manager connection failed");
+                    if let Err(e) = serve(stream, MANAGER_AGENT.to_owned(), coord).await {
-                        }
+                        tracing::warn!(error = ?e, "manager connection failed");
-                    });
+                    }
-                }
+                });
-                Err(e) => {
+            },
-                    tracing::warn!(error = ?e, "manager listener accept failed");
+        )
-                    return;
+        .await;
-                }
+        exit_on_dead_listener(MANAGER_AGENT, &err);
-            }
-        }
     });
     Ok(())
 }
 
+/// Pause after an accept error that is not about a single connection. EMFILE
+/// and ENFILE leave the listener readable, so retrying at once would spin until
+/// some fd is closed. Same policy as axum's `serve`, which the dashboard runs.
+const ACCEPT_RETRY_DELAY: std::time::Duration = std::time::Duration::from_secs(1);
+
+/// Accept errors that mean the listening fd itself is unusable: retrying it
+/// returns the same error forever.
+fn accept_error_is_fatal(e: &std::io::Error) -> bool {
+    matches!(
+        e.raw_os_error(),
+        Some(libc::EBADF | libc::EFAULT | libc::EINVAL | libc::ENOTSOCK | libc::EOPNOTSUPP)
+    )
+}
+
+/// Hand each accepted connection to `on_conn`, retrying every accept error
+/// except a fatal one, which is returned. `agent` labels the log lines.
+async fn accept_until_fatal<A, F>(
+    agent: &str,
+    mut accept: A,
+    mut on_conn: impl FnMut(UnixStream),
+) -> std::io::Error
+where
+    A: FnMut() -> F,
+    F: std::future::Future<Output = std::io::Result<UnixStream>>,
+{
+    loop {
+        match accept().await {
+            Ok(stream) => on_conn(stream),
+            Err(e) if accept_error_is_fatal(&e) => return e,
+            Err(e) => {
+                tracing::warn!(%agent, error = ?e, "socket accept failed; retrying");
+                if !matches!(
+                    e.kind(),
+                    std::io::ErrorKind::ConnectionAborted
+                        | std::io::ErrorKind::ConnectionReset
+                        | std::io::ErrorKind::ConnectionRefused
+                ) {
+                    tokio::time::sleep(ACCEPT_RETRY_DELAY).await;
+                }
+            }
+        }
+    }
+}
+
+/// Exit on a listener that can no longer accept. A running hive-c0re re-binds
+/// an agent socket only on that agent's next start, and the manager socket
+/// never; on restart (`Restart = "on-failure"` on the unit) `start_manager` and
+/// `mcp_sockets::sync_on_start` re-bind every one.
+fn exit_on_dead_listener(agent: &str, e: &std::io::Error) -> ! {
+    tracing::error!(
+        %agent,
+        error = ?e,
+        "socket listener cannot accept; exiting so hive-c0re restarts and re-binds all sockets"
+    );
+    std::process::exit(1)
+}
+
+/// Longest request line accepted on the agent and manager sockets, newline
+/// included. The largest request a client sends is an `OperatorMsg` from the
+/// agent web UI's `/send` (`hive-agent/src/web_ui/actions.rs:21`), whose body
+/// axum's default limit caps at 2 MiB; JSON escaping can roughly double that to
+/// 4 MiB, and this is 4× that.
+const REQUEST_LINE_MAX_BYTES: usize = 16 * 1024 * 1024;
+
+enum RequestLine {
+    Eof,
+    Complete,
+    TooLong,
+}
+
+/// Read one request line into `buf`, consuming at most
+/// [`REQUEST_LINE_MAX_BYTES`] from `reader`.
+async fn read_request_line<R: AsyncBufRead + Unpin>(
+    reader: &mut R,
+    buf: &mut Vec<u8>,
+) -> std::io::Result<RequestLine> {
+    buf.clear();
+    let n = (&mut *reader)
+        .take(REQUEST_LINE_MAX_BYTES as u64)
+        .read_until(b'\n', buf)
+        .await?;
+    Ok(if n == 0 {
+        RequestLine::Eof
+    } else if n == REQUEST_LINE_MAX_BYTES && !buf.ends_with(b"\n") {
+        RequestLine::TooLong
+    } else {
+        RequestLine::Complete
+    })
+}
+
+/// Answer each request line on `stream` until EOF. An over-long line gets a
+/// `Response::Err` and then the connection is closed, because the rest of the
+/// line is still unread.
 async fn serve(stream: UnixStream, agent: String, coord: Arc<Coordinator>) -> Result<()> {
     let (read, mut write) = stream.into_split();
     let mut reader = BufReader::new(read);
-    let mut line = String::new();
+    let mut line = Vec::new();
     loop {
-        line.clear();
+        let (resp, too_long) = match read_request_line(&mut reader, &mut line).await? {
-        let n = reader.read_line(&mut line).await?;
+            RequestLine::Eof => return Ok(()),
-        if n == 0 {
+            RequestLine::TooLong => (
-            return Ok(());
+                Response::Err {
-        }
+                    message: format!(
-        let resp = match serde_json::from_str::<Request>(line.trim()) {
+                        "request too long: no newline within {REQUEST_LINE_MAX_BYTES} bytes; \
-            Ok(req) => dispatch(&req, &agent, &coord).await,
+                         connection closed"
-            Err(e) => Response::Err {
+                    ),
-                message: format!("parse error: {e}"),
+                },
+                true,
+            ),
+            RequestLine::Complete => match serde_json::from_slice::<Request>(&line) {
+                Ok(req) => (dispatch(&req, &agent, &coord).await, false),
+                Err(e) => (
+                    Response::Err {
+                        message: format!("parse error: {e}"),
+                    },
+                    false,
+                ),
             },
         };
         let mut payload = serde_json::to_string(&resp)?;
         payload.push('\n');
         write.write_all(payload.as_bytes()).await?;
         write.flush().await?;
+        if too_long {
+            anyhow::bail!(
+                "`{agent}` sent a request line over {REQUEST_LINE_MAX_BYTES} bytes; closed"
+            );
+        }
     }
 }
 
@@ -791,4 +895,74 @@ mod tests {
         let err = check_can_cancel_approval("nobody-with-no-groups").unwrap_err();
         assert!(err.contains("approvals` tool group"), "{err}");
     }
+
+    #[tokio::test]
+    async fn over_long_line_is_rejected_after_reading_only_the_bound() {
+        let total = 2 * REQUEST_LINE_MAX_BYTES as u64;
+        let mut reader = BufReader::new(tokio::io::repeat(b'x').take(total));
+        let mut buf = Vec::new();
+        let outcome = read_request_line(&mut reader, &mut buf)
+            .await
+            .expect("read");
+        // At most one `BufReader` fill past the bound left the source.
+        let unread = reader.into_inner().limit();
+        let consumed = total - unread;
+        assert!(
+            consumed <= (REQUEST_LINE_MAX_BYTES + 8 * 1024) as u64,
+            "consumed {consumed} of {total} bytes"
+        );
+        assert_eq!(buf.len(), REQUEST_LINE_MAX_BYTES);
+        assert!(matches!(outcome, RequestLine::TooLong));
+    }
+
+    #[tokio::test]
+    async fn serve_answers_an_over_long_line_with_an_error_and_closes() {
+        let (_dir, coord) = schedules::tests::coordinator();
+        let (client, server) = UnixStream::pair().expect("socketpair");
+        let handler = tokio::spawn(serve(server, "iris".to_owned(), coord));
+        let (read, mut write) = client.into_split();
+        let writer = tokio::spawn(async move {
+            let chunk = vec![b'x'; 64 * 1024];
+            for _ in 0..2 * REQUEST_LINE_MAX_BYTES / chunk.len() {
+                if write.write_all(&chunk).await.is_err() {
+                    return;
+                }
+            }
+            let _ = write.shutdown().await;
+        });
+        let mut reply = String::new();
+        BufReader::new(read)
+            .read_line(&mut reply)
+            .await
+            .expect("reply");
+        let resp: Response = serde_json::from_str(&reply).expect("response json");
+        assert!(
+            matches!(&resp, Response::Err { message } if message.starts_with("request too long")),
+            "{reply}"
+        );
+        assert!(
+            handler.await.expect("join").is_err(),
+            "connection not closed"
+        );
+        writer.await.expect("join");
+    }
+
+    #[tokio::test]
+    async fn accept_loop_survives_a_transient_error() {
+        let (client, _peer) = UnixStream::pair().expect("socketpair");
+        let mut script = std::collections::VecDeque::from([
+            Err(std::io::Error::from_raw_os_error(libc::EMFILE)),
+            Ok(client),
+            Err(std::io::Error::from_raw_os_error(libc::EBADF)),
+        ]);
+        let mut served = 0;
+        let err = accept_until_fatal(
+            "iris",
+            || std::future::ready(script.pop_front().expect("accept after a fatal error")),
+            |_stream| served += 1,
+        )
+        .await;
+        assert_eq!(served, 1, "connection after EMFILE not served");
+        assert_eq!(err.raw_os_error(), Some(libc::EBADF));
+    }
 }

hive-c0re/src/socket_server/schedules.rs

@@ -302,7 +302,7 @@ fn schedule_to_wire(s: crate::scheduled_prompts::Schedule) -> hive_sh4re::schedu
 }
 
 #[cfg(test)]
-mod tests {
+pub(super) mod tests {
     use super::*;
 
     fn target(name: &str) -> hive_sh4re::schedule::WireScheduleTarget {
@@ -377,10 +377,10 @@ mod tests {
     /// anybody's ancestor or descendant.
     const STRANGER: &str = "requester-with-no-relations";
 
-    /// A real `Coordinator` over a throwaway sqlite dir. The schedule
+    /// A real `Coordinator` over a throwaway sqlite dir. The socket-server
     /// handlers take `&Arc<Coordinator>`, so there is no lighter way in;
     /// `open` touches nothing outside the db path it is handed.
-    fn coordinator() -> (tempfile::TempDir, Arc<Coordinator>) {
+    pub(in crate::socket_server) fn coordinator() -> (tempfile::TempDir, Arc<Coordinator>) {
         let dir = tempfile::tempdir().expect("tempdir");
         let coord = Coordinator::open(
             &dir.path().join("broker.sqlite"),

hive-c0re/src/workers/mcp_sockets.rs

@@ -10,9 +10,11 @@
 //! - `run_reconcile` calls `register_agent` immediately after `start_with_fallback`.
 //! - `kill`/`destroy` paths call `unregister_agent`.
 //!
-//! No recurring poll is needed because c0re owns the listener lifecycle —
+//! No recurring poll is needed because c0re owns the listener lifecycle. An
-//! a listener can only disappear when c0re itself restarts, which is exactly
+//! accept loop retries every error except one that leaves its fd unusable,
-//! the case `sync_on_start` covers.
+//! and on that one hive-c0re exits (`socket_server::exit_on_dead_listener`),
+//! so a listener only disappears across a restart — exactly the case
+//! `sync_on_start` covers.
 
 use std::sync::Arc;