From 3787709dc4d58a3492994f666df15c1283cfaab7 Mon Sep 17 00:00:00 2001 From: wanghao <1562495626@qq.com> Date: Thu, 9 Jul 2026 11:37:56 +0800 Subject: [PATCH 1/2] fix(sight): avoid false interruption signals Only provider error text and non-2xx response bodies feed runtime interruption classification. Successful assistant output may mention timeout, auth, or rate-limit troubleshooting. Those phrases should not create captured interruption events by themselves. Assisted-by: Codex Signed-off-by: wanghao <1562495626@qq.com> --- src/agentsight/src/interruption/detector.rs | 85 +++++++++++++++++++-- src/agentsight/src/probes/probes.rs | 2 +- src/agentsight/src/probes/sslsniff.rs | 2 +- src/agentsight/src/storage/sqlite/token.rs | 2 +- src/agentsight/src/storage/unified.rs | 2 +- 5 files changed, 84 insertions(+), 9 deletions(-) diff --git a/src/agentsight/src/interruption/detector.rs b/src/agentsight/src/interruption/detector.rs index 76a862794..51e6fe6df 100644 --- a/src/agentsight/src/interruption/detector.rs +++ b/src/agentsight/src/interruption/detector.rs @@ -101,10 +101,15 @@ 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 response_body = call.response.raw_body.as_deref().unwrap_or(""); - let combined_error = format!("{error_text} {response_body}").to_ascii_lowercase(); + 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") @@ -174,9 +179,9 @@ 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, @@ -754,6 +759,76 @@ 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_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}"); } From 9d1e8e2a55e8fe70731eb2aa877f5113ec0997ec Mon Sep 17 00:00:00 2001 From: wanghao <1562495626@qq.com> Date: Thu, 9 Jul 2026 11:38:17 +0800 Subject: [PATCH 2/2] fix(sight): detect SSE stream errors Parse structured error envelopes from successful SSE streams before truncated-stream detection. Keep ordinary HTTP 200 assistant text out of keyword matching. Restore recall for provider errors emitted after the stream status line is committed. The detector only trusts explicit error-shaped events. Assisted-by: Codex Signed-off-by: wanghao <1562495626@qq.com> --- src/agentsight/src/interruption/detector.rs | 151 ++++++++++++++++++-- 1 file changed, 137 insertions(+), 14 deletions(-) diff --git a/src/agentsight/src/interruption/detector.rs b/src/agentsight/src/interruption/detector.rs index 51e6fe6df..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,8 +180,19 @@ impl InterruptionDetector { .and_then(|s| s.parse().ok()) .unwrap_or(200); - 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 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 @@ -135,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, @@ -160,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, @@ -186,7 +276,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::NetworkTimeout, @@ -213,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, @@ -235,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( @@ -263,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, @@ -281,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( @@ -305,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) @@ -829,6 +914,44 @@ mod tests { 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();