hyperhive/hive-ag3nt/src/bin/hive-m1nd.rs

//! 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_spawn`, `kill`, `start`, `restart`,
    /// `request_apply_commit`, `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::<u16>().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 => {
                    serve(
                        &cli.socket,
                        Duration::from_millis(poll_ms),
                        bus,
                        stats,
                        &files,
                        turn_lock,
                    )
                    .await
                }
                LoginState::NeedsLogin => {
                    turn::wait_for_login(&claude_dir, login_state, &bus, poll_ms).await;
                    serve(
                        &cli.socket,
                        Duration::from_millis(poll_ms),
                        bus,
                        stats,
                        &files,
                        turn_lock,
                    )
                    .await
                }
            }
        }
        Cmd::Mcp => mcp::serve_manager_stdio(cli.socket).await,
    }
}

async fn serve(
    socket: &Path,
    interval: Duration,
    bus: Bus,
    stats: Option<TurnStats>,
    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<ManagerResponse> =
            // 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");
                handle_manager_turn(socket, &bus, stats.as_ref(), files, &turn_lock, first).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 { .. },
            ) => {
                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.
async fn handle_manager_turn(
    socket: &Path,
    bus: &Bus,
    stats: Option<&TurnStats>,
    files: &turn::TurnFiles,
    turn_lock: &TurnLock,
    first: hive_sh4re::DeliveredMessage,
) {
    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::<HelperEvent>(&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 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");
    }
}

/// 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<u64>, Option<u64>) {
    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)
}