diff --git a/hive-c0re/src/agent_server.rs b/hive-c0re/src/agent_server.rs index 030d01b9..1be7f9df 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`. -const RECV_LONG_POLL_MAX: std::time::Duration = std::time::Duration::from_secs(180); +pub(crate) 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,29 +104,40 @@ const RECV_LONG_POLL_MAX: std::time::Duration = std::time::Duration::from_secs(1 /// seen in practice (post-rebuild rescue, multi-agent reply storms) /// and well under the per-message `MESSAGE_MAX_BYTES` * N envelope /// budget. -const RECV_BATCH_MAX: u32 = 32; +pub(crate) const RECV_BATCH_MAX: u32 = 32; -fn recv_timeout(wait_seconds: Option) -> std::time::Duration { +pub(crate) 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, } } -#[allow(clippy::too_many_lines)] -async fn dispatch(req: &AgentRequest, agent: &str, coord: &Arc) -> AgentResponse { +/// 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 { let broker = &coord.broker; - match req { - AgentRequest::Send { to, body, in_reply_to } => { + Some(match req { + hive_sh4re::Request::Send { to, body, in_reply_to } => { handle_send(coord, agent, to, body, *in_reply_to) } - AgentRequest::Recv { wait_seconds, max } => { + hive_sh4re::Request::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) => AgentResponse::Messages { + Ok(deliveries) => hive_sh4re::Response::Messages { messages: deliveries .into_iter() .map(|d| hive_sh4re::DeliveredMessage { @@ -138,46 +149,46 @@ async fn dispatch(req: &AgentRequest, agent: &str, coord: &Arc) -> }) .collect(), }, - Err(e) => AgentResponse::Err { + Err(e) => hive_sh4re::Response::Err { message: format!("{e:#}"), }, } } - AgentRequest::Status => match broker.count_pending(agent) { - Ok(unread) => AgentResponse::Status { unread }, - Err(e) => AgentResponse::Err { + hive_sh4re::Request::Status => match broker.count_pending(agent) { + Ok(unread) => hive_sh4re::Response::Status { unread }, + Err(e) => hive_sh4re::Response::Err { message: format!("{e:#}"), }, }, - AgentRequest::OperatorMsg { body } => match broker.send(&Message { + hive_sh4re::Request::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(()) => AgentResponse::Ok, - Err(e) => AgentResponse::Err { + Ok(()) => hive_sh4re::Response::Ok, + Err(e) => hive_sh4re::Response::Err { message: format!("{e:#}"), }, }, - AgentRequest::Wake { from, body } => match broker.send(&Message { + hive_sh4re::Request::Wake { from, body } => match broker.send(&Message { from: from.clone(), to: agent.to_owned(), body: body.clone(), in_reply_to: None, }) { - Ok(()) => AgentResponse::Ok, - Err(e) => AgentResponse::Err { + Ok(()) => hive_sh4re::Response::Ok, + Err(e) => hive_sh4re::Response::Err { message: format!("{e:#}"), }, }, - AgentRequest::Recent { limit } => match broker.recent_for(agent, *limit) { - Ok(rows) => AgentResponse::Recent { rows }, - Err(e) => AgentResponse::Err { + hive_sh4re::Request::Recent { limit } => match broker.recent_for(agent, *limit) { + Ok(rows) => hive_sh4re::Response::Recent { rows }, + Err(e) => hive_sh4re::Response::Err { message: format!("{e:#}"), }, }, - AgentRequest::Ask { + hive_sh4re::Request::Ask { question, options, multi, @@ -193,21 +204,108 @@ async fn dispatch(req: &AgentRequest, agent: &str, coord: &Arc) -> to.as_deref(), ) .map_or_else( - |message| AgentResponse::Err { message }, - |id| AgentResponse::QuestionQueued { id }, + |message| hive_sh4re::Response::Err { message }, + |id| hive_sh4re::Response::QuestionQueued { id }, ), - AgentRequest::Answer { id, answer } => crate::questions::handle_answer( - coord, agent, *id, answer, - ) - .map_or_else( - |message| AgentResponse::Err { message }, - |()| AgentResponse::Ok, - ), - AgentRequest::Remind { + 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 { 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 { @@ -230,89 +328,6 @@ 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(), @@ -394,7 +409,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). -fn fan_out_send( +pub(crate) fn fan_out_send( coord: &Arc, from: &str, body: &str, @@ -421,9 +436,8 @@ 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. -/// Pulled out of `dispatch` to keep that function under the clippy -/// too-many-lines limit; the behaviour is identical to inlining. -fn handle_send( +/// `pub(crate)` so `dispatch_shared` (and via it, `manager_server`) can use it. +pub(crate) 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 52e97786..2dec41d2 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, Message}; +use hive_sh4re::{MANAGER_AGENT, ManagerRequest, ManagerResponse}; use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader}; use tokio::net::{UnixListener, UnixStream}; @@ -74,123 +74,13 @@ 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); @@ -375,31 +265,6 @@ 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 @@ -439,20 +304,6 @@ 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, @@ -500,94 +351,9 @@ async fn dispatch(req: &ManagerRequest, coord: &Arc) -> ManagerResp }, } } - 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:#}"), - }, + _ => ManagerResponse::Err { + message: "request not handled on manager socket".to_owned(), }, - 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 - } } }