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

feat: enhance responsiveness by implementing abort checks in streaming API calls

Browse Source
This commit is contained in:
asepharyana
2026-07-13 14:39:39 +07:00
parent 8388a83af0
commit 0d6f558b2b
3 changed files with 73 additions and 38 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/app/runtime/actions/mod.rs
+35 -29
View File
@@ -1166,7 +1166,11 @@ fn run_agent_turn(
let token_estimate = total_chars / 4;
let max_wire_tokens = tc.context_window;
let wire_msgs = if crate::app::runtime::shortsend::should_shape(token_estimate, max_wire_tokens, prev_shaped) {
// Skip message compaction if abort was requested — the non-streaming
// LLM call for summarization would block without checking abort_flag.
let wire_msgs = if !tc.abort_flag.load(std::sync::atomic::Ordering::SeqCst)
&& crate::app::runtime::shortsend::should_shape(token_estimate, max_wire_tokens, prev_shaped)
{
prev_shaped = true;
let compacted = crate::app::runtime::shortsend::shape_messages(&msgs, token_estimate, max_wire_tokens, false, Some(&tc.client));
@@ -1240,42 +1244,44 @@ fn run_agent_turn(
let (response, final_usage) = match result {
Ok((msg, u)) => (msg, u.or(usage)),
Err(e) => {
// If abort was requested, return immediately.
if tc.abort_flag.load(std::sync::atomic::Ordering::SeqCst) || e.to_string().contains("aborted") {
if let Ok(mut q) = events_q.lock() {
q.push_back(TurnEvent::Error("Generation aborted by user".to_string()));
}
return Ok(());
}
match tc.client.chat_with_tools_non_streaming(&wire_msgs, Some(tc.tdefs.clone())) {
Ok((msg, usage_fb)) => (msg, usage_fb),
Err(api_err) => {
let todo_path = tc.ctx.session_dir.join("todo.md");
let mut has_unfinished = false;
if let Ok(todo_text) = std::fs::read_to_string(&todo_path) {
if todo_text.lines().any(|l| l.trim_start().starts_with("- [ ]")) {
has_unfinished = true;
}
}
if has_unfinished {
todo_retry_count += 1;
if todo_retry_count > MAX_TODO_RETRIES {
anyhow::bail!(
"exhausted {MAX_TODO_RETRIES} todo-retries — giving up on unfinished tasks. \
Edit todo.md manually or ask me to focus on specific items.",
);
}
if let Ok(mut q) = events_q.lock() {
q.push_back(TurnEvent::SystemNote {
kind: "task_retry".to_string(),
message: format!("Network/API error: {api_err}. Auto-retrying to finish tasks... (retry {todo_retry_count}/{MAX_TODO_RETRIES})"),
});
}
std::thread::sleep(std::time::Duration::from_secs(5));
continue;
}
return Err(api_err);
// Streaming-only: no non-streaming fallback.
// Non-streaming blocks up to 1 minute without checking
// abort_flag, making cancellation unresponsive.
// If the API supports streaming (which it must), this
// path handles transient errors via the retry loop below.
let api_err = e;
let todo_path = tc.ctx.session_dir.join("todo.md");
let mut has_unfinished = false;
if let Ok(todo_text) = std::fs::read_to_string(&todo_path) {
if todo_text.lines().any(|l| l.trim_start().starts_with("- [ ]")) {
has_unfinished = true;
}
}
if has_unfinished {
todo_retry_count += 1;
if todo_retry_count > MAX_TODO_RETRIES {
anyhow::bail!(
"exhausted {MAX_TODO_RETRIES} todo-retries — giving up on unfinished tasks. \
Edit todo.md manually or ask me to focus on specific items.",
);
}
if let Ok(mut q) = events_q.lock() {
q.push_back(TurnEvent::SystemNote {
kind: "task_retry".to_string(),
message: format!("Network/API error: {api_err}. Auto-retrying to finish tasks... (retry {todo_retry_count}/{MAX_TODO_RETRIES})"),
});
}
std::thread::sleep(std::time::Duration::from_secs(5));
continue;
}
return Err(api_err);
}
};