systemprompt_runtime/trace/repository/
list.rs

1//! Trace listing queries with filter-driven pagination.
2//!
3//! Copyright (c) systemprompt.io — Business Source License 1.1.
4//! See <https://systemprompt.io> for licensing details.
5
6use chrono::{DateTime, Utc};
7
8use systemprompt_identifiers::TraceId;
9
10use super::{Result, TraceRepository};
11use crate::trace::models::{TraceListFilter, TraceListItem};
12
13struct TraceRow {
14    trace_id: TraceId,
15    first_timestamp: DateTime<Utc>,
16    last_timestamp: DateTime<Utc>,
17    agent: Option<String>,
18    status: String,
19    ai_requests: Option<i64>,
20    mcp_calls: Option<i64>,
21}
22
23impl TraceRepository {
24    pub async fn list_traces(&self, filter: &TraceListFilter) -> Result<Vec<TraceListItem>> {
25        let rows = self.fetch_trace_rows(filter).await?;
26
27        Ok(rows
28            .into_iter()
29            .map(|r| TraceListItem {
30                trace_id: r.trace_id,
31                first_timestamp: r.first_timestamp,
32                last_timestamp: r.last_timestamp,
33                agent: r.agent,
34                status: r.status,
35                ai_requests: r.ai_requests.unwrap_or(0),
36                mcp_calls: r.mcp_calls.unwrap_or(0),
37            })
38            .collect())
39    }
40
41    async fn fetch_trace_rows(&self, filter: &TraceListFilter) -> Result<Vec<TraceRow>> {
42        let tool_pattern = filter.tool.as_deref();
43        let agent_pattern = filter.agent.as_ref().map(|a| format!("%{a}%"));
44        let agent_pat = agent_pattern.as_deref();
45        let status_lower = filter.status.as_ref().map(|s| s.to_lowercase());
46        let status_val = status_lower.as_deref();
47        let exclude_system = (!filter.include_system).then_some("1");
48        let require_tracked: Option<&str> = None;
49
50        sqlx::query_as!(
51        TraceRow,
52        r#"
53        WITH tool_traces AS (
54            SELECT DISTINCT trace_id FROM mcp_tool_executions
55            WHERE ($1::text IS NULL OR tool_name ILIKE $1) AND ($2::timestamptz IS NULL OR started_at >= $2)
56        ),
57        all_traces AS (
58            SELECT trace_id, timestamp as ts FROM logs
59            WHERE trace_id IS NOT NULL AND ($2::timestamptz IS NULL OR timestamp >= $2) AND ($1::text IS NULL OR trace_id IN (SELECT trace_id FROM tool_traces))
60            UNION ALL
61            SELECT trace_id, created_at as ts FROM ai_requests
62            WHERE trace_id IS NOT NULL AND ($2::timestamptz IS NULL OR created_at >= $2) AND ($1::text IS NULL OR trace_id IN (SELECT trace_id FROM tool_traces))
63            UNION ALL
64            SELECT trace_id, started_at as ts FROM mcp_tool_executions
65            WHERE trace_id IS NOT NULL AND ($2::timestamptz IS NULL OR started_at >= $2) AND ($1::text IS NULL OR trace_id IN (SELECT trace_id FROM tool_traces))
66            UNION ALL
67            SELECT trace_id, created_at as ts FROM agent_tasks
68            WHERE trace_id IS NOT NULL AND trace_id <> '' AND ($2::timestamptz IS NULL OR created_at >= $2) AND ($1::text IS NULL OR trace_id IN (SELECT trace_id FROM tool_traces))
69        ),
70        grouped AS (
71            SELECT
72                t.trace_id,
73                MIN(t.ts) as first_ts,
74                MAX(t.ts) as last_ts,
75                (SELECT at.agent_name FROM agent_tasks at WHERE at.trace_id = t.trace_id AND at.trace_id <> '' ORDER BY at.updated_at DESC LIMIT 1) as agent,
76                COALESCE(
77                    CASE WHEN EXISTS (SELECT 1 FROM agent_tasks at WHERE at.trace_id = t.trace_id AND at.trace_id <> '' AND at.status IN ('TASK_STATE_PENDING','TASK_STATE_SUBMITTED','TASK_STATE_WORKING','TASK_STATE_INPUT_REQUIRED','TASK_STATE_AUTH_REQUIRED'))
78                        OR EXISTS (SELECT 1 FROM ai_requests ar WHERE ar.trace_id = t.trace_id AND ar.status = 'pending')
79                        OR EXISTS (SELECT 1 FROM mcp_tool_executions mte WHERE mte.trace_id = t.trace_id AND mte.status = 'pending') THEN 'running' END,
80                    CASE WHEN EXISTS (SELECT 1 FROM agent_tasks at WHERE at.trace_id = t.trace_id AND at.trace_id <> '' AND at.status IN ('TASK_STATE_FAILED','TASK_STATE_REJECTED'))
81                        OR EXISTS (SELECT 1 FROM ai_requests ar WHERE ar.trace_id = t.trace_id AND ar.status = 'failed')
82                        OR EXISTS (SELECT 1 FROM mcp_tool_executions mte WHERE mte.trace_id = t.trace_id AND mte.status IN ('failed','timeout'))
83                        OR (NOT EXISTS (SELECT 1 FROM agent_tasks at WHERE at.trace_id = t.trace_id AND at.trace_id <> '') AND NOT EXISTS (SELECT 1 FROM ai_requests ar WHERE ar.trace_id = t.trace_id) AND NOT EXISTS (SELECT 1 FROM mcp_tool_executions mte WHERE mte.trace_id = t.trace_id) AND EXISTS (SELECT 1 FROM logs l WHERE l.trace_id = t.trace_id AND l.level = 'ERROR')) THEN 'failed' END,
84                    CASE WHEN EXISTS (SELECT 1 FROM agent_tasks at WHERE at.trace_id = t.trace_id AND at.trace_id <> '' AND at.status = 'TASK_STATE_CANCELED') THEN 'canceled' END,
85                    'completed'
86                ) as status,
87                (SELECT COUNT(*) FROM ai_requests ar WHERE ar.trace_id = t.trace_id) as ai_requests,
88                (SELECT COUNT(*) FROM mcp_tool_executions mte WHERE mte.trace_id = t.trace_id) as mcp_calls
89            FROM all_traces t
90            GROUP BY t.trace_id
91        )
92        SELECT trace_id as "trace_id!: TraceId", first_ts as "first_timestamp!", last_ts as "last_timestamp!",
93               agent as "agent", status as "status!", ai_requests as "ai_requests", mcp_calls as "mcp_calls"
94        FROM grouped
95        WHERE ($3::text IS NULL OR trace_id != 'system') AND ($4::text IS NULL OR agent ILIKE $4)
96          AND ($5::text IS NULL OR LOWER(status) = $5) AND ($6::boolean IS NOT TRUE OR mcp_calls > 0)
97          AND ($7::text IS NULL OR status IS NOT NULL)
98          AND ($3::text IS NULL OR ai_requests > 0 OR mcp_calls > 0 OR agent IS NOT NULL)
99        ORDER BY first_ts DESC
100        LIMIT $8
101        "#,
102        tool_pattern,
103        filter.since,
104        exclude_system,
105        agent_pat,
106        status_val,
107        Some(filter.has_mcp),
108        require_tracked,
109        filter.limit
110    )
111    .fetch_all(&*self.pool)
112    .await
113    .map_err(Into::into)
114    }
115}
systemprompt_runtime/trace/repository/list.rs

systemprompt_runtime/trace/repository/
list.rs
