mara/hyperhive
mara/hyperhive
Watch
0
0
Fork
You've already forked hyperhive
 0
Code Activity

refactor(sock): one socket client, retry as a policy value

Browse source 
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.
This commit is contained in:
atlas 2026-07-26 22:20:01 +02:00 • committed by mara
commit 7a826f9ee2
23 changed files with 498 additions and 501 deletions
Download patch file Download diff file Expand all files Collapse all files

9
CLAUDE.md
View file

@ -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.

16
Cargo.lock generated
View file

@ -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"

2
Cargo.toml
View file

@ -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"] }

1
hive-agent-mcp/Cargo.toml
View file

@ -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

152
hive-agent-mcp/src/client.rs
View file

@ -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<Req, Resp>(socket: &Path, req: &Req) -> Result<(Resp, u32)>
where
Req: Serialize + ?Sized,
Resp: DeserializeOwned,
{
let mut last_err: Option<anyhow::Error> = None;
let max_retries = u32::try_from(RETRY_BACKOFFS_MS.len()).unwrap();
for attempt in 0..=max_retries {
match try_once::<Req, Resp>(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<Req, Resp>(socket: &Path, req: &Req) -> Result<Resp, RequestError>
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"),
}
}
}

1
hive-agent-mcp/src/main.rs
View file

@ -16,7 +16,6 @@ use std::path::PathBuf;
use anyhow::Result;
use clap::Parser;
mod client;
mod mcp;
mod paths;
mod send_allow;

12
hive-agent-mcp/src/mcp/mod.rs
View file

@ -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<hive_core_agent_sock::Response, anyhow::Error>, 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),

1
hive-agent/Cargo.toml
View file

@ -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

167
hive-agent/src/client.rs
View file

@ -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<Req, Resp>(socket: &Path, req: &Req) -> Result<Resp>
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<Req, Resp>(socket: &Path, req: &Req) -> Result<(Resp, u32)>
where
Req: Serialize + ?Sized,
Resp: DeserializeOwned,
{
let mut last_err: Option<anyhow::Error> = None;
let max_retries = u32::try_from(RETRY_BACKOFFS_MS.len()).unwrap();
for attempt in 0..=max_retries {
match try_once::<Req, Resp>(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<Req, Resp>(socket: &Path, req: &Req) -> Result<Resp, RequestError>
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"),
}
}
}

41
hive-agent/src/main.rs
View file

@ -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<u64>, Option<u64>) {
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<Response> = client::request(
let recv: Result<Response> = hive_sock_client::request(
socket,
&Request::Recv {
wait_seconds: Some(180),
max: None,
},
CONTROL_SOCKET_RETRY,
)
.await;
match recv {

14
hive-agent/src/todo_server.rs
View file

@ -65,17 +65,13 @@ pub(crate) async fn dial(req: &Request) -> Option<Response> {
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

6
hive-agent/src/web_ui/mod.rs
View file

@ -305,7 +305,11 @@ async fn broker_request(
) -> std::result::Result<hive_core_agent_sock::Response, BrokerError> {
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
{

1
hive-bash-mcp/Cargo.toml
View file

@ -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

32
hive-bash-mcp/src/runner.rs
View file

@ -182,33 +182,11 @@ pub fn read_task(id: &str) -> Option<TaskFile> {
/// 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");
}
}

1
hive-forge-notify/Cargo.toml
View file

@ -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

10
hive-forge-notify/src/main.rs
View file

@ -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() {

19
hive-forge-notify/src/notify.rs
View file

@ -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}"),
}

86
hive-forge-notify/src/todo_client.rs
View file

@ -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<Req, Resp>(socket: &Path, req: &Req) -> Result<Resp>
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}");
}
}

1
hive-matrix-mcp/Cargo.toml
View file

@ -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"

46
hive-matrix-mcp/src/wake.rs
View file

@ -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<str>) -> 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.

Some files were not shown because too many files have changed in this diff Show more