codecraft/cli/src/server.rs

//! Connection to a CodeCraft server: login, configuration, model gateway, MCP tools.

use crate::conf::ServerLogin;
use anyhow::{Context, Result, anyhow, bail};
use codecraft_core::provider::{StreamEvent, ToolDef};
use codecraft_core::types::{Message, Usage};
use eventsource_stream::Eventsource;
use futures::StreamExt;
use serde_json::{Value, json};
use tokio::sync::mpsc;
use tokio_util::sync::CancellationToken;

#[derive(Clone)]
pub struct Server {
    pub url: String,
    token: String,
    http: reqwest::Client,
}

pub fn http() -> reqwest::Client {
    reqwest::Client::builder()
        .user_agent(concat!("codecraft-cli/", env!("CARGO_PKG_VERSION")))
        .connect_timeout(std::time::Duration::from_secs(15))
        .build()
        .expect("http client")
}

async fn check(resp: reqwest::Response) -> Result<reqwest::Response> {
    if resp.status().is_success() {
        return Ok(resp);
    }
    let status = resp.status();
    let body = resp.text().await.unwrap_or_default();
    let msg = serde_json::from_str::<Value>(&body).ok().and_then(|v| v["error"].as_str().map(String::from)).unwrap_or(body);
    match status.as_u16() {
        401 => bail!("the server rejected the login – run `codecraft login` again"),
        _ => bail!("server: {} {}", status.as_u16(), msg.trim()),
    }
}

/// Sign in with username and password; the server creates an API token for this machine.
pub async fn login(url: &str, username: &str, password: &str) -> Result<ServerLogin> {
    let url = url.trim().trim_end_matches('/').to_string();
    let url = if url.starts_with("http://") || url.starts_with("https://") { url } else { format!("https://{url}") };
    let host = hostname();
    let resp = http()
        .post(format!("{url}/api/auth/cli-login"))
        .json(&json!({"username": username, "password": password, "hostname": host}))
        .send()
        .await
        .with_context(|| format!("cannot reach {url}"))?;
    let v: Value = check(resp).await?.json().await?;
    let token = v["token"].as_str().ok_or_else(|| anyhow!("unexpected answer from the server"))?.to_string();
    Ok(ServerLogin { url, token, username: v["user"]["username"].as_str().unwrap_or(username).to_string() })
}

pub fn hostname() -> String {
    std::env::var("HOSTNAME")
        .ok()
        .or_else(|| std::fs::read_to_string("/etc/hostname").ok())
        .or_else(|| std::env::var("COMPUTERNAME").ok())
        .map(|h| h.trim().to_string())
        .filter(|h| !h.is_empty())
        .unwrap_or_else(|| "terminal".into())
}

impl Server {
    pub fn new(login: &ServerLogin) -> Self {
        Self { url: login.url.clone(), token: login.token.clone(), http: http() }
    }

    pub async fn get(&self, path: &str) -> Result<Value> {
        let resp = self.http.get(format!("{}/api{path}", self.url)).bearer_auth(&self.token).send().await.with_context(|| format!("cannot reach {}", self.url))?;
        Ok(check(resp).await?.json().await?)
    }

    pub async fn post(&self, path: &str, body: Value) -> Result<Value> {
        let resp = self.http.post(format!("{}/api{path}", self.url)).bearer_auth(&self.token).json(&body).send().await.with_context(|| format!("cannot reach {}", self.url))?;
        Ok(check(resp).await?.json().await?)
    }

    /// One model request through the server's gateway, as stream events.
    #[allow(clippy::too_many_arguments)]
    pub async fn stream(
        &self,
        model: &str,
        system: String,
        messages: Vec<Message>,
        tools: Vec<ToolDef>,
        effort: Option<String>,
        session: &str,
        agent: &str,
        tx: mpsc::Sender<StreamEvent>,
        cancel: CancellationToken,
    ) -> Result<()> {
        let body = json!({
            "model": model, "system": system, "messages": messages, "tools": tools,
            "reasoning": true, "effort": effort, "session": session, "agent": agent,
        });
        let resp = self.http.post(format!("{}/api/llm/stream", self.url)).bearer_auth(&self.token).json(&body).send().await.with_context(|| format!("cannot reach {}", self.url))?;
        let resp = check(resp).await?;
        let mut events = resp.bytes_stream().eventsource();
        loop {
            let ev = tokio::select! {
                ev = events.next() => ev,
                _ = cancel.cancelled() => return Ok(()),
            };
            let Some(ev) = ev else { break };
            let ev = ev.map_err(|e| anyhow!("stream error: {e}"))?;
            let v: Value = serde_json::from_str(&ev.data).unwrap_or(Value::Null);
            let out = match v["type"].as_str().unwrap_or("") {
                "text" => StreamEvent::TextDelta(v["text"].as_str().unwrap_or("").into()),
                "reasoning" => StreamEvent::ReasoningDelta(v["text"].as_str().unwrap_or("").into()),
                "signature" => StreamEvent::ReasoningSignature(v["signature"].as_str().unwrap_or("").into()),
                "tool_call" => StreamEvent::ToolCall {
                    id: v["id"].as_str().unwrap_or("").into(),
                    name: v["name"].as_str().unwrap_or("").into(),
                    input: v["input"].clone(),
                    extra: v.get("extra").filter(|e| !e.is_null()).cloned(),
                },
                "usage" => StreamEvent::Usage(serde_json::from_value::<Usage>(v["usage"].clone()).unwrap_or_default()),
                "done" => StreamEvent::Done { stop_reason: v["stop_reason"].as_str().unwrap_or("").into() },
                "error" => bail!("{}", v["message"].as_str().unwrap_or("model error")),
                _ => continue,
            };
            if tx.send(out).await.is_err() {
                break;
            }
        }
        Ok(())
    }

    /// Run one of the server's MCP tools.
    pub async fn tool(&self, name: &str, input: Value) -> Result<(String, bool)> {
        let v = self.post("/cli/tool", json!({"name": name, "input": input})).await?;
        Ok((v["content"].as_str().unwrap_or("").to_string(), v["is_error"].as_bool().unwrap_or(false)))
    }
}