diff --git a/Cargo.lock b/Cargo.lock index abf86db0..22ddacbc 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1523,6 +1523,7 @@ dependencies = [ "clap", "forgejo-api", "futures-util", + "hive-agent-sock", "hive-claude", "hive-sh4re", "http-body-util", @@ -1551,6 +1552,7 @@ dependencies = [ "anyhow", "axum", "clap", + "hive-agent-sock", "hive-sh4re", "rmcp", "serde", @@ -1560,13 +1562,21 @@ dependencies = [ "tracing-subscriber", ] +[[package]] +name = "hive-agent-sock" +version = "0.1.0" +dependencies = [ + "hive-sh4re", + "serde", +] + [[package]] name = "hive-agent-wake" version = "0.1.0" dependencies = [ "anyhow", "clap", - "hive-sh4re", + "hive-agent-sock", "serde", "serde_json", "tokio", @@ -1579,6 +1589,7 @@ name = "hive-bash-mcp" version = "0.1.0" dependencies = [ "anyhow", + "hive-agent-sock", "hive-sh4re", "libc", "rmcp", @@ -1604,6 +1615,7 @@ dependencies = [ "clap-markdown", "clap_complete", "forgejo-api", + "hive-agent-sock", "hive-host-sock", "hive-priv-sock", "hive-sh4re", diff --git a/Cargo.toml b/Cargo.toml index 771ae3c3..4e4decc1 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -3,6 +3,7 @@ resolver = "3" members = [ "hive-agent", "hive-agent-mcp", + "hive-agent-sock", "hive-agent-wake", "hive-bash-mcp", "hive-c0re", @@ -48,6 +49,7 @@ clap = { version = "4", features = ["derive"] } clap_complete = "4" indicatif = "0.18" hive-sh4re = { path = "hive-sh4re" } +hive-agent-sock = { path = "hive-agent-sock" } hive-claude = { path = "hive-claude" } hive-host-sock = { path = "hive-host-sock" } hive-priv-sock = { path = "hive-priv-sock" } diff --git a/hive-agent-mcp/Cargo.toml b/hive-agent-mcp/Cargo.toml index 531840e1..7c4ee9c2 100644 --- a/hive-agent-mcp/Cargo.toml +++ b/hive-agent-mcp/Cargo.toml @@ -14,6 +14,7 @@ workspace = true anyhow.workspace = true axum.workspace = true clap.workspace = true +hive-agent-sock.workspace = true hive-sh4re.workspace = true rmcp.workspace = true serde.workspace = true diff --git a/hive-agent-mcp/src/mcp/mod.rs b/hive-agent-mcp/src/mcp/mod.rs index deb06454..6cf8cb1b 100644 --- a/hive-agent-mcp/src/mcp/mod.rs +++ b/hive-agent-mcp/src/mcp/mod.rs @@ -1,7 +1,7 @@ //! Embedded MCP server. Claude Code (running inside the agent container) //! connects to this over streamable-HTTP via `--mcp-config` (the long-lived //! `hive-mcp-http` daemon); tool calls land here and are translated to -//! `AgentRequest::*` / `ManagerRequest::*` against hyperhive's own +//! `Request::*` against hyperhive's own //! per-container unix socket at `/run/hive/mcp.sock`. //! //! Two protocols, two surfaces: @@ -99,11 +99,9 @@ where /// Unified MCP tool surface for both sub-agent and manager roles. /// -/// `AgentRequest = ManagerRequest = Request` and `AgentResponse = -/// ManagerResponse = Response` are type aliases in hive-sh4re, so a single +/// Both sockets speak the same `Request` / `Response` wire, so a single /// `dispatch` call covers both sockets — the only real difference is which /// socket path is used and which tools the flavor enables. -/// #[derive(Debug, Clone)] pub struct AgentServer { socket: PathBuf, @@ -119,13 +117,13 @@ impl AgentServer { /// `Response` plus the retry count so tool handlers can annotate their /// result (see `annotate_retries`). /// - /// `AgentRequest` / `ManagerRequest` / `Request` are all the same type - /// (hive-sh4re type aliases), so this single method covers both sockets. + /// Both sockets speak the same `Request` type, so this single method + /// covers both. async fn dispatch( &self, - req: hive_sh4re::Request, - ) -> (Result, u32) { - match client::request_retried::<_, hive_sh4re::Response>(&self.socket, &req).await { + req: hive_agent_sock::Request, + ) -> (Result, u32) { + match client::request_retried::<_, hive_agent_sock::Response>(&self.socket, &req).await { Ok((r, n)) => (Ok(r), n), Err(e) => (Err(e), 0), } @@ -152,7 +150,7 @@ impl AgentServer { } run_tool_envelope("send", log, async move { let (resp, retries) = self - .dispatch(hive_sh4re::Request::Send { + .dispatch(hive_agent_sock::Request::Send { to: args.to, body: args.body, in_reply_to: args.in_reply_to, @@ -183,7 +181,7 @@ impl AgentServer { let log = format!("{args:?}"); run_tool_envelope("ask", log, async move { let (resp, retries) = self - .dispatch(hive_sh4re::Request::Ask { + .dispatch(hive_agent_sock::Request::Ask { question: args.question, options: args.options, multi: args.multi, @@ -192,7 +190,7 @@ impl AgentServer { }) .await; let s = match resp { - Ok(hive_sh4re::Response::QuestionQueued { id }) => format!( + Ok(hive_agent_sock::Response::QuestionQueued { id }) => format!( "question queued (id={id}); answer will arrive as a system \ `question_answered` event in your inbox" ), @@ -217,7 +215,7 @@ impl AgentServer { let id = args.id; run_tool_envelope("answer", log, async move { let (resp, retries) = self - .dispatch(hive_sh4re::Request::Answer { + .dispatch(hive_agent_sock::Request::Answer { id, answer: args.answer, }) @@ -255,7 +253,7 @@ impl AgentServer { run_tool_envelope("recv", log, async move { let waited = args.wait_seconds.is_some_and(|w| w > 0); let (resp, retries) = self - .dispatch(hive_sh4re::Request::Recv { + .dispatch(hive_agent_sock::Request::Recv { wait_seconds: args.wait_seconds, max: args.max, }) @@ -280,10 +278,10 @@ impl AgentServer { let log = format!("{args:?}"); run_tool_envelope("ack_until", log, async move { let (resp, retries) = self - .dispatch(hive_sh4re::Request::AckUntil { up_to: args.up_to }) + .dispatch(hive_agent_sock::Request::AckUntil { up_to: args.up_to }) .await; let rendered = match resp { - Ok(hive_sh4re::Response::Acked { count }) => { + Ok(hive_agent_sock::Response::Acked { count }) => { format!("acked {count} message(s) up to id {}", args.up_to) } other => reply_err(other, "ack_until"), @@ -312,11 +310,11 @@ impl AgentServer { run_tool_envelope("get_loose_ends", String::new(), async move { let is_self_query = args.agent.is_none(); let (resp, retries) = self - .dispatch(hive_sh4re::Request::GetLooseEnds { agent: args.agent }) + .dispatch(hive_agent_sock::Request::GetLooseEnds { agent: args.agent }) .await; // Extract the vec so we can augment before rendering. let mut loose_ends = match resp { - Ok(hive_sh4re::Response::LooseEnds { loose_ends }) => loose_ends, + Ok(hive_agent_sock::Response::LooseEnds { loose_ends }) => loose_ends, other => return annotate_retries(reply_err(other, "get_loose_ends"), retries), }; // Prepend matrix unread entry for self-queries only (can't @@ -365,7 +363,7 @@ impl AgentServer { return e; } let (resp, retries) = self - .dispatch(hive_sh4re::Request::SetStatus { text: args.text }) + .dispatch(hive_agent_sock::Request::SetStatus { text: args.text }) .await; annotate_retries( format_ack(resp, "set_status", "status updated".to_owned()), @@ -397,7 +395,7 @@ impl AgentServer { let log = args.name.clone().unwrap_or_else(|| "".to_owned()); run_tool_envelope("get_agent_meta", log, async move { let (resp, retries) = self - .dispatch(hive_sh4re::Request::GetAgentMeta { name: args.name }) + .dispatch(hive_agent_sock::Request::GetAgentMeta { name: args.name }) .await; annotate_retries(format_agent_meta(resp), retries) }) @@ -425,7 +423,7 @@ impl AgentServer { }; let kind_label = loose_end_kind_label(kind); let (resp, retries) = self - .dispatch(hive_sh4re::Request::CancelLooseEnd { kind, id }) + .dispatch(hive_agent_sock::Request::CancelLooseEnd { kind, id }) .await; annotate_retries( format_ack( @@ -452,10 +450,10 @@ impl AgentServer { let log = format!("{args:?}"); run_tool_envelope("create_repo", log, async move { let (resp, retries) = self - .dispatch(hive_sh4re::Request::CreateRepo { repo: args.repo }) + .dispatch(hive_agent_sock::Request::CreateRepo { repo: args.repo }) .await; let s = match resp { - Ok(hive_sh4re::Response::RepoCreated { + Ok(hive_agent_sock::Response::RepoCreated { full_name, clone_url, }) => format!("created repo {full_name} — clone: {clone_url}"), @@ -495,7 +493,7 @@ impl AgentServer { (None, Some(t)) => hive_sh4re::ReminderTiming::At { unix_timestamp: t }, }; let (resp, retries) = self - .dispatch(hive_sh4re::Request::Remind { + .dispatch(hive_agent_sock::Request::Remind { message: args.message, timing, file_path: args.file_path, @@ -549,7 +547,7 @@ impl AgentServer { let name = args.name.clone(); run_tool_envelope("restart", log, async move { let (resp, retries) = self - .dispatch(hive_sh4re::Request::Restart { name: args.name }) + .dispatch(hive_agent_sock::Request::Restart { name: args.name }) .await; annotate_retries( format_ack(resp, "restart", format!("restarted {name}")), @@ -571,7 +569,7 @@ impl AgentServer { let name = args.name.clone(); run_tool_envelope("kill", log, async move { let (resp, retries) = self - .dispatch(hive_sh4re::Request::Kill { name: args.name }) + .dispatch(hive_agent_sock::Request::Kill { name: args.name }) .await; annotate_retries(format_ack(resp, "kill", format!("killed {name}")), retries) }) @@ -592,7 +590,7 @@ impl AgentServer { let name = args.name.clone(); run_tool_envelope("update", log, async move { let (resp, retries) = self - .dispatch(hive_sh4re::Request::Update { name: args.name }) + .dispatch(hive_agent_sock::Request::Update { name: args.name }) .await; annotate_retries( format_ack(resp, "update", format!("updated {name}")), @@ -614,9 +612,11 @@ impl AgentServer { )] async fn list_containers(&self) -> String { run_tool_envelope("list_containers", String::new(), async move { - let (resp, retries) = self.dispatch(hive_sh4re::Request::ListDescendants).await; + let (resp, retries) = self + .dispatch(hive_agent_sock::Request::ListDescendants) + .await; let body = match resp { - Ok(hive_sh4re::Response::Containers { containers }) => { + Ok(hive_agent_sock::Response::Containers { containers }) => { if containers.is_empty() { "no descendant containers".to_owned() } else { @@ -659,7 +659,7 @@ impl AgentServer { let log = format!("{args:?}"); run_tool_envelope("get_host_journal", log, async move { let (resp, retries) = self - .dispatch(hive_sh4re::Request::GetHostJournal { + .dispatch(hive_agent_sock::Request::GetHostJournal { unit: args.unit, container: args.container, lines: args.lines, @@ -670,7 +670,7 @@ impl AgentServer { }) .await; let result = match resp { - Ok(hive_sh4re::Response::HostJournal { content }) => content, + Ok(hive_agent_sock::Response::HostJournal { content }) => content, other => reply_err(other, "get_host_journal"), }; annotate_retries(result, retries) @@ -699,7 +699,7 @@ impl AgentServer { let name = args.name.clone(); run_tool_envelope("request_init_config", log, async move { let (resp, retries) = self - .dispatch(hive_sh4re::Request::RequestInitConfig { + .dispatch(hive_agent_sock::Request::RequestInitConfig { name: args.name, description: args.description, }) @@ -727,7 +727,7 @@ impl AgentServer { let name = args.name.clone(); run_tool_envelope("start", log, async move { let (resp, retries) = self - .dispatch(hive_sh4re::Request::Start { name: args.name }) + .dispatch(hive_agent_sock::Request::Start { name: args.name }) .await; annotate_retries( format_ack(resp, "start", format!("started {name}")), @@ -750,13 +750,13 @@ impl AgentServer { run_tool_envelope("get_logs", log, async move { let lines = args.lines.map(|n| n.min(500)); let (resp, retries) = self - .dispatch(hive_sh4re::Request::GetLogs { + .dispatch(hive_agent_sock::Request::GetLogs { agent: agent.clone(), lines, }) .await; let s = match resp { - Ok(hive_sh4re::Response::Logs { content }) => { + Ok(hive_agent_sock::Response::Logs { content }) => { if content.is_empty() { format!("(no journal output for {agent})") } else { @@ -790,7 +790,7 @@ impl AgentServer { args.inputs.join(", ") }; let (resp, retries) = self - .dispatch(hive_sh4re::Request::RequestUpdateMetaInputs { + .dispatch(hive_agent_sock::Request::RequestUpdateMetaInputs { inputs: args.inputs, description: args.description, }) @@ -829,7 +829,7 @@ impl AgentServer { run_tool_envelope("request_schedule_prompt", log, async move { let target_count = args.targets.len(); let (resp, retries) = self - .dispatch(hive_sh4re::Request::RequestSchedulePrompt( + .dispatch(hive_agent_sock::Request::RequestSchedulePrompt( hive_sh4re::SchedulePromptPayload { targets: args.targets, body: args.body, @@ -865,7 +865,7 @@ impl AgentServer { run_tool_envelope("fire_schedule_now", log, async move { let id = args.id; let (resp, retries) = self - .dispatch(hive_sh4re::Request::FireScheduleNow { id }) + .dispatch(hive_agent_sock::Request::FireScheduleNow { id }) .await; annotate_retries( format_ack(resp, "fire_schedule_now", format!("fired #{id} now")), @@ -888,7 +888,7 @@ impl AgentServer { run_tool_envelope("cancel_schedule", log, async move { let id = args.id; let (resp, retries) = self - .dispatch(hive_sh4re::Request::CancelSchedule { + .dispatch(hive_agent_sock::Request::CancelSchedule { id: args.id, targets: args.targets, }) @@ -920,7 +920,7 @@ impl AgentServer { run_tool_envelope("edit_schedule", log, async move { let id = args.id; let (resp, retries) = self - .dispatch(hive_sh4re::Request::EditSchedule { + .dispatch(hive_agent_sock::Request::EditSchedule { id: args.id, body: args.body, description: args.description.map(Some), @@ -947,9 +947,9 @@ impl AgentServer { )] async fn list_schedules(&self) -> String { run_tool_envelope("list_schedules", String::new(), async move { - let (resp, retries) = self.dispatch(hive_sh4re::Request::ListSchedules).await; + let (resp, retries) = self.dispatch(hive_agent_sock::Request::ListSchedules).await; let body = match resp { - Ok(hive_sh4re::Response::Schedules { schedules }) => { + Ok(hive_agent_sock::Response::Schedules { schedules }) => { serde_json::to_string(&schedules) .unwrap_or_else(|e| format!("list_schedules: serialise: {e:#}")) } diff --git a/hive-agent-mcp/src/mcp/render.rs b/hive-agent-mcp/src/mcp/render.rs index 28f55bca..ccce7934 100644 --- a/hive-agent-mcp/src/mcp/render.rs +++ b/hive-agent-mcp/src/mcp/render.rs @@ -10,9 +10,12 @@ /// transport error: …"`. Handlers match their own happy-path variant and route /// everything else here via a catch-all arm (`other => reply_err(other, tool)`), /// so the triplet lives in exactly one place. -pub(super) fn reply_err(resp: Result, tool: &str) -> String { +pub(super) fn reply_err( + resp: Result, + tool: &str, +) -> String { match resp { - Ok(hive_sh4re::Response::Err { message }) => format!("{tool} failed: {message}"), + Ok(hive_agent_sock::Response::Err { message }) => format!("{tool} failed: {message}"), Ok(other) => format!("{tool} unexpected response: {other:?}"), Err(e) => format!("{tool} transport error: {e:#}"), } @@ -23,12 +26,12 @@ pub(super) fn reply_err(resp: Result, tool: /// behavior. #[must_use] pub fn format_ack( - resp: Result, + resp: Result, tool: &str, ok_msg: String, ) -> String { match resp { - Ok(hive_sh4re::Response::Ok) => ok_msg, + Ok(hive_agent_sock::Response::Ok) => ok_msg, other => reply_err(other, tool), } } @@ -44,9 +47,9 @@ pub fn format_ack( /// so the model can tell where one ends and the next begins; /// per-message redelivery banners included. #[must_use] -pub fn format_recv(resp: Result, waited: bool) -> String { +pub fn format_recv(resp: Result, waited: bool) -> String { match resp { - Ok(hive_sh4re::Response::Messages { + Ok(hive_agent_sock::Response::Messages { messages, remaining, }) => render_recv_messages(&messages, remaining, waited), @@ -57,7 +60,7 @@ pub fn format_recv(resp: Result, waited: bo // stop unmissably tells claude to flush + end. `remaining` is forced // to 0 — the inbox is fenced, so a "N more pending" hint would be // misleading. - Ok(hive_sh4re::Response::GracefulStop) => { + Ok(hive_agent_sock::Response::GracefulStop) => { render_recv_messages(&[graceful_stop_message()], 0, waited) } other => reply_err(other, "recv"), @@ -354,9 +357,9 @@ pub(super) fn loose_end_kind_label(kind: hive_sh4re::CancelLooseEndKind) -> &'st /// `running: no` line tells the caller WHY. See /// `docs/turn-loop/mcp.md::Core tools` (`get_agent_meta`). #[must_use] -pub fn format_agent_meta(resp: Result) -> String { +pub fn format_agent_meta(resp: Result) -> String { match resp { - Ok(hive_sh4re::Response::AgentMeta { + Ok(hive_agent_sock::Response::AgentMeta { name, running, hyperhive_rev, @@ -477,7 +480,7 @@ mod tests { #[test] fn empty_recv_after_wait_appends_idle_hint() { let out = format_recv( - Ok(hive_sh4re::Response::Messages { + Ok(hive_agent_sock::Response::Messages { messages: vec![], remaining: 0, }), @@ -490,7 +493,7 @@ mod tests { #[test] fn empty_recv_without_wait_has_no_hint() { let out = format_recv( - Ok(hive_sh4re::Response::Messages { + Ok(hive_agent_sock::Response::Messages { messages: vec![], remaining: 0, }), @@ -502,7 +505,7 @@ mod tests { #[test] fn single_recv_with_remaining_appends_pending_hint() { let out = format_recv( - Ok(hive_sh4re::Response::Messages { + Ok(hive_agent_sock::Response::Messages { messages: vec![msg(7, "alice", "hi")], remaining: 3, }), @@ -516,7 +519,7 @@ mod tests { #[test] fn single_recv_no_remaining_has_no_pending_hint() { let out = format_recv( - Ok(hive_sh4re::Response::Messages { + Ok(hive_agent_sock::Response::Messages { messages: vec![msg(7, "alice", "hi")], remaining: 0, }), @@ -528,7 +531,7 @@ mod tests { #[test] fn batch_recv_with_remaining_appends_pending_hint_once() { let out = format_recv( - Ok(hive_sh4re::Response::Messages { + Ok(hive_agent_sock::Response::Messages { messages: vec![msg(7, "alice", "hi"), msg(8, "bob", "yo")], remaining: 9, }), diff --git a/hive-agent-sock/Cargo.toml b/hive-agent-sock/Cargo.toml new file mode 100644 index 00000000..57a61004 --- /dev/null +++ b/hive-agent-sock/Cargo.toml @@ -0,0 +1,11 @@ +[package] +name = "hive-agent-sock" +edition.workspace = true +version.workspace = true + +[lints] +workspace = true + +[dependencies] +hive-sh4re.workspace = true +serde.workspace = true diff --git a/hive-agent-sock/src/lib.rs b/hive-agent-sock/src/lib.rs new file mode 100644 index 00000000..e3ef2093 --- /dev/null +++ b/hive-agent-sock/src/lib.rs @@ -0,0 +1,399 @@ +//! Per-agent + manager socket wire types (`/run/hive/mcp.sock`). +//! +//! The unified `Request` / `Response` protocol spoken between an agent's +//! in-container harness (and the manager) and the `hive-c0re` daemon over the +//! per-agent mcp socket. Re-homed out of `hive-sh4re` so this one socket owns +//! its own protocol crate, mirroring `hive-host-sock` / `hive-priv-sock`. The +//! shared payload types it references (`Message`, `LooseEnd`, `Approval`, …) +//! stay in `hive-sh4re`, which this crate depends on. + +use hive_sh4re::{ + CancelLooseEndKind, ContainerInfo, DeliveredMessage, InboxRow, JournalPriority, LooseEnd, + MatrixIdentity, ReminderStats, ReminderTiming, SchedulePromptPayload, WireSchedule, +}; +use serde::{Deserialize, Serialize}; + +/// serde `default` helper for `Response::AgentMeta::running` (absent = true). +fn default_true() -> bool { + true +} + +/// Unified request enum for both agent and manager sockets. The agent's +/// identity is the socket it arrived on. Privileged variants are marked +/// `*(privileged)*` — an agent socket returns `Err` for them server-side. +#[derive(Debug, Clone, Serialize, Deserialize)] +#[serde(tag = "cmd", rename_all = "snake_case")] +pub enum Request { + /// Send a message to another agent. + Send { + to: String, + body: String, + /// Optional id of the message being replied to. Stored in the + /// broker DB and returned on `Recv` so the dashboard can render + /// threads. Ignored if the id is unknown or out of retention. + #[serde(default, skip_serializing_if = "Option::is_none")] + in_reply_to: Option, + }, + /// Pop pending messages from this agent's inbox. + /// Delivery + ack cycle: see + /// `docs/conventions.md::Broker delivery + ack cycle`. + Recv { + #[serde(default)] + wait_seconds: Option, + #[serde(default)] + max: Option, + }, + /// Non-mutating: how many pending messages are addressed to me? + /// Used by the harness to render a status line after each tool call. + Status, + /// Operator-injected message TO this agent (from this agent's own web + /// UI). Recipient is implicit — `from` is `"operator"`. Effectively the + /// per-agent equivalent of the old dashboard T4LK form, but scoped to + /// the agent whose page the operator is on. + OperatorMsg { body: String }, + /// Wake-up event injected from inside the container. Recipient is + /// implicit (this agent); `from` is caller-chosen. See + /// `docs/conventions.md::Wake injection` for the trust model and + /// typical callers. The wake is persisted in the sqlite broker + /// like any other message — the agent can ack it via `AckUntil`. + Wake { from: String, body: String }, + /// Last `limit` messages addressed to this agent, newest-first. + /// Non-mutating — pulls from the broker without delivering. The + /// per-agent web UI uses this to render its own inbox section. + Recent { limit: u64 }, + /// Surface a question to either the operator or another agent. + /// Routing + shape: see + /// `docs/conventions.md::Question routing (Ask / Answer)`. + Ask { + question: String, + #[serde(default)] + options: Vec, + #[serde(default)] + multi: bool, + #[serde(default)] + ttl_seconds: Option, + #[serde(default)] + to: Option, + }, + /// Answer a question previously routed to this agent via + /// `HelperEvent::QuestionAsked`. Authorised callers + threading + /// back via `HelperEvent::QuestionAnswered`: see + /// `docs/conventions.md::Question routing (Ask / Answer)`. + Answer { id: i64, answer: String }, + /// Schedule a reminder message to be delivered to this agent at a + /// future time. The reminder lands in the agent's inbox as an auto-sent + /// message from `"reminder"`. Use for agent follow-ups (e.g. check task + /// status, retry failed operation). Message length is limited; pass + /// `file_path` to store in a file and get a path-reference message + /// instead. + Remind { + message: String, + timing: ReminderTiming, + #[serde(default)] + file_path: Option, + }, + /// Loose-ends view. On the agent socket: `None` = self; direct + /// children are always accessible; non-children require the + /// `query_agent_state` capability — rejected with an error otherwise; + /// `"*"` is always rejected (use the manager socket). On the manager + /// socket: `None` = manager self, `"*"` = hive-wide, any name = + /// that agent. See `docs/conventions.md::Loose-ends wire shape`. + GetLooseEnds { + #[serde(default, skip_serializing_if = "Option::is_none")] + agent: Option, + }, + /// Upsert a *todo* (loose-ends v2) from an in-container + /// subsystem (matrix / forge / bash). `subsystem` is the producer + /// marker; `key` is the optional subsystem-specific dedup key (a + /// matrix room id, a bash task id). Re-pushing an identical keyed + /// todo is a no-op; a new-or-changed one coalesces a wake to the + /// agent. Keyless todos always insert as one-offs. + UpsertTodo { + subsystem: String, + #[serde(default, skip_serializing_if = "Option::is_none")] + key: Option, + summary: String, + #[serde(default, skip_serializing_if = "Option::is_none")] + source: Option, + }, + /// Clear producer-resolved todo(s) by `(subsystem, key)`. `key = + /// Some(k)` clears the one keyed row; `key = None` clears **all** of + /// the subsystem's keyless todos (rows with no key can't be told + /// apart — clear a specific one via `MarkTodoDone` by id). `all = + /// true` wipes the producer's whole set (cancel-and-recreate on + /// daemon restart). + ClearTodo { + subsystem: String, + #[serde(default, skip_serializing_if = "Option::is_none")] + key: Option, + #[serde(default)] + all: bool, + }, + /// List todos, optionally filtered to one `subsystem` (a producer + /// enumerating its own set). `None` = all. + ListTodos { + #[serde(default, skip_serializing_if = "Option::is_none")] + subsystem: Option, + }, + /// The agent marks one of its own todos done, by id. + MarkTodoDone { id: i64 }, + /// Count of pending (un-delivered) reminders. On the agent socket: + /// same target rules as `GetLooseEnds` (self/children free; + /// non-children require `query_agent_state`; `"*"` rejected). + /// On the manager socket: `None` = self, any name = that agent. + /// Used by the harness's per-turn stats sink. + CountPendingReminders { + #[serde(default, skip_serializing_if = "Option::is_none")] + agent: Option, + }, + /// Reminder statistics: counts of scheduled, delivered, and pending + /// reminders over a time window. `since_secs` filters to reminders + /// created in the last N seconds (0 = all). On the agent socket: + /// same target rules as `GetLooseEnds` (self/children free; + /// non-children require `query_agent_state`; `"*"` rejected). + /// On the manager socket: `None` = self, any name = that agent. + ReminderRollup { + /// Only count reminders created in the last N seconds from now. + /// Pass 0 to include all reminders. + #[serde(default)] + since_secs: u64, + /// Whose reminders to roll up. `None` = the caller's own. + /// `Some("")` = that agent's (requires `query_agent_state` + /// capability on the agent socket; always available on the manager + /// socket). + #[serde(default, skip_serializing_if = "Option::is_none")] + agent: Option, + }, + /// Set a free-text status string visible on the dashboard. The harness + /// writes `{state_dir}/hyperhive-status` locally before sending this + /// request; hive-c0re just triggers a dashboard rescan on receipt. + /// Pass an empty string to clear the status. + SetStatus { text: String }, + /// Fetch identity + status for an agent. `name = None` = + /// self-introspection; `Some()` = target query. See + /// `docs/conventions.md::Agent metadata`. + GetAgentMeta { + #[serde(default, skip_serializing_if = "Option::is_none")] + name: Option, + }, + /// Cancel an open thread the agent owns. Authorisation + + /// per-kind semantics in + /// `docs/conventions.md::Loose-ends wire shape`. + CancelLooseEnd { kind: CancelLooseEndKind, id: i64 }, + /// Create a git repo *through hive-c0re*. Agents can't create + /// repos with their own forge token (`max_repo_creation = 0`); this is + /// the sanctioned path. hive-c0re creates `repo` in the c0re-owned + /// `agents` org, adds the calling agent as a write collaborator (not + /// owner), and applies operator-team branch protection so the author + /// can't merge its own PRs. Returns the new repo's full name. + CreateRepo { repo: String }, + /// Mark every message popped since the last `AckTurn` as handled. + /// Harness↔broker pairing fired after `TurnOutcome::Ok`. See + /// `docs/conventions.md::Broker delivery + ack cycle`. + AckTurn, + /// Mark every inbox message with broker row id `<= up_to` as + /// handled (`acked_at` set), whether still pending or already + /// delivered. The agent-facing bulk-triage escape hatch for a + /// redelivered / accumulated backlog: instead of popping and + /// re-reading dozens of already-handled messages one turn at a + /// time, the agent acks everything up to the id it has seen. + /// Recipient-scoped — an agent can only ack its own rows. See + /// `docs/conventions.md::Broker delivery + ack cycle`. + AckUntil { up_to: i64 }, + /// Requeue every popped-but-unacked message back into the inbox. + /// Harness fires this once at boot to recover from + /// crashed-mid-turn sessions. See + /// `docs/conventions.md::Broker delivery + ack cycle`. + RequeueInflight, + /// Harness → c0re: "I saw the `GracefulStop` signal, ran my + /// stop-checkpoint turn (durable `/state` flushed) and am exiting my + /// serve loop now." Lets the `GracefulStop` orchestration stop the + /// container immediately instead of waiting out its timeout fallback. + GracefulStopComplete, + /// *(capability-gated: `read_host_journal`)* Fetch recent lines + /// from the host journal. Filters are all optional; omitting all + /// returns the last `lines` entries from the global journal. + GetHostJournal { + /// Filter to a specific systemd unit (e.g. `hive-c0re.service`). + #[serde(default, skip_serializing_if = "Option::is_none")] + unit: Option, + /// Machine name to pass to journalctl `-M` verbatim (e.g. `h-iris`). + /// The caller is responsible for the correct nspawn machine name. + #[serde(default, skip_serializing_if = "Option::is_none")] + container: Option, + /// Number of journal lines to return (default 30, max 100). + #[serde(default, skip_serializing_if = "Option::is_none")] + lines: Option, + /// Minimum syslog priority level. + #[serde(default, skip_serializing_if = "Option::is_none")] + priority: Option, + /// Regex to match against log message fields (journalctl `--grep`). + #[serde(default, skip_serializing_if = "Option::is_none")] + grep: Option, + /// Show entries on or newer than this timestamp (journalctl `--since`). + /// ISO 8601 or journalctl-accepted relative strings (e.g. `"-1h"`). + #[serde(default, skip_serializing_if = "Option::is_none")] + since: Option, + /// Show entries on or older than this timestamp (journalctl `--until`). + #[serde(default, skip_serializing_if = "Option::is_none")] + until: Option, + }, + + // ---- privileged (manager socket only for now) --------------------------- + /// *(privileged)* Initialise a brand-new agent's proposed config repo + /// and queue an approval for the operator to review. + RequestInitConfig { + name: String, + #[serde(default, skip_serializing_if = "Option::is_none")] + description: Option, + }, + /// *(privileged)* Stop a sub-agent (graceful). + Kill { name: String }, + /// *(privileged)* Start a previously-stopped sub-agent container. + Start { name: String }, + /// *(privileged)* Restart a sub-agent container (stop + start). + Restart { name: String }, + /// *(privileged)* Rebuild a sub-agent against the current hyperhive + /// flake + agent.nix. No approval required. + Update { name: String }, + /// *(privileged)* Fetch recent journal lines for a sub-agent container. + GetLogs { + agent: String, + #[serde(default)] + lines: Option, + }, + /// *(privileged)* Queue an approval to run `nix flake update [inputs...]`. + RequestUpdateMetaInputs { + #[serde(default)] + inputs: Vec, + #[serde(default, skip_serializing_if = "Option::is_none")] + description: Option, + }, + /// *(privileged)* Queue an approval to add a scheduled prompt. + RequestSchedulePrompt(SchedulePromptPayload), + /// *(privileged)* Cancel a scheduled prompt. + CancelSchedule { + id: i64, + #[serde(default, skip_serializing_if = "Option::is_none")] + targets: Option>, + }, + /// *(privileged)* List every schedule in the queue. + ListSchedules, + /// List all containers that are topological descendants of the calling + /// agent (direct children + their subtrees). Scoped to the caller's + /// subtree; gated by the `lifecycle` tool group. The result includes all + /// known descendants regardless of whether the container is currently + /// running — use `running` to distinguish. + ListDescendants, + /// *(privileged)* Fire a scheduled prompt out of band immediately. + FireScheduleNow { id: i64 }, + /// *(privileged)* Edit an existing schedule's mutable fields. + EditSchedule { + id: i64, + #[serde(default, skip_serializing_if = "Option::is_none")] + body: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + description: Option>, + #[serde(default, skip_serializing_if = "Option::is_none")] + interval_seconds: Option>, + #[serde(default, skip_serializing_if = "Option::is_none")] + next_fire_at_unix: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + targets_add: Option>, + #[serde(default, skip_serializing_if = "Option::is_none")] + targets_remove: Option>, + }, +} + +/// Unified response enum for both agent and manager sockets. Privileged +/// variants (`Logs`, `Schedules`) are never returned on agent sockets. +#[derive(Debug, Clone, Serialize, Deserialize)] +#[serde(tag = "kind", rename_all = "snake_case")] +pub enum Response { + /// `Send` succeeded. + Ok, + /// Either `Send` failed or `Recv` errored. + Err { message: String }, + /// `Recv` result: zero or more messages, FIFO-ordered, never + /// longer than the `max` the caller passed. Empty vec = nothing + /// pending (the "(empty)" path for the formatter). Per-row `id` + + /// `redelivered` carry the broker's row id (tracked by the harness + /// for `AckTurn`, and surfaced to claude as a `[msg #]` marker + /// so `AckUntil` has something to reference) and the "previously + /// popped, not acked" flag — see `DeliveredMessage` for details. + /// `remaining` is the inbox depth *after* this batch was popped — + /// how many still-pending messages the caller could drain next. The + /// harness surfaces it to claude ("N more pending") so an in-turn + /// `recv` learns whether the inbox is drained, mirroring the count + /// the wake prompt already carries. + Messages { + messages: Vec, + remaining: u64, + }, + /// `Status` result: how many pending messages are in this agent's inbox. + Status { unread: u64 }, + /// `AckUntil` result: how many rows were newly marked handled. + Acked { count: u64 }, + /// `Recent` result: newest-first inbox rows. + Recent { rows: Vec }, + /// `Ask` result: the queued question id. The answer lands later + /// as `HelperEvent::QuestionAnswered` in this agent's inbox. + QuestionQueued { id: i64 }, + /// `GetLooseEnds` result: list of loose ends pending against + /// this agent. Ordered newest-first within each kind. + LooseEnds { loose_ends: Vec }, + /// `CountPendingReminders` result. + PendingRemindersCount { count: u64 }, + /// `ReminderRollup` result: reminder activity stats for the agent. + ReminderRollup(ReminderStats), + /// `GetAgentMeta` result. Per-field semantics + serde defaults + /// live in `docs/conventions.md::Agent metadata`. + AgentMeta { + name: String, + #[serde(default = "default_true")] + running: bool, + #[serde(default, skip_serializing_if = "Option::is_none")] + hyperhive_rev: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + status_text: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + status_set_at: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + hive_name: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + swarm_name: Option, + /// Matrix identities this agent can act as (one per configured + + /// live account). Empty for agents with no matrix provisioning. + #[serde(default, skip_serializing_if = "Vec::is_empty")] + matrix_accounts: Vec, + }, + /// `GetLogs` result: journal lines for the requested container. + /// Returned on the manager socket only. + Logs { content: String }, + /// `GetHostJournal` result: host journal lines matching the + /// requested filters. Returned on the agent socket when the agent + /// holds the `read_host_journal` capability. + HostJournal { content: String }, + /// `ListSchedules` result. Snapshot of every schedule. + /// Returned on the manager socket only. + Schedules { schedules: Vec }, + /// `CreateRepo` result: the new repo's full name (`agents/`) + /// and clone URL, so the agent can immediately `git clone` it. + RepoCreated { + full_name: String, + clone_url: String, + }, + /// `ListDescendants` result: all descendant containers, with running + /// status. Ordered by topology depth (parents before children), then + /// alphabetically within each depth tier. + Containers { containers: Vec }, + /// `Recv` result when a graceful stop is pending for this agent + /// (set by hive-c0re's `GracefulStop` orchestration). Returned in + /// place of `Messages` — it doubles as the inbound fence: the harness + /// stops consuming normal inbox messages and instead runs one + /// stop-checkpoint turn (flush durable `/state`), then reports + /// `GracefulStopComplete` and exits its serve loop so the container + /// can be stopped cleanly. New sends keep queueing in the broker for + /// the agent's next start. + GracefulStop, +} diff --git a/hive-agent-wake/Cargo.toml b/hive-agent-wake/Cargo.toml index 2d1fd492..c672f09d 100644 --- a/hive-agent-wake/Cargo.toml +++ b/hive-agent-wake/Cargo.toml @@ -13,7 +13,7 @@ workspace = true [dependencies] anyhow.workspace = true clap.workspace = true -hive-sh4re.workspace = true +hive-agent-sock.workspace = true serde.workspace = true serde_json.workspace = true tokio.workspace = true diff --git a/hive-agent-wake/src/main.rs b/hive-agent-wake/src/main.rs index 03cf7d3a..d21cfca9 100644 --- a/hive-agent-wake/src/main.rs +++ b/hive-agent-wake/src/main.rs @@ -13,7 +13,7 @@ use std::path::PathBuf; use anyhow::Result; use clap::Parser; -use hive_sh4re::{AgentRequest, AgentResponse}; +use hive_agent_sock::{Request, Response}; /// Per-agent MCP socket, bind-mounted from the host into every container. const DEFAULT_SOCKET: &str = "/run/hive/mcp.sock"; @@ -50,17 +50,17 @@ async fn main() -> Result<()> { } else { cli.body }; - let resp: AgentResponse = client::request( + let resp: Response = client::request( &cli.socket, - &AgentRequest::Wake { + &Request::Wake { from: cli.from, body, }, ) .await?; match resp { - AgentResponse::Ok => Ok(()), - AgentResponse::Err { message } => anyhow::bail!("wake: {message}"), + Response::Ok => Ok(()), + Response::Err { message } => anyhow::bail!("wake: {message}"), other => anyhow::bail!("wake: unexpected response {other:?}"), } } diff --git a/hive-agent/Cargo.toml b/hive-agent/Cargo.toml index a9af635c..f16a7e9a 100644 --- a/hive-agent/Cargo.toml +++ b/hive-agent/Cargo.toml @@ -19,6 +19,7 @@ time.workspace = true futures-util = "0.3" clap.workspace = true hive-claude.workspace = true +hive-agent-sock.workspace = true hive-sh4re.workspace = true rmcp.workspace = true rusqlite.workspace = true diff --git a/hive-agent/src/forge_notify.rs b/hive-agent/src/forge_notify.rs index d6c1de32..e701e4b7 100644 --- a/hive-agent/src/forge_notify.rs +++ b/hive-agent/src/forge_notify.rs @@ -1083,11 +1083,11 @@ async fn poll_once( continue; }; - let req = hive_sh4re::Request::Wake { + let req = hive_agent_sock::Request::Wake { from: "forge".to_owned(), body, }; - let deliver_result = crate::client::request::<_, hive_sh4re::Response>(socket, &req) + let deliver_result = crate::client::request::<_, hive_agent_sock::Response>(socket, &req) .await .map(|_| ()); match deliver_result { diff --git a/hive-agent/src/main.rs b/hive-agent/src/main.rs index a79292c1..7e0d98c4 100644 --- a/hive-agent/src/main.rs +++ b/hive-agent/src/main.rs @@ -42,7 +42,8 @@ use crate::login::LoginState; use crate::turn_stats::TurnStats; use anyhow::Result; use clap::Parser; -use hive_sh4re::{AgentRequest, AgentResponse, HelperEvent, SYSTEM_SENDER}; +use hive_agent_sock::{Request, Response}; +use hive_sh4re::{HelperEvent, SYSTEM_SENDER}; #[derive(Parser)] #[command(name = "hive-agent", about = "hyperhive harness serve loop")] @@ -190,9 +191,8 @@ fn graceful_stop_message() -> hive_sh4re::DeliveredMessage { // ---------- surface trait ---------- -/// What a `Recv` long-poll returned. Decoupled from the per-role -/// Response enum so `serve_loop` can pattern-match without seeing -/// either `AgentResponse` or `ManagerResponse` directly. +/// What a `Recv` long-poll returned. Decoupled from the `Response` +/// enum so `serve_loop` can pattern-match without seeing it directly. enum RecvOutcome { /// Long-poll returned at least one message; first one is detached. Message(hive_sh4re::DeliveredMessage), @@ -212,7 +212,7 @@ enum RecvOutcome { /// Wire surface abstraction. `AgentSurface` is the only impl — the trait /// exists to keep the turn loop generic and testable. Every function that /// talks to the broker goes through this so there are zero hard-coded -/// `AgentRequest` / `AgentResponse` references in the turn loop itself. +/// `Request` / `Response` references in the turn loop itself. trait Surface { /// Ack the in-flight turn. Logs warnings on transport/broker /// errors but never propagates — turn loop continues either way. @@ -251,17 +251,17 @@ trait Surface { // ---------- AgentSurface ---------- /// Zero-sized type tag for the agent wire surface. -/// Talks `AgentRequest` / `AgentResponse`. +/// Talks `Request` / `Response`. struct AgentSurface; /// Issue an `Ok`-expecting fire-and-forget broker request, logging any /// rejection / unexpected response / transport error under `label`. Shared by /// the `Surface` methods that don't need the reply (`ack_turn`, /// `requeue_inflight`, `graceful_stop_complete`). -async fn fire_and_forget(socket: &Path, req: AgentRequest, label: &str) { - match client::request::<_, AgentResponse>(socket, &req).await { - Ok(AgentResponse::Ok) => {} - Ok(AgentResponse::Err { message }) => { +async fn fire_and_forget(socket: &Path, req: Request, label: &str) { + match client::request::<_, Response>(socket, &req).await { + Ok(Response::Ok) => {} + Ok(Response::Err { message }) => { tracing::warn!(%message, "{label} rejected by broker"); } Ok(other) => tracing::warn!(?other, "{label} unexpected response"), @@ -271,55 +271,53 @@ async fn fire_and_forget(socket: &Path, req: AgentRequest, label: &str) { impl Surface for AgentSurface { async fn ack_turn(socket: &Path) { - fire_and_forget(socket, AgentRequest::AckTurn, "ack_turn").await; + fire_and_forget(socket, Request::AckTurn, "ack_turn").await; } async fn requeue_inflight(socket: &Path) { - fire_and_forget(socket, AgentRequest::RequeueInflight, "requeue_inflight").await; + fire_and_forget(socket, Request::RequeueInflight, "requeue_inflight").await; } async fn graceful_stop_complete(socket: &Path) { fire_and_forget( socket, - AgentRequest::GracefulStopComplete, + Request::GracefulStopComplete, "graceful_stop_complete", ) .await; } async fn inbox_unread(socket: &Path) -> u64 { - match client::request::<_, AgentResponse>(socket, &AgentRequest::Status).await { - Ok(AgentResponse::Status { unread }) => unread, + match client::request::<_, Response>(socket, &Request::Status).await { + Ok(Response::Status { unread }) => unread, _ => 0, } } async fn post_turn_counts(socket: &Path) -> (Option, Option) { - let threads = match client::request::<_, AgentResponse>( + 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 reminders = match client::request::<_, Response>( socket, - &AgentRequest::GetLooseEnds { agent: None }, + &Request::CountPendingReminders { agent: None }, ) .await { - Ok(AgentResponse::LooseEnds { loose_ends }) => u64::try_from(loose_ends.len()).ok(), - _ => None, - }; - let reminders = match client::request::<_, AgentResponse>( - socket, - &AgentRequest::CountPendingReminders { agent: None }, - ) - .await - { - Ok(AgentResponse::PendingRemindersCount { count }) => Some(count), + Ok(Response::PendingRemindersCount { count }) => Some(count), _ => None, }; (threads, reminders) } async fn send_to_parent(socket: &Path, body: String) { - let res = client::request::<_, AgentResponse>( + let res = client::request::<_, Response>( socket, - &AgentRequest::Send { + &Request::Send { to: hive_sh4re::PARENT_RECIPIENT.into(), body, in_reply_to: None, @@ -332,22 +330,22 @@ impl Surface for AgentSurface { } async fn recv_next(socket: &Path) -> RecvOutcome { - let recv: Result = client::request( + let recv: Result = client::request( socket, - &AgentRequest::Recv { + &Request::Recv { wait_seconds: Some(180), max: None, }, ) .await; match recv { - Ok(AgentResponse::Messages { messages, .. }) if !messages.is_empty() => { + Ok(Response::Messages { messages, .. }) if !messages.is_empty() => { let first = messages.into_iter().next().expect("checked non-empty"); RecvOutcome::Message(first) } - Ok(AgentResponse::Messages { .. }) => RecvOutcome::Empty, - Ok(AgentResponse::GracefulStop) => RecvOutcome::GracefulStop, - Ok(AgentResponse::Err { message }) => { + Ok(Response::Messages { .. }) => RecvOutcome::Empty, + Ok(Response::GracefulStop) => RecvOutcome::GracefulStop, + Ok(Response::Err { message }) => { tracing::warn!(%message, "recv error"); RecvOutcome::TransportError } diff --git a/hive-agent/src/web_ui/actions.rs b/hive-agent/src/web_ui/actions.rs index c2fb1f71..7e6de725 100644 --- a/hive-agent/src/web_ui/actions.rs +++ b/hive-agent/src/web_ui/actions.rs @@ -23,15 +23,20 @@ pub(super) async fn post_send( if body.is_empty() { return error_response(StatusCode::BAD_REQUEST, "send: `body` required"); } - match super::broker_request(&state.socket, &hive_sh4re::Request::OperatorMsg { body }).await { + match super::broker_request( + &state.socket, + &hive_agent_sock::Request::OperatorMsg { body }, + ) + .await + { // 200 instead of 303 → the client doesn't refetch /api/state. // The operator message becomes a broker `Sent` (already shown // server-side in the dashboard); on the agent side, the // resulting `TurnStart` SSE event drives the terminal + the // inbox row gets consumed by the time `TurnEnd` fires the // existing turn-end refresh. - Ok(hive_sh4re::Response::Ok) => (axum::http::StatusCode::OK, "ok").into_response(), - Ok(hive_sh4re::Response::Err { message }) => error_response( + Ok(hive_agent_sock::Response::Ok) => (axum::http::StatusCode::OK, "ok").into_response(), + Ok(hive_agent_sock::Response::Err { message }) => error_response( StatusCode::INTERNAL_SERVER_ERROR, &format!("send failed: {message}"), ), diff --git a/hive-agent/src/web_ui/mod.rs b/hive-agent/src/web_ui/mod.rs index 7eb137d1..75c0768d 100644 --- a/hive-agent/src/web_ui/mod.rs +++ b/hive-agent/src/web_ui/mod.rs @@ -302,11 +302,11 @@ enum BrokerError { /// web-UI handler that talks to the broker goes through it. async fn broker_request( socket: &Path, - req: &hive_sh4re::Request, -) -> std::result::Result { + req: &hive_agent_sock::Request, +) -> std::result::Result { match tokio::time::timeout( SOCKET_FETCH_TIMEOUT, - crate::client::request::<_, hive_sh4re::Response>(socket, req), + crate::client::request::<_, hive_agent_sock::Response>(socket, req), ) .await { diff --git a/hive-agent/src/web_ui/state.rs b/hive-agent/src/web_ui/state.rs index 13d1c11f..b475c619 100644 --- a/hive-agent/src/web_ui/state.rs +++ b/hive-agent/src/web_ui/state.rs @@ -366,8 +366,8 @@ async fn recent_inbox(socket: &std::path::Path) -> Vec { const LIMIT: u64 = 30; // Deadline-bounded (via `broker_request`): `/api/state` must render even // when hive-c0re is busy — an empty inbox section beats a hung snapshot. - match super::broker_request(socket, &hive_sh4re::Request::Recent { limit: LIMIT }).await { - Ok(hive_sh4re::Response::Recent { rows }) => rows, + match super::broker_request(socket, &hive_agent_sock::Request::Recent { limit: LIMIT }).await { + Ok(hive_agent_sock::Response::Recent { rows }) => rows, _ => Vec::new(), } } diff --git a/hive-agent/src/web_ui/stats.rs b/hive-agent/src/web_ui/stats.rs index bd5982e4..bbb3eb8a 100644 --- a/hive-agent/src/web_ui/stats.rs +++ b/hive-agent/src/web_ui/stats.rs @@ -35,14 +35,14 @@ async fn fetch_reminder_stats( ) -> Option { match super::broker_request( socket, - &hive_sh4re::Request::ReminderRollup { + &hive_agent_sock::Request::ReminderRollup { since_secs: window_secs, agent: None, }, ) .await { - Ok(hive_sh4re::Response::ReminderRollup(stats)) => Some(stats), + Ok(hive_agent_sock::Response::ReminderRollup(stats)) => Some(stats), _ => None, } } @@ -57,14 +57,14 @@ async fn fetch_reminder_stats( pub(super) async fn api_loose_ends(State(state): State) -> Response { match super::broker_request( &state.socket, - &hive_sh4re::Request::GetLooseEnds { agent: None }, + &hive_agent_sock::Request::GetLooseEnds { agent: None }, ) .await { - Ok(hive_sh4re::Response::LooseEnds { loose_ends }) => { + Ok(hive_agent_sock::Response::LooseEnds { loose_ends }) => { axum::Json(serde_json::json!({ "loose_ends": loose_ends })).into_response() } - Ok(hive_sh4re::Response::Err { message }) => error_response( + Ok(hive_agent_sock::Response::Err { message }) => error_response( StatusCode::INTERNAL_SERVER_ERROR, &format!("get_loose_ends: {message}"), ), diff --git a/hive-bash-mcp/Cargo.toml b/hive-bash-mcp/Cargo.toml index 68d470a0..06269257 100644 --- a/hive-bash-mcp/Cargo.toml +++ b/hive-bash-mcp/Cargo.toml @@ -8,6 +8,7 @@ workspace = true [dependencies] anyhow.workspace = true +hive-agent-sock.workspace = true hive-sh4re.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 89045a28..1db267f1 100644 --- a/hive-bash-mcp/src/runner.rs +++ b/hive-bash-mcp/src/runner.rs @@ -760,7 +760,7 @@ pub(crate) async fn send_wake( ); } } - let req = hive_sh4re::AgentRequest::Wake { + let req = hive_agent_sock::Request::Wake { from: format!("bash-task-{id}"), body, }; diff --git a/hive-c0re/Cargo.toml b/hive-c0re/Cargo.toml index b3e71558..6c7edbeb 100644 --- a/hive-c0re/Cargo.toml +++ b/hive-c0re/Cargo.toml @@ -30,6 +30,7 @@ opentelemetry-otlp = { version = "0.32", default-features = false, features = [ "reqwest-rustls", ] } indicatif.workspace = true +hive-agent-sock.workspace = true hive-sh4re.workspace = true hive-host-sock.workspace = true hive-priv-sock.workspace = true diff --git a/hive-c0re/src/agent_config/limits.rs b/hive-c0re/src/agent_config/limits.rs index ad981f63..723fc58f 100644 --- a/hive-c0re/src/agent_config/limits.rs +++ b/hive-c0re/src/agent_config/limits.rs @@ -23,7 +23,7 @@ pub const MESSAGE_MAX_BYTES: usize = 4096; /// Validate that `body` fits under [`MESSAGE_MAX_BYTES`]. Returns a /// caller-ready error string (caller wraps in -/// `AgentResponse::Err`/`ManagerResponse::Err`) on failure. +/// `Response::Err`) on failure. /// /// `label` shows up in the error message verbatim — pass a short /// noun like `"send"`, `"question"`, `"broadcast"` so the model can diff --git a/hive-c0re/src/dashboard/lifecycle_ops.rs b/hive-c0re/src/dashboard/lifecycle_ops.rs index 7960fe49..4f2bdb9f 100644 --- a/hive-c0re/src/dashboard/lifecycle_ops.rs +++ b/hive-c0re/src/dashboard/lifecycle_ops.rs @@ -77,7 +77,7 @@ pub(super) async fn post_kill( // host-side approval queue without the manager up, and // operator-driven meta-input updates work from the dashboard // either way. The MCP-surface self-kill guard in - // `socket_server.rs::ManagerRequest::Kill` stays in place: a + // `socket_server.rs::Request::Kill` stays in place: a // manager calling Kill on its own container is self-suicide // mid-call, not a legitimate operator action. submit::stop( diff --git a/hive-c0re/src/loose_ends.rs b/hive-c0re/src/loose_ends.rs index b4803b37..72a9b0ee 100644 --- a/hive-c0re/src/loose_ends.rs +++ b/hive-c0re/src/loose_ends.rs @@ -1,8 +1,8 @@ //! Loose-ends aggregator. Walks the `approvals` + `operator_questions` //! tables once per call and assembles a `Vec` for either -//! a single agent (`for_agent`) or the whole hive (`hive_wide`). Both -//! `AgentRequest::GetLooseEnds` and `ManagerRequest::GetLooseEnds` -//! land here so the routing logic + age-seconds derivation stay in +//! a single agent (`for_agent`) or the whole hive (`hive_wide`). +//! `Request::GetLooseEnds` from either the agent or manager socket +//! lands here so the routing logic + age-seconds derivation stay in //! one place. //! //! Call frequency is low (an agent doing self-introspection between diff --git a/hive-c0re/src/socket_server/config_approvals.rs b/hive-c0re/src/socket_server/config_approvals.rs index 02ed9b39..3b35f8be 100644 --- a/hive-c0re/src/socket_server/config_approvals.rs +++ b/hive-c0re/src/socket_server/config_approvals.rs @@ -9,7 +9,7 @@ use std::sync::Arc; -use hive_sh4re::AgentResponse; +use hive_agent_sock::Response; use super::require_new_child; use crate::coordinator::Coordinator; @@ -24,14 +24,14 @@ pub(super) fn handle_request_init_config( agent: &str, name: &str, description: Option, -) -> AgentResponse { +) -> Response { if let Some(err) = require_new_child(agent, name, "request_init_config for") { return err; } tracing::info!(%agent, %name, "request_init_config"); match submit_init_config(coord, name, Some(agent), description) { - Ok(_id) => AgentResponse::Ok, - Err(e) => AgentResponse::Err { + Ok(_id) => Response::Ok, + Err(e) => Response::Err { message: format!("{e:#}"), }, } @@ -45,7 +45,7 @@ pub(super) fn handle_request_update_meta_inputs( requester: &str, inputs: &[String], description: Option<&str>, -) -> AgentResponse { +) -> Response { let label = if inputs.is_empty() { "all inputs".to_string() } else { @@ -67,7 +67,7 @@ pub(super) fn handle_request_update_meta_inputs( { Ok(id) => id, Err(e) => { - return AgentResponse::Err { + return Response::Err { message: format!("queue update_meta_inputs approval: {e:#}"), }; } @@ -81,7 +81,7 @@ pub(super) fn handle_request_update_meta_inputs( description: description.map(str::to_owned), pr_number: None, }); - AgentResponse::Ok + Response::Ok } /// Submit-time half of the PR-merge flow: fetch the PR head sha from the diff --git a/hive-c0re/src/socket_server/lifecycle_handlers.rs b/hive-c0re/src/socket_server/lifecycle_handlers.rs index c0ef4d20..ae57b3bd 100644 --- a/hive-c0re/src/socket_server/lifecycle_handlers.rs +++ b/hive-c0re/src/socket_server/lifecycle_handlers.rs @@ -5,18 +5,14 @@ use std::sync::Arc; -use hive_sh4re::AgentResponse; +use hive_agent_sock::Response; use super::require_descendant; use crate::coordinator::Coordinator; /// `Start` — start a container, kicking its next turn. The caller must be an /// ancestor of `name` in the topology (the root covers every agent). -pub(super) async fn handle_start( - coord: &Arc, - agent: &str, - name: &str, -) -> AgentResponse { +pub(super) async fn handle_start(coord: &Arc, agent: &str, name: &str) -> Response { if let Some(err) = require_descendant(agent, name, "start") { return err; } @@ -31,18 +27,14 @@ pub(super) async fn handle_start( format!("agent `{agent}` start tool"), ) .await; - AgentResponse::Ok + Response::Ok } /// `Restart` — enqueue a restart for a container. The caller must be an /// ancestor of `name` in the topology. The infra-container branch is /// orthogonal: it is gated on the `infra_admin` capability and audited, so it /// stays ahead of the topology guard. -pub(super) async fn handle_restart( - coord: &Arc, - agent: &str, - name: &str, -) -> AgentResponse { +pub(super) async fn handle_restart(coord: &Arc, agent: &str, name: &str) -> Response { // Infra-container restart: an agent holding the `infra_admin` // capability can restart a hive infrastructure container (hive-ci / // hive-gateway / hive-forge / hive-matrix) by passing its name to the @@ -63,7 +55,7 @@ pub(super) async fn handle_restart( format!("agent `{agent}` restart tool"), ) .await; - AgentResponse::Ok + Response::Ok } /// Restart a hive infrastructure container on behalf of an agent that @@ -75,7 +67,7 @@ async fn handle_restart_infra( coord: &Arc, agent: &str, container: hive_priv_sock::InfraContainer, -) -> AgentResponse { +) -> Response { let name = container.unit_name(); // Record the attempt in the operator-visible privileged-action audit // trail, then emit a live `AuditEntryAdded` so the dashboard audit view @@ -97,7 +89,7 @@ async fn handle_restart_infra( crate::audit_log::AuditOutcome::Err, Some("denied: missing infra_admin capability"), ); - return AgentResponse::Err { + return Response::Err { message: format!( "restarting infra container `{name}` requires the `infra_admin` capability" ), @@ -107,23 +99,19 @@ async fn handle_restart_infra( match crate::priv_client::restart_infra_container(container).await { Ok(()) => { audit(crate::audit_log::AuditOutcome::Ok, None); - AgentResponse::Ok + Response::Ok } Err(e) => { let msg = format!("{e:#}"); audit(crate::audit_log::AuditOutcome::Err, Some(&msg)); - AgentResponse::Err { message: msg } + Response::Err { message: msg } } } } /// `Kill` — kill a container, unregister it, notify the manager. The caller /// must be an ancestor of `name` in the topology. -pub(super) async fn handle_kill( - coord: &Arc, - agent: &str, - name: &str, -) -> AgentResponse { +pub(super) async fn handle_kill(coord: &Arc, agent: &str, name: &str) -> Response { if let Some(err) = require_descendant(agent, name, "kill") { return err; } @@ -144,9 +132,9 @@ pub(super) async fn handle_kill( coord.notify_manager(&hive_sh4re::HelperEvent::Killed { agent: name.to_owned(), }); - AgentResponse::Ok + Response::Ok } - Err(e) => AgentResponse::Err { + Err(e) => Response::Err { message: format!("{e:#}"), }, } @@ -154,7 +142,7 @@ pub(super) async fn handle_kill( /// `Update` — enqueue a rebuild for a container. The caller must be an /// ancestor of `name` in the topology. -pub(super) fn handle_update(coord: &Arc, agent: &str, name: &str) -> AgentResponse { +pub(super) fn handle_update(coord: &Arc, agent: &str, name: &str) -> Response { if let Some(err) = require_descendant(agent, name, "rebuild") { return err; } @@ -165,12 +153,12 @@ pub(super) fn handle_update(coord: &Arc, agent: &str, name: &str) - crate::job_queue::Source::Manual, format!("agent `{agent}` update tool"), ); - AgentResponse::Ok + Response::Ok } /// `ListDescendants` — every topological descendant of `agent` with /// its running/stopped state, parents before children. -pub(super) async fn handle_list_descendants(agent: &str) -> AgentResponse { +pub(super) async fn handle_list_descendants(agent: &str) -> Response { tracing::debug!(%agent, "agent: list descendants"); // All containers known to nixos-container (running only). let running_set: std::collections::HashSet = match crate::lifecycle::list().await { @@ -182,7 +170,7 @@ pub(super) async fn handle_list_descendants(agent: &str) -> AgentResponse { }) .collect(), Err(e) => { - return AgentResponse::Err { + return Response::Err { message: format!("list containers failed: {e:#}"), }; } @@ -203,5 +191,5 @@ pub(super) async fn handle_list_descendants(agent: &str) -> AgentResponse { hive_sh4re::ContainerInfo { name, running } }) .collect(); - AgentResponse::Containers { containers } + Response::Containers { containers } } diff --git a/hive-c0re/src/socket_server/mod.rs b/hive-c0re/src/socket_server/mod.rs index 3d3f2cba..7b04daf3 100644 --- a/hive-c0re/src/socket_server/mod.rs +++ b/hive-c0re/src/socket_server/mod.rs @@ -13,7 +13,8 @@ use std::path::{Path, PathBuf}; use std::sync::Arc; use anyhow::{Context, Result}; -use hive_sh4re::{AgentRequest, AgentResponse, MANAGER_AGENT, Message}; +use hive_agent_sock::{Request, Response}; +use hive_sh4re::{MANAGER_AGENT, Message}; use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader}; use tokio::net::{UnixListener, UnixStream}; use tokio::task::JoinHandle; @@ -145,9 +146,9 @@ async fn serve(stream: UnixStream, agent: String, coord: Arc) -> Re if n == 0 { return Ok(()); } - let resp = match serde_json::from_str::(line.trim()) { + let resp = match serde_json::from_str::(line.trim()) { Ok(req) => dispatch(&req, &agent, &coord).await, - Err(e) => AgentResponse::Err { + Err(e) => Response::Err { message: format!("parse error: {e}"), }, }; @@ -189,24 +190,24 @@ pub(crate) fn recv_timeout(wait_seconds: Option) -> std::time::Duration { /// The unified `dispatch` calls this first; the remaining arms (which gate /// on topology / capabilities / tool-groups) are handled there. pub(crate) async fn dispatch_shared( - req: &hive_sh4re::Request, + req: &hive_agent_sock::Request, agent: &str, coord: &Arc, -) -> Option { +) -> Option { Some(match req { - hive_sh4re::Request::Send { + hive_agent_sock::Request::Send { to, body, in_reply_to, } => handle_send(coord, agent, to, body, *in_reply_to), - hive_sh4re::Request::Recv { wait_seconds, max } => { + hive_agent_sock::Request::Recv { wait_seconds, max } => { handle_recv(coord, agent, *wait_seconds, *max).await } - hive_sh4re::Request::Status => handle_status(coord, agent), - hive_sh4re::Request::OperatorMsg { body } => handle_operator_msg(coord, agent, body), - hive_sh4re::Request::Wake { from, body } => handle_wake(coord, agent, from, body), - hive_sh4re::Request::Recent { limit } => handle_recent(coord, agent, *limit), - hive_sh4re::Request::Ask { + hive_agent_sock::Request::Status => handle_status(coord, agent), + hive_agent_sock::Request::OperatorMsg { body } => handle_operator_msg(coord, agent, body), + hive_agent_sock::Request::Wake { from, body } => handle_wake(coord, agent, from, body), + hive_agent_sock::Request::Recent { limit } => handle_recent(coord, agent, *limit), + hive_agent_sock::Request::Ask { question, options, multi, @@ -222,42 +223,42 @@ pub(crate) async fn dispatch_shared( to.as_deref(), ) .map_or_else( - |message| hive_sh4re::Response::Err { message }, - |id| hive_sh4re::Response::QuestionQueued { id }, + |message| hive_agent_sock::Response::Err { message }, + |id| hive_agent_sock::Response::QuestionQueued { id }, ), - hive_sh4re::Request::Answer { id, answer } => { + hive_agent_sock::Request::Answer { id, answer } => { crate::questions::handle_answer(coord, agent, *id, answer).map_or_else( - |message| hive_sh4re::Response::Err { message }, - |()| hive_sh4re::Response::Ok, + |message| hive_agent_sock::Response::Err { message }, + |()| hive_agent_sock::Response::Ok, ) } - hive_sh4re::Request::Remind { + hive_agent_sock::Request::Remind { message, timing, file_path, } => handle_remind(coord, agent, message, timing, file_path.as_deref()), - hive_sh4re::Request::SetStatus { text } => handle_set_status(coord, text), - hive_sh4re::Request::GetAgentMeta { name } => { + hive_agent_sock::Request::SetStatus { text } => handle_set_status(coord, text), + hive_agent_sock::Request::GetAgentMeta { name } => { handle_get_agent_meta(coord, agent, name.as_deref()).await } - hive_sh4re::Request::CancelLooseEnd { kind, id } => { + hive_agent_sock::Request::CancelLooseEnd { kind, id } => { crate::questions::handle_cancel_loose_end(coord, agent, *kind, *id).map_or_else( - |message| hive_sh4re::Response::Err { message }, - |()| hive_sh4re::Response::Ok, + |message| hive_agent_sock::Response::Err { message }, + |()| hive_agent_sock::Response::Ok, ) } - hive_sh4re::Request::CreateRepo { repo } => handle_create_repo(agent, repo).await, - hive_sh4re::Request::AckTurn => handle_ack_turn(coord, agent), - hive_sh4re::Request::AckUntil { up_to } => handle_ack_until(coord, agent, *up_to), - hive_sh4re::Request::RequeueInflight => handle_requeue_inflight(coord, agent), - hive_sh4re::Request::GracefulStopComplete => { + hive_agent_sock::Request::CreateRepo { repo } => handle_create_repo(agent, repo).await, + hive_agent_sock::Request::AckTurn => handle_ack_turn(coord, agent), + hive_agent_sock::Request::AckUntil { up_to } => handle_ack_until(coord, agent, *up_to), + hive_agent_sock::Request::RequeueInflight => handle_requeue_inflight(coord, agent), + hive_agent_sock::Request::GracefulStopComplete => { // Harness drained + is exiting: clear the fence so the // `GracefulStop` orchestration (which polls this flag) proceeds // to stop the container without waiting out its timeout. coord.clear_graceful_stop(agent); - hive_sh4re::Response::Ok + hive_agent_sock::Response::Ok } - hive_sh4re::Request::GetHostJournal { + hive_agent_sock::Request::GetHostJournal { unit, container, lines, @@ -292,7 +293,7 @@ async fn handle_recv( agent: &str, wait_seconds: Option, max: Option, -) -> hive_sh4re::Response { +) -> hive_agent_sock::Response { // Graceful-stop fence: while a graceful stop is pending for this agent, // return `GracefulStop` instead of polling the broker. The harness runs // one stop-checkpoint turn then exits; new sends keep queueing in the @@ -300,7 +301,7 @@ async fn handle_recv( // so a flag set between polls is seen on the next Recv — the orchestration // also fires a transient wake to break an in-flight long-poll. if coord.is_graceful_stop_pending(agent) { - return hive_sh4re::Response::GracefulStop; + return hive_agent_sock::Response::GracefulStop; } let cap = max.unwrap_or(1).min(RECV_BATCH_MAX) as usize; match coord @@ -314,7 +315,7 @@ async fn handle_recv( // exactly how many still-pending messages remain to drain. A count // error is non-fatal — fall back to 0 rather than fail the recv. let remaining = coord.broker.count_pending(agent).unwrap_or(0); - hive_sh4re::Response::Messages { + hive_agent_sock::Response::Messages { messages: deliveries .into_iter() .map(|d| hive_sh4re::DeliveredMessage { @@ -328,7 +329,7 @@ async fn handle_recv( remaining, } } - Err(e) => hive_sh4re::Response::Err { + Err(e) => hive_agent_sock::Response::Err { message: format!("{e:#}"), }, } @@ -342,15 +343,15 @@ fn handle_wake( agent: &str, from: &str, body: &str, -) -> hive_sh4re::Response { +) -> hive_agent_sock::Response { match coord.broker.send(&Message { from: from.to_owned(), to: agent.to_owned(), body: body.to_owned(), in_reply_to: None, }) { - Ok(()) => hive_sh4re::Response::Ok, - Err(e) => hive_sh4re::Response::Err { + Ok(()) => hive_agent_sock::Response::Ok, + Err(e) => hive_agent_sock::Response::Err { message: format!("{e:#}"), }, } @@ -360,13 +361,13 @@ fn handle_wake( /// rescan. The harness has already written the status file to its own /// `state/` dir (it runs as the agent user), so this only refreshes the /// dashboard's view. -fn handle_set_status(coord: &Arc, text: &str) -> hive_sh4re::Response { +fn handle_set_status(coord: &Arc, text: &str) -> hive_agent_sock::Response { if let Err(message) = crate::limits::check_status_text(text) { - return hive_sh4re::Response::Err { message }; + return hive_agent_sock::Response::Err { message }; } let coord2 = Arc::clone(coord); tokio::spawn(async move { coord2.rescan_containers_and_emit().await }); - hive_sh4re::Response::Ok + hive_agent_sock::Response::Ok } /// Validate an agent-supplied repo name: a single safe slug segment, no @@ -384,9 +385,9 @@ fn valid_repo_name(name: &str) -> bool { /// `CreateRepo` — create a repo for `agent` *through hive-c0re* in the /// c0re-owned `agents` org with operator-team branch protection. /// The sanctioned create path now that agents can't create repos directly. -async fn handle_create_repo(agent: &str, repo: &str) -> hive_sh4re::Response { +async fn handle_create_repo(agent: &str, repo: &str) -> hive_agent_sock::Response { if !valid_repo_name(repo) { - return hive_sh4re::Response::Err { + return hive_agent_sock::Response::Err { message: format!( "invalid repo name {repo:?} — single segment of letters, digits, '-', '_', '.' \ (no leading '-'/'.', max 100 chars)" @@ -394,16 +395,16 @@ async fn handle_create_repo(agent: &str, repo: &str) -> hive_sh4re::Response { }; } let Some(core_token) = crate::forge::core_token() else { - return hive_sh4re::Response::Err { + return hive_agent_sock::Response::Err { message: "forge unavailable (no core token) — cannot create repo".to_owned(), }; }; match crate::forge::create_agent_repo(agent, repo, &core_token).await { - Ok(full_name) => hive_sh4re::Response::RepoCreated { + Ok(full_name) => hive_agent_sock::Response::RepoCreated { clone_url: format!("{}/{full_name}.git", crate::forge::forge_http_base()), full_name, }, - Err(e) => hive_sh4re::Response::Err { + Err(e) => hive_agent_sock::Response::Err { message: format!("create repo {repo:?} failed: {e:#}"), }, } @@ -416,7 +417,7 @@ async fn handle_get_agent_meta( coord: &Arc, agent: &str, name: Option<&str>, -) -> hive_sh4re::Response { +) -> hive_agent_sock::Response { let target = name.unwrap_or(agent); // `name` is agent-supplied and flows into filesystem reads below // (`read_agent_status_live`, `read_agent_matrix_identities` → @@ -425,14 +426,14 @@ async fn handle_get_agent_meta( // (`target == agent`) is the caller's own authenticated name, already // valid — but validating unconditionally is simplest and harmless. if let Some(reason) = crate::dashboard::validate_agent_name(target) { - return hive_sh4re::Response::Err { + return hive_agent_sock::Response::Err { message: format!("get_agent_meta: invalid agent name {target:?}: {reason}"), }; } let (status_text, status_set_at, running) = crate::container_view::read_agent_status_live(target).await; let (hive_name, swarm_name) = crate::container_view::hive_swarm_names(); - hive_sh4re::Response::AgentMeta { + hive_agent_sock::Response::AgentMeta { name: target.to_owned(), running, hyperhive_rev: crate::auto_update::current_flake_rev(&coord.hyperhive_flake), @@ -466,10 +467,10 @@ fn read_agent_matrix_identities(agent: &str) -> Vec } /// `Status` — count of pending (unread) inbox messages for `agent`. -fn handle_status(coord: &Arc, agent: &str) -> hive_sh4re::Response { +fn handle_status(coord: &Arc, agent: &str) -> hive_agent_sock::Response { match coord.broker.count_pending(agent) { - Ok(unread) => hive_sh4re::Response::Status { unread }, - Err(e) => hive_sh4re::Response::Err { + Ok(unread) => hive_agent_sock::Response::Status { unread }, + Err(e) => hive_agent_sock::Response::Err { message: format!("{e:#}"), }, } @@ -477,15 +478,19 @@ fn handle_status(coord: &Arc, agent: &str) -> hive_sh4re::Response /// `OperatorMsg` — deliver an operator-authored message into `agent`'s /// inbox (from the `operator` recipient). -fn handle_operator_msg(coord: &Arc, agent: &str, body: &str) -> hive_sh4re::Response { +fn handle_operator_msg( + coord: &Arc, + agent: &str, + body: &str, +) -> hive_agent_sock::Response { match coord.broker.send(&Message { from: hive_sh4re::OPERATOR_RECIPIENT.to_owned(), to: agent.to_owned(), body: body.to_owned(), in_reply_to: None, }) { - Ok(()) => hive_sh4re::Response::Ok, - Err(e) => hive_sh4re::Response::Err { + Ok(()) => hive_agent_sock::Response::Ok, + Err(e) => hive_agent_sock::Response::Err { message: format!("{e:#}"), }, } @@ -493,10 +498,10 @@ fn handle_operator_msg(coord: &Arc, agent: &str, body: &str) -> hiv /// `Recent` — the last `limit` inbox rows for `agent` (read-only, /// doesn't consume). -fn handle_recent(coord: &Arc, agent: &str, limit: u64) -> hive_sh4re::Response { +fn handle_recent(coord: &Arc, agent: &str, limit: u64) -> hive_agent_sock::Response { match coord.broker.recent_for(agent, limit) { - Ok(rows) => hive_sh4re::Response::Recent { rows }, - Err(e) => hive_sh4re::Response::Err { + Ok(rows) => hive_agent_sock::Response::Recent { rows }, + Err(e) => hive_agent_sock::Response::Err { message: format!("{e:#}"), }, } @@ -504,10 +509,10 @@ fn handle_recent(coord: &Arc, agent: &str, limit: u64) -> hive_sh4r /// `AckTurn` — mark `agent`'s in-flight delivered messages acked so /// they don't redeliver on the next turn. -fn handle_ack_turn(coord: &Arc, agent: &str) -> hive_sh4re::Response { +fn handle_ack_turn(coord: &Arc, agent: &str) -> hive_agent_sock::Response { match coord.broker.ack_turn(agent) { - Ok(_n) => hive_sh4re::Response::Ok, - Err(e) => hive_sh4re::Response::Err { + Ok(_n) => hive_agent_sock::Response::Ok, + Err(e) => hive_agent_sock::Response::Err { message: format!("{e:#}"), }, } @@ -515,10 +520,14 @@ fn handle_ack_turn(coord: &Arc, agent: &str) -> hive_sh4re::Respons /// `AckUntil` — bulk-ack every message addressed to `agent` with row /// id `<= up_to` (the agent-side backlog-triage escape hatch). -fn handle_ack_until(coord: &Arc, agent: &str, up_to: i64) -> hive_sh4re::Response { +fn handle_ack_until( + coord: &Arc, + agent: &str, + up_to: i64, +) -> hive_agent_sock::Response { match coord.broker.ack_until(agent, up_to) { - Ok(count) => hive_sh4re::Response::Acked { count }, - Err(e) => hive_sh4re::Response::Err { + Ok(count) => hive_agent_sock::Response::Acked { count }, + Err(e) => hive_agent_sock::Response::Err { message: format!("{e:#}"), }, } @@ -526,15 +535,15 @@ fn handle_ack_until(coord: &Arc, agent: &str, up_to: i64) -> hive_s /// `RequeueInflight` — resurface `agent`'s unacked in-flight messages /// (crash recovery on harness boot). -fn handle_requeue_inflight(coord: &Arc, agent: &str) -> hive_sh4re::Response { +fn handle_requeue_inflight(coord: &Arc, agent: &str) -> hive_agent_sock::Response { match coord.broker.requeue_inflight(agent) { Ok(n) => { if n > 0 { tracing::info!(%agent, requeued = %n, "requeued in-flight messages"); } - hive_sh4re::Response::Ok + hive_agent_sock::Response::Ok } - Err(e) => hive_sh4re::Response::Err { + Err(e) => hive_agent_sock::Response::Err { message: format!("{e:#}"), }, } @@ -548,37 +557,37 @@ fn handle_requeue_inflight(coord: &Arc, agent: &str) -> hive_sh4re: /// queries require the `QueryAgentState` capability; hive-wide orchestration /// verbs (schedules / meta-inputs) require the matching tool-group (the /// grantable capability). -async fn dispatch(req: &AgentRequest, agent: &str, coord: &Arc) -> AgentResponse { +async fn dispatch(req: &Request, agent: &str, coord: &Arc) -> Response { if let Some(resp) = dispatch_shared(req, agent, coord).await { return resp; } match req { // Lifecycle + config: caller must be an ancestor of the target // (a parent owns its whole subtree; the root covers every agent). - AgentRequest::Start { name } => handle_start(coord, agent, name).await, - AgentRequest::Restart { name } => handle_restart(coord, agent, name).await, - AgentRequest::Kill { name } => handle_kill(coord, agent, name).await, - AgentRequest::Update { name } => handle_update(coord, agent, name), - AgentRequest::ListDescendants => handle_list_descendants(agent).await, - AgentRequest::RequestInitConfig { name, description } => { + Request::Start { name } => handle_start(coord, agent, name).await, + Request::Restart { name } => handle_restart(coord, agent, name).await, + Request::Kill { name } => handle_kill(coord, agent, name).await, + Request::Update { name } => handle_update(coord, agent, name), + Request::ListDescendants => handle_list_descendants(agent).await, + Request::RequestInitConfig { name, description } => { handle_request_init_config(coord, agent, name, description.clone()) } // Agent-state queries: own subtree is free; other agents + the // hive-wide `"*"` sweep require `QueryAgentState`. - AgentRequest::GetLooseEnds { agent: target } => { + Request::GetLooseEnds { agent: target } => { handle_get_loose_ends(coord, agent, target.as_deref()) } - AgentRequest::CountPendingReminders { agent: target } => { + Request::CountPendingReminders { agent: target } => { handle_count_pending_reminders(coord, agent, target.as_deref()) } - AgentRequest::ReminderRollup { + Request::ReminderRollup { since_secs, agent: target, } => handle_reminder_rollup(coord, agent, target.as_deref(), *since_secs), // Todos (loose-ends v2): in-container subsystems push/clear // their own; the agent lists / marks its own done. Scoped to the // calling agent (the socket identity) — no cross-agent access. - AgentRequest::UpsertTodo { + Request::UpsertTodo { subsystem, key, summary, @@ -591,15 +600,13 @@ async fn dispatch(req: &AgentRequest, agent: &str, coord: &Arc) -> summary, source.as_deref(), ), - AgentRequest::ClearTodo { + Request::ClearTodo { subsystem, key, all, } => handle_clear_todo(coord, agent, subsystem, key.as_deref(), *all), - AgentRequest::ListTodos { subsystem } => { - handle_list_todos(coord, agent, subsystem.as_deref()) - } - AgentRequest::MarkTodoDone { id } => handle_mark_todo_done(coord, agent, *id), + Request::ListTodos { subsystem } => handle_list_todos(coord, agent, subsystem.as_deref()), + Request::MarkTodoDone { id } => handle_mark_todo_done(coord, agent, *id), // Orchestration / diagnostics verbs — gated per-verb on tool-group // membership or topology (see `dispatch_orchestration`). _ => dispatch_orchestration(req, agent, coord).await, @@ -612,13 +619,9 @@ async fn dispatch(req: &AgentRequest, agent: &str, coord: &Arc) -> /// (`scheduling` / `approvals`), or `is_descendant_of` for `get_logs` /// (a parent reads its subtree's logs). Any other variant is a host-admin / /// unknown request invalid on either socket. -async fn dispatch_orchestration( - req: &AgentRequest, - agent: &str, - coord: &Arc, -) -> AgentResponse { +async fn dispatch_orchestration(req: &Request, agent: &str, coord: &Arc) -> Response { match req { - AgentRequest::RequestUpdateMetaInputs { + Request::RequestUpdateMetaInputs { inputs, description, } => { @@ -627,19 +630,19 @@ async fn dispatch_orchestration( } handle_request_update_meta_inputs(coord, agent, inputs, description.as_deref()) } - AgentRequest::RequestSchedulePrompt(payload) => { + Request::RequestSchedulePrompt(payload) => { if let Some(err) = require_group(agent, "scheduling", "schedule a prompt") { return err; } handle_request_schedule_prompt(coord, agent, payload) } - AgentRequest::CancelSchedule { id, targets } => { + Request::CancelSchedule { id, targets } => { if let Some(err) = require_group(agent, "scheduling", "cancel a schedule") { return err; } handle_cancel_schedule(coord, agent, *id, targets.as_deref()) } - AgentRequest::EditSchedule { + Request::EditSchedule { id, body, description, @@ -665,19 +668,19 @@ async fn dispatch_orchestration( }, ) } - AgentRequest::ListSchedules => { + Request::ListSchedules => { if let Some(err) = require_group(agent, "scheduling", "list schedules") { return err; } handle_list_schedules(coord) } - AgentRequest::FireScheduleNow { id } => { + Request::FireScheduleNow { id } => { if let Some(err) = require_group(agent, "scheduling", "fire a schedule") { return err; } handle_fire_schedule_now(coord, agent, *id).await } - AgentRequest::GetLogs { + Request::GetLogs { agent: target, lines, } => { @@ -687,7 +690,7 @@ async fn dispatch_orchestration( handle_get_logs(target, *lines).await } // Host-admin-only / unknown variants: never valid on either socket. - _ => AgentResponse::Err { + _ => Response::Err { message: "request not handled on this socket".to_owned(), }, } @@ -699,11 +702,11 @@ async fn dispatch_orchestration( /// a consequence, with no positional/hardcoded privilege. Returns `Some(Err)` /// to short-circuit the dispatch arm when it isn't, `None` when authorised. /// `action` is the verb phrase for the message (e.g. `"start"`). -fn require_descendant(agent: &str, target: &str, action: &str) -> Option { +fn require_descendant(agent: &str, target: &str, action: &str) -> Option { if crate::topology::is_descendant_of(target, agent) { None } else { - Some(AgentResponse::Err { + Some(Response::Err { message: format!( "agent `{agent}` cannot {action} `{target}`: \ not in its subtree (topology)" @@ -718,14 +721,14 @@ fn require_descendant(agent: &str, target: &str, action: &str) -> Option Option { +fn require_group(agent: &str, group: &str, action: &str) -> Option { if crate::tool_groups::groups_for(agent) .iter() .any(|g| g == group) { None } else { - Some(AgentResponse::Err { + Some(Response::Err { message: format!("agent `{agent}` cannot {action}: requires the `{group}` tool group"), }) } @@ -744,9 +747,9 @@ fn require_group(agent: &str, group: &str, action: &str) -> Option Option { +fn require_new_child(agent: &str, target: &str, action: &str) -> Option { if let Some(reason) = crate::dashboard::validate_agent_name(target) { - return Some(AgentResponse::Err { + return Some(Response::Err { message: format!("agent `{agent}` cannot {action} `{target}`: {reason}"), }); } @@ -762,7 +765,7 @@ fn require_new_child(agent: &str, target: &str, action: &str) -> Option Option, - agent: &str, - target: Option<&str>, -) -> AgentResponse { +fn handle_get_loose_ends(coord: &Arc, agent: &str, target: Option<&str>) -> Response { let result = if target == Some("*") { if !crate::capabilities::has_cap(agent, hive_sh4re::Capability::QueryAgentState) { - return AgentResponse::Err { + return Response::Err { message: "query_agent_state capability required for hive-wide loose ends" .to_owned(), }; @@ -791,12 +790,12 @@ fn handle_get_loose_ends( } else { match resolve_agent_state_target(agent, target) { Ok(name) => crate::loose_ends::for_agent(coord, name), - Err(message) => return AgentResponse::Err { message }, + Err(message) => return Response::Err { message }, } }; match result { - Ok(loose_ends) => AgentResponse::LooseEnds { loose_ends }, - Err(e) => AgentResponse::Err { + Ok(loose_ends) => Response::LooseEnds { loose_ends }, + Err(e) => Response::Err { message: format!("{e:#}"), }, } @@ -812,7 +811,7 @@ fn handle_upsert_todo( key: Option<&str>, summary: &str, source: Option<&str>, -) -> AgentResponse { +) -> Response { match coord.todos.upsert(agent, subsystem, key, summary, source) { Ok((_, changed)) => { if changed { @@ -823,9 +822,9 @@ fn handle_upsert_todo( in_reply_to: None, }); } - AgentResponse::Ok + Response::Ok } - Err(e) => AgentResponse::Err { + Err(e) => Response::Err { message: format!("{e:#}"), }, } @@ -839,17 +838,17 @@ fn handle_clear_todo( subsystem: &str, key: Option<&str>, all: bool, -) -> AgentResponse { +) -> Response { let result = if all { coord.todos.clear_subsystem(agent, subsystem) } else { coord.todos.clear(agent, subsystem, key) }; match result { - Ok(count) => AgentResponse::Acked { + Ok(count) => Response::Acked { count: u64::try_from(count).unwrap_or(0), }, - Err(e) => AgentResponse::Err { + Err(e) => Response::Err { message: format!("{e:#}"), }, } @@ -857,14 +856,10 @@ fn handle_clear_todo( /// `ListTodos` — enumerate this agent's todos (optionally one subsystem's) /// as `LooseEnd::Todo` rows, so a producer can reconcile its own set. -fn handle_list_todos( - coord: &Arc, - agent: &str, - subsystem: Option<&str>, -) -> AgentResponse { +fn handle_list_todos(coord: &Arc, agent: &str, subsystem: Option<&str>) -> Response { match crate::loose_ends::todos_for(coord, agent, subsystem) { - Ok(loose_ends) => AgentResponse::LooseEnds { loose_ends }, - Err(e) => AgentResponse::Err { + Ok(loose_ends) => Response::LooseEnds { loose_ends }, + Err(e) => Response::Err { message: format!("{e:#}"), }, } @@ -872,12 +867,12 @@ fn handle_list_todos( /// `MarkTodoDone` — the agent clears one of its own todos by id (scoped to /// the agent, so it can't touch another agent's). -fn handle_mark_todo_done(coord: &Arc, agent: &str, id: i64) -> AgentResponse { +fn handle_mark_todo_done(coord: &Arc, agent: &str, id: i64) -> Response { match coord.todos.mark_done(agent, id) { - Ok(count) => AgentResponse::Acked { + Ok(count) => Response::Acked { count: u64::try_from(count).unwrap_or(0), }, - Err(e) => AgentResponse::Err { + Err(e) => Response::Err { message: format!("{e:#}"), }, } @@ -889,15 +884,15 @@ fn handle_count_pending_reminders( coord: &Arc, agent: &str, target: Option<&str>, -) -> AgentResponse { +) -> Response { match resolve_agent_state_target(agent, target) { Ok(name) => match coord.broker.count_pending_reminders_for(name) { - Ok(count) => AgentResponse::PendingRemindersCount { count }, - Err(e) => AgentResponse::Err { + Ok(count) => Response::PendingRemindersCount { count }, + Err(e) => Response::Err { message: format!("{e:#}"), }, }, - Err(message) => AgentResponse::Err { message }, + Err(message) => Response::Err { message }, } } @@ -909,15 +904,15 @@ fn handle_reminder_rollup( agent: &str, target: Option<&str>, since_secs: u64, -) -> AgentResponse { +) -> Response { match resolve_agent_state_target(agent, target) { Ok(name) => match coord.broker.reminder_rollup_for(name, since_secs) { - Ok(stats) => AgentResponse::ReminderRollup(stats), - Err(e) => AgentResponse::Err { + Ok(stats) => Response::ReminderRollup(stats), + Err(e) => Response::Err { message: format!("{e:#}"), }, }, - Err(message) => AgentResponse::Err { message }, + Err(message) => Response::Err { message }, } } @@ -940,7 +935,7 @@ pub struct HostJournalArgs<'a> { /// /// The manager is not exempt - grant `read_host_journal` in /// `meta/capabilities.json` to enable it for any agent including the manager. -pub async fn dispatch_host_journal(agent: &str, args: HostJournalArgs<'_>) -> AgentResponse { +pub async fn dispatch_host_journal(agent: &str, args: HostJournalArgs<'_>) -> Response { let HostJournalArgs { unit, container, @@ -951,7 +946,7 @@ pub async fn dispatch_host_journal(agent: &str, args: HostJournalArgs<'_>) -> Ag until, } = args; if !crate::capabilities::has_cap(agent, hive_sh4re::Capability::ReadHostJournal) { - return AgentResponse::Err { + return Response::Err { message: "agent does not have the read_host_journal capability".to_owned(), }; } @@ -979,9 +974,9 @@ pub async fn dispatch_host_journal(agent: &str, args: HostJournalArgs<'_>) -> Ag { Ok((stdout, stderr)) => { let content = if stdout.is_empty() { stderr } else { stdout }; - AgentResponse::HostJournal { content } + Response::HostJournal { content } } - Err(e) => AgentResponse::Err { + Err(e) => Response::Err { message: format!("journal read: {e:#}"), }, }; @@ -1023,9 +1018,9 @@ pub async fn dispatch_host_journal(agent: &str, args: HostJournalArgs<'_>) -> Ag let stderr = String::from_utf8_lossy(&out.stderr); format!("journalctl exited {}: {stderr}", out.status) }; - AgentResponse::HostJournal { content } + Response::HostJournal { content } } - Err(e) => AgentResponse::Err { + Err(e) => Response::Err { message: format!("journalctl spawn failed: {e:#}"), }, } @@ -1068,16 +1063,16 @@ pub(crate) fn handle_send( to: &str, body: &str, in_reply_to: Option, -) -> AgentResponse { +) -> Response { if let Err(message) = crate::limits::check_size("send", body) { - return AgentResponse::Err { message }; + return Response::Err { message }; } if to == "*" { let errors = coord.broadcast_send(agent, body); return if errors.is_empty() { - AgentResponse::Ok + Response::Ok } else { - AgentResponse::Err { + Response::Err { message: format!("broadcast failed for agents: {}", errors.join(", ")), } }; @@ -1090,9 +1085,9 @@ pub(crate) fn handle_send( let children = crate::topology::children_of(agent); let errors = fan_out_send(coord, agent, body, in_reply_to, &children); return if errors.is_empty() { - AgentResponse::Ok + Response::Ok } else { - AgentResponse::Err { + Response::Err { message: format!("children fan-out failed for agents: {}", errors.join(", ")), } }; @@ -1109,7 +1104,7 @@ pub(crate) fn handle_send( // Cross-hive messaging (`name@hive` qualified names) is not routed // through the broker — use the Matrix MCP tools for that instead. if resolved.contains('@') { - return AgentResponse::Err { + return Response::Err { message: format!( "send failed: cross-hive recipient `{resolved}` is not supported \ via the broker — use Matrix MCP tools for cross-hive messaging" @@ -1119,7 +1114,7 @@ pub(crate) fn handle_send( if resolved != hive_sh4re::OPERATOR_RECIPIENT { let state_root = crate::paths::agent_state_dir(&resolved); if !state_root.exists() { - return AgentResponse::Err { + return Response::Err { message: format!( "send failed: unknown recipient `{resolved}` \ (no agent with that name exists on this hive)" @@ -1133,8 +1128,8 @@ pub(crate) fn handle_send( body: body.to_owned(), in_reply_to, }) { - Ok(()) => AgentResponse::Ok, - Err(e) => AgentResponse::Err { + Ok(()) => Response::Ok, + Err(e) => Response::Err { message: format!("{e:#}"), }, } @@ -1143,7 +1138,7 @@ pub(crate) fn handle_send( /// `GetLogs` — read a child container's journal via hive-priv (the /// `-M` read needs root). `journalctl -M` wants the `h-` machine /// name, which `container_name` derives. -async fn handle_get_logs(agent: &str, lines: Option) -> AgentResponse { +async fn handle_get_logs(agent: &str, lines: Option) -> Response { let n = lines.unwrap_or(50); let machine = crate::lifecycle::container_name(agent); tracing::info!(%agent, %machine, %n, "manager: get_logs"); @@ -1158,9 +1153,9 @@ async fn handle_get_logs(agent: &str, lines: Option) -> AgentResponse { { Ok((stdout, stderr)) => { let content = if stdout.is_empty() { stderr } else { stdout }; - AgentResponse::Logs { content } + Response::Logs { content } } - Err(e) => AgentResponse::Err { + Err(e) => Response::Err { message: format!("get_logs: {e:#}"), }, } diff --git a/hive-c0re/src/socket_server/reminders.rs b/hive-c0re/src/socket_server/reminders.rs index 674ebc1a..e2b7b734 100644 --- a/hive-c0re/src/socket_server/reminders.rs +++ b/hive-c0re/src/socket_server/reminders.rs @@ -5,7 +5,7 @@ use std::sync::Arc; -use hive_sh4re::AgentResponse; +use hive_agent_sock::Response; use crate::coordinator::Coordinator; @@ -15,10 +15,10 @@ pub(super) fn handle_remind( message: &str, timing: &hive_sh4re::ReminderTiming, file_path: Option<&str>, -) -> AgentResponse { +) -> Response { match store_remind(coord, agent, message, timing, file_path) { - Ok(()) => AgentResponse::Ok, - Err(message) => AgentResponse::Err { message }, + Ok(()) => Response::Ok, + Err(message) => Response::Err { message }, } } diff --git a/hive-c0re/src/socket_server/schedules.rs b/hive-c0re/src/socket_server/schedules.rs index 96f7adea..9ba19b1b 100644 --- a/hive-c0re/src/socket_server/schedules.rs +++ b/hive-c0re/src/socket_server/schedules.rs @@ -6,17 +6,17 @@ use std::sync::Arc; -use hive_sh4re::AgentResponse; +use hive_agent_sock::Response; use crate::coordinator::Coordinator; /// `ListSchedules` — snapshot every scheduled prompt onto the wire. -pub(super) fn handle_list_schedules(coord: &Arc) -> AgentResponse { +pub(super) fn handle_list_schedules(coord: &Arc) -> Response { match coord.scheduled_prompts.list() { - Ok(schedules) => AgentResponse::Schedules { + Ok(schedules) => Response::Schedules { schedules: schedules.into_iter().map(schedule_to_wire).collect(), }, - Err(e) => AgentResponse::Err { + Err(e) => Response::Err { message: format!("list scheduled prompts: {e:#}"), }, } @@ -32,26 +32,26 @@ pub(super) fn handle_request_schedule_prompt( coord: &Arc, requester: &str, payload: &hive_sh4re::SchedulePromptPayload, -) -> AgentResponse { +) -> Response { if payload.targets.is_empty() { - return AgentResponse::Err { + return Response::Err { message: "schedule must have at least one target".into(), }; } if payload.body.trim().is_empty() { - return AgentResponse::Err { + return Response::Err { message: "schedule body must be non-empty".into(), }; } if let Some(0) = payload.interval_seconds { - return AgentResponse::Err { + return Response::Err { message: "interval_seconds must be > 0 (use None for one-shot)".into(), }; } let commit_ref = match serde_json::to_string(payload) { Ok(s) => s, Err(e) => { - return AgentResponse::Err { + return Response::Err { message: format!("encode SchedulePromptPayload: {e:#}"), }; } @@ -66,7 +66,7 @@ pub(super) fn handle_request_schedule_prompt( ) { Ok(id) => id, Err(e) => { - return AgentResponse::Err { + return Response::Err { message: format!("queue schedule_prompt approval: {e:#}"), }; } @@ -87,7 +87,7 @@ pub(super) fn handle_request_schedule_prompt( description: payload.description.clone(), pr_number: None, }); - AgentResponse::Ok + Response::Ok } /// Cancel a schedule (whole or per-target). Manager-surface @@ -101,22 +101,22 @@ pub(super) fn handle_cancel_schedule( requester: &str, schedule_id: i64, targets: Option<&[String]>, -) -> AgentResponse { +) -> Response { let schedule = match coord.scheduled_prompts.get(schedule_id) { Ok(Some(s)) => s, Ok(None) => { - return AgentResponse::Err { + return Response::Err { message: format!("schedule {schedule_id} not found"), }; } Err(e) => { - return AgentResponse::Err { + return Response::Err { message: format!("read schedule {schedule_id}: {e:#}"), }; } }; if !cancel_authorized(requester, &schedule.owner) { - return AgentResponse::Err { + return Response::Err { message: format!( "not authorized: {requester} cannot cancel schedule owned by {owner}", owner = schedule.owner @@ -136,9 +136,9 @@ pub(super) fn handle_cancel_schedule( match result { Ok(()) => { coord.emit_schedules_snapshot(); - AgentResponse::Ok + Response::Ok } - Err(message) => AgentResponse::Err { message }, + Err(message) => Response::Err { message }, } } @@ -151,22 +151,22 @@ pub(super) async fn handle_fire_schedule_now( coord: &Arc, requester: &str, schedule_id: i64, -) -> AgentResponse { +) -> Response { let schedule = match coord.scheduled_prompts.get(schedule_id) { Ok(Some(s)) => s, Ok(None) => { - return AgentResponse::Err { + return Response::Err { message: format!("schedule {schedule_id} not found"), }; } Err(e) => { - return AgentResponse::Err { + return Response::Err { message: format!("read schedule {schedule_id}: {e:#}"), }; } }; if !cancel_authorized(requester, &schedule.owner) { - return AgentResponse::Err { + return Response::Err { message: format!( "not authorized: {requester} cannot fire schedule owned by {owner}", owner = schedule.owner @@ -178,9 +178,9 @@ pub(super) async fn handle_fire_schedule_now( match crate::scheduled_prompts_worker::fire_now(coord, schedule_id, false).await { Ok(_report) => { coord.emit_schedules_snapshot(); - AgentResponse::Ok + Response::Ok } - Err(e) => AgentResponse::Err { + Err(e) => Response::Err { message: format!("fire schedule {schedule_id} now: {e:#}"), }, } @@ -217,7 +217,7 @@ pub(super) fn handle_edit_schedule( requester: &str, schedule_id: i64, patch: EditSchedulePatch, -) -> AgentResponse { +) -> Response { let EditSchedulePatch { body, description, @@ -229,18 +229,18 @@ pub(super) fn handle_edit_schedule( let schedule = match coord.scheduled_prompts.get(schedule_id) { Ok(Some(s)) => s, Ok(None) => { - return AgentResponse::Err { + return Response::Err { message: format!("schedule {schedule_id} not found"), }; } Err(e) => { - return AgentResponse::Err { + return Response::Err { message: format!("read schedule {schedule_id}: {e:#}"), }; } }; if !cancel_authorized(requester, &schedule.owner) { - return AgentResponse::Err { + return Response::Err { message: format!( "not authorized: {requester} cannot edit schedule owned by {owner}", owner = schedule.owner @@ -258,9 +258,9 @@ pub(super) fn handle_edit_schedule( match coord.scheduled_prompts.update(schedule_id, patch) { Ok(()) => { coord.emit_schedules_snapshot(); - AgentResponse::Ok + Response::Ok } - Err(e) => AgentResponse::Err { + Err(e) => Response::Err { message: format!("edit schedule {schedule_id}: {e:#}"), }, } diff --git a/hive-matrix-mcp/src/wake.rs b/hive-matrix-mcp/src/wake.rs index 93baf9d6..2ce4e737 100644 --- a/hive-matrix-mcp/src/wake.rs +++ b/hive-matrix-mcp/src/wake.rs @@ -3,7 +3,7 @@ //! //! Same wire shape as `hive-agent::forge_notify`'s wake: a single JSON //! line written to the hyperhive control socket (`/run/hive/mcp.sock` -//! by default) carrying an `AgentRequest::Wake { from, body }`. +//! by default) carrying an `Request::Wake { from, body }`. //! The agent harness's `agent_server` parses it and treats it as a //! `Wake` from the matrix subsystem. //! @@ -19,12 +19,12 @@ use anyhow::{Context, Result}; use tokio::io::AsyncWriteExt; use tokio::net::UnixStream; -/// Send an `AgentRequest::Wake { from: "matrix", body }` to the hyperhive +/// Send an `Request::Wake { from: "matrix", body }` to the hyperhive /// control socket at `socket`. Best-effort: returns Err on any plumbing /// failure; callers log + ignore so a wake delivery hiccup doesn't tear /// down the matrix sync loop. /// -/// Wire format matches `hive_sh4re::Request` tagged with `"cmd"` per +/// Wire format matches `hive_agent_sock::Request` tagged with `"cmd"` per /// `#[serde(tag = "cmd", rename_all = "snake_case")]`. Must be `"cmd"`, /// not `"kind"` — the harness deserialises against the hive-sh4re type /// and silently discards requests that don't match. diff --git a/hive-sh4re/src/lib.rs b/hive-sh4re/src/lib.rs index 73180185..c8bf4c15 100644 --- a/hive-sh4re/src/lib.rs +++ b/hive-sh4re/src/lib.rs @@ -8,7 +8,7 @@ pub mod jobs; pub mod paths; pub mod wire_time; -/// Server-side hard cap on `Recv.max` (see `AgentRequest::Recv`). Bounds +/// Server-side hard cap on `Recv.max` (see the `Recv` request). Bounds /// the size of a single round-trip so a confused caller can't drain the /// entire inbox in one go and blow past wire-buffer sizes; everything /// above the cap silently clamps. 5 keeps individual turns small — a big @@ -352,401 +352,6 @@ pub struct TaskFile { pub stderr_tail: Option, } -/// Unified request enum for both agent and manager sockets. The agent's -/// identity is the socket it arrived on. Privileged variants are marked -/// `*(privileged)*` — an agent socket returns `Err` for them server-side. -/// -/// `AgentRequest` and `ManagerRequest` are type aliases for this enum; -/// existing callers continue to compile unchanged. -#[derive(Debug, Clone, Serialize, Deserialize)] -#[serde(tag = "cmd", rename_all = "snake_case")] -pub enum Request { - /// Send a message to another agent. - Send { - to: String, - body: String, - /// Optional id of the message being replied to. Stored in the - /// broker DB and returned on `Recv` so the dashboard can render - /// threads. Ignored if the id is unknown or out of retention. - #[serde(default, skip_serializing_if = "Option::is_none")] - in_reply_to: Option, - }, - /// Pop pending messages from this agent's inbox. - /// Delivery + ack cycle: see - /// `docs/conventions.md::Broker delivery + ack cycle`. - Recv { - #[serde(default)] - wait_seconds: Option, - #[serde(default)] - max: Option, - }, - /// Non-mutating: how many pending messages are addressed to me? - /// Used by the harness to render a status line after each tool call. - Status, - /// Operator-injected message TO this agent (from this agent's own web - /// UI). Recipient is implicit — `from` is `"operator"`. Effectively the - /// per-agent equivalent of the old dashboard T4LK form, but scoped to - /// the agent whose page the operator is on. - OperatorMsg { body: String }, - /// Wake-up event injected from inside the container. Recipient is - /// implicit (this agent); `from` is caller-chosen. See - /// `docs/conventions.md::Wake injection` for the trust model and - /// typical callers. The wake is persisted in the sqlite broker - /// like any other message — the agent can ack it via `AckUntil`. - Wake { from: String, body: String }, - /// Last `limit` messages addressed to this agent, newest-first. - /// Non-mutating — pulls from the broker without delivering. The - /// per-agent web UI uses this to render its own inbox section. - Recent { limit: u64 }, - /// Surface a question to either the operator or another agent. - /// Routing + shape: see - /// `docs/conventions.md::Question routing (Ask / Answer)`. - Ask { - question: String, - #[serde(default)] - options: Vec, - #[serde(default)] - multi: bool, - #[serde(default)] - ttl_seconds: Option, - #[serde(default)] - to: Option, - }, - /// Answer a question previously routed to this agent via - /// `HelperEvent::QuestionAsked`. Authorised callers + threading - /// back via `HelperEvent::QuestionAnswered`: see - /// `docs/conventions.md::Question routing (Ask / Answer)`. - Answer { id: i64, answer: String }, - /// Schedule a reminder message to be delivered to this agent at a - /// future time. The reminder lands in the agent's inbox as an auto-sent - /// message from `"reminder"`. Use for agent follow-ups (e.g. check task - /// status, retry failed operation). Message length is limited; pass - /// `file_path` to store in a file and get a path-reference message - /// instead. - Remind { - message: String, - timing: ReminderTiming, - #[serde(default)] - file_path: Option, - }, - /// Loose-ends view. On the agent socket: `None` = self; direct - /// children are always accessible; non-children require the - /// `query_agent_state` capability — rejected with an error otherwise; - /// `"*"` is always rejected (use the manager socket). On the manager - /// socket: `None` = manager self, `"*"` = hive-wide, any name = - /// that agent. See `docs/conventions.md::Loose-ends wire shape`. - GetLooseEnds { - #[serde(default, skip_serializing_if = "Option::is_none")] - agent: Option, - }, - /// Upsert a *todo* (loose-ends v2) from an in-container - /// subsystem (matrix / forge / bash). `subsystem` is the producer - /// marker; `key` is the optional subsystem-specific dedup key (a - /// matrix room id, a bash task id). Re-pushing an identical keyed - /// todo is a no-op; a new-or-changed one coalesces a wake to the - /// agent. Keyless todos always insert as one-offs. - UpsertTodo { - subsystem: String, - #[serde(default, skip_serializing_if = "Option::is_none")] - key: Option, - summary: String, - #[serde(default, skip_serializing_if = "Option::is_none")] - source: Option, - }, - /// Clear producer-resolved todo(s) by `(subsystem, key)`. `key = - /// Some(k)` clears the one keyed row; `key = None` clears **all** of - /// the subsystem's keyless todos (rows with no key can't be told - /// apart — clear a specific one via `MarkTodoDone` by id). `all = - /// true` wipes the producer's whole set (cancel-and-recreate on - /// daemon restart). - ClearTodo { - subsystem: String, - #[serde(default, skip_serializing_if = "Option::is_none")] - key: Option, - #[serde(default)] - all: bool, - }, - /// List todos, optionally filtered to one `subsystem` (a producer - /// enumerating its own set). `None` = all. - ListTodos { - #[serde(default, skip_serializing_if = "Option::is_none")] - subsystem: Option, - }, - /// The agent marks one of its own todos done, by id. - MarkTodoDone { id: i64 }, - /// Count of pending (un-delivered) reminders. On the agent socket: - /// same target rules as `GetLooseEnds` (self/children free; - /// non-children require `query_agent_state`; `"*"` rejected). - /// On the manager socket: `None` = self, any name = that agent. - /// Used by the harness's per-turn stats sink. - CountPendingReminders { - #[serde(default, skip_serializing_if = "Option::is_none")] - agent: Option, - }, - /// Reminder statistics: counts of scheduled, delivered, and pending - /// reminders over a time window. `since_secs` filters to reminders - /// created in the last N seconds (0 = all). On the agent socket: - /// same target rules as `GetLooseEnds` (self/children free; - /// non-children require `query_agent_state`; `"*"` rejected). - /// On the manager socket: `None` = self, any name = that agent. - ReminderRollup { - /// Only count reminders created in the last N seconds from now. - /// Pass 0 to include all reminders. - #[serde(default)] - since_secs: u64, - /// Whose reminders to roll up. `None` = the caller's own. - /// `Some("")` = that agent's (requires `query_agent_state` - /// capability on the agent socket; always available on the manager - /// socket). - #[serde(default, skip_serializing_if = "Option::is_none")] - agent: Option, - }, - /// Set a free-text status string visible on the dashboard. The harness - /// writes `{state_dir}/hyperhive-status` locally before sending this - /// request; hive-c0re just triggers a dashboard rescan on receipt. - /// Pass an empty string to clear the status. - SetStatus { text: String }, - /// Fetch identity + status for an agent. `name = None` = - /// self-introspection; `Some()` = target query. See - /// `docs/conventions.md::Agent metadata`. - GetAgentMeta { - #[serde(default, skip_serializing_if = "Option::is_none")] - name: Option, - }, - /// Cancel an open thread the agent owns. Authorisation + - /// per-kind semantics in - /// `docs/conventions.md::Loose-ends wire shape`. - CancelLooseEnd { kind: CancelLooseEndKind, id: i64 }, - /// Create a git repo *through hive-c0re*. Agents can't create - /// repos with their own forge token (`max_repo_creation = 0`); this is - /// the sanctioned path. hive-c0re creates `repo` in the c0re-owned - /// `agents` org, adds the calling agent as a write collaborator (not - /// owner), and applies operator-team branch protection so the author - /// can't merge its own PRs. Returns the new repo's full name. - CreateRepo { repo: String }, - /// Mark every message popped since the last `AckTurn` as handled. - /// Harness↔broker pairing fired after `TurnOutcome::Ok`. See - /// `docs/conventions.md::Broker delivery + ack cycle`. - AckTurn, - /// Mark every inbox message with broker row id `<= up_to` as - /// handled (`acked_at` set), whether still pending or already - /// delivered. The agent-facing bulk-triage escape hatch for a - /// redelivered / accumulated backlog: instead of popping and - /// re-reading dozens of already-handled messages one turn at a - /// time, the agent acks everything up to the id it has seen. - /// Recipient-scoped — an agent can only ack its own rows. See - /// `docs/conventions.md::Broker delivery + ack cycle`. - AckUntil { up_to: i64 }, - /// Requeue every popped-but-unacked message back into the inbox. - /// Harness fires this once at boot to recover from - /// crashed-mid-turn sessions. See - /// `docs/conventions.md::Broker delivery + ack cycle`. - RequeueInflight, - /// Harness → c0re: "I saw the `GracefulStop` signal, ran my - /// stop-checkpoint turn (durable `/state` flushed) and am exiting my - /// serve loop now." Lets the `GracefulStop` orchestration stop the - /// container immediately instead of waiting out its timeout fallback. - GracefulStopComplete, - /// *(capability-gated: `read_host_journal`)* Fetch recent lines - /// from the host journal. Filters are all optional; omitting all - /// returns the last `lines` entries from the global journal. - GetHostJournal { - /// Filter to a specific systemd unit (e.g. `hive-c0re.service`). - #[serde(default, skip_serializing_if = "Option::is_none")] - unit: Option, - /// Machine name to pass to journalctl `-M` verbatim (e.g. `h-iris`). - /// The caller is responsible for the correct nspawn machine name. - #[serde(default, skip_serializing_if = "Option::is_none")] - container: Option, - /// Number of journal lines to return (default 30, max 100). - #[serde(default, skip_serializing_if = "Option::is_none")] - lines: Option, - /// Minimum syslog priority level. - #[serde(default, skip_serializing_if = "Option::is_none")] - priority: Option, - /// Regex to match against log message fields (journalctl `--grep`). - #[serde(default, skip_serializing_if = "Option::is_none")] - grep: Option, - /// Show entries on or newer than this timestamp (journalctl `--since`). - /// ISO 8601 or journalctl-accepted relative strings (e.g. `"-1h"`). - #[serde(default, skip_serializing_if = "Option::is_none")] - since: Option, - /// Show entries on or older than this timestamp (journalctl `--until`). - #[serde(default, skip_serializing_if = "Option::is_none")] - until: Option, - }, - - // ---- privileged (manager socket only for now) --------------------------- - /// *(privileged)* Initialise a brand-new agent's proposed config repo - /// and queue an approval for the operator to review. - RequestInitConfig { - name: String, - #[serde(default, skip_serializing_if = "Option::is_none")] - description: Option, - }, - /// *(privileged)* Stop a sub-agent (graceful). - Kill { name: String }, - /// *(privileged)* Start a previously-stopped sub-agent container. - Start { name: String }, - /// *(privileged)* Restart a sub-agent container (stop + start). - Restart { name: String }, - /// *(privileged)* Rebuild a sub-agent against the current hyperhive - /// flake + agent.nix. No approval required. - Update { name: String }, - /// *(privileged)* Fetch recent journal lines for a sub-agent container. - GetLogs { - agent: String, - #[serde(default)] - lines: Option, - }, - /// *(privileged)* Queue an approval to run `nix flake update [inputs...]`. - RequestUpdateMetaInputs { - #[serde(default)] - inputs: Vec, - #[serde(default, skip_serializing_if = "Option::is_none")] - description: Option, - }, - /// *(privileged)* Queue an approval to add a scheduled prompt. - RequestSchedulePrompt(SchedulePromptPayload), - /// *(privileged)* Cancel a scheduled prompt. - CancelSchedule { - id: i64, - #[serde(default, skip_serializing_if = "Option::is_none")] - targets: Option>, - }, - /// *(privileged)* List every schedule in the queue. - ListSchedules, - /// List all containers that are topological descendants of the calling - /// agent (direct children + their subtrees). Scoped to the caller's - /// subtree; gated by the `lifecycle` tool group. The result includes all - /// known descendants regardless of whether the container is currently - /// running — use `running` to distinguish. - ListDescendants, - /// *(privileged)* Fire a scheduled prompt out of band immediately. - FireScheduleNow { id: i64 }, - /// *(privileged)* Edit an existing schedule's mutable fields. - EditSchedule { - id: i64, - #[serde(default, skip_serializing_if = "Option::is_none")] - body: Option, - #[serde(default, skip_serializing_if = "Option::is_none")] - description: Option>, - #[serde(default, skip_serializing_if = "Option::is_none")] - interval_seconds: Option>, - #[serde(default, skip_serializing_if = "Option::is_none")] - next_fire_at_unix: Option, - #[serde(default, skip_serializing_if = "Option::is_none")] - targets_add: Option>, - #[serde(default, skip_serializing_if = "Option::is_none")] - targets_remove: Option>, - }, -} - -/// Backwards-compatible aliases. Both sockets now speak the unified `Request` -/// / `Response` wire; the server-side privilege gate rejects privileged -/// variants on agent sockets with `Err { message: "privileged variant..." }`. -pub type AgentRequest = Request; -pub type ManagerRequest = Request; - -/// Unified response enum for both agent and manager sockets. Privileged -/// variants (`Logs`, `Schedules`) are never returned on agent sockets. -/// -/// `AgentResponse` and `ManagerResponse` are type aliases for this enum. -#[derive(Debug, Clone, Serialize, Deserialize)] -#[serde(tag = "kind", rename_all = "snake_case")] -pub enum Response { - /// `Send` succeeded. - Ok, - /// Either `Send` failed or `Recv` errored. - Err { message: String }, - /// `Recv` result: zero or more messages, FIFO-ordered, never - /// longer than the `max` the caller passed. Empty vec = nothing - /// pending (the "(empty)" path for the formatter). Per-row `id` + - /// `redelivered` carry the broker's row id (tracked by the harness - /// for `AckTurn`, and surfaced to claude as a `[msg #]` marker - /// so `AckUntil` has something to reference) and the "previously - /// popped, not acked" flag — see `DeliveredMessage` for details. - /// `remaining` is the inbox depth *after* this batch was popped — - /// how many still-pending messages the caller could drain next. The - /// harness surfaces it to claude ("N more pending") so an in-turn - /// `recv` learns whether the inbox is drained, mirroring the count - /// the wake prompt already carries. - Messages { - messages: Vec, - remaining: u64, - }, - /// `Status` result: how many pending messages are in this agent's inbox. - Status { unread: u64 }, - /// `AckUntil` result: how many rows were newly marked handled. - Acked { count: u64 }, - /// `Recent` result: newest-first inbox rows. - Recent { rows: Vec }, - /// `Ask` result: the queued question id. The answer lands later - /// as `HelperEvent::QuestionAnswered` in this agent's inbox. - QuestionQueued { id: i64 }, - /// `GetLooseEnds` result: list of loose ends pending against - /// this agent. Ordered newest-first within each kind. - LooseEnds { loose_ends: Vec }, - /// `CountPendingReminders` result. - PendingRemindersCount { count: u64 }, - /// `ReminderRollup` result: reminder activity stats for the agent. - ReminderRollup(ReminderStats), - /// `GetAgentMeta` result. Per-field semantics + serde defaults - /// live in `docs/conventions.md::Agent metadata`. - AgentMeta { - name: String, - #[serde(default = "default_true")] - running: bool, - #[serde(default, skip_serializing_if = "Option::is_none")] - hyperhive_rev: Option, - #[serde(default, skip_serializing_if = "Option::is_none")] - status_text: Option, - #[serde(default, skip_serializing_if = "Option::is_none")] - status_set_at: Option, - #[serde(default, skip_serializing_if = "Option::is_none")] - hive_name: Option, - #[serde(default, skip_serializing_if = "Option::is_none")] - swarm_name: Option, - /// Matrix identities this agent can act as (one per configured + - /// live account). Empty for agents with no matrix provisioning. - #[serde(default, skip_serializing_if = "Vec::is_empty")] - matrix_accounts: Vec, - }, - /// `GetLogs` result: journal lines for the requested container. - /// Returned on the manager socket only. - Logs { content: String }, - /// `GetHostJournal` result: host journal lines matching the - /// requested filters. Returned on the agent socket when the agent - /// holds the `read_host_journal` capability. - HostJournal { content: String }, - /// `ListSchedules` result. Snapshot of every schedule. - /// Returned on the manager socket only. - Schedules { schedules: Vec }, - /// `CreateRepo` result: the new repo's full name (`agents/`) - /// and clone URL, so the agent can immediately `git clone` it. - RepoCreated { - full_name: String, - clone_url: String, - }, - /// `ListDescendants` result: all descendant containers, with running - /// status. Ordered by topology depth (parents before children), then - /// alphabetically within each depth tier. - Containers { containers: Vec }, - /// `Recv` result when a graceful stop is pending for this agent - /// (set by hive-c0re's `GracefulStop` orchestration). Returned in - /// place of `Messages` — it doubles as the inbound fence: the harness - /// stops consuming normal inbox messages and instead runs one - /// stop-checkpoint turn (flush durable `/state`), then reports - /// `GracefulStopComplete` and exits its serve loop so the container - /// can be stopped cleanly. New sends keep queueing in the broker for - /// the agent's next start. - GracefulStop, -} - -/// Backwards-compatible response aliases. -pub type AgentResponse = Response; -pub type ManagerResponse = Response; - /// One entry in a `ListDescendants` result. #[derive(Debug, Clone, Serialize, Deserialize)] pub struct ContainerInfo { @@ -802,13 +407,6 @@ pub struct MatrixIdentity { pub homeserver: String, } -/// Serde default for the `running` field; keeps wire backwards-compat -/// with pre-running-field payloads. See -/// `docs/conventions.md::Agent metadata`. -fn default_true() -> bool { - true -} - // ----------------------------------------------------------------------------- // Manager socket — /run/hyperhive/manager/mcp.sock on the host, bind-mounted // into the manager container at /run/hive/mcp.sock.