hiboss 1.0.0

CLI for AI agents to communicate with their boss
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127

// Purpose: Provide the hiboss bot command that auto-replies with an external handler.
// Exports: BotArgs CLI parser and run() to stream messages via SSE and reply.
// Dependencies: clap, tokio, crate::client, crate::config, crate::types, crate::sse.

use crate::{client::HiBossClient, config::Config, helpers::{short_id, truncate}, session, sse, types::Message};
use clap::Args;
use std::{error::Error, process::Stdio};
use tokio::{
    io::AsyncWriteExt,
    process::Command as TokioCommand,
    signal,
    time::{Duration, sleep},
};

#[derive(Debug, Args)]
pub struct BotArgs {
    #[arg(long)]
    pub handler: String,
    #[arg(long, default_value_t = 5)]
    pub interval: u64,
}

pub async fn run(args: &BotArgs, config: &Config, client: &HiBossClient) -> Result<(), Box<dyn Error>> {
    let server = config.require_server()?;
    let key = config.require_key()?;
    eprintln!(
        "Bot mode active. Handler: {} (Ctrl+C to stop)",
        args.handler
    );

    let (tx, mut rx) = tokio::sync::mpsc::channel::<sse::SseEvent>(32);
    let mut sse_url = format!("{}/api/messages/stream", server);
    if let Some(sid) = session::read_session_id() {
        sse_url = format!("{}?session={}", sse_url, sid);
    }
    let sse_client = reqwest::Client::new();
    let sse_key = key;

    tokio::spawn(async move {
        let mut backoff = 5u64;
        loop {
            eprintln!("Connecting to SSE stream...");
            match sse::connect_sse(&sse_client, &sse_url, &sse_key, tx.clone()).await {
                Err(e) => eprintln!("SSE error: {}, retrying in {}s...", e, backoff),
                Ok(_) => {
                    backoff = 5;
                    eprintln!("SSE stream closed, reconnecting in 5s...");
                }
            }
            sleep(Duration::from_secs(backoff)).await;
            backoff = (backoff * 2).min(60);
        }
    });

    loop {
        tokio::select! {
            _ = signal::ctrl_c() => {
                eprintln!("Bot mode stopping");
                break;
            }
            Some(event) = rx.recv() => {
                if event.event_type == "message" {
                    if let Ok(message) = serde_json::from_str::<Message>(&event.data) {
                        handle_message(&args.handler, client, &message).await;
                    }
                }
            }
        }
    }

    Ok(())
}

async fn handle_message(handler: &str, client: &HiBossClient, message: &Message) {
    let id = &message.id;
    let agent = message.agent_name.as_deref().unwrap_or("hiboss");
    let body = message.body.as_deref().unwrap_or("");
    eprintln!("[recv] {}: {}", agent, body);

    match run_handler(handler, body).await {
        Ok(Some(reply)) => {
            if let Err(err) = client.reply_to(id, &reply).await {
                eprintln!("Failed to reply to {}: {}", short_id(id), err);
            } else {
                eprintln!("[reply] {}", truncate(&reply, 80));
            }
        }
        Ok(None) => {
            eprintln!("Handler produced no reply for {}", short_id(id));
        }
        Err(err) => {
            eprintln!("Handler failed for {}: {}", short_id(id), err);
        }
    }

    if let Err(err) = client.update_status(id, "read").await {
        eprintln!("Failed to mark {} as read: {}", short_id(id), err);
    }
}

async fn run_handler(handler: &str, input: &str) -> Result<Option<String>, Box<dyn Error>> {
    let mut child = TokioCommand::new("sh")
        .arg("-c")
        .arg(handler)
        .stdin(Stdio::piped())
        .stdout(Stdio::piped())
        .spawn()?;

    if let Some(mut stdin) = child.stdin.take() {
        if !input.is_empty() {
            stdin.write_all(input.as_bytes()).await?;
        }
    }

    let output = child.wait_with_output().await?;
    if !output.status.success() {
        return Ok(None);
    }

    let reply = String::from_utf8_lossy(&output.stdout).to_string();
    let trimmed = reply.trim_end();
    if trimmed.is_empty() {
        return Ok(None);
    }

    Ok(Some(trimmed.to_string()))
}