From dcaf1838e164937ad5f6e1d68cc530e0c00e1992 Mon Sep 17 00:00:00 2001 From: damocles Date: Sun, 31 May 2026 03:21:06 +0200 Subject: [PATCH] harness: unify hive-ag3nt + hive-m1nd into one binary picking role from HIVE_ROLE (#598) --- hive-ag3nt/Cargo.toml | 14 +- hive-ag3nt/src/bin/hive-ag3nt.rs | 474 --------------------- hive-ag3nt/src/bin/hive-m1nd.rs | 358 ---------------- hive-ag3nt/src/bin/hive.rs | 698 +++++++++++++++++++++++++++++++ hive-ag3nt/src/turn.rs | 2 +- nix/templates/harness-base.nix | 11 +- 6 files changed, 717 insertions(+), 840 deletions(-) delete mode 100644 hive-ag3nt/src/bin/hive-ag3nt.rs delete mode 100644 hive-ag3nt/src/bin/hive-m1nd.rs create mode 100644 hive-ag3nt/src/bin/hive.rs diff --git a/hive-ag3nt/Cargo.toml b/hive-ag3nt/Cargo.toml index 2446dec4..bc33727c 100644 --- a/hive-ag3nt/Cargo.toml +++ b/hive-ag3nt/Cargo.toml @@ -28,9 +28,11 @@ tracing-subscriber.workspace = true tempfile = "3" [[bin]] -name = "hive-ag3nt" -path = "src/bin/hive-ag3nt.rs" - -[[bin]] -name = "hive-m1nd" -path = "src/bin/hive-m1nd.rs" +# Unified harness binary (#598). Replaces the pre-#598 split into +# `hive-ag3nt` (sub-agent) + `hive-m1nd` (manager). Both code paths +# live here; the binary picks its role at startup from `HIVE_ROLE` +# (set by `harness-base.nix` from `hyperhive.role` — `"agent"` or +# `"manager"`). The privilege boundary is enforced server-side at +# the socket, so shipping both surfaces in one binary is safe. +name = "hive" +path = "src/bin/hive.rs" diff --git a/hive-ag3nt/src/bin/hive-ag3nt.rs b/hive-ag3nt/src/bin/hive-ag3nt.rs deleted file mode 100644 index 22e196cb..00000000 --- a/hive-ag3nt/src/bin/hive-ag3nt.rs +++ /dev/null @@ -1,474 +0,0 @@ -use std::path::{Path, PathBuf}; -use std::sync::{Arc, Mutex}; -use std::time::Duration; - -use hive_ag3nt::web_ui::TurnLock; - -use anyhow::Result; -use clap::{Parser, Subcommand}; -use hive_ag3nt::events::{Bus, LiveEvent, TurnState}; -use hive_ag3nt::login::{self, LoginState}; -use hive_ag3nt::turn_stats::TurnStats; -use hive_ag3nt::{DEFAULT_SOCKET, DEFAULT_WEB_PORT, client, mcp, plugins, serve_common, turn, web_ui}; -use hive_sh4re::{AgentRequest, AgentResponse}; - -#[derive(Parser)] -#[command(name = "hive-ag3nt", about = "hyperhive sub-agent harness")] -struct Cli { - /// Path to the per-agent MCP socket (bind-mounted from the host). - #[arg(long, global = true, default_value = DEFAULT_SOCKET)] - socket: PathBuf, - - #[command(subcommand)] - cmd: Cmd, -} - -#[derive(Subcommand)] -enum Cmd { - /// Run the long-lived harness loop. Polls inbox; replies via `claude --print` - /// when available, falling back to a simple echo otherwise. - Serve { - /// Inbox poll interval in milliseconds. - #[arg(long, default_value_t = 1000)] - poll_ms: u64, - }, - /// Run the agent's MCP server on stdio. Spawned by `claude` via - /// `--mcp-config`; tools dispatch through `/run/hive/mcp.sock` back into - /// the hyperhive broker. - Mcp, - /// Inject a wake-up event into this agent's inbox so the next turn - /// fires with the given body. Intended for extra MCP servers / - /// helpers running inside the container (matrix bridge, scraper, - /// webhook listener) that need to nudge claude on external events. - /// `from` is the sender label that appears in the wake prompt - /// (claude sees "from: matrix" etc.). - Wake { - #[arg(long)] - from: String, - /// Body of the wake message. Pass `-` to read from stdin. - #[arg(long)] - body: String, - }, -} - -#[tokio::main] -async fn main() -> Result<()> { - tracing_subscriber::fmt() - .with_env_filter( - tracing_subscriber::EnvFilter::try_from_default_env() - .unwrap_or_else(|_| tracing_subscriber::EnvFilter::new("info")), - ) - .init(); - - let cli = Cli::parse(); - match cli.cmd { - Cmd::Serve { poll_ms } => { - 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 claude_dir = login::default_dir(); - let initial = LoginState::from_dir(&claude_dir); - tracing::info!(state = ?initial, claude_dir = %claude_dir.display(), "harness 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(&cli.socket, &label, mcp::Flavor::Agent).await?; - let turn_lock: TurnLock = Arc::new(tokio::sync::Mutex::new(())); - plugins::install_configured(&cli.socket, Some("manager")).await; - tokio::spawn(hive_ag3nt::forge_notify::run(cli.socket.clone(), false)); - tokio::spawn(web_ui::serve( - label.clone(), - port, - login_state.clone(), - bus.clone(), - cli.socket.clone(), - files.clone(), - turn_lock.clone(), - )); - match initial { - LoginState::Online => { - // #682: clear any stale `hyperhive-needs-login` - // sentinel from a prior boot that parked in - // `wait_for_login`. Without this the - // dashboard's `needs_login` chip would survive a - // healthy re-spawn since the `Online` branch goes - // straight into `serve()` and `emit_status("online")` - // — the only callsite that removes the sentinel — is - // never invoked on this path. Idempotent: when no - // sentinel exists the inner `remove_file` is a no-op. - bus.emit_status("online"); - serve( - &cli.socket, - Duration::from_millis(poll_ms), - login_state, - claude_dir, - bus, - stats, - &files, - turn_lock, - &label, - ) - .await - } - LoginState::NeedsLogin => { - // Partial-run mode: keep the harness alive (so the web UI - // stays bound) but don't drive the turn loop. Poll the - // claude dir; once a session lands we enter `serve`. - turn::wait_for_login(&claude_dir, login_state.clone(), &bus, poll_ms).await; - serve( - &cli.socket, - Duration::from_millis(poll_ms), - login_state, - claude_dir, - bus, - stats, - &files, - turn_lock, - &label, - ) - .await - } - } - } - Cmd::Mcp => mcp::serve_agent_stdio(cli.socket).await, - Cmd::Wake { from, body } => { - // Read body from stdin if caller passed `-`. Same convention - // many CLI tools use; keeps multi-line / shell-quoting - // friction out of the body content. - 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: AgentResponse = - client::request(&cli.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:?}"), - } - } - } -} - -#[allow(clippy::too_many_arguments)] -async fn serve( - socket: &Path, - interval: Duration, - login_state: Arc>, - claude_dir: std::path::PathBuf, - bus: Bus, - stats: Option, - files: &turn::TurnFiles, - turn_lock: TurnLock, - label: &str, -) -> Result<()> { - tracing::info!(socket = %socket.display(), "hive-ag3nt serve"); - requeue_inflight(socket).await; - loop { - let recv: Result = - // Explicit long-poll: park until a message arrives (180s cap). - // `max: None` (= 1) — one turn per wake; claude calls - // recv(max: N) in-turn to drain bursts. - 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"); - let auth_failed = - handle_agent_turn(socket, &bus, stats.as_ref(), files, &turn_lock, label, first) - .await; - if auth_failed { - // Park: flip LoginState + wait for the operator's - // re-auth to repopulate claude_dir. wait_for_login - // emits `online` on resume, which clears the - // needs_login sentinel. - *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(AgentResponse::Messages { .. }) => { - // Idle: empty list = nothing pending. Brief sleep - // before next poll so a stretch of empty long-poll - // returns doesn't tight-loop. - 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"); - } - } - } -} - -/// Drive one turn for a received agent-inbox message. Returns `true` -/// when the turn ended with `AuthFailed` so the caller knows to park -/// in `wait_for_login`. -async fn handle_agent_turn( - socket: &Path, - bus: &Bus, - stats: Option<&TurnStats>, - files: &turn::TurnFiles, - turn_lock: &TurnLock, - label: &str, - first: hive_sh4re::DeliveredMessage, -) -> bool { - let from = first.from; - let body = first.body; - let redelivered = first.redelivered; - tracing::info!(%from, %body, %redelivered, "inbox"); - let unread = 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(); - let started_instant = std::time::Instant::now(); - let model_at_start = bus.model(); - let prompt = serve_common::format_wake_prompt(&from, &body, unread, redelivered); - 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); - // Ack only on a clean turn-end. `Failed` leaves every message popped - // during the turn in the unacked list; next harness boot requeues them. - if matches!(outcome, turn::TurnOutcome::Ok | turn::TurnOutcome::Compacted) { - 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; - requeue_inflight(socket).await; - bus.emit_status("online"); - } - // 401: flip into needs_login + requeue the message that triggered - // the turn so it survives the re-auth. The serve loop's outer - // login-state watcher parks until the operator's `/login` flow - // completes; once it does, the requeued message replays the turn - // (closes #419). - 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"); - requeue_inflight(socket).await; - } - // Real crash: PromptTooLong is absorbed by compaction inside drive_turn. - if let turn::TurnOutcome::Failed(e) = &outcome { - notify_manager_of_failure(socket, label, 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) = fetch_agent_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 = inbox_unread(socket).await; - if pending > 0 { - tracing::info!(%pending, "pending messages after turn; fetching next"); - } - // `request_next_turn` MCP tool: agent wrote a sentinel requesting - // an immediate self-continuation. Clear and inject synthetic wake. - check_and_inject_continue(socket, label).await; - matches!(outcome, turn::TurnOutcome::AuthFailed) -} - -// Per-turn user prompt: the role/tools/etc. is in the system prompt -// (`prompts/system.md` filtered to this agent's role-block via -// `hive_ag3nt::prompt::render` → `claude --system-prompt-file`); this -// is just the wake signal claude reacts to. `unread` is the count of *other* -// messages in the inbox right after this one was popped. -// `redelivered` flags messages that were popped in a prior harness -// session, never acked, and resurfaced after a restart — a banner -// at the top of the wake prompt warns that any side-effects of -// previous handling may already have happened. - -/// Best-effort: tell the broker every message we popped during the -/// turn is now fully handled (turn-end-OK). Swallows transport -/// errors — the worst case is a redundant requeue on next boot. -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"), - } -} - -/// Boot-time recovery: ask the broker to resurface anything we -/// popped in a previous harness session but never acked. The broker -/// resets `delivered_at = NULL` on those rows and remembers their -/// ids so the next `Recv` carries `redelivered: true`. Swallows -/// transport errors — they degrade to "no recovery this boot", -/// which is no worse than the pre-feature behaviour (silent drop). -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"), - } -} - -/// Best-effort: tell the manager that this agent's last turn crashed -/// (claude exited non-zero, compaction didn't help, etc.). Routed -/// through the normal send path so the manager's inbox surfaces it -/// as a system-style event; `label` is included explicitly in the -/// body so the manager can identify the failing agent without having -/// to look at the `from` field (which is broker-stamped and may -/// differ from what the operator sees in the dashboard). Swallows -/// transport errors — we just logged the failure, the worst case is -/// the manager learns about the crash from the dashboard instead of -/// inbox. -async fn notify_manager_of_failure(socket: &Path, label: &str, err: &anyhow::Error) { - let body = format!("[system] agent `{label}` claude turn failed:\n{err:#}"); - let res = client::request::<_, AgentResponse>( - socket, - &AgentRequest::Send { - to: "manager".into(), - body, - in_reply_to: None, - }, - ) - .await; - if let Err(e) = res { - tracing::warn!(error = ?e, "failed to notify manager of turn failure"); - } -} - -/// Best-effort: ask our own per-agent socket how many messages are still -/// pending after the wake-up Recv. Returns 0 if anything goes wrong. -async fn inbox_unread(socket: &Path) -> u64 { - match client::request::<_, AgentResponse>(socket, &AgentRequest::Status).await { - Ok(AgentResponse::Status { unread }) => unread, - _ => 0, - } -} - - -/// Best-effort: ask hive-c0re for this agent's open thread count + pending -/// reminder count, after the turn finishes. Either roundtrip can fail -/// (transport hiccup, race with hive-c0re restart) — in those cases we -/// just drop a `None` into the stats row rather than blocking the loop. -async fn fetch_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) -} - -/// Check for the `request_next_turn` sentinel file. If present, remove it -/// and inject a synthetic `from: "self", body: "continue"` message so the -/// serve loop fires an immediate follow-up turn even when the inbox is empty. -/// Best-effort: any I/O error is logged and ignored (the agent just waits -/// for a real message as normal). -async fn check_and_inject_continue(socket: &Path, label: &str) { - let sentinel = hive_ag3nt::paths::state_dir().join("hyperhive-continue"); - if !sentinel.exists() { - return; - } - if let Err(e) = std::fs::remove_file(&sentinel) { - tracing::warn!(error = %e, "check_and_inject_continue: remove sentinel failed"); - return; - } - // Sentinel was present: inject a wake so the outer loop fires immediately. - // Route through the `Wake` request which is already wired in agent_server. - let res = client::request::<_, AgentResponse>( - socket, - &AgentRequest::Wake { - from: "self".into(), - body: "continue".into(), - }, - ) - .await; - match res { - Ok(AgentResponse::Ok) => { - tracing::info!(%label, "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"); - } - _ => {} - } -} - diff --git a/hive-ag3nt/src/bin/hive-m1nd.rs b/hive-ag3nt/src/bin/hive-m1nd.rs deleted file mode 100644 index ab3bba64..00000000 --- a/hive-ag3nt/src/bin/hive-m1nd.rs +++ /dev/null @@ -1,358 +0,0 @@ -//! Manager harness. Talks to the manager socket (bind-mounted from the host -//! at `/run/hive/mcp.sock` inside the `hm1nd` container). Two surfaces: -//! `serve` (long-lived turn loop) and `mcp` (stdio MCP server claude spawns). - -use std::path::{Path, PathBuf}; -use std::sync::{Arc, Mutex}; -use std::time::Duration; - -use hive_ag3nt::web_ui::TurnLock; - -use anyhow::Result; -use clap::{Parser, Subcommand}; -use hive_ag3nt::events::{Bus, LiveEvent, TurnState}; -use hive_ag3nt::login::{self, LoginState}; -use hive_ag3nt::turn_stats::TurnStats; -use hive_ag3nt::{DEFAULT_SOCKET, DEFAULT_WEB_PORT, client, mcp, plugins, serve_common, turn, web_ui}; -use hive_sh4re::{HelperEvent, ManagerRequest, ManagerResponse, SYSTEM_SENDER}; - -#[derive(Parser)] -#[command(name = "hive-m1nd", about = "hyperhive manager harness")] -struct Cli { - /// Path to the manager MCP socket (bind-mounted from the host). - #[arg(long, global = true, default_value = DEFAULT_SOCKET)] - socket: PathBuf, - - #[command(subcommand)] - cmd: Cmd, -} - -#[derive(Subcommand)] -enum Cmd { - /// Long-lived loop polling the manager inbox. - Serve { - #[arg(long, default_value_t = 1000)] - poll_ms: u64, - }, - /// Run the manager MCP server on stdio. Spawned by claude via - /// `--mcp-config`; same shape as `hive-ag3nt mcp` but with the - /// manager tool surface (`request_init_config`, `request_apply_commit`, - /// `kill`, `start`, `restart`, `ask`, `answer`, `remind`, …). - Mcp, -} - -#[tokio::main] -async fn main() -> Result<()> { - tracing_subscriber::fmt() - .with_env_filter( - tracing_subscriber::EnvFilter::try_from_default_env() - .unwrap_or_else(|_| tracing_subscriber::EnvFilter::new("info")), - ) - .init(); - - let cli = Cli::parse(); - match cli.cmd { - Cmd::Serve { poll_ms } => { - 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(&cli.socket, &label, mcp::Flavor::Manager).await?; - let turn_lock: TurnLock = Arc::new(tokio::sync::Mutex::new(())); - plugins::install_configured(&cli.socket, None).await; - tokio::spawn(web_ui::serve( - label, - port, - login_state.clone(), - bus.clone(), - cli.socket.clone(), - files.clone(), - turn_lock.clone(), - )); - tokio::spawn(hive_ag3nt::forge_notify::run(cli.socket.clone(), true)); - match initial { - LoginState::Online => { - // #682: clear any stale `hyperhive-needs-login` - // sentinel from a prior boot that parked in - // `wait_for_login`. Same bug as the sub-agent - // binary — without this the dashboard's - // `needs_login` chip survives a healthy re-spawn - // because the `Online` branch never invokes - // `emit_status("online")`. Mirror of the fix in - // `hive-ag3nt.rs`. Idempotent. - bus.emit_status("online"); - serve( - &cli.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; - serve( - &cli.socket, - Duration::from_millis(poll_ms), - login_state, - claude_dir, - bus, - stats, - &files, - turn_lock, - ) - .await - } - } - } - Cmd::Mcp => mcp::serve_manager_stdio(cli.socket).await, - } -} - -#[allow(clippy::too_many_arguments)] -async fn serve( - 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"); - // Same boot-time recovery as hive-ag3nt — see that loop for the - // rationale. Manager-flavour socket so we requeue only manager - // inflight rows. - requeue_inflight(socket).await; - loop { - let recv: Result = - // Explicit long-poll: see hive-ag3nt's serve loop for the - // rationale — recv now defaults to peek when wait_seconds - // is None. `max: None` (= 1) keeps the serve loop driving - // one turn per wake; claude calls recv(max: N) in-turn to - // drain a burst when the wake prompt mentions pending. - 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 { .. }) => { - // Idle: empty list = nothing pending. Brief sleep - // before the next long-poll attempt. - 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"); - } - } - } -} - -/// Drive one turn for a received manager-inbox message. Called from the -/// serve loop for the non-empty-messages arm to keep that loop readable. -/// Returns `true` when the turn ended with `AuthFailed` so the caller -/// can park in `wait_for_login`. -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; - if from == SYSTEM_SENDER { - // Helper events (ApprovalResolved / Spawned / Rebuilt / - // Killed / Destroyed) — surface in the live view and drive a - // normal turn so the manager can react. - let parsed = serde_json::from_str::(&body).ok(); - if let Some(event) = parsed { - tracing::info!(?event, "helper event"); - } else { - tracing::info!(%from, %body, "system message"); - } - bus.emit(LiveEvent::Note { text: format!("[system] {body}") }); - } - tracing::info!(%from, %body, %redelivered, "manager inbox"); - let unread = 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); - // Ack only on a clean turn-end; Failed / RateLimited leave the - // popped ids in-flight for the next boot's requeue. - if matches!(outcome, turn::TurnOutcome::Ok | turn::TurnOutcome::Compacted) { - 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; - 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"); - requeue_inflight(socket).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) = fetch_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 = inbox_unread(socket).await; - if pending > 0 { - tracing::info!(%pending, "pending messages after turn; fetching next"); - } - matches!(outcome, turn::TurnOutcome::AuthFailed) -} - -/// Best-effort: tell the broker every message popped during the turn -/// is now handled. Mirror of `hive-ag3nt::ack_turn` on the manager -/// surface. -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"), - } -} - -/// Boot-time recovery: ask the broker to resurface any inflight (popped -/// but not acked) messages so the next `Recv` re-delivers them with -/// the redelivery banner. Mirror of `hive-ag3nt::requeue_inflight`. -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, - } -} - -/// Manager-flavour equivalent of the agent helper. Mirror shape, just -/// uses ManagerRequest/ManagerResponse instead of the agent variants. -async fn fetch_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) -} - diff --git a/hive-ag3nt/src/bin/hive.rs b/hive-ag3nt/src/bin/hive.rs new file mode 100644 index 00000000..d4872c45 --- /dev/null +++ b/hive-ag3nt/src/bin/hive.rs @@ -0,0 +1,698 @@ +//! Unified hyperhive harness binary (#598). +//! +//! Replaces the pre-#598 `hive-ag3nt` + `hive-m1nd` split. One binary, +//! picks its role at startup from `HIVE_ROLE` (set by post-#676 +//! `harness-base.nix` from `hyperhive.role` — `"agent"` or +//! `"manager"`). Both agent + manager wire surfaces live in the same +//! binary because the privilege boundary is enforced server-side at +//! the socket (`/run/hive/mcp.sock`): an agent socket refuses +//! `ManagerRequest` calls regardless of who sends them, so there's no +//! escalation risk in shipping the same code to both. +//! +//! `HIVE_ROLE` defaults to `"agent"` if unset to keep the standalone +//! `nix run .#hive-ag3nt` shape working without env plumbing. + +use std::path::{Path, PathBuf}; +use std::sync::{Arc, Mutex}; +use std::time::Duration; + +use hive_ag3nt::web_ui::TurnLock; + +use anyhow::{Result, bail}; +use clap::{Parser, Subcommand}; +use hive_ag3nt::events::{Bus, LiveEvent, TurnState}; +use hive_ag3nt::login::{self, LoginState}; +use hive_ag3nt::turn_stats::TurnStats; +use hive_ag3nt::{DEFAULT_SOCKET, DEFAULT_WEB_PORT, client, mcp, plugins, serve_common, turn, web_ui}; +use hive_sh4re::{ + AgentRequest, AgentResponse, HelperEvent, ManagerRequest, ManagerResponse, SYSTEM_SENDER, +}; + +#[derive(Parser)] +#[command( + name = "hive", + about = "hyperhive harness — role from $HIVE_ROLE (agent|manager)" +)] +struct Cli { + /// Path to the per-agent MCP socket (bind-mounted from the host). + #[arg(long, global = true, default_value = DEFAULT_SOCKET)] + socket: PathBuf, + + #[command(subcommand)] + cmd: Cmd, +} + +#[derive(Subcommand)] +enum Cmd { + /// Run the long-lived harness loop. Polls inbox; replies via + /// `claude --print` when available. + Serve { + /// Inbox poll interval in milliseconds. + #[arg(long, default_value_t = 1000)] + poll_ms: u64, + }, + /// Run this role's MCP server on stdio. Spawned by `claude` via + /// `--mcp-config`; tools dispatch through `/run/hive/mcp.sock` back + /// into the hyperhive broker. + Mcp, + /// Agent-only: inject a wake-up event into this agent's inbox so + /// the next turn fires with the given body. Intended for extra MCP + /// servers / helpers (matrix bridge, scraper, webhook listener) + /// that need to nudge claude on external events. Refused when + /// `HIVE_ROLE=manager` — the manager surface has no `Wake` + /// equivalent. + Wake { + #[arg(long)] + from: String, + /// Body of the wake message. Pass `-` to read from stdin. + #[arg(long)] + body: String, + }, +} + +#[derive(Copy, Clone)] +enum Role { + Agent, + Manager, +} + +fn resolve_role() -> Result { + match std::env::var("HIVE_ROLE").as_deref() { + Ok("agent") | Err(_) => Ok(Role::Agent), + Ok("manager") => Ok(Role::Manager), + Ok(other) => bail!("unknown HIVE_ROLE={other:?}; expected 'agent' or 'manager'"), + } +} + +#[tokio::main] +async fn main() -> Result<()> { + tracing_subscriber::fmt() + .with_env_filter( + tracing_subscriber::EnvFilter::try_from_default_env() + .unwrap_or_else(|_| tracing_subscriber::EnvFilter::new("info")), + ) + .init(); + + let cli = Cli::parse(); + let role = resolve_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::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::Manager, Cmd::Wake { .. }) => { + bail!("wake is agent-only — manager has no equivalent surface") + } + } +} + +// ---------- agent role ---------- + +async fn agent_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 claude_dir = login::default_dir(); + let initial = LoginState::from_dir(&claude_dir); + tracing::info!(state = ?initial, claude_dir = %claude_dir.display(), "harness 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::Agent).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)); + tokio::spawn(web_ui::serve( + label.clone(), + port, + login_state.clone(), + bus.clone(), + socket.to_path_buf(), + files.clone(), + turn_lock.clone(), + )); + match initial { + LoginState::Online => { + agent_serve_loop( + socket, + Duration::from_millis(poll_ms), + login_state, + claude_dir, + bus, + stats, + &files, + turn_lock, + &label, + ) + .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, + &label, + ) + .await + } + } +} + +#[allow(clippy::too_many_arguments)] +async fn agent_serve_loop( + socket: &Path, + interval: Duration, + login_state: Arc>, + claude_dir: std::path::PathBuf, + bus: Bus, + stats: Option, + files: &turn::TurnFiles, + turn_lock: TurnLock, + label: &str, +) -> Result<()> { + tracing::info!(socket = %socket.display(), "hive-ag3nt serve"); + agent_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"); + let auth_failed = + handle_agent_turn(socket, &bus, stats.as_ref(), files, &turn_lock, label, 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(AgentResponse::Messages { .. }) => { + 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"); + } + } + } +} + +async fn handle_agent_turn( + socket: &Path, + bus: &Bus, + stats: Option<&TurnStats>, + files: &turn::TurnFiles, + turn_lock: &TurnLock, + label: &str, + first: hive_sh4re::DeliveredMessage, +) -> bool { + let from = first.from; + let body = first.body; + let redelivered = first.redelivered; + tracing::info!(%from, %body, %redelivered, "inbox"); + let unread = agent_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(); + let started_instant = std::time::Instant::now(); + let model_at_start = bus.model(); + let prompt = serve_common::format_wake_prompt(&from, &body, unread, redelivered); + 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) { + agent_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; + agent_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"); + agent_requeue_inflight(socket).await; + } + if let turn::TurnOutcome::Failed(e) = &outcome { + agent_notify_manager_of_failure(socket, label, 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 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 = agent_inbox_unread(socket).await; + if pending > 0 { + tracing::info!(%pending, "pending messages after turn; fetching next"); + } + agent_check_and_inject_continue(socket, label).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"), + } +} + +async fn agent_notify_manager_of_failure(socket: &Path, label: &str, err: &anyhow::Error) { + let body = format!("[system] agent `{label}` claude turn failed:\n{err:#}"); + let res = client::request::<_, AgentResponse>( + socket, + &AgentRequest::Send { + to: "manager".into(), + body, + in_reply_to: None, + }, + ) + .await; + if let Err(e) = res { + tracing::warn!(error = ?e, "failed to notify manager 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, label: &str) { + let sentinel = hive_ag3nt::paths::state_dir().join("hyperhive-continue"); + if !sentinel.exists() { + return; + } + if let Err(e) = std::fs::remove_file(&sentinel) { + tracing::warn!(error = %e, "check_and_inject_continue: remove sentinel failed"); + return; + } + let res = client::request::<_, AgentResponse>( + socket, + &AgentRequest::Wake { + from: "self".into(), + body: "continue".into(), + }, + ) + .await; + match res { + Ok(AgentResponse::Ok) => { + tracing::info!(%label, "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<()> { + 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: 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 => { + 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; + if from == SYSTEM_SENDER { + let parsed = serde_json::from_str::(&body).ok(); + if let Some(event) = parsed { + tracing::info!(?event, "helper event"); + } else { + tracing::info!(%from, %body, "system message"); + } + bus.emit(LiveEvent::Note { text: format!("[system] {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 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"); + } + matches!(outcome, turn::TurnOutcome::AuthFailed) +} + +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) +} diff --git a/hive-ag3nt/src/turn.rs b/hive-ag3nt/src/turn.rs index 95c9f0d8..5982764f 100644 --- a/hive-ag3nt/src/turn.rs +++ b/hive-ag3nt/src/turn.rs @@ -141,7 +141,7 @@ pub async fn write_mcp_config(socket: &Path) -> Result { let path = parent.join("claude-mcp-config.json"); let exe = std::env::current_exe() .ok() - .map_or_else(|| "hive-ag3nt".into(), |p| p.display().to_string()); + .map_or_else(|| "hive".into(), |p| p.display().to_string()); let body = mcp::render_claude_config(&exe, socket); tokio::fs::write(&path, body).await?; tracing::info!(path = %path.display(), "wrote claude MCP config"); diff --git a/nix/templates/harness-base.nix b/nix/templates/harness-base.nix index fbc5c9d7..09ed177c 100644 --- a/nix/templates/harness-base.nix +++ b/nix/templates/harness-base.nix @@ -1240,7 +1240,12 @@ in systemd.services.${if config.hyperhive.role == "manager" then "hive-m1nd" else "hive-ag3nt"} = let isManager = config.hyperhive.role == "manager"; - binary = if isManager then "hive-m1nd" else "hive-ag3nt"; + # Post-#598 there is exactly one harness binary (`hive`), and + # it picks its surface from `HIVE_ROLE` at startup. We still + # name the systemd unit `hive-ag3nt` / `hive-m1nd` so dashboard + # log queries + ExecStartPre paths + ancestor PR diffs keep + # working without a unit rename cascade. + binary = "hive"; in { description = "${binary}${lib.optionalString isManager " manager"} harness"; @@ -1283,6 +1288,10 @@ in # unit directly — `environment.variables` only populates # /etc/profile, which systemd services don't inherit. HIVE_ASSETS_DIR = "${pkgs.hyperhive-assets}/share/hyperhive"; + # Post-#598: the unified `hive` binary picks its surface from + # this env var at startup. Default (`"agent"`) matches the + # binary's standalone fallback when this is unset. + HIVE_ROLE = config.hyperhive.role; } // lib.optionalAttrs isManager { # Standalone-eval fallbacks for `nixosConfigurations.manager`.