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

use std::path::{Path, PathBuf};
use std::sync::{Arc, Mutex};
use std::time::Duration;

use anyhow::{Result, bail};
use clap::{Parser, Subcommand};
use hive_ag3nt::login::{self, LoginState};
use hive_ag3nt::{DEFAULT_SOCKET, DEFAULT_WEB_PORT, client, web_ui};
use hive_sh4re::{AgentRequest, AgentResponse};
use tokio::process::Command;

#[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,
    },
    /// Send a message to another agent.
    Send { to: String, body: String },
    /// Pop one message from the inbox.
    Recv,
}

#[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(|_| "hive-ag3nt".into());
            let claude_dir = PathBuf::from(login::DEFAULT_CLAUDE_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 ui_state = login_state.clone();
            tokio::spawn(async move {
                if let Err(e) = web_ui::serve(label, port, ui_state).await {
                    tracing::error!(error = ?e, "web ui failed");
                }
            });
            match initial {
                LoginState::Online => {
                    serve(&cli.socket, Duration::from_millis(poll_ms), login_state).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 periodically so a successful login (whether
                    // from the dashboard PTY path in step 4 or via
                    // `root-login` + `claude /login` in the meantime)
                    // transitions us into the turn loop without a restart.
                    needs_login_loop(&cli.socket, &claude_dir, login_state, poll_ms).await
                }
            }
        }
        Cmd::Send { to, body } => {
            let resp: AgentResponse =
                client::request(&cli.socket, &AgentRequest::Send { to, body }).await?;
            render(&resp)?;
            check(&resp)
        }
        Cmd::Recv => {
            let resp: AgentResponse = client::request(&cli.socket, &AgentRequest::Recv).await?;
            render(&resp)?;
            check(&resp)
        }
    }
}

/// Re-checks `claude_dir` every `poll_ms` ms. As soon as it contains a session
/// (login completed), flips `state` to `Online` and enters the turn loop.
async fn needs_login_loop(
    socket: &Path,
    claude_dir: &Path,
    state: Arc<Mutex<LoginState>>,
    poll_ms: u64,
) -> Result<()> {
    tracing::warn!(
        claude_dir = %claude_dir.display(),
        "no claude session — staying in partial-run mode (web UI only)"
    );
    let probe = Duration::from_millis(poll_ms.max(2000));
    loop {
        tokio::time::sleep(probe).await;
        if login::has_session(claude_dir) {
            tracing::info!("claude session detected — entering turn loop");
            *state.lock().unwrap() = LoginState::Online;
            return serve(socket, Duration::from_millis(poll_ms), state).await;
        }
    }
}

async fn serve(socket: &Path, interval: Duration, state: Arc<Mutex<LoginState>>) -> Result<()> {
    tracing::info!(socket = %socket.display(), "hive-ag3nt serve");
    let _ = state; // reserved for future state transitions (turn-loop -> needs-login)
    loop {
        let recv: Result<AgentResponse> = client::request(socket, &AgentRequest::Recv).await;
        match recv {
            Ok(AgentResponse::Message { from, body }) => {
                tracing::info!(%from, %body, "inbox");
                // Don't auto-reply to echoes — prevents infinite ping-pong when
                // both ends are falling back to echo. Real loop control is the
                // manager's job (Phase 4+).
                if !body.starts_with("echo: ") {
                    let reply = compute_reply(&body).await;
                    let send: Result<AgentResponse> = client::request(
                        socket,
                        &AgentRequest::Send {
                            to: from,
                            body: reply,
                        },
                    )
                    .await;
                    if let Err(e) = send {
                        tracing::warn!(error = ?e, "send reply failed");
                    }
                }
            }
            Ok(AgentResponse::Empty) => {}
            Ok(AgentResponse::Ok) => {
                tracing::warn!("recv produced Ok (unexpected)");
            }
            Ok(AgentResponse::Err { message }) => {
                tracing::warn!(%message, "recv error");
            }
            Err(e) => {
                tracing::warn!(error = ?e, "recv failed; retrying");
            }
        }
        tokio::time::sleep(interval).await;
    }
}

async fn compute_reply(prompt: &str) -> String {
    match invoke_claude(prompt).await {
        Ok(s) => s,
        Err(e) => {
            tracing::warn!(error = %format!("{e:#}"), "claude failed; falling back to echo");
            format!("echo: {prompt}")
        }
    }
}

async fn invoke_claude(prompt: &str) -> Result<String> {
    let out = Command::new("claude")
        .arg("--print")
        .arg(prompt)
        .output()
        .await?;
    if !out.status.success() {
        bail!(
            "claude exited {}: {}",
            out.status,
            String::from_utf8_lossy(&out.stderr).trim()
        );
    }
    let text = String::from_utf8_lossy(&out.stdout).trim().to_owned();
    if text.is_empty() {
        bail!("claude produced empty output");
    }
    Ok(text)
}

fn render(resp: &AgentResponse) -> Result<()> {
    println!("{}", serde_json::to_string_pretty(resp)?);
    Ok(())
}

fn check(resp: &AgentResponse) -> Result<()> {
    if let AgentResponse::Err { message } = resp {
        bail!("{message}");
    }
    Ok(())
}