diff --git a/hive-c0re/src/agent_server.rs b/hive-c0re/src/agent_server.rs index 11bdd5dc..030d01b9 100644 --- a/hive-c0re/src/agent_server.rs +++ b/hive-c0re/src/agent_server.rs @@ -95,7 +95,7 @@ async fn serve(stream: UnixStream, agent: String, coord: Arc) -> Re /// cheap "is there anything pending?" check without blocking the /// turn for 30 seconds. To actually park, the caller passes a /// positive `wait_seconds`. -pub(crate) const RECV_LONG_POLL_MAX: std::time::Duration = std::time::Duration::from_secs(180); +const RECV_LONG_POLL_MAX: std::time::Duration = std::time::Duration::from_secs(180); /// Server-side hard cap on `Recv.max`. Bounds the size of a single /// round-trip so a confused caller can't drain the entire inbox in @@ -104,9 +104,9 @@ pub(crate) const RECV_LONG_POLL_MAX: std::time::Duration = std::time::Duration:: /// seen in practice (post-rebuild rescue, multi-agent reply storms) /// and well under the per-message `MESSAGE_MAX_BYTES` * N envelope /// budget. -pub(crate) const RECV_BATCH_MAX: u32 = 32; +const RECV_BATCH_MAX: u32 = 32; -pub(crate) fn recv_timeout(wait_seconds: Option) -> std::time::Duration { +fn recv_timeout(wait_seconds: Option) -> std::time::Duration { match wait_seconds { Some(s) => std::time::Duration::from_secs(s).min(RECV_LONG_POLL_MAX), None => std::time::Duration::ZERO, @@ -114,31 +114,19 @@ pub(crate) fn recv_timeout(wait_seconds: Option) -> std::time::Duration { } #[allow(clippy::too_many_lines)] -/// Handle the subset of `Request` variants that are identical on both -/// the agent socket and the manager socket. Returns `Some(response)` for -/// every variant it handles; returns `None` for variants with socket-specific -/// semantics (e.g. `GetLooseEnds` / `CountPendingReminders` / `ReminderRollup` -/// where the manager can target other agents) or for manager-only variants. -/// -/// Both `agent_server::dispatch` and `manager_server::dispatch` call this -/// first; each then handles its own remaining arms. -pub(crate) async fn dispatch_shared( - req: &hive_sh4re::Request, - agent: &str, - coord: &Arc, -) -> Option { +async fn dispatch(req: &AgentRequest, agent: &str, coord: &Arc) -> AgentResponse { let broker = &coord.broker; - Some(match req { - hive_sh4re::Request::Send { to, body, in_reply_to } => { + match req { + AgentRequest::Send { to, body, in_reply_to } => { handle_send(coord, agent, to, body, *in_reply_to) } - hive_sh4re::Request::Recv { wait_seconds, max } => { + AgentRequest::Recv { wait_seconds, max } => { let cap = max.unwrap_or(1).min(RECV_BATCH_MAX) as usize; match broker .recv_blocking_batch(agent, recv_timeout(*wait_seconds), cap) .await { - Ok(deliveries) => hive_sh4re::Response::Messages { + Ok(deliveries) => AgentResponse::Messages { messages: deliveries .into_iter() .map(|d| hive_sh4re::DeliveredMessage { @@ -150,46 +138,46 @@ pub(crate) async fn dispatch_shared( }) .collect(), }, - Err(e) => hive_sh4re::Response::Err { + Err(e) => AgentResponse::Err { message: format!("{e:#}"), }, } } - hive_sh4re::Request::Status => match broker.count_pending(agent) { - Ok(unread) => hive_sh4re::Response::Status { unread }, - Err(e) => hive_sh4re::Response::Err { + AgentRequest::Status => match broker.count_pending(agent) { + Ok(unread) => AgentResponse::Status { unread }, + Err(e) => AgentResponse::Err { message: format!("{e:#}"), }, }, - hive_sh4re::Request::OperatorMsg { body } => match broker.send(&Message { + AgentRequest::OperatorMsg { body } => match broker.send(&Message { from: hive_sh4re::OPERATOR_RECIPIENT.to_owned(), to: agent.to_owned(), body: body.clone(), in_reply_to: None, }) { - Ok(()) => hive_sh4re::Response::Ok, - Err(e) => hive_sh4re::Response::Err { + Ok(()) => AgentResponse::Ok, + Err(e) => AgentResponse::Err { message: format!("{e:#}"), }, }, - hive_sh4re::Request::Wake { from, body } => match broker.send(&Message { + AgentRequest::Wake { from, body } => match broker.send(&Message { from: from.clone(), to: agent.to_owned(), body: body.clone(), in_reply_to: None, }) { - Ok(()) => hive_sh4re::Response::Ok, - Err(e) => hive_sh4re::Response::Err { + Ok(()) => AgentResponse::Ok, + Err(e) => AgentResponse::Err { message: format!("{e:#}"), }, }, - hive_sh4re::Request::Recent { limit } => match broker.recent_for(agent, *limit) { - Ok(rows) => hive_sh4re::Response::Recent { rows }, - Err(e) => hive_sh4re::Response::Err { + AgentRequest::Recent { limit } => match broker.recent_for(agent, *limit) { + Ok(rows) => AgentResponse::Recent { rows }, + Err(e) => AgentResponse::Err { message: format!("{e:#}"), }, }, - hive_sh4re::Request::Ask { + AgentRequest::Ask { question, options, multi, @@ -205,108 +193,21 @@ 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| AgentResponse::Err { message }, + |id| AgentResponse::QuestionQueued { id }, ), - hive_sh4re::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, - ) - } - hive_sh4re::Request::Remind { + AgentRequest::Answer { id, answer } => crate::questions::handle_answer( + coord, agent, *id, answer, + ) + .map_or_else( + |message| AgentResponse::Err { message }, + |()| AgentResponse::Ok, + ), + AgentRequest::Remind { message, timing, file_path, } => handle_remind(coord, agent, message, timing, file_path.as_deref()), - hive_sh4re::Request::SetStatus { text } => { - if let Err(message) = crate::limits::check_status_text(text) { - return Some(hive_sh4re::Response::Err { message }); - } - let path = crate::coordinator::Coordinator::agent_notes_dir(agent) - .join("hyperhive-status"); - let result = if text.trim().is_empty() { - std::fs::remove_file(&path).or_else(|e| { - if e.kind() == std::io::ErrorKind::NotFound { - Ok(()) - } else { - Err(e) - } - }) - } else { - std::fs::write(&path, format!("{}\n", text.trim())) - }; - match result { - Ok(()) => { - let coord2 = Arc::clone(coord); - tokio::spawn(async move { coord2.rescan_containers_and_emit().await }); - hive_sh4re::Response::Ok - } - Err(e) => { - hive_sh4re::Response::Err { - message: format!("set_status write failed: {e}"), - } - } - } - } - hive_sh4re::Request::GetAgentMeta { name } => { - let target = name.as_deref().unwrap_or(agent); - let (status_text, status_set_at, running) = - crate::container_view::read_agent_status_live(target).await; - let role = if target == hive_sh4re::MANAGER_AGENT { - "manager" - } else { - "agent" - } - .to_owned(); - let (hive_name, swarm_name) = crate::container_view::hive_swarm_names(); - hive_sh4re::Response::AgentMeta { - name: target.to_owned(), - role, - running, - hyperhive_rev: crate::auto_update::current_flake_rev(&coord.hyperhive_flake), - status_text, - status_set_at, - hive_name, - swarm_name, - } - } - hive_sh4re::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, - ) - } - hive_sh4re::Request::AckTurn => match broker.ack_turn(agent) { - Ok(_n) => hive_sh4re::Response::Ok, - Err(e) => hive_sh4re::Response::Err { - message: format!("{e:#}"), - }, - }, - hive_sh4re::Request::RequeueInflight => match broker.requeue_inflight(agent) { - Ok(n) => { - if n > 0 { - tracing::info!(%agent, requeued = %n, "requeued in-flight messages"); - } - hive_sh4re::Response::Ok - } - Err(e) => hive_sh4re::Response::Err { - message: format!("{e:#}"), - }, - }, - hive_sh4re::Request::GetHostJournal { unit, container, lines, priority, grep, since, until } => { - dispatch_host_journal(agent, unit, container, lines, priority, grep, since, until).await - } - // Not a shared variant. - _ => return None, - }) -} - -async fn dispatch(req: &AgentRequest, agent: &str, coord: &Arc) -> AgentResponse { - if let Some(resp) = dispatch_shared(req, agent, coord).await { - return resp; - } - match req { AgentRequest::GetLooseEnds { .. } => match crate::loose_ends::for_agent(coord, agent) { Ok(loose_ends) => AgentResponse::LooseEnds { loose_ends }, Err(e) => AgentResponse::Err { @@ -329,6 +230,89 @@ async fn dispatch(req: &AgentRequest, agent: &str, coord: &Arc) -> }, } } + AgentRequest::SetStatus { text } => { + // Cap length + reject multi-line so a confused caller + // can't dump a multi-paragraph session report into the + // dashboard chip. + if let Err(message) = crate::limits::check_status_text(text) { + return AgentResponse::Err { message }; + } + let path = crate::coordinator::Coordinator::agent_notes_dir(agent) + .join("hyperhive-status"); + let result = if text.trim().is_empty() { + // Empty = clear: remove the file (ignore missing). + std::fs::remove_file(&path) + .or_else(|e| if e.kind() == std::io::ErrorKind::NotFound { + Ok(()) + } else { + Err(e) + }) + } else { + std::fs::write(&path, format!("{}\n", text.trim())) + }; + match result { + Ok(()) => { + // Kick a container rescan so the dashboard updates live. + let coord2 = Arc::clone(coord); + tokio::spawn(async move { coord2.rescan_containers_and_emit().await }); + AgentResponse::Ok + } + Err(e) => AgentResponse::Err { message: format!("set_status write failed: {e}") }, + } + } + AgentRequest::GetAgentMeta { name } => { + let target = name.as_deref().unwrap_or(agent); + // Gate status on the target's running state so a stopped + // container's stale on-disk status doesn't leak through. + // Also surface `running` itself so callers can tell + // (e.g. "iris is down" vs "iris has no status set"). + let (status_text, status_set_at, running) = + crate::container_view::read_agent_status_live(target).await; + let role = if target == hive_sh4re::MANAGER_AGENT { + "manager" + } else { + "agent" + } + .to_owned(); + let (hive_name, swarm_name) = crate::container_view::hive_swarm_names(); + AgentResponse::AgentMeta { + name: target.to_owned(), + role, + running, + hyperhive_rev: crate::auto_update::current_flake_rev(&coord.hyperhive_flake), + status_text, + status_set_at, + hive_name, + swarm_name, + } + } + AgentRequest::CancelLooseEnd { kind, id } => crate::questions::handle_cancel_loose_end( + coord, agent, *kind, *id, + ) + .map_or_else( + |message| AgentResponse::Err { message }, + |()| AgentResponse::Ok, + ), + AgentRequest::AckTurn => match broker.ack_turn(agent) { + Ok(_n) => AgentResponse::Ok, + Err(e) => AgentResponse::Err { + message: format!("{e:#}"), + }, + }, + AgentRequest::RequeueInflight => match broker.requeue_inflight(agent) { + Ok(n) => { + if n > 0 { + tracing::info!(%agent, requeued = %n, "requeued in-flight messages"); + } + AgentResponse::Ok + } + Err(e) => AgentResponse::Err { + message: format!("{e:#}"), + }, + }, + AgentRequest::GetHostJournal { unit, container, lines, priority, grep, since, until } => { + dispatch_host_journal(agent, unit, container, lines, priority, grep, since, until).await + } // Manager-only variants are not valid on the agent socket. _ => AgentResponse::Err { message: "request not supported on agent socket".to_owned(), @@ -410,7 +394,7 @@ pub async fn dispatch_host_journal( /// Fan out one message to each recipient in `targets`. Skips the sender /// itself. Returns a list of `": "` strings for any delivery /// failures (empty = all good). -pub(crate) fn fan_out_send( +fn fan_out_send( coord: &Arc, from: &str, body: &str, @@ -437,8 +421,9 @@ pub(crate) fn fan_out_send( /// Common Send handler shared between dispatch arms. Applies the /// 4 KiB body cap, then routes broadcast (`to == "*"`) / children fan-out /// (`to == ""`) / unicast through their respective broker calls. -/// `pub(crate)` so `dispatch_shared` (and via it, `manager_server`) can use it. -pub(crate) fn handle_send( +/// Pulled out of `dispatch` to keep that function under the clippy +/// too-many-lines limit; the behaviour is identical to inlining. +fn handle_send( coord: &Arc, agent: &str, to: &str, diff --git a/hive-c0re/src/manager_server.rs b/hive-c0re/src/manager_server.rs index 2dec41d2..52e97786 100644 --- a/hive-c0re/src/manager_server.rs +++ b/hive-c0re/src/manager_server.rs @@ -6,7 +6,7 @@ use std::sync::Arc; use anyhow::{Context, Result}; -use hive_sh4re::{MANAGER_AGENT, ManagerRequest, ManagerResponse}; +use hive_sh4re::{MANAGER_AGENT, ManagerRequest, ManagerResponse, Message}; use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader}; use tokio::net::{UnixListener, UnixStream}; @@ -74,13 +74,123 @@ async fn serve(stream: UnixStream, coord: Arc) -> Result<()> { } } +/// Max long-poll window for manager `Recv`. Same semantics as the +/// sub-agent socket: omitted `wait_seconds` (or `0`) = peek and +/// return immediately, positive value = park up to that many +/// seconds (clamped at MAX). +const MANAGER_RECV_LONG_POLL_MAX: std::time::Duration = std::time::Duration::from_secs(180); + +/// Same shape + rationale as `agent_server::RECV_BATCH_MAX`. Kept +/// numerically aligned across surfaces so a tool description that +/// quotes the cap stays accurate either way. +const MANAGER_RECV_BATCH_MAX: u32 = 32; + +fn manager_recv_timeout(wait_seconds: Option) -> std::time::Duration { + match wait_seconds { + Some(s) => std::time::Duration::from_secs(s).min(MANAGER_RECV_LONG_POLL_MAX), + None => std::time::Duration::ZERO, + } +} + #[allow(clippy::too_many_lines)] async fn dispatch(req: &ManagerRequest, coord: &Arc) -> ManagerResponse { - // Delegate all variants shared with the agent socket to the common handler. - if let Some(resp) = crate::agent_server::dispatch_shared(req, MANAGER_AGENT, coord).await { - return resp; - } match req { + ManagerRequest::Send { + to, + body, + in_reply_to, + } => { + if let Err(message) = crate::limits::check_size("send", body) { + return ManagerResponse::Err { message }; + } + if to == "*" { + let errors = coord.broadcast_send(MANAGER_AGENT, body); + if errors.is_empty() { + ManagerResponse::Ok + } else { + ManagerResponse::Err { + message: format!("broadcast failed for agents: {}", errors.join(", ")), + } + } + } else { + // Resolve magic-recipient sentinels (currently ``) + // against topology.json; no-op for ordinary names. The + // manager has no parent in topology, so `` + // resolves to OPERATOR_RECIPIENT — the "no parent → tell + // the operator" fallback. See `docs/conventions.md:: + // Recipient sentinels`. + let resolved = crate::topology::resolve_recipient(MANAGER_AGENT, to); + match coord.broker.send(&Message { + from: MANAGER_AGENT.to_owned(), + to: resolved, + body: body.clone(), + in_reply_to: *in_reply_to, + }) { + Ok(()) => ManagerResponse::Ok, + Err(e) => ManagerResponse::Err { + message: format!("{e:#}"), + }, + } + } + } + ManagerRequest::Wake { from, body } => match coord.broker.send(&Message { + from: from.clone(), + to: MANAGER_AGENT.to_owned(), + body: body.clone(), + in_reply_to: None, + }) { + Ok(()) => ManagerResponse::Ok, + Err(e) => ManagerResponse::Err { + message: format!("{e:#}"), + }, + }, + ManagerRequest::OperatorMsg { body } => match coord.broker.send(&Message { + from: hive_sh4re::OPERATOR_RECIPIENT.to_owned(), + to: MANAGER_AGENT.to_owned(), + body: body.clone(), + in_reply_to: None, + }) { + Ok(()) => ManagerResponse::Ok, + Err(e) => ManagerResponse::Err { + message: format!("{e:#}"), + }, + }, + ManagerRequest::Status => match coord.broker.count_pending(MANAGER_AGENT) { + Ok(unread) => ManagerResponse::Status { unread }, + Err(e) => ManagerResponse::Err { + message: format!("{e:#}"), + }, + }, + ManagerRequest::Recent { limit } => match coord.broker.recent_for(MANAGER_AGENT, *limit) { + Ok(rows) => ManagerResponse::Recent { rows }, + Err(e) => ManagerResponse::Err { + message: format!("{e:#}"), + }, + }, + ManagerRequest::Recv { wait_seconds, max } => { + let cap = max.unwrap_or(1).min(MANAGER_RECV_BATCH_MAX) as usize; + match coord + .broker + .recv_blocking_batch(MANAGER_AGENT, manager_recv_timeout(*wait_seconds), cap) + .await + { + Ok(deliveries) => ManagerResponse::Messages { + messages: deliveries + .into_iter() + .map(|d| hive_sh4re::DeliveredMessage { + from: d.message.from, + body: d.message.body, + id: d.id, + redelivered: d.redelivered, + in_reply_to: d.message.in_reply_to, + }) + .collect(), + }, + Err(e) => ManagerResponse::Err { + message: format!("{e:#}"), + }, + } + } ManagerRequest::RequestInitConfig { name, description } => { tracing::info!(%name, "manager: request_init_config"); let proposed_dir = crate::coordinator::Coordinator::agent_proposed_dir(name); @@ -265,6 +375,31 @@ async fn dispatch(req: &ManagerRequest, coord: &Arc) -> ManagerResp ManagerRequest::FireScheduleNow { id } => { handle_fire_schedule_now(coord, hive_sh4re::MANAGER_AGENT, *id).await } + ManagerRequest::Ask { + question, + options, + multi, + ttl_seconds, + to, + } => crate::questions::handle_ask( + coord, + MANAGER_AGENT, + question, + options, + *multi, + *ttl_seconds, + to.as_deref(), + ) + .map_or_else( + |message| ManagerResponse::Err { message }, + |id| ManagerResponse::QuestionQueued { id }, + ), + ManagerRequest::Answer { id, answer } => { + crate::questions::handle_answer(coord, MANAGER_AGENT, *id, answer).map_or_else( + |message| ManagerResponse::Err { message }, + |()| ManagerResponse::Ok, + ) + } ManagerRequest::GetLogs { agent, lines } => { let n = lines.unwrap_or(50); // `journalctl -M` wants the *machine* name, not the @@ -304,6 +439,20 @@ async fn dispatch(req: &ManagerRequest, coord: &Arc) -> ManagerResp }, } } + ManagerRequest::Remind { + message, + timing, + file_path, + } => match crate::agent_server::store_remind( + coord, + MANAGER_AGENT, + message, + timing, + file_path.as_deref(), + ) { + Ok(()) => ManagerResponse::Ok, + Err(message) => ManagerResponse::Err { message }, + }, ManagerRequest::RequestApplyCommit { agent, commit_ref, @@ -351,9 +500,94 @@ async fn dispatch(req: &ManagerRequest, coord: &Arc) -> ManagerResp }, } } - _ => ManagerResponse::Err { - message: "request not handled on manager socket".to_owned(), + ManagerRequest::SetStatus { text } => { + // Cap length + reject multi-line so a confused caller + // can't dump a multi-paragraph session report into the + // dashboard chip. + if let Err(message) = crate::limits::check_status_text(text) { + return ManagerResponse::Err { message }; + } + let path = Coordinator::agent_notes_dir(MANAGER_AGENT).join("hyperhive-status"); + let result = if text.trim().is_empty() { + std::fs::remove_file(&path).or_else(|e| { + if e.kind() == std::io::ErrorKind::NotFound { + Ok(()) + } else { + Err(e) + } + }) + } else { + std::fs::write(&path, format!("{}\n", text.trim())) + }; + match result { + Ok(()) => { + let coord2 = Arc::clone(coord); + tokio::spawn(async move { coord2.rescan_containers_and_emit().await }); + ManagerResponse::Ok + } + Err(e) => ManagerResponse::Err { + message: format!("set_status write failed: {e}"), + }, + } + } + ManagerRequest::GetAgentMeta { name } => { + let target = name.as_deref().unwrap_or(MANAGER_AGENT); + // Gate status on the target's running state so a stopped + // container's stale on-disk status doesn't leak through. + // Also surface `running` itself so callers can tell + // (e.g. "iris is down" vs "iris has no status set"). + let (status_text, status_set_at, running) = + crate::container_view::read_agent_status_live(target).await; + let role = if target == MANAGER_AGENT { + "manager" + } else { + "agent" + } + .to_owned(); + let (hive_name, swarm_name) = crate::container_view::hive_swarm_names(); + ManagerResponse::AgentMeta { + name: target.to_owned(), + role, + running, + hyperhive_rev: crate::auto_update::current_flake_rev(&coord.hyperhive_flake), + status_text, + status_set_at, + hive_name, + swarm_name, + } + } + ManagerRequest::CancelLooseEnd { kind, id } => { + crate::questions::handle_cancel_loose_end(coord, MANAGER_AGENT, *kind, *id).map_or_else( + |message| ManagerResponse::Err { message }, + |()| ManagerResponse::Ok, + ) + } + ManagerRequest::AckTurn => match coord.broker.ack_turn(MANAGER_AGENT) { + Ok(_n) => ManagerResponse::Ok, + Err(e) => ManagerResponse::Err { + message: format!("{e:#}"), + }, }, + ManagerRequest::RequeueInflight => match coord.broker.requeue_inflight(MANAGER_AGENT) { + Ok(n) => { + if n > 0 { + tracing::info!(agent = %MANAGER_AGENT, requeued = %n, "requeued in-flight messages"); + } + ManagerResponse::Ok + } + Err(e) => ManagerResponse::Err { + message: format!("{e:#}"), + }, + }, + // GetHostJournal is an agent-socket-only capability-gated variant. + // The manager can use the existing GetLogs tool for per-container + // logs. Route to the agent_server handler for consistency. + ManagerRequest::GetHostJournal { unit, container, lines, priority, grep, since, until } => { + crate::agent_server::dispatch_host_journal( + MANAGER_AGENT, unit, container, lines, priority, grep, since, until, + ) + .await + } } }