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

feat(tui): enhance agent turn handling by grouping parameters and improving message management

Browse Source
This commit is contained in:
asepharyana
2026-07-20 15:21:23 +07:00
parent 9bfb95d795
commit 07b217cf4a
6 changed files with 62 additions and 78 deletions
Download Patch File Download Diff File Expand all files Collapse all files
apps/gateway/src/bin/test_load.rs
-1
View File
@@ -1,6 +1,5 @@
use zesdex_domain::cms::AppConfigRepository; use zesdex_domain::cms::AppConfigRepository;
use zesdex_infrastructure::persistence::cms::app_config_repo::JsonAppConfigRepository; use zesdex_infrastructure::persistence::cms::app_config_repo::JsonAppConfigRepository;
use std::path::PathBuf;
fn main() { fn main() {
let base_dir = dirs::home_dir().unwrap().join(".local/share/zesdex"); let base_dir = dirs::home_dir().unwrap().join(".local/share/zesdex");
apps/infrastructure/src/llm/provider.rs
+6 -33
View File
@@ -13,7 +13,7 @@ const DEFAULT_BASE_URL: &str = "https://opencode.ai/zen/v1";
const DEFAULT_MODEL: &str = "deepseek-v4-flash-free"; const DEFAULT_MODEL: &str = "deepseek-v4-flash-free";
pub const DEFAULT_API_KEY: &str = ""; pub const DEFAULT_API_KEY: &str = "";
const CONNECT_TIMEOUT: Duration = Duration::from_secs(10); const CONNECT_TIMEOUT: Duration = Duration::from_secs(10);
const REQUEST_TIMEOUT: Duration = Duration::from_secs(60); const REQUEST_TIMEOUT: Duration = Duration::from_secs(600);
// --------------------------------------------------------------------------- // ---------------------------------------------------------------------------
// Retry helpers // Retry helpers
@@ -209,9 +209,8 @@ impl LlmClient {
temperature: Option<f32>, temperature: Option<f32>,
max_tokens: Option<u32>, max_tokens: Option<u32>,
mut on_event: impl FnMut(&StreamEvent) -> bool, mut on_event: impl FnMut(&StreamEvent) -> bool,
abort_flag: Option<&AtomicBool>, _abort_flag: Option<&AtomicBool>,
) -> anyhow::Result<(ChatMessage, Option<(u64, u64)>)> { ) -> anyhow::Result<(ChatMessage, Option<(u64, u64)>)> {
let tools_for_fallback = tools.clone();
let req = ChatRequest { let req = ChatRequest {
model: self.model.clone(), model: self.model.clone(),
messages: messages.to_vec(), messages: messages.to_vec(),
@@ -228,21 +227,18 @@ impl LlmClient {
}; };
let url = format!("{}/chat/completions", self.base_url); let url = format!("{}/chat/completions", self.base_url);
let max_retries_stream = 5; let max_retries_stream = 3;
let mut attempt = 0u32; let mut attempt = 0u32;
let mut meaningful_content = false;
loop { loop {
attempt += 1; attempt += 1;
let mut captured_content = false; let mut captured_content = false;
let mut wrapped = |event: &StreamEvent| -> bool { let mut wrapped = |event: &StreamEvent| -> bool {
match event { match event {
StreamEvent::Token(_) | StreamEvent::Reasoning(_) => { StreamEvent::Token(_) | StreamEvent::Reasoning(_) | StreamEvent::ToolCallDelta { .. } => {
captured_content = true; captured_content = true;
} }
_ => { _ => {}
tracing::debug!("unhandled stream event type in wrapped closure");
}
} }
on_event(event) on_event(event)
}; };
@@ -250,13 +246,9 @@ impl LlmClient {
Ok(result) => return Ok(result), Ok(result) => return Ok(result),
Err(e) => { Err(e) => {
let err_str = e.to_string(); let err_str = e.to_string();
if is_auth_error(&err_str) { if is_auth_error(&err_str) || captured_content {
return Err(e); return Err(e);
} }
if captured_content || (attempt >= max_retries_stream) {
meaningful_content = captured_content || meaningful_content;
break;
}
if attempt >= max_retries_stream { if attempt >= max_retries_stream {
return Err(e); return Err(e);
} }
@@ -265,25 +257,6 @@ impl LlmClient {
} }
} }
} }
if meaningful_content {
if let Some(flag) = abort_flag {
if flag.load(std::sync::atomic::Ordering::Relaxed) {
return Err(anyhow::anyhow!("aborted"));
}
}
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"
))
} }
fn try_stream_once( fn try_stream_once(
apps/infrastructure/src/tools/mod.rs
+7
View File
@@ -123,6 +123,13 @@ impl ToolCtxBuilder {
self.origin = v; self.origin = v;
self self
} }
pub fn turn_events(
mut self,
v: Arc<Mutex<std::collections::VecDeque<crate::TurnEvent>>>,
) -> Self {
self.turn_events = Some(v);
self
}
pub fn workflow_findings(mut self, v: Option<Arc<Mutex<Vec<String>>>>) -> Self { pub fn workflow_findings(mut self, v: Option<Arc<Mutex<Vec<String>>>>) -> Self {
self.workflow_findings = v; self.workflow_findings = v;
self self
apps/interfaces/tui/src/action.rs
+1
View File
@@ -153,6 +153,7 @@ pub fn apply_action(state: &mut crate::state::AppStateRest, action: Action) {
if let Some(ref mut rt) = state.session_runtime { if let Some(ref mut rt) = state.session_runtime {
rt.usage.tokens_in = rt.usage.tokens_in.saturating_add(tokens_in); rt.usage.tokens_in = rt.usage.tokens_in.saturating_add(tokens_in);
rt.usage.tokens_out = rt.usage.tokens_out.saturating_add(tokens_out); rt.usage.tokens_out = rt.usage.tokens_out.saturating_add(tokens_out);
rt.usage.api_calls = rt.usage.api_calls.saturating_add(1);
} }
} }
zesdex_infrastructure::TurnEvent::Error(msg) => { zesdex_infrastructure::TurnEvent::Error(msg) => {
apps/interfaces/tui/src/state.rs
+1 -1
View File
@@ -975,7 +975,7 @@ impl AppStateRest {
abort_flag: Arc::new(AtomicBool::new(false)), abort_flag: Arc::new(AtomicBool::new(false)),
dir_cache: Arc::new(tokio::sync::RwLock::new(DirCache::new())), dir_cache: Arc::new(tokio::sync::RwLock::new(DirCache::new())),
mention_index: MentionIndex::new(), mention_index: MentionIndex::new(),
session_runtime: None, session_runtime: Some(zesdex_infrastructure::SessionRuntime::new(session_dir.to_path_buf())),
workflow_engine: SimpleWorkflowEngine::new(), workflow_engine: SimpleWorkflowEngine::new(),
transcript_cache: TranscriptCache::new(200), transcript_cache: TranscriptCache::new(200),
scroll: ScrollState::new(), scroll: ScrollState::new(),
apps/interfaces/tui/src/turn.rs
+47 -43
View File
@@ -70,56 +70,62 @@ pub fn spawn_agent_turn(state: &mut AppStateRest, text: String) {
info!("spawning agent turn with {} messages (model: {})", messages.len(), model); info!("spawning agent turn with {} messages (model: {})", messages.len(), model);
std::thread::spawn(move || { std::thread::spawn(move || {
run_turn( let params = TurnParams {
&mut messages, messages: &mut messages,
&session_dir, session_dir: &session_dir,
&workspace_roots, workspace_roots: &workspace_roots,
&turn_events, turn_events: &turn_events,
&in_flight, in_flight: &in_flight,
&abort, abort: &abort,
api_key, api_key,
model, model,
api_base, api_base,
); };
run_turn(params);
}); });
} }
/// The core agent turn — LLM call → tool execution → repeat. /// Parameters for a turn, grouped to avoid too-many-arguments lint.
fn run_turn( struct TurnParams<'a> {
messages: &mut Vec<ChatMessage>, messages: &'a mut Vec<ChatMessage>,
session_dir: &Path, session_dir: &'a Path,
workspace_roots: &[PathBuf], workspace_roots: &'a [PathBuf],
turn_events: &Arc<Mutex<VecDeque<TurnEvent>>>, turn_events: &'a Arc<Mutex<VecDeque<TurnEvent>>>,
in_flight: &Arc<AtomicBool>, in_flight: &'a Arc<AtomicBool>,
abort: &Arc<AtomicBool>, abort: &'a Arc<AtomicBool>,
api_key: String, api_key: String,
model: String, model: String,
api_base: Option<String>, api_base: Option<String>,
) { }
let client = LlmClient::new(api_key, model, api_base);
/// The core agent turn — LLM call → tool execution → repeat.
fn run_turn(params: TurnParams) {
let client = LlmClient::new(params.api_key, params.model, params.api_base);
let tools = all_tools(); let tools = all_tools();
let defs = tool_defs(&tools); let defs = tool_defs(&tools);
// Prepend system message
let sys_msg = ChatMessage::system( let sys_msg = ChatMessage::system(
"You are Zesdex, an AI coding assistant. You have access to various tools via native function calling to help the user. \ "You are Zesdex, an AI coding assistant. You have access to various tools via native function calling to help the user. \
For complex problems, use `seq_think` to reason step-by-step. \
For any non-trivial tasks, you MUST prioritize creating a structured plan (using `plan_enter`) and a list of TODOs (using `todowrite`) BEFORE executing any other tools or modifying files. \ For any non-trivial tasks, you MUST prioritize creating a structured plan (using `plan_enter`) and a list of TODOs (using `todowrite`) BEFORE executing any other tools or modifying files. \
For large multi-step operations or delegating tasks, you MUST prioritize using `workflow_run` (to run a yaml workflow script) or `hive_mind` (to orchestrate multiple agents) to complete the task efficiently. \
Use other tools when necessary. If a tool returns an error, fix the issue before retrying. \ Use other tools when necessary. If a tool returns an error, fix the issue before retrying. \
Respond conversationally, concisely, and helpfully." Respond conversationally, concisely, and helpfully."
.to_string(), .to_string(),
); );
messages.insert(0, sys_msg); params.messages.insert(0, sys_msg);
let tool_ctx = ToolCtx::builder() let tool_ctx = ToolCtx::builder()
.session_dir(session_dir.to_path_buf()) .session_dir(params.session_dir.to_path_buf())
.workspaces(workspace_roots.to_vec()) .workspaces(params.workspace_roots.to_vec())
.turn_events(Arc::clone(params.turn_events))
.build(); .build();
for iteration in 0..50 { for iteration in 0..50 {
if abort.load(Ordering::SeqCst) { if params.abort.load(Ordering::SeqCst) {
abort.store(false, Ordering::SeqCst); params.abort.store(false, Ordering::SeqCst);
push_event(turn_events, TurnEvent::SystemNote { push_event(params.turn_events, TurnEvent::SystemNote {
kind: "info".into(), kind: "info".into(),
message: "Turn aborted by user".into(), message: "Turn aborted by user".into(),
}); });
@@ -129,29 +135,29 @@ fn run_turn(
debug!("agent turn iteration {iteration}"); debug!("agent turn iteration {iteration}");
// Stream the LLM response // Stream the LLM response
push_event(turn_events, TurnEvent::StreamStart); push_event(params.turn_events, TurnEvent::StreamStart);
let result = client.chat_with_tools_streaming( let result = client.chat_with_tools_streaming(
messages, params.messages,
Some(defs.clone()), Some(defs.clone()),
Some(0.7), Some(0.7),
Some(4096), Some(4096),
|event| { |event| {
if abort.load(Ordering::SeqCst) { if params.abort.load(Ordering::SeqCst) {
return false; return false;
} }
match event { match event {
zesdex_domain::core::StreamEvent::Token(s) => { zesdex_domain::core::StreamEvent::Token(s) => {
push_event(turn_events, TurnEvent::StreamToken(s.clone())); push_event(params.turn_events, TurnEvent::StreamToken(s.clone()));
} }
zesdex_domain::core::StreamEvent::Reasoning(s) => { zesdex_domain::core::StreamEvent::Reasoning(s) => {
push_event(turn_events, TurnEvent::StreamReasoning(s.clone())); push_event(params.turn_events, TurnEvent::StreamReasoning(s.clone()));
} }
_ => {} _ => {}
} }
true true
}, },
Some(&abort), Some(params.abort),
); );
match result { match result {
@@ -159,21 +165,21 @@ fn run_turn(
let content = assistant_msg.content.clone().unwrap_or_default(); let content = assistant_msg.content.clone().unwrap_or_default();
let tool_calls = assistant_msg.tool_calls.clone().unwrap_or_default(); let tool_calls = assistant_msg.tool_calls.clone().unwrap_or_default();
push_event(turn_events, TurnEvent::StreamDone(assistant_msg.clone())); push_event(params.turn_events, TurnEvent::StreamDone(assistant_msg.clone()));
if let Some((tokens_in, tokens_out)) = usage { if let Some((tokens_in, tokens_out)) = usage {
push_event(turn_events, TurnEvent::Usage { push_event(params.turn_events, TurnEvent::Usage {
tokens_in, tokens_in,
tokens_out, tokens_out,
}); });
} }
if tool_calls.is_empty() { if tool_calls.is_empty() {
messages.push(ChatMessage::assistant(Some(content))); params.messages.push(ChatMessage::assistant(Some(content)));
break; break;
} }
messages.push(assistant_msg); params.messages.push(assistant_msg);
for tc in &tool_calls { for tc in &tool_calls {
let name = &tc.function.name; let name = &tc.function.name;
@@ -192,7 +198,7 @@ fn run_turn(
let is_error = output.starts_with("Error:"); let is_error = output.starts_with("Error:");
push_event(turn_events, TurnEvent::ToolResult { push_event(params.turn_events, TurnEvent::ToolResult {
tool_call_id: tc.id.clone(), tool_call_id: tc.id.clone(),
tool_name: name.clone(), tool_name: name.clone(),
output: output.clone(), output: output.clone(),
@@ -200,12 +206,12 @@ fn run_turn(
path: None, path: None,
}); });
messages.push(ChatMessage::tool(tc.id.clone(), output.clone())); params.messages.push(ChatMessage::tool(tc.id.clone(), output.clone()));
} }
} }
Err(e) => { Err(e) => {
warn!("LLM call failed: {e}"); warn!("LLM call failed: {e}");
push_event(turn_events, TurnEvent::Error(format!("LLM error: {e}"))); push_event(params.turn_events, TurnEvent::Error(format!("LLM error: {e}")));
break; break;
} }
} }
@@ -213,11 +219,9 @@ fn run_turn(
// Propagate accumulated messages back to session_runtime so the next // Propagate accumulated messages back to session_runtime so the next
// turn starts with the full history (assistant replies + tool results). // turn starts with the full history (assistant replies + tool results).
// TurnEvent::Compacted already exists on the enum and is handled in push_event(params.turn_events, TurnEvent::Compacted(params.messages.clone()));
// action.rs to write back to state.session_runtime.messages. push_event(params.turn_events, TurnEvent::Done);
push_event(turn_events, TurnEvent::Compacted(messages.clone())); mark_done(params.in_flight);
push_event(turn_events, TurnEvent::Done);
mark_done(in_flight);
} }
fn push_event(queue: &Arc<Mutex<VecDeque<TurnEvent>>>, event: TurnEvent) { fn push_event(queue: &Arc<Mutex<VecDeque<TurnEvent>>>, event: TurnEvent) {