fix(sight): avoid false interruption signals

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<String> {
+    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<String> {
+    serde_json::from_str::<serde_json::Value>(candidate)
+        .ok()
+        .and_then(|value| structured_error_value_text(&value))
+}
+
+fn structured_error_value_text(value: &serde_json::Value) -> Option<String> {
+    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<String> {
+    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::<Vec<_>>()
+                .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();

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",

src/agentsight/src/probes/sslsniff.rs

@@ -622,7 +622,7 @@ fn find_boringssl_offsets(path: &str) -> Option<BoringSslOffsets> {
         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 {

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);

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}");
                 }