diff --git a/src/agentsight/src/interruption/detector.rs b/src/agentsight/src/interruption/detector.rs index 76a862794..3309bc8fd 100644 --- a/src/agentsight/src/interruption/detector.rs +++ b/src/agentsight/src/interruption/detector.rs @@ -20,6 +20,85 @@ fn is_token_limit_finish(reason: Option<&str>) -> bool { matches!(reason, Some("length" | "max_tokens")) } +fn structured_stream_error_text(raw_body: &str) -> Option { + if let Some(error) = structured_error_json_text(raw_body.trim()) { + return Some(error); + } + + raw_body.lines().find_map(|line| { + let payload = line + .trim() + .strip_prefix("data:") + .map(str::trim) + .unwrap_or_else(|| line.trim()); + if payload.is_empty() || payload == "[DONE]" || payload.starts_with("event:") { + return None; + } + structured_error_json_text(payload) + }) +} + +fn structured_error_json_text(candidate: &str) -> Option { + serde_json::from_str::(candidate) + .ok() + .and_then(|value| structured_error_value_text(&value)) +} + +fn structured_error_value_text(value: &serde_json::Value) -> Option { + match value { + serde_json::Value::Array(items) => items.iter().find_map(structured_error_value_text), + serde_json::Value::Object(map) => { + let event_type = map.get("type").and_then(|value| value.as_str()); + let is_error_event = event_type.is_some_and(|kind| { + kind == "error" || kind.ends_with(".failed") || kind.ends_with(".error") + }); + + if let Some(error) = map.get("error") { + return summarize_error_value(error); + } + + let response_error = map + .get("response") + .and_then(|response| response.get("error")) + .and_then(summarize_error_value); + if let Some(error) = response_error { + return Some(error); + } + + if is_error_event { + return summarize_error_value(value); + } + + None + } + _ => None, + } +} + +fn summarize_error_value(value: &serde_json::Value) -> Option { + match value { + serde_json::Value::String(text) => { + let trimmed = text.trim(); + if trimmed.is_empty() { + None + } else { + Some(trimmed.to_string()) + } + } + serde_json::Value::Object(map) => { + let text = ["type", "code", "message"] + .iter() + .filter_map(|key| map.get(*key).and_then(|value| value.as_str())) + .filter(|text| !text.trim().is_empty()) + .map(str::trim) + .collect::>() + .join(" "); + if text.is_empty() { None } else { Some(text) } + } + _ => None, + } +} + /// Configuration for the interruption detector pub struct DetectorConfig { /// Ratio of output_tokens / max_tokens that triggers token_limit (default: 0.95) @@ -101,10 +180,26 @@ impl InterruptionDetector { .and_then(|s| s.parse().ok()) .unwrap_or(200); - // 修复：从 call.response.raw_body 读取响应体，而非 call.metadata（builder 不会写入 metadata） - let error_text = call.error.as_deref().unwrap_or(""); + let is_sse = call + .metadata + .get("is_sse") + .map(|s| s == "true") + .unwrap_or(false); let response_body = call.response.raw_body.as_deref().unwrap_or(""); - let combined_error = format!("{error_text} {response_body}").to_ascii_lowercase(); + let stream_error = if status_code < 400 && is_sse { + structured_stream_error_text(response_body) + } else { + None + }; + let effective_error = call.error.as_deref().or(stream_error.as_deref()); + let error_text = effective_error.unwrap_or(""); + let structured_error = error_text.to_ascii_lowercase(); + let response_error_body = if status_code >= 400 { + response_body + } else { + "" + }; + let combined_error = format!("{error_text} {response_error_body}").to_ascii_lowercase(); let is_context_overflow = combined_error.contains("context_length_exceeded") || combined_error.contains("maximum context length") @@ -130,7 +225,7 @@ impl InterruptionDetector { let detail = serde_json::json!({ "model": call.model, "status_code": status_code, - "error": call.error, + "error": effective_error, }); events.push(InterruptionEvent::new( InterruptionType::AuthError, @@ -155,7 +250,7 @@ impl InterruptionDetector { let detail = serde_json::json!({ "model": call.model, "status_code": status_code, - "error": call.error, + "error": effective_error, }); events.push(InterruptionEvent::new( InterruptionType::RateLimit, @@ -174,14 +269,14 @@ impl InterruptionDetector { // ── 3. NetworkTimeout (408/504 / timeout) ───────────────────────────── if status_code == 408 || status_code == 504 - || combined_error.contains("timeout") - || combined_error.contains("timed out") - || combined_error.contains("deadline exceeded") + || structured_error.contains("timeout") + || structured_error.contains("timed out") + || structured_error.contains("deadline exceeded") { let detail = serde_json::json!({ "model": call.model, "status_code": status_code, - "error": call.error, + "error": effective_error, }); events.push(InterruptionEvent::new( InterruptionType::NetworkTimeout, @@ -208,7 +303,7 @@ impl InterruptionDetector { let detail = serde_json::json!({ "model": call.model, "status_code": status_code, - "error": call.error, + "error": effective_error, }); events.push(InterruptionEvent::new( InterruptionType::ServiceUnavailable, @@ -230,7 +325,7 @@ impl InterruptionDetector { let detail = serde_json::json!({ "model": call.model, "status_code": status_code, - "error": call.error, + "error": effective_error, "input_tokens": call.token_usage.as_ref().map(|u| u.input_tokens), }); events.push(InterruptionEvent::new( @@ -258,7 +353,7 @@ impl InterruptionDetector { let detail = serde_json::json!({ "model": call.model, "finish_reason": "content_filter", - "error": call.error, + "error": effective_error, }); events.push(InterruptionEvent::new( InterruptionType::SafetyFilter, @@ -276,10 +371,10 @@ impl InterruptionDetector { // ── 7. LLM error (non-context HTTP/API errors) ──────────────────────── // 通用兜底：所有 HTTP >= 400 且未被上述规则匹配的错误 - if status_code >= 400 || call.error.is_some() { + if status_code >= 400 || call.error.is_some() || stream_error.is_some() { let detail = serde_json::json!({ "status_code": status_code, - "error": call.error, + "error": effective_error, "model": call.model, }); events.push(InterruptionEvent::new( @@ -300,11 +395,6 @@ impl InterruptionDetector { // 严格条件：SSE 流 + 持续时间 >= 阈值 + 无正常终止标志 + 非 token-limit // 正常终止标志：finish_reason 为 stop/tool_calls/end_turn/tool_use/stop_sequence // token-limit (length/max_tokens) 由 rule 9/10 单独处理 - let is_sse = call - .metadata - .get("is_sse") - .map(|s| s == "true") - .unwrap_or(false); if is_sse && !is_normal_finish(finish_reason) && !is_token_limit_finish(finish_reason) @@ -754,6 +844,114 @@ mod tests { ); } + #[test] + fn test_ignores_timeout_text_in_successful_response_body() { + let detector = InterruptionDetector::default(); + let mut call = make_base_call(); + call.response.raw_body = Some( + r#"{"content":"Run journalctl -u app | grep -i \"timeout\" to inspect timeout logs."}"# + .to_string(), + ); + + let events = detector.detect(&call); + + assert!(events.is_empty()); + } + + #[test] + fn test_ignores_auth_text_in_successful_response_body() { + let detector = InterruptionDetector::default(); + let mut call = make_base_call(); + call.response.raw_body = Some( + r#"{"content":"Use rejectUnauthorized:false only for local TLS smoke tests."}"# + .to_string(), + ); + + let events = detector.detect(&call); + + assert!(events.is_empty()); + } + + #[test] + fn test_ignores_rate_limit_text_in_successful_response_body() { + let detector = InterruptionDetector::default(); + let mut call = make_base_call(); + call.response.raw_body = Some( + r#"{"content":"Document rate limit and too many requests troubleshooting steps."}"# + .to_string(), + ); + + let events = detector.detect(&call); + + assert!(events.is_empty()); + } + + #[test] + fn test_ignores_service_unavailable_text_in_successful_response_body() { + let detector = InterruptionDetector::default(); + let mut call = make_base_call(); + call.response.raw_body = Some( + r#"{"content":"Explain service_unavailable and overloaded model symptoms."}"# + .to_string(), + ); + + let events = detector.detect(&call); + + assert!(events.is_empty()); + } + + #[test] + fn test_ignores_context_overflow_text_in_successful_response_body() { + let detector = InterruptionDetector::default(); + let mut call = make_base_call(); + call.response.raw_body = Some( + r#"{"content":"Compare context window, input is too long, and prompt is too long errors."}"# + .to_string(), + ); + + let events = detector.detect(&call); + + assert!(events.is_empty()); + } + + #[test] + fn test_detects_anthropic_sse_error_event_in_200_stream() { + let detector = InterruptionDetector::default(); + let mut call = make_base_call(); + call.metadata + .insert("is_sse".to_string(), "true".to_string()); + call.duration_ns = 500_000_000; + call.response.raw_body = Some( + "event: error\n\ + data: {\"type\":\"error\",\"error\":{\"type\":\"overloaded_error\",\"message\":\"Overloaded\"}}\n\n" + .to_string(), + ); + + let events = detector.detect(&call); + + assert_eq!(events.len(), 1); + assert_eq!( + events[0].interruption_type, + InterruptionType::ServiceUnavailable + ); + } + + #[test] + fn test_detects_structured_sse_error_object_in_200_stream() { + let detector = InterruptionDetector::default(); + let mut call = make_base_call(); + call.metadata + .insert("is_sse".to_string(), "true".to_string()); + call.duration_ns = 100_000_000; + call.response.raw_body = + Some(r#"data: {"error":{"message":"upstream failed before completion"}}"#.to_string()); + + let events = detector.detect(&call); + + assert_eq!(events.len(), 1); + assert_eq!(events[0].interruption_type, InterruptionType::LlmError); + } + #[test] fn test_detect_service_unavailable_503() { let detector = InterruptionDetector::default(); diff --git a/src/agentsight/src/probes/probes.rs b/src/agentsight/src/probes/probes.rs index 89da014af..a863f0dfe 100644 --- a/src/agentsight/src/probes/probes.rs +++ b/src/agentsight/src/probes/probes.rs @@ -345,7 +345,7 @@ impl Probes { } ChannelPolicy::Sample(n) => { let idx = drop_counter.fetch_add(1, Ordering::Relaxed); - if idx % n == 0 { + if idx.is_multiple_of(n) { if event_tx.try_send(e).is_err() { log::warn!( "Probes event channel full (capacity={}); dropping sampled event", diff --git a/src/agentsight/src/probes/sslsniff.rs b/src/agentsight/src/probes/sslsniff.rs index 6d00ded54..83304ca9d 100644 --- a/src/agentsight/src/probes/sslsniff.rs +++ b/src/agentsight/src/probes/sslsniff.rs @@ -622,7 +622,7 @@ fn find_boringssl_offsets(path: &str) -> Option { hs_matches[0] } else { // Multiple matches: choose the one closest before read_off. - match hs_matches.iter().filter(|&&o| o < read_off).next_back() { + match hs_matches.iter().rfind(|&&o| o < read_off) { Some(&o) => o, None => { if verbose { diff --git a/src/agentsight/src/storage/sqlite/token.rs b/src/agentsight/src/storage/sqlite/token.rs index 4b79e6df6..6c82c9492 100644 --- a/src/agentsight/src/storage/sqlite/token.rs +++ b/src/agentsight/src/storage/sqlite/token.rs @@ -208,7 +208,7 @@ pub fn format_tokens_with_commas(count: u64) -> String { let s = count.to_string(); let mut result = String::new(); for (i, c) in s.chars().enumerate() { - if i > 0 && (s.len() - i) % 3 == 0 { + if i > 0 && (s.len() - i).is_multiple_of(3) { result.push(','); } result.push(c); diff --git a/src/agentsight/src/storage/unified.rs b/src/agentsight/src/storage/unified.rs index d275091ab..a6e029b52 100644 --- a/src/agentsight/src/storage/unified.rs +++ b/src/agentsight/src/storage/unified.rs @@ -272,7 +272,7 @@ impl Storage { // Auto-purge check: trigger every `purge_interval` inserts if self.purge_interval > 0 && self.retention_days > 0 { let count = self.insert_count.fetch_add(1, Ordering::Relaxed) + 1; - if count % self.purge_interval == 0 { + if count.is_multiple_of(self.purge_interval) { if let Err(e) = self.purge_expired() { log::warn!("Auto-purge failed: {e}"); }