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

refactor: DRY cleanup — extract shared helpers, remove duplication across tools, LSP, overlays, and runtime

Browse Source 
Eliminate ~500 lines of duplicate code across 31 files by extracting
shared functions, helpers, and consolidating repeated patterns.

Highlights:
- Toast helpers (toast_info/success/warning/error) on AppStateRest
- push_event() helper for turn-event queue (19 callers consolidated)
- log_write_edit_tool() shared fn (turn.rs + engine.rs ~50 lines saved)
- resolve_api_key() shared fn (spawn.rs + provider.rs)
- LSP call_positional() helper on LspClient
- lsp_cursor_params() shared schema for 4 tool files
-overlay_block() helper for consistent overlay title/border styling
- cycle_selected_index(), path_not_found/a_directory() helpers
- mark_dirty(), save_settings() on AppStateRest
- Remove redundant Err(e) => Err(e) arms in LSP tools
- Consolidate generate_workspace_tree (turn.rs → workspace.rs)
- Simplify background-review wrapper args in auto/mod.rs

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
asepharyana
2026-07-18 03:18:27 +07:00
co-authored by Claude Opus 4.8
parent b02754acd2
commit 9a6ab62562
31 changed files with 628 additions and 870 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/zesdex-backend/src/service/provider.rs
+74 -107
View File
@@ -25,6 +25,7 @@
//! caller-level fallback handles that case.
use anyhow::Result;
use std::sync::atomic::AtomicBool;
use std::time::{Duration, SystemTime, UNIX_EPOCH};
use crate::app::runtime::stream::turn::StreamedTurn;
@@ -43,11 +44,15 @@ const REQUEST_TIMEOUT: Duration = Duration::from_mins(1);
// ---------------------------------------------------------------------------
/// Return a pseudo-random jitter offset in the range [0, range_ns).
///
/// Uses the full epoch nanoseconds (wrapped to u64) instead of the
/// sub-second component so the jitter range scales with `range_ns`
/// rather than being capped at ~1 s.
fn jitter_ns(range_ns: u64) -> u64 {
let nanos = SystemTime::now()
.duration_since(UNIX_EPOCH)
.unwrap_or_default()
.subsec_nanos() as u64;
.as_nanos() as u64;
nanos % range_ns
}
@@ -56,10 +61,11 @@ fn jitter_ns(range_ns: u64) -> u64 {
/// `attempt` is 1-based (first retry → attempt=1).
fn backoff_duration(attempt: u32) -> Duration {
let base_secs = (2u64).pow(attempt).min(30);
let quarter = (base_secs * 250_000_000).max(100_000_000); // ±25%, min 100ms
let offset = jitter_ns(quarter);
// ±25% jitter: sometimes slightly less, sometimes slightly more
let ns = base_secs * 1_000_000_000 + offset - quarter / 2;
let half_range = (base_secs * 250_000_000).max(100_000_000); // 25% of base, min 100ms
let offset = jitter_ns(half_range * 2); // [0, 50% of base)
// ±25% jitter: subtract half_range so the result varies
// between base-25% and base+25%.
let ns = base_secs * 1_000_000_000 + offset - half_range;
Duration::from_nanos(ns)
}
@@ -91,9 +97,9 @@ fn backoff_for_error(attempt: u32, err_str: &str) -> Duration {
if is_rate_limit(err_str) {
// Rate limits need more time to drain — start at 5s instead of 2s.
let base_secs = (5u64 * (2u64).pow(attempt.saturating_sub(1))).min(60);
let quarter = (base_secs * 250_000_000).max(100_000_000);
let offset = jitter_ns(quarter);
let ns = base_secs * 1_000_000_000 + offset - quarter / 2;
let half_range = (base_secs * 250_000_000).max(100_000_000);
let offset = jitter_ns(half_range * 2);
let ns = base_secs * 1_000_000_000 + offset - half_range;
Duration::from_nanos(ns)
} else {
backoff_duration(attempt)
@@ -188,12 +194,15 @@ impl LlmClient {
&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(4096),
temperature: Some(0.7),
max_tokens: Some(max_tokens.unwrap_or(4096)),
temperature: Some(temperature.unwrap_or(0.7)),
tools,
stream: Some(false),
stop: None,
@@ -209,6 +218,12 @@ impl LlmClient {
loop {
attempt += 1;
// Check abort before each retry so user cancellation is
// responsive even during a long non-streaming backoff chain.
if abort_flag.is_some_and(|f| f.load(std::sync::atomic::Ordering::SeqCst)) {
anyhow::bail!("aborted");
}
let mut http_req = self
.client
.post(&url)
@@ -298,7 +313,11 @@ impl LlmClient {
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(),
@@ -374,19 +393,23 @@ impl LlmClient {
// 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 abort_flag.is_some_and(|f| f.load(std::sync::atomic::Ordering::SeqCst)) {
return Err(anyhow::anyhow!("aborted"));
}
tracing::warn!(
"streaming failed after meaningful content — falling back to non-streaming retry",
"streaming failed after meaningful content — falling back to non-streaming call",
);
// Use the same messages; pass None for tools (streaming already
// included them) and let the non-streaming path handle retries.
// The on_event callback is irrelevant for non-streaming, but we
// signal a special synthetic Done event so callers aren't left
// hanging waiting for stream completion.
// Rebuild ChatRequest without streaming options.
return self.chat_with_tools_non_streaming_retry(
// 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,
max_tokens.unwrap_or(4096),
temperature.unwrap_or(0.7),
tools_for_fallback,
max_tokens,
temperature,
abort_flag,
);
}
@@ -395,93 +418,6 @@ impl LlmClient {
))
}
/// Non-streaming fallback used by the streaming method after a partial
/// stream failure. Same retry policy as `chat_with_tools_non_streaming`.
fn chat_with_tools_non_streaming_retry(
&self,
messages: &[ChatMessage],
max_tokens: u32,
temperature: f32,
) -> Result<(ChatMessage, Option<(u64, u64)>)> {
let req = ChatRequest {
model: self.model.clone(),
messages: messages.to_vec(),
max_tokens: Some(max_tokens),
temperature: Some(temperature),
tools: None,
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 = (|| -> 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 [non-streaming fallback]: {}. Retrying {}/{}, sleeping {delay:?}...",
e,
attempt,
max_retries,
);
std::thread::sleep(delay);
}
}
}
}
/// Perform one streaming chat completion request, parsing SSE events until completion.
///
/// Flow: POST → read body in chunks → advance past valid UTF-8 boundary →
@@ -590,3 +526,34 @@ impl LlmClient {
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
}