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/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 ─────────────────────────────────────────── /** 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: diff --git a/src/agentsight/src/aggregator/http/aggregator.rs b/src/agentsight/src/aggregator/http/aggregator.rs index 25a0a943f..39baa9fbf 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,81 @@ 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 { .. })); + } + + #[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/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/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(); 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..97f8f8eb1 --- /dev/null +++ b/src/agentsight/src/grader/evidence.rs @@ -0,0 +1,302 @@ +//! 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(value_has_error_signal) + } + _ => 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..ba8a2250c --- /dev/null +++ b/src/agentsight/src/grader/input.rs @@ -0,0 +1,257 @@ +//! 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, + interruption_store: Option<&InterruptionStore>, + 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(interruption_store, 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( + interruption_store: Option<&InterruptionStore>, + conversation_id: &str, +) -> Result, GraderError> { + match interruption_store { + Some(store) => store + .list_by_conversation(conversation_id) + .map_err(|error| GraderError::Storage(error.to_string())), + None => Ok(Vec::new()), + } +} + +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, + }) +} + +#[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/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..fd58e98c1 --- /dev/null +++ b/src/agentsight/src/grader/storage.rs @@ -0,0 +1,321 @@ +//! 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 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 + 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) => 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 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 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 + 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) => Some( + read_raw_record(row) + .map_err(|error| GraderError::Storage(error.to_string()))?, + ), + None => None, + } + }; + raw.map(raw_to_record).transpose() + } +} + +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/interruption/detector.rs b/src/agentsight/src/interruption/detector.rs index 76a862794..51e6fe6df 100644 --- a/src/agentsight/src/interruption/detector.rs +++ b/src/agentsight/src/interruption/detector.rs @@ -101,10 +101,15 @@ impl InterruptionDetector { .and_then(|s| s.parse().ok()) .unwrap_or(200); - // 修复：从 call.response.raw_body 读取响应体，而非 call.metadata（builder 不会写入 metadata） let error_text = call.error.as_deref().unwrap_or(""); let response_body = call.response.raw_body.as_deref().unwrap_or(""); - let combined_error = format!("{error_text} {response_body}").to_ascii_lowercase(); + let structured_error = error_text.to_ascii_lowercase(); + let response_error_body = if status_code >= 400 { + response_body + } else { + "" + }; + let combined_error = format!("{error_text} {response_error_body}").to_ascii_lowercase(); let is_context_overflow = combined_error.contains("context_length_exceeded") || combined_error.contains("maximum context length") @@ -174,9 +179,9 @@ impl InterruptionDetector { // ── 3. NetworkTimeout (408/504 / timeout) ───────────────────────────── if status_code == 408 || status_code == 504 - || combined_error.contains("timeout") - || combined_error.contains("timed out") - || combined_error.contains("deadline exceeded") + || structured_error.contains("timeout") + || structured_error.contains("timed out") + || structured_error.contains("deadline exceeded") { let detail = serde_json::json!({ "model": call.model, @@ -754,6 +759,76 @@ mod tests { ); } + #[test] + fn test_ignores_timeout_text_in_successful_response_body() { + let detector = InterruptionDetector::default(); + let mut call = make_base_call(); + call.response.raw_body = Some( + r#"{"content":"Run journalctl -u app | grep -i \"timeout\" to inspect timeout logs."}"# + .to_string(), + ); + + let events = detector.detect(&call); + + assert!(events.is_empty()); + } + + #[test] + fn test_ignores_auth_text_in_successful_response_body() { + let detector = InterruptionDetector::default(); + let mut call = make_base_call(); + call.response.raw_body = Some( + r#"{"content":"Use rejectUnauthorized:false only for local TLS smoke tests."}"# + .to_string(), + ); + + let events = detector.detect(&call); + + assert!(events.is_empty()); + } + + #[test] + fn test_ignores_rate_limit_text_in_successful_response_body() { + let detector = InterruptionDetector::default(); + let mut call = make_base_call(); + call.response.raw_body = Some( + r#"{"content":"Document rate limit and too many requests troubleshooting steps."}"# + .to_string(), + ); + + let events = detector.detect(&call); + + assert!(events.is_empty()); + } + + #[test] + fn test_ignores_service_unavailable_text_in_successful_response_body() { + let detector = InterruptionDetector::default(); + let mut call = make_base_call(); + call.response.raw_body = Some( + r#"{"content":"Explain service_unavailable and overloaded model symptoms."}"# + .to_string(), + ); + + let events = detector.detect(&call); + + assert!(events.is_empty()); + } + + #[test] + fn test_ignores_context_overflow_text_in_successful_response_body() { + let detector = InterruptionDetector::default(); + let mut call = make_base_call(); + call.response.raw_body = Some( + r#"{"content":"Compare context window, input is too long, and prompt is too long errors."}"# + .to_string(), + ); + + let events = detector.detect(&call); + + assert!(events.is_empty()); + } + #[test] fn test_detect_service_unavailable_503() { let detector = InterruptionDetector::default(); diff --git a/src/agentsight/src/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/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/server/handlers.rs b/src/agentsight/src/server/handlers.rs index 5e09893d8..5fc58c00b 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, 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, + data.interruption_store.as_deref(), + &body.target_id, + body.force, + ) { + Ok(input) => input, + Err(error) => return grader_error_response(error), + }; + let store = &data.evaluation_store; + + 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"})); + } + + 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), + } +} + +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 @@ -609,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::*; @@ -822,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 @@ -844,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 956ef77d7..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, } @@ -129,6 +132,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) @@ -194,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() { @@ -242,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, }); @@ -283,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::{ @@ -341,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(), } } 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}"); } 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.