From dcd6f81eb77c4b579b631b5fa3631dd0515b1a65 Mon Sep 17 00:00:00 2001 From: wanghao <1562495626@qq.com> Date: Mon, 6 Jul 2026 15:22:45 +0800 Subject: [PATCH 1/9] feat(sight): add conversation grader api Add a rule-based conversation evaluation pipeline with stable input hashing, SQLite persistence, and manual API endpoints. This keeps the MVP deterministic over captured GenAI and interruption evidence so repeated evaluations can reuse completed runs. LLM and agent graders remain reserved extension points. Assisted-by: Codex Signed-off-by: wanghao <1562495626@qq.com> --- src/agentsight/src/grader.rs | 19 + src/agentsight/src/grader/evidence.rs | 304 +++++++++++++ src/agentsight/src/grader/input.rs | 159 +++++++ src/agentsight/src/grader/rule.rs | 593 ++++++++++++++++++++++++++ src/agentsight/src/grader/storage.rs | 315 ++++++++++++++ src/agentsight/src/grader/types.rs | 372 ++++++++++++++++ src/agentsight/src/lib.rs | 1 + src/agentsight/src/server/handlers.rs | 160 +++++++ src/agentsight/src/server/mod.rs | 2 + 9 files changed, 1925 insertions(+) create mode 100644 src/agentsight/src/grader.rs create mode 100644 src/agentsight/src/grader/evidence.rs create mode 100644 src/agentsight/src/grader/input.rs create mode 100644 src/agentsight/src/grader/rule.rs create mode 100644 src/agentsight/src/grader/storage.rs create mode 100644 src/agentsight/src/grader/types.rs diff --git a/src/agentsight/src/grader.rs b/src/agentsight/src/grader.rs new file mode 100644 index 000000000..eb616a205 --- /dev/null +++ b/src/agentsight/src/grader.rs @@ -0,0 +1,19 @@ +//! Conversation quality evaluation for AgentSight. +//! +//! The MVP is a manual, rule-based grader for conversation snapshots. + +mod evidence; +pub mod input; +pub mod rule; +pub mod storage; +pub mod types; + +pub use input::{EvaluationInput, load_conversation_input}; +pub use rule::RuleGrader; +pub use storage::EvaluationStore; +pub use types::{ + EvaluationDimension, EvaluationFinding, EvaluationMetadata, EvaluationRef, EvaluationRequest, + EvaluationResponse, EvaluationResult, EvaluationRunRecord, EvaluationStatus, EvidenceDeeplink, + EvidenceTarget, EvidenceType, GraderError, GraderType, RULE_GRADER_VERSION, RootCause, + TargetType, Verdict, +}; diff --git a/src/agentsight/src/grader/evidence.rs b/src/agentsight/src/grader/evidence.rs new file mode 100644 index 000000000..9d1336322 --- /dev/null +++ b/src/agentsight/src/grader/evidence.rs @@ -0,0 +1,304 @@ +//! Evidence reference helpers for grader dimensions and findings. + +use super::input::EvaluationInput; +use super::types::{EvaluationRef, EvidenceDeeplink, EvidenceTarget, EvidenceType}; +use crate::storage::sqlite::InterruptionRecord; +use crate::storage::sqlite::genai::TraceEventDetail; + +pub(super) fn has_usable_output(event: &TraceEventDetail) -> bool { + if let Some(raw) = event.output_messages.as_deref() { + return raw_contains_content(raw); + } + + event.output_tokens > 0 +} + +pub(super) fn looks_like_tool_failure(event: &TraceEventDetail) -> bool { + event + .output_messages + .as_deref() + .is_some_and(contains_tool_failure_signal) + || event + .input_messages + .as_deref() + .is_some_and(contains_structured_tool_failure_signal) +} + +fn contains_tool_failure_signal(raw: &str) -> bool { + if contains_structured_tool_failure_signal(raw) { + return true; + } + + let text = raw.to_ascii_lowercase(); + text.contains("tool_call_response") + && (text.contains("\"error\"") + || text.contains("traceback") + || text.contains("exception") + || text.contains("failed")) +} + +fn contains_structured_tool_failure_signal(raw: &str) -> bool { + serde_json::from_str::(raw) + .map(|value| json_has_tool_failure(&value)) + .unwrap_or(false) +} + +fn json_has_tool_failure(value: &serde_json::Value) -> bool { + match value { + serde_json::Value::Array(items) => items.iter().any(json_has_tool_failure), + serde_json::Value::Object(map) => { + let is_tool_response = map + .get("type") + .and_then(|value| value.as_str()) + .is_some_and(|kind| matches!(kind, "tool_call_response" | "tool_result")) + || map.contains_key("tool_call_response"); + + if is_tool_response && tool_response_has_error(map) { + return true; + } + + map.values().any(json_has_tool_failure) + } + _ => false, + } +} + +fn tool_response_has_error(map: &serde_json::Map) -> bool { + if map.get("is_error").and_then(|value| value.as_bool()) == Some(true) { + return true; + } + + ["response", "content", "error"] + .iter() + .any(|key| map.get(*key).is_some_and(value_has_error_signal)) +} + +fn value_has_error_signal(value: &serde_json::Value) -> bool { + match value { + serde_json::Value::String(text) => text_has_error_signal(text), + serde_json::Value::Array(items) => items.iter().any(value_has_error_signal), + serde_json::Value::Object(map) => { + map.get("is_error").and_then(|value| value.as_bool()) == Some(true) + || map + .values() + .any(|nested_value| value_has_error_signal(nested_value)) + } + _ => false, + } +} + +fn text_has_error_signal(text: &str) -> bool { + let lower = text.to_ascii_lowercase(); + lower.contains("traceback") + || lower.contains("exception") + || lower.contains("failed") + || lower.contains("exit code 1") + || lower.contains("no such file or directory") + || lower.contains("permission denied") + || lower.contains("command not found") + || lower.contains("\"error\"") + || lower.contains("error:") +} + +pub(super) fn first_event_refs(input: &EvaluationInput, label: &str) -> Vec { + input + .events + .first() + .map(|event| vec![genai_ref(&input.target_id, event, label)]) + .unwrap_or_default() +} + +pub(super) fn genai_ref( + conversation_id: &str, + event: &TraceEventDetail, + label: &str, +) -> EvaluationRef { + let id = event + .call_id + .clone() + .unwrap_or_else(|| format!("genai-event-{}", event.id)); + EvaluationRef { + evidence_type: EvidenceType::GenaiEvent, + id, + label: label.to_string(), + severity: event.interruption_type.clone(), + target: EvidenceTarget { + conversation_id: conversation_id.to_string(), + trace_id: event.trace_id.clone(), + call_id: event.call_id.clone(), + step_id: None, + }, + deeplink: Some(EvidenceDeeplink { + route: "/atif".to_string(), + query: serde_json::json!({ + "type": "conversation", + "id": conversation_id, + "highlight_call_id": &event.call_id, + }), + }), + metadata: serde_json::json!({ + "event_id": event.id, + "model": &event.model, + "status": &event.status, + }), + } +} + +pub(super) fn interruption_ref( + conversation_id: &str, + record: &InterruptionRecord, +) -> EvaluationRef { + EvaluationRef { + evidence_type: EvidenceType::Interruption, + id: record.interruption_id.clone(), + label: record.interruption_type.clone(), + severity: Some(record.severity.clone()), + target: EvidenceTarget { + conversation_id: conversation_id.to_string(), + trace_id: record.trace_id.clone(), + call_id: record.call_id.clone(), + step_id: None, + }, + deeplink: Some(EvidenceDeeplink { + route: "/atif".to_string(), + query: serde_json::json!({ + "type": "conversation", + "id": conversation_id, + "highlight_call_id": &record.call_id, + "interruption_id": &record.interruption_id, + }), + }), + metadata: serde_json::json!({ + "occurred_at_ns": record.occurred_at_ns, + "detail": &record.detail, + "resolved": record.resolved, + }), + } +} + +fn raw_contains_content(raw: &str) -> bool { + let trimmed = raw.trim(); + if trimmed.is_empty() || trimmed == "[]" || trimmed == "{}" { + return false; + } + serde_json::from_str::(trimmed) + .map(|value| json_has_text(&value)) + .unwrap_or_else(|_| !trimmed.is_empty()) +} + +fn json_has_text(value: &serde_json::Value) -> bool { + match value { + serde_json::Value::String(text) => !text.trim().is_empty(), + serde_json::Value::Array(values) => values.iter().any(json_has_text), + serde_json::Value::Object(map) => map.iter().any(|(key, value)| { + matches!( + key.as_str(), + "content" | "text" | "message" | "output" | "response" + ) && json_has_text(value) + || key == "parts" && json_has_text(value) + || key == "Text" && json_has_text(value) + || key == "Reasoning" && json_has_text(value) + }), + _ => false, + } +} + +#[cfg(test)] +mod tests { + use super::*; + + fn event( + input_messages: Option<&str>, + output_messages: Option<&str>, + event_json: Option<&str>, + ) -> TraceEventDetail { + TraceEventDetail { + id: 1, + call_id: Some("call-1".to_string()), + start_timestamp_ns: 100, + end_timestamp_ns: Some(200), + model: Some("test-model".to_string()), + input_tokens: 10, + output_tokens: 10, + total_tokens: 20, + input_messages: input_messages.map(str::to_string), + output_messages: output_messages.map(str::to_string), + system_instructions: None, + agent_name: Some("agent".to_string()), + process_name: None, + pid: Some(1), + user_query: Some("do work".to_string()), + event_json: event_json.map(str::to_string), + trace_id: Some("trace-1".to_string()), + conversation_id: Some("conv-1".to_string()), + cache_read_tokens: None, + status: Some("complete".to_string()), + interruption_type: None, + } + } + + #[test] + fn ignores_historical_tool_failure_text_in_raw_event_json() { + let event = event( + Some(r#"[{"role":"user","content":"write a Linux troubleshooting guide"}]"#), + Some(r#"[{"role":"assistant","content":"step 1: check the process"}]"#), + Some( + r#"{"request":{"messages":[{"role":"user","content":"tool_call_response: {\"error\":\"failed\", \"traceback\":\"FileNotFoundError\"}"}]},"response":{"messages":[{"role":"assistant","content":"step 1: check the process"}]},"error":null}"#, + ), + ); + + assert!(!looks_like_tool_failure(&event)); + } + + #[test] + fn detects_tool_failure_in_current_assistant_output() { + let event = event( + Some(r#"[{"role":"user","content":"run the tool"}]"#), + Some( + r#"[{"role":"assistant","content":"tool_call_response: {\"error\":\"failed to read config\", \"traceback\":\"FileNotFoundError\"}"}]"#, + ), + None, + ); + + assert!(looks_like_tool_failure(&event)); + } + + #[test] + fn detects_tool_failure_in_structured_input_tool_result() { + let event = event( + Some( + r#"[{"role":"user","parts":[{"type":"tool_call_response","id":"toolu_1","response":{"content":"Exit code 1\ncat: /tmp/missing.txt: No such file or directory","is_error":true}}]}]"#, + ), + Some(r#"[{"role":"assistant","parts":[{"type":"text","content":"file missing"}]}]"#), + None, + ); + + assert!(looks_like_tool_failure(&event)); + } + + #[test] + fn detects_tool_failure_from_nested_is_error_flag() { + let event = event( + Some( + r#"[{"role":"user","parts":[{"type":"tool_call_response","id":"toolu_1","response":{"is_error":true}}]}]"#, + ), + Some(r#"[{"role":"assistant","parts":[{"type":"text","content":"tool failed"}]}]"#), + None, + ); + + assert!(looks_like_tool_failure(&event)); + } + + #[test] + fn ignores_tool_failure_text_in_user_prompt() { + let event = event( + Some( + r#"[{"role":"user","content":"please quote this: tool_call_response: {\"error\":\"failed\", \"traceback\":\"FileNotFoundError\"}"}]"#, + ), + Some(r#"[{"role":"assistant","content":"quoted text omitted"}]"#), + None, + ); + + assert!(!looks_like_tool_failure(&event)); + } +} diff --git a/src/agentsight/src/grader/input.rs b/src/agentsight/src/grader/input.rs new file mode 100644 index 000000000..ca6fb2e95 --- /dev/null +++ b/src/agentsight/src/grader/input.rs @@ -0,0 +1,159 @@ +//! Snapshot loading and stable input hashing for grader runs. + +use std::path::Path; + +use serde::Serialize; +use sha2::{Digest, Sha256}; + +use super::types::{GraderError, RULE_GRADER_VERSION, TargetType}; +use crate::storage::sqlite::genai::TraceEventDetail; +use crate::storage::sqlite::{GenAISqliteStore, InterruptionRecord, InterruptionStore}; + +/// Evidence snapshot used by a grader run. +pub struct EvaluationInput { + /// Evaluated target kind. + pub target_type: TargetType, + /// Evaluated conversation id. + pub target_id: String, + /// Captured LLM call rows for the conversation. + pub events: Vec, + /// Captured interruption rows for the conversation. + pub interruptions: Vec, + /// Stable hash over the evaluated snapshot. + pub input_hash: String, + /// True when the snapshot contains pending calls and was forced. + pub evaluated_with_pending: bool, + /// Number of pending LLM calls in the snapshot. + pub pending_call_count: usize, +} + +/// Load a conversation snapshot and compute its stable input hash. +pub fn load_conversation_input( + storage_path: &Path, + conversation_id: &str, + force: bool, +) -> Result { + let genai_store = GenAISqliteStore::new_with_path(storage_path) + .map_err(|error| GraderError::Storage(error.to_string()))?; + let events = genai_store + .get_events_by_conversation(conversation_id) + .map_err(|error| GraderError::Storage(error.to_string()))?; + + if events.is_empty() { + return Err(GraderError::ConversationNotFound( + conversation_id.to_string(), + )); + } + + let pending_call_count = events + .iter() + .filter(|event| event.status.as_deref() == Some("pending")) + .count(); + if pending_call_count > 0 && !force { + return Err(GraderError::ConversationNotReady { + pending_count: pending_call_count, + }); + } + + let interruptions = load_conversation_interruptions(storage_path, conversation_id)?; + let input_hash = compute_input_hash(conversation_id, &events, &interruptions)?; + + Ok(EvaluationInput { + target_type: TargetType::Conversation, + target_id: conversation_id.to_string(), + events, + interruptions, + input_hash, + evaluated_with_pending: pending_call_count > 0, + pending_call_count, + }) +} + +fn load_conversation_interruptions( + storage_path: &Path, + conversation_id: &str, +) -> Result, GraderError> { + if storage_path == Path::new(":memory:") { + return Ok(Vec::new()); + } + let parent = storage_path + .parent() + .filter(|path| !path.as_os_str().is_empty()) + .unwrap_or_else(|| Path::new(".")); + let interruption_path = parent.join("interruption_events.db"); + let store = InterruptionStore::new_with_path(&interruption_path) + .map_err(|error| GraderError::Storage(error.to_string()))?; + store + .list_by_conversation(conversation_id) + .map_err(|error| GraderError::Storage(error.to_string())) +} + +fn compute_input_hash( + conversation_id: &str, + events: &[TraceEventDetail], + interruptions: &[InterruptionRecord], +) -> Result { + #[derive(Serialize)] + struct HashPayload<'a> { + schema: &'static str, + grader_version: &'static str, + conversation_id: &'a str, + events: Vec, + interruptions: Vec, + } + + let payload = HashPayload { + schema: "agentsight-grader-input-v1", + grader_version: RULE_GRADER_VERSION, + conversation_id, + events: events.iter().map(event_hash_value).collect(), + interruptions: interruptions.iter().map(interruption_hash_value).collect(), + }; + let bytes = serde_json::to_vec(&payload)?; + let mut hasher = Sha256::new(); + hasher.update(bytes); + Ok(format!("{:x}", hasher.finalize())) +} + +fn event_hash_value(event: &TraceEventDetail) -> serde_json::Value { + serde_json::json!({ + "id": event.id, + "call_id": &event.call_id, + "start_timestamp_ns": event.start_timestamp_ns, + "end_timestamp_ns": &event.end_timestamp_ns, + "model": &event.model, + "input_tokens": event.input_tokens, + "output_tokens": event.output_tokens, + "total_tokens": event.total_tokens, + "input_messages": &event.input_messages, + "output_messages": &event.output_messages, + "system_instructions": &event.system_instructions, + "agent_name": &event.agent_name, + "process_name": &event.process_name, + "pid": &event.pid, + "user_query": &event.user_query, + "event_json": &event.event_json, + "trace_id": &event.trace_id, + "conversation_id": &event.conversation_id, + "cache_read_tokens": &event.cache_read_tokens, + "status": &event.status, + "interruption_type": &event.interruption_type, + }) +} + +fn interruption_hash_value(record: &InterruptionRecord) -> serde_json::Value { + serde_json::json!({ + "interruption_id": &record.interruption_id, + "session_id": &record.session_id, + "trace_id": &record.trace_id, + "conversation_id": &record.conversation_id, + "call_id": &record.call_id, + "pid": &record.pid, + "agent_name": &record.agent_name, + "interruption_type": &record.interruption_type, + "severity": &record.severity, + "occurred_at_ns": record.occurred_at_ns, + "detail": &record.detail, + "resolved": record.resolved, + }) +} diff --git a/src/agentsight/src/grader/rule.rs b/src/agentsight/src/grader/rule.rs new file mode 100644 index 000000000..07a5da46b --- /dev/null +++ b/src/agentsight/src/grader/rule.rs @@ -0,0 +1,593 @@ +//! Deterministic rule-based conversation grader. + +use super::input::EvaluationInput; +use super::types::{ + EvaluationDimension, EvaluationFinding, EvaluationMetadata, EvaluationRef, EvaluationResult, + GraderType, RULE_GRADER_VERSION, RootCause, Verdict, +}; +use crate::grader::evidence::{ + first_event_refs, genai_ref, has_usable_output, interruption_ref, looks_like_tool_failure, +}; +use uuid::Uuid; + +/// Deterministic MVP grader for conversation snapshots. +pub struct RuleGrader; + +impl RuleGrader { + /// Evaluate a conversation snapshot with the current deterministic rule set. + pub fn evaluate(input: &EvaluationInput) -> EvaluationResult { + let completion = score_completion(input); + let runtime = score_runtime_health(input); + let tool_use = score_tool_use(input); + let efficiency = score_efficiency(input); + let safety = score_safety(input); + + let dimensions = vec![ + completion.clone(), + runtime.clone(), + tool_use.clone(), + efficiency.clone(), + safety.clone(), + ]; + let weighted_score = round_score( + completion.score * 0.35 + + runtime.score * 0.25 + + tool_use.score * 0.20 + + efficiency.score * 0.10 + + safety.score * 0.10, + ); + let findings = build_findings(input, &dimensions); + let root_cause = select_root_cause(input, &dimensions, &findings); + let verdict = select_verdict(input, weighted_score, root_cause, &findings); + + EvaluationResult { + target_type: input.target_type, + target_id: input.target_id.clone(), + run_id: Uuid::new_v4().to_string(), + input_hash: input.input_hash.clone(), + verdict, + score: weighted_score, + summary: summary_for(verdict, root_cause), + root_cause, + recommended_action: recommended_action_for(verdict, root_cause).to_string(), + dimensions, + findings, + metadata: EvaluationMetadata { + evaluated_with_pending: input.evaluated_with_pending, + pending_call_count: input.pending_call_count, + input_event_count: input.events.len(), + grader_type: GraderType::Rule, + grader_version: RULE_GRADER_VERSION.to_string(), + rubric_version: None, + judge_model: None, + prompt_hash: None, + confidence: None, + }, + } + } +} + +fn score_completion(input: &EvaluationInput) -> EvaluationDimension { + let output_refs: Vec = input + .events + .iter() + .filter(|event| has_usable_output(event)) + .map(|event| genai_ref(&input.target_id, event, "Assistant output")) + .collect(); + + if output_refs.is_empty() { + return dimension( + "completion", + 0.0, + "No usable assistant output was captured.", + first_event_refs(input, "No output"), + ); + } + + let pending_penalty = if input.evaluated_with_pending { + 0.15 + } else { + 0.0 + }; + dimension( + "completion", + 1.0 - pending_penalty, + if input.evaluated_with_pending { + "A usable output exists, but the snapshot still has pending calls." + } else { + "A usable assistant output was captured." + }, + output_refs, + ) +} + +fn score_runtime_health(input: &EvaluationInput) -> EvaluationDimension { + let interrupted_refs: Vec = input + .events + .iter() + .filter(|event| event.status.as_deref() == Some("interrupted")) + .map(|event| genai_ref(&input.target_id, event, "Interrupted LLM call")) + .collect(); + if !interrupted_refs.is_empty() { + return dimension( + "runtime_health", + 0.0, + "One or more LLM calls were interrupted.", + interrupted_refs, + ); + } + + let unresolved: Vec = input + .interruptions + .iter() + .filter(|record| !record.resolved) + .map(|record| interruption_ref(&input.target_id, record)) + .collect(); + if !unresolved.is_empty() { + return dimension( + "runtime_health", + 0.45, + "Unresolved interruption signals were captured for this conversation.", + unresolved, + ); + } + + if input.evaluated_with_pending { + return dimension( + "runtime_health", + 0.75, + "The snapshot contains pending calls and may still change.", + first_event_refs(input, "Pending call"), + ); + } + + dimension( + "runtime_health", + 1.0, + "No runtime interruption was detected.", + Vec::new(), + ) +} + +fn score_tool_use(input: &EvaluationInput) -> EvaluationDimension { + let failed_tool_refs: Vec = input + .events + .iter() + .filter(|event| looks_like_tool_failure(event)) + .map(|event| genai_ref(&input.target_id, event, "Tool failure signal")) + .collect(); + if !failed_tool_refs.is_empty() { + return dimension( + "tool_use", + 0.45, + "Tool output contains deterministic error signals.", + failed_tool_refs, + ); + } + + let call_count = input.events.len(); + if call_count > 12 { + return dimension( + "tool_use", + 0.55, + "The conversation required an unusually large number of LLM calls.", + first_event_refs(input, "Repeated calls"), + ); + } + + dimension( + "tool_use", + 1.0, + "No deterministic tool failure was detected.", + Vec::new(), + ) +} + +fn score_efficiency(input: &EvaluationInput) -> EvaluationDimension { + let total_tokens: i64 = input.events.iter().map(|event| event.total_tokens).sum(); + if total_tokens >= 200_000 || input.events.len() > 20 { + return dimension( + "efficiency", + 0.35, + "Token usage or call count is unusually high for a single conversation.", + first_event_refs(input, "High cost"), + ); + } + if total_tokens >= 64_000 || input.events.len() > 10 { + return dimension( + "efficiency", + 0.65, + "Token usage or call count is elevated for a single conversation.", + first_event_refs(input, "Elevated cost"), + ); + } + + dimension( + "efficiency", + 1.0, + "Token usage and call count are within normal bounds.", + Vec::new(), + ) +} + +fn score_safety(input: &EvaluationInput) -> EvaluationDimension { + let safety_refs: Vec = input + .interruptions + .iter() + .filter(|record| record.interruption_type.contains("safety")) + .map(|record| interruption_ref(&input.target_id, record)) + .collect(); + if !safety_refs.is_empty() { + return dimension( + "safety", + 0.0, + "Safety-related interruption signal was captured.", + safety_refs, + ); + } + + dimension( + "safety", + 1.0, + "No safety-specific signal was available or triggered.", + Vec::new(), + ) +} + +fn build_findings( + input: &EvaluationInput, + dimensions: &[EvaluationDimension], +) -> Vec { + let mut findings = Vec::new(); + + if !dimensions + .iter() + .any(|dimension| dimension.name == "completion" && dimension.score > 0.0) + { + findings.push(finding( + "no_final_answer", + "critical", + "The conversation has no usable assistant output.", + first_event_refs(input, "No output"), + )); + } + + for event in input + .events + .iter() + .filter(|event| event.status.as_deref() == Some("interrupted")) + { + findings.push(finding( + "interrupted_main_call", + "critical", + "An LLM call was interrupted before normal completion.", + vec![genai_ref(&input.target_id, event, "Interrupted call")], + )); + } + + if input.evaluated_with_pending { + findings.push(finding( + "partial_snapshot", + "medium", + "Evaluation was forced while LLM calls were still pending.", + first_event_refs(input, "Pending snapshot"), + )); + } + + for record in input.interruptions.iter().filter(|record| !record.resolved) { + findings.push(finding( + &record.interruption_type, + severity_to_finding(&record.severity), + "An unresolved interruption was recorded for this conversation.", + vec![interruption_ref(&input.target_id, record)], + )); + } + + if input.events.iter().any(looks_like_tool_failure) { + findings.push(finding( + "tool_failure", + "medium", + "Tool output contains an error-like signal.", + input + .events + .iter() + .filter(|event| looks_like_tool_failure(event)) + .map(|event| genai_ref(&input.target_id, event, "Tool failure")) + .collect(), + )); + } + + if input.events.len() > 12 { + findings.push(finding( + "loop_detected", + "medium", + "The conversation used many LLM calls and may need loop inspection.", + first_event_refs(input, "Repeated calls"), + )); + } + + findings +} + +fn select_root_cause( + input: &EvaluationInput, + dimensions: &[EvaluationDimension], + findings: &[EvaluationFinding], +) -> RootCause { + let completion_failed = dimensions.iter().any(|dimension| { + dimension.name == "completion" && (dimension.score - 0.0).abs() < f64::EPSILON + }); + if completion_failed { + return RootCause::NoFinalAnswer; + } + if input + .events + .iter() + .any(|event| event.status.as_deref() == Some("interrupted")) + { + return RootCause::InterruptedMainCall; + } + if findings.iter().any(|finding| finding.code == "agent_crash") { + return RootCause::AgentCrash; + } + if findings.iter().any(|finding| { + matches!( + finding.code.as_str(), + "llm_error" | "sse_truncated" | "network_timeout" | "service_unavailable" + ) + }) { + return RootCause::RuntimeError; + } + if findings + .iter() + .any(|finding| finding.code == "tool_failure") + { + return RootCause::ToolFailure; + } + if findings + .iter() + .any(|finding| finding.code.contains("safety")) + { + return RootCause::SafetyRisk; + } + if findings + .iter() + .any(|finding| finding.code == "loop_detected") + { + return RootCause::LoopDetected; + } + if dimensions + .iter() + .any(|dimension| dimension.name == "efficiency" && dimension.score < 0.5) + { + return RootCause::ExcessiveCost; + } + if input.evaluated_with_pending { + return RootCause::PartialSnapshot; + } + RootCause::None +} + +fn select_verdict( + input: &EvaluationInput, + score: f64, + root_cause: RootCause, + findings: &[EvaluationFinding], +) -> Verdict { + if matches!( + root_cause, + RootCause::NoFinalAnswer | RootCause::InterruptedMainCall + ) { + return Verdict::Fail; + } + if score < 0.5 { + return Verdict::Fail; + } + if input.evaluated_with_pending + || score < 0.8 + || findings.iter().any(|finding| finding.severity != "low") + { + return Verdict::Warn; + } + Verdict::Pass +} + +fn dimension( + name: &str, + score: f64, + reason: &str, + evidence_refs: Vec, +) -> EvaluationDimension { + EvaluationDimension { + name: name.to_string(), + score: round_score(score), + verdict: verdict_for_score(score), + reason: reason.to_string(), + evidence_refs, + } +} + +fn finding( + code: &str, + severity: &str, + message: &str, + evidence_refs: Vec, +) -> EvaluationFinding { + EvaluationFinding { + code: code.to_string(), + severity: severity.to_string(), + message: message.to_string(), + evidence_refs, + } +} + +fn verdict_for_score(score: f64) -> Verdict { + if score >= 0.8 { + Verdict::Pass + } else if score >= 0.5 { + Verdict::Warn + } else { + Verdict::Fail + } +} + +fn summary_for(verdict: Verdict, root_cause: RootCause) -> String { + match verdict { + Verdict::Pass => { + "Conversation completed successfully with no deterministic quality issue.".to_string() + } + Verdict::Warn => format!( + "Conversation is usable but needs review for {}.", + root_cause.as_str() + ), + Verdict::Fail => format!( + "Conversation failed quality evaluation because of {}.", + root_cause.as_str() + ), + } +} + +fn recommended_action_for(verdict: Verdict, root_cause: RootCause) -> &'static str { + match (verdict, root_cause) { + (Verdict::Pass, _) => "No immediate action required.", + (_, RootCause::NoFinalAnswer) => { + "Inspect the final LLM call and provider response parsing." + } + (_, RootCause::InterruptedMainCall) => { + "Inspect interruption evidence and retry the conversation after fixing runtime stability." + } + (_, RootCause::AgentCrash) => "Inspect agent health and crash diagnostics before retrying.", + (_, RootCause::RuntimeError) => { + "Inspect provider errors, network stability, and retry behavior." + } + (_, RootCause::ToolFailure) => "Inspect failing tool calls and tool response parsing.", + (_, RootCause::SafetyRisk) => { + "Review safety/security findings before re-running the agent." + } + (_, RootCause::LoopDetected) => "Inspect repeated calls and tighten stopping conditions.", + (_, RootCause::ExcessiveCost) => { + "Review prompts, tool outputs, and token-saving opportunities." + } + (_, RootCause::PartialSnapshot) => { + "Wait for pending calls to complete or keep the partial result marked as forced." + } + (_, RootCause::None) => "Review warnings and supporting evidence.", + } +} + +fn severity_to_finding(severity: &str) -> &'static str { + match severity { + "critical" | "high" => "high", + "medium" => "medium", + _ => "low", + } +} + +fn round_score(score: f64) -> f64 { + let clamped = score.clamp(0.0, 1.0); + (clamped * 100.0).round() / 100.0 +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::grader::types::TargetType; + use crate::storage::sqlite::genai::TraceEventDetail; + + fn event(id: i64, status: &str, output_tokens: i64) -> TraceEventDetail { + TraceEventDetail { + id, + call_id: Some(format!("call-{id}")), + start_timestamp_ns: id * 100, + end_timestamp_ns: Some(id * 100 + 50), + model: Some("gpt-test".to_string()), + input_tokens: 100, + output_tokens, + total_tokens: 100 + output_tokens, + input_messages: None, + output_messages: if output_tokens > 0 { + Some(r#"[{"role":"assistant","content":"done"}]"#.to_string()) + } else { + Some("[]".to_string()) + }, + system_instructions: None, + agent_name: Some("agent".to_string()), + process_name: None, + pid: Some(1), + user_query: Some("do work".to_string()), + event_json: None, + trace_id: Some(format!("trace-{id}")), + conversation_id: Some("conv-1".to_string()), + cache_read_tokens: None, + status: Some(status.to_string()), + interruption_type: None, + } + } + + fn input(events: Vec) -> EvaluationInput { + EvaluationInput { + target_type: TargetType::Conversation, + target_id: "conv-1".to_string(), + events, + interruptions: Vec::new(), + input_hash: "hash".to_string(), + evaluated_with_pending: false, + pending_call_count: 0, + } + } + + #[test] + fn passes_completed_conversation_with_output() { + let result = RuleGrader::evaluate(&input(vec![event(1, "complete", 10)])); + + assert_eq!(result.verdict, Verdict::Pass); + assert_eq!(result.root_cause, RootCause::None); + assert!(result.score >= 0.8); + } + + #[test] + fn fails_when_no_usable_output_exists() { + let result = RuleGrader::evaluate(&input(vec![event(1, "complete", 0)])); + + assert_eq!(result.verdict, Verdict::Fail); + assert_eq!(result.root_cause, RootCause::NoFinalAnswer); + } + + #[test] + fn ignores_event_json_metadata_when_no_output_exists() { + let mut no_output = event(1, "complete", 0); + no_output.output_messages = None; + no_output.event_json = Some( + r#"{"model":"gpt-test","user_query":"do work","response":{"messages":[]}}"#.to_string(), + ); + + let result = RuleGrader::evaluate(&input(vec![no_output])); + + assert_eq!(result.verdict, Verdict::Fail); + assert_eq!(result.root_cause, RootCause::NoFinalAnswer); + } + + #[test] + fn ignores_role_only_output_messages() { + let mut no_output = event(1, "complete", 0); + no_output.output_messages = Some(r#"[{"role":"assistant"}]"#.to_string()); + + let result = RuleGrader::evaluate(&input(vec![no_output])); + + assert_eq!(result.verdict, Verdict::Fail); + assert_eq!(result.root_cause, RootCause::NoFinalAnswer); + } + + #[test] + fn forced_pending_snapshot_warns_without_hard_failure() { + let mut snapshot = input(vec![event(1, "pending", 10)]); + snapshot.evaluated_with_pending = true; + snapshot.pending_call_count = 1; + + let result = RuleGrader::evaluate(&snapshot); + + assert_eq!(result.verdict, Verdict::Warn); + assert_eq!(result.root_cause, RootCause::PartialSnapshot); + assert!(result.metadata.evaluated_with_pending); + } +} diff --git a/src/agentsight/src/grader/storage.rs b/src/agentsight/src/grader/storage.rs new file mode 100644 index 000000000..23d63a429 --- /dev/null +++ b/src/agentsight/src/grader/storage.rs @@ -0,0 +1,315 @@ +//! SQLite persistence for grader evaluation runs. + +use std::path::Path; +use std::sync::Mutex; + +use rusqlite::{Connection, params}; + +use super::types::{ + EvaluationResult, EvaluationRunRecord, EvaluationStatus, GraderError, GraderType, RootCause, + TargetType, Verdict, +}; +use crate::storage::sqlite::create_connection; + +/// SQLite-backed persistence for evaluation runs. +pub struct EvaluationStore { + conn: Mutex, +} + +impl EvaluationStore { + /// Open an evaluation store using the given SQLite path. + /// + /// The MVP stores `evaluation_runs` beside GenAI events so `serve --db` + /// controls both conversation evidence and evaluation results. + pub fn new_with_path(path: &Path) -> Result { + let conn = + create_connection(path).map_err(|error| GraderError::Storage(error.to_string()))?; + let store = EvaluationStore { + conn: Mutex::new(conn), + }; + store.init_tables()?; + Ok(store) + } + + fn init_tables(&self) -> Result<(), GraderError> { + let conn = self + .conn + .lock() + .map_err(|error| GraderError::Storage(error.to_string()))?; + conn.execute_batch( + "CREATE TABLE IF NOT EXISTS evaluation_runs ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + run_id TEXT NOT NULL UNIQUE, + target_type TEXT NOT NULL, + target_id TEXT NOT NULL, + input_hash TEXT NOT NULL, + grader_type TEXT NOT NULL, + grader_version TEXT NOT NULL, + rubric_version TEXT, + judge_model TEXT, + prompt_hash TEXT, + confidence REAL, + status TEXT NOT NULL, + verdict TEXT, + score REAL, + root_cause TEXT, + created_at DATETIME DEFAULT CURRENT_TIMESTAMP, + completed_at DATETIME, + result_json TEXT, + UNIQUE(target_type, target_id, input_hash, grader_type, grader_version) + ); + CREATE INDEX IF NOT EXISTS idx_evaluation_runs_target_latest + ON evaluation_runs(target_type, target_id, created_at DESC); + CREATE INDEX IF NOT EXISTS idx_evaluation_runs_run_id + ON evaluation_runs(run_id);", + ) + .map_err(|error| GraderError::Storage(error.to_string()))?; + Ok(()) + } + + /// Return an existing completed run for the idempotency key, if present. + pub fn find_completed( + &self, + target_type: TargetType, + target_id: &str, + input_hash: &str, + grader_type: GraderType, + grader_version: &str, + ) -> Result, GraderError> { + let conn = self + .conn + .lock() + .map_err(|error| GraderError::Storage(error.to_string()))?; + let mut stmt = conn + .prepare( + "SELECT id, run_id, target_type, target_id, input_hash, grader_type, + grader_version, status, verdict, score, root_cause, created_at, + completed_at, result_json + FROM evaluation_runs + WHERE target_type=?1 + AND target_id=?2 + AND input_hash=?3 + AND grader_type=?4 + AND grader_version=?5 + AND status='completed' + LIMIT 1", + ) + .map_err(|error| GraderError::Storage(error.to_string()))?; + let mut rows = stmt + .query(params![ + target_type.as_str(), + target_id, + input_hash, + grader_type.as_str(), + grader_version, + ]) + .map_err(|error| GraderError::Storage(error.to_string()))?; + match rows + .next() + .map_err(|error| GraderError::Storage(error.to_string()))? + { + Some(row) => raw_to_record( + read_raw_record(row).map_err(|error| GraderError::Storage(error.to_string()))?, + ) + .map(Some), + None => Ok(None), + } + } + + /// Insert a completed evaluation result. + /// + /// Returns `false` when an equivalent completed run already exists. + pub fn insert_completed(&self, result: &EvaluationResult) -> Result { + let conn = self + .conn + .lock() + .map_err(|error| GraderError::Storage(error.to_string()))?; + let result_json = serde_json::to_string(result)?; + let inserted = conn + .execute( + "INSERT OR IGNORE INTO evaluation_runs ( + run_id, target_type, target_id, input_hash, grader_type, grader_version, + rubric_version, judge_model, prompt_hash, confidence, status, verdict, + score, root_cause, completed_at, result_json + ) VALUES (?1,?2,?3,?4,?5,?6,?7,?8,?9,?10,?11,?12,?13,?14,CURRENT_TIMESTAMP,?15)", + params![ + &result.run_id, + result.target_type.as_str(), + &result.target_id, + &result.input_hash, + result.metadata.grader_type.as_str(), + &result.metadata.grader_version, + &result.metadata.rubric_version, + &result.metadata.judge_model, + &result.metadata.prompt_hash, + result.metadata.confidence, + EvaluationStatus::Completed.as_str(), + result.verdict.as_str(), + result.score, + result.root_cause.as_str(), + result_json, + ], + ) + .map_err(|error| GraderError::Storage(error.to_string()))?; + Ok(inserted > 0) + } + + /// Return the latest completed run for a target. + pub fn latest_completed( + &self, + target_type: TargetType, + target_id: &str, + ) -> Result, GraderError> { + let conn = self + .conn + .lock() + .map_err(|error| GraderError::Storage(error.to_string()))?; + let mut stmt = conn + .prepare( + "SELECT id, run_id, target_type, target_id, input_hash, grader_type, + grader_version, status, verdict, score, root_cause, created_at, + completed_at, result_json + FROM evaluation_runs + WHERE target_type=?1 + AND target_id=?2 + AND status='completed' + ORDER BY created_at DESC, id DESC + LIMIT 1", + ) + .map_err(|error| GraderError::Storage(error.to_string()))?; + let mut rows = stmt + .query(params![target_type.as_str(), target_id]) + .map_err(|error| GraderError::Storage(error.to_string()))?; + match rows + .next() + .map_err(|error| GraderError::Storage(error.to_string()))? + { + Some(row) => raw_to_record( + read_raw_record(row).map_err(|error| GraderError::Storage(error.to_string()))?, + ) + .map(Some), + None => Ok(None), + } + } +} + +struct RawEvaluationRunRecord { + id: i64, + run_id: String, + target_type: String, + target_id: String, + input_hash: String, + grader_type: String, + grader_version: String, + status: String, + verdict: Option, + score: Option, + root_cause: Option, + created_at: String, + completed_at: Option, + result_json: Option, +} + +fn read_raw_record(row: &rusqlite::Row<'_>) -> rusqlite::Result { + Ok(RawEvaluationRunRecord { + id: row.get(0)?, + run_id: row.get(1)?, + target_type: row.get(2)?, + target_id: row.get(3)?, + input_hash: row.get(4)?, + grader_type: row.get(5)?, + grader_version: row.get(6)?, + status: row.get(7)?, + verdict: row.get(8)?, + score: row.get(9)?, + root_cause: row.get(10)?, + created_at: row.get(11)?, + completed_at: row.get(12)?, + result_json: row.get(13)?, + }) +} + +fn raw_to_record(raw: RawEvaluationRunRecord) -> Result { + let target_type = parse_target_type(raw.target_type)?; + let grader_type = parse_grader_type(raw.grader_type)?; + let status = parse_status(raw.status)?; + let verdict = raw.verdict.map(parse_verdict).transpose()?; + let root_cause = raw.root_cause.map(parse_root_cause).transpose()?; + let result = raw + .result_json + .as_deref() + .map(serde_json::from_str) + .transpose()?; + + Ok(EvaluationRunRecord { + id: raw.id, + run_id: raw.run_id, + target_type, + target_id: raw.target_id, + input_hash: raw.input_hash, + grader_type, + grader_version: raw.grader_version, + status, + verdict, + score: raw.score, + root_cause, + created_at: raw.created_at, + completed_at: raw.completed_at, + result, + }) +} + +fn parse_target_type(value: String) -> Result { + match value.as_str() { + "conversation" => Ok(TargetType::Conversation), + _ => Err(GraderError::Storage(format!( + "unknown target_type: {value}" + ))), + } +} + +fn parse_grader_type(value: String) -> Result { + match value.as_str() { + "rule" => Ok(GraderType::Rule), + "llm" => Ok(GraderType::Llm), + "agent" => Ok(GraderType::Agent), + _ => Err(GraderError::Storage(format!( + "unknown grader_type: {value}" + ))), + } +} + +fn parse_status(value: String) -> Result { + match value.as_str() { + "completed" => Ok(EvaluationStatus::Completed), + "failed" => Ok(EvaluationStatus::Failed), + _ => Err(GraderError::Storage(format!( + "unknown evaluation status: {value}" + ))), + } +} + +fn parse_verdict(value: String) -> Result { + match value.as_str() { + "pass" => Ok(Verdict::Pass), + "warn" => Ok(Verdict::Warn), + "fail" => Ok(Verdict::Fail), + _ => Err(GraderError::Storage(format!("unknown verdict: {value}"))), + } +} + +fn parse_root_cause(value: String) -> Result { + match value.as_str() { + "none" => Ok(RootCause::None), + "no_final_answer" => Ok(RootCause::NoFinalAnswer), + "interrupted_main_call" => Ok(RootCause::InterruptedMainCall), + "agent_crash" => Ok(RootCause::AgentCrash), + "runtime_error" => Ok(RootCause::RuntimeError), + "tool_failure" => Ok(RootCause::ToolFailure), + "safety_risk" => Ok(RootCause::SafetyRisk), + "loop_detected" => Ok(RootCause::LoopDetected), + "excessive_cost" => Ok(RootCause::ExcessiveCost), + "partial_snapshot" => Ok(RootCause::PartialSnapshot), + _ => Err(GraderError::Storage(format!("unknown root_cause: {value}"))), + } +} diff --git a/src/agentsight/src/grader/types.rs b/src/agentsight/src/grader/types.rs new file mode 100644 index 000000000..dfddd565d --- /dev/null +++ b/src/agentsight/src/grader/types.rs @@ -0,0 +1,372 @@ +//! Public data types shared by grader storage, API handlers, and dashboard clients. + +use serde::{Deserialize, Serialize}; +use thiserror::Error; + +/// Rule-based grader version used by the MVP. +pub const RULE_GRADER_VERSION: &str = "rule-v3"; + +/// Evaluation target kind. +#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "snake_case")] +pub enum TargetType { + /// A grouped Agent conversation. + Conversation, +} + +impl TargetType { + /// Stable string used in SQLite idempotency keys. + pub fn as_str(self) -> &'static str { + match self { + TargetType::Conversation => "conversation", + } + } +} + +/// Evaluator implementation kind. +#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "snake_case")] +pub enum GraderType { + /// Deterministic rules over captured evidence. + Rule, + /// Reserved for future LLM-as-a-Judge. + Llm, + /// Reserved for future Agent-as-a-Judge. + Agent, +} + +impl GraderType { + /// Stable string used in SQLite idempotency keys. + pub fn as_str(self) -> &'static str { + match self { + GraderType::Rule => "rule", + GraderType::Llm => "llm", + GraderType::Agent => "agent", + } + } +} + +/// Top-level evaluation verdict. +#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "snake_case")] +pub enum Verdict { + /// The conversation appears successful. + Pass, + /// The conversation is usable but has notable risks. + Warn, + /// The conversation did not produce a usable outcome. + Fail, +} + +impl Verdict { + /// Stable string used in SQLite summary columns. + pub fn as_str(self) -> &'static str { + match self { + Verdict::Pass => "pass", + Verdict::Warn => "warn", + Verdict::Fail => "fail", + } + } +} + +/// Stored run status. +#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "snake_case")] +pub enum EvaluationStatus { + /// Evaluation completed and `result_json` is available. + Completed, + /// Evaluation failed before a result was produced. + Failed, +} + +impl EvaluationStatus { + /// Stable string used in SQLite summary columns. + pub fn as_str(self) -> &'static str { + match self { + EvaluationStatus::Completed => "completed", + EvaluationStatus::Failed => "failed", + } + } +} + +/// Single primary cause selected for the top-level result. +#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "snake_case")] +pub enum RootCause { + /// No actionable cause was detected. + None, + /// No usable assistant output was captured. + NoFinalAnswer, + /// The primary LLM call was interrupted. + InterruptedMainCall, + /// The agent process crashed. + AgentCrash, + /// Runtime or provider errors were observed. + RuntimeError, + /// Tool calls failed or repeated abnormally. + ToolFailure, + /// Security or safety signal was non-pass. + SafetyRisk, + /// Repeated calls indicate a likely loop. + LoopDetected, + /// Token or call count was unusually high. + ExcessiveCost, + /// The user intentionally evaluated an incomplete snapshot. + PartialSnapshot, +} + +impl RootCause { + /// Stable string used in SQLite summary columns. + pub fn as_str(self) -> &'static str { + match self { + RootCause::None => "none", + RootCause::NoFinalAnswer => "no_final_answer", + RootCause::InterruptedMainCall => "interrupted_main_call", + RootCause::AgentCrash => "agent_crash", + RootCause::RuntimeError => "runtime_error", + RootCause::ToolFailure => "tool_failure", + RootCause::SafetyRisk => "safety_risk", + RootCause::LoopDetected => "loop_detected", + RootCause::ExcessiveCost => "excessive_cost", + RootCause::PartialSnapshot => "partial_snapshot", + } + } +} + +/// Evidence source kind. +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "snake_case")] +pub enum EvidenceType { + /// Captured GenAI LLM call row. + GenaiEvent, + /// Captured interruption row. + Interruption, + /// Reserved for agent-sec security events. + SecurityEvent, + /// Trace-level navigation target. + Trace, + /// Tool call inside a GenAI message payload. + ToolCall, + /// Reserved for persisted ATIF step identifiers. + AtifStep, +} + +/// Navigation target for an evidence reference. +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +pub struct EvidenceTarget { + /// Conversation detail route anchor. + pub conversation_id: String, + /// Optional trace identifier when known. + #[serde(skip_serializing_if = "Option::is_none")] + pub trace_id: Option, + /// Optional LLM call identifier when known. + #[serde(skip_serializing_if = "Option::is_none")] + pub call_id: Option, + /// Reserved ATIF step identifier. + #[serde(skip_serializing_if = "Option::is_none")] + pub step_id: Option, +} + +/// UI deeplink hint for a piece of evidence. +#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] +pub struct EvidenceDeeplink { + /// Client-side route name. + pub route: String, + /// Route parameters and highlight hints. + pub query: serde_json::Value, +} + +/// Evidence reference included in dimensions and findings. +#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] +pub struct EvaluationRef { + /// Evidence source kind. + #[serde(rename = "type")] + pub evidence_type: EvidenceType, + /// Source-local evidence identifier. + pub id: String, + /// Short user-facing label. + pub label: String, + /// Optional severity string from the source. + #[serde(skip_serializing_if = "Option::is_none")] + pub severity: Option, + /// Target used by Dashboard navigation. + pub target: EvidenceTarget, + /// Optional deeplink target for richer Dashboard routes. + #[serde(skip_serializing_if = "Option::is_none")] + pub deeplink: Option, + /// Source-specific structured metadata. + #[serde(default, skip_serializing_if = "serde_json::Value::is_null")] + pub metadata: serde_json::Value, +} + +/// Score and explanation for one evaluation dimension. +#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] +pub struct EvaluationDimension { + /// Dimension key, such as `completion` or `runtime_health`. + pub name: String, + /// Dimension score in `[0, 1]`. + pub score: f64, + /// Dimension-level verdict. + pub verdict: Verdict, + /// Human-readable reason. + pub reason: String, + /// Supporting evidence references. + #[serde(default)] + pub evidence_refs: Vec, +} + +/// Actionable issue found during evaluation. +#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] +pub struct EvaluationFinding { + /// Stable finding code. + pub code: String, + /// `critical`, `high`, `medium`, or `low`. + pub severity: String, + /// Human-readable finding message. + pub message: String, + /// Supporting evidence references. + #[serde(default)] + pub evidence_refs: Vec, +} + +/// Extra metadata that keeps future LLM/Agent judge fields stable. +#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] +pub struct EvaluationMetadata { + /// True when `force=true` evaluated an incomplete snapshot. + pub evaluated_with_pending: bool, + /// Number of pending LLM calls in the evaluated input. + pub pending_call_count: usize, + /// Number of GenAI LLM call rows used as input. + pub input_event_count: usize, + /// Evaluator kind. + pub grader_type: GraderType, + /// Evaluator version. + pub grader_version: String, + /// Reserved rubric version for LLM/Agent judges. + pub rubric_version: Option, + /// Reserved judge model name. + pub judge_model: Option, + /// Reserved prompt hash for judge prompts. + pub prompt_hash: Option, + /// Reserved confidence score. + pub confidence: Option, +} + +/// Full persisted evaluation result. +#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] +pub struct EvaluationResult { + /// Evaluated target kind. + pub target_type: TargetType, + /// Evaluated target identifier. + pub target_id: String, + /// Unique evaluation run identifier. + pub run_id: String, + /// Stable hash of the evaluated input snapshot. + pub input_hash: String, + /// Top-level verdict. + pub verdict: Verdict, + /// Weighted score in `[0, 1]`. + pub score: f64, + /// One-sentence summary. + pub summary: String, + /// Single primary root cause. + pub root_cause: RootCause, + /// Suggested next action. + pub recommended_action: String, + /// Per-dimension scores. + pub dimensions: Vec, + /// Actionable findings. + pub findings: Vec, + /// Additional run metadata. + pub metadata: EvaluationMetadata, +} + +/// Stored run projection returned by persistence queries. +#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] +pub struct EvaluationRunRecord { + /// SQLite row id. + pub id: i64, + /// Unique evaluation run identifier. + pub run_id: String, + /// Evaluated target kind. + pub target_type: TargetType, + /// Evaluated target id. + pub target_id: String, + /// Stable input hash. + pub input_hash: String, + /// Evaluator kind. + pub grader_type: GraderType, + /// Evaluator version. + pub grader_version: String, + /// Stored run status. + pub status: EvaluationStatus, + /// Top-level verdict. + pub verdict: Option, + /// Weighted score. + pub score: Option, + /// Single primary root cause. + pub root_cause: Option, + /// Creation timestamp as stored by SQLite. + pub created_at: String, + /// Completion timestamp as stored by SQLite. + pub completed_at: Option, + /// Full result payload for completed runs. + pub result: Option, +} + +/// Evaluation request body. +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +pub struct EvaluationRequest { + /// Evaluation target kind. + pub target_type: String, + /// Evaluation target identifier. + pub target_id: String, + /// Evaluate incomplete snapshots instead of returning 409. + #[serde(default)] + pub force: bool, +} + +/// Evaluation API response body. +#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] +pub struct EvaluationResponse { + /// Completed evaluation result. + pub result: EvaluationResult, + /// True when an idempotent completed run was reused. + pub reused_existing_run: bool, +} + +/// Errors returned by grader components. +#[derive(Debug, Error)] +pub enum GraderError { + /// The requested conversation has no captured events. + #[error("conversation not found: {0}")] + ConversationNotFound(String), + /// The requested conversation still has pending LLM calls. + #[error("conversation still has {pending_count} pending LLM call(s)")] + ConversationNotReady { + /// Number of pending calls found. + pending_count: usize, + }, + /// The request uses a target kind this MVP does not support. + #[error("unsupported target type: {0}")] + UnsupportedTarget(String), + /// SQLite or storage-layer failure. + #[error("storage error: {0}")] + Storage(String), + /// JSON serialization or deserialization failure. + #[error("json error: {0}")] + Json(#[from] serde_json::Error), +} + +impl GraderError { + /// Stable machine-readable error code for HTTP handlers and tests. + pub fn code(&self) -> &'static str { + match self { + GraderError::ConversationNotFound(_) => "conversation_not_found", + GraderError::ConversationNotReady { .. } => "conversation_not_ready", + GraderError::UnsupportedTarget(_) => "unsupported_target", + GraderError::Storage(_) => "storage_error", + GraderError::Json(_) => "json_error", + } + } +} diff --git a/src/agentsight/src/lib.rs b/src/agentsight/src/lib.rs index d79944b0c..dc9b1f70e 100644 --- a/src/agentsight/src/lib.rs +++ b/src/agentsight/src/lib.rs @@ -52,6 +52,7 @@ pub mod discovery; pub mod event; pub mod ffi; pub mod genai; +pub mod grader; pub mod health; pub mod interruption; pub mod parser; diff --git a/src/agentsight/src/server/handlers.rs b/src/agentsight/src/server/handlers.rs index 5e09893d8..d82633147 100644 --- a/src/agentsight/src/server/handlers.rs +++ b/src/agentsight/src/server/handlers.rs @@ -9,6 +9,10 @@ use serde_json::{Value, json}; use super::AppState; use crate::agent_sec::{AgentSecClient, AgentSecClientError, DaemonResponse}; +use crate::grader::{ + EvaluationRequest, EvaluationResponse, EvaluationStore, GraderError, GraderType, + RULE_GRADER_VERSION, RuleGrader, TargetType, load_conversation_input, +}; use crate::health::AgentHealthStatus; use crate::storage::sqlite::GenAISqliteStore; use crate::storage::sqlite::genai::{ModelTimeseriesBucket, TimeseriesBucket}; @@ -154,6 +158,162 @@ pub async fn get_conversation_events( } } +// ─── Grader endpoints ──────────────────────────────────────────────────────── + +/// Query parameters for GET /api/grader/latest. +#[derive(Debug, Deserialize)] +pub struct GraderLatestQuery { + pub target_type: String, + pub target_id: String, +} + +/// POST /api/grader/evaluate +/// +/// Manually evaluate a conversation snapshot with the rule-based grader. +#[post("/grader/evaluate")] +pub async fn evaluate_grader( + data: web::Data, + body: web::Json, +) -> impl Responder { + let target_type = match parse_grader_target_type(&body.target_type) { + Ok(target_type) => target_type, + Err(error) => return grader_error_response(error), + }; + + if body.target_id.trim().is_empty() { + return HttpResponse::BadRequest() + .json(json!({"error": "bad_request", "message": "target_id is required"})); + } + + let input = match load_conversation_input(&data.storage_path, &body.target_id, body.force) { + Ok(input) => input, + Err(error) => return grader_error_response(error), + }; + let store = match EvaluationStore::new_with_path(&data.storage_path) { + Ok(store) => store, + Err(error) => return grader_error_response(error), + }; + + match store.find_completed( + target_type, + &body.target_id, + &input.input_hash, + GraderType::Rule, + RULE_GRADER_VERSION, + ) { + Ok(Some(record)) => { + if let Some(result) = record.result { + return HttpResponse::Ok().json(EvaluationResponse { + result, + reused_existing_run: true, + }); + } + } + Ok(None) => {} + Err(error) => return grader_error_response(error), + } + + let result = RuleGrader::evaluate(&input); + match store.insert_completed(&result) { + Ok(true) => {} + Ok(false) => { + return match store.find_completed( + target_type, + &body.target_id, + &input.input_hash, + GraderType::Rule, + RULE_GRADER_VERSION, + ) { + Ok(Some(record)) => { + if let Some(result) = record.result { + HttpResponse::Ok().json(EvaluationResponse { + result, + reused_existing_run: true, + }) + } else { + grader_error_response(GraderError::Storage( + "existing evaluation run is missing result_json".to_string(), + )) + } + } + Ok(None) => grader_error_response(GraderError::Storage( + "evaluation insert was ignored but no completed run was found".to_string(), + )), + Err(error) => grader_error_response(error), + }; + } + Err(error) => return grader_error_response(error), + } + + HttpResponse::Ok().json(EvaluationResponse { + result, + reused_existing_run: false, + }) +} + +/// GET /api/grader/latest?target_type=conversation&target_id= +/// +/// Return the latest completed evaluation result for a conversation. +#[get("/grader/latest")] +pub async fn latest_grader( + data: web::Data, + query: web::Query, +) -> impl Responder { + let target_type = match parse_grader_target_type(&query.target_type) { + Ok(target_type) => target_type, + Err(error) => return grader_error_response(error), + }; + + if query.target_id.trim().is_empty() { + return HttpResponse::BadRequest() + .json(json!({"error": "bad_request", "message": "target_id is required"})); + } + + let store = match EvaluationStore::new_with_path(&data.storage_path) { + Ok(store) => store, + Err(error) => return grader_error_response(error), + }; + + match store.latest_completed(target_type, &query.target_id) { + Ok(Some(record)) => HttpResponse::Ok().json(record.result), + Ok(None) => HttpResponse::Ok().json(serde_json::Value::Null), + Err(error) => grader_error_response(error), + } +} + +fn parse_grader_target_type(value: &str) -> Result { + match value { + "conversation" => Ok(TargetType::Conversation), + other => Err(GraderError::UnsupportedTarget(other.to_string())), + } +} + +fn grader_error_response(error: GraderError) -> HttpResponse { + match error { + GraderError::ConversationNotFound(id) => HttpResponse::NotFound().json(json!({ + "error": "conversation_not_found", + "message": format!("Conversation not found: {id}"), + })), + GraderError::ConversationNotReady { pending_count } => HttpResponse::Conflict().json(json!({ + "error": "conversation_not_ready", + "message": "Conversation still has pending LLM calls. Retry after completion or use force=true.", + "pending_call_count": pending_count, + })), + GraderError::UnsupportedTarget(target) => HttpResponse::BadRequest().json(json!({ + "error": "unsupported_target", + "message": format!("Unsupported target_type: {target}. MVP supports only conversation."), + })), + GraderError::Storage(message) => HttpResponse::InternalServerError().json(json!({ + "error": "storage_error", + "message": message, + })), + GraderError::Json(error) => HttpResponse::InternalServerError().json(json!({ + "error": "json_error", + "message": error.to_string(), + })), + } +} + // ─── Agent-name & time-series endpoints ──────────────────────────────────── /// Query parameters shared by agent-name and time-series endpoints diff --git a/src/agentsight/src/server/mod.rs b/src/agentsight/src/server/mod.rs index 956ef77d7..8940ca801 100644 --- a/src/agentsight/src/server/mod.rs +++ b/src/agentsight/src/server/mod.rs @@ -129,6 +129,8 @@ fn configure_routes(cfg: &mut web::ServiceConfig) { .service(handlers::list_traces_by_session) .service(handlers::get_trace_detail) .service(handlers::get_conversation_events) + .service(handlers::evaluate_grader) + .service(handlers::latest_grader) .service(handlers::list_agent_names) .service(handlers::get_timeseries) .service(handlers::export_atif_trace) From 7168f973c38f2da721a89711a30343991c61e7f5 Mon Sep 17 00:00:00 2001 From: wanghao <1562495626@qq.com> Date: Mon, 6 Jul 2026 15:23:22 +0800 Subject: [PATCH 2/9] feat(sight): add grader dashboard controls Add dashboard controls for fetching, running, and displaying persisted conversation evaluations with evidence deeplinks into ATIF. The UI keeps evaluation manual so users decide when to grade a snapshot, while still surfacing saved verdicts in conversation rows. Pending snapshots require explicit force evaluation from the panel. Assisted-by: Codex Signed-off-by: wanghao <1562495626@qq.com> --- .../src/components/EvaluationBadge.tsx | 32 ++ .../src/components/EvaluationPanel.tsx | 336 ++++++++++++++++++ .../dashboard/src/pages/AtifViewerPage.tsx | 27 +- .../dashboard/src/pages/ConversationList.tsx | 95 ++++- .../src/test/AtifViewerPage.test.tsx | 16 +- .../src/test/ConversationList.test.tsx | 67 ++++ .../src/test/EvaluationBadge.test.tsx | 27 ++ .../src/test/EvaluationPanel.test.tsx | 134 +++++++ .../dashboard/src/test/apiClient.test.ts | 55 +++ .../dashboard/src/utils/apiClient.ts | 132 +++++++ 10 files changed, 908 insertions(+), 13 deletions(-) create mode 100644 src/agentsight/dashboard/src/components/EvaluationBadge.tsx create mode 100644 src/agentsight/dashboard/src/components/EvaluationPanel.tsx create mode 100644 src/agentsight/dashboard/src/test/EvaluationBadge.test.tsx create mode 100644 src/agentsight/dashboard/src/test/EvaluationPanel.test.tsx diff --git a/src/agentsight/dashboard/src/components/EvaluationBadge.tsx b/src/agentsight/dashboard/src/components/EvaluationBadge.tsx new file mode 100644 index 000000000..383aacb42 --- /dev/null +++ b/src/agentsight/dashboard/src/components/EvaluationBadge.tsx @@ -0,0 +1,32 @@ +import React from 'react'; +import { EvaluationResult } from '../utils/apiClient'; + +interface EvaluationBadgeProps { + result: Pick | null; +} + +const STYLE_BY_VERDICT = { + pass: 'bg-green-50 text-green-700 border-green-200', + warn: 'bg-amber-50 text-amber-700 border-amber-200', + fail: 'bg-red-50 text-red-700 border-red-200', +} as const; + +const LABEL_BY_VERDICT = { + pass: '通过', + warn: '需复核', + fail: '未通过', +} as const; + +export const EvaluationBadge: React.FC = ({ result }) => { + if (!result) return null; + + return ( + + {LABEL_BY_VERDICT[result.verdict]} + {Math.round(result.score * 100)} + + ); +}; diff --git a/src/agentsight/dashboard/src/components/EvaluationPanel.tsx b/src/agentsight/dashboard/src/components/EvaluationPanel.tsx new file mode 100644 index 000000000..020316a32 --- /dev/null +++ b/src/agentsight/dashboard/src/components/EvaluationPanel.tsx @@ -0,0 +1,336 @@ +import React, { useState } from 'react'; +import { useNavigate } from 'react-router-dom'; +import { + EvaluationNotReadyError, + EvaluationRef, + EvaluationResult, + evaluateConversation, +} from '../utils/apiClient'; +import { EvaluationBadge } from './EvaluationBadge'; + +interface EvaluationPanelProps { + conversationId: string; + initialResult: EvaluationResult | null; + onResult?: (result: EvaluationResult) => void; +} + +export const EvaluationPanel: React.FC = ({ + conversationId, + initialResult, + onResult, +}) => { + const navigate = useNavigate(); + const [result, setResult] = useState(initialResult); + const [expanded, setExpanded] = useState(false); + const [loading, setLoading] = useState(false); + const [pendingCount, setPendingCount] = useState(null); + const [error, setError] = useState(null); + + const runEvaluation = async (force: boolean) => { + setLoading(true); + setError(null); + try { + const response = await evaluateConversation(conversationId, force); + setResult(response.result); + setPendingCount(null); + onResult?.(response.result); + } catch (err) { + if (err instanceof EvaluationNotReadyError) { + setPendingCount(err.pendingCallCount); + } else { + setError(err instanceof Error ? err.message : '质量评估失败'); + } + } finally { + setLoading(false); + } + }; + + const renderEvidenceLinks = (refs: EvaluationRef[]) => { + if (refs.length === 0) return null; + + return ( +
+ {refs.slice(0, 3).map((ref) => { + const path = evidencePath(ref); + return ( + + ); + })} + {refs.length > 3 && +{refs.length - 3}} +
+ ); + }; + + return ( +
+
+
+
+ 质量评估 + +
+ {result ? ( +
+

{summaryText(result)}

+

+ 根因：{rootCauseLabel(result.root_cause)} +

+

{recommendedActionText(result)}

+
+ ) : ( +

暂无质量评估结果。

+ )} +
+ +
+ + {pendingCount !== null && ( +
+ {pendingCount} 个 LLM 调用仍未完成。 + +
+ )} + + {result?.metadata.evaluated_with_pending && ( +
+ 评估时仍有 {result.metadata.pending_call_count} 个 LLM 调用未完成。 +
+ )} + + {error && ( +
+ {error} +
+ )} + + {result && ( +
+ + {expanded && ( +
+
+

评估维度

+
+ {result.dimensions.map((dimension) => ( +
+
+ {dimensionLabel(dimension.name)} + + {Math.round(dimension.score * 100)} + +
+

{reasonText(dimension.reason)}

+ {renderEvidenceLinks(dimension.evidence_refs)} +
+ ))} +
+
+
+

问题发现

+
+ {result.findings.length === 0 ? ( +

未发现问题。

+ ) : ( + result.findings.map((finding) => ( +
+
+ + {findingLabel(finding.code)} + + {severityLabel(finding.severity)} +
+

{findingMessageText(finding.message)}

+ {renderEvidenceLinks(finding.evidence_refs)} +
+ )) + )} +
+
+
+ )} +
+ )} +
+ ); +}; + +function evidencePath(ref: EvaluationRef): string | null { + if (!ref.deeplink) return null; + + const params = new URLSearchParams(); + for (const [key, value] of Object.entries(ref.deeplink.query ?? {})) { + if (value !== null && value !== undefined) { + params.set(key, String(value)); + } + } + const query = params.toString(); + return query ? `${ref.deeplink.route}?${query}` : ref.deeplink.route; +} + +function summaryText(result: EvaluationResult): string { + if (result.verdict === 'pass') { + return '会话已完成，未发现确定性的质量问题。'; + } + if (result.verdict === 'warn') { + return `当前会话可用，但需要复核：${rootCauseLabel(result.root_cause)}。`; + } + return `质量评估未通过，主要原因：${rootCauseLabel(result.root_cause)}。`; +} + +function recommendedActionText(result: EvaluationResult): string { + if (result.verdict === 'pass') { + return '暂无需要立即处理的动作。'; + } + + const actions: Record = { + none: '复核告警项和支撑证据。', + no_final_answer: '检查最后一次 LLM 调用和服务端响应解析。', + interrupted_main_call: '检查中断证据，修复运行稳定性后再重试会话。', + agent_crash: '重试前先检查 Agent 健康状态和崩溃诊断。', + runtime_error: '检查模型服务错误、网络稳定性和重试行为。', + tool_failure: '检查失败的工具调用和工具响应解析。', + safety_risk: '重新运行 Agent 前先复核安全相关发现。', + loop_detected: '检查重复调用并收紧停止条件。', + excessive_cost: '复核提示词、工具输出和 Token 节省空间。', + partial_snapshot: '等待 pending 调用完成，或保留强制评估标记。', + }; + + return actions[result.root_cause]; +} + +function rootCauseLabel(value: EvaluationResult['root_cause']): string { + const labels: Record = { + none: '未发现明确根因', + no_final_answer: '未生成最终回答', + interrupted_main_call: '主调用被中断', + agent_crash: 'Agent 崩溃', + runtime_error: '运行时错误', + tool_failure: '工具调用失败', + safety_risk: '安全风险', + loop_detected: '疑似循环调用', + excessive_cost: '成本过高', + partial_snapshot: '快照未完成', + }; + + return labels[value]; +} + +function dimensionLabel(value: string): string { + const labels: Record = { + completion: '完成度', + runtime_health: '运行健康', + tool_use: '工具使用', + efficiency: '效率', + safety: '安全', + }; + + return labels[value] ?? value; +} + +function reasonText(value: string): string { + const labels: Record = { + 'No usable assistant output was captured.': '未捕获到可用的助手输出。', + 'A usable output exists.': '已捕获可用输出。', + 'A usable output exists, but the snapshot still has pending calls.': '已捕获可用输出，但快照仍有未完成调用。', + 'A usable assistant output was captured.': '已捕获可用的助手输出。', + 'One or more LLM calls were interrupted.': '一个或多个 LLM 调用被中断。', + 'Unresolved interruption signals were captured for this conversation.': '当前会话存在未解决的中断信号。', + 'The snapshot contains pending calls and may still change.': '快照包含未完成调用，结果仍可能变化。', + 'No runtime interruption was detected.': '未检测到运行时中断。', + 'Tool output contains deterministic error signals.': '工具输出包含确定性错误信号。', + 'The conversation required an unusually large number of LLM calls.': '当前会话的 LLM 调用次数异常偏高。', + 'No deterministic tool failure was detected.': '未检测到确定性工具故障。', + 'Token usage or call count is unusually high for a single conversation.': '单个会话的 Token 用量或调用次数异常偏高。', + 'Token usage or call count is elevated for a single conversation.': '单个会话的 Token 用量或调用次数偏高。', + 'Token usage and call count are within normal bounds.': 'Token 用量和调用次数处于正常范围。', + 'Safety-related interruption signal was captured.': '捕获到安全相关中断信号。', + 'No safety-specific signal was available or triggered.': '未发现安全专项信号触发。', + }; + + return labels[value] ?? value; +} + +function findingLabel(value: string): string { + const labels: Record = { + no_final_answer: '未生成最终回答', + interrupted_main_call: '主调用被中断', + partial_snapshot: '快照未完成', + tool_failure: '工具调用失败', + loop_detected: '疑似循环调用', + llm_error: 'LLM 错误', + sse_truncated: 'SSE 流截断', + network_timeout: '网络超时', + service_unavailable: '服务不可用', + agent_crash: 'Agent 崩溃', + }; + + return labels[value] ?? value; +} + +function findingMessageText(value: string): string { + const labels: Record = { + 'The conversation has no usable assistant output.': '会话没有可用的助手输出。', + 'An LLM call was interrupted before normal completion.': 'LLM 调用在正常完成前被中断。', + 'Evaluation was forced while LLM calls were still pending.': '仍有 LLM 调用未完成时执行了强制评估。', + 'Evaluation was forced while calls were pending.': '仍有调用未完成时执行了强制评估。', + 'An unresolved interruption was recorded for this conversation.': '当前会话存在未解决的中断记录。', + 'Tool output contains an error-like signal.': '工具输出包含疑似错误信号。', + 'The conversation used many LLM calls and may need loop inspection.': '会话使用了较多 LLM 调用，可能需要检查循环行为。', + }; + + return labels[value] ?? value; +} + +function severityLabel(value: string): string { + const labels: Record = { + critical: '严重', + high: '高', + medium: '中', + low: '低', + }; + + return labels[value] ?? value; +} + +function evidenceLabel(value: string): string { + const labels: Record = { + 'Assistant output': '助手输出', + 'No output': '无输出', + 'Interrupted LLM call': '中断的 LLM 调用', + 'Interrupted call': '中断调用', + 'Pending call': '未完成调用', + 'Tool failure signal': '工具故障信号', + 'Repeated calls': '重复调用', + 'High cost': '高成本', + 'Elevated cost': '成本偏高', + 'Pending snapshot': '未完成快照', + 'Tool failure': '工具故障', + }; + + return labels[value] ?? findingLabel(value); +} diff --git a/src/agentsight/dashboard/src/pages/AtifViewerPage.tsx b/src/agentsight/dashboard/src/pages/AtifViewerPage.tsx index 8e44b016b..4cf81c7e1 100644 --- a/src/agentsight/dashboard/src/pages/AtifViewerPage.tsx +++ b/src/agentsight/dashboard/src/pages/AtifViewerPage.tsx @@ -28,6 +28,21 @@ function shortId(id: string, len = 20): string { return id.length > len ? id.slice(0, len) + '\u2026' : id; } +function highlightedSections(doc: AtifDocument, callId: string | null): Set { + const sections = new Set(); + if (!callId) return sections; + + for (const step of doc.steps) { + if (step.tool_calls?.some((toolCall) => toolCall.tool_call_id === callId)) { + sections.add(`${step.step_id}-toolcalls`); + } + if (step.observation?.results.some((result) => result.source_call_id === callId)) { + sections.add(`${step.step_id}-observation`); + } + } + return sections; +} + // ─── Strategy label config (shared with TokenSavingsPage) ──────────────────── const STRATEGY_LABELS: Record = { @@ -392,7 +407,14 @@ export const AtifViewerPage: React.FC = () => { const i = id ?? queryId; if (!i.trim()) return; - setSearchParams({ type: t, id: i.trim() }, { replace: true }); + const nextParams: Record = { type: t, id: i.trim() }; + if (searchParams.get('id') === i.trim()) { + const highlightCallId = searchParams.get('highlight_call_id'); + const interruptionId = searchParams.get('interruption_id'); + if (highlightCallId) nextParams.highlight_call_id = highlightCallId; + if (interruptionId) nextParams.interruption_id = interruptionId; + } + setSearchParams(nextParams, { replace: true }); setLoading(true); setError(null); setDoc(null); @@ -406,6 +428,7 @@ export const AtifViewerPage: React.FC = () => { data = await fetchAtifBySession(i.trim()); } setDoc(data); + setExpandedSections(highlightedSections(data, nextParams.highlight_call_id ?? null)); // Fetch savings data for the session if (data.session_id) { fetchSessionSavings(data.session_id) @@ -417,7 +440,7 @@ export const AtifViewerPage: React.FC = () => { } finally { setLoading(false); } - }, [queryType, queryId, setSearchParams]); + }, [queryType, queryId, searchParams, setSearchParams]); // Auto-load from URL on mount useEffect(() => { diff --git a/src/agentsight/dashboard/src/pages/ConversationList.tsx b/src/agentsight/dashboard/src/pages/ConversationList.tsx index 393f3cf4b..c5b17e27b 100644 --- a/src/agentsight/dashboard/src/pages/ConversationList.tsx +++ b/src/agentsight/dashboard/src/pages/ConversationList.tsx @@ -1,4 +1,4 @@ -import React, { useState, useEffect, useCallback, useRef } from 'react'; +import React, { useState, useEffect, useCallback, useMemo, useRef } from 'react'; import { useNavigate, useSearchParams } from 'react-router-dom'; import { LineChart, Line, BarChart, Bar, @@ -6,6 +6,8 @@ import { } from 'recharts'; import { InterruptionBadge } from '../components/InterruptionBadge'; import { InterruptionPanel, ResolvedEventInfo } from '../components/InterruptionPanel'; +import { EvaluationBadge } from '../components/EvaluationBadge'; +import { EvaluationPanel } from '../components/EvaluationPanel'; import { DateTimePicker } from '../components/DateTimePicker'; import { SessionIdHelp } from '../components/SessionIdHelp'; import { @@ -18,6 +20,7 @@ import { fetchInterruptionStats, fetchInterruptionSessionCounts, fetchInterruptionConversationCounts, + fetchLatestEvaluation, fetchTokenSavings, SessionSummary, TraceSummary, @@ -28,6 +31,7 @@ import { InterruptionTypeStat, SessionInterruptionCount, ConversationInterruptionCount, + EvaluationResult, INTERRUPTION_TYPE_CN, } from '../utils/apiClient'; @@ -316,16 +320,63 @@ const TraceSubTable: React.FC = ({ sessionId, conversationIn const [error, setError] = useState(null); const [page, setPage] = useState(0); // 0-based const [expandedTracePanel, setExpandedTracePanel] = useState(null); + const [expandedEvaluationPanel, setExpandedEvaluationPanel] = useState(null); + const [evaluations, setEvaluations] = useState>(new Map()); + const [evaluationLookupDone, setEvaluationLookupDone] = useState>(new Set()); useEffect(() => { setLoading(true); setPage(0); + setEvaluations(new Map()); + setEvaluationLookupDone(new Set()); fetchTraces(sessionId, startNs, endNs) .then(setTraces) .catch((e: Error) => setError(e.message)) .finally(() => setLoading(false)); }, [sessionId, startNs, endNs]); + const totalPages = Math.max(1, Math.ceil(traces.length / PAGE_SIZE)); + const pageTraces = useMemo( + () => traces.slice(page * PAGE_SIZE, (page + 1) * PAGE_SIZE), + [page, traces] + ); + + useEffect(() => { + if (loading || pageTraces.length === 0) return; + + const missing = pageTraces.filter( + (trace) => !evaluationLookupDone.has(trace.conversation_id) + ); + if (missing.length === 0) return; + + let cancelled = false; + Promise.all( + missing.map((trace) => + fetchLatestEvaluation(trace.conversation_id) + .then((result) => result ? [trace.conversation_id, result] as const : null) + .catch(() => null) + ) + ).then((entries) => { + if (cancelled) return; + setEvaluations((prev) => { + const next = new Map(prev); + for (const entry of entries) { + if (entry) next.set(entry[0], entry[1]); + } + return next; + }); + setEvaluationLookupDone((prev) => { + const next = new Set(prev); + for (const trace of missing) next.add(trace.conversation_id); + return next; + }); + }); + + return () => { + cancelled = true; + }; + }, [evaluationLookupDone, loading, pageTraces]); + if (loading) return ( @@ -343,21 +394,19 @@ const TraceSubTable: React.FC = ({ sessionId, conversationIn ); - const totalPages = Math.max(1, Math.ceil(traces.length / PAGE_SIZE)); - const pageTraces = traces.slice(page * PAGE_SIZE, (page + 1) * PAGE_SIZE); - return ( <> {/* Sub-header */} -
+
Conversation ID
用户请求
输入 Token
输出 Token
开始时间
操作
+
质量评估
中断
@@ -375,7 +424,7 @@ const TraceSubTable: React.FC = ({ sessionId, conversationIn -
+
{/* Col 1: Conversation ID */}
@@ -416,6 +465,20 @@ const TraceSubTable: React.FC = ({ sessionId, conversationIn 详情
+
+ +
{(() => { const ic = conversationInterruptionCounts.get(tr.conversation_id); @@ -432,8 +495,24 @@ const TraceSubTable: React.FC = ({ sessionId, conversationIn })()}
- - + + + {/* Trace evaluation panel */} + {expandedEvaluationPanel === tr.conversation_id && ( + + + setEvaluations((prev) => { + const next = new Map(prev); + next.set(tr.conversation_id, result); + return next; + })} + /> + + + )} {/* Trace interruption panel */} {expandedTracePanel === tr.trace_id && ( diff --git a/src/agentsight/dashboard/src/test/AtifViewerPage.test.tsx b/src/agentsight/dashboard/src/test/AtifViewerPage.test.tsx index 226611816..c8e4318ab 100644 --- a/src/agentsight/dashboard/src/test/AtifViewerPage.test.tsx +++ b/src/agentsight/dashboard/src/test/AtifViewerPage.test.tsx @@ -55,13 +55,12 @@ const mockAtifDoc = { tool_calls: [ { tool_call_id: 'tc-1', - tool_name: 'search', + function_name: 'search', arguments: { query: 'greeting' }, - result: 'found: hello', }, ], observation: { - results: [{ output: 'search result' }], + results: [{ source_call_id: 'tc-1', content: 'search result' }], }, metrics: { prompt_tokens: 100, @@ -274,6 +273,17 @@ describe('AtifViewerPage', () => { expect(mockFetchAtifBySession).toHaveBeenCalledWith('sess-from-url'); }); + it('should expand highlighted evidence sections from URL params', async () => { + mockFetchAtifByConversation.mockResolvedValue(mockAtifDoc); + await act(async () => { + renderPage('/atif?type=conversation&id=conv-1&highlight_call_id=tc-1'); + }); + + expect(mockFetchAtifByConversation).toHaveBeenCalledWith('conv-1'); + expect(screen.getByText('search')).toBeInTheDocument(); + expect(screen.getByText('search result')).toBeInTheDocument(); + }); + it('should show Token savings comparison card when savings data exists', async () => { mockFetchAtifBySession.mockResolvedValue(mockAtifDoc); mockFetchSessionSavings.mockResolvedValue({ diff --git a/src/agentsight/dashboard/src/test/ConversationList.test.tsx b/src/agentsight/dashboard/src/test/ConversationList.test.tsx index a125a3fa9..1503432ec 100644 --- a/src/agentsight/dashboard/src/test/ConversationList.test.tsx +++ b/src/agentsight/dashboard/src/test/ConversationList.test.tsx @@ -29,6 +29,16 @@ vi.mock('../utils/apiClient', () => ({ fetchInterruptionSessionCounts: vi.fn(), fetchInterruptionConversationCounts: vi.fn(), fetchTokenSavings: vi.fn(), + fetchLatestEvaluation: vi.fn(), + evaluateConversation: vi.fn(), + EvaluationNotReadyError: class EvaluationNotReadyError extends Error { + pendingCallCount: number; + constructor(message: string, pendingCallCount: number) { + super(message); + this.name = 'EvaluationNotReadyError'; + this.pendingCallCount = pendingCallCount; + } + }, INTERRUPTION_TYPE_CN: { llm_error: 'LLM 错误', sse_truncated: 'SSE 中断', @@ -54,6 +64,14 @@ vi.mock('../components/InterruptionPanel', () => ({ ResolvedEventInfo: undefined, })); +vi.mock('../components/EvaluationBadge', () => ({ + EvaluationBadge: ({ result }: any) => result ? {result.verdict} : null, +})); + +vi.mock('../components/EvaluationPanel', () => ({ + EvaluationPanel: ({ conversationId }: any) =>
质量评估 {conversationId}
, +})); + import { fetchSessions, fetchAgentNames, @@ -64,6 +82,7 @@ import { fetchInterruptionConversationCounts, fetchTokenSavings, fetchTraces, + fetchLatestEvaluation, } from '../utils/apiClient'; import { ConversationList } from '../pages/ConversationList'; @@ -76,6 +95,7 @@ const mockFetchInterruptionSessionCounts = fetchInterruptionSessionCounts as Ret const mockFetchInterruptionConversationCounts = fetchInterruptionConversationCounts as ReturnType; const mockFetchTokenSavings = fetchTokenSavings as ReturnType; const mockFetchTraces = fetchTraces as ReturnType; +const mockFetchLatestEvaluation = fetchLatestEvaluation as ReturnType; function setupMocks() { mockFetchAgentNames.mockResolvedValue(['agent-a', 'agent-b']); @@ -87,6 +107,7 @@ function setupMocks() { mockFetchInterruptionConversationCounts.mockResolvedValue([]); mockFetchTokenSavings.mockResolvedValue({ sessions: [], summary: null, stats_available: false }); mockFetchTraces.mockResolvedValue([]); + mockFetchLatestEvaluation.mockResolvedValue(null); } function renderPage(route = '/') { @@ -251,6 +272,52 @@ describe('ConversationList', () => { expect(screen.getByText('Conversation ID')).toBeInTheDocument(); }); + it('should show latest grader badge and panel for a conversation', async () => { + mockFetchSessions.mockResolvedValue([ + { + session_id: 'sess-graded', + agent_name: 'GradeAgent', + model: 'gpt-4o', + conversation_count: 1, + total_input_tokens: 500, + total_output_tokens: 300, + last_seen_ns: Date.now() * 1_000_000, + }, + ]); + mockFetchTraces.mockResolvedValue([ + { + trace_id: 'trace-graded', + conversation_id: 'conv-graded', + user_query: 'grade this', + total_input_tokens: 300, + total_output_tokens: 200, + start_ns: Date.now() * 1_000_000, + end_ns: Date.now() * 1_000_000, + model: 'gpt-4o', + }, + ]); + mockFetchLatestEvaluation.mockResolvedValue({ + target_id: 'conv-graded', + verdict: 'warn', + score: 0.7, + }); + + await act(async () => { renderPage(); }); + await act(async () => { + fireEvent.click(screen.getByText('查询')); + }); + await act(async () => { + fireEvent.click(screen.getByText('GradeAgent').closest('tr')!); + }); + + expect(mockFetchLatestEvaluation).toHaveBeenCalledWith('conv-graded'); + const badge = await screen.findByTestId('evaluation-badge'); + expect(badge).toHaveTextContent('warn'); + + fireEvent.click(badge.closest('button')!); + expect(screen.getByTestId('evaluation-panel')).toHaveTextContent('conv-graded'); + }); + it('should show agent dropdown with loaded names', async () => { await act(async () => { renderPage(); }); expect(screen.getByText('全部 Agent')).toBeInTheDocument(); diff --git a/src/agentsight/dashboard/src/test/EvaluationBadge.test.tsx b/src/agentsight/dashboard/src/test/EvaluationBadge.test.tsx new file mode 100644 index 000000000..a6d46d7d9 --- /dev/null +++ b/src/agentsight/dashboard/src/test/EvaluationBadge.test.tsx @@ -0,0 +1,27 @@ +import React from 'react'; +import { describe, it, expect } from 'vitest'; +import { render, screen } from '@testing-library/react'; +import { EvaluationBadge } from '../components/EvaluationBadge'; + +describe('EvaluationBadge', () => { + it('renders nothing when result is null', () => { + const { container } = render(); + expect(container.innerHTML).toBe(''); + }); + + it('renders pass verdict with score', () => { + render(); + expect(screen.getByText('通过')).toBeInTheDocument(); + expect(screen.getByText('93')).toBeInTheDocument(); + }); + + it('renders warn verdict', () => { + render(); + expect(screen.getByText('需复核')).toBeInTheDocument(); + }); + + it('renders fail verdict', () => { + render(); + expect(screen.getByText('未通过')).toBeInTheDocument(); + }); +}); diff --git a/src/agentsight/dashboard/src/test/EvaluationPanel.test.tsx b/src/agentsight/dashboard/src/test/EvaluationPanel.test.tsx new file mode 100644 index 000000000..2de8068fb --- /dev/null +++ b/src/agentsight/dashboard/src/test/EvaluationPanel.test.tsx @@ -0,0 +1,134 @@ +import React from 'react'; +import { describe, it, expect, vi, beforeEach } from 'vitest'; +import { render, screen, fireEvent, waitFor } from '@testing-library/react'; +import { MemoryRouter } from 'react-router-dom'; + +vi.mock('../utils/apiClient', async () => { + const actual = await vi.importActual('../utils/apiClient'); + return { + ...actual, + evaluateConversation: vi.fn(), + }; +}); + +import { EvaluationNotReadyError, evaluateConversation, EvaluationResult } from '../utils/apiClient'; +import { EvaluationPanel } from '../components/EvaluationPanel'; + +const mockEvaluate = evaluateConversation as ReturnType; + +const result: EvaluationResult = { + target_type: 'conversation', + target_id: 'conv-1', + run_id: 'run-1', + input_hash: 'hash-1', + verdict: 'warn', + score: 0.72, + summary: 'Conversation is usable but needs review.', + root_cause: 'partial_snapshot', + recommended_action: 'Wait for pending calls to complete.', + dimensions: [ + { + name: 'completion', + score: 0.85, + verdict: 'pass', + reason: 'A usable output exists.', + evidence_refs: [ + { + type: 'genai_event', + id: 'call-1', + label: 'Assistant output', + target: { + conversation_id: 'conv-1', + call_id: 'call-1', + }, + deeplink: { + route: '/atif', + query: { + type: 'conversation', + id: 'conv-1', + highlight_call_id: 'call-1', + }, + }, + metadata: null, + }, + ], + }, + ], + findings: [ + { + code: 'partial_snapshot', + severity: 'medium', + message: 'Evaluation was forced while calls were pending.', + evidence_refs: [], + }, + ], + metadata: { + evaluated_with_pending: true, + pending_call_count: 1, + input_event_count: 2, + grader_type: 'rule', + grader_version: 'rule-v1', + rubric_version: null, + judge_model: null, + prompt_hash: null, + confidence: null, + }, +}; + +beforeEach(() => { + mockEvaluate.mockReset(); +}); + +function renderPanel(ui: React.ReactElement) { + return render({ui}); +} + +describe('EvaluationPanel', () => { + it('renders evaluate button when no result exists', () => { + renderPanel(); + expect(screen.getByText('开始评估')).toBeInTheDocument(); + }); + + it('renders compact summary and pending warning', () => { + renderPanel(); + expect(screen.getByText('需复核')).toBeInTheDocument(); + expect(screen.getByText('72')).toBeInTheDocument(); + expect(screen.getByText('当前会话可用，但需要复核：快照未完成。')).toBeInTheDocument(); + expect(screen.getByText('评估时仍有 1 个 LLM 调用未完成。')).toBeInTheDocument(); + expect(screen.getByText('等待 pending 调用完成，或保留强制评估标记。')).toBeInTheDocument(); + }); + + it('reveals dimensions and findings', () => { + renderPanel(); + fireEvent.click(screen.getByText('查看详情')); + expect(screen.getByText('完成度')).toBeInTheDocument(); + expect(screen.getAllByText('快照未完成').length).toBeGreaterThanOrEqual(2); + expect(screen.getByText('助手输出')).toBeInTheDocument(); + }); + + it('runs evaluation and emits the new result', async () => { + const onResult = vi.fn(); + mockEvaluate.mockResolvedValue({ result, reused_existing_run: false }); + renderPanel(); + + fireEvent.click(screen.getByText('开始评估')); + + await waitFor(() => expect(mockEvaluate).toHaveBeenCalledWith('conv-1', false)); + await waitFor(() => expect(onResult).toHaveBeenCalledWith(result)); + expect(screen.getByText('需复核')).toBeInTheDocument(); + }); + + it('shows force action after pending conflict', async () => { + mockEvaluate + .mockRejectedValueOnce(new EvaluationNotReadyError('pending', 2)) + .mockResolvedValueOnce({ result, reused_existing_run: false }); + renderPanel(); + + fireEvent.click(screen.getByText('开始评估')); + await waitFor(() => expect(screen.getByText(/2 个 LLM 调用仍未完成/)).toBeInTheDocument()); + + fireEvent.click(screen.getByText('强制评估')); + await waitFor(() => expect(mockEvaluate).toHaveBeenLastCalledWith('conv-1', true)); + expect(await screen.findByText('需复核')).toBeInTheDocument(); + }); +}); diff --git a/src/agentsight/dashboard/src/test/apiClient.test.ts b/src/agentsight/dashboard/src/test/apiClient.test.ts index 7b8a6fe56..6a0ff6f9d 100644 --- a/src/agentsight/dashboard/src/test/apiClient.test.ts +++ b/src/agentsight/dashboard/src/test/apiClient.test.ts @@ -20,6 +20,9 @@ import { fetchAgentHealth, deleteAgentHealth, restartAgentHealth, + fetchLatestEvaluation, + evaluateConversation, + EvaluationNotReadyError, INTERRUPTION_TYPE_CN, fetchSkillMetrics, fetchSecurityStatus, @@ -137,6 +140,58 @@ describe('apiClient', () => { }); }); + describe('Grader APIs', () => { + it('fetchLatestEvaluation should fetch latest conversation evaluation', async () => { + mockFetch.mockResolvedValueOnce(mockJsonResponse(null)); + const result = await fetchLatestEvaluation('conv-1'); + + expect(result).toBeNull(); + const url = mockFetch.mock.calls[0][0]; + expect(url).toContain('/api/grader/latest'); + expect(url).toContain('target_type=conversation'); + expect(url).toContain('target_id=conv-1'); + }); + + it('evaluateConversation should post a manual evaluation request', async () => { + const response = { result: { target_id: 'conv-1', verdict: 'pass' }, reused_existing_run: false }; + mockFetch.mockResolvedValueOnce(mockJsonResponse(response)); + const result = await evaluateConversation('conv-1', true); + + expect(result).toEqual(response); + expect(mockFetch.mock.calls[0][0]).toContain('/api/grader/evaluate'); + expect(mockFetch.mock.calls[0][1]).toMatchObject({ + method: 'POST', + headers: { 'Content-Type': 'application/json' }, + }); + expect(JSON.parse(mockFetch.mock.calls[0][1].body)).toEqual({ + target_type: 'conversation', + target_id: 'conv-1', + force: true, + }); + }); + + it('evaluateConversation should expose pending conflicts as EvaluationNotReadyError', async () => { + mockFetch.mockResolvedValueOnce(mockJsonResponse({ + error: 'conversation_not_ready', + pending_call_count: 2, + message: 'pending', + }, 409)); + + let caught: unknown; + try { + await evaluateConversation('conv-1'); + } catch (error) { + caught = error; + } + + expect(caught).toMatchObject({ + name: 'EvaluationNotReadyError', + pendingCallCount: 2, + }); + expect(caught).toBeInstanceOf(EvaluationNotReadyError); + }); + }); + describe('fetchAgentNames', () => { it('should fetch agent names', async () => { mockFetch.mockResolvedValueOnce(mockJsonResponse(['agent-a', 'agent-b'])); diff --git a/src/agentsight/dashboard/src/utils/apiClient.ts b/src/agentsight/dashboard/src/utils/apiClient.ts index e7cb96508..e9bd20dee 100644 --- a/src/agentsight/dashboard/src/utils/apiClient.ts +++ b/src/agentsight/dashboard/src/utils/apiClient.ts @@ -137,6 +137,138 @@ export async function fetchConversationDetail(conversationId: string): Promise; + } | null; + metadata?: Record | null; +} + +export interface EvaluationDimension { + name: string; + score: number; + verdict: EvaluationVerdict; + reason: string; + evidence_refs: EvaluationRef[]; +} + +export interface EvaluationFinding { + code: string; + severity: string; + message: string; + evidence_refs: EvaluationRef[]; +} + +export interface EvaluationMetadata { + evaluated_with_pending: boolean; + pending_call_count: number; + input_event_count: number; + grader_type: 'rule' | 'llm' | 'agent'; + grader_version: string; + rubric_version: string | null; + judge_model: string | null; + prompt_hash: string | null; + confidence: number | null; +} + +export interface EvaluationResult { + target_type: 'conversation'; + target_id: string; + run_id: string; + input_hash: string; + verdict: EvaluationVerdict; + score: number; + summary: string; + root_cause: EvaluationRootCause; + recommended_action: string; + dimensions: EvaluationDimension[]; + findings: EvaluationFinding[]; + metadata: EvaluationMetadata; +} + +export interface EvaluationResponse { + result: EvaluationResult; + reused_existing_run: boolean; +} + +export class EvaluationNotReadyError extends Error { + readonly pendingCallCount: number; + + constructor(message: string, pendingCallCount: number) { + super(message); + this.name = 'EvaluationNotReadyError'; + this.pendingCallCount = pendingCallCount; + } +} + +/** + * Fetch the latest persisted evaluation for a conversation. + */ +export async function fetchLatestEvaluation(conversationId: string): Promise { + const params = new URLSearchParams({ + target_type: 'conversation', + target_id: conversationId, + }); + return apiFetch(`${API_BASE}/api/grader/latest?${params.toString()}`); +} + +/** + * Manually evaluate a conversation with the rule-based grader. + */ +export async function evaluateConversation( + conversationId: string, + force = false, +): Promise { + const res = await fetch(`${API_BASE}/api/grader/evaluate`, { + method: 'POST', + headers: { 'Content-Type': 'application/json' }, + body: JSON.stringify({ + target_type: 'conversation', + target_id: conversationId, + force, + }), + }); + const body = await res.json().catch(() => ({})); + if (!res.ok) { + if (res.status === 409 && body?.error === 'conversation_not_ready') { + throw new EvaluationNotReadyError( + body.message ?? 'Conversation still has pending LLM calls.', + Number(body.pending_call_count ?? 0), + ); + } + throw new Error(`POST /api/grader/evaluate -> ${res.status}: ${body?.message ?? res.statusText}`); + } + return body as EvaluationResponse; +} + // ─── Agent-name & time-series APIs ─────────────────────────────────────────── /** From 277458cedc38c85cebcf696f7a33c903ca74e8e6 Mon Sep 17 00:00:00 2001 From: wanghao <1562495626@qq.com> Date: Mon, 6 Jul 2026 15:23:53 +0800 Subject: [PATCH 3/9] fix(sight): persist idle streamed calls Drain idle in-flight HTTP states and persist them as pending GenAI calls when a stream is abandoned while the agent process stays alive. Manual output interruption does not necessarily kill the process, so dead-PID draining cannot recover these sessions. This preserves the captured request as pending evidence; unsupported states are skipped. Assisted-by: Codex Signed-off-by: wanghao <1562495626@qq.com> --- .../src/aggregator/http/aggregator.rs | 108 +++++++++++++++--- src/agentsight/src/aggregator/unified.rs | 8 ++ src/agentsight/src/unified.rs | 52 +++++++++ 3 files changed, 155 insertions(+), 13 deletions(-) diff --git a/src/agentsight/src/aggregator/http/aggregator.rs b/src/agentsight/src/aggregator/http/aggregator.rs index 25a0a943f..1eaa4c8dc 100644 --- a/src/agentsight/src/aggregator/http/aggregator.rs +++ b/src/agentsight/src/aggregator/http/aggregator.rs @@ -173,11 +173,11 @@ impl HttpConnectionAggregator { } } - /// Evict connections that have been idle for longer than `self.idle_timeout` - /// or whose buffered body exceeds `self.max_body_bytes`. + /// Evict discardable connections that have been idle for longer than + /// `self.idle_timeout` or whose buffered body exceeds `self.max_body_bytes`. /// - /// Called periodically from the main event loop (via `UnifiedAggregator`) - /// to prevent stale or oversized connection states from accumulating. + /// In-flight request/response states are preserved here so callers can + /// drain and persist them instead of losing a manually interrupted stream. pub fn evict_idle_and_oversized(&mut self) { let now = Instant::now(); let timeout = self.idle_timeout; @@ -196,16 +196,20 @@ impl HttpConnectionAggregator { }) .collect(); + let mut evicted_idle = 0usize; for key in &to_evict { - self.connections.pop(key); - self.sse_continuation_buffers.pop(key); - self.last_appended_src_ptr.pop(key); - self.last_activity.pop(key); + if matches!(self.connections.peek(key), Some(ConnectionState::Idle)) { + self.connections.pop(key); + self.sse_continuation_buffers.pop(key); + self.last_appended_src_ptr.pop(key); + self.last_activity.pop(key); + evicted_idle += 1; + } } - if !to_evict.is_empty() { + if evicted_idle > 0 { log::info!( "[HttpAggregator] evicted {} idle connections (timeout={}s)", - to_evict.len(), + evicted_idle, timeout.as_secs() ); } @@ -217,9 +221,6 @@ impl HttpConnectionAggregator { .filter_map(|(k, state)| { let body_len = match state { ConnectionState::RequestBodyPending { body_buffer, .. } => body_buffer.len(), - ConnectionState::SseActive { - compressed_buffer, .. - } => compressed_buffer.as_ref().map_or(0, |b| b.len()), _ => 0, }; if body_len > max_bytes { Some(*k) } else { None } @@ -241,6 +242,50 @@ impl HttpConnectionAggregator { } } + /// Drain non-idle connections that exceeded `idle_timeout`. + /// + /// These states can represent manually interrupted or abandoned LLM + /// streams. Returning them lets the caller persist a pending GenAI row + /// instead of silently dropping the session evidence. + pub fn drain_idle_connections(&mut self) -> Vec<(ConnectionId, ConnectionState)> { + let now = Instant::now(); + let timeout = self.idle_timeout; + let keys: Vec = self + .last_activity + .iter() + .filter_map(|(k, ts)| { + if now.duration_since(*ts) > timeout { + Some(*k) + } else { + None + } + }) + .collect(); + + let mut result = Vec::new(); + for key in keys { + self.sse_continuation_buffers.pop(&key); + self.last_appended_src_ptr.pop(&key); + self.last_activity.pop(&key); + if let Some(state) = self.connections.pop(&key) { + match state { + ConnectionState::Idle => {} + _ => result.push((key, state)), + } + } + } + + if !result.is_empty() { + log::info!( + "[HttpAggregator] drained {} idle in-flight connection(s) (timeout={}s)", + result.len(), + timeout.as_secs() + ); + } + + result + } + /// Record activity timestamp for a connection. fn touch(&mut self, key: &ConnectionId) { self.last_activity.push(*key, Instant::now()); @@ -2374,4 +2419,41 @@ mod tests { agg.evict_idle_and_oversized(); assert!(agg.connections.peek(&conn_id).is_none()); } + + #[test] + fn test_idle_eviction_preserves_sse_active_for_drain() { + let mut agg = HttpConnectionAggregator::with_limits(10, 8192, Duration::from_millis(50)); + let event = create_mock_ssl_event(1234, 0x7000); + let request = ParsedRequest { + method: "POST".to_string(), + path: "/v1/messages".to_string(), + version: 11, + headers: HashMap::new(), + body_offset: 0, + body_len: 0, + source_event: event.clone(), + reassembled_body: None, + }; + agg.process_request(request); + + let mut headers = HashMap::new(); + headers.insert("content-type".to_string(), "text/event-stream".to_string()); + let response = ParsedResponse { + version: 11, + status_code: 200, + reason: "OK".to_string(), + headers, + body_offset: 0, + body_len: 0, + source_event: event, + }; + assert!(agg.process_response(response).is_none()); + + std::thread::sleep(Duration::from_millis(60)); + agg.evict_idle_and_oversized(); + let drained = agg.drain_connections_for_pid(1234); + + assert_eq!(drained.len(), 1); + assert!(matches!(drained[0].1, ConnectionState::SseActive { .. })); + } } diff --git a/src/agentsight/src/aggregator/unified.rs b/src/agentsight/src/aggregator/unified.rs index d27c46b87..e4c28e6f7 100644 --- a/src/agentsight/src/aggregator/unified.rs +++ b/src/agentsight/src/aggregator/unified.rs @@ -177,4 +177,12 @@ impl Aggregator { pub fn drain_dead_pid_connections(&mut self) -> Vec<(ConnectionId, ConnectionState)> { self.http.drain_dead_pid_connections() } + + /// Drain in-flight HTTP connections that exceeded the idle timeout. + /// + /// Used to persist evidence for manually interrupted streams where the + /// agent process remains alive, so dead-PID draining would never run. + pub fn drain_idle_connections(&mut self) -> Vec<(ConnectionId, ConnectionState)> { + self.http.drain_idle_connections() + } } diff --git a/src/agentsight/src/unified.rs b/src/agentsight/src/unified.rs index e7a5798a6..fc4538c3e 100644 --- a/src/agentsight/src/unified.rs +++ b/src/agentsight/src/unified.rs @@ -909,6 +909,8 @@ impl AgentSight { self.flush_expired_pending_genai(); // Drain orphaned connections from dead PIDs and persist as pending self.drain_and_persist_dead_connections(); + // Drain idle in-flight streams whose owning process is still alive. + self.drain_and_persist_idle_connections(); // Check if config watcher deposited a new LogtailExporter self.check_pending_logtail(); // Periodically purge old/oversized interruption DB entries @@ -1278,6 +1280,56 @@ impl AgentSight { } } + /// Drain idle in-flight streams and persist them as `pending` records. + /// + /// This covers manual output interruption where the agent process stays + /// alive, so dead-PID draining cannot discover the abandoned stream. + fn drain_and_persist_idle_connections(&mut self) { + let drained = self.aggregator.drain_idle_connections(); + if drained.is_empty() { + return; + } + + use crate::aggregator::ConnectionState; + + for (conn_id, state) in drained { + let (_state_name, request) = match state { + ConnectionState::RequestPending { request } => ("RequestPending", request), + ConnectionState::SseActive { + request: Some(req), .. + } => ("SseActive", req), + _ => continue, + }; + + if let Some(pending) = self.genai_builder.build_pending_from_request( + &request, + &conn_id, + &self.pid_agent_name_cache, + ) { + if let Some(ref store) = self.genai_sqlite_store { + let call_id = pending.call_id.clone(); + if let Err(e) = store.insert_pending(&pending) { + log::warn!("[IdleDrain] Failed to persist pending call: {e}"); + continue; + } + log::info!( + "[IdleDrain] persisted idle in-flight call as pending: call_id={} pid={} path={}", + call_id, + conn_id.pid, + request.path, + ); + } + } else { + log::debug!( + "[IdleDrain] build_pending returned None: pid={} path={} body_len={}", + conn_id.pid, + request.path, + request.body_len + ); + } + } + } + /// Drain aggregator connections whose PID is no longer alive and persist /// them as `pending` records in `genai_events`. Rate-limited to once per /// second to avoid excessive `/proc` scanning. From 3cd71743c7fdecf4cf64036604b8b3d5df2c3bd1 Mon Sep 17 00:00:00 2001 From: wanghao <1562495626@qq.com> Date: Mon, 6 Jul 2026 15:24:27 +0800 Subject: [PATCH 4/9] fix(sight): avoid false interruption signals Classify timeout keywords only from structured call errors and inspect response bodies for provider errors only on non-success status codes. Successful assistant text can legitimately mention timeout, auth, rate limits, or context limits as troubleshooting content. This keeps interruption detection focused on runtime failures; provider-specific wording still depends on captured error fields. Assisted-by: Codex Signed-off-by: wanghao <1562495626@qq.com> --- src/agentsight/src/interruption/detector.rs | 85 +++++++++++++++++++-- 1 file changed, 80 insertions(+), 5 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(); From 4a8d98c3a1adfa271ed7e89132e1dbf7197b4cb8 Mon Sep 17 00:00:00 2001 From: wanghao <1562495626@qq.com> Date: Mon, 6 Jul 2026 15:24:58 +0800 Subject: [PATCH 5/9] fix(sight): preserve Anthropic tool results Convert Anthropic request content blocks into unified message parts, including tool_use, tool_result, and thinking blocks. The grader relies on structured input evidence instead of raw event JSON to avoid prompt-context false positives. Historical rows are unchanged; new captures preserve tool failures for deterministic scoring. Assisted-by: Codex Signed-off-by: wanghao <1562495626@qq.com> --- src/agentsight/src/genai/call_builder.rs | 138 ++++++++++++++++++++++- 1 file changed, 135 insertions(+), 3 deletions(-) diff --git a/src/agentsight/src/genai/call_builder.rs b/src/agentsight/src/genai/call_builder.rs index 1dc45e08d..0b0cf54a0 100644 --- a/src/agentsight/src/genai/call_builder.rs +++ b/src/agentsight/src/genai/call_builder.rs @@ -291,9 +291,7 @@ impl GenAIBuilder { let role = format!("{:?}", m.role).to_lowercase(); InputMessage { role, - parts: vec![MessagePart::Text { - content: m.content.as_text(), - }], + parts: Self::anthropic_message_content_to_parts(&m.content), name: None, } }) @@ -403,6 +401,77 @@ impl GenAIBuilder { // openai_msg_to_output / parse_openai_tool_call_value / parse_sse_response_body / // extract_parts_from_sse_body live in `openai_parse.rs` (same impl block). + fn anthropic_message_content_to_parts( + content: &crate::analyzer::message::AnthropicMessageContent, + ) -> Vec { + match content { + crate::analyzer::message::AnthropicMessageContent::Text(text) => { + if text.is_empty() { + Vec::new() + } else { + vec![MessagePart::Text { + content: text.clone(), + }] + } + } + crate::analyzer::message::AnthropicMessageContent::Blocks(blocks) => blocks + .iter() + .filter_map(Self::anthropic_content_block_to_part) + .collect(), + } + } + + fn anthropic_content_block_to_part( + block: &crate::analyzer::message::AnthropicContentBlock, + ) -> Option { + match block { + crate::analyzer::message::AnthropicContentBlock::Text { text, .. } => { + if text.is_empty() { + None + } else { + Some(MessagePart::Text { + content: text.clone(), + }) + } + } + crate::analyzer::message::AnthropicContentBlock::ToolUse { id, name, input } => { + Some(MessagePart::ToolCall { + id: Some(id.clone()), + name: name.clone(), + arguments: Some(input.clone()), + }) + } + crate::analyzer::message::AnthropicContentBlock::ToolResult { + tool_use_id, + content, + is_error, + } => { + let response = match (content.clone(), *is_error) { + (Some(value), Some(is_error)) => { + serde_json::json!({"content": value, "is_error": is_error}) + } + (Some(value), None) => value, + (None, Some(is_error)) => serde_json::json!({"is_error": is_error}), + (None, None) => serde_json::Value::Null, + }; + Some(MessagePart::ToolCallResponse { + id: Some(tool_use_id.clone()), + response, + }) + } + crate::analyzer::message::AnthropicContentBlock::Thinking { thinking, .. } => { + if thinking.is_empty() { + None + } else { + Some(MessagePart::Reasoning { + content: thinking.clone(), + }) + } + } + _ => None, + } + } + /// Build LLMResponse from parsed message or HTTP record fn build_response( &self, @@ -936,6 +1005,69 @@ mod tests { assert!(call.request.tools.is_some()); } + #[test] + fn test_build_request_anthropic_preserves_tool_result_blocks() { + let builder = GenAIBuilder::new(); + let anth_req = AnthropicRequest { + model: "claude-3".to_string(), + messages: vec![ + AnthMsg { + role: MessageRole::Assistant, + content: AnthropicMessageContent::Blocks(vec![ + AnthropicContentBlock::ToolUse { + id: "toolu_1".to_string(), + name: "Bash".to_string(), + input: serde_json::json!({"command": "cat /tmp/missing.txt"}), + }, + ]), + }, + AnthMsg { + role: MessageRole::User, + content: AnthropicMessageContent::Blocks(vec![ + AnthropicContentBlock::ToolResult { + tool_use_id: "toolu_1".to_string(), + content: Some(serde_json::json!( + "Exit code 1\ncat: /tmp/missing.txt: No such file or directory" + )), + is_error: Some(true), + }, + ]), + }, + ], + max_tokens: 200, + system: None, + stream: Some(false), + temperature: None, + top_p: None, + top_k: None, + stop_sequences: None, + metadata: None, + tools: None, + tool_choice: None, + }; + let parsed = ParsedApiMessage::AnthropicMessage { + request: Some(anth_req), + response: None, + }; + let http = make_http("/v1/messages", None, None); + let call = build_call( + &builder, + &[AnalysisResult::Http(http), AnalysisResult::Message(parsed)], + ) + .unwrap(); + + assert!(call.request.messages.iter().any(|message| { + message.parts.iter().any(|part| { + matches!( + part, + MessagePart::ToolCallResponse { id, response } + if id.as_deref() == Some("toolu_1") + && response.get("is_error").and_then(|v| v.as_bool()) == Some(true) + ) + }) + })); + } + #[test] fn test_build_request_sysom_full() { let builder = GenAIBuilder::new(); From ba8c19f67605cb59d4c246ff0fc4c2bebdbf7af6 Mon Sep 17 00:00:00 2001 From: wanghao <1562495626@qq.com> Date: Mon, 6 Jul 2026 15:25:33 +0800 Subject: [PATCH 6/9] docs(sight): document conversation grader Document the manual conversation grader APIs, persistence model, current rule version, and dashboard-facing changelog entries. Keeping the operational guide and semantic design note together makes the MVP discoverable without changing runtime behavior. The repository documentation-standard file was not present in this worktree, so this keeps the doc change narrowly scoped. Assisted-by: Codex Signed-off-by: wanghao <1562495626@qq.com> --- src/agentsight/CHANGELOG.md | 3 +++ .../docs/agent-diagnostics-guide.md | 16 +++++++++++++++ .../docs/design-docs/genai-semantic.md | 20 +++++++++++++++++++ 3 files changed, 39 insertions(+) diff --git a/src/agentsight/CHANGELOG.md b/src/agentsight/CHANGELOG.md index d2a68e69d..ffb7ea17b 100644 --- a/src/agentsight/CHANGELOG.md +++ b/src/agentsight/CHANGELOG.md @@ -11,6 +11,7 @@ - Add `call_kind` classification (chat / completion / embedding / tool_use) to GenAI semantic events. - Add `--exclude` filter to `agentsight audit` CLI for noise reduction, and show non-streaming LLM calls in audit output. - Add unified `agentsight summary` command for one-shot status overview. +- Add manual conversation grader with persisted quality verdicts in the dashboard. - Enhance token savings page with baseline comparison, strategy breakdown, line-level diff highlighting and optimization tips. - Upload skill metrics via SLS Logtail exporter. - Improve agent health UX: role badges (P1/P2), TTL-based cleanup, process-ancestry grouping, and Session ID help tooltip. @@ -33,6 +34,8 @@ - Respect dynamic sysom path in SLS exporter mode selection; replace removed `sysom_logtail_path` with `logtail_path` filter. - Validate ring buffer size is power-of-two at startup. - Wire feature flags and runtime limits to actual runtime code paths. +- Avoid treating historical raw event context as a current tool failure in conversation quality evaluation. +- Avoid treating successful assistant text that mentions timeout as a network timeout interruption. ### Refactoring - Split `genai/builder.rs` into 4 focused modules and `genai.rs` into 5 submodules. diff --git a/src/agentsight/docs/agent-diagnostics-guide.md b/src/agentsight/docs/agent-diagnostics-guide.md index 325db18f0..0980a1bbe 100644 --- a/src/agentsight/docs/agent-diagnostics-guide.md +++ b/src/agentsight/docs/agent-diagnostics-guide.md @@ -441,6 +441,20 @@ curl http://127.0.0.1:7396/api/export/atif/session/{session_id} curl http://127.0.0.1:7396/api/export/atif/conversation/{conversation_id} ``` +### 10.4 Conversation Grader + +Conversation Grader 对指定 `conversation_id` 执行手动质量评估，并将结果持久化到 SQLite。当前版本只支持 `target_type=conversation`，评估器为规则版 `rule-v3`。 + +```bash +curl -X POST http://127.0.0.1:7396/api/grader/evaluate \ + -H 'Content-Type: application/json' \ + -d '{"target_type":"conversation","target_id":"{conversation_id}","force":false}' + +curl "http://127.0.0.1:7396/api/grader/latest?target_type=conversation&target_id={conversation_id}" +``` + +如果 conversation 仍有 pending LLM 调用，评估接口返回 HTTP 409。确认需要评估当前不完整快照时，可以将 `force` 设置为 `true`，结果会标记 `evaluated_with_pending=true`。 + --- ## 11. 支持的 LLM 提供商 @@ -466,6 +480,8 @@ AgentSight 自动识别并解析以下 LLM API 格式： | `/api/sessions/{id}/traces` | GET | 会话下的对话列表 | | `/api/traces/{id}` | GET | Trace 详情 | | `/api/conversations/{id}` | GET | 对话详情 | +| `/api/grader/evaluate` | POST | 手动评估 Conversation 质量 | +| `/api/grader/latest` | GET | 查询最新 Conversation 评估结果 | | `/api/agent-names` | GET | Agent 名称列表 | | `/api/agent-health` | GET | Agent 健康状态 | | `/api/interruptions` | GET | 中断事件列表 | diff --git a/src/agentsight/docs/design-docs/genai-semantic.md b/src/agentsight/docs/design-docs/genai-semantic.md index 2cd7764b8..b7049a300 100644 --- a/src/agentsight/docs/design-docs/genai-semantic.md +++ b/src/agentsight/docs/design-docs/genai-semantic.md @@ -144,11 +144,31 @@ graph LR - **By trace**: `/api/export/atif/trace/{trace_id}` - **By session**: `/api/export/atif/session/{session_id}` +- **By conversation**: `/api/export/atif/conversation/{conversation_id}` **ATIF structure**: `AtifAgent` → `Vec` → `AtifToolCall` + `AtifObservation` **Source**: `src/atif/converter.rs`, `src/atif/schema.rs` +## Conversation Grader + +The conversation grader is a manual post-run evaluation layer over the GenAI SQLite evidence. The MVP only supports `target_type = "conversation"` and uses deterministic `rule-v3` scoring. + +Inputs: + +- GenAI LLM call rows from `GenAISqliteStore::get_events_by_conversation()`. +- Interruption rows from `InterruptionStore::list_by_conversation()`. +- A stable input hash derived from the conversation evidence and grader version. + +Outputs are stored in `evaluation_runs` in the GenAI SQLite database. The idempotency key is `target_type`, `target_id`, `input_hash`, `grader_type`, and `grader_version`, so repeated evaluation of the same snapshot reuses the completed run. + +Manual APIs: + +- `POST /api/grader/evaluate` +- `GET /api/grader/latest?target_type=conversation&target_id={conversation_id}` + +If the conversation still has pending LLM calls, `POST /api/grader/evaluate` returns HTTP 409 unless `force=true` is provided. Forced snapshots are marked with `metadata.evaluated_with_pending = true`. + ## Session & Trace Model The GenAI semantic layer introduces session and trace concepts: From 1b05d76e5e5d89d3e063eec0b74bb7531f1488e4 Mon Sep 17 00:00:00 2001 From: wanghao <1562495626@qq.com> Date: Mon, 6 Jul 2026 16:18:37 +0800 Subject: [PATCH 7/9] fix(sight): satisfy clippy lints Apply current Clippy mechanical suggestions in AgentSight code touched by the PR and adjacent existing paths. This keeps the community preflight check `cargo clippy -- -D warnings` passing on the Linux test host without changing runtime behavior. Assisted-by: Codex Signed-off-by: wanghao <1562495626@qq.com> --- src/agentsight/src/grader/evidence.rs | 4 +--- 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, 5 insertions(+), 7 deletions(-) diff --git a/src/agentsight/src/grader/evidence.rs b/src/agentsight/src/grader/evidence.rs index 9d1336322..97f8f8eb1 100644 --- a/src/agentsight/src/grader/evidence.rs +++ b/src/agentsight/src/grader/evidence.rs @@ -79,9 +79,7 @@ fn value_has_error_signal(value: &serde_json::Value) -> bool { serde_json::Value::Array(items) => items.iter().any(value_has_error_signal), serde_json::Value::Object(map) => { map.get("is_error").and_then(|value| value.as_bool()) == Some(true) - || map - .values() - .any(|nested_value| value_has_error_signal(nested_value)) + || map.values().any(value_has_error_signal) } _ => false, } 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 6ed35fc19420767bb7134bf32b8a8119ec800a5a Mon Sep 17 00:00:00 2001 From: wanghao <1562495626@qq.com> Date: Wed, 8 Jul 2026 15:00:32 +0800 Subject: [PATCH 8/9] fix(sight): narrow grader store locks Keep only SQLite reads and writes under the EvaluationStore mutex. JSON serialization and record conversion now run outside the lock, which addresses the review concern without changing the rule grader execution path. Assisted-by: Codex Signed-off-by: wanghao <1562495626@qq.com> --- src/agentsight/src/grader/storage.rs | 108 ++++++++++++++------------- 1 file changed, 57 insertions(+), 51 deletions(-) diff --git a/src/agentsight/src/grader/storage.rs b/src/agentsight/src/grader/storage.rs index 23d63a429..fd58e98c1 100644 --- a/src/agentsight/src/grader/storage.rs +++ b/src/agentsight/src/grader/storage.rs @@ -76,13 +76,14 @@ impl EvaluationStore { grader_type: GraderType, grader_version: &str, ) -> Result, GraderError> { - let conn = self - .conn - .lock() - .map_err(|error| GraderError::Storage(error.to_string()))?; - let mut stmt = conn - .prepare( - "SELECT id, run_id, target_type, target_id, input_hash, grader_type, + let raw = { + let conn = self + .conn + .lock() + .map_err(|error| GraderError::Storage(error.to_string()))?; + let mut stmt = conn + .prepare( + "SELECT id, run_id, target_type, target_id, input_hash, grader_type, grader_version, status, verdict, score, root_cause, created_at, completed_at, result_json FROM evaluation_runs @@ -93,38 +94,40 @@ impl EvaluationStore { AND grader_version=?5 AND status='completed' LIMIT 1", - ) - .map_err(|error| GraderError::Storage(error.to_string()))?; - let mut rows = stmt - .query(params![ - target_type.as_str(), - target_id, - input_hash, - grader_type.as_str(), - grader_version, - ]) - .map_err(|error| GraderError::Storage(error.to_string()))?; - match rows - .next() - .map_err(|error| GraderError::Storage(error.to_string()))? - { - Some(row) => raw_to_record( - read_raw_record(row).map_err(|error| GraderError::Storage(error.to_string()))?, - ) - .map(Some), - None => Ok(None), - } + ) + .map_err(|error| GraderError::Storage(error.to_string()))?; + let mut rows = stmt + .query(params![ + target_type.as_str(), + target_id, + input_hash, + grader_type.as_str(), + grader_version, + ]) + .map_err(|error| GraderError::Storage(error.to_string()))?; + match rows + .next() + .map_err(|error| GraderError::Storage(error.to_string()))? + { + Some(row) => Some( + read_raw_record(row) + .map_err(|error| GraderError::Storage(error.to_string()))?, + ), + None => None, + } + }; + raw.map(raw_to_record).transpose() } /// Insert a completed evaluation result. /// /// Returns `false` when an equivalent completed run already exists. pub fn insert_completed(&self, result: &EvaluationResult) -> Result { + let result_json = serde_json::to_string(result)?; let conn = self .conn .lock() .map_err(|error| GraderError::Storage(error.to_string()))?; - let result_json = serde_json::to_string(result)?; let inserted = conn .execute( "INSERT OR IGNORE INTO evaluation_runs ( @@ -160,13 +163,14 @@ impl EvaluationStore { target_type: TargetType, target_id: &str, ) -> Result, GraderError> { - let conn = self - .conn - .lock() - .map_err(|error| GraderError::Storage(error.to_string()))?; - let mut stmt = conn - .prepare( - "SELECT id, run_id, target_type, target_id, input_hash, grader_type, + let raw = { + let conn = self + .conn + .lock() + .map_err(|error| GraderError::Storage(error.to_string()))?; + let mut stmt = conn + .prepare( + "SELECT id, run_id, target_type, target_id, input_hash, grader_type, grader_version, status, verdict, score, root_cause, created_at, completed_at, result_json FROM evaluation_runs @@ -175,21 +179,23 @@ impl EvaluationStore { AND status='completed' ORDER BY created_at DESC, id DESC LIMIT 1", - ) - .map_err(|error| GraderError::Storage(error.to_string()))?; - let mut rows = stmt - .query(params![target_type.as_str(), target_id]) - .map_err(|error| GraderError::Storage(error.to_string()))?; - match rows - .next() - .map_err(|error| GraderError::Storage(error.to_string()))? - { - Some(row) => raw_to_record( - read_raw_record(row).map_err(|error| GraderError::Storage(error.to_string()))?, - ) - .map(Some), - None => Ok(None), - } + ) + .map_err(|error| GraderError::Storage(error.to_string()))?; + let mut rows = stmt + .query(params![target_type.as_str(), target_id]) + .map_err(|error| GraderError::Storage(error.to_string()))?; + match rows + .next() + .map_err(|error| GraderError::Storage(error.to_string()))? + { + Some(row) => Some( + read_raw_record(row) + .map_err(|error| GraderError::Storage(error.to_string()))?, + ), + None => None, + } + }; + raw.map(raw_to_record).transpose() } } From 154c1f0ec24575b312be922275604824722d4e0b Mon Sep 17 00:00:00 2001 From: wanghao <1562495626@qq.com> Date: Wed, 8 Jul 2026 15:26:37 +0800 Subject: [PATCH 9/9] fix(sight): address grader review Reuse the grader evaluation store from server state instead of reopening SQLite on each API request. Load interruption evidence from the existing interruption store rather than deriving a sibling database path, and add regression coverage for oversized request-body eviction. Assisted-by: Codex Signed-off-by: wanghao <1562495626@qq.com> --- .../src/aggregator/http/aggregator.rs | 40 ++++++ src/agentsight/src/grader/input.rs | 126 ++++++++++++++++-- src/agentsight/src/server/handlers.rs | 107 +++++++++++++-- src/agentsight/src/server/mod.rs | 13 ++ src/agentsight/src/server/token_savings.rs | 4 +- 5 files changed, 262 insertions(+), 28 deletions(-) diff --git a/src/agentsight/src/aggregator/http/aggregator.rs b/src/agentsight/src/aggregator/http/aggregator.rs index 1eaa4c8dc..39baa9fbf 100644 --- a/src/agentsight/src/aggregator/http/aggregator.rs +++ b/src/agentsight/src/aggregator/http/aggregator.rs @@ -2456,4 +2456,44 @@ mod tests { assert_eq!(drained.len(), 1); assert!(matches!(drained[0].1, ConnectionState::SseActive { .. })); } + + #[test] + fn test_oversized_request_body_pending_is_evicted() { + let mut agg = HttpConnectionAggregator::with_limits(10, 1024, Duration::from_secs(60)); + let conn_id = ConnectionId { + pid: 1234, + ssl_ptr: 0x7100, + }; + let event = create_mock_ssl_event(conn_id.pid, conn_id.ssl_ptr); + let request = ParsedRequest { + method: "POST".to_string(), + path: "/v1/messages".to_string(), + version: 11, + headers: HashMap::new(), + body_offset: 0, + body_len: 0, + source_event: event, + reassembled_body: None, + }; + + agg.connections.push( + conn_id, + ConnectionState::RequestBodyPending { + request, + expected_body_len: Some(4096), + body_buffer: vec![b'x'; 2048], + }, + ); + agg.last_activity.push(conn_id, Instant::now()); + agg.sse_continuation_buffers + .push(conn_id, b"stale".to_vec()); + agg.last_appended_src_ptr.push(conn_id, 42); + + agg.evict_idle_and_oversized(); + + assert!(agg.connections.peek(&conn_id).is_none()); + assert!(agg.last_activity.peek(&conn_id).is_none()); + assert!(agg.sse_continuation_buffers.peek(&conn_id).is_none()); + assert!(agg.last_appended_src_ptr.peek(&conn_id).is_none()); + } } diff --git a/src/agentsight/src/grader/input.rs b/src/agentsight/src/grader/input.rs index ca6fb2e95..ba8a2250c 100644 --- a/src/agentsight/src/grader/input.rs +++ b/src/agentsight/src/grader/input.rs @@ -30,6 +30,7 @@ pub struct EvaluationInput { /// Load a conversation snapshot and compute its stable input hash. pub fn load_conversation_input( storage_path: &Path, + interruption_store: Option<&InterruptionStore>, conversation_id: &str, force: bool, ) -> Result { @@ -55,7 +56,7 @@ pub fn load_conversation_input( }); } - let interruptions = load_conversation_interruptions(storage_path, conversation_id)?; + let interruptions = load_conversation_interruptions(interruption_store, conversation_id)?; let input_hash = compute_input_hash(conversation_id, &events, &interruptions)?; Ok(EvaluationInput { @@ -70,22 +71,15 @@ pub fn load_conversation_input( } fn load_conversation_interruptions( - storage_path: &Path, + interruption_store: Option<&InterruptionStore>, conversation_id: &str, ) -> Result, GraderError> { - if storage_path == Path::new(":memory:") { - return Ok(Vec::new()); + match interruption_store { + Some(store) => store + .list_by_conversation(conversation_id) + .map_err(|error| GraderError::Storage(error.to_string())), + None => Ok(Vec::new()), } - let parent = storage_path - .parent() - .filter(|path| !path.as_os_str().is_empty()) - .unwrap_or_else(|| Path::new(".")); - let interruption_path = parent.join("interruption_events.db"); - let store = InterruptionStore::new_with_path(&interruption_path) - .map_err(|error| GraderError::Storage(error.to_string()))?; - store - .list_by_conversation(conversation_id) - .map_err(|error| GraderError::Storage(error.to_string())) } fn compute_input_hash( @@ -157,3 +151,107 @@ fn interruption_hash_value(record: &InterruptionRecord) -> serde_json::Value { "resolved": record.resolved, }) } + +#[cfg(test)] +mod tests { + use std::path::{Path, PathBuf}; + use std::time::{SystemTime, UNIX_EPOCH}; + + use crate::genai::GenAIExporter; + use crate::genai::semantic::{GenAISemanticEvent, LLMCall, LLMRequest}; + use crate::interruption::{InterruptionEvent, InterruptionType}; + + use super::*; + + #[test] + fn load_conversation_input_uses_injected_interruption_store() { + let root = temp_root("grader_input_interruption_store"); + let genai_path = root.join("genai").join("events.db"); + let interruption_path = root.join("interruptions").join("events.db"); + write_conversation_event(&genai_path, "conv-injected"); + + let interruption_store = InterruptionStore::new_with_path(&interruption_path).unwrap(); + let event = InterruptionEvent::new( + InterruptionType::NetworkTimeout, + Some("session-1".to_string()), + Some("trace-1".to_string()), + Some("conv-injected".to_string()), + Some("call-1".to_string()), + Some(1234), + Some("Codex".to_string()), + 1_700_000_000_000_000_100, + None, + ); + interruption_store.insert(&event).unwrap(); + + let input = load_conversation_input( + &genai_path, + Some(&interruption_store), + "conv-injected", + false, + ) + .unwrap(); + + assert_eq!(input.events.len(), 1); + assert_eq!(input.interruptions.len(), 1); + assert_eq!( + input.interruptions[0].interruption_type, + InterruptionType::NetworkTimeout.as_str() + ); + + cleanup_db(&genai_path); + cleanup_db(&interruption_path); + let _ = std::fs::remove_dir_all(&root); + } + + fn write_conversation_event(path: &Path, conversation_id: &str) { + let store = GenAISqliteStore::new_with_path(path).unwrap(); + let mut call = LLMCall::new( + "call-1".to_string(), + 1_700_000_000_000_000_000, + "anthropic".to_string(), + "claude".to_string(), + LLMRequest { + messages: Vec::new(), + temperature: None, + max_tokens: None, + frequency_penalty: None, + presence_penalty: None, + top_p: None, + top_k: None, + seed: None, + stop_sequences: None, + stream: false, + tools: None, + raw_body: None, + }, + 1234, + "claude".to_string(), + ); + call.metadata + .insert("conversation_id".to_string(), conversation_id.to_string()); + call.metadata + .insert("response_id".to_string(), "trace-1".to_string()); + call.metadata + .insert("user_query".to_string(), "hello".to_string()); + + store.export(&[GenAISemanticEvent::LLMCall(call)]); + store.flush(); + } + + fn temp_root(label: &str) -> PathBuf { + std::env::temp_dir().join(format!( + "agentsight_{label}_{}", + SystemTime::now() + .duration_since(UNIX_EPOCH) + .unwrap() + .as_nanos() + )) + } + + fn cleanup_db(path: &Path) { + let _ = std::fs::remove_file(path); + let _ = std::fs::remove_file(format!("{}-wal", path.display())); + let _ = std::fs::remove_file(format!("{}-shm", path.display())); + } +} diff --git a/src/agentsight/src/server/handlers.rs b/src/agentsight/src/server/handlers.rs index d82633147..5fc58c00b 100644 --- a/src/agentsight/src/server/handlers.rs +++ b/src/agentsight/src/server/handlers.rs @@ -10,8 +10,8 @@ use serde_json::{Value, json}; use super::AppState; use crate::agent_sec::{AgentSecClient, AgentSecClientError, DaemonResponse}; use crate::grader::{ - EvaluationRequest, EvaluationResponse, EvaluationStore, GraderError, GraderType, - RULE_GRADER_VERSION, RuleGrader, TargetType, load_conversation_input, + EvaluationRequest, EvaluationResponse, GraderError, GraderType, RULE_GRADER_VERSION, + RuleGrader, TargetType, load_conversation_input, }; use crate::health::AgentHealthStatus; use crate::storage::sqlite::GenAISqliteStore; @@ -185,14 +185,16 @@ pub async fn evaluate_grader( .json(json!({"error": "bad_request", "message": "target_id is required"})); } - let input = match load_conversation_input(&data.storage_path, &body.target_id, body.force) { + let input = match load_conversation_input( + &data.storage_path, + data.interruption_store.as_deref(), + &body.target_id, + body.force, + ) { Ok(input) => input, Err(error) => return grader_error_response(error), }; - let store = match EvaluationStore::new_with_path(&data.storage_path) { - Ok(store) => store, - Err(error) => return grader_error_response(error), - }; + let store = &data.evaluation_store; match store.find_completed( target_type, @@ -269,12 +271,10 @@ pub async fn latest_grader( .json(json!({"error": "bad_request", "message": "target_id is required"})); } - let store = match EvaluationStore::new_with_path(&data.storage_path) { - Ok(store) => store, - Err(error) => return grader_error_response(error), - }; - - match store.latest_completed(target_type, &query.target_id) { + match data + .evaluation_store + .latest_completed(target_type, &query.target_id) + { Ok(Some(record)) => HttpResponse::Ok().json(record.result), Ok(None) => HttpResponse::Ok().json(serde_json::Value::Null), Err(error) => grader_error_response(error), @@ -769,6 +769,7 @@ mod tests { use actix_web::test as awtest; use crate::agent_sec::DaemonErrorPayload; + use crate::grader::EvaluationStore; use crate::health::HealthStore; use super::*; @@ -982,6 +983,46 @@ mod tests { } } + #[actix_web::test] + async fn latest_grader_uses_shared_evaluation_store() { + let root = temp_root("latest_grader_shared_store"); + let evaluation_path = root.join("evaluation.db"); + let evaluation_store = Arc::new(EvaluationStore::new_with_path(&evaluation_path).unwrap()); + let result = test_evaluation_result("conv-shared"); + evaluation_store.insert_completed(&result).unwrap(); + + let blocked_parent = root.join("not-a-directory"); + std::fs::write(&blocked_parent, b"file").unwrap(); + let data = web::Data::new(AppState { + storage_path: blocked_parent.join("genai.db"), + start_time: Instant::now(), + health_store: Arc::new(RwLock::new(HealthStore::new())), + interruption_store: None, + evaluation_store: Arc::clone(&evaluation_store), + security_observability: super::super::SecurityObservabilityConfig { timeout_ms: 0 }, + }); + let app = awtest::init_service(App::new().app_data(data).service(latest_grader)).await; + + let response = awtest::call_service( + &app, + awtest::TestRequest::get() + .uri("/grader/latest?target_type=conversation&target_id=conv-shared") + .to_request(), + ) + .await; + + assert_eq!(response.status(), StatusCode::OK); + let body: Value = serde_json::from_slice( + &actix_web::body::to_bytes(response.into_body()) + .await + .unwrap(), + ) + .unwrap(); + assert_eq!(body["run_id"], "run-shared"); + + let _ = std::fs::remove_dir_all(&root); + } + async fn response_json(response: HttpResponse) -> Value { let body = to_bytes(response.into_body()) .await @@ -1004,12 +1045,52 @@ mod tests { } } + fn test_evaluation_result(target_id: &str) -> crate::grader::EvaluationResult { + crate::grader::EvaluationResult { + target_type: TargetType::Conversation, + target_id: target_id.to_string(), + run_id: "run-shared".to_string(), + input_hash: "input-hash-shared".to_string(), + verdict: crate::grader::Verdict::Pass, + score: 1.0, + summary: "ok".to_string(), + root_cause: crate::grader::RootCause::None, + recommended_action: "none".to_string(), + dimensions: Vec::new(), + findings: Vec::new(), + metadata: crate::grader::EvaluationMetadata { + evaluated_with_pending: false, + pending_call_count: 0, + input_event_count: 1, + grader_type: GraderType::Rule, + grader_version: RULE_GRADER_VERSION.to_string(), + rubric_version: None, + judge_model: None, + prompt_hash: None, + confidence: Some(1.0), + }, + } + } + + fn temp_root(label: &str) -> PathBuf { + std::env::temp_dir().join(format!( + "agentsight_{label}_{}", + std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .unwrap() + .as_nanos() + )) + } + fn test_app_state(timeout_ms: u64) -> web::Data { web::Data::new(AppState { storage_path: PathBuf::from(":memory:"), start_time: Instant::now(), health_store: Arc::new(RwLock::new(HealthStore::new())), interruption_store: None, + evaluation_store: Arc::new( + EvaluationStore::new_with_path(std::path::Path::new(":memory:")).unwrap(), + ), security_observability: super::super::SecurityObservabilityConfig { timeout_ms }, }) } diff --git a/src/agentsight/src/server/mod.rs b/src/agentsight/src/server/mod.rs index 8940ca801..b8fd8322c 100644 --- a/src/agentsight/src/server/mod.rs +++ b/src/agentsight/src/server/mod.rs @@ -14,6 +14,7 @@ use actix_cors::Cors; use actix_web::{App, HttpRequest, HttpResponse, HttpServer, Responder, get, web}; use include_dir::{Dir, include_dir}; +use crate::grader::EvaluationStore; use crate::health::{HealthChecker, HealthStore}; use crate::storage::sqlite::InterruptionStore; @@ -45,6 +46,8 @@ pub struct AppState { pub health_store: Arc>, /// Interruption events store pub interruption_store: Option>, + /// Grader evaluation store + pub evaluation_store: Arc, /// agent-sec security observability integration configuration pub security_observability: SecurityObservabilityConfig, } @@ -196,6 +199,11 @@ async fn api_not_found() -> impl Responder { pub async fn run_server(host: &str, port: u16, storage_path: PathBuf) -> std::io::Result<()> { let security_observability = SecurityObservabilityConfig::default(); + let evaluation_store = Arc::new( + EvaluationStore::new_with_path(&storage_path) + .map_err(|error| std::io::Error::other(error.to_string()))?, + ); + // Initialize GenAI SQLite store (needed for HealthChecker to query pending calls) let genai_store: Option> = match crate::storage::sqlite::GenAISqliteStore::new() { @@ -244,6 +252,7 @@ pub async fn run_server(host: &str, port: u16, storage_path: PathBuf) -> std::io start_time: Instant::now(), health_store, interruption_store, + evaluation_store, security_observability, }); @@ -285,6 +294,7 @@ mod tests { use actix_web::test as awtest; use actix_web::{App, web}; + use crate::grader::EvaluationStore; use crate::health::HealthStore; use super::{ @@ -343,6 +353,9 @@ mod tests { start_time: Instant::now(), health_store: Arc::new(RwLock::new(HealthStore::new())), interruption_store: None, + evaluation_store: Arc::new( + EvaluationStore::new_with_path(std::path::Path::new(":memory:")).unwrap(), + ), security_observability: SecurityObservabilityConfig { timeout_ms }, }) } diff --git a/src/agentsight/src/server/token_savings.rs b/src/agentsight/src/server/token_savings.rs index ee5a1a6bc..920aaafc3 100644 --- a/src/agentsight/src/server/token_savings.rs +++ b/src/agentsight/src/server/token_savings.rs @@ -863,6 +863,7 @@ pub async fn get_session_savings( #[cfg(test)] mod tests { use super::*; + use crate::grader::EvaluationStore; use actix_web::test as actix_test; use actix_web::{App, web}; use std::sync::{Arc, Mutex, RwLock}; @@ -929,10 +930,11 @@ mod tests { fn make_app_state(db_path: std::path::PathBuf) -> AppState { AppState { - storage_path: db_path, + storage_path: db_path.clone(), start_time: Instant::now(), health_store: Arc::new(RwLock::new(crate::health::HealthStore::default())), interruption_store: None, + evaluation_store: Arc::new(EvaluationStore::new_with_path(&db_path).unwrap()), security_observability: crate::server::SecurityObservabilityConfig::default(), } }