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_infrastructure::persistence::cms::app_config_repo::JsonAppConfigRepository;
use std::path::PathBuf;
fn main() {
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";
pub const DEFAULT_API_KEY: &str = "";
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
@@ -209,9 +209,8 @@ impl LlmClient {
temperature: Option<f32>,
max_tokens: Option<u32>,
mut on_event: impl FnMut(&StreamEvent) -> bool,
abort_flag: Option<&AtomicBool>,
_abort_flag: Option<&AtomicBool>,
) -> anyhow::Result<(ChatMessage, Option<(u64, u64)>)> {
let tools_for_fallback = tools.clone();
let req = ChatRequest {
model: self.model.clone(),
messages: messages.to_vec(),
@@ -228,21 +227,18 @@ impl LlmClient {
};
let url = format!("{}/chat/completions", self.base_url);
let max_retries_stream = 5;
let max_retries_stream = 3;
let mut attempt = 0u32;
let mut meaningful_content = false;
loop {
attempt += 1;
let mut captured_content = false;
let mut wrapped = |event: &StreamEvent| -> bool {
match event {
StreamEvent::Token(_) | StreamEvent::Reasoning(_) => {
StreamEvent::Token(_) | StreamEvent::Reasoning(_) | StreamEvent::ToolCallDelta { .. } => {
captured_content = true;
}
_ => {
tracing::debug!("unhandled stream event type in wrapped closure");
}
_ => {}
}
on_event(event)
};
@@ -250,13 +246,9 @@ impl LlmClient {
Ok(result) => return Ok(result),
Err(e) => {
let err_str = e.to_string();
if is_auth_error(&err_str) {
if is_auth_error(&err_str) || captured_content {
return Err(e);
}
if captured_content || (attempt >= max_retries_stream) {
meaningful_content = captured_content || meaningful_content;
break;
}
if attempt >= max_retries_stream {
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(
apps/infrastructure/src/tools/mod.rs
+7
View File
@@ -123,6 +123,13 @@ impl ToolCtxBuilder {
self.origin = v;
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 {
self.workflow_findings = v;
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 {
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.api_calls = rt.usage.api_calls.saturating_add(1);
}
}
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)),
dir_cache: Arc::new(tokio::sync::RwLock::new(DirCache::new())),
mention_index: MentionIndex::new(),
session_runtime: None,
session_runtime: Some(zesdex_infrastructure::SessionRuntime::new(session_dir.to_path_buf())),
workflow_engine: SimpleWorkflowEngine::new(),
transcript_cache: TranscriptCache::new(200),
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);
std::thread::spawn(move || {
run_turn(
&mut messages,
&session_dir,
&workspace_roots,
&turn_events,
&in_flight,
&abort,
let params = TurnParams {
messages: &mut messages,
session_dir: &session_dir,
workspace_roots: &workspace_roots,
turn_events: &turn_events,
in_flight: &in_flight,
abort: &abort,
api_key,
model,
api_base,
);
};
run_turn(params);
});
}
/// The core agent turn — LLM call → tool execution → repeat.
fn run_turn(
messages: &mut Vec<ChatMessage>,
session_dir: &Path,
workspace_roots: &[PathBuf],
turn_events: &Arc<Mutex<VecDeque<TurnEvent>>>,
in_flight: &Arc<AtomicBool>,
abort: &Arc<AtomicBool>,
/// Parameters for a turn, grouped to avoid too-many-arguments lint.
struct TurnParams<'a> {
messages: &'a mut Vec<ChatMessage>,
session_dir: &'a Path,
workspace_roots: &'a [PathBuf],
turn_events: &'a Arc<Mutex<VecDeque<TurnEvent>>>,
in_flight: &'a Arc<AtomicBool>,
abort: &'a Arc<AtomicBool>,
api_key: String,
model: 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 defs = tool_defs(&tools);
// Prepend system message
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. \
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 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. \
Respond conversationally, concisely, and helpfully."
.to_string(),
);
messages.insert(0, sys_msg);
params.messages.insert(0, sys_msg);
let tool_ctx = ToolCtx::builder()
.session_dir(session_dir.to_path_buf())
.workspaces(workspace_roots.to_vec())
.session_dir(params.session_dir.to_path_buf())
.workspaces(params.workspace_roots.to_vec())
.turn_events(Arc::clone(params.turn_events))
.build();
for iteration in 0..50 {
if abort.load(Ordering::SeqCst) {
abort.store(false, Ordering::SeqCst);
push_event(turn_events, TurnEvent::SystemNote {
if params.abort.load(Ordering::SeqCst) {
params.abort.store(false, Ordering::SeqCst);
push_event(params.turn_events, TurnEvent::SystemNote {
kind: "info".into(),
message: "Turn aborted by user".into(),
});
@@ -129,29 +135,29 @@ fn run_turn(
debug!("agent turn iteration {iteration}");
// Stream the LLM response
push_event(turn_events, TurnEvent::StreamStart);
push_event(params.turn_events, TurnEvent::StreamStart);
let result = client.chat_with_tools_streaming(
messages,
params.messages,
Some(defs.clone()),
Some(0.7),
Some(4096),
|event| {
if abort.load(Ordering::SeqCst) {
if params.abort.load(Ordering::SeqCst) {
return false;
}
match event {
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) => {
push_event(turn_events, TurnEvent::StreamReasoning(s.clone()));
push_event(params.turn_events, TurnEvent::StreamReasoning(s.clone()));
}
_ => {}
}
true
},
Some(&abort),
Some(params.abort),
);
match result {
@@ -159,21 +165,21 @@ fn run_turn(
let content = assistant_msg.content.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 {
push_event(turn_events, TurnEvent::Usage {
push_event(params.turn_events, TurnEvent::Usage {
tokens_in,
tokens_out,
});
}
if tool_calls.is_empty() {
messages.push(ChatMessage::assistant(Some(content)));
params.messages.push(ChatMessage::assistant(Some(content)));
break;
}
messages.push(assistant_msg);
params.messages.push(assistant_msg);
for tc in &tool_calls {
let name = &tc.function.name;
@@ -192,7 +198,7 @@ fn run_turn(
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_name: name.clone(),
output: output.clone(),
@@ -200,12 +206,12 @@ fn run_turn(
path: None,
});
messages.push(ChatMessage::tool(tc.id.clone(), output.clone()));
params.messages.push(ChatMessage::tool(tc.id.clone(), output.clone()));
}
}
Err(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;
}
}
@@ -213,11 +219,9 @@ fn run_turn(
// Propagate accumulated messages back to session_runtime so the next
// turn starts with the full history (assistant replies + tool results).
// TurnEvent::Compacted already exists on the enum and is handled in
// action.rs to write back to state.session_runtime.messages.
push_event(turn_events, TurnEvent::Compacted(messages.clone()));
push_event(turn_events, TurnEvent::Done);
mark_done(in_flight);
push_event(params.turn_events, TurnEvent::Compacted(params.messages.clone()));
push_event(params.turn_events, TurnEvent::Done);
mark_done(params.in_flight);
}
fn push_event(queue: &Arc<Mutex<VecDeque<TurnEvent>>>, event: TurnEvent) {