From 59dd3a0aa7021c88971e6b7d7ed650f83daabf57 Mon Sep 17 00:00:00 2001 From: damocles Date: Tue, 14 Jul 2026 23:26:06 +0200 Subject: [PATCH] refactor(#2456): split hive-agent-wake into its own bin crate --- Cargo.lock | 14 ++ Cargo.toml | 1 + hive-ag3nt/src/bin/hive-agent-wake.rs | 60 -------- hive-agent-wake/Cargo.toml | 21 +++ hive-agent-wake/src/main.rs | 189 ++++++++++++++++++++++++++ 5 files changed, 225 insertions(+), 60 deletions(-) delete mode 100644 hive-ag3nt/src/bin/hive-agent-wake.rs create mode 100644 hive-agent-wake/Cargo.toml create mode 100644 hive-agent-wake/src/main.rs diff --git a/Cargo.lock b/Cargo.lock index c530ea47..83e62455 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1489,6 +1489,20 @@ dependencies = [ "url", ] +[[package]] +name = "hive-agent-wake" +version = "0.1.0" +dependencies = [ + "anyhow", + "clap", + "hive-sh4re", + "serde", + "serde_json", + "tokio", + "tracing", + "tracing-subscriber", +] + [[package]] name = "hive-bash-mcp" version = "0.1.0" diff --git a/Cargo.toml b/Cargo.toml index f4e3e7c5..c5442811 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -2,6 +2,7 @@ resolver = "3" members = [ "hive-ag3nt", + "hive-agent-wake", "hive-bash-mcp", "hive-c0re", "hive-claude", diff --git a/hive-ag3nt/src/bin/hive-agent-wake.rs b/hive-ag3nt/src/bin/hive-agent-wake.rs deleted file mode 100644 index f5e599bf..00000000 --- a/hive-ag3nt/src/bin/hive-agent-wake.rs +++ /dev/null @@ -1,60 +0,0 @@ -//! Wake CLI: inject a wake-up event into this container's harness inbox -//! so the next turn fires with the given body. Intended for extra MCP -//! servers / helpers (scraper, webhook listener, etc.) that need to -//! nudge claude on external events; the built-in daemons (matrix, bash) -//! talk to the socket directly instead. Sibling of `hive-agent` and -//! `hive-agent-mcp`. - -use std::path::PathBuf; - -use anyhow::Result; -use clap::Parser; -use hive_ag3nt::{DEFAULT_SOCKET, client}; -use hive_sh4re::{AgentRequest, AgentResponse}; - -#[derive(Parser)] -#[command(name = "hive-agent-wake", about = "hyperhive harness wake signal")] -struct Cli { - /// Path to the per-agent MCP socket (bind-mounted from the host). - #[arg(long, default_value = DEFAULT_SOCKET)] - socket: PathBuf, - - #[arg(long)] - from: String, - - /// Body of the wake message. Pass `-` to read from stdin. - #[arg(long)] - body: String, -} - -#[tokio::main] -async fn main() -> Result<()> { - tracing_subscriber::fmt() - .with_env_filter( - tracing_subscriber::EnvFilter::try_from_default_env() - .unwrap_or_else(|_| tracing_subscriber::EnvFilter::new("info")), - ) - .init(); - - let cli = Cli::parse(); - let body = if cli.body == "-" { - let mut buf = String::new(); - std::io::Read::read_to_string(&mut std::io::stdin(), &mut buf)?; - buf - } else { - cli.body - }; - let resp: AgentResponse = client::request( - &cli.socket, - &AgentRequest::Wake { - from: cli.from, - body, - }, - ) - .await?; - match resp { - AgentResponse::Ok => Ok(()), - AgentResponse::Err { message } => anyhow::bail!("wake: {message}"), - other => anyhow::bail!("wake: unexpected response {other:?}"), - } -} diff --git a/hive-agent-wake/Cargo.toml b/hive-agent-wake/Cargo.toml new file mode 100644 index 00000000..2d1fd492 --- /dev/null +++ b/hive-agent-wake/Cargo.toml @@ -0,0 +1,21 @@ +[package] +name = "hive-agent-wake" +version.workspace = true +edition.workspace = true + +[[bin]] +name = "hive-agent-wake" +path = "src/main.rs" + +[lints] +workspace = true + +[dependencies] +anyhow.workspace = true +clap.workspace = true +hive-sh4re.workspace = true +serde.workspace = true +serde_json.workspace = true +tokio.workspace = true +tracing.workspace = true +tracing-subscriber.workspace = true diff --git a/hive-agent-wake/src/main.rs b/hive-agent-wake/src/main.rs new file mode 100644 index 00000000..4afce17f --- /dev/null +++ b/hive-agent-wake/src/main.rs @@ -0,0 +1,189 @@ +//! Wake CLI: inject a wake-up event into this container's harness inbox +//! so the next turn fires with the given body. Intended for extra MCP +//! servers / helpers (scraper, webhook listener, etc.) that need to +//! nudge claude on external events; the built-in daemons (matrix, bash) +//! talk to the socket directly instead. +//! +//! Standalone bin crate: it dials the per-agent MCP socket directly and +//! carries its own copy of the retrying request client (below), so it +//! does not link the whole `hive-ag3nt` harness lib — a helper author +//! wiring up an `extraMcpServers` binary only needs this one small crate. + +use std::path::PathBuf; + +use anyhow::Result; +use clap::Parser; +use hive_sh4re::{AgentRequest, AgentResponse}; + +/// Per-agent MCP socket, bind-mounted from the host into every container. +const DEFAULT_SOCKET: &str = "/run/hive/mcp.sock"; + +#[derive(Parser)] +#[command(name = "hive-agent-wake", about = "hyperhive harness wake signal")] +struct Cli { + /// Path to the per-agent MCP socket (bind-mounted from the host). + #[arg(long, default_value = DEFAULT_SOCKET)] + socket: PathBuf, + + #[arg(long)] + from: String, + + /// Body of the wake message. Pass `-` to read from stdin. + #[arg(long)] + body: String, +} + +#[tokio::main] +async fn main() -> Result<()> { + tracing_subscriber::fmt() + .with_env_filter( + tracing_subscriber::EnvFilter::try_from_default_env() + .unwrap_or_else(|_| tracing_subscriber::EnvFilter::new("info")), + ) + .init(); + + let cli = Cli::parse(); + let body = if cli.body == "-" { + let mut buf = String::new(); + std::io::Read::read_to_string(&mut std::io::stdin(), &mut buf)?; + buf + } else { + cli.body + }; + let resp: AgentResponse = client::request( + &cli.socket, + &AgentRequest::Wake { + from: cli.from, + body, + }, + ) + .await?; + match resp { + AgentResponse::Ok => Ok(()), + AgentResponse::Err { message } => anyhow::bail!("wake: {message}"), + other => anyhow::bail!("wake: unexpected response {other:?}"), + } +} + +/// Self-contained retrying unix-socket JSON request client. A trimmed +/// copy of `hive_ag3nt::client` (no `request_retried` variant — the wake +/// CLI never needs the retry-count) so this bin does not link the harness +/// lib. Keeping the retry matters: the socket can be briefly absent while +/// hive-c0re restarts under an operator redeploy, and a helper firing a +/// wake shouldn't spuriously fail then. +mod client { + 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 caller having to handle the + /// transient itself. + 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. + /// + /// # 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, + { + let mut last_err: Option = None; + // One attempt per backoff entry, plus a final attempt with no sleep + // after it — so `len + 1` tries, retrying only transient failures. + for attempt in 0..=RETRY_BACKOFFS_MS.len() { + match try_once::(socket, req).await { + Ok(resp) => return Ok(resp), + Err(RequestError::Fatal(e)) => return Err(e), + Err(RequestError::Transient(e)) => { + if let Some(&sleep_ms) = RETRY_BACKOFFS_MS.get(attempt) { + tracing::warn!( + attempt = attempt + 1, + sleep_ms, + error = %e, + "hive socket attempt failed; retrying" + ); + tokio::time::sleep(Duration::from_millis(sleep_ms)).await; + } + last_err = Some(e); + } + } + } + // The final iteration always sets `last_err` on a transient failure. + 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 the retry loop rides it out. + 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 CLI 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())) + } +}