MythEclipse/zesdex
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Refactor subagent and workflow domain models; migrate access tiers and events to domain module

Browse Source 
- Moved `AccessTier` and `SubagentEvent` enums to `zesdex_domain::subagent`.
- Consolidated workflow-related types into `zesdex_domain::workflow`.
- Updated references across the codebase to use the new domain models.
- Refactored tool execution logic to utilize a new `ToolExecutor` trait.
- Enhanced `AgentTurnService` to handle tool calls and events more effectively.
- Adjusted API handlers and state management to align with new domain structure.
This commit is contained in:
asepharyana
2026-07-21 06:42:53 +07:00
parent 552bc5bc63
commit 802346f909
31 changed files with 968 additions and 870 deletions
Download Patch File Download Diff File Expand all files Collapse all files
apps/infrastructure/src/llm/provider.rs
+161 -185
View File
@@ -1,13 +1,15 @@
//! Blocking HTTP client for OpenAI/Anthropic-compatible chat completion APIs,
//! Async HTTP client for OpenAI/Anthropic-compatible chat completion APIs,
//! supporting both non-streaming and SSE-streaming requests with automatic retry.
use rand_core::RngCore;
use std::sync::atomic::AtomicBool;
use std::time::Duration;
use std::future::Future;
use zesdex_domain::core::{
ChatMessage, ChatRequest, ChatResponse, SseParser, StreamEvent, StreamOptions, ToolDef,
};
use zesdex_application::ports::ProviderService;
const DEFAULT_BASE_URL: &str = "https://opencode.ai/zen/v1";
const DEFAULT_MODEL: &str = "deepseek-v4-flash-free";
@@ -22,12 +24,10 @@ const REQUEST_TIMEOUT: Duration = Duration::from_secs(600);
fn backoff_seconds(attempt: u32, cap: u64) -> Duration {
let base = 2u64.pow(attempt.saturating_sub(1));
let delay = std::cmp::min(base, cap);
// ±25% jitter
let jitter_factor = 0.75 + (rand_core::OsRng.next_u32() % 51) as f64 / 100.0;
Duration::from_secs_f64(delay as f64 * jitter_factor)
}
/// Is the error an auth / billing failure that retrying won't fix?
pub fn is_auth_error(err_str: &str) -> bool {
let err_lower = err_str.to_lowercase();
(err_str.contains("API error 401")
@@ -54,9 +54,9 @@ fn backoff_for_error(attempt: u32, err_str: &str) -> Duration {
// Client
// ---------------------------------------------------------------------------
/// Blocking HTTP client for a single LLM provider endpoint.
/// Async HTTP client for a single LLM provider endpoint.
pub struct LlmClient {
pub client: reqwest::blocking::Client,
pub client: reqwest::Client,
pub api_key: String,
pub base_url: String,
pub model: String,
@@ -72,7 +72,7 @@ impl LlmClient {
} else {
model
};
let client = match reqwest::blocking::Client::builder()
let client = match reqwest::Client::builder()
.timeout(REQUEST_TIMEOUT)
.connect_timeout(CONNECT_TIMEOUT)
.build()
@@ -84,14 +84,14 @@ impl LlmClient {
retrying without connect timeout",
e,
);
match reqwest::blocking::Client::builder()
match reqwest::Client::builder()
.timeout(REQUEST_TIMEOUT)
.build()
{
Ok(c) => c,
Err(e2) => {
tracing::warn!("also failed: {e2}. using default client");
reqwest::blocking::Client::new()
reqwest::Client::new()
}
}
}
@@ -106,167 +106,12 @@ impl LlmClient {
}
}
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>,
) -> anyhow::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;
if let Some(flag) = abort_flag {
if flag.load(std::sync::atomic::Ordering::Relaxed) {
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 =
(|| -> anyhow::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: 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);
std::thread::sleep(delay);
}
}
}
}
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>,
) -> anyhow::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(true),
stop: None,
stream_options: Some(StreamOptions {
include_usage: true,
}),
tool_choice: None,
top_p: None,
};
let url = format!("{}/chat/completions", self.base_url);
let max_retries_stream = 10;
let mut attempt = 0u32;
loop {
attempt += 1;
let mut captured_content = false;
let mut wrapped = |event: &StreamEvent| -> bool {
match event {
StreamEvent::Token(_) | StreamEvent::Reasoning(_) | StreamEvent::ToolCallDelta { .. } => {
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) || captured_content {
return Err(e);
}
if attempt >= max_retries_stream {
return Err(e);
}
let delay = backoff_for_error(attempt, &err_str);
std::thread::sleep(delay);
}
}
}
}
fn try_stream_once(
async fn try_stream_once(
&self,
req: &ChatRequest,
url: &str,
on_event: &mut dyn FnMut(&StreamEvent) -> bool,
on_event: &mut (dyn FnMut(&StreamEvent) -> bool + Send),
) -> anyhow::Result<(ChatMessage, Option<(u64, u64)>)> {
use std::io::Read;
let mut http_req = self
.client
.post(url)
@@ -276,7 +121,7 @@ impl LlmClient {
http_req.header("Authorization", format!("Bearer {}", self.api_key));
}
let resp = http_req.json(req).send().map_err(|e| {
let mut resp = http_req.json(req).send().await.map_err(|e| {
if e.is_timeout() {
anyhow::anyhow!(
"API request timed out after {REQUEST_TIMEOUT:?}. \
@@ -295,7 +140,7 @@ impl LlmClient {
if !resp.status().is_success() {
let status = resp.status();
let body = resp.text().unwrap_or_default();
let body = resp.text().await.unwrap_or_default();
anyhow::bail!("API error {} from {}: {}", status, self.base_url, body);
}
@@ -324,8 +169,6 @@ impl LlmClient {
name,
arguments_delta,
} => {
// Cap the index to prevent memory exhaustion from
// maliciously large indices.
const MAX_TOOL_CALLS: usize = 64;
let index = usize::min(*index, MAX_TOOL_CALLS.saturating_sub(1));
@@ -380,27 +223,16 @@ impl LlmClient {
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)?;
if n == 0 {
break;
}
byte_buf.extend_from_slice(&chunk_buf[..n]);
while let Some(chunk) = resp.chunk().await? {
byte_buf.extend_from_slice(&chunk);
// Drain any bytes that are not valid UTF-8 to prevent
// infinite loop when a non-UTF-8 sequence is received.
let valid_len = match std::str::from_utf8(&byte_buf) {
Ok(s) => s.len(),
Err(e) => {
let n = e.valid_up_to();
if n == 0 {
// No valid UTF-8 prefix; skip the first byte (likely
// a partial multi-byte sequence or stray byte).
byte_buf.drain(..1);
continue;
}
@@ -432,7 +264,6 @@ impl LlmClient {
return Ok((turn.build_assistant_message(), usage));
}
other => {
tracing::debug!("unhandled stream event type: {other:?}");
turn.apply_event(other);
}
}
@@ -443,8 +274,153 @@ impl LlmClient {
}
}
/// Resolve the API key for the currently configured provider, falling back
/// through settings -> env var -> provider default.
impl ProviderService for LlmClient {
async fn chat(
&self,
messages: &[ChatMessage],
tools: Option<Vec<ToolDef>>,
max_tokens: Option<u32>,
temperature: Option<f32>,
) -> anyhow::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;
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 = async {
let resp = http_req.json(&req).send().await.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().await.unwrap_or_default();
anyhow::bail!("API error {} from {}: {}", status, self.base_url, body);
}
let data: ChatResponse = resp.json().await?;
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))
}.await;
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);
tokio::time::sleep(delay).await;
}
}
}
}
async fn chat_stream(
&self,
messages: &[ChatMessage],
tools: Option<Vec<ToolDef>>,
max_tokens: Option<u32>,
temperature: Option<f32>,
mut on_event: Box<dyn FnMut(&StreamEvent) -> bool + Send>,
) -> anyhow::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(true),
stop: None,
stream_options: Some(StreamOptions {
include_usage: true,
}),
tool_choice: None,
top_p: None,
};
let url = format!("{}/chat/completions", self.base_url);
let max_retries_stream = 10;
let mut attempt = 0u32;
loop {
attempt += 1;
let mut captured_content = false;
let mut wrapped = |event: &StreamEvent| -> bool {
match event {
StreamEvent::Token(_) | StreamEvent::Reasoning(_) | StreamEvent::ToolCallDelta { .. } => {
captured_content = true;
}
_ => {}
}
on_event(event)
};
match self.try_stream_once(&req, &url, &mut wrapped).await {
Ok(result) => return Ok(result),
Err(e) => {
let err_str = e.to_string();
if is_auth_error(&err_str) || captured_content {
return Err(e);
}
if attempt >= max_retries_stream {
return Err(e);
}
let delay = backoff_for_error(attempt, &err_str);
tokio::time::sleep(delay).await;
}
}
}
}
}
pub fn resolve_api_key(
settings: &zesdex_domain::cms::Settings,
app_config: &zesdex_domain::cms::AppConfig,