diff --git a/hive-ag3nt/src/client.rs b/hive-ag3nt/src/client.rs index 64e12932..5a314bd4 100644 --- a/hive-ag3nt/src/client.rs +++ b/hive-ag3nt/src/client.rs @@ -1,109 +1,33 @@ use std::path::Path; -use std::time::Duration; -use anyhow::{Context, Result, anyhow}; +use anyhow::{Context, Result, bail}; 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. +/// Generic JSON-line request/response over a unix socket. One request, one +/// response, then drop. Used by both the agent and manager harnesses. 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. -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); - } - } - } - } - Err(last_err.unwrap_or_else(|| anyhow!("hive socket: retries exhausted"))) -} - -/// 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 = UnixStream::connect(socket) .await - .with_context(|| format!("connect to {}", socket.display())) - .map_err(RequestError::Transient)?; + .with_context(|| format!("connect to {}", socket.display()))?; let (read, mut write) = stream.into_split(); - let mut payload = serde_json::to_string(req).map_err(|e| RequestError::Fatal(e.into()))?; + let mut payload = serde_json::to_string(req)?; 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()))?; + write.write_all(payload.as_bytes()).await?; + write.flush().await?; 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" - ))); + reader.read_line(&mut line).await?; + if line.is_empty() { + bail!("server closed connection without responding"); } - serde_json::from_str(line.trim()).map_err(|e| RequestError::Fatal(e.into())) + Ok(serde_json::from_str(line.trim())?) } diff --git a/hive-ag3nt/src/mcp.rs b/hive-ag3nt/src/mcp.rs index 0a726395..90a8e739 100644 --- a/hive-ag3nt/src/mcp.rs +++ b/hive-ag3nt/src/mcp.rs @@ -107,22 +107,6 @@ where result } -/// Append a short note to a tool result when the underlying socket call -/// took retries to land. Lets claude distinguish "my request was wrong" -/// from "c0re flickered and the harness rode it out" — without the -/// hint, a tool result that took 30s to come back looks identical to a -/// content failure and the model would burn a turn retrying it. -pub fn annotate_retries(mut s: String, retries: u32) -> String { - if retries > 0 { - let suffix = if retries == 1 { "retry" } else { "retries" }; - s.push_str(&format!( - "\n\n(note: hive socket connect needed {retries} {suffix} — c0re likely \ - restarted. Your request did succeed on the final attempt; no action needed.)" - )); - } - s -} - #[derive(Debug, serde::Deserialize, schemars::JsonSchema)] pub struct SendArgs { /// Logical agent name to deliver the message to (e.g. `"manager"`, @@ -154,19 +138,6 @@ impl AgentServer { pub fn new(socket: PathBuf) -> Self { Self { socket } } - - /// Issue any `AgentRequest` through the retry-aware client and pull - /// the reply through `SocketReply`. Returns the retry count so tool - /// handlers can annotate their result (see `annotate_retries`). - async fn dispatch( - &self, - req: hive_sh4re::AgentRequest, - ) -> (Result, u32) { - match client::request_retried::<_, hive_sh4re::AgentResponse>(&self.socket, &req).await { - Ok((r, n)) => (Ok(SocketReply::from(r)), n), - Err(e) => (Err(e), 0), - } - } } #[tool_router] @@ -182,13 +153,16 @@ impl AgentServer { return run_tool_envelope("send", log, async move { refusal }).await; } run_tool_envelope("send", log, async move { - let (resp, retries) = self - .dispatch(hive_sh4re::AgentRequest::Send { + let resp = client::request::<_, hive_sh4re::AgentResponse>( + &self.socket, + &hive_sh4re::AgentRequest::Send { to: args.to, body: args.body, - }) - .await; - annotate_retries(format_ack(resp, "send", format!("sent to {to}")), retries) + }, + ) + .await + .map(SocketReply::from); + format_ack(resp, "send", format!("sent to {to}")) }) .await } @@ -208,15 +182,18 @@ impl AgentServer { async fn ask_operator(&self, Parameters(args): Parameters) -> String { let log = format!("{args:?}"); run_tool_envelope("ask_operator", log, async move { - let (resp, retries) = self - .dispatch(hive_sh4re::AgentRequest::AskOperator { + let resp = client::request::<_, hive_sh4re::AgentResponse>( + &self.socket, + &hive_sh4re::AgentRequest::AskOperator { question: args.question, options: args.options, multi: args.multi, ttl_seconds: args.ttl_seconds, - }) - .await; - let s = match resp { + }, + ) + .await + .map(SocketReply::from); + match resp { Ok(SocketReply::QuestionQueued(id)) => format!( "question queued (id={id}); operator's answer will arrive as a system \ `operator_answered` event in your inbox" @@ -224,8 +201,7 @@ impl AgentServer { Ok(SocketReply::Err(m)) => format!("ask_operator failed: {m}"), Ok(other) => format!("ask_operator unexpected response: {other:?}"), Err(e) => format!("ask_operator transport error: {e:#}"), - }; - annotate_retries(s, retries) + } }) .await } @@ -243,12 +219,15 @@ impl AgentServer { async fn recv(&self, Parameters(args): Parameters) -> String { let log = format!("{args:?}"); run_tool_envelope("recv", log, async move { - let (resp, retries) = self - .dispatch(hive_sh4re::AgentRequest::Recv { + let resp = client::request::<_, hive_sh4re::AgentResponse>( + &self.socket, + &hive_sh4re::AgentRequest::Recv { wait_seconds: args.wait_seconds, - }) - .await; - annotate_retries(format_recv(resp), retries) + }, + ) + .await + .map(SocketReply::from); + format_recv(resp) }) .await } @@ -362,18 +341,15 @@ impl ManagerServer { Self { socket } } - /// Helper: issue any `ManagerRequest` through the retry-aware - /// client, convert the reply through `SocketReply`, and return the - /// retry count alongside so the tool handler can `annotate_retries` - /// on the final string. + /// Helper: issue any `ManagerRequest`, convert the reply through + /// `SocketReply`. Manager tools that just need an `Ok` ack share this. async fn dispatch( &self, req: hive_sh4re::ManagerRequest, - ) -> (Result, u32) { - match client::request_retried::<_, hive_sh4re::ManagerResponse>(&self.socket, &req).await { - Ok((r, n)) => (Ok(SocketReply::from(r)), n), - Err(e) => (Err(e), 0), - } + ) -> Result { + client::request::<_, hive_sh4re::ManagerResponse>(&self.socket, &req) + .await + .map(SocketReply::from) } } @@ -387,13 +363,13 @@ impl ManagerServer { let log = format!("{args:?}"); let to = args.to.clone(); run_tool_envelope("send", log, async move { - let (resp, retries) = self + let resp = self .dispatch(hive_sh4re::ManagerRequest::Send { to: args.to, body: args.body, }) .await; - annotate_retries(format_ack(resp, "send", format!("sent to {to}")), retries) + format_ack(resp, "send", format!("sent to {to}")) }) .await } @@ -408,12 +384,12 @@ impl ManagerServer { async fn recv(&self, Parameters(args): Parameters) -> String { let log = format!("{args:?}"); run_tool_envelope("recv", log, async move { - let (resp, retries) = self + let resp = self .dispatch(hive_sh4re::ManagerRequest::Recv { wait_seconds: args.wait_seconds, }) .await; - annotate_retries(format_recv(resp), retries) + format_recv(resp) }) .await } @@ -426,19 +402,16 @@ impl ManagerServer { let log = format!("{args:?}"); let name = args.name.clone(); run_tool_envelope("request_spawn", log, async move { - let (resp, retries) = self + let resp = self .dispatch(hive_sh4re::ManagerRequest::RequestSpawn { name: args.name, description: args.description, }) .await; - annotate_retries( - format_ack( - resp, - "request_spawn", - format!("spawn approval queued for {name}"), - ), - retries, + format_ack( + resp, + "request_spawn", + format!("spawn approval queued for {name}"), ) }) .await @@ -452,10 +425,10 @@ impl ManagerServer { let log = format!("{args:?}"); let name = args.name.clone(); run_tool_envelope("kill", log, async move { - let (resp, retries) = self + let resp = self .dispatch(hive_sh4re::ManagerRequest::Kill { name: args.name }) .await; - annotate_retries(format_ack(resp, "kill", format!("killed {name}")), retries) + format_ack(resp, "kill", format!("killed {name}")) }) .await } @@ -468,10 +441,10 @@ impl ManagerServer { let log = format!("{args:?}"); let name = args.name.clone(); run_tool_envelope("start", log, async move { - let (resp, retries) = self + let resp = self .dispatch(hive_sh4re::ManagerRequest::Start { name: args.name }) .await; - annotate_retries(format_ack(resp, "start", format!("started {name}")), retries) + format_ack(resp, "start", format!("started {name}")) }) .await } @@ -481,13 +454,10 @@ impl ManagerServer { let log = format!("{args:?}"); let name = args.name.clone(); run_tool_envelope("restart", log, async move { - let (resp, retries) = self + let resp = self .dispatch(hive_sh4re::ManagerRequest::Restart { name: args.name }) .await; - annotate_retries( - format_ack(resp, "restart", format!("restarted {name}")), - retries, - ) + format_ack(resp, "restart", format!("restarted {name}")) }) .await } @@ -501,13 +471,10 @@ impl ManagerServer { let log = format!("{args:?}"); let name = args.name.clone(); run_tool_envelope("update", log, async move { - let (resp, retries) = self + let resp = self .dispatch(hive_sh4re::ManagerRequest::Update { name: args.name }) .await; - annotate_retries( - format_ack(resp, "update", format!("updated {name}")), - retries, - ) + format_ack(resp, "update", format!("updated {name}")) }) .await } @@ -527,7 +494,7 @@ impl ManagerServer { async fn ask_operator(&self, Parameters(args): Parameters) -> String { let log = format!("{args:?}"); run_tool_envelope("ask_operator", log, async move { - let (resp, retries) = self + let resp = self .dispatch(hive_sh4re::ManagerRequest::AskOperator { question: args.question, options: args.options, @@ -535,7 +502,7 @@ impl ManagerServer { ttl_seconds: args.ttl_seconds, }) .await; - let s = match resp { + match resp { Ok(SocketReply::QuestionQueued(id)) => format!( "question queued (id={id}); operator's answer will arrive as a system \ `operator_answered` event in your inbox" @@ -543,8 +510,7 @@ impl ManagerServer { Ok(SocketReply::Err(m)) => format!("ask_operator failed: {m}"), Ok(other) => format!("ask_operator unexpected response: {other:?}"), Err(e) => format!("ask_operator transport error: {e:#}"), - }; - annotate_retries(s, retries) + } }) .await } @@ -562,20 +528,17 @@ impl ManagerServer { let agent = args.agent.clone(); let commit_ref = args.commit_ref.clone(); run_tool_envelope("request_apply_commit", log, async move { - let (resp, retries) = self + let resp = self .dispatch(hive_sh4re::ManagerRequest::RequestApplyCommit { agent: args.agent, commit_ref: args.commit_ref, description: args.description, }) .await; - annotate_retries( - format_ack( - resp, - "request_apply_commit", - format!("apply approval queued for {agent} @ {commit_ref}"), - ), - retries, + format_ack( + resp, + "request_apply_commit", + format!("apply approval queued for {agent} @ {commit_ref}"), ) }) .await diff --git a/hive-ag3nt/src/turn.rs b/hive-ag3nt/src/turn.rs index b9fe6fe0..651f2a72 100644 --- a/hive-ag3nt/src/turn.rs +++ b/hive-ag3nt/src/turn.rs @@ -290,11 +290,6 @@ async fn run_claude(prompt: &str, files: &TurnFiles, bus: &Bus) -> Result if line.contains(PROMPT_TOO_LONG_MARKER) { flag_err.store(true, Ordering::Relaxed); } - // Mirror to journald so post-mortems work without the web UI - // or the events sqlite. The bus event is what the dashboard - // renders; the tracing line is what `journalctl -M -b` - // surfaces when claude exits non-zero. - tracing::warn!(line = %line, "claude stderr"); bus_err.emit(LiveEvent::Note(format!("stderr: {line}"))); } });