From 7a826f9ee2f7d7362638f082c0121a9ea0beb19c Mon Sep 17 00:00:00 2001 From: atlas Date: Sun, 26 Jul 2026 22:20:01 +0200 Subject: [PATCH] refactor(sock): one socket client, retry as a policy value MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Six places in the tree hand-rolled the same connect / write one JSON line / read one JSON line back. Two of them — the harness serve loop's client and the MCP server's — were byte-identical apart from a six-line wrapper, ~145 lines of literal copy-paste. The other four each reimplemented a subset, and the subsets had drifted: some named the socket path in their errors and some did not, one classified transient against fatal failures and the rest retried nothing at all, two drained the response and two decoded it. That duplication was defended when the daemons were split out, on the grounds that a daemon's socket etiquette should stay visible in the crate that depends on it. The etiquette genuinely does differ. The code does not, and five copies is where "each daemon documents its own etiquette" stops paying for itself. `hive-sock-client` now owns the transport once, generic over the request and response types so it is protocol-agnostic: the host-served control socket and the harness's in-agent socket both use it with their own wire-type crates. The two real differences become values instead of forks. Retry is `Retry::RideOutRestart` (2/4/8/16/30s, sized to ride out a service restart) for callers with no natural retry of their own, or `Retry::None` for callers already inside a poll loop where the poll interval is the retry — and the reason each caller picked one is a comment at the call site rather than a reimplementation. The response is either decoded (`request`) or half-closed and drained (`notify`, where the drain exists so the server's write-back doesn't land on a closed socket). Whether a failure propagates or is logged and swallowed stays at the call site, because that is the caller's choice and not a property of the transport. Errors always name the socket path now, everywhere. That detail is load-bearing: a permission problem on a socket that reads as "is the daemon running?" sends the operator to fix the wrong thing. The transient-against-fatal enum is gone rather than moved. Serialising happens before the retry loop and deserialising after it, so only connect, I/O and short-read failures can reach the loop at all — a deterministic failure is now unretryable by construction instead of by classification. It is deliberately a new crate and not part of `hive-agent-sock`. The `*-sock` crates are pure wire types by convention — `hive-agent-sock` depends on serde and nothing else — and the two largest copies talk to the host socket, whose types live in a different crate entirely. A transport in either wire-type crate would drag tokio into it and point the wrong way besides. No wire-format change: same JSON line in, same line out. --- CLAUDE.md | 9 + Cargo.lock | 16 ++ Cargo.toml | 2 + hive-agent-mcp/Cargo.toml | 1 + hive-agent-mcp/src/client.rs | 152 ------------- hive-agent-mcp/src/main.rs | 1 - hive-agent-mcp/src/mcp/mod.rs | 12 +- hive-agent/Cargo.toml | 1 + hive-agent/src/client.rs | 167 --------------- hive-agent/src/main.rs | 41 ++-- hive-agent/src/todo_server.rs | 14 +- hive-agent/src/web_ui/mod.rs | 6 +- hive-bash-mcp/Cargo.toml | 1 + hive-bash-mcp/src/runner.rs | 32 +-- hive-forge-notify/Cargo.toml | 1 + hive-forge-notify/src/main.rs | 10 +- hive-forge-notify/src/notify.rs | 19 +- hive-forge-notify/src/todo_client.rs | 86 -------- hive-matrix-mcp/Cargo.toml | 1 + hive-matrix-mcp/src/wake.rs | 46 +--- hive-sock-client/Cargo.toml | 21 ++ hive-sock-client/README.md | 52 +++++ hive-sock-client/src/lib.rs | 308 +++++++++++++++++++++++++++ 23 files changed, 498 insertions(+), 501 deletions(-) delete mode 100644 hive-agent-mcp/src/client.rs delete mode 100644 hive-agent/src/client.rs delete mode 100644 hive-forge-notify/src/todo_client.rs create mode 100644 hive-sock-client/Cargo.toml create mode 100644 hive-sock-client/README.md create mode 100644 hive-sock-client/src/lib.rs diff --git a/CLAUDE.md b/CLAUDE.md index 12d57fb8..e5d4cc41 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -87,6 +87,15 @@ hand-maintained per-file tree drifts out of sync with the code. - **`hive-priv-sock/`** — wire types for the `hive-priv` privileged-helper socket (`/run/hive/priv.sock`), shared by `hive-priv` (server) and `hive-c0re` (client). Also split out of `hive-sh4re`. +- **`hive-sock-client/`** — the shared JSON-line-over-unix-socket client + every daemon uses to talk to a hyperhive socket. Generic over the + request/response types, so the host-served control socket and the + harness's in-agent socket both use it with their own wire-type crates. + Retry is a policy value (`Retry::None` for callers already inside a poll + loop, `Retry::RideOutRestart` for callers with no natural retry), and + the response is either decoded (`request`) or drained (`notify`). + Deliberately separate from the `*-sock` crates — those stay + dependency-free wire types. - **`hive-metric/`** — small CLI to push a single labeled metric to the OTEL collector via the OpenTelemetry Rust SDK / OTLP HTTP exporter. diff --git a/Cargo.lock b/Cargo.lock index 5eddf387..24803d40 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1526,6 +1526,7 @@ dependencies = [ "hive-claude", "hive-core-agent-sock", "hive-sh4re", + "hive-sock-client", "http-body-util", "hyper", "hyper-util", @@ -1554,6 +1555,7 @@ dependencies = [ "hive-agent-sock", "hive-core-agent-sock", "hive-sh4re", + "hive-sock-client", "hive-types", "reqwest 0.13.1", "rmcp", @@ -1581,6 +1583,7 @@ dependencies = [ "clap", "hive-agent-sock", "hive-sh4re", + "hive-sock-client", "hive-types", "libc", "rmcp", @@ -1675,6 +1678,7 @@ dependencies = [ "anyhow", "forgejo-api", "hive-agent-sock", + "hive-sock-client", "reqwest 0.13.1", "serde", "serde_json", @@ -1712,6 +1716,7 @@ dependencies = [ "axum", "clap", "futures-util", + "hive-sock-client", "matrix-sdk", "mime", "mime_guess", @@ -1780,6 +1785,17 @@ dependencies = [ "serde_json", ] +[[package]] +name = "hive-sock-client" +version = "0.1.0" +dependencies = [ + "anyhow", + "serde", + "serde_json", + "tokio", + "tracing", +] + [[package]] name = "hive-types" version = "0.1.0" diff --git a/Cargo.toml b/Cargo.toml index a673bd9a..9cd6ab8c 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -18,6 +18,7 @@ members = [ "hive-priv", "hive-priv-sock", "hive-sh4re", + "hive-sock-client", "hive-types", "hivectl", ] @@ -58,6 +59,7 @@ hive-core-agent-sock = { path = "hive-core-agent-sock" } hive-claude = { path = "hive-claude" } hive-host-sock = { path = "hive-host-sock" } hive-priv-sock = { path = "hive-priv-sock" } +hive-sock-client = { path = "hive-sock-client" } hive-types = { path = "hive-types" } thiserror = "2" tower-http = { version = "0.7", features = ["fs"] } diff --git a/hive-agent-mcp/Cargo.toml b/hive-agent-mcp/Cargo.toml index 9cf72d6d..886d6adb 100644 --- a/hive-agent-mcp/Cargo.toml +++ b/hive-agent-mcp/Cargo.toml @@ -18,6 +18,7 @@ clap.workspace = true hive-agent-sock.workspace = true hive-core-agent-sock.workspace = true hive-sh4re.workspace = true +hive-sock-client.workspace = true hive-types.workspace = true reqwest.workspace = true rmcp.workspace = true diff --git a/hive-agent-mcp/src/client.rs b/hive-agent-mcp/src/client.rs deleted file mode 100644 index c175cd11..00000000 --- a/hive-agent-mcp/src/client.rs +++ /dev/null @@ -1,152 +0,0 @@ -use std::path::Path; -use std::time::Duration; - -use anyhow::{Result, anyhow}; -use serde::Serialize; -use serde::de::DeserializeOwned; -use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader}; -use tokio::net::UnixStream; - -/// Backoff schedule between attempts. Five entries → up to 5 retries on -/// top of the initial attempt; total wall-clock cap = 2+4+8+16+30 = 60s. -/// Sized to ride out a hive-c0re restart (systemd usually has the unix -/// socket back inside ~5s) without the agent-side claude session having -/// to handle the transient itself — burning tokens on a tool-error retry -/// loop is more expensive than 60s of in-harness sleep. -const RETRY_BACKOFFS_MS: &[u64] = &[2_000, 4_000, 8_000, 16_000, 30_000]; - -/// Send `req` over the unix socket and decode the single-line JSON -/// response, retrying transient connect/IO failures on the backoff -/// schedule above and reporting how many retries it took past the initial -/// attempt (0 = succeeded first try). MCP tool handlers -/// use this so they can append a one-line hint to the tool result when -/// retries happened — that way claude knows the prior socket flake -/// wasn't a content error and shouldn't trigger an LLM-level retry of -/// its own. -/// -/// # Errors -/// -/// Returns an error if all retries are exhausted, or on a fatal protocol -/// error (serialization / deserialization failure). -/// -/// # Panics -/// -/// Panics if `RETRY_BACKOFFS_MS.len()` does not fit in a `u32`, which -/// cannot happen with the current compile-time constant. -pub async fn request_retried(socket: &Path, req: &Req) -> Result<(Resp, u32)> -where - Req: Serialize + ?Sized, - Resp: DeserializeOwned, -{ - let mut last_err: Option = None; - let max_retries = u32::try_from(RETRY_BACKOFFS_MS.len()).unwrap(); - for attempt in 0..=max_retries { - match try_once::(socket, req).await { - Ok(resp) => return Ok((resp, attempt)), - Err(RequestError::Fatal(e)) => return Err(e), - Err(RequestError::Transient(e)) => { - if attempt < max_retries { - let sleep_ms = RETRY_BACKOFFS_MS[attempt as usize]; - tracing::warn!( - attempt = attempt + 1, - sleep_ms, - error = %e, - "hive socket attempt failed; retrying" - ); - last_err = Some(e); - tokio::time::sleep(Duration::from_millis(sleep_ms)).await; - } else { - last_err = Some(e); - } - } - } - } - // Reaching here means the final attempt returned `Transient`, which always - // sets `last_err` — so this is infallible. - Err(last_err.expect("a transient failure on the final attempt set last_err")) -} - -/// Transient = connect / IO error worth a retry (server restart, broken -/// pipe). Fatal = serialization / deserialization / protocol error -/// where retrying would just repeat the same failure. -enum RequestError { - Transient(anyhow::Error), - Fatal(anyhow::Error), -} - -async fn try_once(socket: &Path, req: &Req) -> Result -where - Req: Serialize + ?Sized, - Resp: DeserializeOwned, -{ - let stream = match UnixStream::connect(socket).await { - Ok(stream) => stream, - Err(e) => { - // A refused or missing socket usually means hive-c0re is - // mid-restart (operator redeploy / rebuild) — the socket is - // recreated on its boot and `request_retried` rides it out. When - // the error *does* surface (retries exhausted, or a non-retried - // caller) add that context so claude reads it as a likely - // transient rather than a hard failure worth escalating. - let restarting = matches!( - e.kind(), - std::io::ErrorKind::ConnectionRefused | std::io::ErrorKind::NotFound - ); - let mut err = anyhow::Error::new(e).context(format!("connect to {}", socket.display())); - if restarting { - err = err.context( - "hive-c0re may be restarting (e.g. an operator redeploy); \ - the harness already retried ~60s before surfacing this", - ); - } - return Err(RequestError::Transient(err)); - } - }; - let (read, mut write) = stream.into_split(); - - let mut payload = serde_json::to_string(req).map_err(|e| RequestError::Fatal(e.into()))?; - payload.push('\n'); - write - .write_all(payload.as_bytes()) - .await - .map_err(|e| RequestError::Transient(e.into()))?; - write - .flush() - .await - .map_err(|e| RequestError::Transient(e.into()))?; - - let mut reader = BufReader::new(read); - let mut line = String::new(); - let read_bytes = reader - .read_line(&mut line) - .await - .map_err(|e| RequestError::Transient(e.into()))?; - if read_bytes == 0 || line.is_empty() { - return Err(RequestError::Transient(anyhow!( - "server closed connection without responding" - ))); - } - serde_json::from_str(line.trim()).map_err(|e| RequestError::Fatal(e.into())) -} - -#[cfg(test)] -mod tests { - use super::{RequestError, try_once}; - - /// A connect to a non-existent socket path (ENOENT → `NotFound`) is - /// classified transient AND annotated with the "hive-c0re is restarting" - /// hint, so a surfaced tool error reads as the expected transient. - #[tokio::test] - async fn missing_socket_connect_is_transient_with_restart_hint() { - let bogus = std::path::Path::new("/nonexistent/hive/mcp.sock"); - match try_once::<(), serde_json::Value>(bogus, &()).await { - Err(RequestError::Transient(e)) => { - let msg = format!("{e:#}"); - assert!(msg.contains("restarting"), "missing restart hint: {msg}"); - assert!(msg.contains("connect to"), "missing connect context: {msg}"); - } - Err(RequestError::Fatal(e)) => panic!("expected transient, got fatal: {e:#}"), - Ok(_) => panic!("expected connect failure to a non-existent socket"), - } - } -} diff --git a/hive-agent-mcp/src/main.rs b/hive-agent-mcp/src/main.rs index 4ff69ef8..2eae80a9 100644 --- a/hive-agent-mcp/src/main.rs +++ b/hive-agent-mcp/src/main.rs @@ -16,7 +16,6 @@ use std::path::PathBuf; use anyhow::Result; use clap::Parser; -mod client; mod mcp; mod paths; mod send_allow; diff --git a/hive-agent-mcp/src/mcp/mod.rs b/hive-agent-mcp/src/mcp/mod.rs index b9708b06..0dcad96a 100644 --- a/hive-agent-mcp/src/mcp/mod.rs +++ b/hive-agent-mcp/src/mcp/mod.rs @@ -20,8 +20,6 @@ use std::path::PathBuf; use anyhow::Result; use rmcp::{ServerHandler, handler::server::wrapper::Parameters, tool, tool_handler, tool_router}; -use crate::client; - mod args; mod render; @@ -124,7 +122,15 @@ impl AgentServer { &self, req: hive_core_agent_sock::Request, ) -> (Result, u32) { - match client::request_retried::<_, hive_core_agent_sock::Response>(&self.socket, &req).await + // Ride out a hive-c0re restart rather than surface it: the caller + // here is claude, and a tool error costs a whole retry turn's + // tokens where 60s of in-daemon patience costs nothing. + match hive_sock_client::request_retried::<_, hive_core_agent_sock::Response>( + &self.socket, + &req, + hive_sock_client::Retry::RideOutRestart, + ) + .await { Ok((r, n)) => (Ok(r), n), Err(e) => (Err(e), 0), diff --git a/hive-agent/Cargo.toml b/hive-agent/Cargo.toml index 4f8b2f79..3dd41c25 100644 --- a/hive-agent/Cargo.toml +++ b/hive-agent/Cargo.toml @@ -20,6 +20,7 @@ hive-claude.workspace = true hive-agent-sock.workspace = true hive-core-agent-sock.workspace = true hive-sh4re.workspace = true +hive-sock-client.workspace = true libc.workspace = true rmcp.workspace = true rusqlite.workspace = true diff --git a/hive-agent/src/client.rs b/hive-agent/src/client.rs deleted file mode 100644 index 0cfd36f9..00000000 --- a/hive-agent/src/client.rs +++ /dev/null @@ -1,167 +0,0 @@ -use std::path::Path; -use std::time::Duration; - -use anyhow::{Result, anyhow}; -use serde::Serialize; -use serde::de::DeserializeOwned; -use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader}; -use tokio::net::UnixStream; - -/// Backoff schedule between attempts. Five entries → up to 5 retries on -/// top of the initial attempt; total wall-clock cap = 2+4+8+16+30 = 60s. -/// Sized to ride out a hive-c0re restart (systemd usually has the unix -/// socket back inside ~5s) without the agent-side claude session having -/// to handle the transient itself — burning tokens on a tool-error retry -/// loop is more expensive than 60s of in-harness sleep. -const RETRY_BACKOFFS_MS: &[u64] = &[2_000, 4_000, 8_000, 16_000, 30_000]; - -/// Transparent retry wrapper around [`request_retried`] that throws away -/// the retry count. Use this from non-tool callers (the harness serve -/// loop, web UI, CLI subcommands) where we just want the socket-restart -/// resilience without surfacing the bookkeeping. -/// -/// # Errors -/// -/// Returns an error if the socket is unreachable after all retries, or if -/// serialization / deserialization of the request or response fails. -pub async fn request(socket: &Path, req: &Req) -> Result -where - Req: Serialize + ?Sized, - Resp: DeserializeOwned, -{ - request_retried(socket, req).await.map(|(resp, _)| resp) -} - -/// Same wire shape as [`request`], but reports how many retries it took -/// past the initial attempt (0 = succeeded first try). MCP tool handlers -/// use this so they can append a one-line hint to the tool result when -/// retries happened — that way claude knows the prior socket flake -/// wasn't a content error and shouldn't trigger an LLM-level retry of -/// its own. -/// -/// # Errors -/// -/// Returns an error if all retries are exhausted, or on a fatal protocol -/// error (serialization / deserialization failure). -/// -/// # Panics -/// -/// Panics if `RETRY_BACKOFFS_MS.len()` does not fit in a `u32`, which -/// cannot happen with the current compile-time constant. -pub async fn request_retried(socket: &Path, req: &Req) -> Result<(Resp, u32)> -where - Req: Serialize + ?Sized, - Resp: DeserializeOwned, -{ - let mut last_err: Option = None; - let max_retries = u32::try_from(RETRY_BACKOFFS_MS.len()).unwrap(); - for attempt in 0..=max_retries { - match try_once::(socket, req).await { - Ok(resp) => return Ok((resp, attempt)), - Err(RequestError::Fatal(e)) => return Err(e), - Err(RequestError::Transient(e)) => { - if attempt < max_retries { - let sleep_ms = RETRY_BACKOFFS_MS[attempt as usize]; - tracing::warn!( - attempt = attempt + 1, - sleep_ms, - error = %e, - "hive socket attempt failed; retrying" - ); - last_err = Some(e); - tokio::time::sleep(Duration::from_millis(sleep_ms)).await; - } else { - last_err = Some(e); - } - } - } - } - // Reaching here means the final attempt returned `Transient`, which always - // sets `last_err` — so this is infallible. - Err(last_err.expect("a transient failure on the final attempt set last_err")) -} - -/// Transient = connect / IO error worth a retry (server restart, broken -/// pipe). Fatal = serialization / deserialization / protocol error -/// where retrying would just repeat the same failure. -enum RequestError { - Transient(anyhow::Error), - Fatal(anyhow::Error), -} - -async fn try_once(socket: &Path, req: &Req) -> Result -where - Req: Serialize + ?Sized, - Resp: DeserializeOwned, -{ - let stream = match UnixStream::connect(socket).await { - Ok(stream) => stream, - Err(e) => { - // A refused or missing socket usually means hive-c0re is - // mid-restart (operator redeploy / rebuild) — the socket is - // recreated on its boot and `request_retried` rides it out. When - // the error *does* surface (retries exhausted, or a non-retried - // caller) add that context so claude reads it as a likely - // transient rather than a hard failure worth escalating. - let restarting = matches!( - e.kind(), - std::io::ErrorKind::ConnectionRefused | std::io::ErrorKind::NotFound - ); - let mut err = anyhow::Error::new(e).context(format!("connect to {}", socket.display())); - if restarting { - err = err.context( - "hive-c0re may be restarting (e.g. an operator redeploy); \ - the harness already retried ~60s before surfacing this", - ); - } - return Err(RequestError::Transient(err)); - } - }; - let (read, mut write) = stream.into_split(); - - let mut payload = serde_json::to_string(req).map_err(|e| RequestError::Fatal(e.into()))?; - payload.push('\n'); - write - .write_all(payload.as_bytes()) - .await - .map_err(|e| RequestError::Transient(e.into()))?; - write - .flush() - .await - .map_err(|e| RequestError::Transient(e.into()))?; - - let mut reader = BufReader::new(read); - let mut line = String::new(); - let read_bytes = reader - .read_line(&mut line) - .await - .map_err(|e| RequestError::Transient(e.into()))?; - if read_bytes == 0 || line.is_empty() { - return Err(RequestError::Transient(anyhow!( - "server closed connection without responding" - ))); - } - serde_json::from_str(line.trim()).map_err(|e| RequestError::Fatal(e.into())) -} - -#[cfg(test)] -mod tests { - use super::{RequestError, try_once}; - - /// A connect to a non-existent socket path (ENOENT → `NotFound`) is - /// classified transient AND annotated with the "hive-c0re is restarting" - /// hint, so a surfaced tool error reads as the expected transient. - #[tokio::test] - async fn missing_socket_connect_is_transient_with_restart_hint() { - let bogus = std::path::Path::new("/nonexistent/hive/mcp.sock"); - match try_once::<(), serde_json::Value>(bogus, &()).await { - Err(RequestError::Transient(e)) => { - let msg = format!("{e:#}"); - assert!(msg.contains("restarting"), "missing restart hint: {msg}"); - assert!(msg.contains("connect to"), "missing connect context: {msg}"); - } - Err(RequestError::Fatal(e)) => panic!("expected transient, got fatal: {e:#}"), - Ok(_) => panic!("expected connect failure to a non-existent socket"), - } - } -} diff --git a/hive-agent/src/main.rs b/hive-agent/src/main.rs index 3d1af3b7..f3a0f53c 100644 --- a/hive-agent/src/main.rs +++ b/hive-agent/src/main.rs @@ -9,7 +9,6 @@ //! Single bin crate: the module tree below (formerly this crate's `lib.rs`, //! before lib + bin were collapsed into one) plus the serve loop. -mod client; mod db_migrate; mod disk_watch; mod events; @@ -37,6 +36,12 @@ mod web_ui; /// Default socket path inside the container — bind-mounted by `hive-c0re`. const DEFAULT_SOCKET: &str = "/run/hive/mcp.sock"; +/// Retry policy for every request to the host-served control socket. +/// Nothing on this side of the socket has a natural retry — the serve loop +/// and the web UI both hand a failure straight to a human or to claude — +/// so a hive-c0re restart is worth waiting out rather than surfacing. +const CONTROL_SOCKET_RETRY: Retry = Retry::RideOutRestart; + /// Default web UI port — used when `HIVE_PORT` env is unset. const DEFAULT_WEB_PORT: u16 = 8042; @@ -57,6 +62,7 @@ use anyhow::Result; use clap::Parser; use hive_core_agent_sock::{Request, Response}; use hive_sh4re::{HelperEvent, SYSTEM_SENDER}; +use hive_sock_client::Retry; #[derive(Parser)] #[command(name = "hive-agent", about = "hyperhive harness serve loop")] @@ -358,7 +364,7 @@ struct AgentSurface; /// the `Surface` methods that don't need the reply (`ack_turn`, /// `requeue_inflight`, `graceful_stop_complete`). async fn fire_and_forget(socket: &Path, req: Request, label: &str) { - match client::request::<_, Response>(socket, &req).await { + match hive_sock_client::request::<_, Response>(socket, &req, CONTROL_SOCKET_RETRY).await { Ok(Response::Ok) => {} Ok(Response::Err { message }) => { tracing::warn!(%message, "{label} rejected by broker"); @@ -387,20 +393,29 @@ impl Surface for AgentSurface { } async fn inbox_unread(socket: &Path) -> u64 { - match client::request::<_, Response>(socket, &Request::Status).await { + match hive_sock_client::request::<_, Response>( + socket, + &Request::Status, + CONTROL_SOCKET_RETRY, + ) + .await + { Ok(Response::Status { unread }) => unread, _ => 0, } } async fn post_turn_counts(socket: &Path) -> (Option, Option) { - let threads = - match client::request::<_, Response>(socket, &Request::GetLooseEnds { agent: None }) - .await - { - Ok(Response::LooseEnds { loose_ends }) => u64::try_from(loose_ends.len()).ok(), - _ => None, - }; + let threads = match hive_sock_client::request::<_, Response>( + socket, + &Request::GetLooseEnds { agent: None }, + CONTROL_SOCKET_RETRY, + ) + .await + { + Ok(Response::LooseEnds { loose_ends }) => u64::try_from(loose_ends.len()).ok(), + _ => None, + }; // Reminders are harness-local — dial the in-agent socket directly // instead of the broker. let reminders = @@ -412,13 +427,14 @@ impl Surface for AgentSurface { } async fn send_to_parent(socket: &Path, body: String) { - let res = client::request::<_, Response>( + let res = hive_sock_client::request::<_, Response>( socket, &Request::Send { to: hive_sh4re::PARENT_RECIPIENT.into(), body, in_reply_to: None, }, + CONTROL_SOCKET_RETRY, ) .await; if let Err(e) = res { @@ -427,12 +443,13 @@ impl Surface for AgentSurface { } async fn recv_next(socket: &Path) -> RecvOutcome { - let recv: Result = client::request( + let recv: Result = hive_sock_client::request( socket, &Request::Recv { wait_seconds: Some(180), max: None, }, + CONTROL_SOCKET_RETRY, ) .await; match recv { diff --git a/hive-agent/src/todo_server.rs b/hive-agent/src/todo_server.rs index 835b8850..f9a81167 100644 --- a/hive-agent/src/todo_server.rs +++ b/hive-agent/src/todo_server.rs @@ -65,17 +65,13 @@ pub(crate) async fn dial(req: &Request) -> Option { if !path.exists() { return None; } - tokio::time::timeout(std::time::Duration::from_secs(3), async move { - let mut stream = UnixStream::connect(&path).await.ok()?; - let mut line = serde_json::to_string(req).ok()?; - line.push('\n'); - stream.write_all(line.as_bytes()).await.ok()?; - let mut lines = BufReader::new(stream).lines(); - let resp_line = lines.next_line().await.ok()??; - serde_json::from_str(&resp_line).ok() - }) + tokio::time::timeout( + std::time::Duration::from_secs(3), + hive_sock_client::request::<_, Response>(&path, req, hive_sock_client::Retry::None), + ) .await .ok()? + .ok() } /// Run the in-agent socket server: bind + accept loop, one request/response diff --git a/hive-agent/src/web_ui/mod.rs b/hive-agent/src/web_ui/mod.rs index 756a405f..5b06e15d 100644 --- a/hive-agent/src/web_ui/mod.rs +++ b/hive-agent/src/web_ui/mod.rs @@ -305,7 +305,11 @@ async fn broker_request( ) -> std::result::Result { match tokio::time::timeout( SOCKET_FETCH_TIMEOUT, - crate::client::request::<_, hive_core_agent_sock::Response>(socket, req), + hive_sock_client::request::<_, hive_core_agent_sock::Response>( + socket, + req, + crate::CONTROL_SOCKET_RETRY, + ), ) .await { diff --git a/hive-bash-mcp/Cargo.toml b/hive-bash-mcp/Cargo.toml index 984f700b..a11592d8 100644 --- a/hive-bash-mcp/Cargo.toml +++ b/hive-bash-mcp/Cargo.toml @@ -13,6 +13,7 @@ axum.workspace = true clap.workspace = true hive-agent-sock.workspace = true hive-sh4re.workspace = true +hive-sock-client.workspace = true hive-types.workspace = true libc.workspace = true rmcp.workspace = true diff --git a/hive-bash-mcp/src/runner.rs b/hive-bash-mcp/src/runner.rs index edff25a5..21527da5 100644 --- a/hive-bash-mcp/src/runner.rs +++ b/hive-bash-mcp/src/runner.rs @@ -182,33 +182,11 @@ pub fn read_task(id: &str) -> Option { /// on the next task transition, and a standalone daemon without the socket /// simply has nowhere to push. async fn send_todo(socket: &Path, req: &TodoReq) { - use tokio::io::{AsyncBufReadExt as _, BufReader}; - use tokio::net::UnixStream; - let line = match serde_json::to_string(req) { - Ok(mut s) => { - s.push('\n'); - s - } - Err(e) => { - tracing::warn!(error = ?e, "bash_runner: serialise todo failed"); - return; - } - }; - match UnixStream::connect(socket).await { - Ok(stream) => { - let (read, mut write) = stream.into_split(); - if write.write_all(line.as_bytes()).await.is_err() { - tracing::warn!("bash_runner: write todo failed"); - return; - } - let _ = write.shutdown().await; - // Drain the response so the server doesn't get ECONNRESET. - let mut resp = String::new(); - let _ = BufReader::new(read).read_line(&mut resp).await; - } - Err(e) => { - tracing::warn!(error = ?e, socket = %socket.display(), "bash_runner: connect todo socket failed"); - } + // Fail-fast rather than back off: the next task transition pushes the + // todo again, and a runner blocked in a retry schedule would delay the + // task bookkeeping behind it. + if let Err(e) = hive_sock_client::notify(socket, req, hive_sock_client::Retry::None).await { + tracing::warn!(error = ?e, socket = %socket.display(), "bash_runner: todo send failed"); } } diff --git a/hive-forge-notify/Cargo.toml b/hive-forge-notify/Cargo.toml index 1b48664d..cb2bb303 100644 --- a/hive-forge-notify/Cargo.toml +++ b/hive-forge-notify/Cargo.toml @@ -17,6 +17,7 @@ workspace = true anyhow.workspace = true forgejo-api.workspace = true hive-agent-sock.workspace = true +hive-sock-client.workspace = true reqwest.workspace = true serde.workspace = true serde_json.workspace = true diff --git a/hive-forge-notify/src/main.rs b/hive-forge-notify/src/main.rs index 1c253f3c..988d1c3e 100644 --- a/hive-forge-notify/src/main.rs +++ b/hive-forge-notify/src/main.rs @@ -13,7 +13,15 @@ //! forge-less agent settles instead of restart-looping. mod notify; -mod todo_client; + +/// Retry policy for the harness's in-agent socket. Deliberately fail-fast: +/// both callers are inside the 30s poll loop and both treat a failed +/// request as "leave the thread unread and try again next tick", so the +/// poll interval *is* the retry — a second, in-request backoff would only +/// stack sleeps on top of it and delay the rest of the batch. That is the +/// opposite trade-off from the serve loop's client, which rides out a +/// hive-c0re restart because its callers have no natural retry of their own. +const TODO_SOCKET_RETRY: hive_sock_client::Retry = hive_sock_client::Retry::None; #[tokio::main] async fn main() { diff --git a/hive-forge-notify/src/notify.rs b/hive-forge-notify/src/notify.rs index 356b30c4..2dce745f 100644 --- a/hive-forge-notify/src/notify.rs +++ b/hive-forge-notify/src/notify.rs @@ -1078,10 +1078,13 @@ async fn poll_once( summary: body, source: None, }; - let deliver_result = - crate::todo_client::request::<_, hive_agent_sock::Response>(socket, &req) - .await - .map(|_| ()); + let deliver_result = hive_sock_client::request::<_, hive_agent_sock::Response>( + socket, + &req, + crate::TODO_SOCKET_RETRY, + ) + .await + .map(|_| ()); match deliver_result { Ok(()) => { debug!(%id, "forge_notify: todo upserted"); @@ -1245,7 +1248,13 @@ async fn update_assigned_rollup( } }; - match crate::todo_client::request::<_, hive_agent_sock::Response>(socket, &req).await { + match hive_sock_client::request::<_, hive_agent_sock::Response>( + socket, + &req, + crate::TODO_SOCKET_RETRY, + ) + .await + { Ok(_) => debug!(total, "forge_notify: assigned rollup todo updated"), Err(e) => debug!("forge_notify: assigned rollup todo update failed: {e}"), } diff --git a/hive-forge-notify/src/todo_client.rs b/hive-forge-notify/src/todo_client.rs deleted file mode 100644 index b778b8a4..00000000 --- a/hive-forge-notify/src/todo_client.rs +++ /dev/null @@ -1,86 +0,0 @@ -//! One-shot JSON-line client for the harness's in-agent socket -//! (`HIVE_AGENT_SOCKET`) — the only channel this daemon has back into the -//! harness. -//! -//! Deliberately has no retry/backoff schedule. The two callers are both -//! inside the 30s poll loop and both treat a failed request as "leave the -//! thread unread and try again next tick", so the poll interval *is* the -//! retry — a second, in-request backoff would only stack sleeps on top of -//! it and delay the rest of the batch. That is the opposite trade-off from -//! the serve loop's client, which rides out a hive-c0re restart because -//! its callers have no natural retry of their own. -//! -//! Each sibling per-agent daemon carries its own small helper like this -//! one rather than sharing the harness's, so a daemon's socket etiquette -//! stays visible in the crate that depends on it. - -use std::path::Path; - -use anyhow::{Context, Result, anyhow}; -use serde::Serialize; -use serde::de::DeserializeOwned; -use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader}; -use tokio::net::UnixStream; - -/// Write one JSON request line to `socket` and read the single JSON -/// response line back. -/// -/// # Errors -/// -/// Returns an error if the socket cannot be connected, if the request -/// cannot be serialised, if the write or read fails, if the server closes -/// without responding, or if the response does not deserialise into -/// `Resp`. -pub async fn request(socket: &Path, req: &Req) -> Result -where - Req: Serialize + ?Sized, - Resp: DeserializeOwned, -{ - let stream = UnixStream::connect(socket) - .await - .with_context(|| format!("connect to {}", socket.display()))?; - let (read, mut write) = stream.into_split(); - - let mut payload = serde_json::to_string(req)?; - payload.push('\n'); - write - .write_all(payload.as_bytes()) - .await - .with_context(|| format!("write to {}", socket.display()))?; - write - .flush() - .await - .with_context(|| format!("flush {}", socket.display()))?; - - let mut reader = BufReader::new(read); - let mut line = String::new(); - let read_bytes = reader - .read_line(&mut line) - .await - .with_context(|| format!("read from {}", socket.display()))?; - if read_bytes == 0 || line.is_empty() { - return Err(anyhow!( - "{} closed the connection without responding", - socket.display() - )); - } - Ok(serde_json::from_str(line.trim())?) -} - -#[cfg(test)] -mod tests { - use super::request; - - /// Connecting to a path that does not exist is an error, not a hang — - /// the caller (a poll tick) needs it to come back promptly so the rest - /// of the batch still runs. - #[tokio::test] - async fn missing_socket_is_an_error() { - let bogus = std::path::Path::new("/nonexistent/hive/agent.sock"); - let err = request::<(), serde_json::Value>(bogus, &()) - .await - .expect_err("connect to a non-existent socket must fail"); - let msg = format!("{err:#}"); - assert!(msg.contains("connect to"), "missing connect context: {msg}"); - } -} diff --git a/hive-matrix-mcp/Cargo.toml b/hive-matrix-mcp/Cargo.toml index e059e214..5b83916d 100644 --- a/hive-matrix-mcp/Cargo.toml +++ b/hive-matrix-mcp/Cargo.toml @@ -12,6 +12,7 @@ anyhow.workspace = true axum.workspace = true clap.workspace = true futures-util.workspace = true +hive-sock-client.workspace = true matrix-sdk.workspace = true mime = "0.3" mime_guess = "2" diff --git a/hive-matrix-mcp/src/wake.rs b/hive-matrix-mcp/src/wake.rs index 4598af2c..237785bf 100644 --- a/hive-matrix-mcp/src/wake.rs +++ b/hive-matrix-mcp/src/wake.rs @@ -10,11 +10,14 @@ //! agent then reads the unmarked event via the `read_room` MCP tool. //! Truncation to ~100 chars keeps the summary focused. -use std::path::Path; +use anyhow::Result; +use hive_sock_client::{Retry, notify}; -use anyhow::{Context, Result}; -use tokio::io::AsyncWriteExt; -use tokio::net::UnixStream; +/// Retry policy for the in-agent socket. Fail-fast: every caller here is +/// inside the sync loop, which re-derives the whole todo set on its next +/// pass — that pass *is* the retry, and it carries fresher state than a +/// backoff replaying a stale summary would. +const TODO_SOCKET_RETRY: Retry = Retry::None; /// The harness-served in-agent socket (`HIVE_AGENT_SOCKET`) where todo ops /// go — distinct from the host-served control socket used by [`send_wake`]. @@ -46,7 +49,7 @@ pub async fn send_todo_upsert(key: &str, summary: impl AsRef) -> Result<()> "key": key, "summary": summary.as_ref(), }); - send_line(&socket, &payload).await + notify(&socket, &payload, TODO_SOCKET_RETRY).await } /// Clear matrix-subsystem todos on the harness's in-agent socket. `key = @@ -68,38 +71,7 @@ pub async fn send_todo_clear(key: Option<&str>, all: bool) -> Result<()> { "key": key, "all": all, }); - send_line(&socket, &payload).await -} - -/// Write one JSON request line to the hyperhive control socket and drain -/// the response line (best-effort — the reply is not acted on, we just -/// read it so the server doesn't get ECONNRESET on its write-back). -/// Shared by [`send_wake`] and the todo senders. -/// -/// # Errors -/// -/// Returns an error on socket connect failure, serialisation failure, -/// or I/O error writing to or reading from the socket. -async fn send_line(socket: &Path, payload: &serde_json::Value) -> Result<()> { - use tokio::io::AsyncBufReadExt; - - let line = format!("{}\n", serde_json::to_string(payload)?); - let stream = UnixStream::connect(socket) - .await - .with_context(|| format!("connect hyperhive socket {}", socket.display()))?; - let (read, mut write) = stream.into_split(); - write - .write_all(line.as_bytes()) - .await - .with_context(|| format!("write to {}", socket.display()))?; - write - .shutdown() - .await - .with_context(|| format!("shutdown write to {}", socket.display()))?; - let mut reader = tokio::io::BufReader::new(read); - let mut resp = String::new(); - let _ = reader.read_line(&mut resp).await; - Ok(()) + notify(&socket, &payload, TODO_SOCKET_RETRY).await } /// Format a wake-message body from a list of per-room unread summaries. diff --git a/hive-sock-client/Cargo.toml b/hive-sock-client/Cargo.toml new file mode 100644 index 00000000..66799d36 --- /dev/null +++ b/hive-sock-client/Cargo.toml @@ -0,0 +1,21 @@ +# `hive-sock-client` — the one JSON-line-over-unix-socket client shared by +# every daemon that talks to a hyperhive socket. Protocol-agnostic on +# purpose: it is generic over the request/response types, so it works for +# the host-served control socket and the harness's in-agent socket alike. +# Deliberately NOT folded into one of the `*-sock` wire-type crates — +# those stay dependency-free type definitions. +[package] +name = "hive-sock-client" +edition.workspace = true +version.workspace = true +readme = "README.md" + +[lints] +workspace = true + +[dependencies] +anyhow.workspace = true +serde.workspace = true +serde_json.workspace = true +tokio.workspace = true +tracing.workspace = true diff --git a/hive-sock-client/README.md b/hive-sock-client/README.md new file mode 100644 index 00000000..9eb9bdfe --- /dev/null +++ b/hive-sock-client/README.md @@ -0,0 +1,52 @@ +# hive-sock-client + +One JSON-line-over-unix-socket client, shared by every daemon that speaks +to a hyperhive socket. + +The wire protocol is the same everywhere: connect, write one line of JSON, +read one line of JSON back. Before this crate existed, five daemons each +carried their own copy of that — two of them byte-identical — and the +retry, response-handling and error-context behaviour drifted between them. + +The crate is generic over the request and response types, so it is +protocol-agnostic: the host-served control socket and the harness's +in-agent socket both use it, with their own wire-type crates. + +## Shapes + +- `request` / `request_retried` — write, then decode the response line. + `request_retried` additionally reports how many retries it took, for + callers (MCP tool handlers) that want to tell the model a socket flake + happened so it doesn't retry at the LLM level. +- `notify` — write, half-close, drain the response line and discard it. + For fire-and-forget ops where the reply carries nothing the caller acts + on. The drain is not optional: without it the server's write-back lands + on a closed socket. + +## Retry is a policy value, not a fork + +`Retry::RideOutRestart` backs off 2/4/8/16/30s (60s total), sized to ride +out a service restart. It is for callers with no natural retry of their +own — a stall there costs a surfaced tool error and the tokens to handle +it. + +`Retry::None` fails fast. It is for callers already inside a poll loop, +where the poll interval *is* the retry and a second backoff would only +stack sleeps and delay the rest of the batch. + +Two named policies, not a configurable schedule: nobody needs a third yet, +and naming them keeps the *reason* for the choice at the call site. + +## Error contract + +Errors always name the socket path. That detail is load-bearing: a +permission problem on a socket that reads as "is the daemon running?" +sends the operator to fix the wrong thing. + +A refused or missing socket additionally gets a "may be restarting" hint, +and an exhausted retry schedule records how long it tried, so a surfaced +error reads as the likely transient it usually is. + +Serialisation and deserialisation failures are never retried — retrying +identical bytes reproduces the same failure. They are raised outside the +retry loop, so only connect/IO/short-read failures ever reach it. diff --git a/hive-sock-client/src/lib.rs b/hive-sock-client/src/lib.rs new file mode 100644 index 00000000..a4e536f8 --- /dev/null +++ b/hive-sock-client/src/lib.rs @@ -0,0 +1,308 @@ +//! JSON-line-over-unix-socket client, shared by every daemon that talks to +//! a hyperhive socket. +//! +//! The wire protocol is identical everywhere — connect, write one line of +//! JSON, read one line of JSON back — so this crate is generic over the +//! request and response types and knows nothing about either protocol. The +//! host-served control socket and the harness's in-agent socket both use +//! it with their own wire-type crates. +//! +//! Two axes of behaviour are real and stay caller-selectable; everything +//! else is shared: +//! +//! - **retry**: [`Retry::RideOutRestart`] for callers with no natural +//! retry of their own, [`Retry::None`] for callers already inside a poll +//! loop whose interval *is* the retry. +//! - **response**: [`request`] decodes it, [`notify`] drains and discards +//! it. +//! +//! Whether a failure propagates or is logged and swallowed is the caller's +//! choice and stays at the call site — it is not a property of the +//! transport. + +use std::path::Path; +use std::time::Duration; + +use anyhow::{Context, Result, anyhow}; +use serde::Serialize; +use serde::de::DeserializeOwned; +use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader}; +use tokio::net::UnixStream; + +/// Backoff schedule for [`Retry::RideOutRestart`]. Five entries → up to 5 +/// retries on top of the initial attempt; total wall-clock cap = +/// 2+4+8+16+30 = 60s. Sized to ride out a service restart (systemd usually +/// has the unix socket back inside ~5s) without the caller having to +/// handle the transient itself. +const RIDE_OUT_RESTART_BACKOFFS_MS: &[u64] = &[2_000, 4_000, 8_000, 16_000, 30_000]; + +/// What to do when a connect or I/O attempt fails. +/// +/// Deliberately two named policies rather than a configurable schedule: +/// only two behaviours exist in the tree, and naming them keeps the +/// *reason* for each choice readable at the call site. +#[derive(Clone, Copy, Debug, Eq, PartialEq)] +pub enum Retry { + /// Fail on the first failure. For callers inside a poll loop, where + /// the poll interval already is the retry — a second, in-request + /// backoff would stack sleeps on top of it and delay the rest of the + /// batch. + None, + /// Back off on [`RIDE_OUT_RESTART_BACKOFFS_MS`] (~60s total). For + /// callers with no natural retry of their own, where a surfaced + /// transient costs more than the wait. + RideOutRestart, +} + +impl Retry { + /// Sleep schedule between attempts; its length is the retry budget. + fn backoffs(self) -> &'static [u64] { + match self { + Self::None => &[], + Self::RideOutRestart => RIDE_OUT_RESTART_BACKOFFS_MS, + } + } +} + +/// Send `req` over `socket` and decode the single-line JSON response. +/// +/// # Errors +/// +/// Returns an error if `req` cannot be serialised, if the socket is +/// unreachable after the retry budget is spent, if the server closes +/// without responding, or if the response does not deserialise into +/// `Resp`. +pub async fn request(socket: &Path, req: &Req, retry: Retry) -> Result +where + Req: Serialize + ?Sized, + Resp: DeserializeOwned, +{ + request_retried(socket, req, retry) + .await + .map(|(resp, _)| resp) +} + +/// Same as [`request`], but also reports how many retries it took past the +/// initial attempt (0 = succeeded first try). +/// +/// MCP tool handlers use this to append a one-line hint to the tool result +/// when retries happened, so claude reads the earlier socket flake as a +/// transient rather than a content error worth an LLM-level retry. +/// +/// # Errors +/// +/// Same as [`request`]. +pub async fn request_retried( + socket: &Path, + req: &Req, + retry: Retry, +) -> Result<(Resp, u32)> +where + Req: Serialize + ?Sized, + Resp: DeserializeOwned, +{ + let payload = encode(req)?; + let (line, retries) = with_retry(socket, &payload, Mode::Decode, retry).await?; + let resp = serde_json::from_str(line.trim()) + .with_context(|| format!("decode response from {}", socket.display()))?; + Ok((resp, retries)) +} + +/// Send `req` over `socket`, half-close, and drain the response line +/// without decoding it. +/// +/// For fire-and-forget ops whose reply carries nothing the caller acts on. +/// The drain is not optional: without it the server's write-back lands on +/// a closed socket. +/// +/// # Errors +/// +/// Returns an error if `req` cannot be serialised, or if the socket is +/// unreachable / the write fails after the retry budget is spent. A +/// failure to read the discarded response is not an error — the request +/// was already delivered. +pub async fn notify(socket: &Path, req: &Req, retry: Retry) -> Result<()> +where + Req: Serialize + ?Sized, +{ + let payload = encode(req)?; + with_retry(socket, &payload, Mode::Drain, retry).await?; + Ok(()) +} + +/// What to do with the server's response line. +#[derive(Clone, Copy)] +enum Mode { + /// Flush the write half, read the response, hand it back to be parsed. + Decode, + /// Half-close the write half, read-and-discard the response. + Drain, +} + +/// Serialise `req` into one newline-terminated JSON line. +/// +/// Kept outside the retry loop on purpose: a serialisation failure is +/// deterministic, so retrying it would only reproduce it. +fn encode(req: &Req) -> Result> +where + Req: Serialize + ?Sized, +{ + let mut payload = serde_json::to_string(req).context("serialise request")?; + payload.push('\n'); + Ok(payload.into_bytes()) +} + +/// Run [`try_once`] until it succeeds or the retry budget is spent, +/// returning the response line and the number of retries it took. +async fn with_retry( + socket: &Path, + payload: &[u8], + mode: Mode, + retry: Retry, +) -> Result<(String, u32)> { + let backoffs = retry.backoffs(); + let mut attempt: usize = 0; + loop { + match try_once(socket, payload, mode).await { + Ok(line) => return Ok((line, u32::try_from(attempt).unwrap_or(u32::MAX))), + Err(e) => { + let Some(&sleep_ms) = backoffs.get(attempt) else { + return Err(exhausted(e, backoffs)); + }; + tracing::warn!( + attempt = attempt + 1, + sleep_ms, + socket = %socket.display(), + error = %e, + "hive socket attempt failed; retrying" + ); + tokio::time::sleep(Duration::from_millis(sleep_ms)).await; + attempt += 1; + } + } + } +} + +/// Annotate the final failure with how long the retry schedule tried, so a +/// surfaced error says whether it was one shot or a full ride-out. +fn exhausted(err: anyhow::Error, backoffs: &[u64]) -> anyhow::Error { + if backoffs.is_empty() { + return err; + } + let total_s = backoffs.iter().sum::() / 1_000; + err.context(format!( + "gave up after {} retries over ~{total_s}s", + backoffs.len() + )) +} + +/// One connect / write / read cycle. Every error it returns is retryable +/// by construction — the deterministic failures (serialise, deserialise) +/// happen outside the retry loop. +async fn try_once(socket: &Path, payload: &[u8], mode: Mode) -> Result { + let stream = UnixStream::connect(socket).await.map_err(|e| { + // A refused or missing socket usually means the listener is + // mid-restart (operator redeploy, harness restart) — it is + // recreated on boot and a retrying caller rides it out. When the + // error does surface (budget spent, or a fail-fast caller) the + // hint marks it as a likely transient rather than a hard failure + // worth escalating. The path stays in the error either way: a + // permission problem that reads as "is the daemon running?" sends + // the operator to fix the wrong thing. + let restarting = matches!( + e.kind(), + std::io::ErrorKind::ConnectionRefused | std::io::ErrorKind::NotFound + ); + let err = anyhow::Error::new(e).context(format!("connect to {}", socket.display())); + if restarting { + err.context("the listener may be restarting (e.g. an operator redeploy)") + } else { + err + } + })?; + let (read, mut write) = stream.into_split(); + + write + .write_all(payload) + .await + .with_context(|| format!("write to {}", socket.display()))?; + match mode { + Mode::Decode => write + .flush() + .await + .with_context(|| format!("flush {}", socket.display()))?, + Mode::Drain => write + .shutdown() + .await + .with_context(|| format!("shutdown write to {}", socket.display()))?, + } + + let mut reader = BufReader::new(read); + let mut line = String::new(); + match mode { + Mode::Decode => { + let read_bytes = reader + .read_line(&mut line) + .await + .with_context(|| format!("read from {}", socket.display()))?; + if read_bytes == 0 || line.is_empty() { + return Err(anyhow!( + "{} closed the connection without responding", + socket.display() + )); + } + } + Mode::Drain => { + // Discarded, including any error reading it: the request is + // already delivered and the caller acts on nothing here. + let _ = reader.read_line(&mut line).await; + line.clear(); + } + } + Ok(line) +} + +#[cfg(test)] +mod tests { + use super::{Retry, notify, request}; + + /// A connect to a non-existent socket path (ENOENT → `NotFound`) is + /// annotated with both the socket path and the "may be restarting" + /// hint, so a surfaced tool error reads as the expected transient. + #[tokio::test] + async fn missing_socket_names_the_path_and_hints_restart() { + let bogus = std::path::Path::new("/nonexistent/hive/mcp.sock"); + let err = request::<(), serde_json::Value>(bogus, &(), Retry::None) + .await + .expect_err("connect to a non-existent socket must fail"); + let msg = format!("{err:#}"); + assert!(msg.contains("restarting"), "missing restart hint: {msg}"); + assert!(msg.contains("connect to"), "missing connect context: {msg}"); + } + + /// `Retry::None` returns promptly rather than sleeping the ride-out + /// schedule — its callers are poll ticks that need the rest of the + /// batch to still run this cycle. + #[tokio::test] + async fn no_retry_fails_fast() { + let bogus = std::path::Path::new("/nonexistent/hive/agent.sock"); + let started = std::time::Instant::now(); + notify(bogus, &(), Retry::None) + .await + .expect_err("connect to a non-existent socket must fail"); + assert!( + started.elapsed() < std::time::Duration::from_secs(1), + "Retry::None slept: {:?}", + started.elapsed() + ); + } + + /// The ride-out schedule is a full minute of patience — the value the + /// harness's tool callers depend on to not surface a restart. + #[test] + fn ride_out_restart_budget_is_a_minute() { + let total_ms: u64 = Retry::RideOutRestart.backoffs().iter().sum(); + assert_eq!(total_ms, 60_000); + assert!(Retry::None.backoffs().is_empty()); + } +}