diff --git a/apps/gateway/src/bin/test_load.rs b/apps/gateway/src/bin/test_load.rs index 2f14d80..32acd13 100644 --- a/apps/gateway/src/bin/test_load.rs +++ b/apps/gateway/src/bin/test_load.rs @@ -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"); diff --git a/apps/infrastructure/src/llm/provider.rs b/apps/infrastructure/src/llm/provider.rs index 2b6bbf5..d9686b0 100644 --- a/apps/infrastructure/src/llm/provider.rs +++ b/apps/infrastructure/src/llm/provider.rs @@ -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, max_tokens: Option, 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( diff --git a/apps/infrastructure/src/tools/mod.rs b/apps/infrastructure/src/tools/mod.rs index ae6131f..6cf03b9 100644 --- a/apps/infrastructure/src/tools/mod.rs +++ b/apps/infrastructure/src/tools/mod.rs @@ -123,6 +123,13 @@ impl ToolCtxBuilder { self.origin = v; self } + pub fn turn_events( + mut self, + v: Arc>>, + ) -> Self { + self.turn_events = Some(v); + self + } pub fn workflow_findings(mut self, v: Option>>>) -> Self { self.workflow_findings = v; self diff --git a/apps/interfaces/tui/src/action.rs b/apps/interfaces/tui/src/action.rs index 88963c1..b4671c8 100644 --- a/apps/interfaces/tui/src/action.rs +++ b/apps/interfaces/tui/src/action.rs @@ -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) => { diff --git a/apps/interfaces/tui/src/state.rs b/apps/interfaces/tui/src/state.rs index 6323ce2..c1a6b28 100644 --- a/apps/interfaces/tui/src/state.rs +++ b/apps/interfaces/tui/src/state.rs @@ -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(), diff --git a/apps/interfaces/tui/src/turn.rs b/apps/interfaces/tui/src/turn.rs index 8a7fdfe..ac603e8 100644 --- a/apps/interfaces/tui/src/turn.rs +++ b/apps/interfaces/tui/src/turn.rs @@ -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, - session_dir: &Path, - workspace_roots: &[PathBuf], - turn_events: &Arc>>, - in_flight: &Arc, - abort: &Arc, +/// Parameters for a turn, grouped to avoid too-many-arguments lint. +struct TurnParams<'a> { + messages: &'a mut Vec, + session_dir: &'a Path, + workspace_roots: &'a [PathBuf], + turn_events: &'a Arc>>, + in_flight: &'a Arc, + abort: &'a Arc, api_key: String, model: String, api_base: Option, -) { - 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>>, event: TurnEvent) {