diff --git a/hive-ag3nt/src/bin/hive.rs b/hive-ag3nt/src/bin/hive.rs index e078fd03..964884f1 100644 --- a/hive-ag3nt/src/bin/hive.rs +++ b/hive-ag3nt/src/bin/hive.rs @@ -188,6 +188,14 @@ trait Surface { /// system-prompt block + tool registration goes into the spawned /// `claude` process. const FLAVOR: mcp::Flavor; + /// `is_manager` flag passed to `forge_notify::run`. Picks which + /// wire enum (`AgentRequest::Wake` vs `ManagerRequest::Wake`) the + /// poller uses to push notifications into the harness inbox — the + /// per-role broker socket rejects the wrong type. Lifting + /// `Surface` into the lib crate to make `forge_notify::run` + /// generic is deferred to its own issue. + const FORGE_IS_MANAGER: bool; + /// Ack the in-flight turn. Logs warnings on transport/broker /// errors but never propagates — turn loop continues either way. fn ack_turn(socket: &Path) -> impl Future; @@ -240,6 +248,7 @@ struct AgentSurface; impl Surface for AgentSurface { const FLAVOR: mcp::Flavor = mcp::Flavor::Agent; + const FORGE_IS_MANAGER: bool = false; async fn ack_turn(socket: &Path) { match client::request::<_, AgentResponse>(socket, &AgentRequest::AckTurn).await { @@ -272,7 +281,7 @@ impl Surface for AgentSurface { async fn post_turn_counts(socket: &Path) -> (Option, Option) { let threads = - match client::request::<_, AgentResponse>(socket, &AgentRequest::GetLooseEnds { agent: None }).await { + match client::request::<_, AgentResponse>(socket, &AgentRequest::GetLooseEnds).await { Ok(AgentResponse::LooseEnds { loose_ends }) => { u64::try_from(loose_ends.len()).ok() } @@ -280,7 +289,7 @@ impl Surface for AgentSurface { }; let reminders = match client::request::<_, AgentResponse>( socket, - &AgentRequest::CountPendingReminders { agent: None }, + &AgentRequest::CountPendingReminders, ) .await { @@ -377,6 +386,7 @@ struct ManagerSurface; impl Surface for ManagerSurface { const FLAVOR: mcp::Flavor = mcp::Flavor::Manager; + const FORGE_IS_MANAGER: bool = true; async fn ack_turn(socket: &Path) { match client::request::<_, ManagerResponse>(socket, &ManagerRequest::AckTurn).await { @@ -550,7 +560,10 @@ async fn serve_main(socket: &Path, poll_ms: u64) -> Result<()> { for failure in plugins::install_configured(socket).await { S::send_to_parent(socket, failure).await; } - tokio::spawn(hive_ag3nt::forge_notify::run(socket.to_path_buf())); + tokio::spawn(hive_ag3nt::forge_notify::run( + socket.to_path_buf(), + S::FORGE_IS_MANAGER, + )); // Log web_ui::serve's error instead of dropping it. A bare // `tokio::spawn(web_ui::serve(...))` discards the JoinHandle, so // any Err (e.g. EACCES from `bind_unix` when HIVE_WEB_SOCKET points diff --git a/hive-ag3nt/src/forge_notify.rs b/hive-ag3nt/src/forge_notify.rs index 7a86c711..64f74168 100644 --- a/hive-ag3nt/src/forge_notify.rs +++ b/hive-ag3nt/src/forge_notify.rs @@ -26,7 +26,10 @@ const BODY_TRUNCATE: usize = 500; /// configured. Otherwise loops forever, polling every /// `POLL_INTERVAL_SECS` seconds. Errors are never fatal. /// -pub async fn run(socket: PathBuf) { +/// `is_manager`: when true, wakes the inbox via `ManagerRequest::Wake` +/// instead of `AgentRequest::Wake` (the manager socket rejects the agent +/// request type). +pub async fn run(socket: PathBuf, is_manager: bool) { let forge_url = match std::env::var("HIVE_FORGE_URL") { Ok(u) if !u.is_empty() => u, _ => { @@ -119,6 +122,7 @@ pub async fn run(socket: PathBuf) { &forge_url, &token, &socket, + is_manager, keep_subscriptions, &mut unsubbed_repos, &own_login, @@ -619,6 +623,7 @@ async fn poll_once( forge_url: &str, token: &str, socket: &Path, + is_manager: bool, keep_subscriptions: bool, unsubbed_repos: &mut HashSet, own_login: &str, @@ -685,13 +690,23 @@ async fn poll_once( continue; }; - let req = hive_sh4re::Request::Wake { - from: "forge".to_owned(), - body, + let delivered = if is_manager { + let req = hive_sh4re::ManagerRequest::Wake { + from: "forge".to_owned(), + body, + }; + crate::client::request::<_, hive_sh4re::ManagerResponse>(socket, &req) + .await + .map(|_| ()) + } else { + let req = hive_sh4re::AgentRequest::Wake { + from: "forge".to_owned(), + body, + }; + crate::client::request::<_, hive_sh4re::AgentResponse>(socket, &req) + .await + .map(|_| ()) }; - let delivered = crate::client::request::<_, hive_sh4re::Response>(socket, &req) - .await - .map(|_| ()); match delivered { Ok(()) => { debug!(%id, "forge_notify: delivered"); diff --git a/hive-ag3nt/src/mcp.rs b/hive-ag3nt/src/mcp.rs index 614fd79b..3b806fcd 100644 --- a/hive-ag3nt/src/mcp.rs +++ b/hive-ag3nt/src/mcp.rs @@ -64,23 +64,60 @@ pub enum SocketReply { }, } -impl From for SocketReply { - fn from(r: hive_sh4re::Response) -> Self { +impl From for SocketReply { + fn from(r: hive_sh4re::AgentResponse) -> Self { match r { - hive_sh4re::Response::Ok => Self::Ok, - hive_sh4re::Response::Err { message } => Self::Err(message), - hive_sh4re::Response::Messages { messages } => Self::Messages(messages), - hive_sh4re::Response::Status { unread } => Self::Status(unread), - hive_sh4re::Response::Recent { rows } => Self::Recent(rows), - hive_sh4re::Response::QuestionQueued { id } => Self::QuestionQueued(id), - hive_sh4re::Response::LooseEnds { loose_ends } => Self::LooseEnds(loose_ends), - hive_sh4re::Response::PendingRemindersCount { count } => { + hive_sh4re::AgentResponse::Ok => Self::Ok, + hive_sh4re::AgentResponse::Err { message } => Self::Err(message), + hive_sh4re::AgentResponse::Messages { messages } => Self::Messages(messages), + hive_sh4re::AgentResponse::Status { unread } => Self::Status(unread), + hive_sh4re::AgentResponse::Recent { rows } => Self::Recent(rows), + hive_sh4re::AgentResponse::QuestionQueued { id } => Self::QuestionQueued(id), + hive_sh4re::AgentResponse::LooseEnds { loose_ends } => Self::LooseEnds(loose_ends), + hive_sh4re::AgentResponse::PendingRemindersCount { count } => { Self::PendingRemindersCount(count) } - hive_sh4re::Response::ReminderRollup(stats) => Self::ReminderRollup(stats), - hive_sh4re::Response::Logs { content } => Self::Logs(content), - hive_sh4re::Response::Schedules { schedules } => Self::Schedules(schedules), - hive_sh4re::Response::AgentMeta { + hive_sh4re::AgentResponse::ReminderRollup(stats) => Self::ReminderRollup(stats), + hive_sh4re::AgentResponse::AgentMeta { + name, + role, + running, + hyperhive_rev, + status_text, + status_set_at, + hive_name, + swarm_name, + } => Self::AgentMeta { + name, + role, + running, + hyperhive_rev, + status_text, + status_set_at, + hive_name, + swarm_name, + }, + } + } +} + +impl From for SocketReply { + fn from(r: hive_sh4re::ManagerResponse) -> Self { + match r { + hive_sh4re::ManagerResponse::Ok => Self::Ok, + hive_sh4re::ManagerResponse::Err { message } => Self::Err(message), + hive_sh4re::ManagerResponse::Messages { messages } => Self::Messages(messages), + hive_sh4re::ManagerResponse::Status { unread } => Self::Status(unread), + hive_sh4re::ManagerResponse::QuestionQueued { id } => Self::QuestionQueued(id), + hive_sh4re::ManagerResponse::Recent { rows } => Self::Recent(rows), + hive_sh4re::ManagerResponse::Logs { content } => Self::Logs(content), + hive_sh4re::ManagerResponse::Schedules { schedules } => Self::Schedules(schedules), + hive_sh4re::ManagerResponse::LooseEnds { loose_ends } => Self::LooseEnds(loose_ends), + hive_sh4re::ManagerResponse::PendingRemindersCount { count } => { + Self::PendingRemindersCount(count) + } + hive_sh4re::ManagerResponse::ReminderRollup(stats) => Self::ReminderRollup(stats), + hive_sh4re::ManagerResponse::AgentMeta { name, role, running, @@ -610,7 +647,7 @@ impl AgentServer { )] async fn get_loose_ends(&self) -> String { run_tool_envelope("get_loose_ends", String::new(), async move { - let (resp, retries) = self.dispatch(hive_sh4re::AgentRequest::GetLooseEnds { agent: None }).await; + let (resp, retries) = self.dispatch(hive_sh4re::AgentRequest::GetLooseEnds).await; annotate_retries(format_loose_ends(resp), retries) }) .await diff --git a/hive-ag3nt/src/web_ui.rs b/hive-ag3nt/src/web_ui.rs index 3a84f613..ed3c192c 100644 --- a/hive-ag3nt/src/web_ui.rs +++ b/hive-ag3nt/src/web_ui.rs @@ -63,7 +63,17 @@ struct AppState { gui_vnc_port: Option, } -/// Re-export so callers in `turn.rs` can name the type via `web_ui::Flavor`. +impl AppState { + fn flavor(&self) -> Flavor { + self.files.flavor + } +} + +/// Which wire protocol the per-agent UI's `/send` handler should speak. +/// Sub-agent → `AgentRequest::OperatorMsg`; manager → +/// `ManagerRequest::OperatorMsg`. Reuses the MCP-side enum so a +/// single value drives both the send protocol and (in +/// `post_compact`) the allowed-tools surface claude sees. pub type Flavor = mcp::Flavor; /// Bind the per-container web listener and serve the SPA. @@ -369,7 +379,7 @@ async fn api_stats( // filters its counts to the same time range as the chart data. let window_secs = window.span_secs(); let window_secs_u = u64::try_from(window_secs).unwrap_or(0); - snapshot.reminder_stats = fetch_reminder_stats(&state.socket, window_secs_u).await; + snapshot.reminder_stats = fetch_reminder_stats(&state.socket, state.flavor(), window_secs_u).await; axum::Json(snapshot) } @@ -495,18 +505,39 @@ struct SessionView { /// the `mcp__hyperhive__get_loose_ends` tool sees from inside the /// container. async fn api_loose_ends(State(state): State) -> Response { - let loose_ends: Vec = match client::request::<_, hive_sh4re::Response>( - &state.socket, - &hive_sh4re::Request::GetLooseEnds { agent: None }, - ) - .await - { - Ok(hive_sh4re::Response::LooseEnds { loose_ends }) => loose_ends, - Ok(hive_sh4re::Response::Err { message }) => { - return error_response(&format!("get_loose_ends: {message}")); + let loose_ends: Vec = match state.flavor() { + Flavor::Agent => { + match client::request::<_, hive_sh4re::AgentResponse>( + &state.socket, + &hive_sh4re::AgentRequest::GetLooseEnds, + ) + .await + { + Ok(hive_sh4re::AgentResponse::LooseEnds { loose_ends }) => loose_ends, + Ok(hive_sh4re::AgentResponse::Err { message }) => { + return error_response(&format!("get_loose_ends: {message}")); + } + Ok(other) => return error_response(&format!("unexpected response: {other:?}")), + Err(e) => return error_response(&format!("transport: {e:#}")), + } + } + Flavor::Manager => { + match client::request::<_, hive_sh4re::ManagerResponse>( + &state.socket, + // Manager's own loose ends — the web page is the + // manager's page, not a hive-wide console. + &hive_sh4re::ManagerRequest::GetLooseEnds { agent: None }, + ) + .await + { + Ok(hive_sh4re::ManagerResponse::LooseEnds { loose_ends }) => loose_ends, + Ok(hive_sh4re::ManagerResponse::Err { message }) => { + return error_response(&format!("get_loose_ends: {message}")); + } + Ok(other) => return error_response(&format!("unexpected response: {other:?}")), + Err(e) => return error_response(&format!("transport: {e:#}")), + } } - Ok(other) => return error_response(&format!("unexpected response: {other:?}")), - Err(e) => return error_response(&format!("transport: {e:#}")), }; axum::Json(serde_json::json!({ "loose_ends": loose_ends })).into_response() } @@ -536,7 +567,7 @@ async fn api_state(State(state): State) -> axum::Json { .ok() .and_then(|s| s.parse::().ok()) .unwrap_or(7000); - let inbox = recent_inbox(&state.socket).await; + let inbox = recent_inbox(&state.socket, state.flavor()).await; let (turn_state, turn_state_since) = state.bus.state_snapshot(); let model = state.bus.model(); let context_window_tokens = state @@ -645,34 +676,67 @@ struct ExtraLink { /// Best-effort: pull the last 30 messages addressed to us via the /// per-agent / manager socket. Empty list on any transport / decode /// failure — the inbox section is decorative, not authoritative. -async fn recent_inbox(socket: &std::path::Path) -> Vec { +async fn recent_inbox(socket: &std::path::Path, flavor: Flavor) -> Vec { const LIMIT: u64 = 30; - match client::request::<_, hive_sh4re::Response>( - socket, - &hive_sh4re::Request::Recent { limit: LIMIT }, - ) - .await - { - Ok(hive_sh4re::Response::Recent { rows }) => rows, - _ => Vec::new(), + match flavor { + Flavor::Agent => { + match client::request::<_, hive_sh4re::AgentResponse>( + socket, + &hive_sh4re::AgentRequest::Recent { limit: LIMIT }, + ) + .await + { + Ok(hive_sh4re::AgentResponse::Recent { rows }) => rows, + _ => Vec::new(), + } + } + Flavor::Manager => { + match client::request::<_, hive_sh4re::ManagerResponse>( + socket, + &hive_sh4re::ManagerRequest::Recent { limit: LIMIT }, + ) + .await + { + Ok(hive_sh4re::ManagerResponse::Recent { rows }) => rows, + _ => Vec::new(), + } + } } } /// Fetch reminder activity stats from the broker via the per-agent / /// manager socket. Returns None on any transport / decode failure — the /// stats are decorative, not authoritative. -async fn fetch_reminder_stats(socket: &std::path::Path, window_secs: u64) -> Option { - match client::request::<_, hive_sh4re::Response>( - socket, - &hive_sh4re::Request::ReminderRollup { - since_secs: window_secs, - agent: None, - }, - ) - .await - { - Ok(hive_sh4re::Response::ReminderRollup(stats)) => Some(stats), - _ => None, +async fn fetch_reminder_stats(socket: &std::path::Path, flavor: Flavor, window_secs: u64) -> Option { + match flavor { + Flavor::Agent => { + match client::request::<_, hive_sh4re::AgentResponse>( + socket, + &hive_sh4re::AgentRequest::ReminderRollup { + since_secs: window_secs, + }, + ) + .await + { + Ok(hive_sh4re::AgentResponse::ReminderRollup(stats)) => Some(stats), + _ => None, + } + } + Flavor::Manager => { + match client::request::<_, hive_sh4re::ManagerResponse>( + socket, + &hive_sh4re::ManagerRequest::ReminderRollup { + since_secs: window_secs, + // Manager's own stats page — its own reminders. + agent: None, + }, + ) + .await + { + Ok(hive_sh4re::ManagerResponse::ReminderRollup(stats)) => Some(stats), + _ => None, + } + } } } @@ -690,16 +754,29 @@ async fn post_send(State(state): State, Form(form): Form) -> if body.is_empty() { return error_response("send: `body` required"); } - let result = match client::request::<_, hive_sh4re::Response>( - &state.socket, - &hive_sh4re::Request::OperatorMsg { body }, - ) - .await - { - Ok(hive_sh4re::Response::Ok) => Ok(()), - Ok(hive_sh4re::Response::Err { message }) => Err(message), - Ok(other) => Err(format!("unexpected response: {other:?}")), - Err(e) => Err(format!("transport: {e:#}")), + let result = match state.flavor() { + Flavor::Agent => match client::request::<_, hive_sh4re::AgentResponse>( + &state.socket, + &hive_sh4re::AgentRequest::OperatorMsg { body }, + ) + .await + { + Ok(hive_sh4re::AgentResponse::Ok) => Ok(()), + Ok(hive_sh4re::AgentResponse::Err { message }) => Err(message), + Ok(other) => Err(format!("unexpected response: {other:?}")), + Err(e) => Err(format!("transport: {e:#}")), + }, + Flavor::Manager => match client::request::<_, hive_sh4re::ManagerResponse>( + &state.socket, + &hive_sh4re::ManagerRequest::OperatorMsg { body }, + ) + .await + { + Ok(hive_sh4re::ManagerResponse::Ok) => Ok(()), + Ok(hive_sh4re::ManagerResponse::Err { message }) => Err(message), + Ok(other) => Err(format!("unexpected response: {other:?}")), + Err(e) => Err(format!("transport: {e:#}")), + }, }; match result { // 200 instead of 303 → the client doesn't refetch /api/state. diff --git a/hive-c0re/src/agent_server.rs b/hive-c0re/src/agent_server.rs index c27130cc..c658fb21 100644 --- a/hive-c0re/src/agent_server.rs +++ b/hive-c0re/src/agent_server.rs @@ -208,13 +208,13 @@ async fn dispatch(req: &AgentRequest, agent: &str, coord: &Arc) -> timing, file_path, } => handle_remind(coord, agent, message, timing, file_path.as_deref()), - AgentRequest::GetLooseEnds { .. } => match crate::loose_ends::for_agent(coord, agent) { + AgentRequest::GetLooseEnds => match crate::loose_ends::for_agent(coord, agent) { Ok(loose_ends) => AgentResponse::LooseEnds { loose_ends }, Err(e) => AgentResponse::Err { message: format!("{e:#}"), }, }, - AgentRequest::CountPendingReminders { .. } => { + AgentRequest::CountPendingReminders => { match coord.broker.count_pending_reminders_for(agent) { Ok(count) => AgentResponse::PendingRemindersCount { count }, Err(e) => AgentResponse::Err { @@ -222,7 +222,7 @@ async fn dispatch(req: &AgentRequest, agent: &str, coord: &Arc) -> }, } } - AgentRequest::ReminderRollup { since_secs, .. } => { + AgentRequest::ReminderRollup { since_secs } => { match coord.broker.reminder_rollup_for(agent, *since_secs) { Ok(stats) => AgentResponse::ReminderRollup(stats), Err(e) => AgentResponse::Err { @@ -310,10 +310,6 @@ async fn dispatch(req: &AgentRequest, agent: &str, coord: &Arc) -> message: format!("{e:#}"), }, }, - // Manager-only variants are not valid on the agent socket. - _ => AgentResponse::Err { - message: "request not supported on agent socket".to_owned(), - }, } } diff --git a/hive-sh4re/src/lib.rs b/hive-sh4re/src/lib.rs index 45488fcf..160ee5c9 100644 --- a/hive-sh4re/src/lib.rs +++ b/hive-sh4re/src/lib.rs @@ -295,15 +295,11 @@ pub enum CancelLooseEndKind { Approval, } -/// 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. +/// Requests on a per-agent socket. The agent's identity is the socket +/// it came in on; `Send.from` is filled in by the server, not the client. #[derive(Debug, Clone, Serialize, Deserialize)] #[serde(tag = "cmd", rename_all = "snake_case")] -pub enum Request { +pub enum AgentRequest { /// Send a message to another agent. Send { to: String, @@ -371,37 +367,23 @@ pub enum Request { #[serde(default)] file_path: Option, }, - /// Loose-ends view. On the agent socket, scoped to the calling agent - /// (the `agent` field is ignored — agents can only see their own - /// loose ends). On the manager socket, `agent = None` scopes to the - /// manager itself, `Some("*")` is hive-wide, `Some("")` is - /// that agent's loose ends. See + /// Loose-ends view: every pending row against THIS agent. + /// Per-flavour scoping in /// `docs/conventions.md::Loose-ends wire shape`. - GetLooseEnds { - #[serde(default, skip_serializing_if = "Option::is_none")] - agent: Option, - }, - /// Count of pending (un-delivered) reminders. On the agent socket - /// always scoped to the calling agent. On the manager socket, - /// `agent = None` means self, `Some("")` means 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 manager socket - /// `agent = None` means self, `Some("")` means that agent. + GetLooseEnds, + /// Count of this agent's pending (un-delivered) reminders. Used + /// by the harness's per-turn stats sink to snapshot "what was + /// queued at turn-end time" without paying for a full list. + CountPendingReminders, + /// Reminder statistics for this agent: counts of scheduled, delivered, + /// and pending reminders over a time window. Used by the stats page + /// to display reminder activity. `since_secs` filters to reminders + /// created in the last N seconds (0 = all reminders). 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. - /// Manager socket only: `Some("")` = that agent's. - #[serde(default, skip_serializing_if = "Option::is_none")] - agent: Option, }, /// Set a free-text status string visible on the dashboard. Persisted /// to `{state_dir}/hyperhive-status` so it survives harness restarts. @@ -427,88 +409,12 @@ pub enum Request { /// crashed-mid-turn sessions. See /// `docs/conventions.md::Broker delivery + ack cycle`. RequeueInflight, - - // ---- 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)* Submit a config commit for the operator to approve. - RequestApplyCommit { - agent: String, - commit_ref: String, - #[serde(default, skip_serializing_if = "Option::is_none")] - description: Option, - }, - /// *(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, - /// *(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. +/// Responses on a per-agent socket. #[derive(Debug, Clone, Serialize, Deserialize)] #[serde(tag = "kind", rename_all = "snake_case")] -pub enum Response { +pub enum AgentResponse { /// `Send` succeeded. Ok, /// Either `Send` failed or `Recv` errored. @@ -552,18 +458,8 @@ pub enum Response { #[serde(default, skip_serializing_if = "Option::is_none")] swarm_name: Option, }, - /// `GetLogs` result: journal lines for the requested container. - /// Returned on the manager socket only. - Logs { content: String }, - /// `ListSchedules` result. Snapshot of every schedule. - /// Returned on the manager socket only. - Schedules { schedules: Vec }, } -/// Backwards-compatible response aliases. -pub type AgentResponse = Response; -pub type ManagerResponse = Response; - /// Serde default for the `running` field; keeps wire backwards-compat /// with pre-running-field payloads. See /// `docs/conventions.md::Agent metadata`. @@ -686,6 +582,240 @@ pub enum HelperEvent { }, } +/// Requests on the manager socket. Manager has the agent surface (send/recv) +/// plus privileged lifecycle verbs. +#[derive(Debug, Clone, Serialize, Deserialize)] +#[serde(tag = "cmd", rename_all = "snake_case")] +pub enum ManagerRequest { + Send { + to: String, + body: String, + /// Optional id of the message being replied to. Mirror of + /// `AgentRequest::Send.in_reply_to`; see that doc. + #[serde(default, skip_serializing_if = "Option::is_none")] + in_reply_to: Option, + }, + /// Same shape as `AgentRequest::Recv` — caller-tunable + /// `wait_seconds` (capped at 60s server-side, default 30s when + /// None) for first-message long-poll, plus `max` (default 1, cap + /// 32) to drain up to N popped rows in one round-trip. + Recv { + #[serde(default)] + wait_seconds: Option, + #[serde(default)] + max: Option, + }, + /// Non-mutating: pending message count, used to render a status line + /// after each MCP tool call (mirrors `AgentRequest::Status`). + Status, + /// Operator-injected message TO the manager (from the manager's own web + /// UI). Same shape as `AgentRequest::OperatorMsg`. + OperatorMsg { body: String }, + /// Last `limit` messages addressed to the manager, newest-first. + /// Non-mutating; mirror of `AgentRequest::Recent`. + Recent { limit: u64 }, + /// Initialise a brand-new agent's proposed config repo and queue an + /// approval for the operator to review. On approval hive-c0re seeds + /// `/agents//config/` with the default `agent.nix` template, + /// giving the manager RW access so it can customise the config and + /// commit changes. After the `ConfigReady` event arrives, edit + /// `agent.nix`, commit, and call `request_apply_commit` — which + /// creates the container on the first deploy. Fails if a proposed + /// repo for this name already exists (use `request_apply_commit` to + /// update an existing agent's config). + RequestInitConfig { + name: String, + /// Optional description shown on the dashboard approval card. + #[serde(default, skip_serializing_if = "Option::is_none")] + description: Option, + }, + /// Stop a sub-agent (graceful). + Kill { name: String }, + /// Start a previously-stopped sub-agent container. + Start { name: String }, + /// Restart a sub-agent container (stop + start). + Restart { name: String }, + /// Rebuild a sub-agent: re-applies the current hyperhive flake + + /// agent.nix, restarts the container. No approval required — + /// it's idempotent and the manager owns its own update cadence. + Update { name: String }, + /// Submit a config commit for the user to approve. `commit_ref` must + /// be a commit sha (7-40 hex chars, short or full) in the agent's + /// proposed config repo — a branch or tag name is rejected so the + /// approval pins an immutable commit. On approval the host applies + /// the change via `nixos-container update`. + RequestApplyCommit { + agent: String, + commit_ref: String, + /// Optional description shown on the dashboard approval card so the + /// operator knows what the change does without opening the diff. + #[serde(default, skip_serializing_if = "Option::is_none")] + description: Option, + }, + /// Surface a question to either the operator or another agent. + /// Manager-flavour mirror of `AgentRequest::Ask` — routing + shape + /// docs in `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 the manager via + /// `HelperEvent::QuestionAsked`. Mirror of `AgentRequest::Answer`; + /// see `docs/conventions.md::Question routing (Ask / Answer)`. + Answer { id: i64, answer: String }, + /// Fetch recent journal lines for a sub-agent container. `agent` + /// is the logical agent name; hive-c0re resolves it to the + /// machine name (`gui` → `h-gui`) and runs `journalctl -M + /// -n --no-pager`, returning the output as a + /// string. Useful for diagnosing MCP registration failures, + /// startup crashes, and harness errors. + /// + /// `lines` defaults to 50 when omitted. + GetLogs { + agent: String, + #[serde(default)] + lines: Option, + }, + /// Mirror of `AgentRequest::Remind` on the manager surface — schedule + /// a reminder addressed to the manager itself. Same semantics: body + /// soft-caps at 4 KiB, oversize bodies auto-persist to + /// `/state/reminders/auto-.md` (the manager container's own state + /// mount) and the inbox sees a pointer. + Remind { + message: String, + timing: ReminderTiming, + #[serde(default)] + file_path: Option, + }, + /// Loose-ends view for the manager surface. The optional `agent` + /// field selects scope: + /// - `None` — the manager's own loose ends: approvals it + /// submitted + questions where it is asker/target + its own + /// pending reminders. This is the default. + /// - `Some("*")` — hive-wide: EVERY pending approval, unanswered + /// question, and pending reminder across the swarm. + /// - `Some("")` — that specific agent's loose ends. + GetLooseEnds { + #[serde(default)] + agent: Option, + }, + /// Count of pending reminders. `agent` selects whose: `None` = + /// the manager's own, `Some("")` = that agent's. Mirror of + /// `AgentRequest::CountPendingReminders` on the manager surface. + CountPendingReminders { + #[serde(default)] + agent: Option, + }, + /// Reminder statistics: counts of scheduled, delivered, and pending + /// reminders (manager-flavour). Mirror of `AgentRequest::ReminderRollup`. + 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 manager's own, + /// `Some("")` = that agent's. + #[serde(default)] + agent: Option, + }, + /// Mirror of `AgentRequest::SetStatus` on the manager surface. + SetStatus { text: String }, + /// Mirror of `AgentRequest::GetAgentMeta` on the manager surface. + /// See `docs/conventions.md::Agent metadata`. + GetAgentMeta { + #[serde(default, skip_serializing_if = "Option::is_none")] + name: Option, + }, + /// Cancel an open thread (question or reminder). Manager surface + /// can cancel any row (no owner check) — same dispatch as + /// `AgentRequest::CancelLooseEnd` but with privileged auth. + CancelLooseEnd { kind: CancelLooseEndKind, id: i64 }, + /// Mirror of `AgentRequest::AckTurn` on the manager surface — fired + /// by the manager harness after `TurnOutcome::Ok` to close out + /// every message popped during the turn. + AckTurn, + /// Mirror of `AgentRequest::RequeueInflight` on the manager + /// surface — fired exactly once on manager harness boot. + RequeueInflight, + /// Mirror of `AgentRequest::Wake` on the manager surface. See + /// `docs/conventions.md::Wake injection`. + Wake { from: String, body: String }, + /// Queue an approval to run `nix flake update [inputs...]` on the + /// meta flake. `inputs` is the list of named inputs to update + /// (e.g. `["bitburner-agent", "nixpkgs"]`). Pass an empty list to + /// update ALL inputs. On operator approval hive-c0re runs the lock + /// update and commits the result. The `UpdateMetaInputs` approval + /// resolves with `ApprovalResolved` in the manager inbox. + RequestUpdateMetaInputs { + #[serde(default)] + inputs: Vec, + /// Optional description shown on the dashboard approval card. + #[serde(default, skip_serializing_if = "Option::is_none")] + description: Option, + }, + /// Queue an approval to add a scheduled prompt. The requester + /// (caller of this request) is recorded as the schedule owner; on + /// operator approval hive-c0re inserts the schedule and the worker + /// fans the body out at fire time. Even agent-self schedules go + /// through approval — the existing `remind` MCP tool is the + /// unapproved self-wake path. + RequestSchedulePrompt(SchedulePromptPayload), + /// Cancel a scheduled prompt. `targets = None` cancels the whole + /// schedule; `Some(list)` cancels just those recipients, + /// auto-cancelling the parent when no active targets remain. + /// Authorization: manager can cancel its own schedules + any + /// sub-agent schedules (i.e. owner reachable via topology); the + /// operator surface bypasses this check. + CancelSchedule { + id: i64, + #[serde(default, skip_serializing_if = "Option::is_none")] + targets: Option>, + }, + /// List every schedule in the queue. Manager-side this is + /// unfiltered — the dashboard does the topology-filter for the + /// per-agent view. + ListSchedules, + /// Fire a scheduled prompt out of band. Runs the per-target + /// fan-out once immediately without touching + /// `next_fire_at_unix` on recurring schedules; one-shots are + /// consumed by the manual fire. Authorization mirrors + /// `CancelSchedule`: the manager can fire its own schedules and + /// any owned by a sub-agent in its subtree per topology.json; + /// the operator surface bypasses the check. + FireScheduleNow { id: i64 }, + /// Edit an existing schedule's mutable fields. Partial PATCH + /// semantics: `None` / missing JSON key = leave alone, + /// `Some(_)` = set. `interval_seconds` and `description` are + /// doubly-wrapped so `Some(None)` (set explicit null) can flip + /// a recurring schedule back to one-shot / clear the + /// description, while plain `None` keeps the current value. + /// `targets_add` / `targets_remove` mutate the recipient list + /// in the same transaction; re-adding a previously-cancelled + /// target drops the tombstone (replace-on-conflict). Refuses + /// cancelled rows. Authorization mirrors `CancelSchedule`. + 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>, + }, +} /// Submission payload for `RequestSchedulePrompt`. Lives outside the /// enum so it can also serialize into the approval row's `commit_ref` @@ -752,3 +882,72 @@ pub struct WireScheduleTarget { pub last_result: Option, } +#[derive(Debug, Clone, Serialize, Deserialize)] +#[serde(tag = "kind", rename_all = "snake_case")] +pub enum ManagerResponse { + Ok, + Err { + message: String, + }, + /// Mirror of `AgentResponse::Messages` on the manager surface. + /// Always-list shape: 0..=max popped rows, FIFO-ordered. Carries + /// per-row `id` + `redelivered` so the manager harness drives the + /// same ack + requeue-with-hint flow as a sub-agent. + Messages { + messages: Vec, + }, + Status { + unread: u64, + }, + /// Result of `Ask`: the queued question id. The actual answer + /// arrives later as a `HelperEvent::QuestionAnswered` in the + /// asker's inbox, so this returns immediately rather than blocking + /// the turn. + QuestionQueued { + id: i64, + }, + /// `Recent` result: mirror of `AgentResponse::Recent`. + Recent { + rows: Vec, + }, + /// `GetLogs` result: journal lines for the requested container. + Logs { + content: String, + }, + /// `ListSchedules` result. Snapshot of every schedule (active + + /// cancelled-but-not-yet-reaped); the dashboard does the + /// per-agent topology filter on top. + Schedules { + schedules: Vec, + }, + /// `GetLooseEnds` result: hive-wide loose ends (approvals + + /// unanswered questions). Same `LooseEnd` variants as the + /// agent surface; the manager's view is unfiltered. + LooseEnds { + loose_ends: Vec, + }, + /// `CountPendingReminders` result. + PendingRemindersCount { + count: u64, + }, + /// `ReminderRollup` result: reminder activity stats for the manager. + ReminderRollup(ReminderStats), + /// Mirror of `AgentResponse::AgentMeta` on the manager surface. + /// See `docs/conventions.md::Agent metadata`. + AgentMeta { + name: String, + role: 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, + }, +}