538 lines
22 KiB
Rust
538 lines
22 KiB
Rust
//! Blocking HTTP client for OpenAI/Anthropic-compatible chat completion APIs,
|
|
//! supporting both non-streaming and SSE-streaming requests with automatic retry.
|
|
//!
|
|
//! # Retry policy
|
|
//!
|
|
//! Both paths use exponential backoff with ±25% jitter so retries spread out
|
|
//! naturally instead of hammering the server in lockstep. Auth errors
|
|
//! (401/402/403) are never retried — they indicate a bad key or billing issue
|
|
//! that retrying won't fix. Rate-limit (429) responses get a longer backoff
|
|
//! (base 5s instead of the usual 1s) so the server has time to drain its queue.
|
|
//!
|
|
//! ## Non-streaming (`chat_with_tools_non_streaming`)
|
|
//! - Up to **10** attempts
|
|
//! - Backoff: `1s, 2s, 4s, 8s, 16s, 30s(capped), 30s, …` + jitter
|
|
//! - Auth errors → abort immediately on the **status code** embedded in the
|
|
//! error message (avoids false positives from port numbers, model names etc.)
|
|
//!
|
|
//! ## Streaming (`chat_with_tools_streaming`)
|
|
//! - Up to **5** attempts *before* any meaningful content (tokens / reasoning)
|
|
//! - After meaningful content arrives, falls back to a **non-streaming retry**
|
|
//! (the non-streaming call carries 10 retries of its own), so a mid-stream
|
|
//! network blip is recovered instead of killing the whole turn.
|
|
//! - The `started` flag still prevents retries on the raw SSE call once the
|
|
//! stream has begun (partial content cannot be safely replayed), but the
|
|
//! caller-level fallback handles that case.
|
|
|
|
use anyhow::Result;
|
|
use std::sync::atomic::AtomicBool;
|
|
use std::time::Duration;
|
|
|
|
use crate::app::runtime::stream::turn::StreamedTurn;
|
|
use crate::app::util::backoff::backoff_seconds;
|
|
use crate::app::runtime::stream::{SseParser, StreamEvent};
|
|
use crate::dto::chat::message::ChatMessage;
|
|
use crate::dto::provider::request::{ChatRequest, StreamOptions, ToolDef};
|
|
|
|
pub(crate) const DEFAULT_BASE_URL: &str = "https://opencode.ai/zen/v1";
|
|
const DEFAULT_MODEL: &str = "deepseek-v4-flash-free";
|
|
pub const DEFAULT_API_KEY: &str = "";
|
|
const CONNECT_TIMEOUT: Duration = Duration::from_secs(10);
|
|
const REQUEST_TIMEOUT: Duration = Duration::from_mins(1);
|
|
|
|
// ---------------------------------------------------------------------------
|
|
// Retry helpers
|
|
// ---------------------------------------------------------------------------
|
|
|
|
/// Exponential backoff with ±25% jitter, capped at 30 seconds.
|
|
///
|
|
/// `attempt` is 1-based (first retry → attempt=1).
|
|
fn backoff_duration(attempt: u32) -> Duration {
|
|
backoff_seconds(attempt, 30)
|
|
}
|
|
|
|
/// Is the error an auth / billing failure that retrying won't fix?
|
|
///
|
|
/// Matches the structured "API error {status} from …" format used by the
|
|
/// request builders below, plus well-known auth keywords in case the body
|
|
/// contains them. This is intentionally tighter than `contains("401")`,
|
|
/// which could false-positive on a URL port, model name, or body text.
|
|
pub fn is_auth_error(err_str: &str) -> bool {
|
|
let err_lower = err_str.to_lowercase();
|
|
// Structured HTTP status patterns
|
|
(err_str.contains("API error 401")
|
|
|| err_str.contains("API error 402")
|
|
|| err_str.contains("API error 403"))
|
|
// Keyword fallback for non-standard error formats
|
|
|| err_lower.contains("unauthorized")
|
|
|| err_lower.contains("forbidden")
|
|
|| err_lower.contains("authentication failed")
|
|
}
|
|
|
|
/// Is the error a rate-limit response?
|
|
fn is_rate_limit(err_str: &str) -> bool {
|
|
err_str.contains("API error 429") || err_str.to_lowercase().contains("rate limit")
|
|
}
|
|
|
|
/// Return a rate-appropriate backoff (longer for 429).
|
|
fn backoff_for_error(attempt: u32, err_str: &str) -> Duration {
|
|
if is_rate_limit(err_str) {
|
|
// Rate limits need more time to drain — backoff capped at 60s.
|
|
backoff_seconds(attempt, 60)
|
|
} else {
|
|
backoff_duration(attempt)
|
|
}
|
|
}
|
|
|
|
// ---------------------------------------------------------------------------
|
|
// Client
|
|
// ---------------------------------------------------------------------------
|
|
|
|
/// Blocking HTTP client for a single LLM provider endpoint.
|
|
///
|
|
/// Holds the reqwest client, credentials, and model/base URL selection used
|
|
/// by both the non-streaming and streaming chat completion calls.
|
|
pub struct LlmClient {
|
|
pub client: reqwest::blocking::Client,
|
|
pub api_key: String,
|
|
pub base_url: String,
|
|
pub model: String,
|
|
}
|
|
|
|
impl LlmClient {
|
|
/// Construct a client, falling back to built-in defaults for empty inputs.
|
|
///
|
|
/// Flow: empty `api_key/model` → substitute defaults → build reqwest client
|
|
/// with connect/request timeouts → if TLS config fails, retry with just
|
|
/// request timeout (no connect timeout) → normalize `base_url`.
|
|
///
|
|
/// Why: empty strings are treated as "unset" rather than errors so callers
|
|
/// can pass through unconfigured settings without special-casing them.
|
|
/// Timeouts are always enforced — the pure-default-client fallback is only
|
|
/// used as a last resort when even the no-connect-timeout build fails.
|
|
pub fn new(mut api_key: String, model: String, base_url: Option<String>) -> Self {
|
|
if api_key.is_empty() {
|
|
api_key = DEFAULT_API_KEY.to_string();
|
|
}
|
|
let model = if model.is_empty() {
|
|
DEFAULT_MODEL.to_string()
|
|
} else {
|
|
model
|
|
};
|
|
let client = match reqwest::blocking::Client::builder()
|
|
.timeout(REQUEST_TIMEOUT)
|
|
.connect_timeout(CONNECT_TIMEOUT)
|
|
.build()
|
|
{
|
|
Ok(c) => c,
|
|
Err(e) => {
|
|
tracing::warn!(
|
|
"failed to build reqwest client with connect timeout: {}. \
|
|
retrying without connect timeout",
|
|
e,
|
|
);
|
|
match reqwest::blocking::Client::builder()
|
|
.timeout(REQUEST_TIMEOUT)
|
|
.build()
|
|
{
|
|
Ok(c) => c,
|
|
Err(e2) => {
|
|
tracing::warn!(
|
|
"also failed: {}. using default client (no configured timeouts)",
|
|
e2,
|
|
);
|
|
reqwest::blocking::Client::new()
|
|
}
|
|
}
|
|
}
|
|
};
|
|
LlmClient {
|
|
client,
|
|
api_key,
|
|
base_url: base_url
|
|
.filter(|s| !s.is_empty())
|
|
.unwrap_or_else(|| DEFAULT_BASE_URL.to_string()),
|
|
model,
|
|
}
|
|
}
|
|
|
|
/// Send a non-streaming chat completion request and return the assistant's reply.
|
|
///
|
|
/// Flow: build request → POST with retry loop (up to 10 attempts, exponential
|
|
/// backoff with jitter) → parse JSON response → extract first choice's message
|
|
/// and token usage.
|
|
///
|
|
/// Why: retries transient failures but aborts immediately on 401/402/403 (bad
|
|
/// API key / billing issue — retrying won't fix). 429 (rate-limit) responses
|
|
/// get a longer backoff so the server has time to recover.
|
|
///
|
|
/// Return: `Err` if all retries are exhausted, an auth error occurs, or the
|
|
/// response has no choices.
|
|
pub fn chat_with_tools_non_streaming(
|
|
&self,
|
|
messages: &[ChatMessage],
|
|
tools: Option<Vec<ToolDef>>,
|
|
max_tokens: Option<u32>,
|
|
temperature: Option<f32>,
|
|
abort_flag: Option<&AtomicBool>,
|
|
) -> Result<(ChatMessage, Option<(u64, u64)>)> {
|
|
let req = ChatRequest {
|
|
model: self.model.clone(),
|
|
messages: messages.to_vec(),
|
|
max_tokens: Some(max_tokens.unwrap_or(4096)),
|
|
temperature: Some(temperature.unwrap_or(0.7)),
|
|
tools,
|
|
stream: Some(false),
|
|
stop: None,
|
|
stream_options: None,
|
|
tool_choice: None,
|
|
top_p: None,
|
|
};
|
|
|
|
let url = format!("{}/chat/completions", self.base_url);
|
|
let max_retries = 10;
|
|
let mut attempt = 0u32;
|
|
|
|
loop {
|
|
attempt += 1;
|
|
|
|
// Check abort before each retry so user cancellation is
|
|
// responsive even during a long non-streaming backoff chain.
|
|
if crate::app::util::abort::is_aborted_ref(abort_flag) {
|
|
anyhow::bail!("aborted");
|
|
}
|
|
|
|
let mut http_req = self
|
|
.client
|
|
.post(&url)
|
|
.header("Content-Type", "application/json");
|
|
|
|
if !self.api_key.is_empty() {
|
|
http_req = http_req.header("Authorization", format!("Bearer {}", self.api_key));
|
|
}
|
|
|
|
let result = (|| -> Result<(ChatMessage, Option<(u64, u64)>)> {
|
|
let resp = http_req.json(&req).send().map_err(|e| {
|
|
if e.is_timeout() {
|
|
anyhow::anyhow!("API request timed out after {REQUEST_TIMEOUT:?}. Check your network or try again.")
|
|
} else if e.is_connect() {
|
|
anyhow::anyhow!("Could not connect to {}. Is the URL correct and is the service reachable?", self.base_url)
|
|
} else {
|
|
anyhow::anyhow!("API request failed: {e}")
|
|
}
|
|
})?;
|
|
|
|
if !resp.status().is_success() {
|
|
let status = resp.status();
|
|
let body = resp.text().unwrap_or_default();
|
|
anyhow::bail!("API error {} from {}: {}", status, self.base_url, body);
|
|
}
|
|
|
|
let data: crate::dto::provider::response::ChatResponse = resp.json()?;
|
|
let usage = data
|
|
.usage
|
|
.map(|u| (u64::from(u.prompt_tokens), u64::from(u.completion_tokens)));
|
|
let message = data
|
|
.choices
|
|
.into_iter()
|
|
.next()
|
|
.and_then(|c| c.message)
|
|
.ok_or_else(|| anyhow::anyhow!("API response had no choices"))?;
|
|
Ok((message, usage))
|
|
})();
|
|
|
|
match result {
|
|
Ok((msg, usage)) => return Ok((msg, usage)),
|
|
Err(e) => {
|
|
let err_str = e.to_string();
|
|
if attempt >= max_retries || is_auth_error(&err_str) {
|
|
return Err(e);
|
|
}
|
|
let delay = backoff_for_error(attempt, &err_str);
|
|
tracing::warn!(
|
|
"Warning: {}. Retrying {}/{}, sleeping {delay:?}...",
|
|
e,
|
|
attempt,
|
|
max_retries,
|
|
);
|
|
std::thread::sleep(delay);
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
/// Streaming variant of `chat_with_tools`. Feeds SSE chunks into an
|
|
/// `SseParser` / `StreamedTurn` and invokes `on_event` for every parsed
|
|
/// `StreamEvent` as it arrives, so the caller can push incremental UI
|
|
/// updates in real time.
|
|
///
|
|
/// Returns the fully assembled assistant message plus token usage (prompt,
|
|
/// completion) if the server reported it.
|
|
///
|
|
/// # Retry semantics
|
|
///
|
|
/// Retries the raw SSE request only *before* any meaningful content (text
|
|
/// tokens or reasoning tokens) has been received — once the LLM has started
|
|
/// generating, a partial stream cannot be safely replayed without duplicating
|
|
/// or garbling output.
|
|
///
|
|
/// Once meaningful content has arrived and the stream fails, **this method
|
|
/// falls back to a non-streaming call** (which carries its own 10-retry
|
|
/// loop). The non-streaming call uses the same `messages` independently
|
|
/// (no SSE state to replay), so the caller always gets a complete result if
|
|
/// the provider is reachable.
|
|
///
|
|
/// Auth errors (401/402/403) are never retried on either path. Rate-limit
|
|
/// (429) responses get a longer backoff.
|
|
pub fn chat_with_tools_streaming(
|
|
&self,
|
|
messages: &[ChatMessage],
|
|
tools: Option<Vec<ToolDef>>,
|
|
temperature: Option<f32>,
|
|
max_tokens: Option<u32>,
|
|
mut on_event: impl FnMut(&StreamEvent) -> bool,
|
|
abort_flag: Option<&AtomicBool>,
|
|
) -> Result<(ChatMessage, Option<(u64, u64)>)> {
|
|
// Clone tools for the non-streaming fallback path — the original
|
|
// is moved into the ChatRequest below and cannot be used again.
|
|
let tools_for_fallback = tools.clone();
|
|
let req = ChatRequest {
|
|
model: self.model.clone(),
|
|
messages: messages.to_vec(),
|
|
max_tokens: Some(max_tokens.unwrap_or(4096)),
|
|
temperature: Some(temperature.unwrap_or(0.7)),
|
|
tools,
|
|
stream: Some(true),
|
|
stop: None,
|
|
stream_options: Some(StreamOptions {
|
|
include_usage: true,
|
|
}),
|
|
tool_choice: None,
|
|
top_p: None,
|
|
};
|
|
|
|
let url = format!("{}/chat/completions", self.base_url);
|
|
|
|
// Phase 1: Retry the raw SSE call up to 5 times, but only before
|
|
// meaningful content arrives. After that, fall back to non-streaming.
|
|
let max_retries_stream = 5;
|
|
let mut attempt = 0u32;
|
|
let mut started = false;
|
|
// Track whether we've emitted text/reasoning tokens (meaningful
|
|
// content). Non-meaningful events (role/usage/done) are safe to
|
|
// ignore for the retry decision.
|
|
let mut meaningful_content = false;
|
|
|
|
loop {
|
|
attempt += 1;
|
|
let mut captured_content = false;
|
|
let mut wrapped = |event: &StreamEvent| -> bool {
|
|
started = true;
|
|
match event {
|
|
StreamEvent::Token(_) | StreamEvent::Reasoning(_) => {
|
|
captured_content = true;
|
|
}
|
|
_ => {}
|
|
}
|
|
on_event(event)
|
|
};
|
|
match self.try_stream_once(&req, &url, &mut wrapped) {
|
|
Ok(result) => return Ok(result),
|
|
Err(e) => {
|
|
let err_str = e.to_string();
|
|
if is_auth_error(&err_str) {
|
|
return Err(e);
|
|
}
|
|
// Once meaningful content has been streamed, a raw SSE
|
|
// retry would produce a different sequence — fall back
|
|
// to non-streaming so the caller gets a clean,
|
|
// reproducible answer.
|
|
if captured_content || started && (attempt >= max_retries_stream) {
|
|
meaningful_content = captured_content || meaningful_content;
|
|
break;
|
|
}
|
|
if attempt >= max_retries_stream {
|
|
return Err(e);
|
|
}
|
|
let delay = backoff_for_error(attempt, &err_str);
|
|
tracing::warn!(
|
|
"Warning: {}. Retrying stream {}/{}, sleeping {delay:?}...",
|
|
e,
|
|
attempt,
|
|
max_retries_stream,
|
|
);
|
|
std::thread::sleep(delay);
|
|
}
|
|
}
|
|
}
|
|
|
|
// Phase 2: If we got meaningful content via SSE but the stream
|
|
// failed before completion, fall back to a non-streaming retry.
|
|
// This preserves the conversation state because the messages
|
|
// passed in are the same — we don't need the partial SSE output.
|
|
if meaningful_content {
|
|
// Check abort before entering the blocking non-streaming
|
|
// call — otherwise the fallback ignores user cancellation.
|
|
if crate::app::util::abort::is_aborted_ref(abort_flag) {
|
|
return Err(anyhow::anyhow!("aborted"));
|
|
}
|
|
tracing::warn!(
|
|
"streaming failed after meaningful content — falling back to non-streaming call",
|
|
);
|
|
// Use the same messages and tools so the fallback produces
|
|
// a response compatible with what the streaming request
|
|
// would have returned (including tool definitions).
|
|
return self.chat_with_tools_non_streaming(
|
|
messages,
|
|
tools_for_fallback,
|
|
max_tokens,
|
|
temperature,
|
|
abort_flag,
|
|
);
|
|
}
|
|
|
|
Err(anyhow::anyhow!(
|
|
"streaming request failed after {max_retries_stream} attempts"
|
|
))
|
|
}
|
|
|
|
/// Perform one streaming chat completion request, parsing SSE events until completion.
|
|
///
|
|
/// Flow: POST → read body in chunks → advance past valid UTF-8 boundary →
|
|
/// feed into `SseParser` → dispatch each `StreamEvent` to `on_event` and
|
|
/// accumulate in `StreamedTurn` → return assembled assistant message on `Done`.
|
|
///
|
|
/// Why: chunk-by-chunk UTF-8-aware reads avoid splitting multi-byte sequences;
|
|
/// returns `aborted` error if `on_event` returns false so the caller can cancel.
|
|
///
|
|
/// Return: assembled message + optional usage on success, `Err` on read
|
|
/// failure, non-2xx status, or callback-initiated abort.
|
|
fn try_stream_once(
|
|
&self,
|
|
req: &ChatRequest,
|
|
url: &str,
|
|
on_event: &mut dyn FnMut(&StreamEvent) -> bool,
|
|
) -> Result<(ChatMessage, Option<(u64, u64)>)> {
|
|
use std::io::Read;
|
|
|
|
let mut http_req = self
|
|
.client
|
|
.post(url)
|
|
.header("Content-Type", "application/json");
|
|
if !self.api_key.is_empty() {
|
|
http_req = http_req.header("Authorization", format!("Bearer {}", self.api_key));
|
|
}
|
|
|
|
let resp = http_req.json(req).send().map_err(|e| {
|
|
if e.is_timeout() {
|
|
anyhow::anyhow!("API request timed out after {REQUEST_TIMEOUT:?}. Check your network or try again.")
|
|
} else if e.is_connect() {
|
|
anyhow::anyhow!("Could not connect to {}. Is the URL correct and is the service reachable?", self.base_url)
|
|
} else {
|
|
anyhow::anyhow!("API request failed: {e}")
|
|
}
|
|
})?;
|
|
|
|
if !resp.status().is_success() {
|
|
let status = resp.status();
|
|
let body = resp.text().unwrap_or_default();
|
|
anyhow::bail!("API error {} from {}: {}", status, self.base_url, body);
|
|
}
|
|
|
|
let mut turn = StreamedTurn::new();
|
|
let mut usage: Option<(u64, u64)> = None;
|
|
let mut parser = SseParser::new();
|
|
let mut reader = resp;
|
|
let mut byte_buf: Vec<u8> = Vec::new();
|
|
let mut chunk_buf = [0u8; 4096];
|
|
|
|
loop {
|
|
let n = reader
|
|
.read(&mut chunk_buf)
|
|
.map_err(|e| anyhow::anyhow!("stream read error: {e}"))?;
|
|
if n == 0 {
|
|
break;
|
|
}
|
|
byte_buf.extend_from_slice(&chunk_buf[..n]);
|
|
let valid_len = match std::str::from_utf8(&byte_buf) {
|
|
Ok(s) => s.len(),
|
|
Err(e) => e.valid_up_to(),
|
|
};
|
|
if valid_len == 0 {
|
|
continue;
|
|
}
|
|
let text = String::from_utf8_lossy(&byte_buf[..valid_len]).into_owned();
|
|
byte_buf.drain(..valid_len);
|
|
|
|
for event in parser.feed(&text) {
|
|
if !on_event(&event) {
|
|
anyhow::bail!("aborted");
|
|
}
|
|
match &event {
|
|
StreamEvent::Usage {
|
|
prompt_tokens,
|
|
completion_tokens,
|
|
..
|
|
} => {
|
|
usage = Some((*prompt_tokens, *completion_tokens));
|
|
}
|
|
StreamEvent::Error(msg) => {
|
|
anyhow::bail!("stream error: {msg}");
|
|
}
|
|
StreamEvent::Done => {
|
|
turn.apply_event(&event);
|
|
turn.done_received = true;
|
|
return Ok((turn.build_assistant_message(), usage));
|
|
}
|
|
_ => turn.apply_event(&event),
|
|
}
|
|
}
|
|
}
|
|
|
|
// The connection closed without an explicit `[DONE]` event. Some
|
|
// providers legitimately omit it, so EOF alone isn't an error —
|
|
// but if it leaves a tool call's arguments as unparsable JSON, the
|
|
// response was truncated mid-generation, not finished. Report that
|
|
// honestly instead of silently double-stringifying the fragment
|
|
// into a tool call that will misbehave (e.g. a `write` call with a
|
|
// half-written file body).
|
|
if let Some((name, err)) = turn.incomplete_tool_call() {
|
|
anyhow::bail!("stream ended before tool call '{name}' arguments were complete: {err}");
|
|
}
|
|
|
|
turn.is_complete = true;
|
|
Ok((turn.build_assistant_message(), usage))
|
|
}
|
|
}
|
|
|
|
/// Resolve the API key for the currently configured provider, falling back
|
|
/// through settings → env var → provider default.
|
|
///
|
|
/// Used by both the main agent turn loop (`spawn.rs`) and subagent provider
|
|
/// resolution (`subagent/provider.rs`) to share the identical fallback chain.
|
|
///
|
|
/// Flow: try `settings.api_keys[provider]` → try `api_key_env` env var →
|
|
/// try `default_api_key` from config → return empty string if all paths
|
|
/// exhausted (callers must check and reject the empty case).
|
|
pub fn resolve_api_key(
|
|
settings: &zesdex_cms::domain::settings::Settings,
|
|
app_config: &zesdex_cms::domain::app_config::AppConfig,
|
|
) -> String {
|
|
let mut api_key = settings
|
|
.api_keys
|
|
.get(&settings.provider)
|
|
.cloned()
|
|
.unwrap_or_default();
|
|
if api_key.is_empty() {
|
|
if let Some(provider_cfg) = app_config.providers.get(&settings.provider) {
|
|
api_key = provider_cfg
|
|
.api_key_env
|
|
.as_ref()
|
|
.and_then(|env| std::env::var(env).ok())
|
|
.or_else(|| provider_cfg.default_api_key.clone())
|
|
.unwrap_or_default();
|
|
}
|
|
}
|
|
api_key
|
|
}
|