hanzoai/dev
2
0
Fork 0
mirror of https://github.com/hanzoai/dev.git synced 2026-10-25 03:32:32 +00:00
Code Issues Packages Projects Releases Wiki Activity

enhance(stream): carry request id, classify drops, log exhausted retries

Browse Source
This commit is contained in:
James Peter
2025-11-25 08:46:44 +10:00
parent b9ab1a1211
commit 936cca8f74
6 changed files with 43 additions and 12 deletions
Download Patch File Download Diff File Expand all files Collapse all files
code-rs/code-auto-drive-core/src/auto_coordinator.rs
+1 -1
View File
@@ -2076,7 +2076,7 @@ fn should_retry_with_default_model(err: &anyhow::Error) -> bool {
pub(crate) fn classify_model_error(error: &anyhow::Error) -> RetryDecision {
if let Some(code_err) = find_in_chain::<CodexErr>(error) {
match code_err {
CodexErr::Stream(message, _) => {
CodexErr::Stream(message, _, _) => {
return RetryDecision::RetryAfterBackoff {
reason: format!("model stream error: {message}"),
};
code-rs/core/src/chat_completions.rs
+7 -2
View File
@@ -536,7 +536,11 @@ async fn process_chat_sse<S>(
Ok(Some(Ok(ev))) => ev,
Ok(Some(Err(e))) => {
let _ = tx_event
.send(Err(CodexErr::Stream(e.to_string(), None)))
.send(Err(CodexErr::Stream(
format!("[transport] {e}"),
None,
Some(request_id.clone()),
)))
.await;
return;
}
@@ -567,8 +571,9 @@ async fn process_chat_sse<S>(
Err(_) => {
let _ = tx_event
.send(Err(CodexErr::Stream(
"idle timeout waiting for SSE".into(),
"[idle] timeout waiting for SSE".into(),
None,
Some(request_id.clone()),
)))
.await;
return;
code-rs/core/src/client.rs
+16 -5
View File
@@ -1361,7 +1361,11 @@ async fn process_sse<S>(
Ok(Some(Ok(sse))) => sse,
Ok(Some(Err(e))) => {
debug!("SSE Error: {e:#}");
let event = CodexErr::Stream(e.to_string(), None);
let event = CodexErr::Stream(
format!("[transport] {e}"),
None,
Some(request_id.clone()),
);
let _ = tx_event.send(Err(event)).await;
return;
}
@@ -1396,6 +1400,7 @@ async fn process_sse<S>(
let error = response_error.unwrap_or(CodexErr::Stream(
"stream closed before response.completed".into(),
None,
Some(request_id.clone()),
));
if let Some(manager) = otel_event_manager.as_ref() {
manager.see_event_completed_failed(&error);
@@ -1412,8 +1417,9 @@ async fn process_sse<S>(
Err(_) => {
let _ = tx_event
.send(Err(CodexErr::Stream(
"idle timeout waiting for SSE".into(),
"[idle] timeout waiting for SSE".into(),
None,
Some(request_id.clone()),
)))
.await;
return;
@@ -1648,6 +1654,7 @@ async fn process_sse<S>(
response_error = Some(CodexErr::Stream(
"response.failed event received".to_string(),
None,
Some(request_id.clone()),
));
let error = resp_val.get("error");
@@ -1660,7 +1667,11 @@ async fn process_sse<S>(
} else {
let retry_after = try_parse_retry_after(&error, Utc::now());
let message = error.message.unwrap_or_default();
response_error = Some(CodexErr::Stream(message, retry_after));
response_error = Some(CodexErr::Stream(
message,
retry_after,
Some(request_id.clone()),
));
}
}
Err(e) => {
@@ -2145,7 +2156,7 @@ mod tests {
);
match &events[1] {
Err(CodexErr::Stream(msg, _)) => {
Err(CodexErr::Stream(msg, _, _)) => {
assert_eq!(msg, "stream closed before response.completed")
}
other => panic!("unexpected second event: {other:?}"),
@@ -2178,7 +2189,7 @@ mod tests {
assert_eq!(events.len(), 1);
match &events[0] {
Err(CodexErr::Stream(msg, Some(retry))) => {
Err(CodexErr::Stream(msg, Some(retry), _)) => {
assert_eq!(
msg,
"Rate limit reached for gpt-5.1 in organization org-AAA on tokens per min (TPM): Limit 30000, Used 22999, Requested 12528. Please try again in 11.054s. Visit https://platform.openai.com/account/rate-limits to learn more."
code-rs/core/src/codex.rs
+16 -2
View File
@@ -5699,7 +5699,7 @@ async fn run_turn(
Err(e) => {
// Detect context-window overflow and auto-run a compact summarization once
if !did_auto_compact {
if let CodexErr::Stream(msg, _maybe_delay) = &e {
if let CodexErr::Stream(msg, _maybe_delay, _req_id) = &e {
let lower = msg.to_ascii_lowercase();
let looks_like_context_overflow =
lower.contains("exceeds the context window")
@@ -5769,10 +5769,14 @@ async fn run_turn(
// Use the configured provider-specific stream retry budget.
let max_retries = tc.client.get_provider().stream_max_retries();
let req_id = match &e {
CodexErr::Stream(_, _, req) => req.clone(),
_ => None,
};
if retries < max_retries {
retries += 1;
let (delay, retry_eta) = match e {
CodexErr::Stream(_, Some(ref retry_after)) => {
CodexErr::Stream(_, Some(ref retry_after), _) => {
let eta = format_retry_eta(&retry_after);
(retry_after.delay, eta)
}
@@ -5780,6 +5784,7 @@ async fn run_turn(
};
warn!(
error = %e,
request_id = req_id.as_deref(),
"stream disconnected - retrying turn ({retries}/{max_retries} in {delay:?})...",
);
@@ -5877,6 +5882,14 @@ async fn run_turn(
tokio::time::sleep(delay).await;
} else {
error!(
retries,
max_retries,
auto_compact_attempted = did_auto_compact,
request_id = req_id.as_deref(),
error = %e,
"stream disconnected - retries exhausted"
);
return Err(e);
}
}
@@ -6087,6 +6100,7 @@ async fn try_run_turn(
return Err(CodexErr::Stream(
"stream closed before response.completed".into(),
None,
None,
));
};
code-rs/core/src/codex/compact.rs
+2 -1
View File
@@ -515,7 +515,7 @@ fn looks_like_context_overflow(message: &str) -> bool {
pub(super) fn is_context_overflow_error(err: &CodexErr) -> bool {
match err {
CodexErr::UnexpectedStatus(resp) => looks_like_context_overflow(&resp.body),
CodexErr::Stream(msg, _) => looks_like_context_overflow(msg),
CodexErr::Stream(msg, _, _) => looks_like_context_overflow(msg),
_ => false,
}
}
@@ -666,6 +666,7 @@ async fn drain_to_completed(
return Err(CodexErr::Stream(
"stream closed before response.completed".into(),
None,
None,
));
};
match event {
code-rs/core/src/error.rs
+1 -1
View File
@@ -78,7 +78,7 @@ pub enum CodexErr {
///
/// Optionally includes the requested delay before retrying the turn.
#[error("stream disconnected before completion: {0}")]
Stream(String, Option<RetryAfter>),
Stream(String, Option<RetryAfter>, Option<String>),
#[error("no conversation with id: {0}")]
ConversationNotFound(Uuid),