diff --git a/hive-ag3nt/src/bin/hive.rs b/hive-ag3nt/src/bin/hive.rs index cb9749dc..873672ad 100644 --- a/hive-ag3nt/src/bin/hive.rs +++ b/hive-ag3nt/src/bin/hive.rs @@ -11,6 +11,16 @@ //! //! `HIVE_ROLE` defaults to `"agent"` if unset to keep the standalone //! `nix run .#hive-ag3nt` shape working without env plumbing. +//! +//! Post-#692 the entire turn loop (`serve_main` / `serve_loop` / +//! `handle_turn` / `wake`) is one generic implementation parameterised +//! by a `Surface` trait. Two zero-sized impls (`AgentSurface` / +//! `ManagerSurface`) wrap the per-role wire enums + boot-time defaults +//! (label fallback, MCP flavor, plugins arg, forge-notify mode); the +//! turn logic itself is written exactly once. Lets `main`'s dispatch +//! pick between `serve_main::` and +//! `serve_main::` and keep both code paths in lockstep +//! by construction. use std::path::{Path, PathBuf}; use std::sync::{Arc, Mutex}; @@ -97,19 +107,29 @@ async fn main() -> Result<()> { let cli = Cli::parse(); let role = resolve_role()?; + // Generic dispatch (#692): one `serve_main` / `wake` body, two + // monomorphisations driven by the `Surface` type parameter. The + // wire-type-disjoint enums live behind the trait methods; the + // turn loop itself is identical regardless of role. match (role, cli.cmd) { - (Role::Agent, Cmd::Serve { poll_ms }) => agent_serve_main(&cli.socket, poll_ms).await, - (Role::Manager, Cmd::Serve { poll_ms }) => manager_serve_main(&cli.socket, poll_ms).await, + (Role::Agent, Cmd::Serve { poll_ms }) => { + serve_main::(&cli.socket, poll_ms).await + } + (Role::Manager, Cmd::Serve { poll_ms }) => { + serve_main::(&cli.socket, poll_ms).await + } (Role::Agent, Cmd::Mcp) => mcp::serve_agent_stdio(cli.socket).await, (Role::Manager, Cmd::Mcp) => mcp::serve_manager_stdio(cli.socket).await, - (Role::Agent, Cmd::Wake { from, body }) => agent_wake(&cli.socket, from, body).await, + (Role::Agent, Cmd::Wake { from, body }) => { + wake::(&cli.socket, from, body).await + } (Role::Manager, Cmd::Wake { from, body }) => { - manager_wake(&cli.socket, from, body).await + wake::(&cli.socket, from, body).await } } } -// ---------- shared turn helpers (#692) ---------- +// ---------- shared turn helpers ---------- /// Surface a `SYSTEM_SENDER` message in the live event bus + tracing /// log. Was manager-only pre-#692; agents receive `QuestionAnswered`, @@ -158,14 +178,390 @@ fn consume_continue_sentinel() -> bool { true } -// ---------- agent role ---------- +// ---------- surface trait (#692) ---------- -async fn agent_serve_main(socket: &Path, poll_ms: u64) -> Result<()> { +/// 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. +enum RecvOutcome { + /// Long-poll returned at least one message; first one is detached. + Message(hive_sh4re::DeliveredMessage), + /// Long-poll timed out cleanly (empty `Messages` response). Caller + /// sleeps then retries. + Empty, + /// Wire returned an error / unexpected variant. Caller logs + + /// retries; the surface impl is responsible for tracing the + /// detail before returning this. + TransportError, +} + +/// Per-role wire surface. Two impls — `AgentSurface`, `ManagerSurface` +/// — wrap the disjoint `Request`/`Response` enums plus a handful of +/// boot-time constants that vary by role. Every other function in this +/// binary that talks to the broker goes through this trait so the turn +/// loop itself has zero per-role branches. +trait Surface { + /// MCP flavor passed to `TurnFiles::prepare`. Picks which static + /// system-prompt block + tool registration goes into the spawned + /// `claude` process. + const FLAVOR: mcp::Flavor; + /// Fallback agent label when `HIVE_LABEL` isn't in the environment. + /// Real deploys always set the env var; the fallback covers + /// standalone `nix run .#hive` invocations. + const DEFAULT_LABEL: &'static str; + /// First-arg to `plugins::install_configured`. `Some("manager")` + /// on sub-agents pulls the manager's plugin allowlist; `None` on + /// the manager loads its own set. + const PLUGINS_PARENT: Option<&'static str>; + /// First-arg to `forge_notify::run`. `true` switches the notifier + /// to mentions-only mode (manager); `false` keeps the full subscribe + /// firehose (sub-agents). + const FORGE_MENTIONS_ONLY: 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; + + /// Requeue any messages that were "in-flight" (delivered but not + /// ack'd) — fires on harness boot to recover from a crash mid-turn. + fn requeue_inflight(socket: &Path) -> impl Future; + + /// Current inbox unread count via `Status`. Returns 0 on any + /// transport/wire error so the caller falls through cleanly. + fn inbox_unread(socket: &Path) -> impl Future; + + /// `(open_threads, open_reminders)` for the post-turn stats row. + /// Either field is `None` when the underlying request errors. + fn post_turn_counts( + socket: &Path, + ) -> impl Future, Option)>; + + /// Send a message addressed to `` (broker resolves the + /// sentinel via `topology::parent_of` at delivery time; root + /// agents/manager fall through to operator). + fn send_to_parent(socket: &Path, body: String) -> impl Future; + + /// Fire a `Wake { from: "self", body: "continue" }` at our own + /// inbox — the request_next_turn sentinel pickup. + fn self_wake(socket: &Path) -> impl Future; + + /// Long-poll the broker for the next message. Wraps the + /// `Messages`/empty/error trichotomy in `RecvOutcome` so the + /// generic `serve_loop` doesn't need the per-role Response enum + /// at all. + fn recv_next(socket: &Path) -> impl Future; + + /// External `wake` subcommand (the `hive wake` CLI command, used + /// by co-process daemons like matrix to push events into the + /// harness inbox). Errors out via `anyhow::bail!` so the calling + /// binary surfaces them on stderr. + fn wake_external( + socket: &Path, + from: String, + body: String, + ) -> impl Future>; +} + +// ---------- AgentSurface (#692) ---------- + +/// Zero-sized type tag for the sub-agent wire surface. +/// Talks `AgentRequest` / `AgentResponse`. +struct AgentSurface; + +impl Surface for AgentSurface { + const FLAVOR: mcp::Flavor = mcp::Flavor::Agent; + const DEFAULT_LABEL: &'static str = "hive-ag3nt"; + const PLUGINS_PARENT: Option<&'static str> = Some("manager"); + const FORGE_MENTIONS_ONLY: bool = false; + + async fn ack_turn(socket: &Path) { + match client::request::<_, AgentResponse>(socket, &AgentRequest::AckTurn).await { + Ok(AgentResponse::Ok) => {} + Ok(AgentResponse::Err { message }) => { + tracing::warn!(%message, "ack_turn rejected by broker"); + } + Ok(other) => tracing::warn!(?other, "ack_turn unexpected response"), + Err(e) => tracing::warn!(error = ?e, "ack_turn transport error"), + } + } + + async fn requeue_inflight(socket: &Path) { + match client::request::<_, AgentResponse>(socket, &AgentRequest::RequeueInflight).await { + Ok(AgentResponse::Ok) => {} + Ok(AgentResponse::Err { message }) => { + tracing::warn!(%message, "requeue_inflight rejected by broker"); + } + Ok(other) => tracing::warn!(?other, "requeue_inflight unexpected response"), + Err(e) => tracing::warn!(error = ?e, "requeue_inflight transport error"), + } + } + + async fn inbox_unread(socket: &Path) -> u64 { + match client::request::<_, AgentResponse>(socket, &AgentRequest::Status).await { + Ok(AgentResponse::Status { unread }) => unread, + _ => 0, + } + } + + async fn post_turn_counts(socket: &Path) -> (Option, Option) { + let threads = + match client::request::<_, AgentResponse>(socket, &AgentRequest::GetLooseEnds).await { + Ok(AgentResponse::LooseEnds { loose_ends }) => { + u64::try_from(loose_ends.len()).ok() + } + _ => None, + }; + let reminders = match client::request::<_, AgentResponse>( + socket, + &AgentRequest::CountPendingReminders, + ) + .await + { + Ok(AgentResponse::PendingRemindersCount { count }) => Some(count), + _ => None, + }; + (threads, reminders) + } + + async fn send_to_parent(socket: &Path, body: String) { + let res = client::request::<_, AgentResponse>( + socket, + &AgentRequest::Send { + to: hive_sh4re::PARENT_RECIPIENT.into(), + body, + in_reply_to: None, + }, + ) + .await; + if let Err(e) = res { + tracing::warn!(error = ?e, "failed to notify parent of turn failure"); + } + } + + async fn self_wake(socket: &Path) { + let res = client::request::<_, AgentResponse>( + socket, + &AgentRequest::Wake { + from: "self".into(), + body: "continue".into(), + }, + ) + .await; + match res { + Ok(AgentResponse::Ok) => { + tracing::info!("request_next_turn: injected self-continue wake"); + } + Ok(AgentResponse::Err { message }) => { + tracing::warn!(%message, "check_and_inject_continue: wake rejected"); + } + Err(e) => { + tracing::warn!(error = ?e, "check_and_inject_continue: wake transport error"); + } + _ => {} + } + } + + async fn recv_next(socket: &Path) -> RecvOutcome { + let recv: Result = client::request( + socket, + &AgentRequest::Recv { + wait_seconds: Some(180), + max: None, + }, + ) + .await; + match recv { + Ok(AgentResponse::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::Err { message }) => { + tracing::warn!(%message, "recv error"); + RecvOutcome::TransportError + } + Ok(other) => { + tracing::warn!(?other, "recv produced unexpected response kind"); + RecvOutcome::TransportError + } + Err(e) => { + tracing::warn!(error = ?e, "recv failed; retrying"); + RecvOutcome::TransportError + } + } + } + + async fn wake_external(socket: &Path, from: String, body: String) -> Result<()> { + let resp: AgentResponse = + client::request(socket, &AgentRequest::Wake { from, body }).await?; + match resp { + AgentResponse::Ok => Ok(()), + AgentResponse::Err { message } => anyhow::bail!("wake: {message}"), + other => anyhow::bail!("wake: unexpected response {other:?}"), + } + } +} + +// ---------- ManagerSurface (#692) ---------- + +/// Zero-sized type tag for the manager wire surface. +/// Talks `ManagerRequest` / `ManagerResponse`. +struct ManagerSurface; + +impl Surface for ManagerSurface { + const FLAVOR: mcp::Flavor = mcp::Flavor::Manager; + const DEFAULT_LABEL: &'static str = "hm1nd"; + /// Manager loads its own plugin allowlist — no parent to inherit from. + const PLUGINS_PARENT: Option<&'static str> = None; + /// Mentions-only forge notifications: manager doesn't want the + /// subscription/participation firehose (see #671's role-driven + /// `forge.skipNotifyReasons` default). + const FORGE_MENTIONS_ONLY: bool = true; + + async fn ack_turn(socket: &Path) { + match client::request::<_, ManagerResponse>(socket, &ManagerRequest::AckTurn).await { + Ok(ManagerResponse::Ok) => {} + Ok(ManagerResponse::Err { message }) => { + tracing::warn!(%message, "ack_turn rejected by broker"); + } + Ok(other) => tracing::warn!(?other, "ack_turn unexpected response"), + Err(e) => tracing::warn!(error = ?e, "ack_turn transport error"), + } + } + + async fn requeue_inflight(socket: &Path) { + match client::request::<_, ManagerResponse>(socket, &ManagerRequest::RequeueInflight).await + { + Ok(ManagerResponse::Ok) => {} + Ok(ManagerResponse::Err { message }) => { + tracing::warn!(%message, "requeue_inflight rejected by broker"); + } + Ok(other) => tracing::warn!(?other, "requeue_inflight unexpected response"), + Err(e) => tracing::warn!(error = ?e, "requeue_inflight transport error"), + } + } + + async fn inbox_unread(socket: &Path) -> u64 { + match client::request::<_, ManagerResponse>(socket, &ManagerRequest::Status).await { + Ok(ManagerResponse::Status { unread }) => unread, + _ => 0, + } + } + + async fn post_turn_counts(socket: &Path) -> (Option, Option) { + let threads = match client::request::<_, ManagerResponse>( + socket, + &ManagerRequest::GetLooseEnds { agent: None }, + ) + .await + { + Ok(ManagerResponse::LooseEnds { loose_ends }) => { + u64::try_from(loose_ends.len()).ok() + } + _ => None, + }; + let reminders = match client::request::<_, ManagerResponse>( + socket, + &ManagerRequest::CountPendingReminders { agent: None }, + ) + .await + { + Ok(ManagerResponse::PendingRemindersCount { count }) => Some(count), + _ => None, + }; + (threads, reminders) + } + + async fn send_to_parent(socket: &Path, body: String) { + let res = client::request::<_, ManagerResponse>( + socket, + &ManagerRequest::Send { + to: hive_sh4re::PARENT_RECIPIENT.into(), + body, + in_reply_to: None, + }, + ) + .await; + if let Err(e) = res { + tracing::warn!(error = ?e, "failed to notify parent of turn failure"); + } + } + + async fn self_wake(socket: &Path) { + let res = client::request::<_, ManagerResponse>( + socket, + &ManagerRequest::Wake { + from: "self".into(), + body: "continue".into(), + }, + ) + .await; + match res { + Ok(ManagerResponse::Ok) => { + tracing::info!("request_next_turn: injected self-continue wake"); + } + Ok(ManagerResponse::Err { message }) => { + tracing::warn!(%message, "check_and_inject_continue: wake rejected"); + } + Err(e) => { + tracing::warn!(error = ?e, "check_and_inject_continue: wake transport error"); + } + _ => {} + } + } + + async fn recv_next(socket: &Path) -> RecvOutcome { + let recv: Result = client::request( + socket, + &ManagerRequest::Recv { + wait_seconds: Some(180), + max: None, + }, + ) + .await; + match recv { + Ok(ManagerResponse::Messages { messages }) if !messages.is_empty() => { + let first = messages.into_iter().next().expect("checked non-empty"); + RecvOutcome::Message(first) + } + Ok(ManagerResponse::Messages { .. }) => RecvOutcome::Empty, + Ok(ManagerResponse::Err { message }) => { + tracing::warn!(%message, "recv error"); + RecvOutcome::TransportError + } + Ok(other) => { + tracing::warn!(?other, "recv produced unexpected response kind"); + RecvOutcome::TransportError + } + Err(e) => { + tracing::warn!(error = ?e, "recv failed; retrying"); + RecvOutcome::TransportError + } + } + } + + async fn wake_external(socket: &Path, from: String, body: String) -> Result<()> { + let resp: ManagerResponse = + client::request(socket, &ManagerRequest::Wake { from, body }).await?; + match resp { + ManagerResponse::Ok => Ok(()), + ManagerResponse::Err { message } => anyhow::bail!("wake: {message}"), + other => anyhow::bail!("wake: unexpected response {other:?}"), + } + } +} + +// ---------- generic turn loop (#692) ---------- + +/// Per-role boot — wires up the web UI, login state, stats, plugins, +/// forge notifier, and either drops into `serve_loop` directly +/// (`Online`) or parks on the login flow first (`NeedsLogin`). +async fn serve_main(socket: &Path, poll_ms: u64) -> Result<()> { let port = std::env::var("HIVE_PORT") .ok() .and_then(|s| s.parse::().ok()) .unwrap_or(DEFAULT_WEB_PORT); - let label = std::env::var("HIVE_LABEL").unwrap_or_else(|_| "hive-ag3nt".into()); + let label = std::env::var("HIVE_LABEL").unwrap_or_else(|_| S::DEFAULT_LABEL.into()); let claude_dir = login::default_dir(); let initial = LoginState::from_dir(&claude_dir); tracing::info!(state = ?initial, claude_dir = %claude_dir.display(), "harness boot"); @@ -178,12 +574,15 @@ async fn agent_serve_main(socket: &Path, poll_ms: u64) -> Result<()> { bus.seed_usage(ctx, cost); } } - let files = turn::TurnFiles::prepare(socket, &label, mcp::Flavor::Agent).await?; + let files = turn::TurnFiles::prepare(socket, &label, S::FLAVOR).await?; let turn_lock: TurnLock = Arc::new(tokio::sync::Mutex::new(())); - plugins::install_configured(socket, Some("manager")).await; - tokio::spawn(hive_ag3nt::forge_notify::run(socket.to_path_buf(), false)); + plugins::install_configured(socket, S::PLUGINS_PARENT).await; + tokio::spawn(hive_ag3nt::forge_notify::run( + socket.to_path_buf(), + S::FORGE_MENTIONS_ONLY, + )); tokio::spawn(web_ui::serve( - label.clone(), + label, port, login_state.clone(), bus.clone(), @@ -191,42 +590,31 @@ async fn agent_serve_main(socket: &Path, poll_ms: u64) -> Result<()> { files.clone(), turn_lock.clone(), )); - match initial { - LoginState::Online => { - // Clear any stale `hyperhive-needs-login` sentinel left - // over from a prior boot (closes #682, see #688). - bus.emit_status("online"); - agent_serve_loop( - socket, - Duration::from_millis(poll_ms), - login_state, - claude_dir, - bus, - stats, - &files, - turn_lock, - ) - .await - } - LoginState::NeedsLogin => { - turn::wait_for_login(&claude_dir, login_state.clone(), &bus, poll_ms).await; - agent_serve_loop( - socket, - Duration::from_millis(poll_ms), - login_state, - claude_dir, - bus, - stats, - &files, - turn_lock, - ) - .await - } + if matches!(initial, LoginState::NeedsLogin) { + turn::wait_for_login(&claude_dir, login_state.clone(), &bus, poll_ms).await; + } else { + // Clear any stale `hyperhive-needs-login` sentinel left over + // from a prior boot (closes #682, see #688). + bus.emit_status("online"); } + serve_loop::( + socket, + Duration::from_millis(poll_ms), + login_state, + claude_dir, + bus, + stats, + &files, + turn_lock, + ) + .await } +/// The long-running message loop. Long-polls the broker via +/// `S::recv_next`, drives a turn per message, parks on auth-failed, +/// otherwise retries. #[allow(clippy::too_many_arguments)] -async fn agent_serve_loop( +async fn serve_loop( socket: &Path, interval: Duration, login_state: Arc>, @@ -236,23 +624,13 @@ async fn agent_serve_loop( files: &turn::TurnFiles, turn_lock: TurnLock, ) -> Result<()> { - tracing::info!(socket = %socket.display(), "hive-ag3nt serve"); - agent_requeue_inflight(socket).await; + tracing::info!(socket = %socket.display(), "harness serve"); + S::requeue_inflight(socket).await; loop { - let recv: Result = client::request( - socket, - &AgentRequest::Recv { - wait_seconds: Some(180), - max: None, - }, - ) - .await; - match recv { - Ok(AgentResponse::Messages { messages }) if !messages.is_empty() => { - let first = messages.into_iter().next().expect("checked non-empty"); + match S::recv_next(socket).await { + RecvOutcome::Message(first) => { let auth_failed = - handle_agent_turn(socket, &bus, stats.as_ref(), files, &turn_lock, first) - .await; + handle_turn::(socket, &bus, stats.as_ref(), files, &turn_lock, first).await; if auth_failed { *login_state.lock().unwrap() = LoginState::NeedsLogin; turn::wait_for_login( @@ -264,32 +642,23 @@ async fn agent_serve_loop( .await; } } - Ok(AgentResponse::Messages { .. }) => { + RecvOutcome::Empty => { tokio::time::sleep(interval).await; } - Ok( - AgentResponse::Ok - | AgentResponse::Status { .. } - | AgentResponse::Recent { .. } - | AgentResponse::QuestionQueued { .. } - | AgentResponse::LooseEnds { .. } - | AgentResponse::PendingRemindersCount { .. } - | AgentResponse::ReminderRollup { .. } - | AgentResponse::AgentMeta { .. }, - ) => { - tracing::warn!("recv produced unexpected response kind"); - } - Ok(AgentResponse::Err { message }) => { - tracing::warn!(%message, "recv error"); - } - Err(e) => { - tracing::warn!(error = ?e, "recv failed; retrying"); + RecvOutcome::TransportError => { + // `recv_next` already logged the detail; just retry. + // No backoff: the long-poll wait is itself the throttle. } } } } -async fn handle_agent_turn( +/// Drive a single turn: emit boot-of-turn events, run claude, ack on +/// success / requeue on rate-limit-or-401 / notify parent on failure, +/// record stats, then pick up the `request_next_turn` sentinel if it's +/// been dropped during the turn. Returns true iff the outcome was +/// `AuthFailed` — the caller flips the harness to needs-login. +async fn handle_turn( socket: &Path, bus: &Bus, stats: Option<&TurnStats>, @@ -302,7 +671,7 @@ async fn handle_agent_turn( let redelivered = first.redelivered; log_system_event(bus, &from, &body); tracing::info!(%from, %body, %redelivered, "inbox"); - let unread = agent_inbox_unread(socket).await; + let unread = S::inbox_unread(socket).await; bus.emit(LiveEvent::TurnStart { from: from.clone(), body: body.clone(), unread }); bus.set_state(TurnState::Thinking); let started_at = serve_common::now_unix(); @@ -316,7 +685,7 @@ async fn handle_agent_turn( turn::emit_turn_end(bus, &outcome); bus.set_state(TurnState::Idle); if matches!(outcome, turn::TurnOutcome::Ok | turn::TurnOutcome::Compacted) { - agent_ack_turn(socket).await; + S::ack_turn(socket).await; } if matches!(outcome, turn::TurnOutcome::RateLimited) { let secs = turn::rate_limit_sleep_secs(); @@ -326,7 +695,7 @@ async fn handle_agent_turn( }); tracing::warn!(sleep_secs = secs, "rate-limited; parking"); tokio::time::sleep(Duration::from_secs(secs)).await; - agent_requeue_inflight(socket).await; + S::requeue_inflight(socket).await; bus.emit_status("online"); } if matches!(outcome, turn::TurnOutcome::AuthFailed) { @@ -335,16 +704,16 @@ async fn handle_agent_turn( text: "API 401 — waiting for re-login via web UI".into(), }); tracing::warn!("auth-failed; parking until re-login"); - agent_requeue_inflight(socket).await; + S::requeue_inflight(socket).await; } if let turn::TurnOutcome::Failed(e) = &outcome { - agent_notify_parent_of_failure(socket, e).await; + S::send_to_parent(socket, format_turn_failure(e)).await; } if let Some(stats) = stats { let ended_at = serve_common::now_unix(); let duration_ms = i64::try_from(started_instant.elapsed().as_millis()).unwrap_or(i64::MAX); - let (open_threads, open_reminders) = agent_post_turn_counts(socket).await; + let (open_threads, open_reminders) = S::post_turn_counts(socket).await; let row = serve_common::build_row( started_at, ended_at, @@ -358,113 +727,20 @@ async fn handle_agent_turn( ); stats.record(&row); } - let pending = agent_inbox_unread(socket).await; + let pending = S::inbox_unread(socket).await; if pending > 0 { tracing::info!(%pending, "pending messages after turn; fetching next"); } - agent_check_and_inject_continue(socket).await; + if consume_continue_sentinel() { + S::self_wake(socket).await; + } matches!(outcome, turn::TurnOutcome::AuthFailed) } -async fn agent_ack_turn(socket: &Path) { - match client::request::<_, AgentResponse>(socket, &AgentRequest::AckTurn).await { - Ok(AgentResponse::Ok) => {} - Ok(AgentResponse::Err { message }) => { - tracing::warn!(%message, "ack_turn rejected by broker"); - } - Ok(other) => { - tracing::warn!(?other, "ack_turn unexpected response"); - } - Err(e) => tracing::warn!(error = ?e, "ack_turn transport error"), - } -} - -async fn agent_requeue_inflight(socket: &Path) { - match client::request::<_, AgentResponse>(socket, &AgentRequest::RequeueInflight).await { - Ok(AgentResponse::Ok) => {} - Ok(AgentResponse::Err { message }) => { - tracing::warn!(%message, "requeue_inflight rejected by broker"); - } - Ok(other) => { - tracing::warn!(?other, "requeue_inflight unexpected response"); - } - Err(e) => tracing::warn!(error = ?e, "requeue_inflight transport error"), - } -} - -/// Notify whoever's structurally watching this agent that the claude -/// turn failed. Pre-#692 this was `agent_notify_manager_of_failure` -/// targeting the literal string `"manager"`; now it routes through -/// the `` sentinel landed in #703 so failures bubble to the -/// real parent (and to operator for root agents). Body identity -/// comes from `format_turn_failure` — no `label` plumbing. -async fn agent_notify_parent_of_failure(socket: &Path, err: &anyhow::Error) { - let res = client::request::<_, AgentResponse>( - socket, - &AgentRequest::Send { - to: hive_sh4re::PARENT_RECIPIENT.into(), - body: format_turn_failure(err), - in_reply_to: None, - }, - ) - .await; - if let Err(e) = res { - tracing::warn!(error = ?e, "failed to notify parent of turn failure"); - } -} - -async fn agent_inbox_unread(socket: &Path) -> u64 { - match client::request::<_, AgentResponse>(socket, &AgentRequest::Status).await { - Ok(AgentResponse::Status { unread }) => unread, - _ => 0, - } -} - -async fn agent_post_turn_counts(socket: &Path) -> (Option, Option) { - let threads = - match client::request::<_, AgentResponse>(socket, &AgentRequest::GetLooseEnds).await { - Ok(AgentResponse::LooseEnds { loose_ends }) => u64::try_from(loose_ends.len()).ok(), - _ => None, - }; - let reminders = match client::request::<_, AgentResponse>( - socket, - &AgentRequest::CountPendingReminders, - ) - .await - { - Ok(AgentResponse::PendingRemindersCount { count }) => Some(count), - _ => None, - }; - (threads, reminders) -} - -async fn agent_check_and_inject_continue(socket: &Path) { - if !consume_continue_sentinel() { - return; - } - let res = client::request::<_, AgentResponse>( - socket, - &AgentRequest::Wake { - from: "self".into(), - body: "continue".into(), - }, - ) - .await; - match res { - Ok(AgentResponse::Ok) => { - tracing::info!("request_next_turn: injected self-continue wake"); - } - Ok(AgentResponse::Err { message }) => { - tracing::warn!(%message, "check_and_inject_continue: wake rejected"); - } - Err(e) => { - tracing::warn!(error = ?e, "check_and_inject_continue: wake transport error"); - } - _ => {} - } -} - -async fn agent_wake(socket: &Path, from: String, body: String) -> Result<()> { +/// External `hive wake` subcommand — push a message into our own +/// inbox so the next turn fires with the given body. Reads the body +/// from stdin when `body == "-"`. +async fn wake(socket: &Path, from: String, body: String) -> Result<()> { let body = if body == "-" { let mut buf = String::new(); std::io::Read::read_to_string(&mut std::io::stdin(), &mut buf)?; @@ -472,343 +748,5 @@ async fn agent_wake(socket: &Path, from: String, body: String) -> Result<()> { } else { body }; - let resp: AgentResponse = - client::request(socket, &AgentRequest::Wake { from, body }).await?; - match resp { - AgentResponse::Ok => Ok(()), - AgentResponse::Err { message } => anyhow::bail!("wake: {message}"), - other => anyhow::bail!("wake: unexpected response {other:?}"), - } -} - -// ---------- manager role ---------- - -async fn manager_serve_main(socket: &Path, poll_ms: u64) -> Result<()> { - let port = std::env::var("HIVE_PORT") - .ok() - .and_then(|s| s.parse::().ok()) - .unwrap_or(DEFAULT_WEB_PORT); - let label = std::env::var("HIVE_LABEL").unwrap_or_else(|_| "hm1nd".into()); - let claude_dir = login::default_dir(); - let initial = LoginState::from_dir(&claude_dir); - tracing::info!(state = ?initial, claude_dir = %claude_dir.display(), "hm1nd boot"); - let login_state = Arc::new(Mutex::new(initial)); - let bus = Bus::new(); - let stats = TurnStats::open_default(); - if let Some(s) = &stats { - let (ctx, cost) = s.last_usage(); - if ctx.is_some() || cost.is_some() { - bus.seed_usage(ctx, cost); - } - } - let files = turn::TurnFiles::prepare(socket, &label, mcp::Flavor::Manager).await?; - let turn_lock: TurnLock = Arc::new(tokio::sync::Mutex::new(())); - plugins::install_configured(socket, None).await; - tokio::spawn(web_ui::serve( - label, - port, - login_state.clone(), - bus.clone(), - socket.to_path_buf(), - files.clone(), - turn_lock.clone(), - )); - tokio::spawn(hive_ag3nt::forge_notify::run(socket.to_path_buf(), true)); - match initial { - LoginState::Online => { - // Clear any stale `hyperhive-needs-login` sentinel left - // over from a prior boot (closes #682, see #688). - bus.emit_status("online"); - manager_serve_loop( - socket, - Duration::from_millis(poll_ms), - login_state, - claude_dir, - bus, - stats, - &files, - turn_lock, - ) - .await - } - LoginState::NeedsLogin => { - turn::wait_for_login(&claude_dir, login_state.clone(), &bus, poll_ms).await; - manager_serve_loop( - socket, - Duration::from_millis(poll_ms), - login_state, - claude_dir, - bus, - stats, - &files, - turn_lock, - ) - .await - } - } -} - -#[allow(clippy::too_many_arguments)] -async fn manager_serve_loop( - socket: &Path, - interval: Duration, - login_state: Arc>, - claude_dir: std::path::PathBuf, - bus: Bus, - stats: Option, - files: &turn::TurnFiles, - turn_lock: TurnLock, -) -> Result<()> { - tracing::info!(socket = %socket.display(), "hive-m1nd serve"); - manager_requeue_inflight(socket).await; - loop { - let recv: Result = client::request( - socket, - &ManagerRequest::Recv { - wait_seconds: Some(180), - max: None, - }, - ) - .await; - match recv { - Ok(ManagerResponse::Messages { messages }) if !messages.is_empty() => { - let first = messages.into_iter().next().expect("checked non-empty"); - let auth_failed = - handle_manager_turn(socket, &bus, stats.as_ref(), files, &turn_lock, first) - .await; - if auth_failed { - *login_state.lock().unwrap() = LoginState::NeedsLogin; - turn::wait_for_login( - &claude_dir, - login_state.clone(), - &bus, - u64::try_from(interval.as_millis()).unwrap_or(2000), - ) - .await; - } - } - Ok(ManagerResponse::Messages { .. }) => { - tokio::time::sleep(interval).await; - } - Ok( - ManagerResponse::Ok - | ManagerResponse::Status { .. } - | ManagerResponse::QuestionQueued { .. } - | ManagerResponse::Recent { .. } - | ManagerResponse::Logs { .. } - | ManagerResponse::LooseEnds { .. } - | ManagerResponse::PendingRemindersCount { .. } - | ManagerResponse::ReminderRollup { .. } - | ManagerResponse::AgentMeta { .. } - | ManagerResponse::Schedules { .. }, - ) => { - tracing::warn!("recv produced unexpected response kind"); - } - Ok(ManagerResponse::Err { message }) => { - tracing::warn!(%message, "recv error"); - } - Err(e) => { - tracing::warn!(error = ?e, "recv failed; retrying"); - } - } - } -} - -async fn handle_manager_turn( - socket: &Path, - bus: &Bus, - stats: Option<&TurnStats>, - files: &turn::TurnFiles, - turn_lock: &TurnLock, - first: hive_sh4re::DeliveredMessage, -) -> bool { - let from = first.from; - let body = first.body; - let redelivered = first.redelivered; - log_system_event(bus, &from, &body); - tracing::info!(%from, %body, %redelivered, "manager inbox"); - let unread = manager_inbox_unread(socket).await; - bus.emit(LiveEvent::TurnStart { from: from.clone(), body: body.clone(), unread }); - let prompt = serve_common::format_wake_prompt(&from, &body, unread, redelivered); - bus.set_state(TurnState::Thinking); - let started_at = serve_common::now_unix(); - let started_instant = std::time::Instant::now(); - let model_at_start = bus.model(); - let outcome = { - let _guard = turn_lock.lock().await; - turn::drive_turn(&prompt, files, bus).await - }; - turn::emit_turn_end(bus, &outcome); - bus.set_state(TurnState::Idle); - if matches!(outcome, turn::TurnOutcome::Ok | turn::TurnOutcome::Compacted) { - manager_ack_turn(socket).await; - } - if matches!(outcome, turn::TurnOutcome::RateLimited) { - let secs = turn::rate_limit_sleep_secs(); - bus.emit_status("rate_limited"); - bus.emit(LiveEvent::Note { - text: format!("API rate-limited — sleeping {secs}s before retry"), - }); - tracing::warn!(sleep_secs = secs, "rate-limited; parking"); - tokio::time::sleep(Duration::from_secs(secs)).await; - manager_requeue_inflight(socket).await; - bus.emit_status("online"); - } - if matches!(outcome, turn::TurnOutcome::AuthFailed) { - bus.emit_status("needs_login_idle"); - bus.emit(LiveEvent::Note { - text: "API 401 — waiting for re-login via web UI".into(), - }); - tracing::warn!("auth-failed; parking until re-login"); - manager_requeue_inflight(socket).await; - } - if let turn::TurnOutcome::Failed(e) = &outcome { - manager_notify_parent_of_failure(socket, e).await; - } - if let Some(stats) = stats { - let ended_at = serve_common::now_unix(); - let duration_ms = - i64::try_from(started_instant.elapsed().as_millis()).unwrap_or(i64::MAX); - let (open_threads, open_reminders) = manager_post_turn_counts(socket).await; - let row = serve_common::build_row( - started_at, - ended_at, - duration_ms, - model_at_start, - from.clone(), - &outcome, - bus, - open_threads, - open_reminders, - ); - stats.record(&row); - } - let pending = manager_inbox_unread(socket).await; - if pending > 0 { - tracing::info!(%pending, "pending messages after turn; fetching next"); - } - manager_check_and_inject_continue(socket).await; - matches!(outcome, turn::TurnOutcome::AuthFailed) -} - -/// Manager mirror of `agent_notify_parent_of_failure`. For a root -/// manager (`topology::parent_of("manager")` is `None`) the -/// `` sentinel resolves to `operator`, which surfaces the -/// failure in the dashboard T4LK box — the only audience above -/// the manager that can act on it. -async fn manager_notify_parent_of_failure(socket: &Path, err: &anyhow::Error) { - let res = client::request::<_, ManagerResponse>( - socket, - &ManagerRequest::Send { - to: hive_sh4re::PARENT_RECIPIENT.into(), - body: format_turn_failure(err), - in_reply_to: None, - }, - ) - .await; - if let Err(e) = res { - tracing::warn!(error = ?e, "failed to notify parent of turn failure"); - } -} - -/// Manager mirror of `agent_check_and_inject_continue`. The -/// `request_next_turn` MCP tool drops the same sentinel from either -/// flavor; both harnesses pick it up the same way and fire their -/// role-specific `Wake` request. -async fn manager_check_and_inject_continue(socket: &Path) { - if !consume_continue_sentinel() { - return; - } - let res = client::request::<_, ManagerResponse>( - socket, - &ManagerRequest::Wake { - from: "self".into(), - body: "continue".into(), - }, - ) - .await; - match res { - Ok(ManagerResponse::Ok) => { - tracing::info!("request_next_turn: injected self-continue wake"); - } - Ok(ManagerResponse::Err { message }) => { - tracing::warn!(%message, "check_and_inject_continue: wake rejected"); - } - Err(e) => { - tracing::warn!(error = ?e, "check_and_inject_continue: wake transport error"); - } - _ => {} - } -} - -async fn manager_ack_turn(socket: &Path) { - match client::request::<_, ManagerResponse>(socket, &ManagerRequest::AckTurn).await { - Ok(ManagerResponse::Ok) => {} - Ok(ManagerResponse::Err { message }) => { - tracing::warn!(%message, "ack_turn rejected by broker"); - } - Ok(other) => { - tracing::warn!(?other, "ack_turn unexpected response"); - } - Err(e) => tracing::warn!(error = ?e, "ack_turn transport error"), - } -} - -async fn manager_requeue_inflight(socket: &Path) { - match client::request::<_, ManagerResponse>(socket, &ManagerRequest::RequeueInflight).await { - Ok(ManagerResponse::Ok) => {} - Ok(ManagerResponse::Err { message }) => { - tracing::warn!(%message, "requeue_inflight rejected by broker"); - } - Ok(other) => { - tracing::warn!(?other, "requeue_inflight unexpected response"); - } - Err(e) => tracing::warn!(error = ?e, "requeue_inflight transport error"), - } -} - -async fn manager_inbox_unread(socket: &Path) -> u64 { - match client::request::<_, ManagerResponse>(socket, &ManagerRequest::Status).await { - Ok(ManagerResponse::Status { unread }) => unread, - _ => 0, - } -} - -async fn manager_post_turn_counts(socket: &Path) -> (Option, Option) { - let threads = match client::request::<_, ManagerResponse>( - socket, - &ManagerRequest::GetLooseEnds { agent: None }, - ) - .await - { - Ok(ManagerResponse::LooseEnds { loose_ends }) => u64::try_from(loose_ends.len()).ok(), - _ => None, - }; - let reminders = match client::request::<_, ManagerResponse>( - socket, - &ManagerRequest::CountPendingReminders { agent: None }, - ) - .await - { - Ok(ManagerResponse::PendingRemindersCount { count }) => Some(count), - _ => None, - }; - (threads, reminders) -} - -async fn manager_wake(socket: &Path, from: String, body: String) -> Result<()> { - let body = if body == "-" { - let mut buf = String::new(); - std::io::Read::read_to_string(&mut std::io::stdin(), &mut buf)?; - buf - } else { - body - }; - let resp: ManagerResponse = - client::request(socket, &ManagerRequest::Wake { from, body }).await?; - match resp { - ManagerResponse::Ok => Ok(()), - ManagerResponse::Err { message } => anyhow::bail!("wake: {message}"), - other => anyhow::bail!("wake: unexpected response {other:?}"), - } + S::wake_external(socket, from, body).await }